LangGraph 中的条件分支与循环编排实战在多智能体Agent工作流编排技术的发展历程中传统的线性执行链如早期的 LangChain Sequential Chain只能表达“步骤 A - 步骤 B - 步骤 C”这种僵化不可变的单向顺序逻辑。然而真实复杂的软件工程与商业决策必然包含大量的**“动态条件分支Conditional Branching”与“试错自纠错循环Cyclic Reflexion Loops”**如果大模型生成的代码通过了语法审查分支流向发布节点如果未通过分支必须循环跳回编码节点进行自我修复最多允许重试 3 轮如果识别出用户提问属于“只读查询”分支流向 RAG 检索节点如果属于“敏感退款”分支必须挂起并流向人工审批节点。基于有向图理论与状态机模型的LangGraph凭借其强大的StateGraph、add_conditional_edges条件边路由与原生循环图表达力成为了当今工业界构建复杂自适应 Agent 系统的首选编排底座。本文将手把手拆解如何基于 LangGraph 构建一个包含**“意图动态分支 编码审查自纠错循环 硬性退出熔断”的生产级状态图工作流**。一、复杂条件分支与循环编排状态图拓扑模型[ 节点 1: 需求解析与意图分类 (Router Node) ] │ ┌────────────────────────────────┴────────────────────────────────┐ ▼ (分支 A: 简单只读查询) ▼ (分支 B: 复杂编码任务) ┌───────────────────────┐ ┌─────────────────────────────────┐ │ 节点 2: RAG 知识检索 │ │ 节点 3: 程序员编码 (Coder Node) │ │ (直接输出结论并结束) │ └────────────────┬────────────────┘ └──────────┬────────────┘ │ │ ▼ │ ┌─────────────────────────────────┐ │ │ 节点 4: 代码质量审查 (Reviewer) │ │ └────────────────┬────────────────┘ │ │ │ ┌─────────────────────────┴─────────────────────────┐ │ ▼ (审查未通过 且 重试次数 3) ▼ (审查通过 或 达到最大重试) │ [ 触发循环回退边 (Loop Edge) ] ┌───────────────────────┐ │ 跳回【节点 3 重新修复代码】! │ 节点 5: 最终交付与归档│ │ └──────────┬────────────┘ └─────────────────────────────────────────────┬──────────────────────────────────────────┘ ▼ [ 终态: END ]二、生产级 LangGraph 强类型状态图完整实现实操from typing import TypedDict, Annotated, Literal from langgraph.graph import StateGraph, END # 1. 定义全局强类型图状态 Schema (State) class AgentWorkflowState(TypedDict): user_requirement: str task_category: str # SIMPLE_QUERY 或 CODE_GENERATION generated_code: str review_feedback: str is_code_approved: bool retry_count: int # 循环计数器防止无限死循环 # 2. 编写图节点业务逻辑函数 (Node Functions) def intent_router_node(state: AgentWorkflowState) - dict: 节点 1: 意图识别与任务分流 req state[user_requirement] if 写代码 in req or 实现 in req or def in req: return {task_category: CODE_GENERATION, retry_count: 0} return {task_category: SIMPLE_QUERY} def simple_rag_node(state: AgentWorkflowState) - dict: 节点 2: 处理简单查询 return {review_feedback: 已为您从知识库中检索到业务答案。} def coder_agent_node(state: AgentWorkflowState) - dict: 节点 3: 程序员生成或修复代码 current_retries state.get(retry_count, 0) print(f▶ [Coder 节点执行] 正在编写/修复代码 (当前重试轮次: {current_retries})...) # 模拟大模型根据上一轮的 review_feedback 修正代码 code f# 修复后的 Python 业务代码 (v{current_retries1})\ndef process_order():\n return SUCCESS return {generated_code: code, retry_count: current_retries 1} def reviewer_agent_node(state: AgentWorkflowState) - dict: 节点 4: 审查专家进行代码质检 current_retries state[retry_count] print(f▶ [Reviewer 节点审查] 正在审查代码质量...) # 模拟在第 2 轮时审查通过 if current_retries 2: return {is_code_approved: True, review_feedback: 代码完全符合安全与规范要求} else: return {is_code_approved: False, review_feedback: 发现缺少异常捕获逻辑请增加 try-except} def final_delivery_node(state: AgentWorkflowState) - dict: 节点 5: 最终交付 return {review_feedback: 任务已完成验收交付生产。} # 3. 编写条件边路由函数 (Conditional Edge Routers) def route_after_intent(state: AgentWorkflowState) - Literal[simple_rag, coder_agent]: 条件边 1: 根据意图分流到不同子链路 if state[task_category] CODE_GENERATION: return coder_agent return simple_rag def route_after_review(state: AgentWorkflowState) - Literal[coder_agent, final_delivery]: 条件边 2: 核心循环与退出判定 (包含最大 3 轮硬性刹车) if state[is_code_approved]: print(✅ [审查通过] 分支流向最终交付) return final_delivery if state[retry_count] 3: print( [达到最大重试上限] 强制跳出循环进入降级交付) return final_delivery print( [审查未通过] 触发循环条件边打回 Coder 重新修复) return coder_agent # 4. 组装与编译有向循环图 (StateGraph Compilation) builder StateGraph(AgentWorkflowState) # 注册所有节点 builder.add_node(intent_router, intent_router_node) builder.add_node(simple_rag, simple_rag_node) builder.add_node(coder_agent, coder_agent_node) builder.add_node(reviewer_agent, reviewer_agent_node) builder.add_node(final_delivery, final_delivery_node) # 设置入口 builder.set_entry_point(intent_router) # 添加条件边 1 builder.add_conditional_edges(intent_router, route_after_intent) builder.add_edge(simple_rag, END) # 添加线性边与循环条件边 2 builder.add_edge(coder_agent, reviewer_agent) builder.add_conditional_edges(reviewer_agent, route_after_review) builder.add_edge(final_delivery, END) # 编译为可执行工作流 compiled_graph builder.compile()三、运行与验证循环自愈执行# 测试复杂编码循环任务 initial_state: AgentWorkflowState { user_requirement: 请帮我写一个安全的订单处理 Python 函数, task_category: , generated_code: , review_feedback: , is_code_approved: False, retry_count: 0 } final_output compiled_graph.invoke(initial_state) print(f\n 【工作流执行终态】: {final_output[review_feedback]}) print(f生成的最终代码:\n{final_output[generated_code]})四、生产治理铁律在 LangGraph 编排循环图时必须死守两条红线任何循环图必须挂载递增计数器与最大硬上限Max Retries $\le 3$坚决杜绝因大模型持续审查不通过引发的无限循环与 Token 账单爆炸状态属性必须最小化内聚仅在 State 中保留控制流转所需的关键标志位与最终产物中间长思考过程在节点内部消化。给发散的大模型逻辑装上严密有向图的铁轨用条件边控制分支用循环边实现自愈LangGraph 是构建工业级高可控多智能体应用的最强基础设施。
