智能体架构 -- 黑板系统，让专家团队动态协作

你有没有遇到过这种情况？你交给一个AI团队一个任务：查一下某公司的最新新闻，如果新闻是积极的，就做技术分析；如果是消极的，就做财务分析。结果，这个团队老老实实地把所有分析都做了一遍，完全没管你的“如果……那么……”指令。

这就是传统顺序多智能体系统（Sequential Multi-Agent System） 的典型问题：它就像工厂流水线，无论来料是什么，都必须经过每一道工序，毫无灵活性可言。在现实工作中，我们需要的是一群能够围在一起、根据情况随时调整分工的专家，而不是只会排队干活的“机器人”。

今天，我们来聊聊一种能真正实现“动态协作”的AI架构——黑板系统（Blackboard System）。它的灵感来源于人类专家围绕一块实体黑板共同解决复杂问题的场景。我们将手把手搭建一个黑板系统，并让它和僵化的顺序系统正面交锋，看看在需要条件判断的任务中，谁才是真正的王者。

推荐阅读：

智能体架构 -- 反思模式，让AI自己当代码审查员

智能体架构 -- 工具使用，秒变真·智能助手

智能体架构 -- ReAct，让AI学会“三思而后行”

智能体架构 -- 反思性元认知智能体，让AI学会“三省吾身”

智能体架构 -- 规划架构，让你的智能体效率翻倍！

智能体架构 -- 多智能体系统，组专家团队

智能体架构 -- 规划器 → 执行器 → 验证器，给AI配个“质检员”

什么是黑板系统？从“流水线”到“圆桌会议”

想象一个场景：你们公司要做一个复杂的技术决策，你召集了产品、开发、测试、运维几个专家开会。会上，每个人都可以在白板上写下自己的观点，其他人看到后，可以补充、质疑，最终形成一个大家公认的结论。这个白板，就是“黑板”。

在AI的世界里，黑板系统就是这样一个共享内存 + 动态调度的协作框架：

• 一块黑板：一个中央共享的数据存储区，存放用户请求、中间发现和部分解决方案。所有专家都能在这里“写”下自己的成果，也能“读”取别人的成果。
• 一群专家智能体：每个智能体都有自己的专长，比如新闻分析师、技术分析师、财务分析师、报告撰写员。它们不按固定顺序工作，而是时刻准备着，只要黑板上的内容触发自己的“专长”，就主动贡献。
• 一个控制器：这是整个系统的“会议主持人”。它持续观察黑板上的最新进展，根据当前状态和最终目标，动态决定下一步应该激活哪个专家。它会问自己：“现在黑板上有新闻了吗？那接下来是该做技术分析还是财务分析？……哦，新闻是积极的，那叫技术分析师进来。”

这个流程是机会主义的——专家们不是被预先排好队，而是由控制器根据“当前最佳下一步”来召唤。整个系统的工作流是涌现式的，而非硬编码的。

黑板系统到底强在哪？

适用场景

• 复杂、结构不良的问题：比如复杂故障诊断、科研探索、商业决策——这些任务的路径无法事先规划，需要根据中间结果动态调整。
• 多模态信息处理：协调处理文本、图像、代码等多种数据类型的智能体，它们都可以将发现发布到共享黑板上，互相参考。
• 动态意义构建：需要综合来自多个异步来源信息的情况，比如舆情监控、金融事件分析。

优势与劣势

优势：

• 灵活性与适应性：工作流是根据问题涌现的，不是硬编码的，所以系统能应对变化。
• 模块化：你可以随时添加或移除专家智能体，而不必重新设计整个系统。

劣势：

• 控制器设计有门槛：整个系统的智能水平很大程度取决于控制器的“脑力”。一个简单的控制器可能导致低效或循环行为。
• 调试可能有点难：由于工作流是非线性的，追踪问题可能比顺序流程更耗时。

动手实战：从僵化的顺序系统到智能的黑板系统

我们分三步走：先搭建一个“流水线”式的顺序多智能体系统作为基线；然后重构出一个真正的黑板系统；最后让它们在同一道复杂的题目上PK。

阶段 0：准备工作

