Multi-Agent编排六大致命反模式拆解

# 显式路由的Supervisor实现（LangGraph）
from langgraph.graph import StateGraph, START, END

# 关键：Supervisor的System Prompt必须包含显式路由规则
SUPERVISOR_PROMPT = """
你是编排主管。根据用户请求决定委派给哪个专家：

路由规则（必须严格遵守）：
- 涉及市场数据、竞品分析 → market_researcher
- 涉及技术规格、代码审查 → tech_analyst  
- 涉及文案撰写、内容生成 → content_writer
- 所有子任务完成 → FINISH

禁止：不分解任务直接转发、创建超过5个子任务、
      子任务之间产生循环依赖
"""

workflow = StateGraph(state_schema=AgentState)
workflow.add_node("supervisor", supervisor_agent)
workflow.add_node("market_researcher", market_agent)
workflow.add_node("tech_analyst", tech_agent)
workflow.add_node("content_writer", writer_agent)

# 所有专家回报Supervisor（星型拓扑）
for agent in ["market_researcher", "tech_analyst", "content_writer"]:
    workflow.add_edge(agent, "supervisor")

workflow.add_conditional_edges("supervisor", route_to_agent, {
    "market_researcher": "market_researcher",
    "tech_analyst": "tech_analyst",
    "content_writer": "content_writer",
    "FINISH": END,
})

# Anthropic Claude Research的子Agent OODA循环（简化版）
RESEARCH_SUBAGENT_PROMPT = """
执行优秀的OODA循环：
(a) Observe: 已收集什么信息？还需要什么？有什么工具可用？
(b) Orient: 什么工具和查询最适合获取所需信息？
               根据已学到的更新信念
(c) Decide: 做出有根据的决策，选择特定工具的使用方式
(d) Act: 使用该工具

重复此循环，高效研究并基于新结果学习。

关键约束：
- 最多5次工具调用
- 每次调用后评估是否已充分回答任务目标
- 如果已充分回答，立即输出结果，不要继续搜索
"""

[Supervisor反思步骤]
1. 子Agent输出是否完整回答了分配的子任务？
2. 输出中是否有事实性错误或矛盾？
3. 是否需要将结果退回子Agent修订？
4. 是否需要委派给另一个专家Agent补充？

from langchain_anthropic import ChatAnthropic

# 异构模型路由：不同Agent使用不同模型
supervisor = create_react_agent(
    model=ChatAnthropic(model="claude-opus-4-6"),  # 最强推理
    tools=[], name="supervisor"
)

researcher = create_react_agent(
    model=ChatAnthropic(model="claude-sonnet-4-6"),  # 中端执行
    tools=[web_search, arxiv_search], name="researcher"
)

classifier = create_react_agent(
    model=ChatAnthropic(model="claude-haiku-4-5"),  # 快速分类
    tools=[], name="classifier"
)

from typing import TypedDict, Annotated
import operator

# 定义严格的阶段间合约
class PipelineState(TypedDict):
    # 每个字段只由特定阶段写入
    raw_sources: list[dict]      # 仅由researcher写入
    analysis_findings: dict       # 仅由analyst写入
    draft_content: str            # 仅由writer写入
    final_report: str             # 仅由reviewer写入
    # 元数据跨阶段共享
    task_id: str
    quality_score: float

def researcher_node(state: PipelineState) -> dict:
    """只返回自己的输出，不转发上游的全部上下文"""
    sources = search_and_collect(state["task_id"])
    return {"raw_sources": sources}  # 只写字段，不传递全部state

def analyst_node(state: PipelineState) -> dict:
    """只读需要的字段，只写自己的字段"""
    findings = analyze(state["raw_sources"])  # 只读raw_sources
    return {"analysis_findings": findings}     # 只写analysis_findings

class BudgetGuard:
    """三层预算保护"""
    def __init__(self):
        self.token_limit = 100_000      # 单次任务Token上限
        self.cost_limit = 2.0           # 单次任务金额上限（USD）
        self.time_limit = 120           # 单次任务时间上限（秒）

    def check(self, usage: dict) -> str:
        if usage["tokens"] > self.token_limit:
            return "TOKEN_BUDGET_EXCEEDED"
        if usage["cost_usd"] > self.cost_limit:
            return "COST_BUDGET_EXCEEDED"
        if usage["elapsed_seconds"] > self.time_limit:
            return "TIME_BUDGET_EXCEEDED"
        return "OK"

