Chuoer47's Blog

Appearance

节点与边的高级用法 ​

掌握了基本的 State、Node、Edge 之后，本篇深入介绍 LangGraph 的高级控制流：条件路由、并行执行、Map-Reduce、子图嵌套等。

条件边（Conditional Edge） ​

条件边是 LangGraph 最强大的特性之一——根据当前状态动态决定下一步执行哪个节点。

基本语法 ​

python
from langgraph.graph import StateGraph, START, END

def route_after_llm(state: AgentState) -> str:
    """根据 LLM 输出决定下一步"""
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "call_tools"    # 有工具调用 → 执行工具
    else:
        return END             # 无工具调用 → 结束

graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)

# 条件边：从 agent 出发，根据路由函数决定去哪
graph.add_conditional_edges(
    "agent",           # 源节点
    route_after_llm,   # 路由函数
    {
        "call_tools": "tools",  # 路由值 → 目标节点
        END: END,
    }
)

路由函数的规则 ​

路由函数必须：

  • 接收 state 作为唯一参数
  • 返回一个字符串（路由键）
  • 返回值必须在 add_conditional_edges 的映射中存在
python
# 路由函数的几种写法

# 写法 1：直接返回节点名
def route(state) -> str:
    return "node_a"  # 直接返回目标节点名

# 写法 2：返回映射键
def route(state) -> str:
    return "path_a"  # 返回映射中的键，由映射转换为节点名

# 写法 3：使用 END 常量
def route(state) -> str:
    if done:
        return END   # LangGraph 特殊常量
    return "next_node"

映射字典的简化 ​

当路由函数直接返回节点名时，可以省略映射字典：

python
# 完整写法
graph.add_conditional_edges("agent", route_fn, {
    "tools": "tools",
    "end": END,
})

# 简化写法（路由函数返回的值就是节点名）
graph.add_conditional_edges("agent", route_fn, ["tools", END])
# 路由函数需返回 "tools" 或 END

多层条件路由 ​

条件边可以有多层，实现复杂的决策树：

python
def classify_input(state) -> str:
    """第一层路由：分类用户输入"""
    if is_question(state):
        return "qa"
    elif is_task(state):
        return "task_planner"
    else:
        return "chitchat"

def route_task(state) -> str:
    """第二层路由：任务分发"""
    task_type = state["task_type"]
    if task_type == "search":
        return "search_agent"
    elif task_type == "code":
        return "code_agent"
    elif task_type == "analysis":
        return "analysis_agent"
    else:
        return "general_agent"

graph.add_conditional_edges("classifier", classify_input, {
    "qa": "qa_node",
    "task_planner": "task_planner",
    "chitchat": "chitchat_node",
})

graph.add_conditional_edges("task_planner", route_task, {
    "search_agent": "search_agent",
    "code_agent": "code_agent",
    "analysis_agent": "analysis_agent",
    "general_agent": "general_agent",
})

并行执行 ​

LangGraph 支持节点并行执行——当一个节点有多个后继节点时，它们会自动并行运行。

自动并行 ​

python
graph = StateGraph(AgentState)

# 三个独立的处理节点
graph.add_node("search", search_node)
graph.add_node("analyze", analyze_node)
graph.add_node("summarize", summarize_node)

# 从 "start" 同时连接到三个节点 → 自动并行
graph.add_edge("start", "search")
graph.add_edge("start", "analyze")
graph.add_edge("start", "summarize")

# 三个节点都执行完后，汇合到 "merge"
graph.add_edge("search", "merge")
graph.add_edge("analyze", "merge")
graph.add_edge("summarize", "merge")

执行流程：

        ┌─→ search  ─┐
start ──┼─→ analyze ─┼──→ merge
        └─→ summarize┘

Fan-out / Fan-in 模式 ​

python
def fan_out(state: AgentState) -> list[str]:
    """并行分发到多个节点"""
    return ["search", "analyze", "summarize"]

# 使用条件边返回多个目标实现 fan-out
graph.add_conditional_edges(
    "dispatcher",
    fan_out,
    ["search", "analyze", "summarize"],  # 多个目标
)

Send API ​

Send API 提供更灵活的并行控制，可以向同一节点发送多个不同的输入：

python
from langgraph.types import Send

def assign_workers(state: AgentState) -> list[Send]:
    """为每个子任务创建一个并行的 worker"""
    sends = []
    for task in state["tasks"]:
        sends.append(Send("worker", {
            "task": task,
            "context": state["context"],
        }))
    return sends

# worker 节点会被并行调用多次，每次处理一个子任务
graph.add_conditional_edges("dispatcher", assign_workers)

Map-Reduce 模式 ​

Send API 天然支持 Map-Reduce 模式：

python
from langgraph.types import Send

class MapReduceState(TypedDict):
    documents: list[str]
    summaries: Annotated[list, operator.add]  # Reducer：追加
    final_summary: str