我们使用 Nebius 作为LLM服务商（类似国内阿里云百炼、智谱AI等平台），Tavily 作为网络搜索工具，LangSmith 用于追踪调试。当然，你也可以根据自己的需求替换成国内的模型服务商。

# 安装依赖库
# !pip install -q -U langchain-nebius langchain langgraph rich python-dotenv langchain-tavily

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

# LangChain 组件
from langchain_nebius import ChatNebius
from langchain_tavily import TavilySearch
from langchain_core.messages import BaseMessage, SystemMessage, HumanMessage
from pydantic import BaseModel, Field
from langchain_core.prompts import ChatPromptTemplate

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

# 美观打印
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 (Nebius)"

for key in ["NEBIUS_API_KEY", "LANGCHAIN_API_KEY", "TAVILY_API_KEY"]:
ifnot os.environ.get(key):
        print(f"⚠️ 未找到{key}。请在.env文件中设置。")

print("✅ 环境变量已加载，追踪已设置。")
console = Console()

阶段 1：基线——一个顺序多智能体系统（修正版）

我们首先搭建一个标准的“流水线”系统：新闻分析师 → 技术分析师 → 财务分析师 → 报告撰写员，严格按照这个顺序执行。注意，我们特意让每个分析师都能访问前一步的成果，这样顺序系统也能产出完整报告，但它的僵化问题依然存在。

llm = ChatNebius(model="mistralai/Mixtral-8x22B-Instruct-v0.1", temperature=0)
search_tool = TavilySearch(max_results=2)

classSequentialState(TypedDict):
    user_request: str
    news_report: Optional[str]
    technical_report: Optional[str]
    financial_report: Optional[str]
    final_report: Optional[str]

defnews_analyst_node_seq(state: SequentialState):
    console.print("--- (顺序) 调用新闻分析师 ---")
    prompt = f"你是一位专家级新闻分析师。请查找用户请求中主题的最新主要新闻，并提供简洁摘要。\n\n用户请求：{state['user_request']}"
    agent = llm.bind_tools([search_tool])
    result = agent.invoke(prompt)
return {"news_report": result.content}

deftechnical_analyst_node_seq(state: SequentialState):
    console.print("--- (顺序) 调用技术分析师 ---")
# 技术分析师依赖新闻报告
    prompt = f"你是一位专家级技术分析师。基于以下新闻报道，对该公司的股票进行技术分析。\n\n新闻报道：\n{state['news_report']}"
    agent = llm.bind_tools([search_tool])
    result = agent.invoke(prompt)
return {"technical_report": result.content}

deffinancial_analyst_node_seq(state: SequentialState):
    console.print("--- (顺序) 调用财务分析师 ---")
# 财务分析师也依赖新闻报告
    prompt = f"你是一位专家级财务分析师。基于以下新闻报道，分析该公司近期的财务表现。\n\n新闻报道：\n{state['news_report']}"
    agent = llm.bind_tools([search_tool])
    result = agent.invoke(prompt)
return {"financial_report": result.content}

defreport_writer_node_seq(state: SequentialState):
    console.print("--- (顺序) 调用报告撰写员 ---")
    prompt = f"""你是一位专业的报告撰写员。综合来自新闻、技术和财务分析师的信息，形成一份连贯的报告，直接回应用户的原始请求。

用户请求：{state['user_request']}

以下是需要整合的报告：
---
新闻报道：{state['news_report']}
---
技术报告：{state['technical_report']}
---
财务报告：{state['financial_report']}
"""
    report = llm.invoke(prompt).content
return {"final_report": report}

# 构建顺序图
seq_graph_builder = StateGraph(SequentialState)
seq_graph_builder.add_node("news", news_analyst_node_seq)
seq_graph_builder.add_node("tech", technical_analyst_node_seq)
seq_graph_builder.add_node("finance", financial_analyst_node_seq)
seq_graph_builder.add_node("writer", report_writer_node_seq)

seq_graph_builder.set_entry_point("news")
seq_graph_builder.add_edge("news", "tech")
seq_graph_builder.add_edge("tech", "finance")
seq_graph_builder.add_edge("finance", "writer")
seq_graph_builder.add_edge("writer", END)

