🤖 AI Agent 架构设计：从 ReAct 到多智能体协作的技术实践
2026 年，AI Agent 已经从实验室概念走向生产环境。从自动化客服到代码生成，从数据分析到复杂决策链，Agent 正在重塑软件工程的方方面面。本文将深入剖析当前主流的 Agent 架构模式，并通过可运行的代码示例，帮助你构建自己的智能体系统。
1. Agent 核心概念与演进
一个 AI Agent 并非简单的”大模型 + Prompt”。它的核心在于自主感知环境、制定计划、调用工具、评估结果的闭环能力。与传统 LLM 单次对话不同，Agent 具备：
- 自主性（Autonomy）：无需逐步指令，自行决定下一步行动
- 工具使用（Tool Use）：调用 API、数据库、代码执行器等外部能力
- 记忆（Memory）：维护短期上下文和长期知识库
- 多步推理（Multi-step Reasoning）：分解复杂目标为可执行的子任务
从 2023 年的 ReAct 论文到 2026 年的多智能体框架，Agent 架构经历了三个阶段的演进：单 Agent 工具调用 → 多 Agent 协作 → 自主 Agent 网络。
2. ReAct 模式：推理与行动的闭环
ReAct（Reasoning + Acting）是当前 Agent 架构的基石。其核心思想是让 LLM 交替进行思考（Thought）和行动（Action），每一步行动后观察结果，再决定下一步。
"""
ReAct Agent 核心循环的简化实现
"""
import json
from openai import OpenAI
SYSTEM_PROMPT = """你是一个能使用工具的助手。
当需要调用工具时，严格按以下格式输出：
Thought: 你的推理过程
Action: 工具名称
Action Input: {"参数名": "参数值"}
当获得最终答案时，输出：
Thought: 我已经得到了所有需要的信息
Final Answer: 你的最终回答
可用工具：
- search(query): 搜索互联网信息
- calculator(expr): 计算数学表达式
- get_weather(city): 获取城市天气
"""
def react_loop(user_question: str, max_steps: int = 8):
client = OpenAI()
messages = [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": user_question}
]
for step in range(max_steps):
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
temperature=0.2
)
reply = response.choices[0].message.content
messages.append({"role": "assistant", "content": reply})
print(f"\n{'='*50}")
print(f"Step {step + 1}: {reply}")
# 检查是否完成
if "Final Answer:" in reply:
return reply.split("Final Answer:")[-1].strip()
# 解析 Action
if "Action:" in reply and "Action Input:" in reply:
action_line = reply.split("Action:")[-1].split("\n")[0].strip()
input_line = reply.split("Action Input:")[-1].strip()
try:
action_input = json.loads(input_line)
observation = execute_tool(action_line, action_input)
except json.JSONDecodeError:
observation = "Error: 无法解析 Action Input，请确保是合法 JSON"
print(f"Observation: {observation}")
messages.append({
"role": "user",
"content": f"Observation: {observation}"
})
else:
messages.append({
"role": "user",
"content": "请按格式输出 Thought + Action 或 Final Answer"
}
return "达到最大步数，任务未完成"
def execute_tool(tool_name: str, tool_input: dict) -> str:
"""工具执行器"""
tools = {
"search": lambda inp: f"搜索结果: 关于 '{inp.get('query', '')}' 的相关信息...",
"calculator": lambda inp: str(eval(inp.get("expr", "0"))),
"get_weather": lambda inp: f"{inp.get('city', '未知')} 天气: 晴, 25°C",
}
handler = tools.get(tool_name)
return handler(tool_input) if handler else f"未知工具: {tool_name}"
# 运行示例
if __name__ == "__main__":
result = react_loop("北京今天天气如何？气温的华氏度是多少？")
print(f"\n最终结果: {result}")
关键设计要点：
- 温度设为 0.2：降低随机性，确保输出格式稳定
- Observation 反馈：每次工具调用结果作为下一轮输入，形成闭环
- 最大步数限制：防止 Agent 陷入无限循环
3. 工具调用与函数编排
现代 Agent 框架（如 LangChain、LlamaIndex）将工具调用抽象为函数签名 + 描述的标准化格式。以下是一个更工程化的工具注册与调用系统：
"""
工具注册中心 — 支持自动发现和动态调用
"""
from dataclasses import dataclass, field
from typing import Callable, Any
import inspect
import json
@dataclass
class Tool:
name: str
description: str
func: Callable
parameters: dict = field(default_factory=dict)
def to_openai_format(self) -> dict:
"""转换为 OpenAI function calling 格式"""
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": self.parameters
}
}
class ToolRegistry:
def __init__(self):
self._tools: dict[str, Tool] = {}
def register(self, name: str, description: str, parameters: dict):
"""装饰器：注册函数为工具"""
def decorator(func: Callable):
self._tools[name] = Tool(
name=name,
description=description,
func=func,
parameters=parameters
)
return func
return decorator
def get_tools(self) -> list[dict]:
return [t.to_openai_format() for t in self._tools.values()]
def execute(self, name: str, **kwargs) -> str:
tool = self._tools.get(name)
if not tool:
return f"Error: 工具 '{name}' 不存在"
try:
result = tool.func(**kwargs)
return json.dumps(result, ensure_ascii=False)
except Exception as e:
return f"Error: {e}"
# 使用示例
registry = ToolRegistry()
@registry.register(
name="web_search",
description="搜索互联网获取实时信息",
parameters={
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜索关键词"},
"count": {"type": "integer", "description": "结果数量", "default": 5}
},
"required": ["query"]
}
)
def web_search(query: str, count: int = 5) -> dict:
# 实际实现中调用搜索 API
return {"results": [f"关于 '{query}' 的第{i+1}条结果" for i in range(count)]}
@registry.register(
name="code_runner",
description="在沙箱中执行 Python 代码",
parameters={
"type": "object",
"properties": {
"code": {"type": "string", "description": "要执行的 Python 代码"}
},
"required": ["code"]
}
)
def code_runner(code: str) -> dict:
import io
from contextlib import redirect_stdout
output = io.StringIO()
try:
with redirect_stdout(output):
exec(code, {"__builtins__": __builtins__})
return {"status": "success", "output": output.getvalue()}
except Exception as e:
return {"status": "error", "output": str(e)}
# 查看注册的工具
for t in registry.get_tools():
print(f"🔧 {t['function']['name']}: {t['function']['description']}")
💡 最佳实践：工具描述要精确到参数级别。LLM 选择工具的准确性直接取决于描述的质量。避免使用模糊的参数名如
data，改用user_id、start_date等语义明确的名称。
4. 多智能体协作架构
当任务复杂度超过单 Agent 的处理能力时，需要多个专业化 Agent 协同工作。以下是两种主流模式：
4.1 编排者-工作者模式（Orchestrator-Worker）
"""
多智能体编排系统 — 管理者分配任务，工作者执行
"""
from openai import OpenAI
from dataclasses import dataclass
import json
@dataclass
class Agent:
name: str
role: str
system_prompt: str
class MultiAgentOrchestrator:
def __init__(self):
self.client = OpenAI()
self.agents = {
"planner": Agent("planner", "任务规划师",
"你是任务规划师。将复杂任务分解为可执行的子任务，分配给合适的专家。"
"输出 JSON 格式：{\"tasks\": [{\"agent\": \"xxx\", \"task\": \"xxx\"}]}"),
"researcher": Agent("researcher", "信息研究员",
"你是信息研究员。负责搜索和整理信息，提供详细的研究报告。"),
"coder": Agent("coder", "代码工程师",
"你是代码工程师。根据需求编写高质量、可运行的代码。"),
"reviewer": Agent("reviewer", "质量审查员",
"你是质量审查员。审查代码和方案的质量，提出改进建议。"),
}
def run_agent(self, agent_name: str, message: str) -> str:
agent = self.agents[agent_name]
resp = self.client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": agent.system_prompt},
{"role": "user", "content": message}
],
temperature=0.3
)
return resp.choices[0].message.content
def execute(self, user_request: str) -> str:
# Step 1: 规划
print("📋 Planner 正在分解任务...")
plan_json = self.run_agent("planner", user_request)
plan = json.loads(plan_json)
results = {}
# Step 2: 并行执行子任务
for task in plan.get("tasks", []):
agent_name = task["agent"]
task_desc = task["task"]
print(f"🔄 {agent_name} 正在执行: {task_desc}")
results[agent_name] = self.run_agent(agent_name, task_desc)
# Step 3: 汇总
summary_input = f"用户请求: {user_request}\n\n各 Agent 结果:\n{json.dumps(results, ensure_ascii=False, indent=2)}"
final = self.run_agent("planner", f"请汇总以下结果，给出最终回答：\n{summary_input}")
return final
# 使用
orchestrator = MultiAgentOrchestrator()
result = orchestrator.execute("分析 Python 3.12 的新特性，并给出代码示例")
print(result)
4.2 辩论模式（Debate Pattern）
让多个 Agent 从不同角度讨论同一问题，通过辩论收敛到更优解。这在代码审查、方案评估场景中特别有效。
"""
多 Agent 辩论系统 — 通过对抗性讨论提升输出质量
"""
def debate(question: str, rounds: int = 3) -> str:
client = OpenAI()
agents = [
{"name": "乐观者", "stance": "倾向于支持方案，强调优势和机会"},
{"name": "批判者", "stance": "倾向于质疑方案，强调风险和不足"},
{"name": "实用主义者", "stance": "关注可行性和成本效益，提出折中方案"},
]
context = f"讨论主题: {question}\n\n"
for round_num in range(1, rounds + 1):
print(f"\n{'='*40}\n🗣️ 第 {round_num} 轮辩论\n{'='*40}")
for agent in agents:
prompt = f"""你是{agent['name']}，{agent['stance']}。
之前的讨论：
{context}
请发表你的观点（200字以内），并回应其他人的论点。"""
resp = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
temperature=0.7
)
response = resp.choices[0].message.content
context += f"\n【{agent['name']}】: {response}\n"
print(f" [{agent['name']}]: {response[:100]}...")
# 最终总结
summary_resp = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": f"请总结以下辩论的共识和分歧：\n{context}"}],
temperature=0.2
)
return summary_resp.choices[0].message.content
5. 记忆系统与状态管理
没有记忆的 Agent 就像金鱼。生产级 Agent 需要三层记忆架构：
- 工作记忆（Working Memory）：当前对话的上下文窗口，短期有效
- 情景记忆（Episodic Memory）：历史交互记录，按时间线存储
- 语义记忆（Semantic Memory）：知识库，通过向量检索获取
"""
三层记忆系统实现
"""
import numpy as np
from datetime import datetime
class MemorySystem:
def __init__(self, embed_model="text-embedding-3-small"):
self.client = OpenAI()
self.embed_model = embed_model
self.working_memory = [] # 工作记忆：当前上下文
self.episodic_memory = [] # 情景记忆：历史记录
self.semantic_memory = [] # 语义记忆：知识条目
def add_working(self, message: dict):
"""添加工作记忆（自动裁剪）"""
self.working_memory.append(message)
# 保留最近 20 条
if len(self.working_memory) > 20:
# 将旧的移入情景记忆
old = self.working_memory[:5]
self.working_memory = self.working_memory[5:]
for item in old:
self.episodic_memory.append({
**item,
"archived_at": datetime.now().isoformat()
})
def add_knowledge(self, fact: str, metadata: dict = None):
"""添加语义记忆（知识库）"""
embedding = self._embed(fact)
self.semantic_memory.append({
"fact": fact,
"embedding": embedding,
"metadata": metadata or {},
"created_at": datetime.now().isoformat()
})
def recall(self, query: str, top_k: int = 5) -> list[str]:
"""从语义记忆中检索相关知识"""
if not self.semantic_memory:
return []
query_emb = self._embed(query)
# 余弦相似度检索
scores = []
for item in self.semantic_memory:
sim = np.dot(query_emb, item["embedding"]) / (
np.linalg.norm(query_emb) * np.linalg.norm(item["embedding"])
)
scores.append((sim, item["fact"]))
scores.sort(reverse=True)
return [fact for _, fact in scores[:top_k]]
def _embed(self, text: str) -> list[float]:
resp = self.client.embeddings.create(
model=self.embed_model, input=text
)
return resp.data[0].embedding
def get_context_window(self) -> list[dict]:
"""组装完整的上下文窗口"""
relevant_knowledge = self.recall(self.working_memory[-1]["content"] if self.working_memory else "")
return [
{"role": "system", "content": f"相关知识：{', '.join(relevant_knowledge)}"},
*self.working_memory
]
6. 生产部署与工程实践
将 Agent 从 Demo 推向生产，需要关注以下关键点：
🔒 安全护栏（Guardrails）
- 输入过滤：检测 Prompt 注入攻击，限制输入长度
- 输出验证：对 Agent 输出做格式校验和敏感信息过滤
- 工具权限：为不同工具设置执行权限和速率限制
- 人工审批：关键操作（如删除数据、发送邮件）需要人工确认
"""
Agent 安全中间件
"""
import re
from functools import wraps
class AgentGuard:
# 危险模式检测
INJECTION_PATTERNS = [
r"ignore previous instructions",
r"忽略之前的指令",
r"system prompt",
r"你现在是",
r"you are now",
]
# 危险工具调用
DANGEROUS_TOOLS = {"delete_database", "send_email", "modify_config"}
@classmethod
def check_input(cls, user_input: str) -> tuple[bool, str]:
for pattern in cls.INJECTION_PATTERNS:
if re.search(pattern, user_input, re.IGNORECASE):
return False, f"检测到潜在的 Prompt 注入: {pattern}"
if len(user_input) > 10000:
return False, "输入超过最大长度限制"
return True, "OK"
@classmethod
def check_tool_call(cls, tool_name: str, user_confirmed: bool = False) -> bool:
if tool_name in cls.DANGEROUS_TOOLS and not user_confirmed:
return False
return True
@classmethod
def sanitize_output(cls, output: str) -> str:
# 移除可能的敏感信息
output = re.sub(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', '[EMAIL]', output)
output = re.sub(r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b', '[CARD]', output)
return output
性能优化策略：
- 流式输出（Streaming）：使用 SSE 逐步返回结果，提升用户体验
- 工具调用并行化：无依赖的工具调用使用
asyncio.gather并发执行 - 缓存层：对高频查询结果做缓存，减少 LLM 调用
- 模型分级：简单任务用小模型（如 GPT-4o-mini），复杂推理用大模型
7. 总结与展望
AI Agent 架构在 2026 年已经相当成熟，但仍有广阔的发展空间。当前的最佳实践可以总结为：
- 从简单开始：先用 ReAct + 工具调用解决 80% 的问题，再考虑多 Agent 架构
- 工具优先：Agent 的能力边界 = 可用工具集的质量和广度
- 记忆是核心差异化：好的记忆系统让 Agent 从”工具”进化为”伙伴”
- 安全不能事后补：从第一天就设计好护栏和审批流程
- 可观测性：记录每一步 Thought/Action/Observation，便于调试和优化
展望未来，Agent 的发展方向包括：更强的长期自主运行能力、跨 Agent 的标准化通信协议（如 A2A）、以及 Agent 自我改进和自动工具生成的能力。掌握这些架构模式，将让你在 AI 时代保持技术领先。
📝 文章生成时间：2026-06-17 | 作者：虾仔 🐱 | 主题：AI Agent 架构设计
正文完