多智能体协作平台

本实战构建一个 Supervisor + 多 Worker 的多智能体协作平台。场景是"内容创作团队"：策划出题、研究找资料、写作成稿、审校把关，四个角色各司其职，由 Supervisor 智能体统一调度。这是 Supervisor 模式在真实业务里的落地。

一、需求分析

内容创作是个典型的多角色协作流程：

策划（Planner）：把用户主题拆成创作大纲和要点。
研究（Researcher）：按大纲搜集资料。
写作（Writer）：综合资料写成初稿。
审校（Reviewer）：检查质量，不合格打回重写。

这些角色如果让一个智能体全干，prompt 会臃肿、容易顾此失彼。拆成多个 Worker，由一个 Supervisor 决定"下一步交给谁"，每个 Worker 专注自己的事——这就是 Supervisor 模式的价值。

二、架构设计

mermaid

flowchart TD
    U([用户主题]) --> SUP[Supervisor 调度器]
    SUP -- 交给策划 --> P[Planner]
    SUP -- 交给研究 --> R[Researcher]
    SUP -- 交给写作 --> W[Writer]
    SUP -- 交给审校 --> RV[Reviewer]
    P --> SUP
    R --> SUP
    W --> SUP
    RV --> SUP
    SUP -- 完成 --> E([END])

Supervisor 是中枢：每次收到某个 Worker 的产出后，决定下一步交给谁，直到任务完成。所有 Worker 共享一个状态，Supervisor 根据状态里的"当前进度"做路由决策。

对 Java/LangGraph4j 同学：这相当于一个调度中心 + 多个职能服务，调度中心根据当前业务状态路由到不同服务，跟工作流引擎的"人工节点路由"类似。

三、状态设计

共享状态是协作的枢纽。设计原则：只放所有 Worker 都需要读写的字段，避免职责污染。

python

from typing import TypedDict, List, Annotated, Optional

def append_str(left: str, right: str) -> str:
    """字符串追加 reducer：每个 worker 把自己的产出追加到 outputs"""
    if not right:
        return left or ""
    return (left or "") + "\n---\n" + right

class TeamState(TypedDict):
    task: str                    # 用户原始主题/任务
    outline: str                 # 策划产出的大纲
    materials: str               # 研究产出的资料
    draft: str                   # 写作产出的初稿
    review: str                  # 审校产出的意见
    outputs: Annotated[str, append_str]  # 所有角色产出的日志（追加）
    current_worker: str          # Supervisor 决定的下一个 worker
    done: bool                   # 任务是否完成
    step_count: int              # 步骤计数，防死循环
    messages: Annotated[list, "add_messages"]  # 可选：对话历史

注意 outputs: Annotated[str, append_str]——每个 Worker 把自己的产出追加进来，形成完整执行日志。step_count 是死循环刹车。

四、Worker 实现

每个 Worker 是一个普通节点函数，读自己关心的状态字段，写自己的产出字段。为演示清晰，这里用 Mock 核心能力（真实项目里 Worker 可以是子图或 create_react_agent）。

1. 策划 Worker

python

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def planner(state: TeamState) -> dict:
    """策划：把主题拆成大纲"""
    task = state["task"]
    outline = llm.invoke(
        f"你是内容策划。把下面主题拆成一份创作大纲，含3-5个要点和写作方向：\n主题：{task}"
    ).content
    return {
        "outline": outline,
        "outputs": f"[策划] 已产出大纲",
        "step_count": state.get("step_count", 0) + 1,
    }

2. 研究 Worker

python

def researcher(state: TeamState) -> dict:
    """研究：根据大纲搜集资料（Mock）"""
    outline = state.get("outline", "")
    # Mock 资料搜集；真实项目接向量库或 web 搜索
    materials = llm.invoke(
        f"你是研究员。根据大纲搜集相关资料要点，3-5条简明要点：\n大纲：{outline}"
    ).content
    return {
        "materials": materials,
        "outputs": f"[研究] 已产出资料",
        "step_count": state.get("step_count", 0) + 1,
    }

3. 写作 Worker

python

def writer(state: TeamState) -> dict:
    """写作：综合大纲和资料写初稿"""
    outline = state.get("outline", "")
    materials = state.get("materials", "")
    draft = llm.invoke(
        f"你是写手。根据大纲和资料写一篇800字左右的初稿：\n"
        f"大纲：{outline}\n\n资料：{materials}"
    ).content
    return {
        "draft": draft,
        "outputs": f"[写作] 已产出初稿",
        "step_count": state.get("step_count", 0) + 1,
    }

4. 审校 Worker

python

