AI Agent 系统设计：从单轮推理到多工具协同的架构演进

一、从 Chatbot 到 Agent：大模型能力的边界与突破

大语言模型在单轮问答场景中表现优异，但面对需要多步推理、外部工具调用和状态管理的复杂任务时，纯文本生成模式暴露出根本性缺陷：模型无法获取实时信息（如当前天气、股票行情），无法执行确定性操作（如数据库查询、API 调用），且在长链推理中容易产生幻觉累积。

以一个"分析竞品定价策略并生成报告"的任务为例，Agent 需要依次完成：搜索竞品信息、调用数据库获取历史定价、用计算工具分析趋势、将结果写入文档。这要求模型从"生成文本"进化为"规划并执行动作序列"，即从 Chatbot 演进为 Agent。

Agent 的核心架构由三个组件构成：规划器（Planner）负责任务分解与步骤编排，执行器（Executor）负责调用工具并处理返回结果，记忆模块（Memory）负责维护上下文状态和对话历史。三者协同的效率决定了 Agent 的任务完成率。

二、Agent 核心架构的底层机制

2.1 ReAct 框架：推理与行动的交织

ReAct（Reasoning + Acting）是目前最主流的 Agent 框架。其核心思想是将推理（Thought）和行动（Action）交替执行，每一步推理都基于前一步的观察结果（Observation），形成 Thought-Action-Observation 的循环。

sequenceDiagram
    participant U as 用户
    participant P as Planner (LLM)
    participant E as Executor
    participant T as 工具集

    U->>P: 提交任务
    P->>P: Thought 1: 需要先获取竞品信息
    P->>E: Action 1: search("竞品A 最新定价")
    E->>T: 调用搜索工具
    T-->>E: Observation 1: 竞品A 定价 ¥299/月
    E-->>P: 返回结果

    P->>P: Thought 2: 需要对比历史数据
    P->>E: Action 2: query_db("竞品A 价格历史")
    E->>T: 调用数据库工具
    T-->>E: Observation 2: 近6个月价格从¥399降至¥299
    E-->>P: 返回结果

    P->>P: Thought 3: 信息已充分，生成分析报告
    P->>E: Action 3: write_report(分析结果)
    E->>T: 调用文档工具
    T-->>E: Observation 3: 报告已生成
    E-->>P: 返回结果

    P-->>U: 任务完成，返回报告

ReAct 的优势在于推理过程可解释：每一步 Thought 都记录了模型的决策依据，便于调试和优化。但其劣势也很明显——每一步都需要调用 LLM 生成 Thought，延迟和成本随步骤数线性增长。

2.2 工具调用的函数签名机制

现代 Agent 通过 Function Calling 实现工具调用。模型接收工具的函数签名描述，在需要时输出结构化的调用请求，而非自由文本。

from pydantic import BaseModel, Field
from typing import Literal
import json


class ToolDefinition(BaseModel):
    """工具定义：描述工具的名称、参数和用途"""
    name: str = Field(description="工具名称，全局唯一标识")
    description: str = Field(description="工具功能描述，供 LLM 理解用途")
    parameters: dict = Field(description="JSON Schema 格式的参数定义")


# 定义工具集
TOOLS = [
    ToolDefinition(
        name="search_web",
        description="搜索互联网获取实时信息，适用于查询新闻、价格、天气等动态数据",
        parameters={
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "搜索关键词"
                },
                "num_results": {
                    "type": "integer",
                    "description": "返回结果数量，默认5",
                    "default": 5
                }
            },
            "required": ["query"]
        }
    ),
    ToolDefinition(
        name="execute_sql",
        description="执行 SQL 查询并返回结果，仅支持 SELECT 语句",
        parameters={
            "type": "object",
            "properties": {
                "sql": {
                    "type": "string",
                    "description": "SQL 查询语句"
                },
                "database": {
                    "type": "string",
                    "enum": ["analytics", "pricing", "users"],
                    "description": "目标数据库名称"
                }
            },
            "required": ["sql", "database"]
        }
    ),
]


def build_tool_prompt(tools: list[ToolDefinition]) -> str:
    """将工具定义转换为 LLM 可理解的系统提示"""
    tool_schemas = [t.model_dump() for t in tools]
    return f"""你可以使用以下工具来完成任务：

{json.dumps(tool_schemas, ensure_ascii=False, indent=2)}

当你需要调用工具时，请输出如下格式：
```json
{{"tool_call": {{"name": "工具名", "arguments": {{参数}}}}}}

如果你已经收集到足够信息，直接输出最终答案，无需调用工具。"""

## 三、生产级 Agent 框架实现

### 3.1 带状态管理与错误恢复的 Agent 循环

