如何用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(计算结果)。
结构:图上仍是单节点,但节点内部有分支判断逻辑。
适用场景:不想把流程拆成多个节点,只想在一个节点里根据参数走不同算法。
