AI Agent 编排架构：从 ReAct 循环到多 Agent 协作的生产级设计

键盘鼓手苏苏

8人浏览 · 2026-06-28 09:26:35

键盘鼓手苏苏 · 2026-06-28 09:26:35 发布

AI Agent 编排架构：从 ReAct 循环到多 Agent 协作的生产级设计

一、Agent 跑飞了——AI Agent 编排的失控困局

AI Agent 上线第一周，运维团队收到了一张巨额 API 账单。排查发现，一个 Agent 在处理用户请求时陷入了无限循环：调用搜索工具 → 结果不满意 → 再次搜索 → 还是找不到 → 继续搜索……一个请求消耗了 47 次 LLM 调用和 120 次工具调用，单次请求成本超过 3 美元。

更严重的问题是可靠性。Agent 的执行路径不可预测——同一个问题，第一次执行 3 步就解决了，第二次却绕了 15 步还没结果。这种不确定性在生产环境中是不可接受的。用户不会为"可能能解决"的 Agent 买单，他们需要的是确定性的结果和可控的成本。

AI Agent 编排的核心挑战：如何在保留 Agent 灵活性的同时，建立执行边界、成本控制和失败恢复机制。本文从单 Agent 的 ReAct 循环优化到多 Agent 协作架构，给出生产级方案。

二、Agent 执行循环的底层机制：从 ReAct 到 Plan-Execute

Agent 的核心执行模式是"感知-推理-行动"循环。ReAct（Reasoning + Acting）是最经典的模式：LLM 先推理下一步该做什么，然后调用工具执行，观察结果后再推理下一步。但 ReAct 的致命问题是缺乏全局规划，容易陷入局部循环。

graph TB
    subgraph ReAct["ReAct 模式（易循环）"]
        R1["推理：需要搜索 X"]
        R2["行动：调用搜索工具"]
        R3["观察：结果不完整"]
        R4["推理：再搜索 Y"]
        R5["行动：调用搜索工具"]
        R6["观察：还是不完整"]
        R7["推理：继续搜索..."]
        R1 --> R2 --> R3 --> R4 --> R5 --> R6 --> R7
    end

    subgraph PlanExec["Plan-Execute 模式（可控）"]
        P1["规划：生成执行计划<br/>步骤1/2/3/4"]
        P2["执行步骤1：搜索 X"]
        P3["执行步骤2：搜索 Y"]
        P4["执行步骤3：整合结果"]
        P5["执行步骤4：生成回答"]
        P6["反思：结果是否满足？<br/>不满足则修订计划"]
        P1 --> P2 --> P3 --> P4 --> P5 --> P6
        P6 -->|需要修订| P1
    end

    style ReAct fill:#ffebee,stroke:#f44336,stroke-width:2px
    style PlanExec fill:#e8f5e9,stroke:#4caf50,stroke-width:2px

关键机制拆解：

ReAct 的循环陷阱。ReAct 模式下，LLM 的每一步推理都依赖上一步的观察结果。如果工具返回的结果不满足预期，LLM 可能反复尝试同一种策略（如不断换关键词搜索），而不是换一种思路。根本原因是 LLM 没有"全局进度"的概念，每一步都是局部决策。

Plan-Execute 的优势。先让 LLM 生成完整的执行计划（步骤列表），然后按计划逐步执行。计划提供了全局视角，LLM 可以在规划阶段就识别潜在的死循环路径并规避。执行阶段如果发现计划不可行，可以回到规划阶段修订计划，而非盲目重试。

多 Agent 协作的分工逻辑。单 Agent 的能力受限于单个 LLM 的上下文窗口和推理能力。多 Agent 架构将复杂任务拆分给专业化的 Agent：规划 Agent 负责制定计划，执行 Agent 负责调用工具，审核 Agent 负责质量把关。每个 Agent 的 System Prompt 更短更聚焦，推理质量更高。

三、生产级 Agent 编排框架实现

3.1 Plan-Execute Agent 核心实现

# agent_orchestrator.py：Plan-Execute Agent 编排器
from typing import List, Optional, Callable
from pydantic import BaseModel
import json

