AI Agent 架构设计：从单次推理到多轮规划的工程演进

一、当 LLM 不再"一问一答"：Agent 的工程挑战

一个客服系统，用户问"帮我退掉昨天的订单"，LLM 直接生成了退款操作。但用户昨天有 3 笔订单，退哪一笔？LLM 不知道，因为它没有"先查询再确认"的规划能力。这不是模型能力不足，而是架构设计缺失——缺少让模型"思考后再行动"的框架。

AI Agent 的核心，是让 LLM 从"单次推理器"进化为"多轮规划执行器"。单次推理是"问什么答什么"，Agent 是"理解意图、拆解任务、调用工具、观察结果、调整策略"。这五步循环，就是 Agent 的基本骨架。

Agent 的设计，像极了易经中的"变"——每一次工具调用都是一次"变"，观察结果是"变"的反馈，调整策略是"应变"。不变的，是目标导向的规划逻辑。本文将从架构层面拆解 Agent 的设计模式，给出生产级的实现方案。

二、从 ReAct 到 Plan-and-Execute：Agent 架构的演进脉络

Agent 架构的核心问题：如何平衡"思考深度"与"执行效率"。

graph TB
    subgraph ReAct 循环
        R1[观察 Observation] --> R2[思考 Thought]
        R2 --> R3[行动 Action]
        R3 --> R4[观察结果]
        R4 --> R2
    end
    subgraph Plan-and-Execute
        P1[用户目标] --> P2[规划器: 生成任务列表]
        P2 --> P3[执行器: 逐步执行]
        P3 --> P4{任务完成?}
        P4 -->|否| P5[重新规划]
        P5 --> P3
        P4 -->|是| P6[输出结果]
    end
    subgraph 多 Agent 协作
        M1[编排 Agent] --> M2[搜索 Agent]
        M1 --> M3[代码 Agent]
        M1 --> M4[分析 Agent]
        M2 --> M5[结果汇总]
        M3 --> M5
        M4 --> M5
    end
    style R2 fill:#fff9c4
    style P2 fill:#fff9c4
    style M1 fill:#fff9c4

1. ReAct：思考-行动-观察的循环

ReAct（Reasoning + Acting）是最经典的 Agent 模式。每一步，LLM 先"思考"当前状态和下一步策略，再选择一个"行动"（工具调用），然后"观察"行动结果，进入下一轮循环。优点是灵活，每一步都可以根据观察调整策略。缺点是效率低——简单任务也需要多轮 LLM 调用。

2. Plan-and-Execute：先规划后执行

Plan-and-Execute 将规划与执行分离。规划器（通常是更强的 LLM）一次性生成任务列表，执行器逐个执行。如果执行过程中发现计划不合理，触发重新规划。优点是减少了 LLM 调用次数，缺点是规划质量依赖 LLM 的推理能力，复杂任务容易规划失败。

3. 多 Agent 协作：分而治之

将不同能力分配给专门的 Agent：搜索 Agent 负责信息检索，代码 Agent 负责编程，分析 Agent 负责数据处理。编排 Agent 负责任务分配和结果汇总。优点是每个 Agent 的 system prompt 更聚焦，缺点是 Agent 间通信成本高，编排逻辑复杂。

三、生产级 Agent 框架：模块化设计与工具管理

import json
import re
import logging
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any, Optional
from enum import Enum

logger = logging.getLogger(__name__)


class ToolCallStatus(Enum):
    """工具调用状态"""
    SUCCESS = "success"
    ERROR = "error"
    TIMEOUT = "timeout"


@dataclass
class ToolCallResult:
    """工具调用结果"""
    status: ToolCallStatus
    output: Any
    error: Optional[str] = None
    execution_time_ms: float = 0.0


class BaseTool(ABC):
    """工具基类：所有 Agent 工具必须实现此接口"""

    @property
    @abstractmethod
    def name(self) -> str:
        """工具名称，用于 LLM 调用"""
        pass

    @property
    @abstractmethod
    def description(self) -> str:
        """工具描述，LLM 据此判断何时调用"""
        pass

    @abstractmethod
    def parameters_schema(self) -> dict:
        """工具参数的 JSON Schema"""
        pass

    @abstractmethod
    def execute(self, **kwargs) -> ToolCallResult:
        """执行工具逻辑"""
        pass

    def to_openai_format(self) -> dict:
        """转换为 OpenAI function calling 格式"""
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": self.parameters_schema(),
            },
        }


