多Agent协作架构实战：从设计模式到生产落地（2026年完整指南）

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

class PipelineState(TypedDict):
    query: str
    retrieved_docs: str
    analysis: str
    final_report: str

def retrieval_agent(state: PipelineState):
    docs = search_knowledge_base(state["query"])
    return {"retrieved_docs": docs}

def analysis_agent(state: PipelineState):
    result = llm.invoke(f"分析以下内容：{state['retrieved_docs']}")
    return {"analysis": result.content}

def writer_agent(state: PipelineState):
    report = llm.invoke(f"根据分析撰写报告：{state['analysis']}")
    return {"final_report": report.content}

builder = StateGraph(PipelineState)
builder.add_node("retriever", retrieval_agent)
builder.add_node("analyzer", analysis_agent)
builder.add_node("writer", writer_agent)
builder.add_edge(START, "retriever")
builder.add_edge("retriever", "analyzer")
builder.add_edge("analyzer", "writer")
builder.add_edge("writer", END)
pipeline = builder.compile()

from langgraph.types import Send
from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
import operator

class ResearchState(TypedDict):
    query: str
    research_results: Annotated[list, operator.add]
    final_synthesis: str

def supervisor(state: ResearchState):
    subtasks = [
        {"query": state["query"], "source": "academic"},
        {"query": state["query"], "source": "industry"},
        {"query": state["query"], "source": "news"},
    ]
    return [Send("research_worker", task) for task in subtasks]

def research_worker(state: dict):
    result = search_by_source(state["query"], state["source"])
    return {"research_results": [result]}

def synthesizer(state: ResearchState):
    combined = "\n".join(state["research_results"])
    synthesis = llm.invoke(f"综合以下研究结果：{combined}")
    return {"final_synthesis": synthesis.content}

builder = StateGraph(ResearchState)
builder.add_node("research_worker", research_worker)
builder.add_node("synthesizer", synthesizer)
builder.add_conditional_edges(START, supervisor, ["research_worker"])
builder.add_edge("research_worker", "synthesizer")
builder.add_edge("synthesizer", END)
graph = builder.compile()

KEYWORD_ROUTING = {
    "代码": "code_agent",
    "code": "code_agent",
    "搜索": "search_agent",
    "查询": "search_agent",
    "资料": "data_agent",
}

def supervisor_with_fast_path(state):
    query = state["query"].lower()
    for keyword, agent_name in KEYWORD_ROUTING.items():
        if keyword in query:
            return {"next": agent_name}
    routing_prompt = f"""
    用户请求：{state['query']}
    可用 Agent：code_agent, search_agent, data_agent
    请返回最合适的 Agent 名称，只返回名称。
    """
    decision = llm.invoke(routing_prompt)
    return {"next": decision.content.strip()}

import autogen

reviewer_1 = autogen.AssistantAgent(
    name="SecurityReviewer",
    system_message="你是一位安全专家，专注於代码中的安全漏洞。"
)
reviewer_2 = autogen.AssistantAgent(
    name="PerformanceReviewer",
    system_message="你是一位性能专家，专注於代码的效率和資源使用。"
)
human_proxy = autogen.UserProxyAgent(
    name="CodeAuthor",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=2,
    is_termination_msg=lambda x: "APPROVED" in x.get("content", "")
)
groupchat = autogen.GroupChat(
    agents=[human_proxy, reviewer_1, reviewer_2],
    messages=[],
    max_round=6
)
manager = autogen.GroupChatManager(groupchat=groupchat)

from mcp.server import Server
from mcp.types import Tool, TextContent

app = Server("data-agent-mcp")

