文章目录前言1. 条件边让图学会“看情况”1.1 路由函数的写法2. 安全的循环Agent 的思考-行动循环2.1 一个极简的 ReAct 循环3. 持久化与检查点为图装上“记忆”3.1 内置检查点3.2 暂停与恢复3.3 时间旅行回溯历史4. 人机交互关键时刻让人来把关4.1 中断节点的执行4.2 暂停后如何处理4.3 完整示例带人工审批的回复发送5. 完整动手实验智能客服审批流6. 小结与下篇预告前言在上一篇博客里我们认识了 LangGraph 的三要素——状态、节点和边并且用它们搭出了一个会循环的简单图。但那个例子离真正的“智能体”还差得远它只会机械地重复同样的话没有根据对话内容做出判断的能力而且一旦跑起来就再也停不下更无法在中间插手。今天这篇我们就为图注入真正的决策能力和记忆。你将学到如何用条件边实现智能路由让 LLM 决定下一步做什么如何安全地构建思考-行动循环模拟一个 Agent 的核心逻辑如何通过检查点让图拥有持久化“记忆”支持暂停、恢复和时间旅行如何嵌入人机交互让关键操作必须经过人工审批才能继续读完这篇你就能搭出真正“活”的流程它会根据情况选择不同的分支会在适当的时候停下来等你点头也会记住自己做过的事。1. 条件边让图学会“看情况”条件边的本质是在节点执行完之后根据状态中的某些信息动态选择下一个节点。这给了我们实现“意图识别”、“超时处理”、“工具调用循环”等高级模式的基础。1.1 路由函数的写法add_conditional_edges需要三个参数源节点名称一个路由函数接收当前状态返回一个字符串代表分支名称一个映射字典把分支名称映射到实际的目标节点或END来看一个实际的例子假设我们的聊天机器人要能区分“普通闲聊”和“需要计算”两类问题。我们先用一个 LLM 节点来分析用户的意图然后根据意图走不同的分支。fromtypingimportTypedDict,Annotated,Literalfromlanggraph.graphimportStateGraph,END,add_messagesfromlangchain_core.messagesimportHumanMessage,AIMessage# 定义状态classState(TypedDict):messages:Annotated[list,add_messages]intent:str# 新增一个字段存放意图分类结果# 意图分类节点defclassify_intent(state:State)-dict:# 模拟分类逻辑如果用户消息包含“计算”则归类为 mathlast_msgstate[messages][-1].contentif计算inlast_msg:return{intent:math}else:return{intent:chat}# 闲聊处理节点defchat_node(state:State)-dict:return{messages:[AIMessage(content这是个闲聊问题我简单回复一下。)]}# 数学处理节点defmath_node(state:State)-dict:return{messages:[AIMessage(content这是需要计算的问题我来认真算一下。)]}# 路由函数根据 intent 字段决定下一步defroute_by_intent(state:State)-Literal[chat_node,math_node]:returnstate[intent]# 直接返回 chat 或 math# 构建图graphStateGraph(State)graph.add_node(classify,classify_intent)graph.add_node(chat_node,chat_node)graph.add_node(math_node,math_node)graph.set_entry_point(classify)graph.add_conditional_edges(classify,route_by_intent,{chat:chat_node,math:math_node})# 两个处理节点执行完直接结束graph.add_edge(chat_node,END)graph.add_edge(math_node,END)appgraph.compile()测试一下initial_state{messages:[HumanMessage(content帮我计算 23)],intent:}resultapp.invoke(initial_state)print(result[messages][-1].content)# 输出这是需要计算的问题我来认真算一下。路由函数route_by_intent根据状态里的intent字段决定走哪个分支。注意它的返回值只能是映射字典中存在的键否则会报错。这种模式非常清晰——节点负责产生状态边负责读取状态并决定流向。2. 安全的循环Agent 的思考-行动循环真正的 Agent 往往需要多次调用 LLM每次调用都可能请求新的工具或产生新的思考结果。这个过程天然就是循环的。但循环也带来了风险如果终止条件没写好图就会一直跑下去直到资源耗尽。LangGraph 为我们提供了内建的最大步数限制recursion_limit但更优雅的做法是在循环体内自己控制何时跳出就像在 while 循环里写break。2.1 一个极简的 ReAct 循环我们用伪代码模拟一个“智能体”做决策每次运行“思考”节点如果它觉得可以结束了就返回一条不包含工具调用的消息否则继续调用工具然后再次思考。classAgentState(TypedDict):messages:Annotated[list,add_messages]defthink(state:AgentState)-dict:# 模拟 LLM 的思考如果消息数量已经较多就决定停止msg_countlen(state[messages])ifmsg_count6:# 返回一个要求调用工具的消息模拟return{messages:[AIMessage(content我需要更多信息调用 search 工具。)]}else:return{messages:[AIMessage(content我已收集足够信息最终答案……)]}defshould_continue(state:AgentState)-Literal[action,end]:last_messagestate[messages][-1]# 如果最后一条消息是最终答案则结束if最终答案inlast_message.content:returnendelse:returnaction# 假设的工具节点deftake_action(state:AgentState)-dict:# 模拟执行工具并返回结果return{messages:[AIMessage(content搜索结果……)]}graphStateGraph(AgentState)graph.add_node(think,think)graph.add_node(action,take_action)graph.set_entry_point(think)graph.add_conditional_edges(think,should_continue,{action:action,end:END})graph.add_edge(action,think)# 行动后再次思考appgraph.compile()执行这个图它会按照think → action → think → action → ...的路径循环直到think节点产出一条包含“最终答案”的消息才走向END。这就实现了一个安全的、由自身状态决定的终止条件。在实际开发中你可以把think替换成真正的 LLM 调用让模型自己决定是调用工具还是直接回答终止条件则变为“消息列表最后一条不是工具调用请求”。这个模式我们会在第三篇博客中完整实现。3. 持久化与检查点为图装上“记忆”到目前为止我们的图每次执行都是无状态的跑完一次invoke后内部状态就消失了。但在真实场景中我们可能需要暂停执行几天后从同一点继续查看某次对话的完整历史轨迹回溯到之前的某个步骤重新选择一条分支这些能力都依赖于检查点Checkpointer。3.1 内置检查点LangGraph 内置了内存级检查点MemorySaver它把所有状态快照保存在内存字典里。使用时只需在compile时传入它并为每次执行提供一个唯一的thread_id。fromlanggraph.checkpoint.memoryimportMemorySaver checkpointerMemorySaver()appgraph.compile(checkpointercheckpointer)config{configurable:{thread_id:user-001}}initial_state{messages:[HumanMessage(content开始会话)]}# 第一次执行每个步骤自动保存状态resultapp.invoke(initial_state,configconfig)MemorySaver会在每一个步骤节点执行之后自动保存一份状态快照键是thread_id和自动生成的checkpoint_id。你可以随时查看某个会话的当前状态current_stateapp.get_state(config)print(current_state.values)# 当前状态字典print(current_state.next)# 下一个将要执行的节点空则表示已结束3.2 暂停与恢复因为每一步都有快照所以你可以随时“重启”一个已经结束的图——只需用相同的config再次调用invoke或stream图会从最后保存的状态开始继续执行。如果图由于某种原因抛出了异常状态也会保存在最后一个成功的步骤处你可以修正输入后用同一个config继续执行通常需要配合人工介入见下一节。3.3 时间旅行回溯历史检查点还允许你重新播放历史。通过get_state_history获取某个线程的所有快照然后取出某个旧快照的config包含checkpoint_id就可以从那个点重新执行选择另一条分支。historylist(app.get_state_history(config))old_statehistory[2]# 第三个快照old_configold_state.config# 用旧配置重新执行从那个时间点开始app.invoke(None,configold_config)这在调试和 A/B 测试中非常有用你可以回到“LLM 生成回复”之前给模型换一个提示词看看回复会怎么变。生产环境中你应当把MemorySaver换成持久化存储如SqliteSaver或AsyncPostgresSaver这样即使服务重启会话也不会丢失。4. 人机交互关键时刻让人来把关在很多业务场景里Agent 不能擅自行动退款申请、对外发送邮件、调用高风险 API……这些动作必须经过人工审核。LangGraph 提供了Human-in-the-loopHITL机制允许图在指定节点前/后暂停等待人类干预。4.1 中断节点的执行在compile时可以传入interrupt_before或interrupt_after参数指定哪些节点要被“打断”。appgraph.compile(checkpointercheckpointer,interrupt_before[take_action]# 在执行 take_action 之前暂停)此时当图运行到take_action节点前时会自动抛出GraphInterrupt异常实际上是暂停并返回当前状态你可以通过get_state获取。此时图处于“等待人工确认”的状态你可以查看当前状态决定是批准、修改还是拒绝。4.2 暂停后如何处理实践中我们通常使用stream或astream来逐步运行图因为它可以产生多个中间状态不会因为中断而抛出异常导致流程终结。当遇到中断时stream会停止产生新的事件你可以在外部代码中捕获当前状态展示给审批人。更直接的方法是用invoke它会抛出一个GraphInterrupt异常你捕获异常后可以获取状态。不过更推荐的是使用update_state和后续的继续执行流程。以下是典型的处理步骤用app.stream(initial_state, config)逐步运行直到某个节点被中断。中断发生后不再有新的流事件。此时调用app.get_state(config)查看当前状态和下一步。如果需要修改状态比如人工编辑了 Agent 生成的草稿调用app.update_state(config, values{messages: [HumanMessage(content审批人修改的回复)]})。然后可以调用app.stream(None, config)继续执行图会从中断的节点之前/之后继续。4.3 完整示例带人工审批的回复发送我们来构建一个真实场景用户提问 → Agent 生成回复草稿 → 人工审批 → 发送回复。fromlanggraph.checkpoint.memoryimportMemorySaverfromlanggraph.graphimportStateGraph,ENDfromtypingimportTypedDict,Annotatedfromlanggraph.graph.messageimportadd_messagesfromlangchain_core.messagesimportHumanMessage,AIMessageclassChatState(TypedDict):messages:Annotated[list,add_messages]draft:strapproved:bool# 节点LLM 生成回复defgenerate_draft(state:ChatState)-dict:# 实际中调用 LLM这里模拟user_msgstate[messages][-1].content draftf这是对 {user_msg} 的自动回复草稿。return{draft:draft,approved:False}# 节点发送回复仅当 approveddefsend_reply(state:ChatState)-dict:return{messages:[AIMessage(contentstate[draft])]}# 路由根据 approved 字段决定是否发送defreview_decision(state:ChatState):ifstate[approved]:returnsendelse:returnend# 未批准直接结束不发graphStateGraph(ChatState)graph.add_node(generate_draft,generate_draft)graph.add_node(send_reply,send_reply)graph.set_entry_point(generate_draft)graph.add_conditional_edges(generate_draft,review_decision,{send:send_reply,end:END})graph.add_edge(send_reply,END)# 在 generate_draft 之后中断让人工审核checkpointerMemorySaver()appgraph.compile(checkpointercheckpointer,interrupt_after[generate_draft]# 生成草稿后暂停)config{configurable:{thread_id:chat-1}}initial_state{messages:[HumanMessage(content请帮我退款)],draft:,approved:False}# 用 stream 运行遇到中断时会停止eventslist(app.stream(initial_state,config))print(流程暂停当前事件,events)stateapp.get_state(config)print(草稿内容,state.values[draft])这时审批人看到草稿觉得不满意想修改回复内容并批准。我们通过update_state把修改后的草稿和approvedTrue写回状态然后继续执行。# 人工审批修改草稿并批准app.update_state(config,values{draft:亲您的退款已处理请查收。,approved:True})# 继续执行从中断点之后开始即条件边判断阶段foreventinapp.stream(None,config):print(event)final_stateapp.get_state(config)print(最终发送的消息,final_state.values[messages][-1].content)输出将是人工修改后的那条消息。整个过程中Agent 绝不可能在审批通过前擅自发送回复。中断位置的选择很灵活interrupt_before在节点执行前暂停让你有机会阻止它运行interrupt_after在节点执行后暂停让你可以审阅它的输出并决定是否继续。你甚至可以在同一张图中组合使用多个中断点。5. 完整动手实验智能客服审批流现在我们把上面的知识点串联起来构建一个稍复杂的智能客服用户提问模型先进行意图分类普通问题 / 投诉如果是投诉必须生成处理草稿并送人工审批审批通过才能发送如果人工驳回则重新生成。如果是普通问题直接回复。整个对话过程被持久化保存支持暂停恢复。这里给出核心代码框架# 状态定义classCustomerState(TypedDict):messages:Annotated[list,add_messages]intent:strdraft:strapproved:bool# 节点意图分类defclassify(state:CustomerState)-dict:laststate[messages][-1].content intentcomplaintif投诉inlastelsegeneralreturn{intent:intent}# 节点生成回复普通defgeneral_reply(state:CustomerState)-dict:return{messages:[AIMessage(content这是普通问题的自动回复。)]}# 节点生成投诉草稿defdraft_complaint_reply(state:CustomerState)-dict:return{draft:【投诉草稿】我们将尽快处理您的问题。,approved:False}# 节点发送草稿defsend_reply(state:CustomerState)-dict:return{messages:[AIMessage(contentstate[draft])]}# 路由函数根据意图分流defroute_intent(state:CustomerState):ifstate[intent]complaint:returndraft_complaint_replyelse:returngeneral_reply# 审批判断检查 approved 字段defcheck_approval(state:CustomerState):ifstate[approved]:returnsend_replyelse:returnend# 构建图builderStateGraph(CustomerState)builder.add_node(classify,classify)builder.add_node(general_reply,general_reply)builder.add_node(draft_complaint_reply,draft_complaint_reply)builder.add_node(send_reply,send_reply)builder.set_entry_point(classify)builder.add_conditional_edges(classify,route_intent,{general_reply:general_reply,draft_complaint_reply:draft_complaint_reply})builder.add_edge(general_reply,END)builder.add_conditional_edges(draft_complaint_reply,check_approval,{send_reply:send_reply,end:END})builder.add_edge(send_reply,END)# 编译并开启持久化和中断checkpointerMemorySaver()appbuilder.compile(checkpointercheckpointer,interrupt_after[draft_complaint_reply]# 草稿生成后暂停等待审批)使用方式与上一节的审批示例完全一致只是这里增加了意图分流。这样我们就拥有了一套完整的、可干预、有记忆的智能客服引擎。6. 小结与下篇预告今天我们把 LangGraph 的控制流和持久化能力挖掘了一遍条件边让图能够根据状态做出动态决策实现意图路由、循环跳出等模式。循环是 Agent 思考-行动逻辑的骨架配合明确的终止条件可以安全运行。检查点为图提供了事务性的状态保存支撑暂停、恢复、时间旅行等能力。人机交互通过中断机制和update_state让人安全地介入关键决策防止模型乱来。有了这些你的图已经是一个“活”的系统了。但它还缺少最后一块拼图如何跟真正的工具搜索、计算、API集成并且以流式方式把执行过程实时推送给前端下一篇博客我们将接上工具调用和流式输出构建一个完整的、生产可用的 LangGraph Agent并探讨部署与可观测性。