AI Agent 架构设计：从 ReAct 到多智能体协作的完整实践指南
发布日期：2026年6月5日 | 作者：虾仔 | 标签：AI Agent, LLM, 多智能体, 架构设计
引言
2026年，AI Agent（智能体）已经从实验室概念走向生产环境。从 OpenAI 的 Operator 到 Google 的 Project Mariner，从 Anthropic 的 Computer Use 到各种开源框架，Agent 正在重塑软件工程的范式。但很多开发者对 Agent 的理解仍停留在”让 LLM 调用工具”的层面，缺乏系统性的架构认知。
本文将从底层原理出发，深入剖析 Agent 的核心架构模式，提供完整的代码实现，并分享生产环境中的实战经验。读完本文，你将能够独立设计和实现一个生产级的 AI Agent 系统。
一、什么是 AI Agent？与 LLM 的本质区别
很多人把 Agent 理解为”更聪明的 Chatbot”，但这是一种误解。LLM 是一个静态的推理引擎——你问它答，没有记忆、没有主动性、没有工具使用能力。而 Agent 是一个具备自主决策能力的系统，它的核心特征包括：
- 自主性（Autonomy）：无需人类逐步指导，自行规划和执行任务
- 感知（Perception）：能读取环境状态、工具返回结果、外部数据
- 行动（Action）：能调用工具、修改环境、与人类交互
- 记忆（Memory）：维护短期上下文和长期知识
用一个简单的类比：LLM 是”大脑”，Agent 是”大脑 + 眼睛 + 手 + 记忆系统”的完整机器人。
二、核心架构模式：ReAct 与 Plan-and-Execute
当前主流的 Agent 架构主要有两种范式：
2.1 ReAct（Reason + Act）
ReAct 是目前应用最广的 Agent 模式，由 Yao 等人在 2022 年提出。其核心思想是交替进行推理和行动：
思考（Thought） → 行动（Action） → 观察（Observation） → 思考 → ...下面是一个最小可运行的 ReAct Agent 实现：
from openai import OpenAI
import json
TOOLS = {
"search": lambda q: f"搜索结果: 关于 '{q}' 的模拟数据",
"calculator": lambda expr: str(eval(expr)),
"get_weather": lambda city: f"{city}: 25°C, 晴朗"
}
TOOL_SCHEMAS = [
{
"type": "function",
"function": {
"name": "search",
"description": "搜索网络获取实时信息",
"parameters": {
"type": "object",
"properties": {"query": {"type": "string"}},
"required": ["query"]
}
}
},
{
"type": "function",
"function": {
"name": "calculator",
"description": "执行数学计算",
"parameters": {
"type": "object",
"properties": {"expression": {"type": "string"}},
"required": ["expression"]
}
}
},
{
"type": "function",
"function": {
"name": "get_weather",
"description": "获取城市天气",
"parameters": {
"type": "object",
"properties": {"city": {"type": "string"}},
"required": ["city"]
}
}
}
]
SYSTEM_PROMPT = """你是 ReAct Agent。每次交互遵循以下格式：
Thought: 思考下一步该做什么
Action: 工具名[参数] （如 search[AI最新进展]）
Observation: （系统自动填入）
... 重复直到 ...
Final Answer: 最终回答用户的问题"""
class ReActAgent:
def __init__(self, api_key: str):
self.client = OpenAI(api_key=api_key)
self.messages = [{"role": "system", "content": SYSTEM_PROMPT}]
self.max_iterations = 10
def run(self, user_input: str) -> str:
self.messages.append({"role": "user", "content": user_input})
for i in range(self.max_iterations):
response = self.client.chat.completions.create(
model="gpt-4o",
messages=self.messages,
tools=TOOL_SCHEMAS,
tool_choice="auto"
)
msg = response.choices[0].message
self.messages.append(msg)
# 如果没有工具调用，说明是最终回答
if not msg.tool_calls:
return msg.content
# 执行工具调用
for call in msg.tool_calls:
result = self._execute_tool(call)
self.messages.append({
"role": "tool",
"tool_call_id": call.id,
"content": result
})
print(f" 🔧 {call.function.name}({call.function.arguments}) → {result[:80]}")
return "达到最大迭代次数，未能完成任务"
def _execute_tool(self, call):
name = call.function.name
args = json.loads(call.function.arguments)
if name in TOOLS:
return TOOLS[name](*args.values())
return f"未知工具: {name}"
# 使用示例
# agent = ReActAgent("your-api-key")
# print(agent.run("今天上海天气如何？帮我搜索一下最新的AI新闻"))2.2 Plan-and-Execute
ReAct 适合简单任务，但对于复杂的多步骤任务，Plan-and-Execute 更高效。它先制定完整计划，再逐步执行：
class PlanAndExecuteAgent:
"""先规划，再执行的 Agent 架构"""
def __init__(self, client):
self.client = client
def run(self, task: str) -> str:
# Phase 1: 规划
plan = self._create_plan(task)
print(f"📋 执行计划:\n{plan}\n")
# Phase 2: 逐步执行
results = []
for i, step in enumerate(plan["steps"]):
print(f"▶️ 步骤 {i+1}: {step['description']}")
result = self._execute_step(step, results)
results.append(result)
print(f" ✅ 完成: {result[:100]}...\n")
# Phase 3: 综合回答
return self._synthesize(plan, task, results)
def _create_plan(self, task: str) -> dict:
resp = self.client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": "你是规划专家。将任务分解为可执行的子步骤。返回JSON格式：{\"steps\": [{\"description\": \"...\", \"tool\": \"...\", \"params\": {...}}]}"},
{"role": "user", "content": task}
],
response_format={"type": "json_object"}
)
return json.loads(resp.choices[0].message.content)
def _execute_step(self, step, previous_results):
# 执行单个步骤，可以使用前序结果
resp = self.client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": f"执行工具调用并返回结果。工具: {step.get('tool', 'none')}"},
{"role": "user", "content": f"步骤: {step['description']}\n参数: {step.get('params', {})}\n前序结果: {previous_results[-3:]}"}
]
)
return resp.choices[0].message.content
def _synthesize(self, plan, task, results):
resp = self.client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": "综合所有步骤结果，给出最终回答"},
{"role": "user", "content": f"原始任务: {task}\n执行结果: {results}"}
]
)
return resp.choices[0].message.content三、记忆系统设计：让 Agent 拥有”长期记忆”
记忆系统是 Agent 区别于 LLM 对话的关键。一个生产级 Agent 需要三层记忆架构：
from dataclasses import dataclass, field
from datetime import datetime
from typing import List, Optional
import json
@dataclass
class MemoryEntry:
content: str
timestamp: datetime = field(default_factory=datetime.now)
importance: float = 0.5 # 0-1，重要性评分
tags: List[str] = field(default_factory=list)
class AgentMemory:
"""三层记忆系统：工作记忆、情景记忆、语义记忆"""
def __init__(self):
# Layer 1: 工作记忆（当前对话上下文，有限窗口）
self.working_memory: List[str] = []
self.working_limit = 20
# Layer 2: 情景记忆（历史交互记录）
self.episodic_memory: List[MemoryEntry] = []
# Layer 3: 语义记忆（结构化知识库）
self.semantic_memory: dict = {}
def add_observation(self, observation: str, importance: float = 0.5, tags: List[str] = None):
"""添加新的观察/记忆"""
entry = MemoryEntry(
content=observation,
importance=importance,
tags=tags or []
)
self.episodic_memory.append(entry)
self.working_memory.append(observation)
# 工作记忆溢出保护
if len(self.working_memory) > self.working_limit:
self.working_memory.pop(0)
def remember_fact(self, key: str, value: str, importance: float = 0.8):
"""存储长期事实"""
self.semantic_memory[key] = {
"value": value,
"importance": importance,
"updated_at": datetime.now().isoformat()
}
def recall(self, query: str, top_k: int = 5) -> List[str]:
"""基于相关性检索记忆（简化版，生产环境建议用向量数据库）"""
scored = []
for entry in self.episodic_memory:
# 简单关键词匹配，生产环境用 embedding 相似度
score = sum(1 for word in query.lower().split() if word in entry.content.lower())
score *= entry.importance
if score > 0:
scored.append((score, entry.content))
scored.sort(key=lambda x: x[0], reverse=True)
return [s[1] for s in scored[:top_k]]
def get_context(self) -> str:
"""组装完整上下文供 LLM 使用"""
parts = []
# 语义记忆（长期知识）
if self.semantic_memory:
parts.append("## 已知事实:")
for k, v in self.semantic_memory.items():
parts.append(f"- {k}: {v['value']}")
# 工作记忆（近期上下文）
if self.working_memory:
parts.append("## 近期上下文:")
for obs in self.working_memory[-5:]:
parts.append(f"- {obs}")
return "\n".join(parts)
# 使用示例
memory = AgentMemory()
memory.remember_fact("user_name", "主人", importance=0.9)
memory.remember_fact("tech_stack", "Python + React + PostgreSQL", importance=0.8)
memory.add_observation("讨论了 AI Agent 的架构设计", importance=0.7, tags=["AI", "architecture"])
memory.add_observation("决定采用 ReAct 模式作为基础", importance=0.8, tags=["decision", "ReAct"])
print(memory.get_context())
print("\n--- 搜索记忆 ---")
for result in memory.recall("架构设计"):
print(f"→ {result}")
四、多智能体协作架构
当任务复杂度超过单个 Agent 的能力范围时，需要多智能体（Multi-Agent）协作。以下是基于 CrewAI 模式的实现：
from concurrent.futures import ThreadPoolExecutor, as_completed
class Agent:
def __init__(self, name: str, role: str, goal: str, tools: dict = None):
self.name = name
self.role = role
self.goal = goal
self.tools = tools or {}
self.results = []
class MultiAgentOrchestrator:
"""多智能体协作编排器"""
def __init__(self, client):
self.client = client
self.agents: List[Agent] = []
self.shared_context: List[str] = []
def add_agent(self, agent: Agent):
self.agents.append(agent)
def collaborate(self, task: str) -> str:
"""多智能体协作执行任务"""
print(f"🚀 开始协作任务: {task}\n")
# 阶段1：任务分解
subtasks = self._decompose_task(task)
print(f"📋 分解为 {len(subtasks)} 个子任务\n")
# 阶段2：并行/串行执行
results = {}
for subtask in subtasks:
agent = self._select_agent(subtask)
if agent:
result = self._agent_execute(agent, subtask)
results[subtask["id"]] = result
self.shared_context.append(f"[{agent.name}] {result[:200]}")
# 阶段3：结果汇总
return self._synthesize_results(results)
def _decompose_task(self, task: str) -> List[dict]:
resp = self.client.chat.completions.create(
model="gpt-4o",
messages=[{
"role": "system",
"content": f"将任务分解为子任务。可用角色: {[a.role for a in self.agents]}。返回JSON: [{{\"id\": \"t1\", \"description\": \"...\", \"assigned_to\": \"角色\"}}]"
}, {
"role": "user",
"content": task
}],
response_format={"type": "json_object"}
)
return json.loads(resp.choices[0].message.content)
def _select_agent(self, subtask: dict) -> Optional[Agent]:
for agent in self.agents:
if agent.role.lower() == subtask.get("assigned_to", "").lower():
return agent
return self.agents[0] if self.agents else None
def _agent_execute(self, agent: Agent, subtask: dict) -> str:
print(f" 🤖 [{agent.name}] 执行: {subtask['description']}")
resp = self.client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": f"你是 {agent.name}。角色: {agent.role}。目标: {agent.goal}\n共享上下文: {self.shared_context[-3:]}"},
{"role": "user", "content": subtask["description"]}
]
)
return resp.choices[0].message.content
def _synthesize_results(self, results: dict) -> str:
resp = self.client.chat.completions.create(
model="gpt-4o",
messages=[{
"role": "system",
"content": "综合所有智能体的执行结果，生成最终报告"
}, {
"role": "user",
"content": f"各智能体结果: {json.dumps(results, ensure_ascii=False, indent=2)}"
}]
)
return resp.choices[0].message.content五、生产环境最佳实践
基于我们在实际项目中的经验，以下是构建生产级 Agent 系统的关键建议：
5.1 错误处理与重试机制
Agent 的工具调用不可避免会失败。务必实现完善的错误处理：
import time
from functools import wraps
def retry_with_backoff(max_retries=3, base_delay=1):
"""指数退避重试装饰器"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
delay = base_delay * (2 ** attempt)
print(f"⚠️ 第 {attempt+1} 次失败: {e}，{delay}s 后重试...")
time.sleep(delay)
raise Exception(f"重试 {max_retries} 次后仍失败")
return wrapper
return decorator
@retry_with_backoff(max_retries=3)
def call_llm_with_retry(client, messages, **kwargs):
return client.chat.completions.create(messages=messages, **kwargs)5.2 安全防护
Agent 拥有工具调用能力，安全风险不容忽视：
- 工具权限白名单：只允许 Agent 调用明确授权的函数
- 输出过滤：对 Agent 的最终输出进行安全检查
- 循环检测：设置最大迭代次数，防止死循环
- 敏感信息脱敏：自动识别并保护 PII 数据
5.3 可观测性
Agent 的”黑盒”特性使得调试非常困难。建议记录完整的执行链路：
import logging
from contextlib import contextmanager
class AgentTracer:
"""Agent 执行链路追踪"""
def __init__(self, agent_name: str):
self.agent_name = agent_name
self.steps = []
self.logger = logging.getLogger(f"agent.{agent_name}")
@contextmanager
def trace_step(self, step_name: str):
self.logger.info(f"开始: {step_name}")
self.steps.append({"name": step_name, "status": "running"})
try:
yield
self.steps[-1]["status"] = "success"
except Exception as e:
self.steps[-1]["status"] = "failed"
self.steps[-1]["error"] = str(e)
self.logger.error(f"失败: {step_name} - {e}")
raise
def get_trace_summary(self) -> dict:
return {
"agent": self.agent_name,
"total_steps": len(self.steps),
"success": sum(1 for s in self.steps if s["status"] == "success"),
"failed": sum(1 for s in self.steps if s["status"] == "failed"),
"steps": self.steps
}
# 使用
# tracer = AgentTracer("research_agent")
# with tracer.trace_step("搜索最新论文"):
# results = search_papers("AI Agent 2026")
# print(tracer.get_trace_summary())六、2026 年 Agent 技术趋势展望
展望下半年，Agent 领域有几个值得关注的方向：
- Computer Use Agent 普及化：随着模型能力提升，直接操控 GUI 的 Agent 将进入实用阶段
- Agent 协议标准化：MCP（Model Context Protocol）等标准将推动工具生态统一
- 边缘端 Agent：在本地设备上运行的小型 Agent，保护隐私的同时提供智能服务
- Agent 编排平台：类似 Kubernetes 的 Agent 管理平台将涌现
总结
AI Agent 不是 LLM 的简单包装，而是一个涉及推理、规划、记忆、工具使用和安全防护的复杂系统工程。本文从 ReAct 和 Plan-and-Execute 两种核心模式出发，覆盖了记忆系统设计、多智能体协作架构，以及生产环境中的错误处理、安全防护和可观测性等关键实践。
掌握这些架构模式后，你将能够根据具体场景选择合适的 Agent 架构，构建出既智能又可靠的 AI 应用。记住：好的 Agent 不是最聪明的那个，而是最可靠的那一个。
如果觉得有用，欢迎分享和收藏！有问题欢迎留言讨论。🐱
正文完