LangGraph Conditional Edge：条件分支与循环控制

在构建搜索型 Agent 时，一个常见的需求是：如果搜索结果不够充分，就换个关键词再搜一次；如果内容足够了，就直接生成答案。

"如果……就……否则……"——这是最朴素的条件逻辑。在 LangGraph 之前，要用 LLM Agent 实现这个逻辑，要么完全依赖 LLM 自己判断要不要继续搜索（不可靠），要么在 LangChain 外面手写一个 while 循环套着 Agent（逻辑混乱，而且 Agent 的上下文管理容易出问题）。

LangGraph 的 Conditional Edge 就是为这种场景设计的。

1.1 Conditional Edge 是什么

LangGraph 条件分支与循环控制 — 路由函数根据 State 值动态决定执行路径

普通的边是固定的：节点 A 执行完，一定跳到节点 B。

Conditional Edge 是动态的：节点 A 执行完，根据当前 State 的值，决定跳到节点 B 还是节点 C 还是直接结束。

决定跳到哪里的逻辑叫路由函数（Router Function），由开发者编写，LangGraph 来调用。

1.2 add_conditional_edges 的用法

python

graph.add_conditional_edges(
    "node_a",           # 从哪个节点出发
    router_function,    # 路由函数，返回一个字符串
    {                   # 字符串到节点名称的映射
        "continue": "node_b",
        "done": END,
    }
)

路由函数接收当前 State，返回一个字符串，这个字符串必须是映射字典里的某个 key。

python

def router_function(state: AgentState) -> str:
    if state["search_results"] and len(state["search_results"]) >= 3:
        return "done"      # 搜索结果够用，结束循环
    return "continue"      # 搜索结果不够，继续搜索

注意这里有一个常见坑：路由函数的返回值必须和映射字典的 key 完全一致，包括大小写、空格。如果返回了一个不存在的 key，LangGraph 会报错，而且错误信息有时候不够直白，容易造成困惑。

1.3 实现循环：节点指向自己

Conditional Edge 可以让一个节点在特定条件下跳回到自己（或者跳回到前面的某个节点），这样就形成了循环。

图里的循环路径：search → evaluate → search 就是 Conditional Edge 实现的。evaluate 节点本身不跳转，跳转逻辑在 evaluate 之后的 Conditional Edge 里。

1.4 防止无限循环

有循环就有死循环的风险。LLM 有时候会持续认为"信息不够"，一直触发继续搜索，永远不结束。

解决方案是在 State 里加一个计数器，在路由函数里检查它。

python

class AgentState(TypedDict):
    user_input: str
    search_query: str
    search_results: List[str]
    final_answer: str
    iteration_count: int    # 循环计数器
    max_iterations: int     # 最大循环次数

路由函数里加上强制终止的逻辑：

python

def should_continue(state: AgentState) -> str:
    # 超过最大次数，强制结束
    if state["iteration_count"] >= state["max_iterations"]:
        return "done"

    # 正常判断逻辑
    results = state["search_results"]
    if results and len(results) >= 3:
        return "done"

    return "continue"

这是生产环境中必须遵守的防御性编程习惯。

1.5 一个完整示例：搜索-评估-循环 Agent

下面是一个完整的可运行示例。Agent 会不断搜索，直到收集到足够多的搜索结果（模拟为找到 3 条以上）或者达到最大循环次数为止。

python

from typing import TypedDict, List
from langgraph.graph import StateGraph, START, END
import random


# 1. 定义 State
class AgentState(TypedDict):
    user_input: str
    search_query: str
    search_results: List[str]
    final_answer: str
    iteration_count: int
    max_iterations: int


# 2. 模拟搜索工具（实际项目换成 Tavily 等真实 API）
def mock_search(query: str, attempt: int) -> List[str]:
    """
    模拟搜索：前两次故意返回少量结果，第三次返回足够多的结果
    用来演示循环逻辑
    """
    if attempt <= 2:
        return [f"结果{attempt}-1：{query} 的部分信息"]
    else:
        return [
            f"结果{attempt}-1：{query} 的详细介绍",
            f"结果{attempt}-2：{query} 的实际案例",
            f"结果{attempt}-3：{query} 的深度分析",
            f"结果{attempt}-4：{query} 的专家观点",
        ]


# 3. 定义节点

def search_node(state: AgentState) -> dict:
    """搜索节点：执行搜索，把结果累积到 search_results 里"""
    iteration = state["iteration_count"] + 1
    query = state["search_query"] or state["user_input"]

    # 每次迭代可以略微调整搜索词（实际场景里由 LLM 生成新的搜索词）
    if iteration > 1:
        query = f"{state['user_input']} 深入分析 第{iteration}轮"

    print(f"\n[搜索节点] 第 {iteration} 次搜索，关键词：{query}")
    new_results = mock_search(query, iteration)
    print(f"[搜索节点] 本次找到 {len(new_results)} 条结果")

    # 累积搜索结果
    all_results = state["search_results"] + new_results

    return {
        "search_query": query,
        "search_results": all_results,
        "iteration_count": iteration,
    }


def evaluate_node(state: AgentState) -> dict:
    """
    评估节点：判断搜索结果是否足够
    注意：这个节点本身不做跳转，跳转由后面的 Conditional Edge 决定
    """
    results = state["search_results"]
    print(f"[评估节点] 当前共有 {len(results)} 条搜索结果，"
          f"迭代次数 {state['iteration_count']}/{state['max_iterations']}")

    # 这里可以用 LLM 来判断结果质量
    # 简单起见用结果数量代替
    if len(results) >= 3:
        print("[评估节点] 信息充足，准备生成答案")
    else:
        print("[评估节点] 信息不足，需要继续搜索")

    return {}  # 评估节点不更新 State，只做判断


