引言：Agent 不是单模型的循环

做 Agent 应用时很多人一开始就想"让模型多跑几次"——给个工具清单、一个 while 循环，让模型自己决定下一步。这在玩具 Demo 里够用，但一旦任务变成"先检索再分析、查不到就换策略、最后写成结构化报告"，单循环就兜不住了。原因很简单：复杂任务天然有依赖关系、有分支、有先后顺序，而单循环只能表达"一条线走到底"。这篇文章用可跑的 Python 代码，从 ReAct 讲起，一路到多 Agent 协作和图工作流，把几种主流编排架构拆开给你看。

文中所有代码用 openai + anthropic 客户端均可运行，也可换成任意 OpenAI 兼容网关。核心是编排思想，不是某个框架。

一、ReAct：把"思考-行动-观察"循环显式化

ReAct 把模型的每一步拆成 Thought（想什么）、Action（调什么工具）、Observation（工具返回什么），模型按这个节奏一路推进直到拿到答案。

一个最简实现，核心就一个循环：

from openai import OpenAI

client = OpenAI()

def run_react(prompt, tools, max_steps=10):
    messages = [{"role": "system",
                 "content": "你是ReAct代理。请按如下格式回答：\n"
                            "Thought: 你现在的推理\n"
                            "Action: 工具名(参数)\n"
                            "Observation: 工具结果\n"
                            "当得到最终答案时输出 Final Answer: ..."}]
    messages.append({"role": "user", "content": prompt})
    for _ in range(max_steps):
        resp = client.chat.completions.create(
            model="gpt-4o", messages=messages)
        text = resp.choices[0].message.content
        print(">>>", text)
        if "Final Answer" in text:
            return text
        # 解析 Action 行，调用对应工具
        action = parse_action(text)  # 比如 "Action: search(\"python\")"
        result = run_tool(action, tools)
        messages.append({"role": "assistant", "content": text})
        messages.append({"role": "user", "content": f"Observation: {result}"})
    return "达到步数上限"

def parse_action(text):
    import re
    m = re.search(r"Action:\s*(\w+)\((.*)\)", text)
    return (m.group(1), m.group(2)) if m else (None, None)

def run_tool(action, tools):
    name, arg = action
    return tools.get(name, lambda _: "未知工具")(arg)

这段代码的价值在于把 Agent 的"自主性"摊开给你看：循环、记忆（messages）、工具路由三件事就构成了一个最小的自主 Agent。生产里你不需要手写正则解析，LangChain 的 create_react_agent 或 OpenAI 的 Agent SDK 都封装好了，但理解这个循环，后面看多 Agent 才有基础。

优点：实现直观、调试容易、适合工具数量少（几个到十几个）的检索问答。 局限：步数多了上下文迅速膨胀；没有并行；任务一旦需要"分工"，单循环无能为力。

二、Router/路由：先分诊，再派活

任务种类一多，第一步往往是分诊——用一个轻量模型判断任务类型，再路由到对应的专用流程。这在客服、工单、内容生成场景最常见。

def route(user_input):
    route_prompt = (
        "判断下面用户问题的类别，只输出一个词："
        "summary | extract | code | default\n\n" + user_input)
    kind = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": route_prompt}]
    ).choices[0].message.content.strip()
    return {"summary": summarize, "extract": extract,
            "code": write_code, "default": faq_answer}[kind]

result = route("帮我总结这份会议纪要的关键决定")

路由架构的好处是每个分支可以各自优化：摘要走长上下文模型、代码走带沙箱的流程、FAQ 直接查库，互不拖累。代价是要维护分类质量和路由兜底——分类错一次，整个下游都错。

三、Pipeline/流水线：固定的有序多阶段

当任务阶段是确定的（A 完成后才能做 B），流水线是最朴素的编排：输入依次流过若干阶段，每阶段一个职责。

def pipeline(pdf_path):
    text   = extract_text(pdf_path)      # 1. 抽取
    chunks = split_into_chunks(text)     # 2. 切块
    nodes  = [embed_and_lookup(c) for c in chunks]  # 3. 检索
    report = synthesize(nodes)           # 4. 生成
    return report

业界标准实现有 LangGraph 的链式节点、CrewAI 的顺序任务（SequentialProcess）。流水线适合流程固定、质量要求稳定的场景，比如文档处理、定时报表。缺点也很直白：中间某阶段失败整条链断，且阶段间不能复用中间结果到别处。