def map_phase(state: MapReduceState) -> list[Send]:
    """Map：为每个文档分配一个 summarizer"""
    return [
        Send("summarizer", {"document": doc})
        for doc in state["documents"]
    ]

def summarizer(state: dict) -> dict:
    """处理单个文档"""
    summary = llm.invoke(f"总结以下内容：{state['document']}")
    return {"summaries": [summary.content]}

def reducer(state: MapReduceState) -> dict:
    """Reduce：合并所有摘要"""
    combined = "\n".join(state["summaries"])
    final = llm.invoke(f"综合以下摘要：{combined}")
    return {"final_summary": final.content}

graph = StateGraph(MapReduceState)
graph.add_node("summarizer", summarizer)
graph.add_node("reducer", reducer)

graph.add_conditional_edges(START, map_phase)
graph.add_edge("summarizer", "reducer")
graph.add_edge("reducer", END)

执行流程：

documents: [doc1, doc2, doc3]
              │
         map_phase (Send API)
         ╱    │    ╲
   summarizer summarizer summarizer  ← 并行执行
         ╲    │    ╱
          reducer (等待所有结果)
              │
         final_summary

子图（Subgraph） ​

子图允许将一个图嵌入另一个图中，实现模块化和代码复用。

基本子图 ​

python
# 定义子图
def create_search_subgraph():
    sg = StateGraph(SearchState)
    sg.add_node("query", query_node)
    sg.add_node("retrieve", retrieve_node)
    sg.add_edge(START, "query")
    sg.add_edge("query", "retrieve")
    sg.add_edge("retrieve", END)
    return sg.compile()

# 在主图中使用子图
search_graph = create_search_subgraph()

main_graph = StateGraph(MainState)
main_graph.add_node("search", search_graph)  # 子图作为节点
main_graph.add_edge(START, "search")
main_graph.add_edge("search", END)

子图的状态映射 ​

子图和主图可以有不同的状态结构，LangGraph 会自动映射同名字段：

python
class MainState(TypedDict):
    query: str
    search_results: list      # ← 主图的字段
    answer: str

class SearchState(TypedDict):
    query: str                 # ← 自动从主图映射
    search_results: list       # ← 自动映射回主图

# 子图节点会自动：
# 1. 从主图状态中提取 search_state 中同名的字段
# 2. 运行子图
# 3. 将子图状态中同名的字段写回主图状态

自定义状态映射 ​

当字段名不同时，需要手动定义输入/输出映射：

python
def map_input(state: MainState) -> dict:
    return {"search_query": state["user_query"]}

def map_output(state: SearchState) -> dict:
    return {"search_results": state["results"]}

main_graph.add_node(
    "search",
    search_graph,
    input=map_input,
    output=map_output,
)

Entry Point 和 Finish Point ​

LangGraph 提供了快捷方式来设置入口和出口：

python
# 方式 1：显式设置
graph.add_edge(START, "first_node")
graph.add_edge("last_node", END)

# 方式 2：使用 set_entry_point / set_finish_point（旧 API，仍可用）
graph.set_entry_point("first_node")
graph.set_finish_point("last_node")

# 方式 3：compile 时指定
graph = graph.compile(
    entry_point="first_node",  # 非标准 API，通常用 add_edge
)

节点的返回值规范 ​

节点函数的返回值会被合并到状态中。规则如下：

python
# 规则 1：返回字典 → 合并到状态
def node(state) -> dict:
    return {"field1": "value1", "field2": "value2"}

# 规则 2：返回空字典 → 不更新任何字段
def node(state) -> dict:
    return {}

# 规则 3：返回 None → 等价于空字典
def node(state) -> dict:
    return None  # 不推荐，但不会报错

# 规则 4：有 Reducer 的字段 → 通过 Reducer 合并
def node(state) -> dict:
    return {"messages": ["new"]}  # add_messages 会追加

# 规则 5：无 Reducer 的字段 → 直接覆盖
def node(state) -> dict:
    return {"current_step": "done"}  # 直接覆盖旧值

错误处理 ​

节点内 try-except ​

python
def safe_node(state: AgentState) -> dict:
    try:
        result = risky_operation(state)
        return {"output": result}
    except Exception as e:
        return {"error": str(e)}

条件边处理错误 ​

python
def check_error(state: AgentState) -> str:
    if state.get("error"):
        return "error_handler"
    return "continue"

graph.add_conditional_edges("main_node", check_error, {
    "error_handler": "error_handler",
    "continue": "next_node",
})

小结 ​

特性说明
条件边add_conditional_edges + 路由函数
并行执行多个后继节点自动并行
Send API向同一节点发送多个不同输入
Map-ReduceSend + Annotated Reducer 实现
子图图的嵌套和模块化
状态映射子图与主图的字段自动/手动映射

下一篇 ​

→ ReAct Agent 与工具集成：ReAct 循环、ToolNode、create_react_agent 快捷方式、自定义 Agent 实现。