sequential_app = seq_graph_builder.compile()
print("✅ 修正后的顺序多智能体系统编译成功。")

测试顺序系统：一个带有条件逻辑的查询

我们用一个需要条件判断的任务来测试它：

dynamic_query = "查找有关英伟达的最新重大新闻。根据该新闻的情绪，如果新闻是中性或积极的，则进行技术分析；如果新闻是消极的，则对其近期表现进行财务分析。"

console.print(f"\n🎯 [bold yellow]在动态查询上测试顺序智能体：[/bold yellow]\n'{dynamic_query}'\n")

final_seq_output = sequential_app.invoke({"user_request": dynamic_query})

console.print("\n--- [bold red]顺序智能体的最终报告[/bold red] ---")
console.print(Markdown(final_seq_output['final_report']))

运行结果分析：顺序系统确实生成了完整的报告，但它的执行轨迹是“新闻 → 技术 → 财务”。它同时做了技术分析和财务分析，完全忽略了用户“要么……要么……”的条件。这就是僵化流水线的问题——无论新闻是积极还是消极，它都会把所有分析跑一遍，既浪费资源，又可能产生冗余信息。

阶段 2：进阶方法——黑板系统（修正版）

现在，我们重构出一个真正的黑板系统。关键点在于：

1. 共享黑板：用列表存储所有智能体发布的报告。
2. 智能控制器：它会读取黑板内容，基于当前进展和用户请求，动态决定下一个调用哪个专家。
3. 动态路由：专家执行后，控制权总是回到控制器，由它决定下一步。

# 黑板状态：包含用户请求、黑板内容、可用专家列表、控制器决策
classBlackboardState(TypedDict):
    user_request: str
    blackboard: List[str]          # 中央黑板，智能体将报告追加到这里
    available_agents: List[str]    # 可选专家列表
    next_agent: Optional[str]      # 控制器选出的下一个专家

# 控制器决策的结构化输出模型
classControllerDecision(BaseModel):
    next_agent: str = Field(description="下一个要调用的智能体名称。必须是 ['新闻分析师', '技术分析师', '财务分析师', '报告撰写员'] 之一，或 '完成'。")
    reasoning: str = Field(description="选择下一个智能体的简要理由。")

# 工厂函数：为黑板系统创建专家节点
defcreate_blackboard_specialist(persona: str, agent_name: str):
    system_prompt = f"""你是一位专家级智能体：{persona}。
你的任务是通过执行你的特定功能来为更大的目标做出贡献。
阅读初始用户请求和当前黑板以获取上下文。
使用你的工具查找所需信息。
最后，将你的简洁 Markdown 报告发布回黑板。你的报告应署上你的名字 '{agent_name}'。
"""
    prompt_template = ChatPromptTemplate.from_messages([
        ("system", system_prompt),
        ("human", "用户请求：{user_request}\n\n黑板（先前报告）：\n{blackboard_str}")
    ])
    agent = prompt_template | llm.bind_tools([search_tool])

defspecialist_node(state: BlackboardState):
        console.print(f"--- (黑板) 智能体 '{agent_name}' 正在工作... ---")
        blackboard_str = "\n---\n".join(state["blackboard"])
        result = agent.invoke({"user_request": state["user_request"], "blackboard_str": blackboard_str})
        report = f"**来自 {agent_name} 的报告：**\n{result.content}"
return {"blackboard": state["blackboard"] + [report]}
return specialist_node

# 创建黑板专家
news_analyst_bb = create_blackboard_specialist("新闻分析师", "新闻分析师")
technical_analyst_bb = create_blackboard_specialist("技术分析师", "技术分析师")
financial_analyst_bb = create_blackboard_specialist("财务分析师", "财务分析师")
report_writer_bb = create_blackboard_specialist("负责从黑板综合出最终答案的报告撰写员", "报告撰写员")