class PlanStep(BaseModel):
    """执行计划的单个步骤"""
    step_id: int
    description: str
    tool: str  # 使用的工具名称
    tool_input: dict  # 工具输入参数
    expected_output: str  # 预期输出描述
    status: str = "pending"  # pending / running / done / failed

class ExecutionPlan(BaseModel):
    """完整的执行计划"""
    steps: List[PlanStep]
    max_revisions: int = 2  # 最大计划修订次数
    revision_count: int = 0

class AgentOrchestrator:
    def __init__(
        self,
        llm_client,
        tools: dict[str, Callable],
        max_steps: int = 10,  # 最大执行步骤数
        max_cost_usd: float = 0.5,  # 单次请求最大成本
    ):
        self.llm = llm_client
        self.tools = tools
        self.max_steps = max_steps
        self.max_cost_usd = max_cost_usd
        self.total_cost = 0.0

    async def execute(self, query: str) -> dict:
        """
        Plan-Execute 主循环
        规划 → 执行 → 反思 → 修订（如需）
        """
        # 阶段 1：生成执行计划
        plan = await self._generate_plan(query)

        step_count = 0
        results = {}

        # 阶段 2：逐步执行计划
        for step in plan.steps:
            # 安全检查：步骤数和成本控制
            # 为什么需要双重限制？步骤数防止无限循环，
            # 成本限制防止 API 账单失控
            step_count += 1
            if step_count > self.max_steps:
                results["_truncated"] = True
                break
            if self.total_cost > self.max_cost_usd:
                results["_cost_exceeded"] = True
                break

            step.status = "running"
            try:
                # 执行工具调用
                tool_fn = self.tools.get(step.tool)
                if not tool_fn:
                    step.status = "failed"
                    results[step.step_id] = {
                        "error": f"工具 {step.tool} 不存在"
                    }
                    continue

                result = await tool_fn(**step.tool_input)
                step.status = "done"
                results[step.step_id] = result

                # 累计成本估算
                self.total_cost += self._estimate_step_cost(step, result)

            except Exception as e:
                step.status = "failed"
                results[step.step_id] = {"error": str(e)}
                # 单步失败不中断整个计划，继续执行后续步骤
                # 为什么不中断？后续步骤可能不依赖失败步骤的结果

        # 阶段 3：反思与修订
        # 为什么需要反思？计划可能基于不完整信息制定，
        # 执行后可能发现需要补充步骤
        reflection = await self._reflect(query, plan, results)

        if reflection.needs_revision and plan.revision_count < plan.max_revisions:
            plan.revision_count += 1
            revised_plan = await self._revise_plan(
                query, plan, results, reflection.feedback
            )
            # 执行修订后的新增步骤
            for step in revised_plan.steps:
                if step.status == "pending" and step_count < self.max_steps:
                    result = await self.tools[step.tool](**step.tool_input)
                    results[step.step_id] = result
                    step_count += 1

        # 阶段 4：生成最终回答
        answer = await self._synthesize_answer(query, results)
        return {
            "answer": answer,
            "steps_executed": step_count,
            "total_cost_usd": self.total_cost,
            "plan_revisions": plan.revision_count,
        }

    async def _generate_plan(self, query: str) -> ExecutionPlan:
        """让 LLM 生成执行计划"""
        prompt = f"""根据用户问题，生成一个执行计划。

用户问题：{query}

可用工具：{list(self.tools.keys())}

要求：
1. 每个步骤必须指定工具和输入参数
2. 步骤之间有明确的依赖关系
3. 总步骤数不超过 {self.max_steps}
4. 必须包含验证步骤，确认结果是否满足用户需求

输出 JSON 格式的执行计划。"""

        response = await self.llm.generate(prompt)
        plan_data = json.loads(response)
        return ExecutionPlan(**plan_data)

    def _estimate_step_cost(self, step: PlanStep, result: dict) -> float:
        """估算单步执行成本"""
        # LLM 调用成本：约 0.01-0.03 美元/次（GPT-4 级别）
        # 工具调用成本：搜索 0.001 美元/次，代码执行 0.01 美元/次
        base_cost = {"search": 0.001, "code_exec": 0.01, "llm_call": 0.02}
        return base_cost.get(step.tool, 0.005)