四、多 Agent 协作：分工、交流、汇总

任务大到"调研 + 写作 + 校对 + 排版"，单 Agent 力不从心，于是出现多 Agent：每个 Agent 负责一个子任务，通过共享消息或显式交接协作。两种主流模式：

分工（Executor）模式——一个协调者拆分任务，多个执行者并行干，最后汇总：

from concurrent.futures import ThreadPoolExecutor

def research_team(topic):
    subtasks = split(topic)          # 协调者拆解
    with ThreadPoolExecutor(max_workers=4) as pool:
        results = list(pool.map(lambda t: researcher(t), subtasks))
    return synthesize_report(results)

讨论/接力模式——Agent 之间互相传递中间产物。典型如"一个写代码、一个审代码、一个补测试"，每个 Agent 的结果是下一个的输入，多轮收敛到合格答案。

code = coder(spec)
for i in range(3):
    review = reviewer(code)
    if review.ok:
        break
    code = coder(code, review.comments)  # 把意见传回

多 Agent 的核心难点是沟通协议与状态共享：谁拥有最终结果？中间结果放哪？怎么防死循环？这就是为什么工程上要把多 Agent 落在结构化图上，而不是让 Agent 自由聊天。

五、Graph Workflow：把编排显式化为有向图

LangGraph 的出现把上面所有模式统一到一个抽象：状态机 + 有向图。节点是处理函数，边是条件转移，共享状态是一个 dict。前面所有模式都是它的特例。

from typing import TypedDict
from langgraph.graph import StateGraph, END

class State(TypedDict):
    question: str
    route: str
    answer: str

def classify(state: State) -> State:
    state["route"] = "code" if "code" in state["question"] else "summary"
    return state

def run_code(state: State) -> State:
    state["answer"] = f"代码处理: {state['question']}"
    return state

def run_summary(state: State) -> State:
    state["answer"] = f"摘要处理: {state['question']}"
    return state

g = StateGraph(State)
g.add_node("classify", classify)
g.add_node("code", run_code)
g.add_node("summary", run_summary)
g.set_entry_point("classify")
g.add_conditional_edges("classify", lambda s: s["route"],
                        {"code": "code", "summary": "summary"})
g.add_edge("code", END)
g.add_edge("summary", END)
app = g.compile()

这个 20 行示例表达了完整的"分诊 + 分支 + 汇合"：classify 节点根据 state["route"] 条件跳转到不同分支。图编排的威力在于：

1. 显式控制流：谁能走、什么时候走、走哪条边一目了然；
2. 检查点与恢复：状态可持久化，任务中断可从最近节点续跑；
3. 人机回环：可插入人工审批节点，符合生产合规要求；
4. 可观测性：每一步的状态转移都能被记录和追踪。

再往上，多 Agent 在 LangGraph 里就是"多个子图 + 一个 Supervisor"（主管 Agent 动态决定把任务派给哪个子 Agent），对应 create_agent / add_node("supervisor", ...) 的组合。这就是生产级多 Agent 的标准形态。

六、怎么选？一张图看清你的场景

模式	适用场景	并行能力	控制流表达	工程复杂度
ReAct 循环	工具少、检索问答	无	弱（靠模型自觉）	低
Router 路由	任务类型多、分类明确	无	中	低
Pipeline 流水线	阶段固定、流程确定	有限	中	低-中
多 Agent 协作	大任务分工、讨论收敛	强	中	中
Graph 工作流	复杂分支、人机回环、需要可恢复	强	强（显式）	中-高

选型口诀：

• 工具少、单任务 → ReAct 够了，别过度设计；
• 任务种类多但每类独立 → Router；
• 阶段固定不变 → Pipeline；
• 任务大、要并行和分工 → 多 Agent；
• 既要分工又要复杂分支/回环/恢复 → Graph Workflow。

总结

从 ReAct 到 Graph Workflow，本质是把 Agent 的"自由度"一步步收进可控的结构里：自由循环适合探索，路由适合分流，流水线适合固定流程，多 Agent 适合分工，图则把它们统一成可观测、可恢复、可人机回环的工程体系。实际项目中，它们通常组合出现——入口路由分诊、主干走图工作流、叶子节点用 ReAct 单步。先想清楚你的任务依赖关系，再选编排形态，比追新框架重要得多。