多 Agent 协同的拓扑路由与动态图编排基于 LangGraph 的条件分支与循环状态机在复杂的企业级多智能体Multi-Agent业务场景中如自动代码重构与审查、法律合规长文审计、智能投研报表生成智能体之间的协作关系早已超越了传统的“线性链条Linear Chain: Agent A $\rightarrow$ Agent B $\rightarrow$ Agent C”代码审查 Agent 发现代码存在安全缺陷需要将任务打回给编码 Agent 重新修改循环状态机路由 Agent 需要根据用户的意图分类动态分支跳转到不同的专业垂直子 Agent条件分支路由审核 Agent 与合规 Agent 需要并发并行执行并在所有子任务全部完成后汇聚结果扇出与扇入。传统的线性编排框架在面对包含“循环Loops、分支Branches与动态状态持久化State Checkpointing”的复杂工作流时显得捉襟见肘。源自图论状态机思想的LangGraph基于有向图与状态持久化的多 Agent 编排框架成为了解决复杂长程智能体协同的工业级利器。今天我们深入拆解 LangGraph 的StateGraph状态图、Conditional Edges条件边与 Checkpointer断点持久化核心运行机制并给出完整的代码审查与自愈工作流实战代码。一、代码自动审查与重构的多 Agent 循环状态图架构flowchart TD Start([开始: 用户提交代码重构需求]) -- Coder[1. Coder Agent: 编写/重构代码] Coder -- Reviewer[2. Reviewer Agent: 静态扫描与安全审查] Reviewer -- ConditionCheck{审查是否通过?} ConditionCheck -- 未通过 (存在缺陷且循环 3次) -- FixRoute[3. 提取审查意见, 动态循环打回给 Coder Agent!] FixRoute -- Coder ConditionCheck -- 严重超标 / 循环耗尽 -- HumanEscalation[4. 触发 Human-in-the-Loop 人工复核断点] ConditionCheck -- 审查完全通过 -- FinalApproval[5. 最终合并代码, 交付成果!] HumanEscalation -- FinalApproval二、生产级 LangGraph 多 Agent 循环状态机代码实现import operator from typing import Annotated, TypedDict, List, Literal from pydantic import BaseModel, Field from langgraph.graph import StateGraph, END # 1. 定义全局强类型共享状态 (State) class AgentTeamState(TypedDict): task_description: str current_code: str review_comments: List[str] iteration_count: int is_approved: bool # 2. 节点 1: 编码与重构 Agent def coder_agent_node(state: AgentTeamState) - dict: iteration state.get(iteration_count, 0) 1 print(f[*] [Coder Agent] 正在编写/修复代码 (第 {iteration} 轮迭代)...) # 模拟根据历史 review_comments 修复代码 if state.get(review_comments): new_code # 已根据安全意见修复 SQL 注入与未释放连接 Bug\nfunc QueryUser() { ... } else: new_code # 初始重构版本\nfunc QueryUser() { rawSQL : ... } return { current_code: new_code, iteration_count: iteration } # 3. 节点 2: 审查与质检 Reviewer Agent def reviewer_agent_node(state: AgentTeamState) - dict: print(f[*] [Reviewer Agent] 正在审查代码质量与安全漏洞...) code state[current_code] iteration state[iteration_count] # 模拟第一轮审查拦截第二轮放行 if iteration 2 and rawSQL in code: comments [发现高危 SQL 字符串拼接风险必须强制改用参数化预编译查询] return {review_comments: comments, is_approved: False} else: return {review_comments: [], is_approved: True} # 4. 节点 3: 动态条件路由边 (Conditional Routing Edge) def decide_next_step(state: AgentTeamState) - Literal[coder_node, human_escalation, __end__]: # 规则 1: 审查通过直接结束交付 if state[is_approved]: print([✓] [Router] 代码审查完全通过流程正常结束) return END # 规则 2: 超过 3 次迭代依然未通过转人工介入 if state[iteration_count] 3: print([!] [Router] 超过最大迭代上限触发人工审核升级) return human_escalation # 规则 3: 打回给 Coder Agent 继续修复 print([↺] [Router] 发现安全缺陷打回 Coder Agent 循环自愈) return coder_node def human_escalation_node(state: AgentTeamState) - dict: print([⚠️ Human-in-the-Loop] 已将任务挂起并推发送至架构师工单系统等待人工审批...) return {}三、构建与编译 LangGraph 状态图def build_multi_agent_workflow(): # 1. 初始化状态图 workflow StateGraph(AgentTeamState) # 2. 添加处理节点 workflow.add_node(coder_node, coder_agent_node) workflow.add_node(reviewer_node, reviewer_agent_node) workflow.add_node(human_escalation, human_escalation_node) # 3. 设定图入口与固定边 workflow.set_entry_point(coder_node) workflow.add_edge(coder_node, reviewer_node) # 4. 核心添加动态条件分支路由边 (Conditional Edge) workflow.add_conditional_edges( reviewer_node, decide_next_step, { coder_node: coder_node, human_escalation: human_escalation, END: END } ) workflow.add_edge(human_escalation, END) # 5. 编译为可执行 App (支持内存/数据库 Checkpointer) app workflow.compile() return app四、真实运行轨迹测试def test_langgraph_execution(): app build_multi_agent_workflow() initial_input { task_description: 将老旧的 MySQL 查询函数重构为安全高并发版本, current_code: , review_comments: [], iteration_count: 0, is_approved: False } print( * 60) for output in app.stream(initial_input): for node_name, state_update in output.items(): print(f-- 完成节点 [{node_name}] 执行: {state_update}) print( * 60)终端执行流日志-- 完成节点 [coder_node]: 产出初版代码iteration1-- 完成节点 [reviewer_node]: 捕获 SQL 风险is_approvedFalse-- 触发条件边 [decide_next_step]: 动态路由打回coder_node-- 完成节点 [coder_node]: 修复代码iteration2-- 完成节点 [reviewer_node]: 审查通过is_approvedTrue-- 流程优雅终止五、生产治理三大黄金法则设置硬状态迭代步数上限Max Steps Limit在条件边中强制限制最大循环次数如 3 次杜绝 Agent 之间相互推诿陷入无限死循环结合 SQLite / PostgreSQL Checkpointer将图状态每一步持久化到数据库中服务突发重启时可从最后一个 Node 完美恢复断点保持全局 State 结构精炼全局 State 中只存放核心业务决策变量与快照避免把几十 MB 的冗余大文本塞进全局状态引发序列化性能下降。基于 LangGraph 的图状态机与动态条件路由复杂的多智能体系统才能真正具备严密自洽的逻辑控制力与企业级生产可用性。