class DatabaseQueryTool(BaseTool):
    """数据库查询工具示例"""

    @property
    def name(self) -> str:
        return "query_orders"

    @property
    def description(self) -> str:
        return (
            "查询用户订单信息。当用户提到订单、退款、物流等需要查数据库时使用。"
            "返回订单列表，包含订单号、金额、状态、创建时间。"
        )

    def parameters_schema(self) -> dict:
        return {
            "type": "object",
            "properties": {
                "user_id": {
                    "type": "string",
                    "description": "用户ID",
                },
                "date_range": {
                    "type": "string",
                    "description": "日期范围，格式: YYYY-MM-DD~YYYY-MM-DD",
                },
                "status": {
                    "type": "string",
                    "enum": ["all", "pending", "completed", "refunded"],
                    "description": "订单状态筛选，默认 all",
                },
            },
            "required": ["user_id"],
        }

    def execute(self, **kwargs) -> ToolCallResult:
        """执行数据库查询"""
        try:
            user_id = kwargs.get("user_id")
            date_range = kwargs.get("date_range", "")
            status = kwargs.get("status", "all")

            # 生产环境中这里连接真实数据库
            # 此处模拟查询结果
            orders = [
                {"order_id": "ORD-001", "amount": 299.0, "status": "completed"},
                {"order_id": "ORD-002", "amount": 158.0, "status": "pending"},
                {"order_id": "ORD-003", "amount": 89.0, "status": "completed"},
            ]

            if status != "all":
                orders = [o for o in orders if o["status"] == status]

            return ToolCallResult(
                status=ToolCallStatus.SUCCESS,
                output=orders,
            )
        except Exception as e:
            logger.error(f"数据库查询失败: {e}")
            return ToolCallResult(
                status=ToolCallStatus.ERROR,
                output=None,
                error=str(e),
            )


@dataclass
class AgentState:
    """Agent 状态：记录对话历史和执行上下文"""
    messages: list = field(default_factory=list)
    tool_results: list = field(default_factory=list)
    iteration: int = 0
    max_iterations: int = 10
    task_complete: bool = False

    def add_observation(self, content: str):
        """添加观察结果"""
        self.messages.append({"role": "user", "content": content})

    def add_thought(self, content: str):
        """添加思考过程"""
        self.messages.append({"role": "assistant", "content": f"[思考] {content}"})

    def check_iteration_limit(self) -> bool:
        """检查是否超过最大迭代次数"""
        self.iteration += 1
        if self.iteration >= self.max_iterations:
            logger.warning(
                f"Agent 已达最大迭代次数 {self.max_iterations}，强制终止"
            )
            self.task_complete = True
            return True
        return False


