条件分支与循环路由

4.1 条件分支详解

条件分支允许图根据运行时状态动态决定执行路径。

4.1.1 基础条件路由

python
from typing import Literal

def classify_intent(state: AgentState) -> Literal["greeting", "question", "complaint", "other"]:
    last_msg = state["messages"][-1].content.lower()
    
    if any(g in last_msg for g in ["hi", "hello", "你好"]):
        return "greeting"
    elif "?" in last_msg or any(q in last_msg for q in ["怎么", "什么", "如何"]):
        return "question"
    elif any(c in last_msg for c in ["投诉", "不满", "太差"]):
        return "complaint"
    else:
        return "other"

graph.add_conditional_edges(
    "classify",
    classify_intent,
    {
        "greeting": "handle_greeting",
        "question": "answer_question",
        "complaint": "handle_complaint",
        "other": "general_response"
    }
)

4.1.2 多条件组合

python
def complex_router(state: AgentState) -> str:
    msg = state["messages"][-1].content
    urgency = state.get("urgency_level", 0)
    is_vip = state.get("is_vip", False)
    
    # VIP + 高紧急度 → 人工客服
    if is_vip and urgency > 8:
        return "human_agent"
    # 高紧急度 → 快速通道
    elif urgency > 7:
        return "fast_track"
    # VIP → VIP 通道
    elif is_vip:
        return "vip_queue"
    # 普通 → 标准流程
    else:
        return "standard_flow"

4.2 循环与迭代

LangGraph 原生支持循环，这使得实现 ReAct 等推理模式变得简单。

4.2.1 ReAct 模式实现

python
from langgraph.graph import StateGraph, START

graph = StateGraph(ReActState)

# 添加节点
graph.add_node("reasoner", reason_node)      # 推理
graph.add_node("act", act_node)              # 执行工具
graph.add_node("observe", observe_node)       # 观察结果

# 设置循环
graph.add_edge(START, "reasoner")
graph.add_edge("reasoner", "act")
graph.add_edge("act", "observe")
# 关键：循环回到 reasoner
graph.add_edge("observe", "reasoner")

# 终止条件
def should_continue(state: ReActState) -> Literal["reasoner", "END"]:
    if state["should_stop"]:
        return "END"
    return "reasoner"

graph.add_conditional_edges("observe", should_continue, {
    "reasoner": "reasoner",
    "END": END
})

python
# 防止无限循环：添加步骤计数
class SafeState(TypedDict):
    messages: Annotated[list, add]
    step: int

def check_step_limit(state: SafeState) -> Literal["continue", "END"]:
    if state["step"] >= 10:  # 最多循环 10 次
        return "END"
    return "continue"

# 在节点中递增 step
def increment_step(state: SafeState) -> dict:
    return {"step": state["step"] + 1}