# Supervisor层面集成预算检查
def supervisor_with_budget(state, budget_guard):
    budget_status = budget_guard.check(state["usage"])
    if budget_status != "OK":
        # 优雅降级：用已有信息生成部分结果
        return synthesize_partial_result(state)

    # 正常编排逻辑
    return delegate_to_specialist(state)

# Per-Agent评估套件架构
class MultiAgentEvalSuite:
    def __init__(self):
        self.agent_evals = {
            "supervisor": SupervisorEvalSet(
                metrics=["delegation_accuracy", "plan_quality", 
                          "termination_correctness"],
                ground_truth=load_golden_plans()
            ),
            "researcher": AgentEvalSet(
                metrics=["source_relevance", "coverage_score",
                          "hallucination_rate"],
                ground_truth=load_research_benchmarks()
            ),
            "writer": AgentEvalSet(
                metrics=["coherence", "factual_accuracy",
                          "format_compliance"],
                ground_truth=load_writing_samples()
            ),
        }
        self.e2e_eval = EndToEndEval(
            metrics=["task_completion_rate", "total_tokens",
                      "latency_seconds", "cost_usd"]
        )

graph TB
    subgraph "用户层"
        U[用户请求]
    end

    subgraph "编排层"
        S[Supervisor Agent<br/>Claude Opus / GPT-4o<br/>显式路由+OODA循环]
        BG[Budget Guard<br/>Token/成本/时间三重硬顶]
    end

    subgraph "专家层 — 异构模型分配"
        R[Research Agent<br/>Claude Sonnet]
        A[Analysis Agent<br/>Claude Sonnet]
        W[Writer Agent<br/>Claude Sonnet]
        C[Classifier Agent<br/>Claude Haiku]
    end

    subgraph "协议层"
        MCP[MCP Server<br/>工具连接标准化]
        A2A[A2A Protocol<br/>Agent间互操作]
    end

    subgraph "可观测层"
        TR[分布式追踪<br/>Per-Agent Traces]
        EV[评估套件<br/>Per-Agent + E2E]
        LOG[结构化日志<br/>完整决策历史]
    end

    U --> S
    S --> BG
    BG -->|预算OK| R
    BG -->|预算OK| A
    BG -->|预算OK| W
    BG -->|预算超限| PARTIAL[优雅降级<br/>部分结果合成]
    R --> S
    A --> S
    W --> S
    S --> PARTIAL

    R --- MCP
    A --- MCP
    W --- MCP
    R -.->|A2A| A

    S --> TR
    R --> TR
    A --> TR
    W --> TR
    TR --> EV
    TR --> LOG

flowchart TD
    START[收到复杂任务] --> Q1{单Agent能否处理?}
    Q1 -->|是| SINGLE[使用增强型单Agent<br/>+检索+工具+记忆]
    Q1 -->|否| Q2{任务可预定义分解?}

    Q2 -->|是| Q3{阶段间有严格依赖?}
    Q3 -->|是| PIPELINE[Sequential Pipeline<br/>类型化状态通道]
    Q3 -->|否| PARALLEL[Parallel Fan-Out<br/>多专家并行+合成]

    Q2 -->|否| Q4{需要动态规划?}
    Q4 -->|是| Q5{规模大于10个Agent?}
    Q5 -->|否| SUPERVISOR[Supervisor模式<br/>OODA+显式路由]
    Q5 -->|是| HIERARCHY[层级模式<br/>Supervisor of Supervisors]

    Q4 -->|否| Q6{无状态+高吞吐?}
    Q6 -->|是| SWARM[Decentralized Swarm<br/>P2P Handoff]
    Q6 -->|否| SUPERVISOR

    SINGLE --> DEPLOY[部署前检查清单]
    PIPELINE --> DEPLOY
    PARALLEL --> DEPLOY
    SUPERVISOR --> DEPLOY
    HIERARCHY --> DEPLOY
    SWARM --> DEPLOY

    DEPLOY --> CHECK1[✓ 三层预算硬顶]
    DEPLOY --> CHECK2[✓ 分布式追踪]
    DEPLOY --> CHECK3[✓ Per-Agent评估集]
    DEPLOY --> CHECK4[✓ 异构模型路由]
    DEPLOY --> CHECK5[✓ 优雅降级路径]