AI Agent 系统架构设计：多工具协同与记忆机制的工程化实现
·
AI Agent 系统架构设计：多工具协同与记忆机制的工程化实现
一、从单次对话到持续行动：Agent 系统的核心挑战
大语言模型的能力边界正在从"文本生成"向"自主行动"延伸，AI Agent 成为这一趋势的核心载体。然而，将 LLM 从被动的问答工具升级为主动执行任务的 Agent，面临三个关键工程挑战。
第一，工具调用的可靠性问题。LLM 生成的工具调用参数经常出现格式错误、参数缺失或类型不匹配，在生产环境中这类错误率可达 15%-30%。一个格式错误的 JSON 参数就能让整个 Agent 执行链中断。第二，多工具协同的状态管理。当 Agent 需要依次调用搜索、数据库查询、代码执行等多个工具时，前序工具的输出必须可靠地传递给后续工具，而 LLM 的上下文窗口有限，长链路调用容易丢失中间状态。第三，持久化记忆的缺失。标准 LLM 调用是无状态的，Agent 无法记住历史交互中的关键信息，导致重复执行相同操作或遗忘用户偏好。
这些问题的本质是：Agent 系统需要在 LLM 的非确定性输出之上，构建确定性的执行框架。如同在混沌的宇宙中建立秩序，Agent 架构设计的核心任务是将 LLM 的"灵性"约束在工程可控的轨道上。
二、ReAct 循环与工具编排的执行机制
现代 Agent 系统普遍采用 ReAct（Reasoning + Acting）范式：LLM 先推理下一步应该做什么，然后执行对应的工具调用，观察执行结果后再推理下一步。这一循环持续进行，直到任务完成或达到最大步数限制。
graph TD
START[用户输入] --> PLAN[推理：分析任务与规划步骤]
PLAN --> DECIDE{需要调用工具？}
DECIDE -->|是| SELECT[选择工具与参数]
SELECT --> VALID{参数校验}
VALID -->|通过| EXEC[执行工具调用]
VALID -->|失败| RETRY[请求 LLM 修正参数]
RETRY --> SELECT
EXEC --> OBS[观察执行结果]
OBS --> MEM[更新记忆状态]
MEM --> CHECK{任务完成？}
DECIDE -->|否| RESPOND[生成最终回复]
CHECK -->|是| RESPOND
CHECK -->|否| PLAN
RESPOND --> END[返回结果]
subgraph 记忆系统
SM[短期记忆：当前对话上下文]
LM[长期记忆：向量数据库]
WM[工作记忆：中间结果缓存]
end
MEM --> SM
MEM --> LM
MEM --> WM
SM -.-> PLAN
LM -.-> PLAN
WM -.-> PLAN
style EXEC fill:#4ecdc4,color:#fff
style MEM fill:#ff6b6b,color:#fff
style RETRY fill:#ffe66d,color:#333
ReAct 循环的关键设计决策在于：何时终止循环、如何处理工具调用失败、如何管理不断增长的上下文。终止条件通常包括 LLM 输出终止标记、达到最大步数、或连续多次工具调用失败。工具调用失败时，将错误信息注入 LLM 上下文让其自我修正，比直接重试更有效。
记忆系统分为三层：短期记忆存储当前对话的完整上下文，受限于 LLM 的上下文窗口；长期记忆将历史信息向量化后存入向量数据库，通过语义检索按需召回；工作记忆缓存当前任务的中间结果（如数据库查询结果、代码执行输出），避免重复计算。
三、生产级 Agent 框架实现
from __future__ import annotations
import json
import logging
import re
import time
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Optional
logger = logging.getLogger(__name__)
# ============================================================
# 工具定义与参数校验
# ============================================================
class ToolParameterType(str, Enum):
STRING = "string"
INTEGER = "integer"
FLOAT = "float"
BOOLEAN = "boolean"
ARRAY = "array"
@dataclass
class ToolParameter:
"""工具参数定义：名称、类型、是否必填、默认值。
设计动机：LLM 生成的参数经常类型不匹配或缺少必填项，
显式的参数定义使得校验逻辑可自动化执行，
无需在每个工具实现中重复编写校验代码。
"""
name: str
param_type: ToolParameterType
required: bool = True
default: Any = None
description: str = ""
@dataclass
class ToolDefinition:
"""工具完整定义：包含名称、描述、参数列表与执行函数。"""
name: str
description: str
parameters: list[ToolParameter]
execute_fn: Any # Callable
def validate_params(self, params: dict) -> dict:
"""校验并补全 LLM 生成的参数。
设计动机：LLM 可能输出字符串 "123" 而非整数 123，
自动类型转换减少因格式问题导致的调用失败。
缺少必填参数时抛出明确异常，而非让工具执行时才报错。
"""
validated = {}
for p in self.parameters:
if p.name not in params:
if p.required:
raise ValueError(
f"工具 {self.name} 缺少必填参数: {p.name}"
)
validated[p.name] = p.default
continue
value = params[p.name]
# 自动类型转换：容忍 LLM 输出的类型偏差
try:
if p.param_type == ToolParameterType.INTEGER:
value = int(value)
elif p.param_type == ToolParameterType.FLOAT:
value = float(value)
elif p.param_type == ToolParameterType.BOOLEAN:
if isinstance(value, str):
value = value.lower() in ("true", "1", "yes")
value = bool(value)
except (ValueError, TypeError) as e:
raise ValueError(
f"工具 {self.name} 参数 {p.name} 类型转换失败: {e}"
) from e
validated[p.name] = value
return validated
def to_function_schema(self) -> dict:
"""生成 OpenAI Function Calling 格式的工具描述。"""
properties = {}
required = []
for p in self.parameters:
properties[p.name] = {
"type": p.param_type.value,
"description": p.description,
}
if p.required:
required.append(p.name)
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": {
"type": "object",
"properties": properties,
"required": required,
},
},
}
# ============================================================
# 记忆系统
# ============================================================
class AgentMemory:
"""Agent 记忆管理器：短期、长期、工作三层记忆协同。
设计动机：LLM 的上下文窗口有限（通常 4K-128K tokens），
将所有历史信息塞入上下文既浪费 token 又降低推理质量。
三层记忆架构按需加载信息，如同修行者的"三识"——
短期记忆是"意识"，长期记忆是"藏识"，工作记忆是"执识"。
"""
def __init__(self, max_short_term_messages: int = 20):
self._short_term: list[dict] = []
self._working: dict[str, Any] = {}
self._max_short_term = max_short_term_messages
def add_message(self, role: str, content: str) -> None:
"""添加短期记忆消息。"""
self._short_term.append({"role": role, "content": content})
# 滑动窗口：保留最近的对话消息
if len(self._short_term) > self._max_short_term:
# 保留系统消息和最近的消息
system_msgs = [m for m in self._short_term if m["role"] == "system"]
other_msgs = [m for m in self._short_term if m["role"] != "system"]
keep = other_msgs[-self._max_short_term:]
self._short_term = system_msgs + keep
def set_working(self, key: str, value: Any) -> None:
"""存储工作记忆：缓存中间结果避免重复计算。"""
self._working[key] = value
def get_working(self, key: str, default: Any = None) -> Any:
"""检索工作记忆。"""
return self._working.get(key, default)
def get_context(self) -> list[dict]:
"""获取当前完整上下文：短期记忆 + 工作记忆摘要。"""
context = list(self._short_term)
if self._working:
summary = "当前工作记忆缓存:\n"
for k, v in self._working.items():
summary += f"- {k}: {str(v)[:200]}\n"
context.append({"role": "system", "content": summary})
return context
# ============================================================
# Agent 执行引擎
# ============================================================
class AgentStatus(str, Enum):
IDLE = "idle"
RUNNING = "running"
WAITING_TOOL = "waiting_tool"
COMPLETED = "completed"
FAILED = "failed"
@dataclass
class AgentConfig:
"""Agent 配置：控制执行循环的行为边界。"""
max_steps: int = 10
max_retries_per_tool: int = 2
tool_call_timeout: float = 30.0
class Agent:
"""AI Agent 执行引擎：ReAct 循环 + 工具编排 + 记忆管理。
设计动机：Agent 的核心不是 LLM 本身，而是围绕 LLM 构建
确定性的执行框架。LLM 负责"思考"，框架负责"行动"，
二者各司其职，如同阴阳互补——LLM 是变化的"阳"，
框架是守恒的"阴"，阴阳和合方能成事。
"""
def __init__(self, llm_client: Any, config: Optional[AgentConfig] = None):
self.llm = llm_client
self.config = config or AgentConfig()
self.tools: dict[str, ToolDefinition] = {}
self.memory = AgentMemory()
self.status = AgentStatus.IDLE
self._step_count = 0
def register_tool(self, tool: ToolDefinition) -> None:
"""注册工具到 Agent。"""
if tool.name in self.tools:
logger.warning("工具 %s 已存在，将被覆盖", tool.name)
self.tools[tool.name] = tool
def _parse_tool_call(self, llm_output: str) -> Optional[tuple[str, dict]]:
"""从 LLM 输出中解析工具调用。
设计动机：LLM 的工具调用格式可能不稳定，
使用正则 + JSON 解析的双重策略提高鲁棒性。
解析失败时返回 None，由上层逻辑决定是否重试。
"""
# 尝试匹配 <tool_call name="..." params='...'> 格式
pattern = r'<tool_call\s+name="(\w+)"\s+params=[\'"](.+?)[\'"]\s*/?>'
match = re.search(pattern, llm_output, re.DOTALL)
if match:
tool_name = match.group(1)
try:
params = json.loads(match.group(2))
return tool_name, params
except json.JSONDecodeError:
logger.warning("工具调用参数 JSON 解析失败: %s", match.group(2))
return None
# 尝试匹配 JSON 格式 {"tool": "...", "params": {...}}
try:
data = json.loads(llm_output.strip())
if isinstance(data, dict) and "tool" in data:
return data["tool"], data.get("params", {})
except json.JSONDecodeError:
pass
return None
async def execute_tool(self, tool_name: str, params: dict) -> str:
"""执行工具调用：参数校验 + 执行 + 错误处理。"""
if tool_name not in self.tools:
available = ", ".join(self.tools.keys())
return f"错误：未知工具 '{tool_name}'，可用工具: {available}"
tool = self.tools[tool_name]
try:
validated_params = tool.validate_params(params)
except ValueError as e:
return f"参数校验失败: {e}"
try:
result = await tool.execute_fn(**validated_params)
# 将结果存入工作记忆
self.memory.set_working(f"tool_{tool_name}_last_result", result)
return str(result)
except Exception as e:
logger.error("工具 %s 执行失败: %s", tool_name, e)
return f"工具执行错误: {type(e).__name__}: {e}"
async def run(self, user_input: str) -> str:
"""执行 Agent 主循环：ReAct 推理-行动循环。"""
self.status = AgentStatus.RUNNING
self._step_count = 0
self.memory.add_message("user", user_input)
while self._step_count < self.config.max_steps:
self._step_count += 1
context = self.memory.get_context()
try:
# 调用 LLM 进行推理
response = await self.llm.chat(
messages=context,
tools=[t.to_function_schema() for t in self.tools.values()],
)
assistant_msg = response.choices[0].message
content = assistant_msg.content or ""
self.memory.add_message("assistant", content)
# 检查是否包含工具调用
tool_call = self._parse_tool_call(content)
if tool_call is None:
# 无工具调用，视为最终回复
self.status = AgentStatus.COMPLETED
return content
tool_name, params = tool_call
self.status = AgentStatus.WAITING_TOOL
# 执行工具并记录结果
result = await self.execute_tool(tool_name, params)
self.memory.add_message("system", f"工具 {tool_name} 执行结果: {result}")
except Exception as e:
logger.error("Agent 步骤 %d 执行异常: %s", self._step_count, e)
self.status = AgentStatus.FAILED
return f"Agent 执行失败（步骤 {self._step_count}）: {e}"
self.status = AgentStatus.COMPLETED
return "Agent 已达到最大执行步数限制，任务可能未完全完成。"
四、Agent 架构的工程代价与可靠性边界
Agent 系统的工程代价远超普通 LLM 应用，主要体现在三个方面。
Token 消耗的不可预测性。ReAct 循环中，每一步都需要将完整的上下文发送给 LLM，上下文长度随步数线性增长。一个 10 步的 Agent 任务可能消耗 5 万 tokens，是单次对话的 10 倍以上。在成本敏感的生产环境中，必须设置 max_steps 上限和 token 预算。
执行延迟的累积效应。每一步 ReAct 循环包含一次 LLM 推理和一次工具执行，串行累积的延迟可能达到数十秒。对于实时交互场景，这远超用户可接受的响应时间。解决方案包括：将独立工具调用并行化、使用流式输出减少感知延迟、对简单任务跳过 ReAct 直接执行。
错误传播与级联失败。Agent 的每一步都依赖前一步的输出，一步错误可能导致后续所有步骤偏离正确方向。LLM 的幻觉在 Agent 场景中危害更大——一个虚构的工具名称或参数会导致整个执行链崩溃。缓解策略包括：工具调用前的参数校验、执行结果的合理性检查、关键步骤的人工确认。
适用边界：Agent 架构适用于多步骤、多工具、需要动态决策的复杂任务（如数据分析、代码生成、信息检索与整合）；对于单步即可完成的简单任务，直接调用 LLM 即可，Agent 框架只会增加不必要的延迟与成本。
五、总结
AI Agent 系统的架构设计，核心是在 LLM 的非确定性与工程系统的确定性之间建立桥梁。ReAct 循环提供了推理与行动交替的基本范式，工具定义与参数校验确保了 LLM 输出到系统执行的可靠映射，三层记忆架构解决了上下文有限与信息保留的矛盾。
落地路线建议：第一步，从单工具 Agent 开始，验证工具调用与参数校验的可靠性；第二步，引入多工具编排，建立工具选择与执行顺序的评估基线；第三步，实现三层记忆系统，优先完善工作记忆的中间结果缓存；第四步，建立 Agent 执行的监控体系，追踪步数、token 消耗、工具调用成功率等核心指标；第五步，在关键决策节点引入人工确认机制，确保 Agent 行为的可控性。
欢迎加入 MCP 技术社区！与志同道合者携手前行，一同解锁 MCP 技术的无限可能！
更多推荐
0
0- 0
已为社区贡献17条内容
所有评论(0)