# 控制器节点——这是整个系统的“大脑”
defcontroller_node(state: BlackboardState):
    console.print("--- 控制器：分析黑板... ---")
    controller_llm = llm.with_structured_output(ControllerDecision)
    blackboard_content = "\n\n".join(state['blackboard'])
    agent_list = state['available_agents']

    prompt = f"""你是多智能体系统的中央控制器。你的工作是分析共享黑板和原始用户请求，以决定接下来应该运行哪个专家智能体。

**原始用户请求：**
{state['user_request']}

**当前黑板内容：**
---
{blackboard_content if blackboard_content else"黑板当前为空。"}
---

**可用的专家智能体：**
{', '.join(agent_list)}

**你的任务：**
1.  仔细阅读用户请求和当前黑板内容。
2.  确定 *下一个逻辑步骤* 是什么，以便更接近完整答案。
3.  从可用智能体列表中选择执行该步骤的最佳单一智能体。
4.  如果用户的请求已完全满足并且已撰写最终报告，请选择“完成”。在“报告撰写员”提供最终综合答案之前，不要完成。

请以所需格式提供你的决策。
"""
    decision_result = controller_llm.invoke(prompt)
    console.print(f"--- 控制器：决定调用 '{decision_result.next_agent}'。理由：{decision_result.reasoning} ---")
return {"next_agent": decision_result.next_agent}

print("✅ 黑板组件和控制器节点定义完成。")

构建黑板图

现在，我们将节点组装成一个动态图。流程是：控制器 → 选择专家 → 专家执行 → 返回控制器，直到控制器选择“完成”。

bb_graph_builder = StateGraph(BlackboardState)

bb_graph_builder.add_node("Controller", controller_node)
bb_graph_builder.add_node("新闻分析师", news_analyst_bb)
bb_graph_builder.add_node("技术分析师", technical_analyst_bb)
bb_graph_builder.add_node("财务分析师", financial_analyst_bb)
bb_graph_builder.add_node("报告撰写员", report_writer_bb)

bb_graph_builder.set_entry_point("Controller")

defroute_to_agent(state: BlackboardState):
return state["next_agent"]

# 条件边：控制器根据决策路由到专家或结束
bb_graph_builder.add_conditional_edges(
"Controller",
    route_to_agent,
    {
"新闻分析师": "新闻分析师",
"技术分析师": "技术分析师",
"财务分析师": "财务分析师",
"报告撰写员": "报告撰写员",
"完成": END
    }
)

# 任何专家执行后，控制权回到控制器
bb_graph_builder.add_edge("新闻分析师", "Controller")
bb_graph_builder.add_edge("技术分析师", "Controller")
bb_graph_builder.add_edge("财务分析师", "Controller")
bb_graph_builder.add_edge("报告撰写员", "Controller")

blackboard_app = bb_graph_builder.compile()
print("✅ 黑板系统编译成功。")

阶段 3：直接对比

让黑板系统执行同样的条件任务，观察它的动态流程。

console.print(f"\n🎯 [bold green]在相同的动态查询上测试黑板系统：[/bold green]\n'{dynamic_query}'\n")

agent_list = ["新闻分析师", "技术分析师", "财务分析师", "报告撰写员"]
initial_bb_input = {"user_request": dynamic_query, "blackboard": [], "available_agents": agent_list}

final_bb_output = None
for chunk in blackboard_app.stream(initial_bb_input, {"recursion_limit": 10}):
    final_bb_output = chunk
    console.print("\n--- [bold purple]当前黑板状态[/bold purple] ---")
for i, report in enumerate(final_bb_output.get('blackboard', [])):
        console.print(f"--- 报告 {i+1} ---")
        console.print(Markdown(report))
    console.print("\n")

console.print("\n--- [bold green]黑板系统的最终报告[/bold green] ---")
final_report_content = final_bb_output['blackboard'][-1]
console.print(Markdown(final_report_content))

运行结果分析：从输出中，你会看到一条完全不同的执行轨迹：

1. 控制器首先激活新闻分析师，获取最新新闻。
2. 黑板上有新闻报告后，控制器重新评估。它根据新闻情绪（假设是积极的）决定接下来激活技术分析师，而不是财务分析师。
3. 技术分析师执行后，控制器看到已经完成了用户要求的分析（新闻+技术），于是激活报告撰写员综合最终答案。
4. 报告撰写员发布报告后，控制器判断任务完成，结束流程。