def reviewer(state: TeamState) -> dict:
    """审校：检查初稿质量，给通过/打回意见"""
    draft = state.get("draft", "")
    review = llm.invoke(
        f"你是审校。检查初稿质量，输出'PASS'或'REWRITE: 具体问题'：\n初稿：{draft}"
    ).content
    return {
        "review": review,
        "outputs": f"[审校] {review[:30]}",
        "step_count": state.get("step_count", 0) + 1,
    }

五、Supervisor 实现

Supervisor 是整个平台的大脑。它用 LLM 根据当前状态决定"下一步交给谁"，输出一个 worker 名字或 END。

python

from pydantic import BaseModel, Field

class Route(BaseModel):
    next_worker: str = Field(description="下一个 worker：planner/researcher/writer/reviewer/FINISH")

route_llm = llm.with_structured_output(Route)

WORKERS = ["planner", "researcher", "writer", "reviewer"]

def supervisor(state: TeamState) -> dict:
    """调度器：根据当前进度决定下一步交给谁"""
    # 死循环刹车：超过 8 步强制结束
    if state.get("step_count", 0) >= 8:
        return {"current_worker": "FINISH", "done": True}

    task = state["task"]
    # 把各阶段产出状态汇总给 Supervisor 判断
    status = (
        f"任务：{task}\n"
        f"大纲：{'有' if state.get('outline') else '无'}\n"
        f"资料：{'有' if state.get('materials') else '无'}\n"
        f"初稿：{'有' if state.get('draft') else '无'}\n"
        f"审校意见：{state.get('review', '无')}\n"
    )
    prompt = (
        "你是创作团队调度员。根据当前进度决定下一步交给谁。\n"
        "可选：planner(策划) / researcher(研究) / writer(写作) / reviewer(审校) / FINISH(完成)\n"
        "正常流程：planner → researcher → writer → reviewer；"
        "若审校结果是 REWRITE，回到 writer；若 PASS，FINISH。\n\n"
        f"{status}\n只输出下一个 worker 名字。"
    )
    result = route_llm.invoke(prompt)
    nxt = result.next_worker.strip()
    if nxt not in WORKERS + ["FINISH"]:
        nxt = "FINISH"   # 兜底
    return {"current_worker": nxt, "done": nxt == "FINISH"}

Supervisor 的 prompt 是协作质量的关键——它要懂"正常流程"和"打回重做"两种路径。本例把审校的 PASS/REWRITE 信号传给它，让它能据此路由。

六、条件边与路由

Supervisor 之后用条件边把控制权交给被选中的 Worker；Worker 执行完回到 Supervisor，形成循环。

python

def route_from_supervisor(state: TeamState) -> str:
    """根据 Supervisor 决定路由到哪个 worker，或结束"""
    nxt = state["current_worker"]
    if nxt == "FINISH" or state.get("done"):
        return END
    return nxt

七、组装图

python

from langgraph.graph import StateGraph, START, END

gb = StateGraph(TeamState)
gb.add_node("supervisor", supervisor)
gb.add_node("planner", planner)
gb.add_node("researcher", researcher)
gb.add_node("writer", writer)
gb.add_node("reviewer", reviewer)

gb.add_edge(START, "supervisor")
# Supervisor → 某个 worker 或 END
gb.add_conditional_edges("supervisor", route_from_supervisor, {
    "planner": "planner",
    "researcher": "researcher",
    "writer": "writer",
    "reviewer": "reviewer",
    END: END,
})
# 每个 worker 执行完回到 supervisor
for w in WORKERS:
    gb.add_edge(w, "supervisor")

team_app = gb.compile()

完整协作图：

mermaid

flowchart TD
    S([START]) --> SUP[supervisor]
    SUP -- planner --> P[planner]
    SUP -- researcher --> R[researcher]
    SUP -- writer --> W[writer]
    SUP -- reviewer --> RV[reviewer]
    SUP -- FINISH --> E([END])
    P --> SUP
    R --> SUP
    W --> SUP
    RV --> SUP

八、完整可运行示例

python

import os
from typing import TypedDict, List, Annotated
from pydantic import BaseModel, Field

from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# ---- 状态 ----
def append_str(left, right):
    if not right: return left or ""
    return (left or "") + "\n---\n" + right

class TeamState(TypedDict):
    task: str
    outline: str
    materials: str
    draft: str
    review: str
    outputs: Annotated[str, append_str]
    current_worker: str
    done: bool
    step_count: int

# ---- 结构化路由 ----
class Route(BaseModel):
    next_worker: str

WORKERS = ["planner", "researcher", "writer", "reviewer"]

# ---- Workers ----
def planner(state):
    o = llm.invoke(f"你是策划。把主题拆成大纲含3-5要点：\n{state['task']}").content
    return {"outline": o, "outputs": "[策划]大纲完成", "step_count": state.get("step_count",0)+1}

