下面分基础线性图、带分支、带循环、带人工介入人在回路四类场景由浅入深给出完整可运行代码同时逐行讲解适配学习与落地。前置依赖安装bash运行pip install langgraph langchain langchain-openai一、核心前置知识点回顾State全局状态容器用TypedDict定义Annotated operator.add实现列表追加Node函数入参是状态返回「待更新的状态字典」消息在状态中StateGraph图主体负责挂载节点、配置流向START / END图内置起点、终点示例 1最简线性图无分支、无循环纯串行流程起点 → 思考节点 → 回答节点 → 结束python运行from typing import TypedDict, Annotated import operator from langgraph.graph import StateGraph, START, END # 1. 定义全局状态定义state中需要保存哪些信息 class GraphState(TypedDict): # 消息列表自动追加不覆盖 messages: Annotated[list[str], operator.add] query: str # 用户问题 answer: str # 最终回答 # 2. 定义节点执行逻辑 def llm_think(state: GraphState) - dict: 思考节点 print( 进入思考节点 ) return {messages: [完成问题分析]} def generate_answer(state: GraphState) - dict: 生成回答节点 print( 进入回答节点 ) ans f针对问题「{state[query]}」给出回复 return {answer: ans, messages: [生成回答完毕]} # 3. 构建图 # 绑定自定义状态创建图实例基于state的图 workflow StateGraph(GraphState) # 注册节点节点名称 对应节点执行函数 workflow.add_node(llm_think, llm_think) workflow.add_node(generate_answer, generate_answer) # 配置流向普通边把节点连接起来: 起节点 终节点 workflow.add_edge(START, llm_think) # 起点 → 思考节点 workflow.add_edge(llm_think, generate_answer) # 思考 → 回答 workflow.add_edge(generate_answer, END) # 回答 → 结束 # 4. 编译图必须编译才能运行 app workflow.compile() # 5. 执行图 if __name__ __main__: inputs {query: LangGraph是什么, messages: []} result app.invoke(inputs) print(\n最终结果, result)示例 2带条件分支的图根据状态字段动态选择下一个节点使用add_conditional_edges实现分支。流程起点 → 思考 → 【分支判断】→ 调用工具 / 直接回答 → 结束python运行from typing import TypedDict, Annotated import operator from langgraph.graph import StateGraph, START, END # 1. 定义状态 class GraphState(TypedDict): messages: Annotated[list[str], operator.add] query: str need_tool: bool # 标记是否需要调用工具 tool_res: str answer: str # 2. 定义节点 def llm_judge(state: GraphState) - dict: 判断是否需要工具 print( 意图判断节点 ) # 模拟逻辑包含查询则需要工具 need 查询 in state[query] return {need_tool: need, messages: [完成意图判断]} def call_tool(state: GraphState) - dict: 工具调用节点 print( 工具调用节点 ) return {tool_res: 外部数据查询结果, messages: [工具调用完成]} def gen_answer(state: GraphState) - dict: 生成回答 print( 回答生成节点 ) if state[need_tool]: ans f结合工具结果{state[tool_res]} 回答问题 else: ans 直接基于常识回答问题 return {answer: ans, messages: [回答生成完毕]} # 3. 分支路由函数核心返回目标节点名 def route_node(state: GraphState) - str: if state[need_tool]: return call_tool return gen_answer # 4. 构建图 workflow StateGraph(GraphState) workflow.add_node(llm_judge, llm_judge) workflow.add_node(call_tool, call_tool) workflow.add_node(gen_answer, gen_answer) # 基础流向 workflow.add_edge(START, llm_judge) # 配置条件分支 workflow.add_conditional_edges( sourcellm_judge, # 分支起始节点 pathroute_node, # 路由判断函数 path_map{ # 映射函数返回值 → 节点名 call_tool: call_tool, gen_answer: gen_answer } ) # 统一收尾两条分支最终都走到回答节点再结束 workflow.add_edge(call_tool, gen_answer) workflow.add_edge(gen_answer, END) # 编译 执行 app workflow.compile() if __name__ __main__: # 测试1需要工具 res1 app.invoke({query: 查询今日天气, messages: []}) print(- * 30) # 测试2不需要工具 res2 app.invoke({query: 你好, messages: []})示例 3带循环的图有环图利用条件边回跳上游节点实现循环模拟「信息不足 → 重复查询」。流程起点 → 思考 → 工具调用 → 循环判断 → 继续循环 / 结束循环python运行from typing import TypedDict, Annotated import operator from langgraph.graph import StateGraph, START, END class GraphState(TypedDict): messages: Annotated[list[str], operator.add] query: str tool_res: str loop_count: int # 循环计数器 # 节点 def think(state: GraphState) - dict: print(f 第{state[loop_count]}轮思考 ) return {messages: [思考完成]} def search_tool(state: GraphState) - dict: print( 执行查询 ) cnt state[loop_count] 1 # 模拟循环2次后才拿到有效数据 if cnt 2: return {tool_res: 有效数据, loop_count: cnt, messages: [获取到有效数据]} return {tool_res: , loop_count: cnt, messages: [数据不足继续查询]} # 循环路由函数 def check_loop(state: GraphState) - str: # 无结果 → 回到思考节点循环有结果 → 结束循环 if not state[tool_res]: return think return END # 构建图 workflow StateGraph(GraphState) workflow.add_node(think, think) workflow.add_node(search_tool, search_tool) workflow.add_edge(START, think) workflow.add_edge(think, search_tool) # 循环条件边工具执行后判断是否回跳 workflow.add_conditional_edges( sourcesearch_tool, pathcheck_loop, path_map{ think: think, END: END } ) app workflow.compile() if __name__ __main__: inputs {query: 查找资料, messages: [], loop_count: 0, tool_res: } result app.invoke(inputs) print(最终状态, result)示例 4带「人在回路」的图流程暂停 人工修改状态核心编译时指定interrupt_before在节点执行前暂停人工修改状态后继续运行。python运行from typing import TypedDict, Annotated import operator from langgraph.graph import StateGraph, START, END class GraphState(TypedDict): messages: Annotated[list[str], operator.add] content: str result: str # 节点 def pre_process(state: GraphState) - dict: print( 前置处理完成 ) return {messages: [预处理结束]} def human_review(state: GraphState) - dict: print( 正式处理节点 ) res f最终输出{state[content]} return {result: res} # 构建图 workflow StateGraph(GraphState) workflow.add_node(pre_process, pre_process) workflow.add_node(human_review, human_review) workflow.add_edge(START, pre_process) workflow.add_edge(pre_process, human_review) workflow.add_edge(human_review, END) # 关键设置暂停点 # 在 human_review 执行前暂停等待人工介入 app workflow.compile(interrupt_before[human_review]) if __name__ __main__: # 会话ID区分不同会话、持久化状态 config {configurable: {thread_id: session_001}} inputs {content: , messages: []} # 1. 执行到暂停点 print(--- 第一轮执行触发暂停 ---) app.invoke(inputs, configconfig) # 2. 人工查看 修改状态 current_state app.get_state(config) print(当前待处理内容, current_state.values[content]) # 人工补充内容 app.update_state(config, {content: 人工审核后补充的内容}) # 3. 恢复流程继续执行 print(\n--- 人工操作完成恢复执行 ---) final app.invoke(None, configconfig) print(最终结果, final[result])通用编写步骤固定模板记忆套用定义 State用TypedDict列表字段搭配Annotated operator.add实现追加编写 Node 函数入参为状态返回字典更新状态创建 StateGraph传入 State 实例化图注册节点add_node(节点名, 函数)配置流向串行add_edge(起点, 终点)分支 / 循环add_conditional_edges 路由函数编译图compile()需要人在回路则加interrupt_before执行图invoke(输入数据)/stream()运行补充执行方式app.invoke(inputs)同步执行返回最终全量状态for evt in app.stream(inputs)流式逐节点输出适合调试 / 前端展示