参考来源：all-agentic-architectures[1]

---

目录

1. 架构定义
2. 宏观工作流
3. 应用场景
4. 优缺点分析
5. 代码实现详解
6. 评估方法
7. 架构选择思考
8. 总结与核心要点

---

1. 架构定义

Blackboard（黑板） 架构是一种受人类专家协作解决复杂问题启发的智能设计模式。与 Planning 模式的"先规划后执行"不同，Blackboard 系统允许多个专家根据共享状态的演进机会性地贡献知识。

核心理念

特性	描述
共享状态	所有专家围绕一个中央"黑板"读写信息
机会主义	没有固定执行顺序，由 Controller 动态决策
动态调度	下一步行动取决于黑板当前状态而非预设流程

历史溯源：Hearsay-II

Blackboard 模式并非大模型时代的产物，它起源于 20 世纪 70 年代 DARPA 资助的 Hearsay-II 语音识别系统。

• 问题：语音信号极其复杂（声学、音素、词汇、语法多层交织），单一算法无法准确解析。
• 解决方案：将系统分为多个独立的"知识源"（Knowledge Sources）。由于无法确定哪一层级的信息最先被确认（有时听清了一个词能推断出音素，有时反之），系统采用了非线性的黑板结构，允许各层级知识源根据当前确信度动态贡献信息。
• 现代意义：在 LLM 时代，这种模式完美契合了"多模态推理"和"非确定性任务"的需求。

Blackboard vs Planning 核心区别

[!IMPORTANT]
🎯 核心创新

Blackboard 架构的核心创新在于将控制逻辑从专家中解耦：

• Blackboard（黑板）：共享的全局状态存储
• Knowledge Sources（知识源/专家）：独立的领域专家，只知道如何读写黑板
• Controller（控制器）：机会主义调度器，决定下一步激活谁

---

2. 宏观工作流

Blackboard 架构遵循 "状态更新 → 控制器评估 → 专家执行 → 状态更新" 的动态循环流程：

详细步骤说明

步骤	组件	描述	示例
1	初始化黑板	创建共享状态结构，记录用户请求	{"user_request": "分析Nvidia"}
2	控制器评估	读取黑板，判断任务是否完成，选择下一位专家	判断需要先获取新闻
3	选择专家	根据黑板状态动态选择最合适的专家	选择 News Analyst
4	专家执行	专家读取黑板，调用工具，执行任务	调用 web_search 搜索新闻
5	写入黑板	专家将结果写入黑板，供后续专家使用	写入新闻报告
6	综合报告	任务完成后，Report Writer 汇总所有结果	生成最终分析报告

---

3. 应用场景

场景	描述	示例
复杂诊疗	症状模糊，需多科室专家会诊	肿瘤板（Tumor Board）：病理、影像、肿瘤科协作
刑侦破案	线索零碎，拼图式解谜	指纹、DNA、心理侧写专家异步贡献
软件架构设计	模块依赖复杂	安全、数据库、前端架构师围绕设计草稿讨论
多模态融合	音视频文本联合理解	OCR + ASR + LLM 协作完成视频摘要

---

4. 优缺点分析

✅ 优点

优点	说明
极强的动态性	不预设流程，系统能应对"未知的未知"
解耦与模块化	新增专家只需能读写黑板，无需修改其他专家代码
全局上下文	所有决策基于完整历史，减少信息孤岛

❌ 缺点

缺点	说明
控制复杂度高	Controller Prompt 编写难度大，易幻觉或混乱
Token 压力	黑板累积，Token 消耗呈线性/指数增长
不确定性	同输入可能因 Controller 随机性导致不同路径

[!WARNING]
⚠️ 何时不适用

Blackboard 架构不适用于：

• 步骤明确、顺序固定的简单任务
• 对可预测性要求极高的生产环境
• Token 预算极其有限的场景

---

5. 代码实现详解

5.1 环境配置

依赖库安装

pip install -U langchain-openai langchain langgraph python-dotenv rich tavily-python pydantic langchain-core

---

