[智能体-323]：LangGraph如何定义图？如何定义分支？如何循环？如何定义人在回路？如何执行图?

结合实操逻辑与代码示例分步讲解LangGraph 定义图、分支、循环、人在回路、执行流程基于官方标准写法通俗易懂 可落地。一、前置基础核心四要素State全局状态所有节点读写的统一数据容器Node执行单元函数 / 逻辑块Edge流向普通边、条件边Graph状态图本体串联所有节点与流向依赖包langgraph、langchain、langchain-openai二、1. 定义状态State所有节点共享同一份状态是数据流转的载体使用TypedDict定义。python运行from typing import TypedDict, Annotated, Sequence import operator # 定义全局状态 class GraphState(TypedDict): messages: Annotated[Sequence[str], operator.add] # 对话消息自动追加 query: str # 用户问题 tool_result: str # 工具返回结果 final_answer: str # 最终答案Annotated[..., operator.add]列表类型字段新内容追加而非覆盖适配多轮对话。三、2. 定义节点Node节点是独立执行逻辑入参 State出参 更新后的 State 字典。 示例创建 3 个基础节点python运行# 节点1大模型思考判断 def llm_think(state: GraphState) - dict: print(【LLM 思考节点】) return {messages: [LLM 完成思考]} # 节点2调用工具检索/查询 def call_tool(state: GraphState) - dict: print(【工具调用节点】) return {tool_result: 工具查询到的内容} # 节点3生成最终答案 def generate_answer(state: GraphState) - dict: print(【答案生成节点】) return {final_answer: 基于信息生成最终回答}四、3. 定义图Graph本体使用StateGraph基于自定义 State 创建图结构步骤实例化 → 添加节点 → 设置入口。python运行from langgraph.graph import StateGraph, START, END # 1. 实例化状态图绑定自定义State workflow StateGraph(GraphState) # 2. 把上面定义的节点注册到图中名称 函数 workflow.add_node(llm_think, llm_think) workflow.add_node(call_tool, call_tool) workflow.add_node(generate_answer, generate_answer) # 3. 设置图的入口节点流程起点 workflow.add_edge(START, llm_think)START图内置起点END图内置终点。五、4. 定义分支条件路由分支依靠条件边Conditional Edge实现根据 State 中的字段动态选择下一个节点。4.1 编写分支判断函数函数入参为State返回目标节点名称字符串。python运行def route_by_content(state: GraphState) - str: 分支判断逻辑模拟判断是否需要调用工具 # 业务规则有查询问题 → 走工具无 → 直接生成答案 if state.get(query): return call_tool else: return generate_answer4.2 绑定条件分支到图python运行# 从 llm_think 节点出发执行分支判断 workflow.add_conditional_edges( sourcellm_think, # 起始节点 pathroute_by_content, # 分支判断函数 path_map{ # 映射函数返回值 → 对应节点名 call_tool: call_tool, generate_answer: generate_answer } ) # 普通边工具执行完成后走到答案生成节点 workflow.add_edge(call_tool, generate_answer) # 普通边答案生成完成 → 流程结束 workflow.add_edge(generate_answer, END)流程逻辑START→llm_think→ 【分支判断】→call_tool/generate_answer→END六、5. 定义循环有环流程循环本质条件边 回指边执行完节点后重新回到上游节点形成迭代。改造增加循环逻辑模拟「信息不足→重新查询」步骤 1新增循环判断函数python运行def need_loop(state: GraphState) - str: 循环判断模拟信息是否充足 # 模拟规则工具结果为空 → 继续循环不为空 → 结束循环 if not state.get(tool_result): return llm_think # 回到思考节点开启新一轮循环 else: return generate_answer步骤 2修改流向构建环python运行# 清空原有流向重新配置演示循环 workflow StateGraph(GraphState) workflow.add_node(llm_think, llm_think) workflow.add_node(call_tool, call_tool) workflow.add_node(generate_answer, generate_answer) workflow.add_edge(START, llm_think) # 1. 思考 → 固定走到工具节点 workflow.add_edge(llm_think, call_tool) # 2. 工具执行后进入循环判断核心环的入口 workflow.add_conditional_edges( sourcecall_tool, pathneed_loop, path_map{ llm_think: llm_think, # 回跳上游节点 循环 generate_answer: generate_answer } ) # 3. 最终节点走向结束 workflow.add_edge(generate_answer, END)循环流转图START→llm_think→call_tool→ 条件判断信息不足 → 回到llm_think循环执行信息充足 →generate_answer→END七、6. 定义人在回路Human-in-the-Loop核心能力流程暂停、人工介入、恢复执行两种主流实现方式方式 1全局暂停推荐简单易用编译图时开启interrupt_before指定在某个节点执行前暂停等待人工操作。python运行# 编译图并设置执行 call_tool 节点之前暂停 app workflow.compile(interrupt_before[call_tool])运行 人工交互流程首次执行流程走到暂停点返回当前 State人工修改状态补充信息、修正参数调用app.invoke(None)恢复流程继续执行。完整交互示例python运行# 初始输入 inputs {query: 查询数据, tool_result: } # 第一轮执行触发暂停 config {configurable: {thread_id: 1}} # thread_id 区分会话 state app.invoke(inputs, configconfig) print(流程已暂停等待人工操作) # 人工介入修改状态 current_state app.get_state(config) # 人工补充工具结果 app.update_state(config, {tool_result: 人工补充的查询内容}) # 恢复流程继续执行 final_state app.invoke(None, configconfig) print(流程执行完毕, final_state[final_answer])方式 2自定义人工节点新增专属人工节点把人工操作作为普通流程节点灵活性更高适合复杂审核场景。八、7. 执行图完整运行方式图必须先compile编译再执行提供三种执行模式8.1 基础编译python运行# 普通编译无暂停 app workflow.compile() # 带暂停人在回路专用 app workflow.compile(interrupt_before[节点名])8.2 三种执行 API1invoke 同步执行全流程跑完返回最终状态python运行inputs {query: 测试问题} result app.invoke(inputs) print(result)2stream 流式执行逐节点返回结果看中间过程适合调试、前端实时展示python运行for event in app.stream(inputs): print(event)3多会话隔离thread_id通过thread_id区分不同用户 / 会话搭配检查点可持久化状态python运行config {configurable: {thread_id: session_001}} app.invoke(inputs, configconfig)九、核心总结速记定义图用TypedDict定义State编写业务Node函数StateGraph实例化 add_node注册节点 设置START入口定义分支编写条件判断函数→add_conditional_edges绑定路由根据状态动态跳转节点。定义循环基于条件边让流程回跳到上游节点形成有环图实现迭代执行。定义人在回路编译时配置interrupt_before指定暂停节点 → 执行到节点前自动暂停 →update_state人工改状态 → 再次invoke恢复。执行图先compile编译 → 使用invoke/stream运行thread_id实现多会话隔离。