编排

两种编排哲学 在AI Agent的工作流设计中，一直存在两种根本不同的哲学： 状态机模式：预先定义好所有状态和转移条件，Agent按照预定的路径执行任务。 自由流程模式：让LLM自主决定下一步做什么，不预设执行路径。 这不是技术偏好的问题，而是对"AI Agent应该有多大的自主权"这一核心问题的不同回答。 状态机模式：可控优先 工作原理 状态机模式将任务分解为有限的状态集合，每个状态对应一组可执行的操作和转移条件。Agent在任意时刻处于某个状态，执行该状态的操作后根据结果转移到下一个状态。 [接收任务] → [分析意图] → [检索信息] → [生成答案] → [质量检查] → [返回结果] ↓ ↓ [信息不足] → [追问用户] [质量不达标] → [重新生成] 优势 确定性高：执行路径可预测，容易测试和调试。同一个输入永远走同一条路径。 可控性强：每个状态可以设置明确的成功条件和失败处理策略。不会出现Agent"跑偏"的情况。 可观测性好：状态转移日志清晰，问题定位容易——知道在哪个状态出了问题。 合规友好：对于金融、医疗等需要审计的场景，状态机可以提供完整的执行轨迹。 劣势 灵活性差：无法处理预定义路径之外的情况。如果用户的问题不在预设流程内，Agent要么报错要么走一个不太合适的流程。 开发成本高：需要预先枚举所有可能的状态和转移条件，对于复杂任务来说工作量巨大。 维护困难：需求变化时需要修改状态图，可能影响多个状态的逻辑。 适用场景 客服流程（问题类型固定，处理流程标准化） 审批流程（步骤固定，需要审计追踪） 数据处理管道（步骤明确，输入输出格式固定） 自由流程模式：灵活优先 工作原理 自由流程模式不预设执行路径，而是让LLM在每一步根据当前状态自主决策下一步行动。Agent观察当前环境、可用工具和历史信息，选择最优的下一步操作。 while not task_complete: observation = get_current_state() thought = llm.reason(observation, history) action = llm.select_action(thought, available_tools) result = execute_action(action) history.append((observation, thought, action, result)) 优势 灵活性极高：可以处理任意复杂的任务，不受预设路径的限制。 开发成本低：不需要预先设计所有流程，只需提供工具集和总体目标。 适应性强：遇到意外情况可以自主调整策略，不需要人工干预修改流程。 劣势 不可预测：同一个任务可能走完全不同的执行路径，测试和调试困难。 可靠性低：Agent可能陷入循环、选择次优策略、甚至做出不合理的操作。 可观测性差：自由流程的执行轨迹可能很长且混乱，问题定位困难。 安全风险：没有预设的约束，Agent可能执行不安全或超出预期的操作。 适用场景 研究分析（问题开放，探索路径不固定） 创意任务（需要灵活的思维路径） 复杂问题解决（无法预先枚举所有情况） 混合方案：受控自由 在实践中，纯状态机和纯自由流程都不是最优解。我们提出了"受控自由"的混合架构： 双层架构 外层状态机：定义任务的高阶段流程和安全边界。比如"必须先验证用户身份→然后才能执行操作→操作后必须记录日志"。 ...

两种编排哲学 在AI Agent的工作流设计中，一直存在两种根本不同的哲学： 状态机模式：预先定义好所有状态和转移条件，Agent按照预定的路径执行任务。 自由流程模式：让LLM自主决定下一步做什么，不预设执行路径。 这不是技术偏好的问题，而是对"AI Agent应该有多大的自主权"这一核心问题的不同回答。 状态机模式：可控优先 工作原理 状态机模式将任务分解为有限的状态集合，每个状态对应一组可执行的操作和转移条件。Agent在任意时刻处于某个状态，执行该状态的操作后根据结果转移到下一个状态。 [接收任务] → [分析意图] → [检索信息] → [生成答案] → [质量检查] → [返回结果] ↓ ↓ [信息不足] → [追问用户] [质量不达标] → [重新生成] 优势 确定性高：执行路径可预测，容易测试和调试。同一个输入永远走同一条路径。 可控性强：每个状态可以设置明确的成功条件和失败处理策略。不会出现Agent"跑偏"的情况。 可观测性好：状态转移日志清晰，问题定位容易——知道在哪个状态出了问题。 合规友好：对于金融、医疗等需要审计的场景，状态机可以提供完整的执行轨迹。 劣势 灵活性差：无法处理预定义路径之外的情况。如果用户的问题不在预设流程内，Agent要么报错要么走一个不太合适的流程。 开发成本高：需要预先枚举所有可能的状态和转移条件，对于复杂任务来说工作量巨大。 维护困难：需求变化时需要修改状态图，可能影响多个状态的逻辑。 适用场景 客服流程（问题类型固定，处理流程标准化） 审批流程（步骤固定，需要审计追踪） 数据处理管道（步骤明确，输入输出格式固定） 自由流程模式：灵活优先 工作原理 自由流程模式不预设执行路径，而是让LLM在每一步根据当前状态自主决策下一步行动。Agent观察当前环境、可用工具和历史信息，选择最优的下一步操作。 while not task_complete: observation = get_current_state() thought = llm.reason(observation, history) action = llm.select_action(thought, available_tools) result = execute_action(action) history.append((observation, thought, action, result)) 优势 灵活性极高：可以处理任意复杂的任务，不受预设路径的限制。 开发成本低：不需要预先设计所有流程，只需提供工具集和总体目标。 适应性强：遇到意外情况可以自主调整策略，不需要人工干预修改流程。 劣势 不可预测：同一个任务可能走完全不同的执行路径，测试和调试困难。 可靠性低：Agent可能陷入循环、选择次优策略、甚至做出不合理的操作。 可观测性差：自由流程的执行轨迹可能很长且混乱，问题定位困难。 安全风险：没有预设的约束，Agent可能执行不安全或超出预期的操作。 适用场景 研究分析（问题开放，探索路径不固定） 创意任务（需要灵活的思维路径） 复杂问题解决（无法预先枚举所有情况） 混合方案：受控自由 在实践中，纯状态机和纯自由流程都不是最优解。我们提出了"受控自由"的混合架构： 双层架构 外层状态机：定义任务的高阶段流程和安全边界。比如"必须先验证用户身份→然后才能执行操作→操作后必须记录日志"。 ...