整个过程没有执行多余的财务分析，完美遵循了用户的条件逻辑。这就是黑板系统的动态协作能力。

阶段 4：量化评估

为了更客观地比较，我们用“LLM作为评委”对两个系统进行打分，重点考察指令遵循和过程效率。

classProcessLogicEvaluation(BaseModel):
    instruction_following_score: int = Field(description="智能体遵循用户特定条件指令（例如“要么/要么”逻辑）的程度，评分 1-10。")
    process_efficiency_score: int = Field(description="智能体是否采取了最直接的路径并避免了不必要的工作，评分 1-10。")
    justification: str = Field(description="对分数的简要说明，参考智能体采取的具体步骤。")

judge_llm = ChatNebius(model="mistralai/Mixtral-8x22B-Instruct-v0.1", temperature=0).with_structured_output(ProcessLogicEvaluation)

defevaluate_agent_logic(query: str, final_state: dict):
# 从状态中提取轨迹
if'blackboard'in final_state:
        trace = "\n---\n".join(final_state['blackboard'])
        agent_type = "黑板系统"
else:
        trace = f"1. 新闻报道：{final_state.get('news_report')}\n---\n2. 技术报告：{final_state.get('technical_report')}\n---\n3. 财务报告：{final_state.get('financial_report')}"
        agent_type = "顺序系统"

    prompt = f"""你是一位AI智能体过程的专家评判员。你的任务是根据智能体生成的内容轨迹来评估其表现。

**用户的原始任务：**
"{query}"

**智能体的类型：** {agent_type}
**智能体生成的内容轨迹：**
{trace}

**评估标准：**
1.  **指令遵循：** 智能体是否尊重了用户任务中的条件逻辑？（例如：“要么进行技术分析……要么进行财务分析”）。高分意味着它完美地遵循了逻辑。低分意味着它忽略了逻辑。
2.  **流程效率：** 智能体是否避免了做不必要的工作？高分意味着它只运行了所需的专家。低分意味着它运行了用户逻辑明确要求跳过的专家。

基于轨迹，提供你的评估。
"""
return judge_llm.invoke(prompt)

console.print("--- 📊 评估顺序系统的流程 ---")
seq_eval = evaluate_agent_logic(dynamic_query, final_seq_output)
console.print(seq_eval.dict())

console.print("\n--- 📊 评估黑板系统的流程 ---")
bb_eval = evaluate_agent_logic(dynamic_query, final_bb_output)
console.print(bb_eval.dict())

评估输出讨论：

• 顺序系统：指令遵循分数通常很低（比如 2/10），因为它完全无视了条件判断；效率分数也很低（比如 3/10），因为它执行了所有分析师。
• 黑板系统：两个分数都接近满分（10/10）。控制器能够根据新闻情绪智能地选择后续路径，既不漏步骤，也不做无用功。

这个定量评估有力地证明了黑板系统在处理条件逻辑任务时的巨大优势。

写在最后

通过这次实战，我们看到了黑板系统如何让AI团队从僵化的“流水线”进化为灵活的“圆桌会议”。它通过共享黑板和智能控制器，实现了真正的动态协作。

核心回顾

1. 原理：黑板系统通过共享内存和动态调度，让专家智能体基于当前进展涌现出最优工作流。
2. 实践：我们用LangGraph搭建了完整的黑板系统，并对比了顺序系统，证明了其灵活性和效率。
3. 避坑：控制器的质量是系统的天花板，需要精心设计提示词让它具备状态感知能力；同时，黑板内容的清晰度也影响专家智能体的协作效果。

在实际业务中，你可以将黑板系统应用于复杂故障诊断、多模态内容分析、金融舆情监控等场景。如果你用的是国内云服务，比如阿里云百炼、智谱AI的模型，只需要替换 ChatNebius 为相应的客户端，其他逻辑完全通用。

在你的工作或项目中，有没有遇到过“明明有多个专家AI，却不知道如何让它们协同工作”的困惑？你会选择顺序流水线，还是尝试更灵活的黑板架构？欢迎在评论区分享你的思考。