黑板复兴：多Agent共享知识空间六栈实测

# 消息传递模式的典型痛点
class MessagePassingOrchestrator:
    def run(self, agents, task):
        context = task
        for agent in agents:  # 线性链式传递
            result = agent.run(context)
            context = self._serialize(result)  # 每次序列化都损失信息
            if len(context) > self.max_context:
                context = context[:self.max_context]  # 截断！
        return context

# LbMAS的核心循环
class BlackboardController:
    def __init__(self, blackboard, knowledge_sources):
        self.bb = blackboard  # 共享状态空间
        self.ks_list = knowledge_sources  # 多个LLM Agent

    def run(self, problem):
        self.bb.write("problem_layer", problem)
        while not self.bb.is_solved():
            # 控制器选择最合适的知识源
            best_ks = self._select_source(self.bb.get_state())
            # Agent读取黑板、执行、写回黑板
            result = best_ks.execute(self.bb.read_relevant(best_ks.scope))
            self.bb.write(best_ks.output_layer, result)
        return self.bb.get_solution()

# SE-Blackboard的分层黑板结构
SE_LAYERS = {
    "requirements": "PRD和用户故事",
    "architecture": "模块设计和接口定义",
    "code": "源代码和配置文件",
    "tests": "测试用例和覆盖率",
    "review": "代码审查意见和Bug报告",
    "fixes": "修复记录和回归测试结果"
}

class SEAgent:
    def __init__(self, role, read_layers, write_layer):
        self.role = role
        self.read_layers = read_layers  # 可读的黑板层
        self.write_layer = write_layer  # 写入的黑板层

    def execute(self, blackboard):
        # 测试Agent可以同时读取需求和代码
        context = {layer: blackboard.read(layer) for layer in self.read_layers}
        output = self.llm.generate(context)
        blackboard.write(self.write_layer, output)

# blackboard-core实战：5分钟搭建共享知识空间
from blackboard_core import Blackboard, Agent, Controller

bb = Blackboard()
bb.create_section("research")   # 研究Agent的工作区
bb.create_section("writing")    # 写作Agent的工作区
bb.create_section("review")     # 审核Agent的工作区

research_agent = Agent(
    name="researcher",
    read_sections=["research"],
    write_sections=["research"],
    llm=your_llm_client
)

writer_agent = Agent(
    name="writer",
    read_sections=["research", "review"],  # 读研究+审核反馈
    write_sections=["writing"]
)

reviewer_agent = Agent(
    name="reviewer",
    read_sections=["writing", "research"],  # 同时看文章和原始资料
    write_sections=["review"]
)

controller = Controller(
    blackboard=bb,
    agents=[research_agent, writer_agent, reviewer_agent],
    strategy="priority_based"  # 优先级调度
)

result = controller.run("写一篇关于量子计算的科普文章")

// MCP Blackboard的配置示例
{
  "mcpServers": {
    "blackboard": {
      "command": "npx",
      "args": ["-y", "mcp-blackboard"],
      "env": {
        "BLACKBOARD_PERSISTENCE": "redis",
        "BLACKBOARD_TTL": "3600"
      }
    }
  }
}

# 任何支持MCP的Agent都可以读写同一块黑板
# Claude Agent
mcp_tool_call("blackboard", "write", {
    "section": "analysis",
    "key": "market_data",
    "value": {"trend": "bullish", "confidence": 0.87}
})

# GPT Agent（通过MCP协议访问同一块黑板）
mcp_tool_call("blackboard", "read", {
    "section": "analysis",
    "key": "market_data"
})
# → {"trend": "bullish", "confidence": 0.87}

// Agent写入黑板的方式：创建知识图谱节点
CREATE (a:Analysis {agent: "market_analyst", 
                     content: "Q3增长超预期",
                     confidence: 0.92})
CREATE (s:Signal {type: "buy_signal", strength: "strong"})
CREATE (a)-[:PRODUCES]->(s)

// 另一个Agent读取黑板：通过图谱查询获取关联知识
MATCH (a:Analysis)-[:PRODUCES]->(s:Signal {type: "buy_signal"})
WHERE a.confidence > 0.85
RETURN a.content, s.strength

# Neo4j + Google ADK集成示例（2026 Google Codelab）
from neo4j_agent_memory import GraphMemory

shared_memory = GraphMemory(
    uri="neo4j://localhost:7687",
    agent_namespace="trading_team"
)

# 多Agent共享同一个图谱记忆
analyst_agent = Agent(memory=shared_memory, role="analyst")
trader_agent = Agent(memory=shared_memory, role="trader")
risk_agent = Agent(memory=shared_memory, role="risk_manager")

# 分析师Agent写入洞察
analyst_agent.memory.store(
    entity="AAPL",
    relation="revenue_growth",
    value="15.3%",
    metadata={"quarter": "Q3", "source": "earnings_report"}
)

