StateGraph、条件边与有限循环
图中的节点读取状态并返回局部更新,边决定下一节点,reducer 决定更新如何合并。TypedDict 提供类型描述,不自动替你验证不可信运行时输入;入口仍需校验。
安装与第一次执行
新目录中执行,后续本书复用此环境:
bash
python3.12 -m venv .venv
source .venv/bin/activate
python -m pip install 'langgraph==1.0.8' 'langgraph-prebuilt==1.0.7' 'langchain-core==1.6.4'以下为完整 lab.py,不需要网络、模型或密钥。
python
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
class State(TypedDict):
count: int
target: int
def step(state: State):
return {"count":state["count"] + 1}
def route(state: State):
return END if state["count"] >= state["target"] else "step"
builder = StateGraph(State)
builder.add_node("step",step)
builder.add_edge(START,"step")
builder.add_conditional_edges("step",route)
graph = builder.compile()
result = graph.invoke({"count":0,"target":3}, {"recursion_limit":10})
assert result == {"count":3,"target":3}
print(result)每轮只写 count,target 原值保留。条件边在节点更新后读取新 count,第三轮进入 END。若把判断写成相反方向,程序可能无限循环;recursion_limit 是运行时最后防线,不是业务成功条件。业务还要设置 token、时间与无进展限制。
流式更新与动态路由
将最后调用单独替换为 list(graph.stream({"count":0,"target":3}, stream_mode="updates")),会得到三个 step 局部更新;values 模式观察完整状态。它们不是模型 token 流,若节点内部调用模型,需要对应消息流模式。客户端显示中间 count 不表示最终图已经成功。
模型生成路由名时先解析枚举并检查权限,不允许任意字符串选择内部管理节点。固定 DAG 适合已知业务顺序,图循环承载 Agent 的动态工具选择。节点出错不应为了走 END 而吞异常写“成功”;记录错误类别并选择有限重试、降级或人工路径。
旧教程可能使用 MessageGraph 或预构建 Agent;先确认状态是消息列表还是 TypedDict,不能把输入对象直接混用。保持旧程序的输入/输出合同,再将状态明确化,可避免迁移后消息被覆盖。
练习
将 target 改成100、recursion_limit设为3,捕获 GraphRecursionError;不能当作 completed。再用target=2与limit=10验证成功路径。