5.2 导入库并设置密钥

import os
from typing import List, Annotated, TypedDict, Optional
from dotenv import load_dotenv

# LangChain 组件
from langchain_openai import ChatOpenAI
from tavily import TavilyClient
from langchain_core.messages import BaseMessage
from pydantic import BaseModel, Field
from langchain_core.tools import tool
from langchain_core.messages import SystemMessage

# LangGraph 组件
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages

# 输出美化
from rich.console import Console
from rich.markdown import Markdown

# --- API 密钥以及追踪设置 ---
load_dotenv()

os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "Agentic Architecture - Blackboard"

# 确保密钥已载入
for key in ["OPENAI_API_KEY", "LANGCHAIN_API_KEY", "TAVILY_API_KEY"]:
    if not os.environ.get(key):
        print(f"{key} not found. Please create a .env file and set it.")

print("Environment variables loaded and tracing is set up.")

运行结果：

Environment variables loaded and tracing is set up.

---

5.3 Blackboard Agent 核心组件

Blackboard Agent 由三个核心组件组成：

核心组件职责

节点	职责	输入	输出
📋 Blackboard	共享状态存储	各专家的输出	全局上下文
🎮 Controller	动态调度决策	黑板当前状态	下一位专家
👤 Agents	执行特定领域任务	黑板 + 用户请求	专业分析结果

5.3.1 定义状态与专家节点

console = Console()

# 定义黑板状态
class BlackboardState(TypedDict):
    user_request: str
    blackboard: List[str]          # 核心共享内存
    available_agents: List[str]    # 专家花名册
    next_agent: Optional[str]      # 决策结果
    agents_called: List[str]       # 历史记录，防死循环

# 定义基础的 Tavily 搜索工具
client = TavilyClient(os.environ["TAVILY_API_KEY"])

@tool
def web_search(query: str) -> str:
    """基于传入的信息在网络上搜索相关信息"""
    result = client.search(query, max_results=3)
    return str(result["results"])

llm = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0.2,
    api_key=os.environ["OPENAI_API_KEY"],
    base_url="https://api.openai.com/v1"
)

print("基础工具与状态定义完成")

运行结果：

基础工具与状态定义完成

[!TIP]
💡 关键设计

• blackboard 字段存储所有专家的输出，作为全局上下文
• agents_called 字段记录已调用的专家，防止死循环
• next_agent 字段由 Controller 决定，驱动下一步执行

---

5.3.2 定义 Controller 节点

class ControllerDecision(BaseModel):
    """Controller 的结构化输出"""
    next_agent: str = Field(description="The name of the next agent to call, or 'FINISH' if done.")
    reasoning: str = Field(description="Brief reasoning for this decision.")

def controller_node(state: BlackboardState):
    """Controller: 根据黑板状态决定下一步"""
    console.print("--- CONTROLLER: Analyzing blackboard... ---")
    
    controller_llm = llm.with_structured_output(ControllerDecision)
    
    blackboard_content = "\n".join(state["blackboard"]) if state["blackboard"] else "Empty"
    
    prompt = f"""You are the Controller of a Blackboard system. Your job is to decide which agent should act next.

**User Request:** {state['user_request']}

**Available Agents:** {state['available_agents']}

**Agents Already Called:** {state['agents_called']}

**Current Blackboard State:**
{blackboard_content}

**Instructions:**
1. If the user's request is fully satisfied by the blackboard content, return 'Report Writer' to synthesize.
2. Otherwise, choose the most appropriate agent from available_agents that has NOT been called yet.
3. Do NOT repeat agents unless absolutely necessary.
"""
    
    decision = controller_llm.invoke(prompt)
    console.print(f"--- CONTROLLER: Decision is to call '{decision.next_agent}'. Reason: {decision.reasoning} ---")
    
    return {"next_agent": decision.next_agent}

print("Controller 节点定义完成")

运行结果：

Controller 节点定义完成

---

5.3.3 定义专家节点