引言 2026年，多 Agent 系统（Multi-Agent System, MAS）已从学术概念走向生产落地。从 OpenAI 的 Swarm 到 Google 的 A2A 协议，从 AutoGen 的多轮对话到 CrewAI 的角色协作，多 Agent 架构正在重新定义我们构建智能应用的方式。本文将系统梳理当前主流的多 Agent 设计模式，并探讨从编排到涌现的架构演进路径。 一、为什么需要多 Agent？ 单 Agent 架构在面对复杂任务时存在三个结构性瓶颈： Context Window 膨胀：一个 Agent 承担过多职责，导致 Prompt 超长、注意力稀释 工具冲突：50+ 工具注入单一 Agent 时，工具选择准确率下降至 60% 以下（Anthropic, 2025） 验证盲区：自我验证存在系统性偏见，单 Agent 自我纠错的成功率仅 34% 多 Agent 通过任务分解和角色专精，将上述问题分解到可管理的粒度。 二、六大核心设计模式 模式 1：Orchestrator-Worker（编排者-执行者） 最经典也最常用的模式。一个 Orchestrator Agent 负责任务分解和结果聚合，多个 Worker Agent 各司其职。 ┌─────────────────────────────────────┐ │ Orchestrator Agent │ │ (任务分解 / 优先级 / 结果聚合) │ └──────────┬──────────┬───────────────┘ │ │ ┌──────▼──┐ ┌────▼────┐ ┌────────┐ │ Worker A │ │ Worker B │ │Worker C│ │ (搜索) │ │ (分析) │ │ (写作) │ └─────────┘ └─────────┘ └────────┘ 适用场景：内容创作流水线、数据处理 Pipeline ...

