Agent 架构设计模式详解
AI Agent 是大模型应用的核心形态。如何设计高效的 Agent 架构？如何选择合适的设计模式？本文将深入解析 Agent 的主流架构设计模式及实战应用。
一、Agent 架构演进
1.1 架构演进历程
Agent 架构演进：
V1.0 - 简单对话 (2022)
└── 单轮对话，无状态
V2.0 - 工具调用 (2023)
└── Function Calling, Tool Use
V3.0 - 规划执行 (2023)
└── ReAct, Plan-and-Solve
V4.0 - 反思优化 (2024)
└── Self-Reflection, Critique
V5.0 - 多 Agent 协作 (2024)
└── Multi-Agent, Collaboration1.2 核心架构模式
主流 Agent 架构模式：
┌─────────────────────────────────────────┐
│ 模式 │ 复杂度 │ 能力 │ 场景 │
├─────────────────────────────────────────┤
│ ReAct │ 中 │ 强 │ 通用 │
│ Plan-and-Solve│ 中 │ 强 │ 推理 │
│ Reflection │ 高 │ 最强 │ 复杂 │
│ Tool Use │ 低 │ 中 │ 简单 │
│ Multi-Agent │ 最高 │ 最强 │ 协作 │
└─────────────────────────────────────────┘二、ReAct 架构模式
2.1 ReAct 原理
graph LR
A[问题] --> B[思考]
B --> C{需要工具？}
C -->|是 | D[行动]
C -->|否 | E[回答]
D --> F[观察]
F --> B
E --> G[结果]
ReAct 核心思想：
- Reasoning（推理）：逐步思考问题
- Acting（行动）：调用工具获取信息
- 交替进行：推理和行动交替执行
2.2 ReAct 实现
# react_agent.py
from typing import List, Dict, Optional
from dataclasses import dataclass
from enum import Enum
class StepType(Enum):
"""步骤类型"""
THOUGHT = "thought"
ACTION = "action"
OBSERVATION = "observation"
ANSWER = "answer"
@dataclass
class ReActStep:
"""ReAct 步骤"""
type: StepType
content: str
class ReActAgent:
"""ReAct Agent"""
def __init__(
self,
llm,
tools: List[dict],
max_iterations: int = 10
):
"""
初始化
Args:
llm: LLM 模型
tools: 工具列表
max_iterations: 最大迭代次数
"""
self.llm = llm
self.tools = {tool['name']: tool for tool in tools}
self.max_iterations = max_iterations
def run(self, question: str) -> Dict:
"""
运行 Agent
Args:
question: 用户问题
Returns:
执行结果
"""
steps: List[ReActStep] = []
for i in range(self.max_iterations):
# 生成下一步
step = self._generate_step(question, steps)
steps.append(step)
# 如果是思考步骤，继续
if step.type == StepType.THUGHT:
continue
# 如果是行动步骤，执行工具
if step.type == StepType.ACTION:
action = self._parse_action(step.content)
if action:
observation = self._execute_action(action)
steps.append(ReActStep(
type=StepType.OBSERVATION,
content=observation
))
else:
steps.append(ReActStep(
type=StepType.THUGHT,
content="Invalid action format"
))
continue
# 如果是回答，结束
if step.type == StepType.ANSWER:
return {
'answer': step.content,
'steps': steps,
'iterations': i + 1
}
# 达到最大迭代次数
return {
'answer': 'Failed to find answer within max iterations',
'steps': steps,
'iterations': self.max_iterations
}
def _generate_step(
self,
question: str,
steps: List[ReActStep]
) -> ReActStep:
"""生成下一步"""
# 构建 Prompt
prompt = self._build_prompt(question, steps)
# 调用 LLM
response = self.llm.generate(prompt)
# 解析响应
return self._parse_response(response)
def _build_prompt(
self,
question: str,
steps: List[ReActStep]
) -> str:
"""构建 Prompt"""
prompt_parts = [
"你是一个智能助手，使用 ReAct 模式回答问题。",
"",
"你可以使用以下工具：",
]
# 添加工具描述
for tool in self.tools.values():
prompt_parts.append(
f"- {tool['name']}: {tool['description']}"
)
prompt_parts.extend([
"",
"请使用以下格式：",
"Thought: 你的思考",
"Action: 工具名",
"Action Input: 工具参数",
"Observation: 工具返回结果",
"... (重复 Thought/Action/Observation)",
"Thought: 我现在知道答案了",
"Answer: 最终答案",
"",
"开始！",
"",
f"Question: {question}",
])
# 添加历史步骤
for step in steps:
if step.type == StepType.THUGHT:
prompt_parts.append(f"Thought: {step.content}")
elif step.type == StepType.ACTION:
prompt_parts.append(f"Action: {step.content}")
elif step.type == StepType.OBSERVATION:
prompt_parts.append(f"Observation: {step.content}")
return '\n'.join(prompt_parts)
def _parse_action(self, content: str) -> Optional[Dict]:
"""解析行动"""
import json
import re
# 解析 Action Input
match = re.search(r'Action Input:\s*(.+)', content, re.DOTALL)
if not match:
return None
try:
action_input = json.loads(match.group(1).strip())
# 解析 Action
action_match = re.search(r'Action:\s*(\w+)', content)
if action_match:
return {
'action': action_match.group(1),
'input': action_input
}
except:
pass
return None
def _execute_action(self, action: Dict) -> str:
"""执行行动"""
tool_name = action['action']
tool_input = action['input']
if tool_name not in self.tools:
return f"Error: Unknown tool '{tool_name}'"
tool = self.tools[tool_name]
try:
result = tool['function'](**tool_input)
return str(result)
except Exception as e:
return f"Error: {str(e)}"
def _parse_response(self, response: str) -> ReActStep:
"""解析响应"""
response = response.strip()
# 判断类型
if response.startswith('Answer:'):
return ReActStep(
type=StepType.ANSWER,
content=response.replace('Answer:', '').strip()
)
elif response.startswith('Action:'):
return ReActStep(
type=StepType.ACTION,
content=response
)
else:
return ReActStep(
type=StepType.THUGHT,
content=response.replace('Thought:', '').strip()
)2.3 ReAct 工具定义
# react_tools.py
from typing import Callable, Dict
def create_tools() -> Dict:
"""创建工具集"""
def search(query: str) -> str:
"""搜索工具"""
import requests
response = requests.get(
'https://api.example.com/search',
params={'q': query}
)
results = response.json().get('results', [])
return '\n'.join([r['title'] for r in results[:5]])
def calculate(expression: str) -> str:
"""计算器工具"""
try:
result = eval(expression)
return str(result)
except Exception as e:
return f"Error: {str(e)}"
def get_current_time() -> str:
"""获取当前时间"""
from datetime import datetime
return datetime.now().strftime('%Y-%m-%d %H:%M:%S')
return {
'search': {
'name': 'search',
'description': '搜索互联网信息',
'function': search,
'parameters': {
'query': {'type': 'string', 'description': '搜索关键词'}
}
},
'calculate': {
'name': 'calculate',
'description': '执行数学计算',
'function': calculate,
'parameters': {
'expression': {'type': 'string', 'description': '数学表达式'}
}
},
'get_current_time': {
'name': 'get_current_time',
'description': '获取当前时间',
'function': get_current_time,
'parameters': {}
}
}三、Plan-and-Solve 架构
3.1 架构原理
Plan-and-Solve 流程：
┌─────────────────────────────────────┐
│ 1. 理解问题 │
│ - 分析问题需求 │
│ - 识别关键信息 │
├─────────────────────────────────────┤
│ 2. 制定计划 │
│ - 分解为子任务 │
│ - 确定执行顺序 │
├─────────────────────────────────────┤
│ 3. 执行计划 │
│ - 按顺序执行子任务 │
│ - 收集中间结果 │
├─────────────────────────────────────┤
│ 4. 整合答案 │
│ - 汇总所有结果 │
│ - 生成最终答案 │
└─────────────────────────────────────┘3.2 Plan-and-Solve 实现
# plan_and_solve_agent.py
from typing import List, Dict, Optional
from dataclasses import dataclass
@dataclass
class SubTask:
"""子任务"""
id: int
description: str
status: str = 'pending' # pending, completed, failed
result: Optional[str] = None
class PlanAndSolveAgent:
"""Plan-and-Solve Agent"""
def __init__(self, llm):
self.llm = llm
def run(self, question: str) -> Dict:
"""运行 Agent"""
# 1. 理解问题
understanding = self._understand_problem(question)
# 2. 制定计划
plan = self._create_plan(question, understanding)
# 3. 执行计划
results = self._execute_plan(plan)
# 4. 整合答案
answer = self._synthesize_answer(
question,
understanding,
plan,
results
)
return {
'answer': answer,
'understanding': understanding,
'plan': plan,
'results': results
}
def _understand_problem(self, question: str) -> str:
"""理解问题"""
prompt = f"""
请分析以下问题：
问题：{question}
请分析：
1. 问题的核心需求是什么？
2. 需要哪些关键信息？
3. 可能的难点是什么？
"""
return self.llm.generate(prompt)
def _create_plan(
self,
question: str,
understanding: str
) -> List[SubTask]:
"""创建计划"""
prompt = f"""
基于以下问题和分析，制定解决计划：
问题：{question}
分析：
{understanding}
请制定详细的解决计划，将问题分解为多个可执行的子任务。
每个子任务应该：
1. 有明确的目标
2. 可以独立执行
3. 有清晰的输出
请按以下格式输出：
1. [子任务 1 描述]
2. [子任务 2 描述]
...
"""
plan_text = self.llm.generate(prompt)
# 解析计划
tasks = []
for i, line in enumerate(plan_text.split('\n'), 1):
if line.strip() and line[0].isdigit():
task = SubTask(
id=i,
description=line.split('.', 1)[-1].strip()
)
tasks.append(task)
return tasks
def _execute_plan(
self,
plan: List[SubTask]
) -> List[Dict]:
"""执行计划"""
results = []
for task in plan:
# 执行子任务
result = self._execute_subtask(task)
task.result = result['output']
task.status = result['status']
results.append({
'task_id': task.id,
'description': task.description,
'status': task.status,
'output': task.result
})
return results
def _execute_subtask(self, task: SubTask) -> Dict:
"""执行子任务"""
prompt = f"""
请执行以下子任务：
任务：{task.description}
请详细执行该任务，并提供完整的输出。
"""
try:
output = self.llm.generate(prompt)
return {
'status': 'completed',
'output': output
}
except Exception as e:
return {
'status': 'failed',
'output': str(e)
}
def _synthesize_answer(
self,
question: str,
understanding: str,
plan: List[SubTask],
results: List[Dict]
) -> str:
"""整合答案"""
# 构建结果摘要
results_summary = '\n'.join([
f"任务 {r['task_id']}: {r['description']}\n"
f"状态：{r['status']}\n"
f"结果：{r['output']}\n"
for r in results
])
prompt = f"""
基于以下信息，整合最终答案：
问题：{question}
理解：
{understanding}
执行结果：
{results_summary}
请综合以上信息，给出完整、准确的最终答案。
"""
return self.llm.generate(prompt)四、反射架构
4.1 反射原理
反射架构流程：
┌─────────────────────────────────────┐
│ 初始执行 │
│ ↓ │
│ 生成初步答案 │
│ ↓ │
│ 自我反思 │
│ - 检查答案质量 │
│ - 识别潜在问题 │
│ ↓ │
│ 批评改进 │
│ - 提出改进建议 │
│ - 修正答案 │
│ ↓ │
│ 最终输出 │
└─────────────────────────────────────┘4.2 反射 Agent 实现
# reflection_agent.py
from typing import Dict, List
class ReflectionAgent:
"""反射 Agent"""
def __init__(self, llm):
self.llm = llm
def run(self, question: str) -> Dict:
"""运行带反思的 Agent"""
# 1. 初始执行
initial_response = self._initial_execution(question)
# 2. 自我反思
reflection = self._self_reflect(
question,
initial_response
)
# 3. 批评改进
critique = self._critique(
question,
initial_response,
reflection
)
# 4. 修正答案
revised_response = self._revise_answer(
question,
initial_response,
critique
)
# 5. 质量评估
quality_score = self._evaluate_quality(
question,
revised_response
)
return {
'initial_response': initial_response,
'reflection': reflection,
'critique': critique,
'revised_response': revised_response,
'quality_score': quality_score
}
def _initial_execution(self, question: str) -> str:
"""初始执行"""
prompt = f"""
请回答以下问题：
问题：{question}
请给出你的答案。
"""
return self.llm.generate(prompt)
def _self_reflect(
self,
question: str,
response: str
) -> str:
"""自我反思"""
prompt = f"""
请反思以下答案的质量：
问题：{question}
答案：
{response}
请从以下角度反思：
1. 答案是否完整回答了问题？
2. 是否有事实错误？
3. 逻辑是否清晰？
4. 是否有遗漏的重要信息？
5. 表达是否准确？
请详细列出发现的问题。
"""
return self.llm.generate(prompt)
def _critique(
self,
question: str,
response: str,
reflection: str
) -> str:
"""批评改进"""
prompt = f"""
基于以下反思，提出改进建议：
问题：{question}
原答案：
{response}
反思：
{reflection}
请针对反思中发现的问题，提出具体的改进建议。
"""
return self.llm.generate(prompt)
def _revise_answer(
self,
question: str,
initial_response: str,
critique: str
) -> str:
"""修正答案"""
prompt = f"""
请根据批评建议改进答案：
问题：{question}
原答案：
{initial_response}
批评建议：
{critique}
请综合以上信息，给出改进后的完整答案。
"""
return self.llm.generate(prompt)
def _evaluate_quality(
self,
question: str,
response: str
) -> float:
"""评估质量"""
prompt = f"""
请评估以下答案的质量（0-10 分）：
问题：{question}
答案：
{response}
评分标准：
- 完整性：是否完整回答问题
- 准确性：是否有事实错误
- 逻辑性：逻辑是否清晰
- 简洁性：是否简洁明了
请只输出分数（0-10 的数字）。
"""
score_text = self.llm.generate(prompt)
# 提取分数
import re
match = re.search(r'(\d+)', score_text)
if match:
return min(10, max(0, int(match.group(1))))
return 5.0五、架构模式对比
5.1 模式对比表
| 模式 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|
| ReAct | 灵活、可解释 | 迭代次数多 | 需要工具调用 |
| Plan-and-Solve | 结构化、系统 | 规划成本高 | 复杂推理任务 |
| Reflection | 质量高、自优化 | 延迟高 | 高质量要求 |
| Tool Use | 简单、快速 | 能力有限 | 简单任务 |
| Multi-Agent | 能力强、分工 | 复杂度高 | 复杂协作 |
5.2 选择指南
架构选择决策树：
1. 是否需要调用工具？
├── 是 → 2
└── 否 → 3
2. 是否需要多轮推理？
├── 是 → ReAct
└── 否 → Tool Use
3. 任务是否复杂？
├── 是 → 4
└── 否 → 简单对话
4. 是否需要高质量？
├── 是 → Reflection
└── 否 → 5
5. 是否可以分解？
├── 是 → Plan-and-Solve
└── 否 → ReAct六、实战案例
6.1 数据分析 Agent
# data_analysis_agent.py
from react_agent import ReActAgent
def create_data_analysis_agent(llm):
"""创建数据分析 Agent"""
def load_data(file_path: str) -> str:
"""加载数据"""
import pandas as pd
df = pd.read_csv(file_path)
return f"Shape: {df.shape}\nColumns: {list(df.columns)}"
def describe_data() -> str:
"""描述数据"""
return df.describe().to_string()
def run_query(query: str) -> str:
"""运行查询"""
return df.query(query).to_string()
def calculate_statistic(
column: str,
operation: str
) -> str:
"""计算统计"""
if operation == 'mean':
return str(df[column].mean())
elif operation == 'sum':
return str(df[column].sum())
# ... 其他操作
tools = {
'load_data': {
'name': 'load_data',
'description': '加载 CSV 数据文件',
'function': load_data,
'parameters': {
'file_path': {'type': 'string'}
}
},
'describe_data': {
'name': 'describe_data',
'description': '获取数据统计描述',
'function': describe_data,
'parameters': {}
},
'run_query': {
'name': 'run_query',
'description': '执行数据查询',
'function': run_query,
'parameters': {
'query': {'type': 'string'}
}
},
'calculate_statistic': {
'name': 'calculate_statistic',
'description': '计算统计指标',
'function': calculate_statistic,
'parameters': {
'column': {'type': 'string'},
'operation': {'type': 'string'}
}
}
}
return ReActAgent(llm, tools)6.2 代码生成 Agent
# code_generation_agent.py
from plan_and_solve_agent import PlanAndSolveAgent
def create_code_generation_agent(llm):
"""创建代码生成 Agent"""
agent = PlanAndSolveAgent(llm)
# 重写执行子任务方法
original_execute = agent._execute_subtask
def execute_with_validation(task):
"""执行并验证代码"""
result = original_execute(task)
if result['status'] == 'completed':
# 验证生成的代码
code = result['output']
validation = _validate_code(code)
result['validation'] = validation
return result
agent._execute_subtask = execute_with_validation
return agent
def _validate_code(code: str) -> Dict:
"""验证代码"""
import ast
try:
ast.parse(code)
return {
'valid': True,
'message': 'Syntax valid'
}
except SyntaxError as e:
return {
'valid': False,
'message': f'Syntax error: {str(e)}'
}七、总结
7.1 核心要点
-
ReAct
- 推理和行动交替
- 适合工具调用场景
- 可解释性强
-
Plan-and-Solve
- 先规划后执行
- 适合复杂推理
- 结构化好
-
Reflection
- 自我反思改进
- 适合高质量要求
- 自优化能力
7.2 最佳实践
-
根据场景选择架构
- 简单任务：Tool Use
- 中等复杂：ReAct
- 高度复杂：Plan-and-Solve + Reflection
-
优化迭代次数
- 设置合理的 max_iterations
- 添加早期终止条件
- 缓存中间结果
-
提升执行效率
- 并行执行独立任务
- 批量调用工具
- 使用流式输出
参考资料