AI Agent 架构实践：基于状态机与结构化输出的多智能体协作模式

随着大语言模型（LLM）能力的演进，单一 Prompt 的调用逐渐演变为复杂的 Multi-Agent（多智能体）协同系统。在实际业务场景落地中，我们常常会遇到“幻觉失控”、“多 Agent 间上下文污染”以及“执行流程不可预期”等问题。

本文结合近期在智能体工作流设计中的工程实践，分享如何利用**结构化输出（Structured Output）与有限状态机（Finite State Machine, FSM）**构建可控、健壮的多智能体协作架构。

一、为什么需要状态机约束 Agent

传统的纯自然语言链式对话往往依赖模型自主判断“下一步做什么”，但在工业级应用中，完全放开控制流会带来极高的不确定性：

1. 死循环与重复执行：Agent A 与 Agent B 互相客套或反复确认，无法收敛。
2. 上下文漂移：随着会话轮次增加，初始系统约束被稀释。
3. 异常中断难以恢复：单步失败导致整个长任务难以回滚或重试。

引入状态机（FSM）的核心思想是将多 Agent 的协同过程分解为明确的状态集合与转换条件。每个状态对应具体的 Agent 角色与任务边界，Agent 仅负责在当前状态下决策并产出迁移触发信号，主调度器掌控控制权。

二、架构设计与核心组件

一个健壮的 Multi-Agent 系统通常包含以下三个核心组件：

1. Dispatcher（主调度器）：
   • 维护当前会话全局状态（Global State）。
   • 根据 Agent 返回的 Action 决定状态迁移流向。
2. Role Agent（角色智能体）：
   • 专注于单一职责（如需求解析、代码生成、测试验收）。
   • 具备最小化的上下文视窗，避免不必要的冗余信息。
3. Output Contract（结构化契约）：
   • 强制使用 JSON Schema 或 Function Call 规范化返回数据。

三、代码实现示例（Python + Pydantic）

下面是一个精简的实现骨架，展示如何利用 Pydantic 约束 Agent 的决策输出：

from enum import Enum
from typing import Optional, List
from pydantic import BaseModel, Field

class SystemState(str, Enum):
    IDLE = "IDLE"
    ANALYZING = "ANALYZING"
    EXECUTING = "EXECUTING"
    VERIFYING = "VERIFYING"
    COMPLETED = "COMPLETED"
    FAILED = "FAILED"

class AgentDecision(BaseModel):
    current_thought: str = Field(description="当前步骤的思考过程与判断依据")
    next_action: str = Field(description="下一步执行动作，例如 review_code, run_test")
    suggested_next_state: SystemState = Field(description="建议流转的目标状态")
    payload: dict = Field(default_factory=dict, description="传递给下一阶段的结构化参数")

def transition(current_state: SystemState, decision: AgentDecision) -> SystemState:
    # 状态转移硬约束，防止模型越权跳转
    valid_transitions = {
        SystemState.IDLE: [SystemState.ANALYZING],
        SystemState.ANALYZING: [SystemState.EXECUTING, SystemState.FAILED],
        SystemState.EXECUTING: [SystemState.VERIFYING, SystemState.FAILED],
        SystemState.VERIFYING: [SystemState.COMPLETED, SystemState.EXECUTING, SystemState.FAILED],
    }
    
    if decision.suggested_next_state in valid_transitions.get(current_state, []):
        return decision.suggested_next_state
    else:
        # 非法跃迁，保持当前状态并触发重试
        print(f"Warning: Invalid state transition from {current_state} to {decision.suggested_next_state}")
        return current_state

四、工程落地踩坑与优化经验

1. 思考过程与结果分离：
   在大模型响应中，建议保留 <thought> 思考块，但最终输出必须强制要求合法的 JSON 格式。如果模型输出附带额外 markdown 标记，应做容错解析器（Robust Parser）。

2. 单 Agent 上下文“瘦身”：
   不要将所有的历史对话盲目丢给每个 Agent。在状态流转时，Dispatcher 只打包当前步骤所需的 Payload，不仅大幅降低 Token 消耗，还能显著提升模型的专注度与指令遵循率。

3. 设置全局最大步数限制（Max Steps Guard）：
   即使有状态机控制，仍需设置物理安全阈值（如单次任务上限 15 步），防止异常输入导致的边界震荡。

五、总结

将大模型的生成自由度约束在确定性的工程架构中，是推动 AI 智能体走向生产环境的关键。通过状态机定义骨架、通过结构化输出规范接口、通过轻量化上下文保障稳定，能够有效构建出高可靠、易排查的多 Agent 协作系统。