def create_agent_node(agent_name: str, agent_prompt: str):
    """工厂函数：创建专家节点"""
    def agent_node(state: BlackboardState):
        console.print(f"--- (Blackboard) AGENT '{agent_name}' is working... ---")
        
        blackboard_content = "\n".join(state["blackboard"]) if state["blackboard"] else "Empty"
        
        full_prompt = f"""{agent_prompt}

**User Request:** {state['user_request']}

**Current Blackboard:**
{blackboard_content}

Provide your analysis and findings.
"""
        
        # 使用 LLM 并绑定搜索工具
        agent_llm = llm.bind_tools([web_search])
        response = agent_llm.invoke(full_prompt)
        
        # 处理工具调用
        if response.tool_calls:
            console.print("✓ 检测到工具调用，执行搜索...")
            for tool_call in response.tool_calls:
                if tool_call["name"] == "web_search":
                    search_result = web_search.invoke(tool_call["args"]["query"])
                    # 二次调用 LLM 生成报告
                    synthesis_prompt = f"""Based on the search results, provide a structured report.
                    
Search Results: {search_result}

User Request: {state['user_request']}
"""
                    final_response = llm.invoke(synthesis_prompt)
                    report = f"Report from {agent_name}:\n{final_response.content}"
        else:
            report = f"Report from {agent_name}:\n{response.content}"
        
        return {
            "blackboard": state["blackboard"] + [report],
            "agents_called": state["agents_called"] + [agent_name]
        }
    
    return agent_node

# 创建专家节点
news_analyst = create_agent_node("News Analyst", "You are a news analyst. Search for the latest news about the topic.")
technical_analyst = create_agent_node("Technical Analyst", "You are a technical analyst. Analyze technical aspects and specifications.")
financial_analyst = create_agent_node("Financial Analyst", "You are a financial analyst. Analyze financial performance and metrics.")
report_writer = create_agent_node("Report Writer", "You are a report writer. Synthesize all information into a final report.")

print("专家节点创建完毕")

运行结果：

专家节点创建完毕

---

5.3.4 创建 Blackboard Agent 图

def blackboard_router(state: BlackboardState):
    """路由函数：根据 Controller 决策选择下一节点"""
    next_agent = state["next_agent"]
    
    if next_agent == "FINISH":
        return "end"
    elif next_agent == "News Analyst":
        return "news_analyst"
    elif next_agent == "Technical Analyst":
        return "technical_analyst"
    elif next_agent == "Financial Analyst":
        return "financial_analyst"
    elif next_agent == "Report Writer":
        return "report_writer"
    else:
        return "end"

# 构建图
blackboard_graph_builder = StateGraph(BlackboardState)
blackboard_graph_builder.add_node("controller", controller_node)
blackboard_graph_builder.add_node("news_analyst", news_analyst)
blackboard_graph_builder.add_node("technical_analyst", technical_analyst)
blackboard_graph_builder.add_node("financial_analyst", financial_analyst)
blackboard_graph_builder.add_node("report_writer", report_writer)

blackboard_graph_builder.set_entry_point("controller")

# 添加条件边
blackboard_graph_builder.add_conditional_edges(
    "controller",
    blackboard_router,
    {
        "news_analyst": "news_analyst",
        "technical_analyst": "technical_analyst",
        "financial_analyst": "financial_analyst",
        "report_writer": "report_writer",
        "end": END
    }
)

# 所有专家执行完毕后返回 Controller
for agent in ["news_analyst", "technical_analyst", "financial_analyst", "report_writer"]:
    blackboard_graph_builder.add_edge(agent, "controller")

blackboard_app = blackboard_graph_builder.compile()

print("Blackboard Agent compiled successfully.")

运行结果：

Blackboard Agent compiled successfully.

Blackboard Agent 架构图

---

5.4 实际运行测试

使用一个需要动态决策的问题测试 Blackboard Agent：

dynamic_query = """
Find the latest major news about Nvidia. 
Based on the sentiment of that news, conduct either a technical analysis 
(if the news is neutral or positive) or a financial analysis of their 
recent performance (if the news is negative).
"""

