多智能体协作系统六大编排模式深度解析

# LangGraph监督者模式核心代码
from langgraph.graph import StateGraph, END

# 定义共享状态
class AgentState(TypedDict):
    messages: list[dict]
    next_agent: str
    task_result: dict

# 监督者节点：决定下一个执行哪个Agent
def supervisor_node(state: AgentState) -> AgentState:
    last_message = state["messages"][-1]
    decision = llm.invoke(f"根据以下消息决定路由: {last_message}")
    return {"next_agent": decision.agent_name}

# 构建图
workflow = StateGraph(AgentState)
workflow.add_node("supervisor", supervisor_node)
workflow.add_node("researcher", researcher_agent)
workflow.add_node("coder", coder_agent)
workflow.add_node("reviewer", reviewer_agent)

# 条件边：监督者决定路由
workflow.add_conditional_edges(
    "supervisor",
    lambda state: state["next_agent"],
    {"researcher": "researcher", "coder": "coder", "reviewer": "reviewer"}
)

# CrewAI管道模式
from crewai import Agent, Task, Crew, Process

researcher = Agent(
    role="市场研究员",
    goal="收集并整理行业数据",
    backstory="你是一位资深市场分析专家",
    llm="gpt-4o"
)

analyst = Agent(
    role="数据分析师", 
    goal="从数据中提炼洞察",
    backstory="你擅长从复杂数据中发现趋势"
)

writer = Agent(
    role="内容撰写师",
    goal="将分析结果写成报告",
    backstory="你是专业的商业报告撰写人"
)

# 顺序管道：研究 → 分析 → 撰写
crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, writing_task],
    process=Process.sequential  # 管道模式
)

# OpenAI Agents SDK的Swarm式交接
from agents import Agent, Runner

triage_agent = Agent(
    name="分流Agent",
    instructions="你负责理解用户需求，将对话交给专业Agent",
    handoffs=[  # 定义可交接的目标
        {"target": "billing_agent", "condition": "用户提到账单或支付"},
        {"target": "tech_agent", "condition": "用户报告技术问题"},
    ]
)

billing_agent = Agent(
    name="账单Agent",
    instructions="处理账单查询和退款请求",
    handoffs=[{"target": "triage_agent"}]  # 可以交回
)

# Runner自动处理交接链
result = await Runner.run(triage_agent, messages=conversation)

# AutoGen层级模式
from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager

# 顶层：策略规划
planner = AssistantAgent(
    name="策略总监",
    system_message="你负责分解项目目标，分配给部门负责人"
)

# 中层：领域协调
code_lead = AssistantAgent(
    name="开发负责人", 
    system_message="管理开发任务，分配给具体开发者"
)
test_lead = AssistantAgent(
    name="测试负责人",
    system_message="管理测试任务，确保代码质量"
)

# 底层：执行
developer = AssistantAgent(name="开发者")
tester = AssistantAgent(name="测试工程师")

# 群组聊天作为层级协调器
groupchat = GroupChat(
    agents=[planner, code_lead, test_lead, developer, tester],
    messages=[],
    max_round=20
)
manager = GroupChatManager(groupchat=groupchat)

┌─────────────────────────────────────┐
│           Blackboard                │
│  ┌─────────┐ ┌─────────┐ ┌───────┐ │
│  │ Facts   │ │Hypothesis│ │ Plan  │ │
│  │ Layer   │ │  Layer   │ │ Layer │ │
│  └─────────┘ └─────────┘ └───────┘ │
│  ┌─────────────────────────────────┐│
│  │     Solution Layer              ││
│  └─────────────────────────────────┘│
└──────────┬──────────┬───────────────┘
           │          │
     ┌─────┴──┐  ┌───┴─────┐
     │Agent A │  │Agent B  │
     │(观察者)│  │(推理者) │
     └────────┘  └─────────┘

# 黑板模式核心实现
class Blackboard:
    def __init__(self):
        self.facts = []       # 事实层
        self.hypotheses = []  # 假设层
        self.plans = []       # 计划层
        self.solutions = []   # 方案层

    def write(self, layer: str, data: dict, agent_id: str):
        entry = {"data": data, "source": agent_id, "timestamp": time.time()}
        getattr(self, layer).append(entry)
        self._notify_subscribers(layer)

    def read(self, layer: str, filter_fn=None) -> list:
        entries = getattr(self, layer)
        return [e for e in entries if filter_fn(e)] if filter_fn else entries

class KnowledgeAgent:
    def __init__(self, agent_id: str, bb: Blackboard):
        self.id = agent_id
        self.bb = bb

    def observe_and_act(self):
        # 读取黑板，决定是否写入新信息
        current_facts = self.bb.read("facts")
        new_hypothesis = self.llm.generate(current_facts)
        self.bb.write("hypotheses", new_hypothesis, self.id)

// Agent Card 示例
{
  "name": "medical-diagnosis-agent",
  "description": "专业医疗诊断Agent，支持常见疾病初步诊断",
  "capabilities": ["symptom_analysis", "drug_interaction_check"],
  "endpoint": "https://agents.hospital.com/medical/v1",
  "authentication": {"type": "oauth2", "scope": "medical:read"},
  "protocols": ["a2a/v1", "mcp/v1"]
}

# A2A协议下的Agent间通信
from a2a import A2AClient, AgentCard

# 发现其他Agent
client = A2AClient()
available_agents = await client.discover(capability="data_analysis")

# 选择最合适的Agent并通信
selected = await client.select_best(available_agents, criteria="latency")
response = await client.send_message(
    target=selected,
    message={"type": "task_request", "data": analysis_request},
    protocol="a2a/v1"
)

# 结构化交接摘要
def generate_handoff_summary(agent_state: dict) -> dict:
    return {
        "completed_steps": agent_state["finished_tasks"],
        "current_findings": agent_state["key_insights"],
        "open_questions": agent_state["unresolved"],
        "context_for_next": agent_state["next_agent_needs"],
        "confidence_score": agent_state["confidence"]
    }