LangGraph智能体开发进阶：条件分支与循环图构建指南

在AI Agent智能体开发中，条件分支与循环结构是构建复杂业务逻辑的核心能力。无论是处理多轮对话的上下文依赖，还是实现动态决策的分支路径，亦或是处理需要迭代优化的任务（如文档摘要、代码生成），都离不开对状态流转的精准控制。本文将基于LangGraph框架，深入探讨如何通过条件分支图与循环图实现智能体的灵活行为控制。

一、条件分支图：动态决策的核心

1.1 条件分支的本质与适用场景

条件分支的核心是通过状态判断动态决定后续执行路径，其典型场景包括：

• 多轮对话管理：根据用户输入内容（如意图识别结果）切换不同对话分支；
• 任务路由：根据任务类型（如文本生成、数据分析）分配至不同处理流程；
• 异常处理：当检测到错误时（如API调用失败）进入补偿逻辑。

在LangGraph中，条件分支通过ConditionNode或自定义判断逻辑实现，其核心是状态转移条件的定义。例如，在医疗诊断智能体中，可根据症状输入的严重程度（轻度/中度/重度）跳转至不同处理节点。

1.2 条件分支的实现方式

方式1：基于状态属性的显式判断

from langgraph.predefined import State
from langgraph.graph import Graph
state = State()
state.set("symptom_severity", "moderate")  # 假设症状严重程度为"moderate"
graph = Graph()
graph.add_node("start", lambda: print("开始诊断"))
graph.add_node("light_treatment", lambda: print("建议休息"))
graph.add_node("moderate_treatment", lambda: print("建议就医"))
graph.add_node("severe_treatment", lambda: print("立即急诊"))
# 定义条件分支
def condition(state):
    severity = state.get("symptom_severity")
    if severity == "light":
        return "light_treatment"
    elif severity == "moderate":
        return "moderate_treatment"
    else:
        return "severe_treatment"
graph.add_edge("start", condition, to="light_treatment|moderate_treatment|severe_treatment")

方式2：基于工具调用的隐式判断

通过工具（如LLM）的输出结果动态决定分支：

from langgraph.tools import Tool
class DiagnosisTool(Tool):
    def __call__(self, symptoms):
        # 模拟LLM判断症状严重程度
        if "高烧" in symptoms:
            return {"severity": "severe"}
        elif "咳嗽" in symptoms:
            return {"severity": "moderate"}
        else:
            return {"severity": "light"}
graph.add_node("diagnose", DiagnosisTool())
graph.add_edge("start", "diagnose", to="condition_node")
def condition_node(state):
    severity = state["diagnose_output"]["severity"]
    # 根据severity决定后续节点
    ...

1.3 最佳实践与注意事项

• 状态属性命名规范：使用工具名_output（如diagnose_output）避免命名冲突。
• 分支覆盖完整性：确保所有可能条件均有对应处理路径，避免遗漏else分支。
• 可读性优化：对复杂条件逻辑封装为独立函数，例如：

  def get_treatment_node(state):
      severity = state["diagnose_output"]["severity"]
      mapping = {"light": "light_treatment", "moderate": "moderate_treatment", "severe": "severe_treatment"}
      return mapping.get(severity, "default_treatment")  # 默认处理

二、循环图：迭代优化的利器

2.1 循环结构的典型应用场景

循环结构适用于需要重复执行直至满足条件的任务，例如：

• 文档摘要优化：循环生成摘要并对比与原文的相似度，直至达到阈值；
• 代码调试：循环执行代码片段并检查输出，直至通过测试用例；
• 多轮问答：循环追问用户以澄清模糊问题。

2.2 循环图的实现方法

方法1：固定次数循环

通过LoopNode实现预设次数的循环：

from langgraph.graph import LoopNode
graph.add_node("generate_summary", lambda state: ...)  # 生成摘要
graph.add_node("check_quality", lambda state: ...)     # 检查质量
loop = LoopNode(
    max_iterations=5,
    body=[("generate_summary", "check_quality")],
    next_node="finish"
)
graph.add_node("loop_start", loop)

