多Agent黑板架构复兴与云厂商三栈实测

┌─────────────────────────────────────────────────┐
│                 Blackboard（黑板）                 │
│  ┌──────────┬──────────┬──────────┬──────────┐  │
│  │ 假设层H1 │ 假设层H2 │ 假设层H3 │ 证据层E  │  │
│  └──────────┴──────────┴──────────┴──────────┘  │
│                    ↕ 读写                         │
├────────┬────────┬────────┬────────┬──────────────┤
│ Agent A│ Agent B│ Agent C│ Agent D│  Controller  │
│(语法)  │(语义)  │(上下文)│(验证)  │  (触发控制)  │
└────────┴────────┴────────┴────────┴──────────────┘

import asyncio
from typing import Any
from dataclasses import dataclass, field

@dataclass
class BlackboardEntry:
    source: str          # 写入的Agent ID
    layer: str           # 层级（hypothesis/evidence/decision）
    content: Any         # 内容
    confidence: float    # 置信度 0-1
    timestamp: float = 0.0

class SharedBlackboard:
    """共享黑板 - 所有Agent读写的知识空间"""
    def __init__(self):
        self.entries: list[BlackboardEntry] = []
        self._watchers: dict[str, list[asyncio.Event]] = {}

    async def write(self, entry: BlackboardEntry):
        self.entries.append(entry)
        # 通知订阅该层级的Agent
        for evt in self._watchers.get(entry.layer, []):
            evt.set()

    def read(self, layer: str | None = None) -> list[BlackboardEntry]:
        if layer:
            return [e for e in self.entries if e.layer == layer]
        return self.entries

    def subscribe(self, layer: str) -> asyncio.Event:
        event = asyncio.Event()
        self._watchers.setdefault(layer, []).append(event)
        return event

class KnowledgeAgent:
    """知识源Agent - 自主决定何时贡献"""
    def __init__(self, agent_id: str, expertise: str, 
                 bb: SharedBlackboard, trigger_confidence: float = 0.6):
        self.agent_id = agent_id
        self.expertise = expertise
        self.bb = bb
        self.trigger_confidence = trigger_confidence

    async def run(self):
        event = self.bb.subscribe(self.expertise)
        while True:
            await event.wait()
            event.clear()
            # 读取黑板，判断是否需要贡献
            relevant = self.bb.read(layer=self.expertise)
            if any(e.confidence < self.trigger_confidence for e in relevant):
                result = await self._process(relevant)
                await self.bb.write(BlackboardEntry(
                    source=self.agent_id,
                    layer="decision",
                    content=result,
                    confidence=0.85
                ))

    async def _process(self, entries) -> dict:
        # 调用LLM处理黑板内容
        return {"analysis": f"{self.agent_id} processed", "data": entries}

graph TB
    subgraph "Google Multi-Agent Stack"
        U[用户请求] --> OA[Orchestrator Agent<br/>ADK编排]
        OA -->|A2A Protocol| A1[Research Agent]
        OA -->|A2A Protocol| A2[Analysis Agent]
        OA -->|A2A Protocol| A3[Writing Agent]
        A1 -->|MCP| T1[Search Tool]
        A2 -->|MCP| T2[Data Tool]
        A3 -->|MCP| T3[Doc Tool]
    end

    subgraph "External Framework"
        A4[CrewAI Agent] -->|A2A| OA
        A5[LangGraph Agent] -->|A2A| OA
    end

# Google ADK 多Agent示例
from google.adk import Agent, LlmAgent, Runner

researcher = LlmAgent(
    name="researcher",
    model="gemini-2.5-pro",
    instruction="Research the given topic thoroughly",
    tools=[search_tool, web_scraper]
)

writer = LlmAgent(
    name="writer", 
    model="gemini-2.5-pro",
    instruction="Write a comprehensive article based on research"
)

# ADK支持层级式和A2A式编排
coordinator = Agent(
    name="coordinator",
    sub_agents=[researcher, writer],
    # 通过A2A暴露给外部Agent发现
    a2a_enabled=True
)

runner = Runner(agent=coordinator)
result = await runner.run("Write a report on multi-agent systems")

graph TB
    subgraph "AWS Bedrock AgentCore"
        U[用户] --> SA[Supervisor Agent<br/>自动编排]
        SA -->|自动路由| CA1[Collaborator Agent 1]
        SA -->|自动路由| CA2[Collaborator Agent 2]
        SA -->|自动路由| CA3[Collaborator Agent 3]

        SA --> SM[Shared Memory<br/>Session State]
        CA1 --> SM
        CA2 --> SM
        CA3 --> SM

        SA -->|自动生成| Trace[CloudWatch Trace<br/>完整审计链]
    end

