Agent设计模式
概述
Agent设计模式是经过实践验证的、可复用的Agent架构方案。合理运用设计模式可以帮助我们构建更加健壮、高效、可维护的Agent系统。本章将介绍几种主流的Agent设计模式，包括ReAct、Plan-and-Execute、反思模式等。
ReAct模式
模式原理
ReAct（Reasoning + Acting）是一种将推理和行动交替进行的模式。Agent在每一步都先进行思考（Reasoning），然后采取行动（Acting），最后观察结果并继续下一轮循环。
工作流程
思考 → 行动 → 观察 → 思考 → 行动 → 观察 → ... → 最终答案实现代码
python
from typing import Dict, List, Any
from dataclasses import dataclass
from enum import Enum
class ActionType(Enum):
THINK = "think"
ACT = "act"
OBSERVE = "observe"
FINISH = "finish"
@dataclass
class Step:
action_type: ActionType
content: str
result: Any = None
class ReActAgent:
def __init__(self, llm, tools: Dict[str, Any], max_iterations: int = 10):
self.llm = llm
self.tools = tools
self.max_iterations = max_iterations
self.history: List[Step] = []
def run(self, query: str) -> str:
self.history = []
for iteration in range(self.max_iterations):
thought = self.think(query)
self.history.append(Step(ActionType.THINK, thought))
action, action_input = self.decide_action(thought)
if action == "finish":
answer = action_input
self.history.append(Step(ActionType.FINISH, answer))
return answer
observation = self.execute_action(action, action_input)
self.history.append(Step(ActionType.ACT, f"{action}({action_input})"))
self.history.append(Step(ActionType.OBSERVE, str(observation)))
return "达到最大迭代次数，未能完成任务"
def think(self, query: str) -> str:
context = self._build_context()
prompt = f"""
问题：{query}
历史步骤：
{context}
请思考下一步应该做什么。只输出思考内容，不要输出行动。
"""
return self.llm.generate(prompt).strip()
def decide_action(self, thought: str) -> tuple:
prompt = f"""
基于思考：{thought}
可用工具：{list(self.tools.keys())}
请决定下一步行动。格式：
- 如果需要使用工具：Action: tool_name[args]
- 如果可以给出答案：Action: finish[答案]
只输出一行。
"""
response = self.llm.generate(prompt).strip()
return self._parse_action(response)
def execute_action(self, action: str, action_input: Any) -> Any:
if action not in self.tools:
return f"错误：未知工具 {action}"
tool = self.tools[action]
try:
if isinstance(action_input, dict):
return tool(**action_input)
else:
return tool(action_input)
except Exception as e:
return f"执行错误：{str(e)}"
def _build_context(self) -> str:
if not self.history:
return "无"
lines = []
for step in self.history:
if step.action_type == ActionType.THINK:
lines.append(f"思考：{step.content}")
elif step.action_type == ActionType.ACT:
lines.append(f"行动：{step.content}")
elif step.action_type == ActionType.OBSERVE:
lines.append(f"观察：{step.content}")
return "\n".join(lines)
def _parse_action(self, response: str) -> tuple:
import re
match = re.match(r'(\w+)\[(.+)\]', response)
if match:
action = match.group(1)
args = match.group(2)
if action == "finish":
return action, args
try:
import json
args = json.loads(args)
except:
pass
return action, args
return "finish", response
agent = ReActAgent(
llm=llm,
tools={
"search": search_tool,
"calculate": calculate_tool
}
)
answer = agent.run("北京和上海的天气温差是多少？")适用场景
- 需要多步推理的任务
- 需要调用外部工具的任务
- 需要动态决策的任务
Plan-and-Execute模式
模式原理
Plan-and-Execute将任务分解为两个阶段：规划阶段（Plan）和执行阶段（Execute）。先制定完整的执行计划，然后按计划逐步执行。
实现代码
python
from typing import List, Dict
@dataclass
class PlanStep:
id: int
description: str
status: str = "pending"
result: Any = None
class PlanAndExecuteAgent:
def __init__(self, llm, tools: Dict[str, Any]):
self.llm = llm
self.tools = tools
self.plan: List[PlanStep] = []
def run(self, query: str) -> str:
self.plan = self.create_plan(query)
for step in self.plan:
result = self.execute_step(step)
step.result = result
step.status = "completed"
if not self.is_step_successful(result):
revised_plan = self.revise_plan(step, result)
self.plan = self.merge_plans(self.plan, revised_plan)
return self.synthesize_results()
def create_plan(self, query: str) -> List[PlanStep]:
prompt = f"""
为以下任务制定执行计划：
任务：{query}
可用工具：{list(self.tools.keys())}
请以JSON格式返回计划步骤列表，每个步骤包含：
- id: 步骤编号
- description: 步骤描述
"""
response = self.llm.generate(prompt)
plan_data = self._parse_plan(response)
return [PlanStep(**step) for step in plan_data]
def execute_step(self, step: PlanStep) -> Any:
prompt = f"""
执行步骤：{step.description}
可用工具：{list(self.tools.keys())}
请决定使用哪个工具及参数。格式：tool_name(args)
"""
response = self.llm.generate(prompt)
tool_name, args = self._parse_tool_call(response)
if tool_name in self.tools:
return self.tools[tool_name](**args)
return f"未知工具：{tool_name}"
def is_step_successful(self, result: Any) -> bool:
if isinstance(result, dict):
return result.get("success", False)
return result is not None
def revise_plan(self, failed_step: PlanStep, result: Any) -> List[PlanStep]:
prompt = f"""
步骤执行失败：
步骤：{failed_step.description}
结果：{result}
请制定新的计划来解决这个问题。
"""
response = self.llm.generate(prompt)
plan_data = self._parse_plan(response)
return [PlanStep(**step) for step in plan_data]
def merge_plans(self, original: List[PlanStep], revised: List[PlanStep]) -> List[PlanStep]:
failed_idx = next(
i for i, step in enumerate(original)
if step.id == failed_step.id
)
new_plan = original[:failed_idx]
for i, step in enumerate(revised):
step.id = failed_idx + i
new_plan.append(step)
for step in original[failed_idx + 1:]:
step.id = len(new_plan)
new_plan.append(step)
return new_plan
def synthesize_results(self) -> str:
results = [
f"步骤{step.id}：{step.description}\n结果：{step.result}"
for step in self.plan
if step.status == "completed"
]
prompt = f"""
基于以下执行结果，生成最终答案：
{chr(10).join(results)}
"""
return self.llm.generate(prompt)
def _parse_plan(self, response: str) -> List[Dict]:
import json
return json.loads(response)
def _parse_tool_call(self, response: str) -> tuple:
import re
match = re.match(r'(\w+)\((.+)\)', response)
if match:
return match.group(1), eval(f"{{{match.group(2)}}}")
return None, {}
agent = PlanAndExecuteAgent(
llm=llm,
tools={
"search": search_tool,
"analyze": analyze_tool,
"summarize": summarize_tool
}
)
result = agent.run("研究人工智能在医疗领域的最新应用")适用场景
- 任务步骤明确、可预测
- 需要全局优化的任务
- 执行时间较长的任务
反思模式
模式原理
反思模式让Agent在执行任务后进行自我评估和改进，通过多轮迭代不断优化结果。
实现代码
python
from typing import Tuple
class ReflectionAgent:
def __init__(self, llm, tools: Dict[str, Any], max_reflections: int = 3):
self.llm = llm
self.tools = tools
self.max_reflections = max_reflections
def run(self, query: str) -> str:
result = self.initial_execution(query)
for i in range(self.max_reflections):
critique = self.reflect(query, result)
if self.is_satisfactory(critique):
break
result = self.improve(query, result, critique)
return result
def initial_execution(self, query: str) -> str:
prompt = f"请完成以下任务：{query}"
return self.llm.generate(prompt)
def reflect(self, query: str, result: str) -> Dict:
prompt = f"""
任务：{query}
结果：{result}
请评估这个结果：
1. 是否完全回答了问题？
2. 是否有遗漏或错误？
3. 是否可以改进？
以JSON格式返回评估结果，包含：
- score: 评分（0-10）
- issues: 问题列表
- suggestions: 改进建议
"""
response = self.llm.generate(prompt)
return self._parse_critique(response)
def is_satisfactory(self, critique: Dict) -> bool:
return critique.get("score", 0) >= 8
def improve(self, query: str, result: str, critique: Dict) -> str:
prompt = f"""
任务：{query}
当前结果：{result}
评估反馈：
- 问题：{critique.get('issues', [])}
- 建议：{critique.get('suggestions', [])}
请根据反馈改进结果。
"""
return self.llm.generate(prompt)
def _parse_critique(self, response: str) -> Dict:
import json
try:
return json.loads(response)
except:
return {"score": 5, "issues": [], "suggestions": []}
class SelfConsistencyAgent:
def __init__(self, llm, num_samples: int = 5):
self.llm = llm
self.num_samples = num_samples
def run(self, query: str) -> str:
samples = []
for i in range(self.num_samples):
result = self.llm.generate(query, temperature=0.7)
samples.append(result)
return self.aggregate(samples)
def aggregate(self, samples: List[str]) -> str:
prompt = f"""
以下是同一问题的多个回答：
{chr(10).join([f'回答{i+1}: {s}' for i, s in enumerate(samples)])}
请综合这些回答，给出最准确、最完整的答案。
"""
return self.llm.generate(prompt)
agent = ReflectionAgent(
llm=llm,
tools=tools,
max_reflections=3
)
result = agent.run("解释量子计算的基本原理")适用场景
- 需要高质量输出的任务
- 复杂推理任务
- 创作类任务
工具使用模式
模式原理
工具使用模式专注于如何高效地选择和使用工具，包括工具选择、参数构造、结果处理等。
实现代码
python
class ToolUsePattern:
def __init__(self, llm, tools: Dict[str, Any]):
self.llm = llm
self.tools = tools
self.tool_history = []
def select_tool(self, task: str) -> str:
tool_descriptions = {
name: tool.__doc__ or "无描述"
for name, tool in self.tools.items()
}
prompt = f"""
任务：{task}
可用工具：
{tool_descriptions}
请选择最合适的工具。只输出工具名称。
"""
return self.llm.generate(prompt).strip()
def construct_params(self, tool_name: str, task: str) -> Dict:
tool = self.tools[tool_name]
prompt = f"""
任务：{task}
工具：{tool_name}
请构造工具调用参数。以JSON格式返回。
"""
response = self.llm.generate(prompt)
return self._parse_params(response)
def process_result(self, result: Any, task: str) -> str:
prompt = f"""
任务：{task}
工具返回结果：{result}
请处理结果并生成有用的输出。
"""
return self.llm.generate(prompt)
def run(self, task: str) -> str:
tool_name = self.select_tool(task)
params = self.construct_params(tool_name, task)
result = self.tools[tool_name](**params)
self.tool_history.append({
"tool": tool_name,
"params": params,
"result": result
})
return self.process_result(result, task)
def _parse_params(self, response: str) -> Dict:
import json
try:
return json.loads(response)
except:
return {}
class ChainOfTools:
def __init__(self, llm, tools: Dict[str, Any]):
self.llm = llm
self.tools = tools
def run(self, query: str) -> str:
tool_chain = self.plan_tool_chain(query)
result = None
for tool_call in tool_chain:
tool_name = tool_call["tool"]
params = tool_call.get("params", {})
if result and "previous_result" in params:
params["previous_result"] = result
result = self.tools[tool_name](**params)
return str(result)
def plan_tool_chain(self, query: str) -> List[Dict]:
prompt = f"""
任务：{query}
可用工具：{list(self.tools.keys())}
请规划工具调用链。以JSON格式返回列表，每项包含：
- tool: 工具名称
- params: 参数（可使用"previous_result"引用前一步结果）
"""
response = self.llm.generate(prompt)
return self._parse_tool_chain(response)
def _parse_tool_chain(self, response: str) -> List[Dict]:
import json
return json.loads(response)对话模式
模式原理
对话模式专注于多轮对话场景，管理对话状态、上下文和用户意图。
实现代码
python
from enum import Enum
from typing import Optional
class DialogState(Enum):
GREETING = "greeting"
COLLECTING_INFO = "collecting_info"
PROCESSING = "processing"
CONFIRMING = "confirming"
COMPLETED = "completed"
class DialogAgent:
def __init__(self, llm):
self.llm = llm
self.state = DialogState.GREETING
self.context = {}
self.history = []
def process(self, user_input: str) -> str:
self.history.append({"role": "user", "content": user_input})
intent = self.detect_intent(user_input)
response = self.generate_response(intent, user_input)
self.history.append({"role": "assistant", "content": response})
self.update_state(intent)
return response
def detect_intent(self, user_input: str) -> str:
prompt = f"""
用户输入：{user_input}
当前状态：{self.state.value}
已收集信息：{self.context}
请识别用户意图。可能的意图：
- provide_info: 提供信息
- ask_question: 提问
- confirm: 确认
- cancel: 取消
- other: 其他
只输出意图名称。
"""
return self.llm.generate(prompt).strip()
def generate_response(self, intent: str, user_input: str) -> str:
if self.state == DialogState.GREETING:
return self.handle_greeting(intent, user_input)
elif self.state == DialogState.COLLECTING_INFO:
return self.handle_collecting_info(intent, user_input)
elif self.state == DialogState.PROCESSING:
return self.handle_processing(intent, user_input)
elif self.state == DialogState.CONFIRMING:
return self.handle_confirming(intent, user_input)
else:
return "任务已完成，还有其他需要帮助的吗？"
def handle_greeting(self, intent: str, user_input: str) -> str:
self.state = DialogState.COLLECTING_INFO
return "您好！请问有什么可以帮您的？"
def handle_collecting_info(self, intent: str, user_input: str) -> str:
info = self.extract_info(user_input)
self.context.update(info)
missing = self.get_missing_info()
if missing:
return f"请提供{missing}信息。"
else:
self.state = DialogState.PROCESSING
return "信息已收集完毕，正在处理..."
def handle_processing(self, intent: str, user_input: str) -> str:
result = self.execute_task()
self.state = DialogState.CONFIRMING
return f"处理结果：{result}\n是否满意？"
def handle_confirming(self, intent: str, user_input: str) -> str:
if intent == "confirm":
self.state = DialogState.COMPLETED
return "好的，任务已完成！"
else:
self.state = DialogState.COLLECTING_INFO
return "请告诉我需要如何调整。"
def extract_info(self, user_input: str) -> Dict:
prompt = f"从以下文本中提取关键信息：{user_input}"
response = self.llm.generate(prompt)
return self._parse_info(response)
def get_missing_info(self) -> Optional[str]:
required_fields = ["name", "date", "type"]
for field in required_fields:
if field not in self.context:
return field
return None
def execute_task(self) -> str:
return f"基于{self.context}执行任务"
def update_state(self, intent: str):
if intent == "cancel":
self.state = DialogState.GREETING
self.context = {}
def _parse_info(self, response: str) -> Dict:
pass
agent = DialogAgent(llm)
response1 = agent.process("你好")
response2 = agent.process("我想预订明天下午的会议室")
response3 = agent.process("会议室名称是A101")错误处理模式
模式原理
错误处理模式关注如何优雅地处理Agent执行过程中的各种错误和异常情况。
实现代码
python
from typing import Callable, Any
from functools import wraps
class ErrorHandler:
def __init__(self, llm, max_retries: int = 3):
self.llm = llm
self.max_retries = max_retries
def with_retry(self, func: Callable) -> Callable:
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(self.max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
if attempt == self.max_retries - 1:
return self.handle_final_error(e, func.__name__)
recovery = self.attempt_recovery(e, func.__name__, attempt)
if recovery:
kwargs.update(recovery)
return None
return wrapper
def attempt_recovery(self, error: Exception, func_name: str, attempt: int) -> Dict:
prompt = f"""
函数 {func_name} 执行失败：
错误：{str(error)}
尝试次数：{attempt + 1}
请提供恢复策略或参数调整建议。以JSON格式返回。
"""
response = self.llm.generate(prompt)
return self._parse_recovery(response)
def handle_final_error(self, error: Exception, func_name: str) -> Any:
return {
"success": False,
"error": str(error),
"function": func_name,
"message": f"执行{func_name}失败，已达到最大重试次数"
}
def _parse_recovery(self, response: str) -> Dict:
import json
try:
return json.loads(response)
except:
return {}
class FallbackHandler:
def __init__(self, primary_agent, fallback_agents: List):
self.primary_agent = primary_agent
self.fallback_agents = fallback_agents
def run(self, task: Dict) -> Dict:
try:
result = self.primary_agent.process(task)
if self.is_valid_result(result):
return result
except Exception as e:
pass
for fallback in self.fallback_agents:
try:
result = fallback.process(task)
if self.is_valid_result(result):
return result
except Exception:
continue
return {
"success": False,
"error": "所有Agent都执行失败"
}
def is_valid_result(self, result: Any) -> bool:
if isinstance(result, dict):
return result.get("success", False)
return result is not None
class GracefulDegradation:
def __init__(self, llm):
self.llm = llm
self.degradation_levels = []
def add_level(self, level: int, handler: Callable):
self.degradation_levels.append((level, handler))
self.degradation_levels.sort(key=lambda x: x[0])
def execute(self, task: Dict, current_level: int = 0) -> Any:
for level, handler in self.degradation_levels:
if level >= current_level:
try:
result = handler(task)
if self.is_acceptable(result):
return result
except Exception:
continue
return self.default_handler(task)
def is_acceptable(self, result: Any) -> bool:
pass
def default_handler(self, task: Dict) -> Any:
return {"success": False, "message": "服务暂时不可用"}
error_handler = ErrorHandler(llm)
@error_handler.with_retry
def risky_operation(param: str):
pass
result = risky_operation(param="test")组合模式示例
python
class AdvancedAgent:
def __init__(self, llm, tools: Dict[str, Any]):
self.llm = llm
self.tools = tools
self.react_engine = ReActAgent(llm, tools)
self.reflector = ReflectionAgent(llm, tools)
self.error_handler = ErrorHandler(llm)
self.fallback = FallbackHandler(
self.react_engine,
[PlanAndExecuteAgent(llm, tools)]
)
def run(self, query: str) -> str:
@self.error_handler.with_retry
def execute():
initial_result = self.react_engine.run(query)
reflected_result = self.reflector.run(
f"改进以下回答：{initial_result}\n原问题：{query}"
)
return reflected_result
try:
return execute()
except Exception:
return self.fallback.run({"query": query})
def run_with_dialog(self, query: str) -> str:
dialog_agent = DialogAgent(self.llm)
response = dialog_agent.process(query)
if dialog_agent.state == DialogState.PROCESSING:
result = self.run(dialog_agent.context.get("task", query))
dialog_agent.context["result"] = result
return response
agent = AdvancedAgent(llm, tools)
result = agent.run("分析人工智能对就业市场的影响")小结
Agent设计模式是构建高质量Agent系统的重要工具：
- ReAct模式 - 适合需要动态推理和工具调用的任务
- Plan-and-Execute模式 - 适合步骤明确、可预测的任务
- 反思模式 - 适合需要高质量输出的任务
- 工具使用模式 - 适合依赖外部工具的任务
- 对话模式 - 适合多轮交互场景
- 错误处理模式 - 提高系统健壮性
在实际应用中，应该根据具体任务特点选择合适的模式，或组合多种模式以发挥各自优势。