注:以下所提的“文档”,是指LangGraph官方指南(https://langchain-ai.github.io/langgraph/guides/),参考版本是0.6.8。

以下代码的开发环境

[project]
name = "my-langgraph"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
    "dotenv>=0.9.9",
    "langchain>=0.3.27",
    "langchain-openai>=0.3.33",
    "langfuse>=3.5.0",
    "langgraph>=0.6.7",
    "langgraph-checkpoint-sqlite>=2.0.11",
    "numpy>=2.3.3",
]

LangGraph 中,tick(调度步)是调度器(scheduler)在执行图时的最小运行周期调度迭代,它决定了每一轮图中哪些节点可以被触发、状态如何推进。可以把它理解成一次“调度循环”或“心跳”。

一、 tick 的主要作用

1.    扫描可执行节点

每个 tick,调度器会检查当前状态(State)和图的依赖关系,找到那些:
        o    输入条件已满足,
        o    上游节点已经完成,
        o    状态符合该节点触发条件的节点。

2.    触发节点运行

对所有可执行的节点,调度器在同一个 tick 内依次/并发调度执行它们的节点函数,并收集它们的输出。
3.    更新状态并推进图
节点返回的输出会合并到全局 State 中(如果节点返回的不是 State 或未改变状态,就不会更新),从而影响下一个 tick 的调度结果。
4.    判断流程是否结束
如果在某个 tick 里没有任何新节点可以运行且没有挂起的异步工作,图就认为执行完成。

二、 tick 与常见场景的关系

  • 普通线性流程
    一个 tick 可能只执行一个节点,然后推进到下一个;每次 tick 后状态更新,解锁下一个节点。
  • 并行节点
    当多个节点同时就绪时,调度器会在同一 tick 内并发运行它们。
  • 中断/人工干预 (interrupt / human-in-the-loop)
    如果某节点返回 Command.Interrupt 或挂起等待外部输入,调度器会停止在该 tick 推进;用户恢复后再进行下一次 tick。
  • 从 checkpoint 恢复
    如果通过 checkpoint_id 恢复,调度器会在下一次 tick 时以该状态继续扫描可运行的节点。

三、 关键特征总结

特性

描述

最小调度单位

每次 tick 是一次完整的调度扫描与执行循环

状态驱动

依据 State 和图结构决定哪些节点可以执行

支持并发

同一个 tick 可执行多个独立节点

与 checkpoint 配合

从中间恢复时,下一个 tick 会基于恢复后的状态继续运行

与 stream API 对应

graph.stream(...) 每次迭代的产出通常就是每个 tick 的执行结果

直观的解释是,如果在某个节点有多个分支,langgraph判断每个分支都可以执行,就把这些分支的节点放在同一个tick里,最后合并状态的结果。

但如果合并状态结果时,状态中某个属性值节点返回的值有不一样,则会出错:langgraph.errors.InvalidUpdateError: At key 'last_action': Can receive only one value per step. Use an Annotated key to handle multiple values.

四、情况演示:

要点对应你给的图与约束:

  • 图结构:START → flight_agent → finalize,hotel_agent → finalize,finalize → END
  • MState 只有一个键:last_action: str
  • 每个 node 都会写 last_action
  • flight_agent 最后根据 handoff 返回 Command(goto="hotel_agent", update=updates)(这样同一 tick 内既能沿静态边触发 finalize,又能按 goto 触发 hotel_agent,从而多次写同一键)
# pip install langgraph langchain-core typing_extensions
from typing import TypedDict, Optional
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command

# --- 1) 状态定义(单值字段)---
class MState(TypedDict, total=False):
    last_action: str

# --- 2) 节点实现:每个节点都会写 last_action ---
def flight_agent(state: MState) -> Command:
    # 本节点先写一次 last_action
    updates: MState = {"last_action": "Booked flight SFO→JFK on 2025-10-01"}
    # 设定 handoff 给 hotel_agent:同一 tick 内既会触发 finalize(静态边),也会触发 hotel_agent(goto)
    plan = {"handoff": "hotel_agent"}
    if plan["handoff"] == "flight_agent":
        # 这是你给的分支之一:自跳(这里不用来复现冲突)
        return Command(goto="flight_agent", update=updates)
    # 关键:handoff 到 hotel_agent,制造同一 tick 的多写冲突
    return Command(goto="hotel_agent", update=updates)

def hotel_agent(state: MState) -> MState:
    # 同一个 tick 内,hotel_agent 再写一次 last_action
    return {"last_action": "Hotel handoff accepted (hotel_agent running)"}

def finalize(state: MState) -> MState:
    # 同一个 tick 内,finalize 也写 last_action(与 hotel_agent 形成冲突)
    return {"last_action": "finalized summary generated"}

# --- 3) 按你给的图构建 ---
g = StateGraph(MState)
g.add_node("flight_agent", flight_agent)
g.add_node("hotel_agent", hotel_agent)
g.add_node("finalize", finalize)

g.add_edge(START, "flight_agent")
g.add_edge("flight_agent", "finalize")
g.add_edge("hotel_agent", "finalize")
g.add_edge("finalize", END)

app = g.compile()

# --- 4) 运行并观察错误 ---
if __name__ == "__main__":
    try:
        # 任何空/初始 state 均可;关键在于 tick 调度顺序
        for ev in app.stream({}):
            print(ev)
    except Exception as e:
        # 这里会抛出 InvalidUpdateError:
        #   At key 'last_action': Can receive only one value per step...
        print("ERROR =>", repr(e))

为什么这段代码会报错

  • 同一个 tick 内执行了多个节点:flight_agent 完成后,调度器在同一轮判断到
    1. 通过静态边 flight_agent → finalize,finalize 就绪;
    2. 通过 Command(goto="hotel_agent"),hotel_agent 也就绪;
  • flight_agent、hotel_agent、finalize 都写了 last_action(单值字段),因此同一 tick 内对同一键出现多次写 → 触发 InvalidUpdateError。

三种解决思路(任选其一)

1.  把 last_action 改成可收集型字段

from typing_extensions import Annotated
from operator import add

class MState(TypedDict, total=False):
    # 收集多值(同一 tick 聚合)
    last_action: Annotated[list[str], add]

然后各节点写 {"last_action": ["..."]}(注意是列表),即可在同一 tick 聚合。

2.  插入 tick 边界(避免同一 tick 多写)
在 flight_agent handoff 后使用 Command.interrupt(update=updates);恢复时再跑 hotel_agent 或 finalize,把它们分到下一 tick。

3.  条件路由/守卫
移除 flight_agent → finalize 的静态边,改用 add_conditional_edges 按状态决定走 hotel_agent 或 finalize;或让 finalize 在检测到 handoff 时不写 last_action(no-op)。

这样你就能稳定复现并理解这类 “同一 tick 同一键多次写入” 的错误来源与修复方式了。

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