子代理编排概述 子代理编排是 OpenClaw 的核心能力之一。通过子代理编排，龙虾可以同时处理多个任务，提高效率和并发处理能力。 子代理编排的核心思想是将复杂任务拆分为多个子任务，并行执行，最后汇总结果。 子代理架构 子代理类型 1. 隔离子代理 隔离子代理有独立的上下文，不共享父会话的对话历史。 特点： 上下文独立 适合执行独立任务 不会泄露父会话信息 使用场景： 执行敏感操作 处理独立任务 批量处理数据 2. Fork 子代理 Fork 子代理共享父会话的对话历史。 特点： 共享上下文 适合需要父会话信息的任务 可以访问父会话的文件 使用场景： 需要父会话信息的任务 文件操作 信息查询 子代理生命周期 1. 创建 通过 sessions_spawn 创建子代理。 subagent = sessions_spawn( task="执行特定任务", label="子代理标签", context="isolated" # 隔离模式 ) 2. 运行 子代理在独立环境中运行，执行任务。 3. 监控 通过 subagents 监控子代理的状态。 # 获取子代理列表 subagents_list = subagents(action="list", recentMinutes=60) 4. 完成 子代理完成后，通过回调机制通知父会话。 5. 清理 子代理完成后自动清理资源。 并行任务处理 并行执行 OpenClaw 支持并行执行多个子代理，提高任务处理效率。 限制： 并发上限：通常限制为 5 个并发子代理 资源限制：受服务器资源限制 任务依赖：某些任务可能有依赖关系 任务拆分策略 1. 按任务类型拆分 将不同类型的任务拆分为不同的子代理。 ...

引言：智能体编排的挑战 当单个 LLM 调用无法满足复杂任务需求时，智能体工作流编排应运而生。无论是多步推理、工具调用、还是多智能体协作，都需要一个可靠的编排层来管理执行流程。然而，智能体工作流的编排远比传统的数据处理管道复杂——执行路径可能动态变化，中间结果会影响后续步骤，错误处理需要语义级别的理解。 本文将从最基础的 DAG（有向无环图）编排出发，逐步深入到动态执行图、状态机模型，以及 LangGraph 等主流框架的设计哲学与实践方法。 一、静态 DAG：确定性流程的基石 1.1 什么是 DAG 编排 有向无环图（DAG）是最直观的工作流编排模型。节点表示执行单元（如 LLM 调用、工具调用、数据处理），边表示数据流向。DAG 的核心约束是"无环"——数据只能向前流动，不能回头。 一个典型的 RAG DAG 工作流： 用户查询 → 查询改写 → 向量检索 → 重排序 → 上下文构建 → LLM 生成 → 后处理 → 输出 这种线性流程是最简单的 DAG。更复杂的场景可能包含并行分支，例如同时检索多个数据源、同时调用多个工具，然后汇总结果。 1.2 DAG 的优势与局限 优势： 执行路径确定，易于调试和追踪 天然支持并行执行 可以静态分析依赖关系，优化调度策略 错误处理相对简单——只需处理节点级失败 局限： 无法处理需要循环或迭代的场景（如反思-修正循环） 执行路径在编译时确定，无法根据中间结果动态调整 对于条件分支的支持有限，通常只能通过预先定义所有可能路径来实现 二、条件分支与循环：打破 DAG 的限制 2.1 条件路由 实际业务中，执行路径往往需要根据中间结果动态决定。例如： 检索结果置信度低 → 触发补充检索 生成答案包含不确定信息 → 触发事实核查 用户意图分类为"闲聊" → 跳过检索直接回复 条件路由的实现方式通常是在节点输出上定义路由函数，根据输出内容选择下一个执行节点。这虽然打破了纯 DAG 的"无环"约束，但仍保持了流程的可预测性。 ...

1. 引言：为什么需要多智能体编排 单个 LLM Agent 在复杂任务中面临上下文窗口限制、角色混淆、推理链断裂等问题。多智能体架构通过任务分解、角色专精和协作机制，将复杂问题分配给多个专业化 Agent 协同完成。然而，如何编排这些 Agent——谁来调度、如何通信、何时同步——是工程落地的核心挑战。 2. 三种核心编排模式 2.1 中心化编排（Orchestrator Pattern） 一个中央编排器（Orchestrator）负责任务分配、状态管理和结果汇总。所有 Agent 只与编排器通信，互不直接交互。 ┌──────────────────────────────────┐ │ Orchestrator │ │ ┌──────┐ ┌──────┐ ┌──────┐ │ │ │Worker│ │Worker│ │Worker│ │ │ │ A │ │ B │ │ C │ │ │ └──────┘ └──────┘ └──────┘ │ └──────────────────────────────────┘ 核心代码实现： from abc import ABC, abstractmethod from typing import Any, Optional import asyncio class Agent(ABC): def __init__(self, name: str, system_prompt: str): self.name = name self.system_prompt = system_prompt self.message_history: list[dict] = [] @abstractmethod async def execute(self, task: str, context: dict) -> str: pass class Orchestrator: def __init__(self): self.agents: dict[str, Agent] = {} self.task_queue: list[dict] = [] self.results: dict[str, Any] = {} def register(self, agent: Agent): self.agents[agent.name] = agent async def dispatch(self, agent_name: str, task: str, context: dict = None) -> str: agent = self.agents[agent_name] result = await agent.execute(task, context or {}) self.results[f"{agent_name}:{task[:20]}"] = result return result async def run_pipeline(self, plan: list[dict]) -> dict: """按计划顺序执行任务，支持依赖传递""" for step in plan: agent_name = step["agent"] task = step["task"] deps = step.get("depends_on", []) merged_context = {d: self.results.get(d) for d in deps} await self.dispatch(agent_name, task, merged_context) return self.results 适用场景： 工作流明确的任务（如代码审查流水线、文档生成管线） ...

为什么需要 Agent 编排 单个 Agent 能力有限。复杂任务需要多个 Agent 协作：一个负责检索、一个负责分析、一个负责生成报告。如何编排这些 Agent 是生产环境的核心问题。 编排模式分类 模式 结构 适用场景 复杂度 串行 A -> B -> C 流水线任务 低 并行 A,B,C -> 合并 独立子任务 低 循环 A -> B -> 判断 -> (A 或 结束) 迭代优化 中 路由 Input -> Router -> [A/B/C] 分类分发 中 监督者 Supervisor -> [A,B,C] 中心调度 中 分层 Top Supervisor -> Sub-Supervisors -> Workers 大规模团队 高 图式 DAG/状态机 复杂工作流 高 1. Router Pattern 一个路由 Agent 根据输入决定调用哪个专家 Agent： ...