Multi-Agent 编排架构实践：Supervisor + 7 Specialist 的状态机设计

前言

在 AI Agent 从单任务向工程化落地的过程中，多 Agent 协作（Multi-Agent Orchestration） 已经成为一个核心命题。当单个 Agent 的能力边界越来越清晰时，如何让多个专业 Agent 高效协作，共同完成复杂的工程任务，是衡量一个 AI 平台架构水平的关键指标。

本文将基于云萱 AI 原生工程平台的实践，详细拆解一套可落地的 Multi-Agent 编排架构，包括：

• Supervisor + Specialist 的分层设计
• LangGraph 与纯 Python 状态机的双实现
• 强类型状态传递（Pydantic）
• 条件分支与迭代回退机制
• 人工检查点（HITL）的设计

---

一、为什么需要 Multi-Agent？

1.1 单 Agent 的局限性

单个 LLM Agent 在面对复杂工程任务时，会遇到几个核心瓶颈：

1. 注意力分散：一个 Agent 同时要懂需求、写代码、做测试、搞部署，上下文窗口很快被撑爆
2. 专业深度不足：让同一个 Prompt 既当产品经理又当测试工程师，结果往往是”样样通、样样松”
3. 错误难以隔离：代码写错了，同一个 Agent 自己调试自己，容易陷入”当局者迷”的死循环
4. 无法并行：所有任务串行执行，效率低下

1.2 Multi-Agent 的核心优势

Multi-Agent 架构借鉴了人类社会的专业化分工思想：

让专业的 Agent 做专业的事，通过编排器统一协调。

具体来说：

维度	单 Agent	Multi-Agent
角色定位	全栈（泛化）	专业化分工
上下文管理	容易溢出	按阶段隔离
错误隔离	自己发现不了自己的问题	不同 Agent 交叉验证
可扩展性	改 Prompt 风险高	新增 Agent 不影响现有逻辑
质量保障	缺乏制衡	多环节质量关卡

---

二、整体架构：Supervisor + 7 Specialist

2.1 架构分层

我们采用 Supervisor-Worker 模式的变体：

                    ┌─────────────────────┐
                    │   Supervisor Agent  │
                    │   (全局编排器)       │
                    └─────────┬───────────┘
                              │
           ┌──────────────────┼──────────────────┐
           │                  │                  │
    ┌──────▼──────┐   ┌──────▼──────┐   ┌──────▼──────┐
    │ Requirement │   │  Developer  │   │   Tester    │
    │   Agent     │   │   Agent     │   │   Agent     │
    └─────────────┘   └─────────────┘   └─────────────┘
           │                  │                  │
    ┌──────▼──────┐   ┌──────▼──────┐   ┌──────▼──────┐
    │  Debugger   │   │  Evaluator  │   │  Security   │
    │   Agent     │   │   Agent     │   │   Agent     │
    └─────────────┘   └─────────────┘   └─────────────┘
                              │
                    ┌─────────▼──────────┐
                    │   Release Agent    │
                    └────────────────────┘

Supervisor（编排器） 不做具体业务，只负责：

• 全局状态管理
• 阶段流转调度
• 条件分支判断
• 迭代次数控制
• 人工检查点管理

7 个 Specialist Agent 各司其职：

Agent	角色	核心产出
RequirementAgent	需求分析师	结构化需求文档、验收标准、任务拆分
DeveloperAgent	开发工程师	设计文档、代码文件、架构说明
TesterAgent	测试工程师	测试计划、测试用例、执行报告
DebuggerAgent	调试专家	Bug 定位、修复方案、迭代记录
EvaluatorAgent	评估师	质量/安全/性能综合评分
SecurityAgent	安全工程师	漏洞检测、注入测试、合规检查
ReleaseAgent	发布工程师	版本管理、变更日志、CI/CD 状态

2.2 工作流阶段顺序

7 个 Agent 按照软件工程的天然流程排列成一条流水线：

requirement → developer → tester → debugger → evaluator → security → release → done

这不是一条简单的直线——中间有条件分支和回退机制：

• tester → debugger：测试失败时，进入调试阶段（最多 3 次修复迭代）
• debugger → tester：修复完成后，回到测试阶段重新验证
• security → developer：发现高危漏洞时，回到开发阶段修复

这确保了质量问题不会”流到下一关”。

---

三、编排引擎的双实现

3.1 为什么做双实现？

我们的编排引擎同时支持两种实现：

1. LangGraph 模式（优先）：基于 LangChain 官方的 StateGraph，功能强大，生态丰富
2. 纯 Python 状态机模式（Fallback）：零依赖、轻量、可预测、易调试

为什么不直接用 LangGraph？

• 环境兼容性：在某些受限环境下（如离线部署、资源紧张），LangGraph + LangChain 的依赖链可能过重
• 可控性：纯 Python 实现完全可控，出问题时可以快速定位
• 学习成本：团队新人可以先从简单状态机入手，再深入 LangGraph