initial_input = {
    "user_request": dynamic_query,
    "blackboard": [],
    "available_agents": ["News Analyst", "Technical Analyst", "Financial Analyst", "Report Writer"],
    "next_agent": None,
    "agents_called": []
}

完整执行过程

--- CONTROLLER: Analyzing blackboard... ---
--- CONTROLLER: Decision is to call 'News Analyst'. 
Reason: The first required step is to find the latest major news about Nvidia 
and assess its sentiment; that information determines whether to run the 
Technical Analyst (neutral/positive) or Financial Analyst (negative). ---

--- (Blackboard) AGENT 'News Analyst' is working... ---
✓ 检测到工具调用，执行搜索...
Tavily搜索: Nvidia latest major news January 2026...

--- CONTROLLER: Analyzing blackboard... ---
--- CONTROLLER: Decision is to call 'Technical Analyst'. 
Reason: News Analyst found recent positive/neutral developments 
(BioNeMo adoption and new Rubin chips). Per user instructions, 
positive news requires a technical analysis next. ---

--- (Blackboard) AGENT 'Technical Analyst' is working... ---
✓ 检测到工具调用，执行搜索...
Tavily搜索: NVIDIA Rubin chip BioNeMo 2026 news...

--- CONTROLLER: Analyzing blackboard... ---
--- CONTROLLER: Decision is to call 'Report Writer'. 
Reason: News and technical reports indicate positive/neutral developments. 
A technical analysis has already been produced, so the next step is to 
have the Report Writer synthesize the findings. Financial analysis is 
not needed because the news sentiment is not negative. ---

--- (Blackboard) AGENT 'Report Writer' is working... ---

最终报告（节选）

## 简明报告（基于检索结果）

### 要点
- **时间与场合**：2026-01-12（J.P. Morgan Healthcare Conference）公布合作与平台扩展消息
- **核心内容**：NVIDIA 扩展其 BioNeMo 平台，用于加速 AI 驱动的药物发现
- **重点合作**：NVIDIA 与 Eli Lilly 建立"共同创新"实验室（位于南旧金山）
- **计算架构**：礼来使用 1,000+ NVIDIA Blackwell 芯片，结合 Vera Rubin 架构

### 结论
基于正面市场情绪与技术突破，技术面分析显示其在垂直行业（生命科学）的护城河进一步加深。

[!IMPORTANT]
执行结果分析

系统成功执行了 If-Else 逻辑，但不仅仅是代码层面的判断，而是基于语义理解的判断：

• Controller 理解了 "BioNeMo adoption" = "Positive"
• 自主跳过了 "Financial Analyst" 节点
• 这展示了 Blackboard 架构的动态剪枝能力

---

6. 评估方法 (Evaluation)

使用 LLM-as-a-Judge 对 Blackboard Agent 进行量化评估：

6.1 评估代码

class ProcessEvaluation(BaseModel):
    """Schema for evaluating an agent's problem-solving process."""
    task_completion_score: int = Field(description="Score 1-10 on whether the agent completed the task.")
    process_efficiency_score: int = Field(description="Score 1-10 on the efficiency of the process.")
    dynamic_decision_score: int = Field(description="Score 1-10 on how well the agent made dynamic decisions.")
    justification: str = Field(description="A brief justification for the scores.")

judge_llm = llm.with_structured_output(ProcessEvaluation)

def evaluate_blackboard_process(query: str, final_state: dict):
    blackboard_trace = "\n".join(final_state.get('blackboard', []))
    agents_called = final_state.get('agents_called', [])

    prompt = f"""You are an expert judge of AI agents. Evaluate the Blackboard agent's process.

**User's Task:** {query}

**Agents Called (in order):** {agents_called}

**Blackboard Contents:**
{blackboard_trace}

Evaluate on: task completion, process efficiency, and quality of dynamic decisions.
Use Chinese as output.
"""
    return judge_llm.invoke(prompt)

6.2 评估结果

评估维度	得分
任务完成度	9/10
过程效率	8/10
动态决策质量	9/10

