第06章LangGraph 入门版本LangChain v1.3.7 | 讲师汤姆小白1. LangGraph 是什么LangGraph是 LangChain 生态的底层编排运行时专门用于构建有状态、长时间运行的 Agent。简单理解LangChain 做零件LangGraph 做组装线。三者关系产品定位适用场景LangChain基础构件LLM、Tools、Retrieval提供标准化的组件LangGraph编排运行时复杂工作流、多步推理、持久化Deep Agents开箱即用 Agent 套件高级 Agent内置规划/子代理create_agent底层就是基于 LangGraph 实现的。学 LangGraph 就是学 Agent 的底层原理。安装pipinstalllanggraph2. StateGraph定义工作流2.1 最简单的图fromlanggraph.graphimportStateGraph,MessagesState,START,ENDfromlangchain.chat_modelsimportinit_chat_model modelinit_chat_model(openai:gpt-4o-mini)defcall_model(state:MessagesState):节点函数调用模型responsemodel.invoke(state[messages])return{messages:[response]}# 创建图graphStateGraph(MessagesState)# 添加节点graph.add_node(model,call_model)# 添加边graph.add_edge(START,model)graph.add_edge(model,END)# 编译并执行appgraph.compile()resultapp.invoke({messages:[{role:user,content:你好}]})print(result[messages][-1].content)2.2 StateGraph 核心概念概念说明StateGraph图的容器定义工作流的状态类型State在节点间传递的共享数据如MessagesState是预定义的消息列表状态Node节点一个可执行的函数接收 State 并返回 State 的部分更新Edge边节点之间的连接定义执行顺序START / END特殊内置节点表示图的起点和终点2.3 自定义 StatefromtypingimportTypedDict,Annotatedfromlanggraph.graphimportStateGraph,START,ENDclassMyState(TypedDict):messages:listcounter:intfinal_answer:strdefstep1(state:MyState):return{counter:state[counter]1}defstep2(state:MyState):return{final_answer:f执行了{state[counter]}步}graphStateGraph(MyState)graph.add_node(step1,step1)graph.add_node(step2,step2)graph.add_edge(START,step1)graph.add_edge(step1,step2)graph.add_edge(step2,END)appgraph.compile()resultapp.invoke({messages:[],counter:0,final_answer:})print(result[final_answer])# 执行了 1 步3. 条件边根据状态选择路径条件边让工作流能根据运行时状态动态决定下一步。importrandomfromlanggraph.graphimportStateGraph,MessagesState,START,END modelinit_chat_model(openai:gpt-4o-mini)defcall_model(state:MessagesState):responsemodel.invoke(state[messages])return{messages:[response]}defshould_continue(state:MessagesState)-str:条件判断函数返回下一个节点名last_messagestate[messages][-1]# 检查是否需要工具调用ifhasattr(last_message,tool_calls)andlast_message.tool_calls:returntoolsreturnENDdefcall_tools(state:MessagesState):执行工具last_messagestate[messages][-1]# 这里简化处理return{messages:[{role:tool,content:工具结果}]}graphStateGraph(MessagesState)graph.add_node(model,call_model)graph.add_node(tools,call_tools)graph.add_edge(START,model)graph.add_conditional_edges(model,should_continue,{tools:tools,END:END,})graph.add_edge(tools,model)# 工具结果返回模型appgraph.compile()条件边的三个参数(源节点, 判断函数, 路由映射)。判断函数返回的字符串必须在路由映射中存在。4. 持久化断点恢复Agent 执行中可能因网络、API 限流等原因中断。持久化让 Agent 能从断点继续不丢失进度。fromlanggraph.checkpoint.memoryimportMemorySaver# 创建内存检查点生产环境推荐用 SqliteSaver 或 PostgresSavercheckpointerMemorySaver()graphStateGraph(MessagesState)# ... 添加节点和边 ...appgraph.compile(checkpointercheckpointer)# 执行时指定 thread_idconfig{configurable:{thread_id:user_123}}# 第一次调用result1app.invoke({messages:[{role:user,content:列出3个Python框架}]},configconfig)# 即使程序崩溃重启用同一个 thread_id 继续result2app.invoke({messages:[{role:user,content:详细说明第一个}]},configconfig# 同一个 thread_id自动加载历史)5. 人机协同关键节点人工审批在敏感操作如删除文件、发送邮件前暂停等待人工确认。fromlanggraph.typesimportinterruptdefsensitive_operation(state:MessagesState):需要人工审批的节点# 暂停并询问用户approvalinterrupt(即将删除重要数据是否继续)ifapprovalyes:# 执行操作return{messages:[{role:ai,content:操作已执行}]}else:return{messages:[{role:ai,content:操作已取消}]}# 在图中使用graph.add_node(sensitive_op,sensitive_operation)appgraph.compile(checkpointercheckpointer)# 第一次调用会在 interrupt() 处暂停resultapp.invoke(input_data,config)# 此时 result 包含中断信息# 用户确认后恢复执行app.invoke(None,# 无新输入config,# 传入用户确认的 Command)6. Stream实时监控执行过程# 流式输出每个节点的结果foreventinapp.stream({messages:[{role:user,content:你好}]},config{configurable:{thread_id:1}},stream_modevalues,):# event 是当前完整状态last_msgevent[messages][-1]print(f[{last_msg.get(role,node)}]{last_msg.get(content,)})7. LangGraph 与 create_agent 的关系create_agent本质上就是预先配置好的 LangGraph# create_agent 内部做的事 ≈fromlanggraph.prebuiltimportcreate_react_agent# create_agent 自动创建了# - model 节点# - tools 节点# - 条件边有工具调用→继续无→结束# - 中间件机制# - 流式支持理解 LangGraph 后你就能在create_agent不够灵活时用 LangGraph 自定义 Agent 行为添加自定义逻辑到 Agent 循环中实现复杂的多 Agent 协作本章小结概念作用StateGraph定义工作流图的容器Node可执行函数操作 StateEdge节点间的执行顺序Conditional Edge根据状态动态路由Checkpointer持久化状态断点恢复interrupt暂停等待人工审批stream实时监控图执行LangGraph 是 Agent 的骨架create_agent 是 LangGraph 的快捷方式。先学 create_agent 快速上手再用 LangGraph 深入定制。