def generate_answer_node(state: AgentState) -> dict:
    """生成最终答案"""
    print(f"\n[生成节点] 基于 {len(state['search_results'])} 条结果生成答案")

    results_text = "\n".join(
        f"  - {r}" for r in state["search_results"]
    )
    answer = (
        f"关于「{state['user_input']}」的分析报告\n"
        f"（经过 {state['iteration_count']} 轮搜索，共收集 "
        f"{len(state['search_results'])} 条资料）\n\n"
        f"资料汇总：\n{results_text}"
    )

    return {"final_answer": answer}


# 4. 路由函数

def should_continue_search(state: AgentState) -> str:
    """
    路由函数：决定是继续搜索还是生成答案
    返回值必须和 add_conditional_edges 的映射 key 完全一致
    """
    # 防止无限循环：超过最大次数强制结束
    if state["iteration_count"] >= state["max_iterations"]:
        print(f"[路由] 已达最大迭代次数 {state['max_iterations']}，强制结束")
        return "done"

    # 判断搜索结果是否足够
    if len(state["search_results"]) >= 3:
        return "done"

    return "continue"


# 5. 构建图

def build_search_agent():
    graph = StateGraph(AgentState)

    graph.add_node("search", search_node)
    graph.add_node("evaluate", evaluate_node)
    graph.add_node("generate", generate_answer_node)

    graph.add_edge(START, "search")

    # 普通边：search → evaluate
    graph.add_edge("search", "evaluate")

    # 条件边：evaluate 之后，根据路由函数决定下一步
    graph.add_conditional_edges(
        "evaluate",
        should_continue_search,
        {
            "continue": "search",   # 继续搜索，形成循环
            "done": "generate",     # 信息足够，生成答案
        }
    )

    # 普通边：generate → END
    graph.add_edge("generate", END)

    return graph.compile()


# 6. 运行

if __name__ == "__main__":
    app = build_search_agent()

    initial_state = {
        "user_input": "LangGraph 条件边原理",
        "search_query": "",
        "search_results": [],
        "final_answer": "",
        "iteration_count": 0,
        "max_iterations": 5,   # 最多搜 5 次
    }

    print("开始执行搜索 Agent...\n")
    result = app.invoke(initial_state)

    print("\n" + "="*50)
    print("执行完成！")
    print("="*50)
    print(result["final_answer"])

运行输出：

code

开始执行搜索 Agent...

[搜索节点] 第 1 次搜索，关键词：LangGraph 条件边原理
[搜索节点] 本次找到 1 条结果
[评估节点] 当前共有 1 条搜索结果，迭代次数 1/5
[评估节点] 信息不足，需要继续搜索

[搜索节点] 第 2 次搜索，关键词：LangGraph 条件边原理 深入分析 第2轮
[搜索节点] 本次找到 1 条结果
[评估节点] 当前共有 2 条搜索结果，迭代次数 2/5
[评估节点] 信息不足，需要继续搜索

[搜索节点] 第 3 次搜索，关键词：LangGraph 条件边原理 深入分析 第3轮
[搜索节点] 本次找到 4 条结果
[评估节点] 当前共有 6 条搜索结果，迭代次数 3/5
[评估节点] 信息充足，准备生成答案

[生成节点] 基于 6 条结果生成答案

循环精确执行了 3 次，第 3 次搜索到足够的内容后自动跳出循环，生成答案。

1.6 常见坑：路由函数返回值不匹配

这是最高频的报错。举个例子：

python

# 错误示例
graph.add_conditional_edges(
    "evaluate",
    should_continue_search,
    {
        "Continue": "search",   # 注意这里是大写 C
        "done": "generate",
    }
)

def should_continue_search(state):
    return "continue"  # 小写 c，和映射 key "Continue" 不匹配

LangGraph 会抛出类似 Invalid return value for conditional edge 的错误。

解决方法是用常量或者枚举来管理路由函数的返回值，避免手写字符串导致的拼写错误：

python

# 推荐写法：用常量
ROUTE_CONTINUE = "continue"
ROUTE_DONE = "done"

graph.add_conditional_edges(
    "evaluate",
    should_continue_search,
    {
        ROUTE_CONTINUE: "search",
        ROUTE_DONE: "generate",
    }
)

def should_continue_search(state: AgentState) -> str:
    if len(state["search_results"]) >= 3:
        return ROUTE_DONE
    return ROUTE_CONTINUE

1.7 多个出口的条件边

Conditional Edge 不限于两个出口，可以有多个：

python

def classify_query(state: AgentState) -> str:
    query = state["user_input"]
    if "代码" in query or "如何实现" in query:
        return "code_path"
    elif "是什么" in query or "介绍" in query:
        return "explain_path"
    else:
        return "search_path"

graph.add_conditional_edges(
    "classify",
    classify_query,
    {
        "code_path": "code_generator",
        "explain_path": "explainer",
        "search_path": "searcher",
    }
)

这就是一个分类路由，根据用户意图把请求分发到不同的处理分支。实际项目里，这个分类逻辑通常由 LLM 来做，更加灵活。

1.8 小结

没有条件分支，LangGraph 只是一个线性流程的执行器。加上条件边和循环，Agent 才能根据运行时的结果调整路径，而不是跑完固定的几步就结束。

下一篇会讲 Human-in-the-loop，让 Agent 在关键节点暂停下来，等待人工确认后再继续。