1. 项目概述从状态机到智能体的控制流演进如果你已经跟着这个系列走到了第八篇那么恭喜你你已经不再是LangGraph的门外汉了。前面的文章我们搭建了基础的图结构定义了节点和边让消息能够流动起来。但一个只会按固定路线走流程的“图”充其量是个高级版的流程图工具离我们想要的“智能体”还差得远。真正的智能体其核心魅力在于“决策”——它能根据当前的情况动态地选择下一步该做什么。这正是“控制流”要解决的问题。在LangGraph的语境里控制流Control Flow特指决定图执行路径的逻辑。它超越了简单的“if-else”而是将决策逻辑本身也作为图的一部分实现了执行逻辑的可视化、可编排和可复用。这就像给你的智能体装上了“大脑皮层”让它不仅能处理任务还能思考“接下来处理哪个任务更合适”。无论是处理一个复杂、多步骤的用户查询还是构建一个能自主调用工具、反思结果的多智能体系统控制流都是实现这些高级能力的基石。今天我们就来彻底拆解LangGraph中实现控制流的几种核心武器条件边、入口点、中断与继续以及如何将它们组合起来构建真正有“想法”的智能应用。2. 控制流核心机制深度解析2.1 条件边让图学会“分岔路”思考条件边是LangGraph中最基础也最强大的控制流工具。它允许一条边的指向不是固定的下一个节点而是由一个函数动态决定。这个函数我们称之为“路由器”。路由器函数的设计哲学一个合格的路由器函数接收的是整个图的当前状态返回的是一个或多个下一个节点的名称。它的核心职责是“基于状态做路由决策”。这里的关键在于你的决策逻辑必须完全依赖于状态对象中的信息。例如状态里有一个messages列表路由器就可以检查最后一条消息的内容或类型状态里有一个step_count计数器路由器就可以判断是否超过了最大步数。一个实战中的高级技巧路由器函数应该保持“纯净”和“轻量”。它不应该去调用大语言模型LLM或者执行耗时的IO操作。为什么因为路由决策可能在一个对话中被执行非常多次如果每次路由都去问一次LLM延迟和成本都会激增。正确的做法是将需要复杂判断的逻辑前置到某个专门的“决策节点”中这个节点调用LLL将决策结果例如next_step: research写入状态然后路由器函数仅仅读取这个结果并返回对应的节点名。这样昂贵的LLM调用次数就被严格控制了。from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated, Literal from langgraph.graph.message import add_messages import operator class State(TypedDict): messages: Annotated[list, add_messages] next_step: str # 专门用于存储决策结果的字段 def decide_next_step(state: State): 决策节点调用LLM分析情况决定下一步 # 这里模拟一个LLM调用实际中可能是chat_model.invoke(...) last_msg state[messages][-1] if 价格 in last_msg.content: return {next_step: query_price} elif 故障 in last_msg.content: return {next_step: handle_error} else: return {next_step: general_chat} def route_after_decision(state: State) - Literal[query_price, handle_error, general_chat, __end__]: 路由器函数仅读取决策结果不做复杂判断 next_step state.get(next_step) if next_step query_price: return query_price elif next_step handle_error: return handle_error elif next_step general_chat: return general_chat else: # 如果没有决策结果或决策结束可以导向结束 return END # 构建图 builder StateGraph(State) builder.add_node(decide, decide_next_step) builder.add_node(query_price, lambda s: {messages: [(assistant, 正在查询价格...)]}) builder.add_node(handle_error, lambda s: {messages: [(assistant, 开始排查故障...)]}) builder.add_node(general_chat, lambda s: {messages: [(assistant, 我来帮您解答。)]}) # 设置入口和条件边 builder.set_entry_point(decide) # 关键从decide节点出发根据路由函数的结果动态指向下一个节点 builder.add_conditional_edges( decide, route_after_decision, { query_price: query_price, handle_error: handle_error, general_chat: general_chat, } ) # 其他节点执行完后可以回到决策节点或结束 builder.add_edge(query_price, decide) builder.add_edge(handle_error, decide) builder.add_edge(general_chat, END) graph builder.compile()这个模式将“思考”和“路由”分离是构建复杂、高效智能体的黄金法则。2.2 入口点与子图构建模块化与层次化智能体当你的智能体逻辑变得复杂时把所有节点平铺在一张图上会变得难以管理和理解。LangGraph通过入口点和子图的概念支持你将功能模块化。入口点的战略意义set_entry_point方法不仅指定了图开始运行的第一个节点更重要的是它定义了图的“默认启动模式”。你可以有多个不同的入口点通过编译不同的图实例或者通过一个顶层的“路由图”来调用不同的子图实现功能的按需加载。例如你可以有一个CustomerServiceGraph它内部通过条件边调用RefundSubGraph处理退款或TechSupportSubGraph处理技术支持这两个子图。子图作为可复用组件子图本身就是一个编译好的StateGraph对象。你可以像使用普通节点一样使用add_node将它添加到父图中。当执行流进入这个“节点”时实际上是在运行一个完整的子流程。子图与父图通过状态共享进行通信。这意味着你在父图状态中定义的字段在子图中可以直接读取和修改。注意子图与父图的状态结构State Schema必须兼容。通常的做法是定义一个基础的状态类型BaseState所有图和子图都基于这个类型进行扩展。如果子图需要父图中不存在的字段你需要在父图的状态定义中也进行声明即使父图本身可能用不到它否则在运行时会出现状态验证错误。# 假设我们已定义基础状态类 State from langgraph.graph import StateGraph # 1. 定义一个处理订单的子图 order_builder StateGraph(State) order_builder.add_node(validate_order, validate_order_func) order_builder.add_node(check_inventory, check_inventory_func) order_builder.set_entry_point(validate_order) order_builder.add_edge(validate_order, check_inventory) order_builder.add_edge(check_inventory, END) order_subgraph order_builder.compile() # 2. 定义一个处理咨询的子图 inquiry_builder StateGraph(State) inquiry_builder.add_node(classify_inquiry, classify_inquiry_func) inquiry_builder.add_node(generate_response, generate_response_func) inquiry_builder.set_entry_point(classify_inquiry) inquiry_builder.add_edge(classify_inquiry, generate_response) inquiry_builder.add_edge(generate_response, END) inquiry_subgraph inquiry_builder.compile() # 3. 构建主图将子图作为节点加入 main_builder StateGraph(State) main_builder.add_node(router, router_node) # 路由节点决定调用哪个子图 main_builder.add_node(handle_order, order_subgraph) # 子图作为节点 main_builder.add_node(handle_inquiry, inquiry_subgraph) # 子图作为节点 main_builder.add_node(final_summary, final_summary_func) main_builder.set_entry_point(router) # 路由器根据状态决定下一步是进入订单子图还是咨询子图 main_builder.add_conditional_edges( router, lambda s: handle_order if s[intent] order else handle_inquiry ) # 子图执行完毕后都汇聚到总结节点 main_builder.add_edge(handle_order, final_summary) main_builder.add_edge(handle_inquiry, final_summary) main_builder.add_edge(final_summary, END) main_graph main_builder.compile()这种架构让智能体的能力像乐高积木一样可以拼接和扩展极大地提升了代码的可维护性和复用性。2.3 中断、继续与人工干预实现异步与协作式工作流这是LangGraph从“自动化流程”迈向“协作式智能体”的关键一步。interrupt机制允许你在图的特定节点暂停执行将控制权交还给调用者通常是你的应用程序等待外部输入比如用户确认、人工审核、等待另一个系统响应后再继续执行。核心概念检查点LangGraph的运行是建立在状态演进之上的。每次从一个节点到另一个节点状态都可能被更新。interrupt实际上是在某个节点执行后主动地、有意图地停止状态演进并将当前完整的状态“快照”保存下来。后续的resume操作则是从这个保存的快照状态开始继续执行。如何实现一个可中断的节点你不需要在节点函数里做特别的操作。中断是通过在图的边上配置来实现的。你可以在add_edge或add_conditional_edges时指定一个interrupt_before参数。当执行流到达这条边时图会在实际沿着这条边前往下一个节点之前暂停。from langgraph.graph import StateGraph, END from langgraph.checkpoint import MemorySaver class State(TypedDict): messages: Annotated[list, add_messages] approved: bool def human_review_node(state: State): 模拟需要人工审核的节点 return {messages: [(assistant, 申请已提交等待主管审核。)]} def after_approval_node(state: State): 审核通过后的节点 return {messages: [(assistant, 申请已获批流程继续。)]} def after_rejection_node(state: State): 审核拒绝后的节点 return {messages: [(assistant, 申请被拒绝流程终止。)]} # 使用MemorySaver来保存检查点状态快照 memory MemorySaver() builder StateGraph(State, checkpointermemory) builder.add_node(human_review, human_review_node) builder.add_node(after_approval, after_approval_node) builder.add_node(after_rejection, after_rejection_node) builder.set_entry_point(human_review) # 关键从审核节点出来的边配置为中断。 # 这意味着执行完human_review_node后图会暂停等待外部驱动。 builder.add_conditional_edges( human_review, # 路由器函数决定去哪但实际跳转前会中断 lambda s: after_approval if s.get(approved) else after_rejection, # 配置中断 interrupt_beforeTrue ) builder.add_edge(after_approval, END) builder.add_edge(after_rejection, END) graph builder.compile() # 模拟执行流程 config {configurable: {thread_id: thread-1}} # 第一次执行进入human_review节点后中断 initial_result graph.invoke({messages: [(user, 我要申请特批)]}, config) print(initial_result[messages][-1]) # 输出: (assistant, 申请已提交等待主管审核。) # 此时图已暂停。在外部如Web后台人工将 state[approved] 设置为 True。 # 准备继续执行所需的状态更新 updated_state {approved: True} # 继续执行图将从中断点即决定走哪条条件边的时刻恢复 # 并根据当前最新的状态包含approvedTrue决定路由。 next_result graph.invoke(updated_state, config) print(next_result[messages][-1]) # 输出: (assistant, 申请已获批流程继续。)中断的应用场景人工审批环节如上面的例子在关键操作前暂停等待管理员在后台点击“通过”或“拒绝”。等待用户输入在多轮对话智能体中智能体生成一个选项列表后中断等待用户选择。等待外部API回调触发一个长时间运行的任务如生成报告中断并设置一个Webhook当外部API完成回调时再携带结果继续执行。多智能体协作智能体A将任务分派给智能体B后中断等待B完成任务并返回结果。这个机制打破了图的封闭执行使其能够与真实世界的人类和系统进行交互极大地扩展了应用边界。3. 复杂控制流模式实战构建一个任务规划与执行智能体现在让我们综合运用以上所有概念构建一个更贴近真实场景的智能体一个能够分解复杂任务、逐步执行、并在遇到困难时请求帮助的“任务执行智能体”。3.1 智能体状态设计与整体架构这个智能体的状态需要包含任务规划、执行步骤、历史记录以及当前上下文。from typing import TypedDict, Annotated, List, Optional from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages import operator class AgentState(TypedDict): 任务执行智能体的核心状态 # 对话消息流 messages: Annotated[List, add_messages] # 用户的原始复杂任务 original_task: str # 规划好的子任务列表 sub_tasks: List[str] # 当前正在执行或已完成的子任务索引 current_task_index: int # 子任务执行结果的累积 task_results: Annotated[List[str], operator.add] # 标记是否需要人工帮助 need_human_help: bool # 帮助请求的具体内容 help_request: Optional[str]我们的图将包含以下几个关键节点任务规划节点分析原始任务将其拆解为顺序或并行的子任务列表。任务路由节点决定是执行下一个子任务还是请求帮助或是结束。任务执行节点执行具体的子任务这里可能包含工具调用。帮助处理节点处理“需要帮助”的状态可以连接到一个子图或发送通知。结果汇总节点所有子任务完成后汇总最终结果。3.2 节点函数实现与条件边编排from langchain_core.messages import HumanMessage, AIMessage import random def plan_task(state: AgentState): 节点1规划任务。模拟LLM进行任务分解。 task state[original_task] # 模拟LLM的规划能力例如“帮我订下周末去上海的机票和酒店并查好外滩的天气” # 可能被分解为 # 1. 查询上海本周末的机票信息。 # 2. 查询上海本周末的酒店信息。 # 3. 查询上海本周末的天气预报。 simulated_subtasks [ 查询上海本周末的机票信息, 查询上海本周末的酒店信息, 查询上海本周末的天气预报 ] return { sub_tasks: simulated_subtasks, current_task_index: 0, messages: [AIMessage(contentf我已将您的任务分解为{len(simulated_subtasks)}个子步骤。)] } def execute_task(state: AgentState): 节点2执行当前子任务。模拟调用工具或API。 idx state[current_task_index] task state[sub_tasks][idx] # 模拟执行有一定概率“遇到困难” success random.random() 0.3 # 70%成功率 if success: result f子任务{idx1}【{task}】执行成功。模拟结果找到一些选项。 update { task_results: [result], # operator.add会将其追加到列表 messages: [AIMessage(contentf步骤{idx1}完成{task})], need_human_help: False, help_request: None } else: help_msg f在执行【{task}】时遇到模拟障碍如无直达航班、酒店满房。 update { need_human_help: True, help_request: help_msg, messages: [AIMessage(contentf步骤{idx1}遇到问题需要您的决策。)] } return update def handle_human_help(state: AgentState): 节点3处理人工帮助请求。这里可以连接到一个发送通知的子图。 # 在实际应用中这里可能会 # 1. 发送邮件/短信通知 # 2. 将任务挂起到一个后台管理队列 # 3. 调用一个向用户发送澄清消息的对话节点 # 本例中我们模拟人工介入后提供了解决方案 help_resolution 已根据人工决策选择中转航班/更换酒店区域。 return { messages: [AIMessage(contentf收到人工帮助{help_resolution})], need_human_help: False, help_request: None, # 假设人工帮助后当前任务被视为完成结果记录为人工提供的方案 task_results: [f子任务{state[current_task_index]1}通过人工协助解决{help_resolution}] } def summarize_results(state: AgentState): 节点4汇总所有子任务结果形成最终答案。 all_results \n.join(state[task_results]) final_answer f任务【{state[original_task]}】已完成汇总\n{all_results}\n报告完毕。 return { messages: [AIMessage(contentfinal_answer)] } def route_after_plan(state: AgentState) - str: 路由器A规划完成后总是进入执行节点。 return execute_task def route_after_execution(state: AgentState) - str: 路由器B执行一个子任务后决定下一步。 # 情况1需要人工帮助 if state.get(need_human_help): return handle_human_help # 情况2所有子任务都执行完了 if state[current_task_index] len(state[sub_tasks]) - 1: return summarize_results # 情况3还有下一个子任务 else: # 更新索引准备执行下一个 return prepare_next_task def prepare_next_task(state: AgentState): 节点5准备执行下一个任务递增索引。 next_idx state[current_task_index] 1 return { current_task_index: next_idx, messages: [AIMessage(contentf准备开始步骤{next_idx1}...)] } def route_after_help(state: AgentState) - str: 路由器C处理完人工帮助后决定下一步。 # 帮助处理后回到路由判断是继续执行下一个任务还是结束 return route_after_execution # 复用执行后的路由逻辑 # 开始构建图 builder StateGraph(AgentState) # 添加所有节点 builder.add_node(plan_task, plan_task) builder.add_node(execute_task, execute_task) builder.add_node(handle_human_help, handle_human_help) builder.add_node(summarize_results, summarize_results) builder.add_node(prepare_next_task, prepare_next_task) # 注意路由函数本身不是节点它用于指导边的方向。 # 设置入口点 builder.set_entry_point(plan_task) # 添加边和条件边 # 1. 规划后 - 执行 builder.add_conditional_edges( plan_task, route_after_plan, {execute_task: execute_task} ) # 2. 执行后 - 复杂路由可能去帮助节点、汇总节点或准备下一个任务 builder.add_conditional_edges( execute_task, route_after_execution, { handle_human_help: handle_human_help, summarize_results: summarize_results, prepare_next_task: prepare_next_task, } ) # 3. 准备下一个任务后 - 回到执行节点循环 builder.add_edge(prepare_next_task, execute_task) # 4. 处理帮助后 - 复用执行后的路由逻辑这是一个技巧将路由逻辑节点化 # 我们需要一个“虚拟”节点来触发路由判断 builder.add_node(route_after_help, lambda s: s) # 此节点不改变状态仅作为路由跳板 builder.add_edge(handle_human_help, route_after_help) builder.add_conditional_edges( route_after_help, route_after_help, # 这个函数返回的是节点名我们需要映射 { route_after_execution: route_after_execution } ) # 再添加一个名为“route_after_execution”的虚拟节点并为其配置最终的条件边 builder.add_node(route_after_execution, lambda s: s) builder.add_conditional_edges( route_after_execution, route_after_execution, # 再次调用同一个路由函数但此时状态已更新帮助已处理 { handle_human_help: handle_human_help, summarize_results: summarize_results, prepare_next_task: prepare_next_task, } ) # 5. 汇总后 - 结束 builder.add_edge(summarize_results, END) # 编译图 task_agent_graph builder.compile()3.3 执行模拟与流程分析让我们模拟这个智能体的运行观察控制流是如何工作的。# 模拟一个复杂任务 initial_state { messages: [HumanMessage(content帮我订下周末去上海的机票和酒店并查好外滩的天气)], original_task: 帮我订下周末去上海的机票和酒店并查好外滩的天气, sub_tasks: [], current_task_index: 0, task_results: [], need_human_help: False, help_request: None } # 执行图 final_state task_agent_graph.invoke(initial_state) # 打印执行过程中的消息 for msg in final_state[messages]: print(f{msg.type}: {msg.content})一次可能的执行轨迹如下开始-plan_task: 收到任务分解为3个子任务。plan_task-execute_task(路由A): 开始执行第一个子任务“查询机票”。execute_task-prepare_next_task(路由B假设成功): 任务1成功准备下一个。prepare_next_task-execute_task: 开始执行第二个子任务“查询酒店”。execute_task-handle_human_help(路由B假设失败): 任务2失败触发帮助请求。handle_human_help-route_after_help-route_after_execution-prepare_next_task(路由CB): 人工帮助处理完毕系统决定继续下一个任务递增索引。prepare_next_task-execute_task: 开始执行第三个子任务“查询天气”。execute_task-summarize_results(路由B成功且是最后一个任务): 任务3成功进入汇总。summarize_results-END: 流程结束输出最终报告。这个流程清晰地展示了条件边如何实现分支成功/失败、循环执行多个子任务以及中断与恢复的模拟handle_human_help节点模拟了等待外部输入的过程。通过虚拟的路由节点我们实现了路由逻辑的复用使得图结构更加清晰。4. 高级模式、调试与性能优化4.1 循环与递归的精细化控制在上面的例子中我们通过current_task_index和条件边实现了一个for循环。但有时我们需要更复杂的循环控制比如while循环直到满足某个条件或递归调用自身。实现While循环关键在于设计一个“循环条件检查节点”和一个“循环体节点”。检查节点根据状态决定是进入循环体还是跳出循环。def check_condition(state): 检查节点判断是否继续循环 if state[accuracy] 0.95 and state[iteration] 10: return {should_continue: True} else: return {should_continue: False} def loop_body(state): 循环体节点执行一次迭代 # ... 执行一些操作更新state[accuracy] ... new_accuracy state[accuracy] 0.1 new_iteration state[iteration] 1 return {accuracy: new_accuracy, iteration: new_iteration} builder.add_node(check, check_condition) builder.add_node(body, loop_body) builder.set_entry_point(check) builder.add_conditional_edges( check, lambda s: body if s[should_continue] else END ) builder.add_edge(body, check) # 执行完一次循环体后回到检查点递归调用子图调用自身LangGraph支持子图理论上子图可以调用自身但需要谨慎设计递归出口避免无限递归。通常更好的模式是将递归逻辑扁平化为循环。4.2 状态管理与检查点策略对于长时间运行或需要持久化的智能体检查点的管理至关重要。MemorySaver适用于开发和简单场景生产环境则需要更强大的后端如RedisSaver或PostgresSaver。配置检查点在编译图时传入checkpointer参数。每个执行线程thread_id的状态都会被独立保存。from langgraph.checkpoint import RedisSaver import os redis_url os.getenv(REDIS_URL) checkpointer RedisSaver.from_conn_string(redis_url) builder StateGraph(State, checkpointercheckpointer) # ... 构建图 ... graph builder.compile()中断与继续的配置interrupt_before可以配置在add_edge和add_conditional_edges上。你还可以通过config配置在特定节点后强制创建检查点即使没有中断这对于故障恢复非常有用。4.3 可视化、调试与监控可视化图结构使用graph.get_graph().draw_mermaid()可以输出Mermaid代码粘贴到支持Mermaid的编辑器如GitHub Markdown、Obsidian中查看图形化的工作流。这对于复杂流程的沟通和审查不可或缺。调试技巧打印状态在关键节点的函数开头或结尾添加print(f”Node X, State: {state}”)这是最直接的调试方式。使用stream模式graph.stream(state)会返回一个迭代器每执行一个节点就yield一次结果你可以实时看到执行流和状态变化。检查点检查通过checkpointer.get_tuple(config)可以读取任意thread_id在任意时刻保存的状态快照用于复盘执行过程。性能监控关注节点的执行时间和调用次数。对于频繁执行的路由器函数确保其轻量。对于耗时的LLM或工具调用节点考虑异步执行或设置超时。4.4 常见陷阱与最佳实践状态污染多个节点可能修改状态的同一部分。使用Annotated和operator.add这样的归约器reducer来安全地更新列表或字典。对于简单赋值直接覆盖即可。路由函数副作用牢记路由器函数应无副作用只读状态。将决策逻辑放入前驱节点。循环检测LangGraph有基本的循环检测但复杂的条件边可能导致非预期的无限循环。务必在测试中覆盖各种分支情况并考虑设置最大步数限制。子图状态兼容性父图和子图的状态结构必须对齐。建议使用TypedDict继承或组合来定义统一的基础状态。错误处理图中节点可能抛出异常。目前LangGraph原生错误处理能力较弱需要在节点函数内部进行try-catch或将错误信息写入状态由后续节点处理。配置化将节点中可变的参数如模型名称、API密钥、最大循环次数通过config传入而不是硬编码在函数内部提高图的复用性。控制流是LangGraph的灵魂它将静态的图结构赋予了动态的生命力。从简单的分支判断到包含人工干预的复杂协作流程掌握条件边、入口点、中断和子图你就能设计出适应各种现实业务场景的智能体。记住设计图就是设计智能体的思维过程清晰的图结构往往对应着清晰、可维护的业务逻辑。多画图多思考状态流转你的LangGraph技能就会从“会用”升华到“精通”。