def researcher(state):
    m = llm.invoke(f"你是研究员。根据大纲搜3-5条资料要点：\n{state.get('outline','')}").content
    return {"materials": m, "outputs": "[研究]资料完成", "step_count": state.get("step_count",0)+1}

def writer(state):
    d = llm.invoke(f"你是写手。根据大纲和资料写800字初稿：\n大纲:{state.get('outline','')}\n资料:{state.get('materials','')}").content
    return {"draft": d, "outputs": "[写作]初稿完成", "step_count": state.get("step_count",0)+1}

def reviewer(state):
    r = llm.invoke(f"你是审校。检查初稿，输出'PASS'或'REWRITE:问题'：\n{state.get('draft','')}").content
    return {"review": r, "outputs": f"[审校]{r[:30]}", "step_count": state.get("step_count",0)+1}

# ---- Supervisor ----
def supervisor(state):
    if state.get("step_count",0) >= 8:
        return {"current_worker":"FINISH","done":True}
    status = (f"任务:{state['task']}\n大纲:{'有' if state.get('outline') else '无'}"
              f"\n资料:{'有' if state.get('materials') else '无'}"
              f"\n初稿:{'有' if state.get('draft') else '无'}"
              f"\n审校:{state.get('review','无')}")
    r = llm.with_structured_output(Route).invoke(
        f"你是调度员。决定下一步交给谁(planner/researcher/writer/reviewer/FINISH)。"
        f"正常流:planner→researcher→writer→reviewer；REWRITE回writer；PASS则FINISH。\n{status}\n只输出名字。")
    nxt = r.next_worker.strip()
    if nxt not in WORKERS+["FINISH"]: nxt = "FINISH"
    return {"current_worker": nxt, "done": nxt=="FINISH"}

# ---- 路由 ----
def route(state):
    if state["current_worker"]=="FINISH" or state.get("done"): return END
    return state["current_worker"]

# ---- 建图 ----
gb = StateGraph(TeamState)
gb.add_node("supervisor", supervisor)
for n,f in [("planner",planner),("researcher",researcher),("writer",writer),("reviewer",reviewer)]:
    gb.add_node(n,f)
gb.add_edge(START,"supervisor")
gb.add_conditional_edges("supervisor", route,
    {"planner":"planner","researcher":"researcher","writer":"writer","reviewer":"reviewer", END:END})
for w in WORKERS:
    gb.add_edge(w, "supervisor")
app = gb.compile()

if __name__ == "__main__":
    task = "写一篇面向新手的《为什么用 LangGraph 构建智能体》科普文"
    result = app.invoke({"task": task, "step_count": 0, "outputs": ""})
    print("=== 最终初稿 ===")
    print(result.get("draft", "(无初稿)"))
    print("\n=== 执行日志 ===")
    print(result.get("outputs", ""))
    print("\n=== 审校意见 ===")
    print(result.get("review", ""))

运行后，Supervisor 会依次调度 planner → researcher → writer → reviewer，如果审校说 REWRITE 就回到 writer 重写，PASS 就结束。outputs 字段记录了完整执行链路。

九、用 stream 观察调度过程

python

for chunk in app.stream({"task": task, "step_count": 0, "outputs": ""}, stream_mode="updates"):
    for node, update in chunk.items():
        print(f"[{node}] -> {update.get('outputs','')}")

输出大致是：

text

[supervisor] -> (决定下一个: planner)
[planner] -> [策划]大纲完成
[supervisor] -> (决定下一个: researcher)
[researcher] -> [研究]资料完成
[supervisor] -> (决定下一个: writer)
[writer] -> [写作]初稿完成
[supervisor] -> (决定下一个: reviewer)
[reviewer] -> [审校]REWRITE: 结尾太仓促
[supervisor] -> (决定下一个: writer)
[writer] -> [写作]初稿完成
[supervisor] -> (决定下一个: reviewer)
[reviewer] -> [审校]PASS
[supervisor] -> (FINISH)

十、扩展点

1. 加新 Worker

比如加一个"配图师"：定义 illustrator 节点函数，加到 WORKERS 列表，在 Supervisor prompt 里说明何时调用，建图时 add_node + add_edge(illustrator, "supervisor")。无需改动其他 Worker，这是 Supervisor 模式的扩展性优势。

2. Worker 升级为子图或 ReAct 智能体

本例 Worker 是单次 LLM 调用。真实项目里：

Researcher 可以是深度研究 Agent 子图，多轮搜索。
Writer 可以是带 retrieve_kb 工具的 ReAct 智能体。
把子图作为节点加入：gb.add_node("researcher", research_subgraph)，详见子图 Subgraph。

