如何用langgraph实现的循环嵌套子图功能?

摘要:目录 1. 环境准备 2. 统一约定(建议先看) 3. test1:单节点最小图 4. test2:单节点内做策略分流 5. test3:多节点串行执行 6. test4:条件分支图(add_conditional_edges) 7. te
目录 1. 环境准备 2. 统一约定(建议先看) 3. test1:单节点最小图 4. test2:单节点内做策略分流 5. test3:多节点串行执行 6. test4:条件分支图(add_conditional_edges) 7. test5:循环图(猜数字) 8. test6:父图 + 子图(字段不一致时怎么接) 9. 常见报错与排查 1. 环境准备 # 在当前 notebook 内核里安装/升级到 1.x %pip install -U "langgraph>=1.0.0,<2.0.0" from langgraph.graph import StateGraph, START, END from typing import TypedDict, List, NotRequired, Literal 2. 统一约定(建议先看) 2.1 节点函数尽量返回“增量字典” 推荐: def node(state): # 只返回这个节点更新的字段 return {"x": 123} 不推荐(虽然很多时候也能跑): def node(state): state["x"] = 123 return state 原因很简单:后面一旦有并发分支、reducer、子图拼接,返回增量更稳,也更容易看出每个节点到底改了什么。 2.2 START / END 写法 用 graph.add_edge(START, "nodeA") 指定入口 用 graph.add_edge("nodeZ", END) 指定结束 这套写法和 add_conditional_edges(...) 配合最顺手。 2.3 编译时机 add_node / add_edge / add_conditional_edges 都在 compile() 之前完成 图结构改了以后要重新 compile() 3. test1:单节点最小图 目标:输入名字,输出一句问候。 链路说明:START -> greeting -> END 功能说明: 输入:messages(用户名字或原始文案)。 输出:更新后的 messages(在前面拼接问候语)。 结构:只有 1 个节点 greeting,链路是 START -> greeting -> END。 适用场景:先验证环境、验证图是否能跑通、熟悉最小 StateGraph 代码骨架。 from langgraph.graph import StateGraph, START, END from typing import TypedDict class AgentState(TypedDict): # 这个 demo 只有一个字段,最小可运行 messages: str def greeting_node(state: AgentState): # 节点里只做一件事:拼接问候语 # 返回增量字典,表示只更新 messages return {"messages": "Hello! " + state["messages"]} graph = StateGraph(AgentState) # 注册节点:节点名 + 节点函数 graph.add_node("greeting", greeting_node) # 入口从 START 进 greeting graph.add_edge(START, "greeting") # greeting 执行完就结束 graph.add_edge("greeting", END) app = graph.compile() result = app.invoke({"messages": "kunyashaw"}) print(result) 可视化: from IPython.display import Image, display display(Image(app.get_graph().draw_mermaid_png())) 4. test2:单节点内做策略分流 目标:同一个节点里按 operation 选择加法或乘法。 链路说明:START -> smartNode -> END(分流发生在 smartNode 节点函数内部) 功能说明: 输入:numList(数字列表)和 operation(+ 或 *)。 输出:theResult(计算结果)。 结构:图上仍是单节点,但节点内部有分支判断逻辑。 适用场景:不想把流程拆成多个节点,只想在一个节点里根据参数走不同算法。
阅读全文