Agentic OS 设计模式

在软件工程中，设计模式是对常见问题的经过验证的解决方案。在 Agentic OS 的世界里，智能体之间的协作方式远比传统程序复杂——智能体具有自主性、不确定性和涌现行为。如何在保证系统可控性的前提下，充分发挥多智能体协作的优势？本章将介绍 Agentic OS 中最常用的几种设计模式，它们为智能体的组织、通信和协调提供了成熟的架构范式。

编排器模式（Orchestrator Pattern）

编排器模式是 Agentic OS 中最基础也最常用的多智能体协作模式。在这种模式下，一个中央编排器智能体（Orchestrator）负责理解任务目标、拆解子任务、分配给专门的工作智能体（Worker Agents），并汇总最终结果。编排器本身不执行具体的业务逻辑，它只负责"指挥"。

这种模式的优势在于集中控制和清晰的职责分离。编排器可以基于全局视角做出最优的任务分配决策，而工作智能体只需要关注自己的专业领域。劣势在于编排器是单点故障——如果编排器出现问题，整个协作流程就会中断。此外，编排器的决策能力直接决定了整个系统的表现。

编排器模式的典型应用场景包括：复杂项目管理（编排器拆解项目为子任务，分配给不同的专业智能体）、多模态内容生成（编排器协调文本、图像、音频等不同模态的生成智能体）以及客户服务流程（编排器判断客户意图，路由到对应的处理智能体）。

# 编排器模式伪代码实现
class OrchestratorAgent:
    def __init__(self, runtime):
        self.runtime = runtime
        self.workers = {
            "researcher": ResearchAgent(),
            "writer": WriterAgent(),
            "reviewer": ReviewAgent()
        }

    async def handle_task(self, task):
        # 第一步：任务分析与拆解
        plan = await self.plan_task(task)

        # 第二步：按依赖关系调度工作智能体
        results = {}
        for step in plan.steps:
            if step.depends_on:
                step.input = results[step.depends_on]

            worker = self.workers[step.worker_type]
            results[step.id] = await self.runtime.execute(
                worker, step.input
            )

        # 第三步：汇总结果
        final = await self.synthesize(task, results)
        return final

    async def plan_task(self, task):
        """使用 LLM 进行任务规划和拆解"""
        prompt = f"""分析以下任务并制定执行计划：
        任务：{task.description}
        可用智能体：{list(self.workers.keys())}
        请输出步骤列表，每步指定：
        - 步骤ID
        - 负责智能体类型
        - 输入内容
        - 依赖的步骤ID（如有）"""
        return await self.llm.generate_structured(prompt)

黑板模式（Blackboard Pattern）

黑板模式源自人工智能领域的经典问题求解方法。在这种模式中，多个智能体共享一个全局的黑板（Blackboard）——一个公共的知识空间。每个智能体独立观察黑板上的内容，当发现自己能够贡献价值时，就主动"上台"修改黑板内容。没有一个中央控制器来决定谁在何时工作——智能体之间通过黑板进行间接协调。

黑板模式的核心优势是松耦合和涌现性。智能体之间不需要直接通信，它们只需要理解黑板的格式。系统的行为是涌现的——没有哪个智能体"计划"了最终结果，结果是通过多个智能体的逐步贡献自然形成的。这种模式特别适合解决没有明确解法的复杂问题，如创意写作、科学假说生成和复杂系统分析。

管道模式（Pipeline Pattern）

管道模式是最直观的多智能体协作模式。任务像流水线上的产品一样，依次经过多个智能体的处理。每个智能体完成自己负责的处理步骤，然后将结果传递给下一个智能体。这种模式适合处理流程固定、步骤明确的线性任务。

管道模式的关键设计考虑包括：

• 阶段划分：如何将任务合理拆分为多个处理阶段？每个阶段应该由什么类型的智能体负责？
• 数据格式标准化：相邻阶段之间需要约定统一的数据格式，否则会导致信息丢失或误解。
• 错误处理：当某个阶段的智能体输出质量不达标时，是重试、回退还是跳过？
• 并行化：某些阶段可以并行执行多个实例以提高吞吐量，类似传统流水线中的并行工位。

管道模式在数据处理和分析场景中特别常见。例如，一个文档分析管道可能包括：文档解析智能体 -> 实体提取智能体 -> 关系推理智能体 -> 摘要生成智能体 -> 质量审核智能体。每个智能体专注于一项特定任务，整体效率远高于让一个智能体完成所有步骤。

# 管道模式实现：文档分析流水线
class AnalysisPipeline:
    def __init__(self, runtime):
        self.stages = [
            ("parse",     DocumentParserAgent()),
            ("extract",   EntityExtractorAgent()),
            ("reason",    RelationReasonerAgent()),
            ("summarize", SummaryGeneratorAgent()),
            ("review",    QualityReviewerAgent())
        ]
        self.runtime = runtime

    async def process(self, document):
        current_data = {"document": document}

        for stage_name, agent in self.stages:
            # 执行当前阶段
            result = await self.runtime.execute(agent, current_data)

            # 质量检查：如果审核不通过，可以回退
            if stage_name == "review" and not result.passed:
                # 回退到上一个阶段重试
                retry_agent = self.stages[-2][1]
                result = await self.runtime.execute(
                    retry_agent, current_data,
                    feedback = result.feedback
                )

            # 更新管道数据
            current_data[stage_name] = result

        return current_data