@app.list_tools()
async def list_tools():
    return [
        Tool(
            name="query_customer_db",
            description="查询客戶数据库，支持按 ID、姓名、Email 检索",
            inputSchema={
                "type": "object",
                "properties": {
                    "field": {"type": "string", "enum": ["id", "name", "email"]},
                    "value": {"type": "string"}
                },
                "required": ["field", "value"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "query_customer_db":
        result = db.query(arguments["field"], arguments["value"])
        return [TextContent(type="text", text=str(result))]

{
  "name": "ResearchAgent",
  "version": "1.0",
  "description": "专业信息检索与摘要 Agent",
  "url": "https://research-agent.internal/a2a",
  "capabilities": {
    "streaming": true,
    "async": true
  },
  "skills": [
    {
      "id": "web_research",
      "name": "网络信息检索",
      "description": "从国际网络检索并摘要最新信息",
      "tags": ["research", "summarization", "web"]
    },
    {
      "id": "academic_search",
      "name": "学术文獻检索",
      "description": "检索 arXiv、Google Scholar 等学术数据库"
    }
  ]
}

import httpx

async def discover_and_delegate(agent_url: str, task: str):
    card_response = await httpx.get(f"{agent_url}/.well-known/agent.json")
    agent_card = card_response.json()
    available_skills = [s["id"] for s in agent_card["skills"]]
    if "web_research" not in available_skills:
        raise ValueError(f"Agent {agent_card['name']} 不支持 web_research 技能")
    payload = {
        "jsonrpc": "2.0",
        "method": "message/send",
        "id": "task-001",
        "params": {
            "message": {
                "role": "user",
                "parts": [{"type": "text", "text": task}]
            }
        }
    }
    response = await httpx.post(agent_card["url"], json=payload)
    return response.json()

from opentelemetry import trace
import uuid

tracer = trace.get_tracer("multi-agent-system")

def traced_agent_call(agent_name: str, task: dict, correlation_id: str = None):
    if not correlation_id:
        correlation_id = str(uuid.uuid4())
    with tracer.start_as_current_span(f"agent.{agent_name}") as span:
        span.set_attribute("agent.name", agent_name)
        span.set_attribute("correlation.id", correlation_id)
        span.set_attribute("task.type", task.get("type", "unknown"))
        try:
            result = agent_registry[agent_name].run(task)
            span.set_attribute("agent.tokens_used", result.get("tokens", 0))
            span.set_attribute("agent.status", "success")
            return result
        except Exception as e:
            span.set_attribute("agent.status", "error")
            span.set_attribute("error.message", str(e))
            raise

MONITORING_METRICS = {
    "task_success_rate": "端到端任务完成率（目标：>85%）",
    "e2e_latency_p95": "P95 端到端延迟（目标：<30s）",
    "total_cost_per_task": "每次任务平均 Token 成本",
    "agent_error_rate": "各 Agent 错误率（目标：<5%）",
    "agent_retry_count": "重试次数（高重试 = 需要调查）",
    "tool_call_budget_usage": "工具调用次数/预算比",
    "output_quality_score": "输出质量评分",
    "goal_alignment_score": "目标一致性评分",
    "hallucination_rate": "幻觉检测率",
}

import json

def evaluate_agent_output(original_task: str, agent_output: str) -> dict:
    evaluation_prompt = f"""
    你是一位严格的质量评审专家。请评估以下 AI Agent 的输出质量。
    原始任务：{original_task}
    Agent 输出：{agent_output}
    请从完成度、准确性、相关性、幻觉检测四个维度评分（1-5 分）。
    以 JSON 格式返回：
    {{"completeness": x, "accuracy": x, "relevance": x, "hallucination_detected": true/false, "comments": "..."}}
    """
    evaluation = llm.invoke(evaluation_prompt)
    return json.loads(evaluation.content)

你的任务是否有明确的线性依赖步骤？
├─ 是 → 子任务是否可以并行执行？
│        ├─ 否 → 【顺序流水线】
│        └─ 是 → 【并行扇出 + 顺序流水线 混合】
│
└─ 否 → 是否有一个 Agent 具有决策权威？
         ├─ 是 → 规模是否足够大需要子团队？
         │        ├─ 否 → 【Supervisor-Worker 层级模式】
         │        └─ 是 → 【层级式（Supervisors of Supervisors）】
         │
         └─ 否 → 任务是否是长时间异步的？
                  ├─ 是 → 【黑板架构】
                  └─ 否 → Agent 数量是否 ≤ 5？
                           ├─ 是 → 【Swarm（注意设定终止条件）】
                           └─ 否 → 【考虑重新拆分为层级模式】