常见架构模式

LangGraph 的图结构非常灵活，可以表达多种架构模式。本文总结在实际项目中常见的模式，可以直接套用。

1. ReAct Agent 模式

最基本的 Agent 模式：思考 → 行动 → 观察 → 循环。

              → 工具节点 ┐
START → LLM              → LLM（循环）→ END

def agent_loop(state: MessagesState) -> dict:
    response = model.invoke(state["messages"])
    return {"messages": [response]}

def should_continue(state: MessagesState) -> Literal["tools", END]:
    if state["messages"][-1].tool_calls:
        return "tools"
    return END

def tool_executor(state: MessagesState) -> dict:
    results = []
    for tc in state["messages"][-1].tool_calls:
        tool = tools_by_name[tc["name"]]
        result = tool.invoke(tc["args"])
        results.append(ToolMessage(content=result, tool_call_id=tc["id"]))
    return {"messages": results}

# 图
builder.add_node("agent", agent_loop)
builder.add_node("tools", tool_executor)
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_continue, ["tools", END])
builder.add_edge("tools", "agent")  # 工具执行完回到 agent

适用场景：需要工具调用的通用 Agent

2. RAG Agent 模式

检索增强生成，先检索后回答。

START → 检索 → 增强 → 生成 → END

def retrieve(state: RAGState):
    docs = vectorstore.similarity_search(state["question"])
    return {"context": docs}

def augment(state: RAGState):
    context = "\n".join([d.page_content for d in state["context"]])
    augmented_prompt = f"基于以下内容回答问题：\n{context}\n\n问题：{state['question']}"
    return {"augmented_prompt": augmented_prompt}

def generate(state: RAGState):
    response = llm.invoke(state["augmented_prompt"])
    return {"answer": response}

builder.add_edge(START, "retrieve")
builder.add_edge("retrieve", "augment")
builder.add_edge("augment", "generate")
builder.add_edge("generate", END)

适用场景：知识库问答、文档问答

3. 并行 Fan-out 模式

一个任务分发给多个并行处理器，然后汇总结果。

           → Processor A ┐
Dispatcher → Processor B → Aggregator → END
           → Processor C ┘

def dispatcher(state: FanState):
    """分发任务"""
    tasks = split_work(state["input"])
    return {"tasks": tasks}

def processor_a(state: FanState):
    return {"results_a": [process_chunk(task) for task in state["tasks"]]}

def processor_b(state: FanState):
    return {"results_b": [process_chunk_b(task) for task in state["tasks"]]}

def aggregator(state: FanState):
    all_results = state["results_a"] + state["results_b"]
    return {"final": merge_results(all_results)}

# 图
builder.add_edge(START, "dispatcher")
builder.add_edge("dispatcher", "processor_a")
builder.add_edge("dispatcher", "processor_b")
builder.add_edge("processor_a", "aggregator")
builder.add_edge("processor_b", "aggregator")
builder.add_edge("aggregator", END)

适用场景：批量处理、多源验证、并行搜索

4. Quality Check 模式

生成内容后进行质量检查，不合格则重新生成。

                 → 不合格（回退重试）
START → 生成 → 质检 → 合格 → END

def generate(state: QCState):
    content = llm.invoke(state["prompt"])
    return {"draft": content, "attempts": state.get("attempts", 0) + 1}

def quality_check(state: QCState) -> Literal["generate", END]:
    if state["attempts"] >= 3:
        return END  # 重试次数上限，接受当前结果

    criteria = check_quality(state["draft"])
    if criteria.passed:
        return END
    return "generate"  # 未通过，重新生成

builder.add_node("generate", generate)
builder.add_node("quality_check", quality_check)
builder.add_edge(START, "generate")
builder.add_edge("generate", "quality_check")
builder.add_conditional_edges("quality_check", quality_check, ["generate", END])

适用场景：内容生成质检、代码审查、翻译校对

5. Orchestrator-Worker 模式

协调者制定计划，工人执行，协调者评估结果。

                       → Worker A
Orchestrator (计划)      → Worker B → Orchestrator (评估) → 完成/继续
                       → Worker C

def orchestrator_plan(state: OWState):
    plan = llm.invoke(f"为任务制定计划: {state['task']}")
    return {"plan": plan}

def worker_a(state: OWState):
    return {"results": [execute_worker_a(state["plan"])]}

def worker_b(state: OWState):
    return {"results": [execute_worker_b(state["plan"])]}

def orchestrator_evaluate(state: OWState) -> Literal["orchestrator", END]:
    evaluation = llm.invoke(f"评估结果: {state['results']}")
    if "需要调整" in evaluation:
        # 重新规划
        return "orchestrator"
    return END

builder.add_edge(START, "orchestrator_plan")
builder.add_edge("orchestrator_plan", "worker_a")
builder.add_edge("orchestrator_plan", "worker_b")
builder.add_edge("worker_a", "orchestrator_evaluate")
builder.add_edge("worker_b", "orchestrator_evaluate")
builder.add_conditional_edges(
    "orchestrator_evaluate",
    orchestrator_evaluate,
    ["orchestrator_plan", END]
)

适用场景：研究助手、报告生成、软件开发

6. Human-in-the-Loop 模式

关键步骤需要人工审批。

           → 需要审批 → [Interrupt] → 人工审核 → 继续/取消
START → Agent                        → 自动批准 → 继续

from langgraph.types import interrupt

def sensitive_action(state: State):
    action = decide_action(state)

    if action["risk_level"] == "high":
        # 需要人工审批，暂停等待
        approval = interrupt({
            "action": action["description"],
            "reason": action["reason"]
        })

        if approval.get("approved"):
            return {"result": execute(action), "status": "approved"}
        else:
            return {"result": "已取消", "status": "rejected"}

    # 低风险，自动执行
    return {"result": execute(action), "status": "auto_approved"}

适用场景：支付审批、内容发布、敏感操作

选择模式的指南

需求	推荐模式
通用工具调用	ReAct Agent
知识库问答	RAG Agent
批量处理	Fan-out
内容生成	Quality Check
复杂任务分解	Orchestrator-Worker
需要人工审核	HITL
多个专家协作	Supervisor Multi-Agent

模式组合

实际项目中经常组合多种模式：

[RAG + ReAct] → Agent 先检索再根据结果决定是否调用工具
[Orchestrator + HITL] → 协调者规划 → 人工确认计划 → 工人执行
[Fan-out + Quality Check] → 并行处理 → 分别质检 → 汇总

参考

• 官方文档：Workflows & Agents
• 官方文档：LangGraph Overview