方法2：条件终止循环

通过状态判断动态终止循环：

def should_continue(state):
    return state.get("summary_quality", 0) < 0.9  # 质量未达标时继续
graph.add_edge("check_quality", should_continue, to="generate_summary|finish")

2.3 循环中的状态管理

循环过程中需特别注意状态累积与终止条件：

• 状态累积：在每次迭代中更新状态（如累计迭代次数、最佳结果）：

  def generate_summary(state):
      new_summary = ...  # 生成新摘要
      if "best_summary" not in state or similarity(new_summary, state["best_summary"]) > 0.9:
          state["best_summary"] = new_summary
      state["iteration"] += 1
      return state

• 终止条件优化：避免死循环，建议设置最大迭代次数与质量阈值双保险：

  def should_terminate(state):
      return state["iteration"] >= 10 or state.get("summary_quality", 0) >= 0.9

三、综合案例：医疗诊断智能体

以下是一个结合条件分支与循环的完整案例：

from langgraph.predefined import State
from langgraph.graph import Graph
class DiagnosisTool(Tool):
    def __call__(self, symptoms):
        # 模拟LLM诊断
        if "高烧" in symptoms and "咳嗽" in symptoms:
            return {"disease": "flu", "severity": "severe"}
        elif "咳嗽" in symptoms:
            return {"disease": "cold", "severity": "moderate"}
        else:
            return {"disease": "healthy", "severity": "light"}
class TreatmentTool(Tool):
    def __call__(self, disease):
        treatments = {
            "flu": "抗病毒药物+休息",
            "cold": "感冒药+多喝水",
            "healthy": "无需治疗"
        }
        return treatments.get(disease, "未知疾病")
# 初始化状态
state = State()
state["iteration"] = 0
# 构建图
graph = Graph()
graph.add_node("start", lambda: print("请输入症状："))
graph.add_node("get_symptoms", lambda user_input: {"symptoms": user_input})
graph.add_node("diagnose", DiagnosisTool())
graph.add_node("treat", TreatmentTool())
# 条件分支：根据严重程度决定是否循环追问
def severity_condition(state):
    severity = state["diagnose_output"]["severity"]
    if severity == "severe":
        return "emergency_treatment"
    elif severity == "moderate" and state["iteration"] < 3:  # 中度症状且未达最大迭代次数
        state["iteration"] += 1
        return "refine_symptoms"  # 循环追问细节
    else:
        return "treat"
# 循环追问节点
def refine_symptoms(state):
    return input("请补充更多症状细节（如持续时间、伴随症状）：")
# 构建边
graph.add_edge("start", "get_symptoms", to="diagnose")
graph.add_edge("diagnose", severity_condition, to="emergency_treatment|refine_symptoms|treat")
graph.add_edge("refine_symptoms", "get_symptoms", to="diagnose")  # 循环回诊断
graph.add_edge("treat", lambda state: print(f"建议：{state['treat_output']}"))
graph.add_edge("emergency_treatment", lambda state: print("立即就医！"))

四、性能优化与调试技巧

1. 状态可视化：使用graph.render()生成流程图，快速定位分支/循环逻辑问题。
2. 日志记录：在关键节点添加状态日志，例如：

   def log_state(state):
       print(f"当前状态：{state}")
       return state
   graph.add_node("log", log_state)
   graph.add_edge("diagnose", "log", to="treat")

3. 工具调用缓存：对重复调用的工具（如LLM）启用缓存，避免重复计算。
4. 异常处理：为循环和分支添加超时机制，防止长时间运行。

五、总结与延伸思考

条件分支与循环图是构建复杂智能体的核心工具，其设计需兼顾灵活性与可控性。实际开发中，建议：

• 从简单场景入手，逐步增加分支/循环复杂度；
• 通过单元测试验证每个分支/循环路径的正确性；
• 结合监控工具（如Prometheus）观察状态流转效率。

未来，随着LangGraph对动态图支持的完善，开发者可探索更自适应的分支策略（如基于强化学习的路径选择），进一步释放AI Agent的潜力。