# 风控Agent通过图谱查询理解上下文
risk_context = risk_agent.memory.query(
    "MATCH (e:Entity {name: $name})-[r]->(f) RETURN r, f",
    params={"name": "AAPL"}
)

┌─────────────────────────────────────────┐
│          GAM双Agent架构                  │
│                                          │
│  ┌──────────┐      ┌──────────────┐     │
│  │  Worker   │      │ Memory Agent │     │
│  │  Agent    │      │ (存取决策)    │     │
│  └────┬─────┘      └──────┬───────┘     │
│       │                    │             │
│       │   ┌────────────┐  │             │
│       └──→│   AFS      │←─┘             │
│           │ (Agent File │                │
│           │  System)    │                │
│           │ ┌────────┐ │                │
│           │ │短期记忆 │ │                │
│           │ ├────────┤ │                │
│           │ │长期记忆 │ │                │
│           │ ├────────┤ │                │
│           │ │推理链   │ │                │
│           │ └────────┘ │                │
│           └────────────┘                │
└─────────────────────────────────────────┘

graph TB
    subgraph "控制器层 Controller"
        CTRL[调度控制器<br/>Priority/RoundRobin/LLM-Based]
    end

    subgraph "黑板层 Blackboard State Space"
        BB1[问题层<br/>Problem Layer]
        BB2[分析层<br/>Analysis Layer]
        BB3[方案层<br/>Solution Layer]
        BB4[验证层<br/>Verification Layer]
    end

    subgraph "知识源层 Knowledge Sources / Agents"
        A1[研究Agent<br/>Read: 问题层<br/>Write: 分析层]
        A2[设计Agent<br/>Read: 问题层+分析层<br/>Write: 方案层]
        A3[审核Agent<br/>Read: 方案层+问题层<br/>Write: 验证层]
        A4[修复Agent<br/>Read: 验证层+方案层<br/>Write: 方案层]
    end

    subgraph "持久化层"
        P1[内存<br/>最快/易失]
        P2[Redis<br/>低延迟/可扩展]
        P3[图数据库<br/>关系推理/溯源]
    end

    CTRL -->|选择激活| A1
    CTRL -->|选择激活| A2
    CTRL -->|选择激活| A3
    CTRL -->|选择激活| A4

    A1 <-->|读写| BB1
    A1 <-->|读写| BB2
    A2 <-->|读写| BB2
    A2 <-->|读写| BB3
    A3 <-->|读写| BB3
    A3 <-->|读写| BB4
    A4 <-->|读写| BB4
    A4 <-->|读写| BB3

    BB1 & BB2 & BB3 & BB4 <--> P1
    BB1 & BB2 & BB3 & BB4 <--> P2
    BB1 & BB2 & BB3 & BB4 <--> P3

# 混合架构示例
class HybridOrchestrator:
    def __init__(self):
        self.blackboard = Blackboard()  # 共享状态
        self.message_bus = EventBus()    # 事件触发

    def run(self, task):
        self.blackboard.write("task", task)

        # 消息触发：通知研究Agent开始工作
        self.message_bus.publish("task_started", {"task_id": task.id})

        # Agent通过消息获取触发信号，通过黑板读写数据
        # Agent内部：收到消息→读黑板→处理→写黑板→发消息通知完成

你需要共享知识空间吗？
├── 否 → 用消息传递（LangGraph/CrewAI即可）
└── 是 → 你的Agent是同框架还是异构？
    ├── 同框架 → blackboard-core（最快上手）
    └── 异构 → MCP Blackboard（协议级互通）
        → 你的知识有复杂关系吗？
            ├── 否 → Redis/内存黑板
            └── 是 → Neo4j图谱黑板

# 15分钟黑板系统：无需任何外部依赖
import json
from typing import Dict, List, Any

class SimpleBlackboard:
    def __init__(self):
        self._state: Dict[str, Dict] = {}
        self._history: List[Dict] = []

    def create_section(self, name: str):
        self._state[name] = {}

    def write(self, section: str, key: str, value: Any):
        if section not in self._state:
            self.create_section(section)
        self._state[section][key] = value
        self._history.append({
            "action": "write", "section": section,
            "key": key, "value": value
        })

    def read(self, section: str, key: str = None) -> Any:
        if key is None:
            return self._state.get(section, {})
        return self._state.get(section, {}).get(key)

    def read_all(self) -> Dict:
        return json.loads(json.dumps(self._state))

# 使用示例
bb = SimpleBlackboard()
bb.write("research", "findings", "量子纠缠可实现超远距离通信模拟")
bb.write("research", "sources", ["Nature 2026", "Physical Review Letters"])

# 写作Agent读取研究层的发现
research = bb.read("research")