3.2 多 Agent 协作：Supervisor 模式

# multi_agent.py：基于 Supervisor 的多 Agent 协作
from enum import Enum

class AgentRole(Enum):
    PLANNER = "planner"      # 规划 Agent
    EXECUTOR = "executor"    # 执行 Agent
    REVIEWER = "reviewer"    # 审核 Agent

class SupervisorAgent:
    """
    Supervisor Agent：协调多个专业 Agent 的执行
    为什么用 Supervisor 而非对等协作？
    对等协作缺乏全局控制，容易出现 Agent 之间的循环调用；
    Supervisor 模式有明确的控制流，更容易实现成本控制和超时管理
    """

    def __init__(self, llm_client, tools: dict):
        self.planner = PlannerAgent(llm_client)
        self.executor = ExecutorAgent(llm_client, tools)
        self.reviewer = ReviewerAgent(llm_client)
        self.max_iterations = 3  # 最大规划-执行-审核循环次数

    async def run(self, query: str) -> dict:
        iteration = 0
        context = {"query": query, "history": []}

        while iteration < self.max_iterations:
            iteration += 1

            # 步骤 1：规划 Agent 制定执行计划
            plan = await self.planner.plan(query, context)

            # 步骤 2：执行 Agent 按计划执行
            execution_result = await self.executor.execute(plan, context)

            # 步骤 3：审核 Agent 评估结果质量
            review = await self.reviewer.review(
                query=query,
                plan=plan,
                result=execution_result,
            )

            context["history"].append({
                "iteration": iteration,
                "plan": plan,
                "result": execution_result,
                "review": review,
            })

            # 审核通过，返回结果
            if review.passed:
                return {
                    "answer": execution_result.answer,
                    "iterations": iteration,
                    "quality_score": review.score,
                }

            # 审核不通过，将反馈注入上下文，进入下一轮
            # 为什么注入反馈而非直接重试？反馈让规划 Agent 知道
            # 上一轮哪里出了问题，避免重复同样的错误
            context["last_review_feedback"] = review.feedback

        # 超过最大迭代次数，返回当前最佳结果
        return {
            "answer": execution_result.answer,
            "iterations": iteration,
            "quality_score": review.score,
            "warning": "达到最大迭代次数，结果可能不完整",
        }


class ReviewerAgent:
    """审核 Agent：评估执行结果的质量"""

    async def review(self, query: str, plan, result) -> ReviewResult:
        prompt = f"""评估以下执行结果是否完整回答了用户问题。

用户问题：{query}
执行计划：{plan}
执行结果：{result}

评估维度：
1. 完整性：是否覆盖了问题的所有方面
2. 准确性：事实性信息是否可验证
3. 相关性：回答是否与问题直接相关

输出 JSON：{{"passed": bool, "score": float, "feedback": str}}"""

        response = await self.llm.generate(prompt)
        return ReviewResult(**json.loads(response))

3.3 Agent 执行的 K8s 部署配置

apiVersion: apps/v1
kind: Deployment
metadata:
  name: agent-orchestrator
  namespace: ai-serving
spec:
  replicas: 3
  selector:
    matchLabels:
      app: agent-orchestrator
  template:
    metadata:
      labels:
        app: agent-orchestrator
    spec:
      containers:
        - name: orchestrator
          image: ai-registry.internal/agent-orchestrator:v1.5.0
          ports:
            - containerPort: 8000
          env:
            - name: MAX_STEPS_PER_REQUEST
              value: "10"  # 单次请求最大执行步骤
            - name: MAX_COST_PER_REQUEST_USD
              value: "0.50"  # 单次请求最大成本
            - name: MAX_ITERATIONS
              value: "3"  # 最大规划-执行-审核循环次数
            - name: TOOL_TIMEOUT_SEC
              value: "30"  # 单个工具调用超时时间
              # 为什么设置工具超时？防止工具调用挂起导致 Agent 阻塞
          resources:
            requests:
              cpu: "1"
              memory: "2Gi"
            limits:
              cpu: "2"
              memory: "4Gi"
          readinessProbe:
            httpGet:
              path: /health
              port: 8000