事件驱动模式（Event-Driven Pattern）

事件驱动模式是 Agentic OS 中实现松耦合协作的重要模式。在这种模式下，智能体不直接调用其他智能体，而是通过发布事件和订阅事件来间接交互。当某个事件发生时，所有订阅了该事件的智能体都会被自动触发。

事件驱动模式的优势在于极高的灵活性和可扩展性。新增一个智能体只需要订阅感兴趣的事件类型，不需要修改任何现有智能体的代码。这种"即插即用"的特性非常适合开放生态——第三方开发者可以创建新的智能体，只要它们理解事件格式，就能无缝融入现有的协作网络。

在 Agentic OS 中，事件驱动通常通过消息总线（Message Bus）实现。消息总线是运行时层提供的核心服务，负责事件的路由、过滤和投递。为了保证可靠性，消息总线通常支持事件的持久化存储和至少一次投递（At-Least-Once Delivery）语义。

发布-订阅模式（Publish-Subscribe Pattern）

发布-订阅模式是事件驱动模式的一种具体实现形式，在 Agentic OS 中有广泛的应用。与简单的事件驱动不同，发布-订阅模式引入了主题（Topic）的概念——发布者将消息发布到特定主题，订阅者订阅感兴趣的主题。这种机制实现了发布者和订阅者之间的完全解耦。

在智能体系统中，发布-订阅模式的典型应用包括：

• 状态广播：一个智能体完成重要操作后，将状态变化发布到对应主题，其他需要感知该变化的智能体自动收到通知。
• 任务分发：编排器将任务发布到任务队列主题，多个工作智能体竞争订阅，实现负载均衡。
• 知识共享：智能体将新发现的知识发布到知识主题，其他智能体可以订阅并整合到自身的知识库中。
• 异常告警：监控系统检测到异常后发布告警事件，运维智能体自动触发应急响应流程。

Saga 模式：多步骤智能体工作流

Saga 模式源自分布式事务管理领域，在 Agentic OS 中被广泛用于管理多步骤智能体工作流。一个 Saga 由一系列本地事务组成，每个本地事务由一个智能体执行。如果某个步骤失败，系统会执行预先定义的补偿操作来回滚之前步骤的效果。

Saga 模式解决的核心问题是：当多个智能体协作完成一个复杂任务时，如何保证整体的一致性和可靠性？例如，一个"自动化采购"工作流可能涉及需求分析、供应商选择、价格谈判、合同签订和付款执行等多个步骤。如果"合同签订"步骤失败（供应商违约），系统需要回滚"供应商选择"的结果，并通知"需求分析"智能体重新评估方案。

# Saga 模式实现：自动化采购工作流
class ProcurementSaga:
    def __init__(self, runtime):
        self.runtime = runtime
        self.steps = [
            {
                "name": "analyze_requirement",
                "agent": RequirementAnalystAgent(),
                "compensate": lambda ctx: ctx.cancel_requirement()
            },
            {
                "name": "select_supplier",
                "agent": SupplierSelectorAgent(),
                "compensate": lambda ctx: ctx.release_supplier_hold()
            },
            {
                "name": "negotiate_price",
                "agent": PriceNegotiatorAgent(),
                "compensate": lambda ctx: ctx.cancel_negotiation()
            },
            {
                "name": "sign_contract",
                "agent": ContractManagerAgent(),
                "compensate": lambda ctx: ctx.void_contract()
            },
            {
                "name": "execute_payment",
                "agent": PaymentAgent(),
                "compensate": lambda ctx: ctx.refund_payment()
            }
        ]

    async def execute(self, request):
        context = SagaContext(request)
        completed = []

        for step in self.steps:
            try:
                result = await self.runtime.execute(
                    step["agent"], context
                )
                context.update(step["name"], result)
                completed.append(step)
            except AgentError as e:
                # 执行补偿：逆序回滚已完成的步骤
                for prev_step in reversed(completed):
                    try:
                        await prev_step["compensate"](context)
                    except CompensationError as ce:
                        # 补偿失败需要人工介入
                        await self.alert_human(prev_step, ce)
                raise WorkflowError(
                    f"步骤 {step['name']} 失败: {e}"
                )

        return context.finalize()

模式选择指南

在实际的 Agentic OS 应用中，这些设计模式往往不是孤立使用的，而是根据场景需要组合使用。以下是一个简化的选择指南：

• 当任务可以清晰拆解为子任务，且需要集中控制时 -> 编排器模式
• 当问题没有明确解法，需要多个专家逐步贡献时 -> 黑板模式
• 当任务流程固定，数据需要逐步加工时 -> 管道模式
• 当需要松耦合、高可扩展性的通知机制时 -> 事件驱动 / 发布-订阅
• 当多步骤操作需要保证一致性和可回滚时 -> Saga 模式

OpenClaw 框架在其多智能体协作模块中，内置了对上述所有模式的支持。开发者可以通过配置文件声明智能体之间的协作模式，运行时会自动构建相应的通信拓扑和调度逻辑。这种"声明式协作"大大降低了多智能体系统的开发复杂度。