评估说明：

• 任务完成度高：成功获取新闻、判断情绪、执行正确分支
• 过程效率中上：跳过了不需要的 Financial Analyst
• 动态决策优秀：准确理解"正面新闻→技术分析"的条件逻辑

---

7. 架构选择思考

何时使用 Blackboard？

适用场景	原因
不确定性高	无法预知解决问题需要哪些步骤
需要动态决策	下一步取决于前一步的结果
多专家协作	需要不同领域专家异步贡献知识
全局上下文关键	每位专家需要看到完整的历史信息

何时使用其他架构？

架构	适用场景	原因
Planning	步骤已知且固定	预制计划更高效
ReAct	探索性任务	迭代试错更灵活
Chain	简单顺序任务	无需动态决策的开销

架构对比表

判断维度	选择 Blackboard	选择 Planning	选择 ReAct
任务可预测性	❌ 未知	✅ 已知	❌ 未知
需要全局上下文	✅ 是	❌ 否	❌ 否
多专家协作	✅ 是	❌ 否	❌ 否
动态条件分支	✅ 是	❌ 否	✅ 是

---

Blackboard vs Autogen 对比

在了解了 Blackboard 的原理后，我们来看看它与目前流行的多智能体框架（如 Microsoft Autogen）有何不同。

核心拓扑对比

• Blackboard: 星型结构 (Star)。Controller 是绝对核心，所有专家围着黑板转。
• Autogen: 网状/链式结构 (Mesh/Chain)。智能体之间直接对话 (Message Passing)。

详细对比表

维度	Blackboard 模式	Autogen / Chat-based
通信对象	对物说话：Agent 读写中央数据库（黑板）	对人说话：Agent A 发消息给 Agent B
上下文管理	全局共享：任何 Agent 都能看到完整进展	局部/流式：上下文分散在对话历史中
控制流	中央集权 (Controller)：上帝视角调度	去中心化 (Conversational)：轮流发言
适用直觉	拼图：大家各拿一块拼图往桌上凑	开会：大家围在圆桌上你一言我一语

[!TIP]
选型建议

• 如果你需要 "推理 (Reasoning)"，即每一步都要根据全局信息深思熟虑决定下一步，选 Blackboard。
• 如果你需要 "模拟 (Simulation)"，即模拟人类社群交流、谈判、辩论，选 Autogen。

---

8. 总结与核心要点

🎓 核心成果

成果	说明
架构实现	成功实现 Controller + Blackboard + Agents 三元架构
动态决策	展示了根据中间结果动态调整执行路径的能力
评估框架	引入 LLM-as-a-Judge 量化评估动态决策能力

🌟 关键洞察

洞察	描述
状态驱动	Blackboard 的核心是"状态决定行为"，而非预设流程
解耦设计	Controller 与 Agents 完全解耦，易于扩展
Token 代价	全局上下文带来能力，也带来 Token 开销

关键代码速查表

组件	代码
定义黑板状态	class BlackboardState(TypedDict)
Controller 结构化输出	llm.with_structured_output(ControllerDecision)
创建专家节点	create_agent_node(agent_name, agent_prompt)
路由函数	def blackboard_router(state)
条件边	add_conditional_edges("controller", blackboard_router, {...})

三大组件职责

组件	步骤1	步骤2	步骤3
Controller	读取黑板状态	评估任务进度	选择下一位专家
Agent	读取黑板上下文	执行领域任务	写入结果到黑板
Blackboard	存储初始请求	累积中间结果	提供全局上下文

最佳实践

实践	说明
结构化黑板	使用 JSON 结构而非纯文本，便于 Controller 解析
显式推理	要求 Controller 输出 reasoning 字段，降低幻觉
Janitor Agent	对长任务设置清洁工 Agent，定期总结释放 Token
防死循环	记录 agents_called，Prompt 中禁止重复调用

参考链接

[1] all-agentic-architectures: https://github.com/FareedKhan-dev/all-agentic-architectures/tree/main