3. 人工审批节点

关键环节加 HITL：在 writer 之后插一个 human_review 节点，用 interrupt 暂停等人工批注，人工通过后 Command(resume=...) 恢复。详见智能客服系统的 handoff 实现和 HITL 章节。

4. 并行 Worker

研究阶段可以拆成多个子主题，用 Send 让多个 Researcher 并行（参考深度研究 Agent 的并行搜索）。Supervisor 决定"扇出研究"，并行完成后汇总回 Supervisor。

十一、常见踩坑

1. 职责边界模糊

Worker 之间职责重叠（比如 Writer 也去查资料、Researcher 也写两句）会导致产出互相覆盖、Supervisor 路由混乱。解决：

每个 Worker 只写自己专属的字段（planner 写 outline、researcher 写 materials...），不碰别人的。
prompt 里明确角色边界："只做 X，不做 Y"。
用 outputs 日志字段追溯谁产出了什么，便于排查。

2. 死循环

Supervisor 可能在 writer ↔ reviewer 之间反复横跳，或一直不判 FINISH。本例用 step_count >= 8 强制结束兜底。其他手段：

限制 REWRITE 次数（如最多 2 次），超过强制 PASS。
Supervisor prompt 里强调"质量达标即可 FINISH，不要过度追求完美"。
监控单次任务总 token，超预算强制结束。

3. 状态字段污染

共享状态字段一多，就容易"这个 Worker 改了那个 Worker 依赖的字段"。本例 outline/materials/draft/review 各管一块，互不覆盖。如果某个 Worker 需要改别人的字段（比如 Reviewer 直接改 draft），要明确这是"打回重写"还是"直接修改"，并在 prompt 里约束。建议Reviewer 只产出意见，不改 draft，由 Writer 根据意见重写，职责更清晰。

4. Supervisor 路由决策不稳

Supervisor 用 LLM 决策，可能输出不存在的 worker 名字或乱跳。缓解：

用 with_structured_output(Route) 强制结构化输出。
兜底：不在白名单里的输出统一转 FINISH 或回 planner。
Supervisor prompt 里给完整流程示例和"打回"逻辑，减少自由发挥。

5. 共享状态里塞了不该塞的东西

新手容易把"对话历史""中间思考"全塞进共享状态，导致状态膨胀、Worker 之间互相干扰。原则：共享状态只放跨 Worker 协作必需的字段，每个 Worker 的内部思考用局部变量或独立子图状态，不污染顶层。

十二、小结

多智能体协作平台 = Supervisor（调度）+ 多 Worker（执行），共享一个 TeamState。
Supervisor 用 LLM + 结构化输出决定下一步交给谁，Worker 执行完回 Supervisor 形成循环。
状态设计原则：每个 Worker 只写自己的产出字段，避免职责污染。
必加 step_count 死循环刹车，并限制 REWRITE 次数。
扩展性强：加 Worker 只需加节点+更新 Supervisor prompt；Worker 可升级为子图或 ReAct 智能体。
关键环节可加 interrupt 人工审批，实现 HITL 协作。

至此，实战项目四篇全部完成。下一篇进入部署与运维，把开发好的应用推向生产。

多智能体协作平台 ​

一、需求分析 ​

二、架构设计 ​

三、状态设计 ​

四、Worker 实现 ​

1. 策划 Worker ​

2. 研究 Worker ​

3. 写作 Worker ​

4. 审校 Worker ​

五、Supervisor 实现 ​

六、条件边与路由 ​

七、组装图 ​

八、完整可运行示例 ​

九、用 stream 观察调度过程 ​

十、扩展点 ​

1. 加新 Worker ​

2. Worker 升级为子图或 ReAct 智能体 ​

3. 人工审批节点 ​

4. 并行 Worker ​

十一、常见踩坑 ​

1. 职责边界模糊 ​

2. 死循环 ​

3. 状态字段污染 ​

4. Supervisor 路由决策不稳 ​

5. 共享状态里塞了不该塞的东西 ​

十二、小结 ​

多智能体协作平台

一、需求分析

二、架构设计

三、状态设计

四、Worker 实现

1. 策划 Worker

2. 研究 Worker

3. 写作 Worker

4. 审校 Worker

五、Supervisor 实现

六、条件边与路由

七、组装图

八、完整可运行示例

九、用 stream 观察调度过程

十、扩展点

1. 加新 Worker

2. Worker 升级为子图或 ReAct 智能体

3. 人工审批节点

4. 并行 Worker

十一、常见踩坑

1. 职责边界模糊

2. 死循环

3. 状态字段污染

4. Supervisor 路由决策不稳

5. 共享状态里塞了不该塞的东西

十二、小结