Agent 核心架构

本章深入讲解 Agent 系统中最重要的三种架构模式。

ReAct 架构

ReAct (Reasoning + Acting) 是目前最主流的 Agent 架构。核心思想：交替进行推理和行动。

ReAct 的运行过程：

用户: 北京今天天气怎么样？适合户外活动吗？
Thought: 用户想知道北京的天气，我需要先查询天气信息。
Action: search("北京今天天气")
Observation: 北京今天晴，气温 15-25°C，空气质量良好。
Thought: 天气晴朗，气温舒适，空气质量好。适合户外活动。
Action: 直接回答
Answer: 北京今天天气很好！晴天，气温 15-25°C，空气质量良好，非常适合户外活动。

"""
ReAct Agent 实现
"""
from openai import OpenAI
import json
import re
class ReActAgent:
"""ReAct 模式 Agent"""
SYSTEM_PROMPT = """你是一个使用 ReAct 模式的 AI 助手。
在每一步中，你需要：
1. **Thought**: 分析当前情况，决定下一步
2. **Action**: 选择并执行一个工具
3. **Observation**: 观察工具返回的结果
可用工具：
{tools_desc}
格式要求：
Thought: <你的推理过程>
Action: <工具名称>(<参数>)
当你有了最终答案时：
Thought: <总结推理>
Answer: <最终回答>"""
def __init__(self, model: str = "gpt-4o"):
self.client = OpenAI()
self.model = model
self.tools: dict[str, dict] = {}
def register_tool(self, name: str, func, description: str):
"""注册工具"""
self.tools[name] = {"function": func, "description": description}
def run(self, task: str, max_steps: int = 8) -> str:
"""运行 ReAct 循环"""
# 构建工具描述
tools_desc = "\n".join(
f"- {name}: {info['description']}"
for name, info in self.tools.items()
)
system = self.SYSTEM_PROMPT.format(tools_desc=tools_desc)
messages = [
{"role": "system", "content": system},
{"role": "user", "content": task},
]
for step in range(max_steps):
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
temperature=0,
)
output = response.choices[0].message.content
print(f"\n=== Step {step + 1} ===\n{output}")
messages.append({"role": "assistant", "content": output})
# 检查是否得出最终答案
if "Answer:" in output:
answer = output.split("Answer:")[-1].strip()
return answer
# 解析并执行 Action
action_match = re.search(r"Action:\s*(\w+)\((.+?)\)", output)
if action_match:
tool_name = action_match.group(1)
tool_input = action_match.group(2).strip("\"'")
if tool_name in self.tools:
result = self.tools[tool_name]["function"](tool_input)
observation = f"Observation: {result}"
else:
observation = f"Observation: 错误 - 工具 '{tool_name}' 不存在"
print(observation)
messages.append({"role": "user", "content": observation})
else:
messages.append({
"role": "user",
"content": "请使用正确的格式: Action: 工具名(参数) 或给出 Answer:",
})
return "达到最大步骤数，未能完成任务。"

Plan-and-Execute 架构

与 ReAct 边想边做不同，Plan-and-Execute 先制定完整计划，再逐步执行。

"""
Plan-and-Execute Agent 实现
"""
class PlanAndExecuteAgent:
"""先规划再执行的 Agent"""
def __init__(self, model: str = "gpt-4o"):
self.client = OpenAI()
self.model = model
self.tools: dict[str, dict] = {}
def register_tool(self, name: str, func, description: str):
self.tools[name] = {"function": func, "description": description}
def run(self, task: str) -> str:
"""运行 Plan-and-Execute"""
# 阶段 1: 规划
plan = self._create_plan(task)
print(f"\n📋 执行计划 ({len(plan)} 步):")
for i, step in enumerate(plan, 1):
print(f"  {i}. {step}")
# 阶段 2: 逐步执行
results = []
for i, step in enumerate(plan):
print(f"\n🔧 执行步骤 {i + 1}: {step}")
result = self._execute_step(step, results)
results.append({"step": step, "result": result})
print(f"  结果: {result[:100]}")
# 阶段 3: 汇总
final = self._summarize(task, results)
return final
def _create_plan(self, task: str) -> list[str]:
"""创建执行计划"""
tools_desc = "\n".join(
f"- {name}: {info['description']}"
for name, info in self.tools.items()
)
response = self.client.chat.completions.create(
model=self.model,
messages=[
{
"role": "system",
"content": f"""你是一个任务规划器。将复杂任务分解为可执行的步骤。
可用工具: {tools_desc}
请以 JSON 数组格式返回步骤列表: ["步骤1", "步骤2", ...]""",
},
{"role": "user", "content": f"任务: {task}"},
],
temperature=0,
)
try:
plan = json.loads(response.choices[0].message.content)
return plan if isinstance(plan, list) else [task]
except json.JSONDecodeError:
return [task]
def _execute_step(self, step: str, previous: list[dict]) -> str:
"""执行单个步骤"""
context = "\n".join(
f"步骤: {r['step']}\n结果: {r['result']}"
for r in previous[-3:]
)
tools_desc = "\n".join(
f"- {name}: {info['description']}"
for name, info in self.tools.items()
)
response = self.client.chat.completions.create(
model=self.model,
messages=[
{
"role": "system",
"content": f"""执行以下步骤。可用工具: {tools_desc}
如需调用工具，回复: TOOL: 工具名(参数)
如果可以直接回答，直接给出结果。
之前的执行结果:
{context}""",
},
{"role": "user", "content": step},
],
temperature=0,
)
output = response.choices[0].message.content
# 解析工具调用
tool_match = re.search(r"TOOL:\s*(\w+)\((.+?)\)", output)
if tool_match:
name = tool_match.group(1)
arg = tool_match.group(2).strip("\"'")
if name in self.tools:
return str(self.tools[name]["function"](arg))
return output
def _summarize(self, task: str, results: list[dict]) -> str:
"""汇总结果"""
context = "\n".join(
f"- {r['step']}: {r['result'][:200]}" for r in results
)
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "根据执行结果，给出最终回答。"},
{"role": "user", "content": f"任务: {task}\n\n执行结果:\n{context}"},
],
temperature=0,
)
return response.choices[0].message.content

架构对比

选择建议

本章小结

• ReAct 是最基础的架构，交替「推理-行动-观察」
• Plan-and-Execute 更适合复杂的多步骤任务
• 根据任务特点选择合适的架构
• 实际项目中常常混合使用多种模式

下一章：学习工具调用（Function Calling）机制，这是 Agent 的核心能力。