class ReActAgent:
    """
    ReAct 模式 Agent：思考-行动-观察循环

    核心流程：
    1. LLM 根据当前状态生成思考 + 工具调用
    2. 执行工具调用，获取结果
    3. 将结果作为观察反馈给 LLM
    4. 重复直到任务完成或达到迭代上限
    """

    def __init__(
        self,
        llm_client,  # OpenAI 兼容的 LLM 客户端
        tools: list[BaseTool],
        system_prompt: str = "",
        max_iterations: int = 10,
    ):
        self.llm = llm_client
        self.tools = {tool.name: tool for tool in tools}
        self.system_prompt = system_prompt or self._default_system_prompt()
        self.max_iterations = max_iterations

    def _default_system_prompt(self) -> str:
        return (
            "你是一个智能助手，能够通过调用工具完成用户任务。\n"
            "请按以下格式回复：\n"
            "1. 先思考当前情况和下一步策略\n"
            "2. 如果需要调用工具，使用 function calling\n"
            "3. 如果已有足够信息，直接给出最终答案\n"
            "注意：每次只调用一个工具，观察结果后再决定下一步。"
        )

    def run(self, user_input: str) -> str:
        """执行 Agent 循环"""
        state = AgentState(max_iterations=self.max_iterations)
        state.messages = [
            {"role": "system", "content": self.system_prompt},
            {"role": "user", "content": user_input},
        ]

        while not state.task_complete:
            if state.check_iteration_limit():
                break

            # 调用 LLM
            try:
                response = self.llm.chat.completions.create(
                    model="gpt-4",
                    messages=state.messages,
                    tools=[t.to_openai_format() for t in self.tools.values()],
                    tool_choice="auto",
                )
            except Exception as e:
                logger.error(f"LLM 调用失败: {e}")
                return f"抱歉，服务暂时不可用: {str(e)}"

            message = response.choices[0].message

            # 检查是否有工具调用
            if message.tool_calls:
                for tool_call in message.tool_calls:
                    result = self._execute_tool(tool_call)
                    # 将工具结果加入对话历史
                    state.messages.append(message)
                    state.messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": json.dumps(
                            result.output if result.status == ToolCallStatus.SUCCESS
                            else {"error": result.error},
                            ensure_ascii=False,
                        ),
                    })
            else:
                # 没有工具调用，说明 LLM 认为任务完成
                state.task_complete = True
                return message.content or ""

        return "任务执行超时，请稍后重试。"

    def _execute_tool(self, tool_call) -> ToolCallResult:
        """执行单个工具调用"""
        tool_name = tool_call.function.name
        tool = self.tools.get(tool_name)

        if tool is None:
            logger.error(f"未知工具: {tool_name}")
            return ToolCallResult(
                status=ToolCallStatus.ERROR,
                output=None,
                error=f"未知工具: {tool_name}",
            )

        try:
            arguments = json.loads(tool_call.function.arguments)
        except json.JSONDecodeError as e:
            logger.error(f"工具参数解析失败: {e}")
            return ToolCallResult(
                status=ToolCallStatus.ERROR,
                output=None,
                error=f"参数格式错误: {str(e)}",
            )

        logger.info(f"调用工具: {tool_name}, 参数: {arguments}")
        return tool.execute(**arguments)

四、Agent 架构的权衡：灵活性、成本与可靠性的三角博弈

1. LLM 调用成本与延迟

ReAct 模式每一步都需要 LLM 推理，一个复杂任务可能需要 5-10 轮调用。以 GPT-4 的价格计算，单次 Agent 执行成本可能在 $0.5-2 之间。Plan-and-Execute 减少了调用次数，但规划失败时需要重新规划，成本不可预测。生产环境中，必须设置调用次数上限和单次成本上限。

2. 工具调用的可靠性问题

LLM 生成的工具参数可能不符合 schema，比如日期格式错误、枚举值越界。解决方案：在工具执行前做参数校验，校验失败时将错误信息反馈给 LLM 重试。但这增加了调用轮次。更根本的方案是优化工具的 description 和 schema，让 LLM 更容易生成正确参数。

3. 多 Agent 协作的通信开销

Agent 间通过自然语言通信，信息压缩损失大。搜索 Agent 返回的 10 条结果，传递给分析 Agent 时可能只保留摘要。解决方案：定义结构化的 Agent 间通信协议，用 JSON 而非自然语言传递中间结果。

4. 安全边界：Agent 的权限控制

Agent 可以调用工具，就意味着可以执行操作。退款、发邮件、删除数据——这些操作一旦被 LLM 错误触发，后果严重。生产环境中，所有写操作必须经过人工确认，或设置操作白名单和频率限制。

五、总结

AI Agent 的架构设计，核心是在灵活性、成本和可靠性之间找平衡。ReAct 适合需要动态决策的复杂任务，Plan-and-Execute 适合步骤明确的流程性任务，多 Agent 协作适合能力异构的综合性任务。

落地路线建议：第一，从 ReAct 模式起步，它最简单也最灵活。第二，工具设计遵循"单一职责"，每个工具的 description 要精确描述触发条件和参数含义。第三，所有写操作必须加确认环节，禁止 Agent 直接执行不可逆操作。第四，设置调用次数上限和超时机制，防止 Agent 陷入死循环。第五，生产环境必须记录完整的 Agent 执行轨迹，用于调试和审计。Agent 不是万能的，但架构得当，它是 LLM 从"对话工具"走向"工作伙伴"的关键一步。