运行时自动检测：

def _langgraph_available() -> bool:
    """检查 LangGraph 是否可用。"""
    try:
        import langgraph
        return True
    except ImportError:
        return False

3.2 LangGraph 实现

LangGraph 的核心是 StateGraph——一个有状态的图，每个节点是一个 Agent，边是流转条件。

from langgraph.graph import StateGraph, END

# 构建图
graph = StateGraph(AgentState)

# 添加节点（每个节点对应一个 Specialist Agent）
graph.add_node("requirement", self._run_requirement)
graph.add_node("developer", self._run_developer)
graph.add_node("tester", self._run_tester)
graph.add_node("debugger", self._run_debugger)
graph.add_node("evaluator", self._run_evaluator)
graph.add_node("security", self._run_security)
graph.add_node("release", self._run_release)

# 设置入口
graph.set_entry_point("requirement")

# 条件边：测试 → 调试或评估
graph.add_conditional_edges(
    "tester",
    self._should_enter_debugger,  # 判断函数
    {
        "debugger": "debugger",
        "evaluator": "evaluator",
    }
)

# 调试 → 回到测试
graph.add_edge("debugger", "tester")

# 安全 → 开发（高危时）或发布
graph.add_conditional_edges(
    "security",
    self._should_back_to_developer,
    {
        "developer": "developer",
        "release": "release",
    }
)

# 正常流转边
graph.add_edge("requirement", "developer")
graph.add_edge("developer", "tester")
graph.add_edge("evaluator", "security")
graph.add_edge("release", END)

self._graph = graph.compile()

这种声明式的图结构，让复杂的 Agent 协作逻辑变得清晰可读。

3.3 纯 Python 状态机实现

当 LangGraph 不可用时，自动降级为纯 Python 实现。核心逻辑是一个 while 循环 + 状态机：

def run_python_state_machine(self, initial_state: AgentState) -> AgentState:
    """纯 Python 状态机实现。"""
    state = initial_state
    iteration = 0

    while state.current_stage != "done" and iteration < self.max_iterations:
        iteration += 1
        stage = state.current_stage

        # 执行对应阶段的 Agent
        stage_handler = self._stage_handlers.get(stage)
        if stage_handler:
            state = stage_handler(state)

        # 计算下一阶段
        state = self._transition(state)

        # 检查人工检查点
        if stage in self.human_checkpoints and self.human_checkpoints[stage]:
            state = self._wait_for_approval(state, stage)

    state.status = "completed" if state.current_stage == "done" else "max_iterations"
    return state

这种实现虽然”朴素”，但胜在简单、可预测、易调试——所有逻辑都在一个函数里，出问题时单步调试就能搞定。

---

四、强类型状态传递

4.1 为什么需要强类型？

在 Multi-Agent 系统中，状态（State）是 Agent 之间的通用语言。如果状态结构松散、字段定义模糊，Agent 之间的协作就会充满歧义。

想象一下：

• Developer Agent 把代码放在 code_files 字段里
• Tester Agent 却去 generated_code 字段里找

这就是自然语言的歧义性在 Agent 协作中的体现。

我们用 Pydantic 模型来定义强类型状态，从根本上消除歧义：

class AgentState(PlatformBaseModel):
    """Agent 全局状态。"""

    # 基础标识
    task_id: str = ""
    session_id: str = ""
    user_id: str = ""
    user_request: str = ""

    # 流转控制
    current_stage: str = "requirement"
    stage_history: list[str] = []
    iteration_count: int = 0
    max_iterations: int = 10

    # 各阶段状态（每个 Specialist 有专属子状态）
    requirement: dict[str, Any] = {}
    developer: dict[str, Any] = {}
    tester: dict[str, Any] = {}
    debugger: dict[str, Any] = {}
    evaluator: dict[str, Any] = {}
    security: dict[str, Any] = {}
    release: dict[str, Any] = {}

    # 工具调用记录
    tool_calls: list[dict[str, Any]] = []
    tool_results: list[dict[str, Any]] = []

    # 审批与检查点
    pending_approvals: list[str] = []
    approved_checkpoints: list[str] = []

    # 最终结果
    final_answer: str = ""
    status: str = "running"  # running / completed / failed / blocked
    error: str = ""

每个 Specialist Agent 只读写自己的子状态对象，避免互相干扰。Supervisor 负责在阶段流转时，把上一阶段的输出传递给下一阶段需要的输入。

4.2 状态的可追溯性

每一次阶段流转都会记录到 stage_history 中：

def _transition(self, state: AgentState) -> AgentState:
    """阶段流转。"""
    # ... 判断逻辑 ...

    # 记录历史
    state.stage_history.append(state.current_stage)
    state.current_stage = next_stage
    state.iteration_count += 1

    return state