# AWS Bedrock 多Agent协作（boto3）
import boto3

client = boto3.client('bedrock-agent')

# 创建Supervisor Agent
supervisor = client.create_agent(
    agentName="enterprise-coordinator",
    agentResourceRoleArn="arn:aws:iam::xxx:role/AgentRole",
    foundationModel="anthropic.claude-sonnet-4-20250514",
    instruction="""You are a coordinator. Route requests to:
    - legal_agent for contract review
    - finance_agent for cost analysis  
    - tech_agent for architecture review"""
)

# 注册Collaborator Agents
client.associate_agent_collaborator(
    agentId=supervisor['agentId'],
    collaboratorName="legal_agent",
    collaboratorAgentArn="arn:aws:bedrock:xxx:agent/legal-reviewer",
    relayConversationHistory=True  # 共享对话历史
)

# AutoGen MagenticOne 多Agent示例
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import MagenticOneGroupChat
from autogen_ext.models.openai import OpenAIChatCompletionClient

model_client = OpenAIChatCompletionClient(model="gpt-4.1")

researcher = AssistantAgent(
    "researcher",
    model_client=model_client,
    tools=[search_web, read_file],
    system_message="You research topics thoroughly"
)

coder = AssistantAgent(
    "coder", 
    model_client=model_client,
    tools=[write_file, execute_code],
    system_message="You write clean, tested code"
)

# MagenticOne自动编排
team = MagenticOneGroupChat(
    [researcher, coder],
    model_client=model_client,
    max_turns=10
)

result = await team.run("Build a web scraper for news articles")

graph TB
    subgraph "三层图记忆架构"
        L1[Layer 1: 实体层<br/>Entity Graph<br/>人物/地点/概念节点]
        L2[Layer 2: 证据层<br/>Evidence Graph<br/>文档片段/数据源]
        L3[Layer 3: 推理层<br/>Reasoning Graph<br/>Agent假设/推导链]

        L1 --> L2
        L2 --> L3
    end

    subgraph "多Agent读写"
        A1[抽取Agent] -->|写| L1
        A1 -->|写| L2
        A2[推理Agent] -->|读| L1
        A2 -->|读| L2
        A2 -->|写| L3
        A3[验证Agent] -->|读| L3
        A3 -->|写| L3
    end

# 宪法审查协议实现
async def constitutional_review(task: str, 
                                 generator: Agent, 
                                 reviewer: Agent) -> str:
    """两阶段：生成+审查，最低成本的共识方案"""

    # 阶段1：生成
    draft = await generator.run(task)

    # 阶段2：审查
    review_prompt = f"""
    Review the following output for:
    1. Factual accuracy
    2. Logical consistency  
    3. Completeness

    Original task: {task}
    Draft output: {draft}

    If issues found, provide corrected version.
    If acceptable, respond with "APPROVED: <original>"
    """
    result = await reviewer.run(review_prompt)

    if result.startswith("APPROVED:"):
        return draft
    return result  # 返回审查修正版

graph TB
    subgraph "用户交互层"
        UI[用户界面] --> GW[API Gateway]
        GW --> AUTH[认证/授权]
    end

    subgraph "编排层"
        AUTH --> ORC[Orchestrator<br/>Supervisor/Blackboard]
        ORC --> RB[路由器<br/>意图识别+Agent选择]
    end

    subgraph "Agent层"
        RB --> A1[研究Agent]
        RB --> A2[分析Agent]
        RB --> A3[写作Agent]
        RB --> A4[验证Agent]
    end

    subgraph "通信协议层"
        A1 <-->|A2A Protocol| A2
        A2 <-->|A2A Protocol| A3
        A3 <-->|A2A Protocol| A4
    end

    subgraph "记忆层"
        A1 --> KG[知识图谱<br/>MemGraphRAG]
        A2 --> KG
        A3 --> KG
        A4 --> KG
        A1 --> VS[向量存储<br/>长期记忆]
        A2 --> VS
    end

    subgraph "工具层"
        A1 -->|MCP| T1[搜索工具]
        A2 -->|MCP| T2[数据工具]
        A3 -->|MCP| T3[文档工具]
        A4 -->|MCP| T4[验证工具]
    end

    subgraph "可观测层"
        ORC --> TRACE[Trace/Monitor<br/>LangSmith/CloudWatch]
        A1 --> TRACE
        KG --> TRACE
    end