📚 AI 工程与智能体教程站

第 6 章 · 条件边、子图与模块化 ​

本章目标：

  • 掌握条件边（conditional edges）的路由逻辑
  • 学会使用子图（subgraph）实现模块化
  • 理解复杂 Agent 的多步编排模式

6.1 条件边基础 ​

条件边根据当前状态决定下一个节点：

python
from langgraph.graph import StateGraph, MessagesState, START, END

def router(state: MessagesState) -> str:
    """根据最后一条消息决定路由"""
    last_msg = state["messages"][-1]
    
    if "天气" in last_msg.content:
        return "weather_node"
    elif "搜索" in last_msg.content:
        return "search_node"
    elif "计算" in last_msg.content:
        return "calc_node"
    else:
        return "default_node"

graph = StateGraph(MessagesState)
graph.add_node("weather", weather_node)
graph.add_node("search", search_node)
graph.add_node("calc", calc_node)
graph.add_node("default", default_node)

graph.add_edge(START, "default")
graph.add_conditional_edges("default", router, {
    "weather_node": "weather",
    "search_node": "search",
    "calc_node": "calc",
    "default_node": END
})

6.2 子图：模块化设计 ​

子图允许将复杂逻辑封装为独立单元：

python
from langgraph.graph import StateGraph, MessagesState, START, END

# 搜索子图
def create_search_subgraph():
    search_state = MessagesState
    
    def search_node(state):
        return {"messages": [{"role": "assistant", "content": "搜索结果..."}]}
    
    sg = StateGraph(search_state)
    sg.add_node("search", search_node)
    sg.add_edge(START, "search")
    sg.add_edge("search", END)
    return sg.compile()

# 分析子图
def create_analysis_subgraph():
    analysis_state = MessagesState
    
    def analyze_node(state):
        return {"messages": [{"role": "assistant", "content": "分析报告..."}]}
    
    sg = StateGraph(analysis_state)
    sg.add_node("analyze", analyze_node)
    sg.add_edge(START, "analyze")
    sg.add_edge("analyze", END)
    return sg.compile()

# 主图
def create_main_graph():
    main_state = MessagesState
    
    def route_node(state):
        last_msg = state["messages"][-1]
        if "搜索" in last_msg.content:
            return "search_subgraph"
        return "analyze_subgraph"
    
    graph = StateGraph(main_state)
    graph.add_node("route", route_node)
    graph.add_node("search_subgraph", create_search_subgraph())
    graph.add_node("analyze_subgraph", create_analysis_subgraph())
    
    graph.add_edge(START, "route")
    graph.add_conditional_edges("route", lambda s: "search_subgraph" if "搜索" in s["messages"][-1].content else "analyze_subgraph", {
        "search_subgraph": END,
        "analyze_subgraph": END
    })
    
    return graph.compile()

6.3 多步骤搜索 Agent ​

python
from langgraph.graph import StateGraph, MessagesState, START, END
from typing import Annotated, operator

class SearchState(MessagesState):
    queries: Annotated[list, operator.add]
    results: Annotated[list, operator.add]
    need_second_search: bool

def search_node(state: SearchState) -> dict:
    query = state["messages"][-1]["content"]
    # 模拟搜索
    results = [f"搜索结果1: {query}", f"搜索结果2: {query}"]
    return {
        "queries": [query],
        "results": results,
        "need_second_search": len(results) < 2
    }

def should_continue(state: SearchState) -> str:
    if state.get("need_second_search"):
        return "refine_search"
    return "summarize"

def refine_search_node(state: SearchState) -> dict:
    # 基于首次结果生成更精确的查询
    refined_query = f"{state['queries'][-1]} 更详细"
    return {"queries": [refined_query], "results": [f"精确结果: {refined_query}"]}

def summarize_node(state: SearchState) -> dict:
    all_results = "\n".join(state.get("results", []))
    return {
        "messages": [{
            "role": "assistant",
            "content": f"综合搜索结果：\n{all_results}"
        }]
    }

# 构建图
workflow = StateGraph(SearchState)
workflow.add_node("search", search_node)
workflow.add_node("refine", refine_search_node)
workflow.add_node("summarize", summarize_node)

workflow.add_edge(START, "search")
workflow.add_conditional_edges("search", should_continue, {
    "refine_search": "refine",
    "summarize": "summarize"
})
workflow.add_edge("refine", "summarize")
workflow.add_edge("summarize", END)

graph = workflow.compile()

本章小结 ​

  • 条件边根据状态动态决定执行路径
  • 子图实现模块化和代码复用
  • 多步搜索 Agent 是条件边的典型应用

🛠️ 动手实践 ​

  1. 实现一个多步翻译 Agent：先翻译，判断质量，质量差则重新翻译
  2. 创建两个子图（数据处理、可视化），在主图中组合
  3. 实现带回退机制的 Agent：主路径失败时切换到备用路径