这样，事后复盘时可以完整看到 Agent 的决策路径——哪个阶段走了几次、为什么回退、总共迭代了多少轮。

这就是 Agent Trajectory（轨迹） 的基础数据。

---

五、条件分支与迭代控制

5.1 测试-调试循环

测试失败后进入调试，调试完成后回到测试，这是一个经典的反馈循环。

关键是要有最大迭代次数限制，避免死循环：

def _should_enter_debugger(self, state: AgentState) -> str:
    """判断是否进入调试阶段。"""
    # 测试全部通过 → 进入评估
    tester_state = state.tester
    if tester_state.get("all_passed", False):
        return "evaluator"

    # 失败次数过多 → 仍进入评估（标记为失败）
    debug_count = state.stage_history.count("debugger")
    if debug_count >= self.max_fix_iterations:
        logger.warning(f"已达到最大修复迭代次数 ({self.max_fix_iterations})，停止调试")
        return "evaluator"

    # 否则进入调试
    return "debugger"

这种设计模拟了真实的开发流程：修 Bug 不可能无限迭代，到了一定次数就要评估”是否值得继续修”。

5.2 安全回退机制

安全检查发现高危漏洞时，直接打回开发阶段重新修改：

def _should_back_to_developer(self, state: AgentState) -> str:
    """安全检查未通过 → 回到开发阶段。"""
    security_state = state.security
    critical_count = security_state.get("critical_count", 0)
    high_count = security_state.get("high_count", 0)

    if critical_count > 0 or high_count > 2:
        logger.warning(f"安全检查未通过（高危: {high_count}, 严重: {critical_count}），打回开发阶段")
        return "developer"

    return "release"

这是一道安全闸门——确保有严重安全问题的代码永远不会流到发布阶段。

---

六、人工检查点（HITL）

6.1 HITL 的设计哲学

在 AI 工程化落地中，完全自动化是不现实的，也是危险的。我们需要在关键节点设置人工检查点。

我们的平台设置了 4 个默认检查点：

检查点	位置	人工确认内容
before_code_complete	开发完成后	代码质量、架构合理性
before_security_pass	安全评估后	风险接受程度
before_git_push	推送代码前	确认要提交的内容
before_production_deploy	发布前	最终上线确认

这些检查点可以通过配置开启或关闭：

pipeline:
  human_checkpoints:
    before_code_complete: true
    before_security_pass: true
    before_git_push: true
    before_production_deploy: true

6.2 检查点的实现

当工作流到达一个开启的检查点时，Agent 会暂停并等待人工审批：

def _wait_for_approval(self, state: AgentState, stage: str) -> AgentState:
    """等待人工审批。"""
    state.pending_approvals.append(stage)
    state.status = "blocked"

    # 这里可以接入审批系统：发送通知、生成审批页面等
    logger.info(f"⏸️  阶段 '{stage}' 需要人工审批，工作流暂停")

    # 在 CLI 模式下，可以交互式等待输入
    # 在 API 模式下，应该返回状态，由前端触发继续
    return state

这种设计让 AI 系统始终处于人类的可控范围内——AI 做执行，人类做决策。

---

七、实践总结与思考

7.1 架构设计的关键原则

回顾整个 Multi-Agent 编排架构的设计，有几条核心原则：

1. 关注点分离：Supervisor 只管编排，Specialist 只管业务，互不越界
2. 强类型优先：用 Pydantic 定义状态，消除自然语言歧义
3. 防御性设计：最大迭代次数、安全闸门、人工检查点
4. 降级容错：LangGraph 不可用时自动降级为纯 Python 实现
5. 可追溯性：完整的阶段历史 + 工具调用记录

7.2 还可以优化的方向

这套架构仍然有很多可以演进的空间：

• 并行执行：当前是严格的流水线，某些独立任务（如安全扫描和测试）可以并行
• 动态规划：当前阶段顺序是固定的，可以让 Supervisor 根据任务类型动态调整流程
• Agent 间直接通信：当前都通过 Supervisor 转发，可以引入 Agent 间直接协作机制
• RL 优化编排策略：用强化学习学习最优的调度策略（但这需要大量训练数据）

7.3 写在最后

Multi-Agent 不是银弹——它增加了系统复杂度，也引入了新的问题（如状态一致性、通信开销、错误传播）。但在工程化、流程化、质量要求高的场景下，Multi-Agent 的收益远大于成本。

关键是找到适合自己业务的粒度和边界：

• 简单任务 → 单 Agent 就够了
• 有明确阶段划分的复杂任务 → Multi-Agent
• 完全开放、无固定流程的任务 → 可能需要更灵活的架构

希望这篇文章的实践经验能给你带来一些启发。

---

相关阅读：云萱 AI 原生工程平台：从需求到上线的全链路 AI 自动化实践