```python
import openai
from dataclasses import dataclass, field
from typing import Optional


@dataclass
class AgentState:
    """Agent 运行状态：维护对话历史与工具调用记录"""
    messages: list[dict] = field(default_factory=list)
    tool_calls_log: list[dict] = field(default_factory=list)
    max_iterations: int = 10
    current_iteration: int = 0


class ProductionAgent:
    """生产级 Agent：支持工具调用、错误恢复与迭代限制"""

    def __init__(self, model: str = "gpt-4o", tools: list[dict] = None):
        self.model = model
        self.tools = tools or []
        self.tool_implementations = {
            "search_web": self._search_web,
            "execute_sql": self._execute_sql,
        }

    async def run(self, task: str) -> str:
        """执行 Agent 任务，返回最终结果"""
        state = AgentState()
        system_prompt = build_tool_prompt(self.tools)

        state.messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": task},
        ]

        while state.current_iteration < state.max_iterations:
            state.current_iteration += 1

            try:
                response = await openai.ChatCompletion.acreate(
                    model=self.model,
                    messages=state.messages,
                    temperature=0.1,    # 低温度保证推理稳定性
                )
            except openai.error.RateLimitError:
                # 限流时指数退避重试
                import asyncio
                await asyncio.sleep(2 ** state.current_iteration)
                continue
            except openai.error.APIError as e:
                return f"API 调用失败: {e}"

            assistant_msg = response.choices[0].message
            state.messages.append(assistant_msg.to_dict())

            # 检查是否包含工具调用
            tool_call = self._parse_tool_call(assistant_msg.content)
            if tool_call is None:
                # 无工具调用，任务完成
                return assistant_msg.content

            # 执行工具调用
            tool_name = tool_call["name"]
            tool_args = tool_call["arguments"]

            if tool_name not in self.tool_implementations:
                observation = f"错误：未知工具 '{tool_name}'"
            else:
                try:
                    observation = await self.tool_implementations[tool_name](**tool_args)
                except Exception as e:
                    observation = f"工具执行失败: {type(e).__name__}: {e}"

            # 记录工具调用日志
            state.tool_calls_log.append({
                "iteration": state.current_iteration,
                "tool": tool_name,
                "args": tool_args,
                "result_preview": str(observation)[:200],
            })

            # 将观察结果反馈给 LLM
            state.messages.append({
                "role": "user",
                "content": f"工具返回结果：{observation}"
            })

        return "任务未完成：已达到最大迭代次数限制"

    @staticmethod
    def _parse_tool_call(content: str) -> Optional[dict]:
        """从 LLM 输出中解析工具调用请求"""
        import re
        pattern = r'```json\s*(\{.*?\})\s*```'
        match = re.search(pattern, content, re.DOTALL)
        if match:
            try:
                parsed = json.loads(match.group(1))
                if "tool_call" in parsed:
                    return parsed["tool_call"]
            except json.JSONDecodeError:
                pass
        return None

    @staticmethod
    async def _search_web(query: str, num_results: int = 5) -> str:
        """搜索工具实现（示例）"""
        # 实际场景接入搜索 API
        return f"搜索结果：找到 {num_results} 条关于 '{query}' 的信息"

    @staticmethod
    async def _execute_sql(sql: str, database: str) -> str:
        """数据库查询工具实现（示例）"""
        # 安全校验：仅允许 SELECT 语句
        if not sql.strip().upper().startswith("SELECT"):
            return "错误：仅允许执行 SELECT 查询"
        return f"查询结果：从 {database} 执行 {sql[:50]}..."

3.2 多 Agent 协作架构

class MultiAgentOrchestrator:
    """多 Agent 编排器：将复杂任务分配给专业 Agent"""

    def __init__(self):
        self.agents = {
            "researcher": ProductionAgent(
                model="gpt-4o",
                tools=[t for t in TOOLS if t.name in ["search_web"]]
            ),
            "analyst": ProductionAgent(
                model="gpt-4o",
                tools=[t for t in TOOLS if t.name in ["execute_sql"]]
            ),
            "writer": ProductionAgent(
                model="gpt-4o",
                tools=[]   # 写作 Agent 无需工具
            ),
        }

    async def execute(self, task: str) -> str:
        """三阶段流水线：调研 → 分析 → 写作"""
        # 阶段1：调研 Agent 收集信息
        research_result = await self.agents["researcher"].run(task)

        # 阶段2：分析 Agent 处理数据
        analysis_prompt = f"基于以下调研结果进行数据分析：\n{research_result}"
        analysis_result = await self.agents["analyst"].run(analysis_prompt)

        # 阶段3：写作 Agent 生成报告
        writing_prompt = (
            f"基于以下调研和分析结果，撰写专业报告：\n"
            f"调研：{research_result}\n"
            f"分析：{analysis_result}"
        )
        final_report = await self.agents["writer"].run(writing_prompt)

        return final_report

四、Agent 架构的边界分析与权衡

4.1 ReAct 框架的延迟问题

ReAct 的每一步都需要 LLM 推理，10 步任务的端到端延迟可能超过 30 秒。对于实时性要求高的场景（如客服对话），这一延迟不可接受。替代方案是 Plan-and-Execute：先由 LLM 一次性生成完整计划，再逐步执行，减少 LLM 调用次数。但 Plan-and-Execute 的代价是缺乏灵活性——如果某一步的观察结果与预期不符，需要重新规划。

4.2 工具调用的可靠性瓶颈

LLM 生成的工具调用参数可能不符合 Schema 约束（如类型错误、缺少必填参数）。实测中，GPT-4o 的工具调用格式正确率约为 95%，仍有 5% 的失败率需要重试或兜底处理。对于关键操作（如数据库写入），必须在工具实现层增加校验和确认机制。

4.3 上下文窗口的约束

Agent 的对话历史随迭代次数增长，可能超出模型的上下文窗口。以 GPT-4o 的 128K Token 限制为例，10 轮工具调用可能消耗 20K-50K Token。解决方案包括：滑动窗口截断早期历史、对历史进行摘要压缩、或使用向量数据库存储长期记忆。

4.4 适用边界总结

五、总结

AI Agent 的核心价值在于将 LLM 从文本生成器升级为任务执行器。ReAct 框架通过 Thought-Action-Observation 循环实现了推理与行动的交织，Function Calling 机制使工具调用从自由文本变为结构化请求。生产环境中，必须关注迭代限制、错误恢复、工具校验和上下文管理，才能构建可靠的 Agent 系统。

落地路线建议：从单 Agent + ReAct 起步，使用 2-3 个核心工具验证任务完成率；当任务复杂度增加时，引入 Plan-and-Execute 减少延迟；最终根据业务领域拆分为多 Agent 协作架构。始终监控每步工具调用的成功率和平均延迟，以此作为 Agent 迭代优化的核心指标。