LangGraph 第8章：Edges 边与条件路由

Edge（边）定义了节点之间的流转路径。LangGraph 提供了两种边：普通边和条件边。正确使用边和路由是构建复杂工作流的关键。

普通边（Edge）

普通边建立固定的连接，源节点执行完毕后自动进入目标节点：

from langgraph.graph import StateGraph, START, END

builder = StateGraph(MyState)

# 从 START 到第一个节点
builder.add_edge(START, "node_a")

# 节点间固定连接
builder.add_edge("node_a", "node_b")
builder.add_edge("node_b", "node_c")

# 最后一个节点到 END
builder.add_edge("node_c", END)

固定链路的执行顺序永远是确定的：

START -> node_a -> node_b -> node_c -> END

条件边（Conditional Edge）

条件边根据路由函数的返回值动态选择下一个节点：

def router_function(state: MyState) -> str:
    """根据状态决定下一步"""
    if state["score"] >= 90:
        return "excellent"
    elif state["score"] >= 60:
        return "pass"
    else:
        return "fail"

builder.add_conditional_edges(
    "grader",            # 源节点
    router_function,     # 路由函数
    {                    # 映射表：返回值 -> 目标节点
        "excellent": "reward",
        "pass": "review",
        "fail": "retry"
    }
)

路由函数编写

路由函数的核心要求：

1. 接收当前 State，返回目标节点名称
2. 必须是纯函数（同样的输入 -> 同样的输出）
3. 返回值必须在映射表中存在

简化写法（省略映射表）：

def simple_router(state):
    if state["done"]:
        return "end"
    return "continue"

# 路由函数返回值直接作为节点名
builder.add_conditional_edges(
    "decision",
    simple_router
    # 无需映射表
)

使用 LLM 作为路由器

对于复杂决策，可以用 LLM 作为路由器：

def llm_router(state: MyState):
    """使用 LLM 决定下一步"""
    llm = ChatOpenAI(model="gpt-4", temperature=0)

    prompt = f"""
    当前任务: {state['task']}
    已完成步骤: {state['completed_steps']}
    下一步应该: [search / process / respond / finish]
    """

    response = llm.invoke(prompt)
    decision = response.content.strip().lower()

    if "search" in decision:
        return "search_tool"
    elif "process" in decision:
        return "data_processor"
    elif "respond" in decision:
        return "response_generator"
    else:
        return "__end__"

builder.add_conditional_edges(
    "analyzer",
    llm_router
)

并行分支

LangGraph 支持并行执行多个节点：

# 分支：一个节点后并行执行多个节点
builder.add_edge("start_node", "branch_a")
builder.add_edge("start_node", "branch_b")
builder.add_edge("start_node", "branch_c")

# 汇合：多个节点汇聚到一个节点
builder.add_edge("branch_a", "merger")
builder.add_edge("branch_b", "merger")
builder.add_edge("branch_c", "merger")

# merger 会在所有三个分支都到达后才执行

执行流程：

             +--> branch_a --+
             |               |
  start_node +--> branch_b --+--> merger --> end
             |               |
             +--> branch_c --+

条件分支

结合条件边和并行分支：

def branch_router(state):
    """根据条件决定走哪个分支"""
    if state["type"] == "a":
        return ["branch_a"]  # 单分支
    elif state["type"] == "all":
        return ["branch_a", "branch_b", "branch_c"]  # 全部分支
    else:
        return ["branch_b"]

builder.add_conditional_edges(
    "router",
    branch_router
)

完整路由示例

下面是一个完整的示例，展示了一个包含条件路由的问答系统：

from typing import TypedDict, Annotated, List
import operator
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI

# 1. 状态定义
class QnAState(TypedDict):
    question: str
    category: str
    search_results: List[str]
    answer: str
    needs_review: bool
    steps: Annotated[List[str], operator.add]

# 2. LLM 初始化
llm = ChatOpenAI(model="gpt-4")

# 3. 节点定义
def classifier_node(state: QnAState):
    """问题分类器"""
    prompt = f"将以下问题分类为 [知识问答, 代码问题, 数学计算, 其他]: {state['question']}"
    category = llm.invoke(prompt).content.strip()
    return {"category": category, "steps": ["分类完成"]}

def knowledge_node(state: QnAState):
    """知识问答处理"""
    answer = llm.invoke(f"回答知识问题: {state['question']}")
    return {"answer": answer.content, "steps": ["知识问答处理完成"]}

def code_node(state: QnAState):
    """代码问题处理"""
    answer = llm.invoke(f"解决代码问题: {state['question']}")
    return {"answer": answer.content, "steps": ["代码处理完成"]}

def math_node(state: QnAState):
    """数学计算处理"""
    answer = llm.invoke(f"计算: {state['question']}")
    return {"answer": answer.content, "steps": ["数学计算完成"]}

def review_node(state: QnAState):
    """人工审核节点"""
    print(f"需要人工审核的问题: {state['question']}")
    print(f"生成的答案: {state['answer']}")
    return {"needs_review": False, "steps": ["审核中"]}

# 4. 路由函数
def category_router(state: QnAState):
    """根据分类路由"""
    category = state["category"]
    if "知识问答" in category:
        return "knowledge"
    elif "代码" in category:
        return "code"
    elif "数学" in category:
        return "math"
    else:
        return "review"  # 其他类型需要人工审核

def quality_router(state: QnAState):
    """根据答案质量决定是否需要审核"""
    if state.get("needs_review", False):
        return "review"
    return "end"

# 5. 构建图
builder = StateGraph(QnAState)
builder.add_node("classifier", classifier_node)
builder.add_node("knowledge", knowledge_node)
builder.add_node("code", code_node)
builder.add_node("math", math_node)
builder.add_node("review", review_node)

builder.add_edge(START, "classifier")

# 条件边：根据分类路由
builder.add_conditional_edges(
    "classifier",
    category_router
)

# 汇聚到结束
builder.add_edge("knowledge", END)
builder.add_edge("code", END)
builder.add_edge("math", END)
builder.add_edge("review", END)

# 6. 编译
app = builder.compile()

# 7. 执行
result = app.invoke({
    "question": "Python 中如何实现快速排序？",
    "category": "",
    "search_results": [],
    "answer": "",
    "needs_review": False,
    "steps": []
})
print(f"答案: {result['answer']}")

路由设计模式

灵活使用边和路由，可以构建出任意复杂的工作流逻辑。下一章我们将应用这些知识构建一个完整的对话机器人。