LangGraph-Tick概念
注:以下所提的“文档”,是指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 完成后,调度器在同一轮判断到
- 通过静态边 flight_agent → finalize,finalize 就绪;
- 通过 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 同一键多次写入” 的错误来源与修复方式了。
更多推荐
所有评论(0)