自主AI Agent开发完全教程（记忆/规划/工具使用）
一、概述
人工智能正在从"被动响应"走向"主动行动"。传统的AI模型接收输入、产生输出，是一个无状态的函数映射。而AI Agent（智能体） 则是一种能够自主感知环境、制定计划、使用工具、执行行动并从反馈中学习的智能系统。
2024年以来，AI Agent已成为人工智能领域最活跃的研究和工程方向之一。从自动化编程助手到自主研究系统，从智能客服到数据分析Agent，Agent技术正在重塑人机交互的方式。
本教程将系统讲解AI Agent的核心架构、关键技术与工程实践，帮助开发者从零构建具有记忆、规划和工具使用能力的自主Agent系统。
目标读者：有Python基础和LLM使用经验的开发者。
你将学到：
- AI Agent的核心架构与设计模式
- 短期记忆、长期记忆和工作记忆的实现
- 多种规划策略（ReAct、Plan-and-Execute、Reflexion）
- Function Calling与工具集成
- 环境感知与状态管理
- 错误恢复与自纠正机制
- Agent评估方法
- 主流Agent框架的对比与使用
- Agent安全与人类监督机制
- 从零构建一个自主研究Agent
二、AI Agent概述与架构
2.1 什么是AI Agent
AI Agent是一个具有以下核心能力的系统：
- 感知（Perception）：从环境获取信息（用户输入、API返回、传感器数据等）
- 推理（Reasoning）：基于感知信息和内部知识进行分析和决策
- 行动（Action）：执行具体操作（调用工具、生成内容、发送请求等）
- 记忆（Memory）：保持上下文、积累经验、学习改进
用一个简单的公式表示：
Agent = LLM + 记忆 + 规划 + 工具使用 + 行动循环
2.2 Agent核心架构
一个典型的Agent系统架构包含以下组件：
┌─────────────────────────────────────────────┐
│ 用户输入 │
└─────────────────┬───────────────────────────┘
│
┌─────────────────▼───────────────────────────┐
│ 感知层 (Perception) │
│ 解析输入、提取意图、环境状态获取 │
└─────────────────┬───────────────────────────┘
│
┌─────────────────▼───────────────────────────┐
│ 推理层 (Reasoning) │
│ 基于LLM的思考、规划、决策 │
│ ┌─────────┐ ┌──────────┐ ┌──────────┐ │
│ │ 短期记忆 │ │ 长期记忆 │ │ 工作记忆 │ │
│ └─────────┘ └──────────┘ └──────────┘ │
└─────────────────┬───────────────────────────┘
│
┌─────────────────▼───────────────────────────┐
│ 行动层 (Action) │
│ 工具调用、API请求、代码执行 │
└─────────────────┬───────────────────────────┘
│
┌─────────────────▼───────────────────────────┐
│ 反馈层 (Feedback) │
│ 结果评估、错误处理、经验积累 │
└─────────────────────────────────────────────┘
2.3 Agent循环（Agent Loop）
Agent的核心运行模式是一个感知-思考-行动循环：
class SimpleAgent:
"""最简单的Agent循环实现"""
def __init__(self, llm, tools, memory):
self.llm = llm
self.tools = tools
self.memory = memory
def run(self, task: str, max_iterations: int = 10) -> str:
"""执行任务的主循环"""
self.memory.add("user", task)
for i in range(max_iterations):
# 感知：获取当前上下文
context = self.memory.get_context()
# 思考：LLM决定下一步行动
thought = self.think(context)
# 检查是否完成
if thought.get("type") == "final_answer":
answer = thought["content"]
self.memory.add("assistant", answer)
return answer
# 行动：执行工具调用
if thought.get("type") == "action":
result = self.execute_action(thought["action"], thought["input"])
self.memory.add("system", f"工具 {thought['action']} 返回: {result}")
return "达到最大迭代次数，任务未能完成。"
def think(self, context: str) -> dict:
"""LLM推理"""
prompt = f"""基于当前上下文，决定下一步行动。
上下文：
{context}
你可以：
1. 调用工具: {{"type": "action", "action": "工具名", "input": "参数"}}
2. 给出最终答案: {{"type": "final_answer", "content": "答案内容"}}
请以JSON格式回复："""
response = self.llm.generate(prompt)
return parse_json(response)
def execute_action(self, action: str, input_data: str) -> str:
"""执行工具调用"""
if action in self.tools:
return self.tools[action].execute(input_data)
return f"错误：未知工具 {action}"
三、Agent记忆系统
3.1 记忆的三种类型
人类记忆系统为Agent设计提供了重要启发：
短期记忆（Short-term Memory）：当前对话的上下文，通常通过LLM的上下文窗口实现。容量有限，会话结束即消失。
长期记忆（Long-term Memory）：持久化存储的知识和经验，通常使用向量数据库。跨会话保留，支持检索。
工作记忆（Working Memory）：当前任务的中间状态和临时信息，类似人类的"心智工作台"。
3.2 短期记忆实现
from typing import List, Dict
from collections import deque
class ShortTermMemory:
"""短期记忆：基于滑动窗口的对话历史"""
def __init__(self, max_tokens: int = 4000):
self.messages: List[Dict] = []
self.max_tokens = max_tokens
def add(self, role: str, content: str):
self.messages.append({"role": role, "content": content})
self._trim()
def get_context(self) -> str:
"""获取当前上下文"""
return "\n".join(
f"[{m['role']}]: {m['content']}" for m in self.messages
)
def get_messages(self) -> List[Dict]:
return self.messages.copy()
def _trim(self):
"""保持在token限制内"""
total = sum(len(m["content"]) for m in self.messages)
while total > self.max_tokens and len(self.messages) > 1:
removed = self.messages.pop(0)
total -= len(removed["content"])
def clear(self):
self.messages.clear()
3.3 长期记忆实现
import json
import numpy as np
from datetime import datetime
from typing import List, Dict, Optional
class LongTermMemory:
"""长期记忆：基于向量数据库的持久化记忆"""
def __init__(self, embedding_model, vector_store):
self.embedding_model = embedding_model
self.vector_store = vector_store
def store(self, content: str, memory_type: str = "general",
metadata: Dict = None):
"""存储一条记忆"""
embedding = self.embedding_model.encode(content)
memory_entry = {
"id": f"mem_{datetime.now().timestamp()}",
"content": content,
"embedding": embedding,
"type": memory_type,
"timestamp": datetime.now().isoformat(),
"metadata": metadata or {}
}
self.vector_store.add(memory_entry)
def recall(self, query: str, top_k: int = 5,
memory_type: Optional[str] = None) -> List[Dict]:
"""检索相关记忆"""
query_embedding = self.embedding_model.encode(query)
filters = {}
if memory_type:
filters["type"] = memory_type
results = self.vector_store.search(
query_embedding, top_k=top_k, filters=filters
)
return results
def forget(self, memory_id: str):
"""删除特定记忆"""
self.vector_store.delete(memory_id)
def consolidate(self):
"""记忆整合：合并相似记忆，去除冗余"""
all_memories = self.vector_store.get_all()
# 聚类相似记忆
clusters = self._cluster_memories(all_memories)
# 对每个聚类进行摘要
for cluster in clusters:
if len(cluster) > 1:
summary = self._summarize_cluster(cluster)
# 删除原始记忆，存储摘要
for mem in cluster:
self.vector_store.delete(mem["id"])
self.store(summary, memory_type="consolidated")
3.4 工作记忆实现
class WorkingMemory:
"""工作记忆：当前任务的临时状态"""
def __init__(self):
self.scratchpad: Dict = {} # 临时变量
self.plan: List[str] = [] # 当前计划
self.step_results: Dict = {} # 每步结果
self.constraints: List[str] = [] # 约束条件
self.goals: List[str] = [] # 目标
def set_goal(self, goal: str):
self.goals.append(goal)
def set_plan(self, steps: List[str]):
self.plan = steps
self.step_results = {i: None for i in range(len(steps))}
def update_step(self, step_index: int, result: str):
self.step_results[step_index] = result
def get_progress(self) -> Dict:
completed = sum(1 for v in self.step_results.values() if v is not None)
return {
"total_steps": len(self.plan),
"completed": completed,
"current_step": self._get_current_step(),
"progress_pct": completed / len(self.plan) * 100 if self.plan else 0
}
def _get_current_step(self) -> Optional[str]:
for i, result in self.step_results.items():
if result is None:
return self.plan[i]
return None
def set_scratchpad(self, key: str, value):
self.scratchpad[key] = value
def get_scratchpad(self, key: str, default=None):
return self.scratchpad.get(key, default)
def clear(self):
self.scratchpad.clear()
self.plan.clear()
self.step_results.clear()
self.constraints.clear()
self.goals.clear()
3.5 统一记忆管理
class AgentMemory:
"""统一的Agent记忆管理"""
def __init__(self, embedding_model, vector_store, max_context_tokens=4000):
self.short_term = ShortTermMemory(max_tokens=max_context_tokens)
self.long_term = LongTermMemory(embedding_model, vector_store)
self.working = WorkingMemory()
def add_conversation(self, role: str, content: str):
"""添加对话到短期记忆"""
self.short_term.add(role, content)
def remember(self, content: str, memory_type: str = "general"):
"""存储到长期记忆"""
self.long_term.store(content, memory_type)
def recall(self, query: str, top_k: int = 5) -> List[Dict]:
"""从长期记忆中检索"""
return self.long_term.recall(query, top_k)
def get_context(self, include_long_term: bool = True,
query: str = None) -> str:
"""获取完整的记忆上下文"""
context_parts = []
# 短期记忆
context_parts.append("=== 当前对话 ===")
context_parts.append(self.short_term.get_context())
# 相关的长期记忆
if include_long_term and query:
memories = self.recall(query, top_k=3)
if memories:
context_parts.append("\n=== 相关记忆 ===")
for mem in memories:
context_parts.append(f"- {mem['content']}")
# 工作记忆状态
if self.working.plan:
progress = self.working.get_progress()
context_parts.append(f"\n=== 任务进度 ===")
context_parts.append(f"当前步骤: {progress['current_step']}")
context_parts.append(f"完成度: {progress['progress_pct']:.0f}%")
return "\n".join(context_parts)
四、规划能力
4.1 ReAct（Reasoning + Acting）
ReAct是最经典的Agent规划模式，将推理和行动交替进行：
思考(Thought) → 行动(Action) → 观察(Observation) → 思考 → ...
class ReActAgent:
"""基于ReAct模式的Agent"""
def __init__(self, llm, tools, memory):
self.llm = llm
self.tools = tools
self.memory = memory
def run(self, task: str, max_steps: int = 10) -> str:
self.memory.add_conversation("user", task)
for step in range(max_steps):
# 构建提示
prompt = self._build_react_prompt(task)
# 获取LLM响应
response = self.llm.generate(prompt)
# 解析响应
parsed = self._parse_react_response(response)
# 记录思考过程
self.memory.add_conversation("assistant", response)
# 如果是最终答案
if parsed["type"] == "finish":
return parsed["answer"]
# 执行行动
if parsed["type"] == "action":
observation = self._execute_tool(
parsed["tool"],
parsed["input"]
)
self.memory.add_conversation("system",
f"Observation: {observation}")
return "达到最大步骤数限制"
def _build_react_prompt(self, task: str) -> str:
tools_desc = "\n".join(
f"- {name}: {tool.description}"
for name, tool in self.tools.items()
)
return f"""你是一个智能助手。请用以下格式回答问题：
Question: 用户的问题
Thought: 分析当前情况，决定下一步
Action: 工具名称
Action Input: 工具输入
Observation: 工具返回结果
... (可以重复多次)
Thought: 我现在知道答案了
Final Answer: 最终答案
可用工具：
{tools_desc}
对话历史：
{self.memory.short_term.get_context()}
请继续："""
def _parse_react_response(self, response: str) -> dict:
"""解析ReAct格式的响应"""
lines = response.strip().split("\n")
for line in lines:
if line.startswith("Final Answer:"):
return {
"type": "finish",
"answer": line[len("Final Answer:"):].strip()
}
if line.startswith("Action:"):
tool_name = line[len("Action:"):].strip()
# 查找对应的Action Input
for sub_line in lines:
if sub_line.startswith("Action Input:"):
tool_input = sub_line[len("Action Input:"):].strip()
return {
"type": "action",
"tool": tool_name,
"input": tool_input
}
return {"type": "thought", "content": response}
def _execute_tool(self, tool_name: str, tool_input: str) -> str:
if tool_name in self.tools:
try:
return self.tools[tool_name].execute(tool_input)
except Exception as e:
return f"工具执行错误: {str(e)}"
return f"未知工具: {tool_name}"
4.2 Plan-and-Execute
Plan-and-Execute模式将规划和执行分离：先制定完整计划，然后逐步执行。
class PlanAndExecuteAgent:
"""Plan-and-Execute模式Agent"""
def __init__(self, llm, tools, memory):
self.llm = llm
self.tools = tools
self.memory = memory
self.planner = Planner(llm)
self.executor = Executor(llm, tools)
def run(self, task: str) -> str:
# 阶段1: 制定计划
plan = self.planner.create_plan(task)
self.memory.working.set_plan(plan)
print(f"制定计划: {plan}")
# 阶段2: 逐步执行
for i, step in enumerate(plan):
print(f"\n执行步骤 {i+1}: {step}")
# 获取之前步骤的结果
context = self._get_execution_context()
# 执行当前步骤
result = self.executor.execute_step(step, context)
# 记录结果
self.memory.working.update_step(i, result)
print(f"结果: {result}")
# 检查是否需要重新规划
if self._should_replan(task, i):
remaining = plan[i+1:]
new_plan = self.planner.replan(task, remaining,
self.memory.working.step_results)
plan = plan[:i+1] + new_plan
self.memory.working.set_plan(plan)
print(f"重新规划: {new_plan}")
# 阶段3: 综合结果
final_answer = self._synthesize_results(task)
return final_answer
def _get_execution_context(self) -> str:
"""获取执行上下文"""
context_parts = []
for i, result in self.memory.working.step_results.items():
if result is not None:
context_parts.append(
f"步骤{i+1} ({self.memory.working.plan[i]}): {result}"
)
return "\n".join(context_parts)
def _should_replan(self, task: str, current_step: int) -> bool:
"""判断是否需要重新规划"""
# 简单策略：如果当前步骤失败，需要重新规划
current_result = self.memory.working.step_results[current_step]
if current_result and "错误" in current_result:
return True
return False
def _synthesize_results(self, task: str) -> str:
"""综合所有步骤结果生成最终答案"""
context = self._get_execution_context()
prompt = f"""基于以下执行结果，回答原始问题。
原始问题: {task}
执行结果:
{context}
请给出完整的最终答案："""
return self.llm.generate(prompt)
class Planner:
"""计划生成器"""
def __init__(self, llm):
self.llm = llm
def create_plan(self, task: str) -> List[str]:
prompt = f"""请为以下任务制定一个详细的执行计划，将任务分解为具体的步骤。
任务: {task}
要求：
1. 每个步骤应该是具体、可执行的
2. 步骤之间有清晰的逻辑顺序
3. 步骤数量控制在3-7个
请以如下格式输出：
步骤1: xxx
步骤2: xxx
..."""
response = self.llm.generate(prompt)
return self._parse_plan(response)
def replan(self, task: str, remaining_steps: List[str],
completed_results: Dict) -> List[str]:
prompt = f"""原始任务: {task}
已完成的结果:
{json.dumps(completed_results, ensure_ascii=False, indent=2)}
剩余计划: {remaining_steps}
请根据已完成的结果，重新制定剩余步骤的计划。"""
response = self.llm.generate(prompt)
return self._parse_plan(response)
def _parse_plan(self, response: str) -> List[str]:
steps = []
for line in response.strip().split("\n"):
line = line.strip()
if line.startswith("步骤"):
# 提取步骤内容
parts = line.split(":", 1)
if len(parts) > 1:
steps.append(parts[1].strip())
return steps
class Executor:
"""步骤执行器"""
def __init__(self, llm, tools):
self.llm = llm
self.tools = tools
def execute_step(self, step: str, context: str) -> str:
prompt = f"""请执行以下步骤：
步骤: {step}
之前的执行结果:
{context}
你可以使用以下工具:
{self._format_tools()}
如果需要使用工具，请以如下格式输出：
Action: 工具名称
Action Input: 工具输入
如果不需要工具，请直接输出结果。
输出："""
response = self.llm.generate(prompt)
# 检查是否需要调用工具
if "Action:" in response:
return self._handle_tool_call(response)
return response
def _format_tools(self) -> str:
return "\n".join(
f"- {name}: {tool.description}"
for name, tool in self.tools.items()
)
def _handle_tool_call(self, response: str) -> str:
"""处理工具调用"""
lines = response.split("\n")
tool_name = None
tool_input = None
for line in lines:
if line.startswith("Action:"):
tool_name = line[len("Action:"):].strip()
elif line.startswith("Action Input:"):
tool_input = line[len("Action Input:"):].strip()
if tool_name and tool_name in self.tools:
return self.tools[tool_name].execute(tool_input or "")
return f"工具调用失败: {tool_name}"
4.3 Reflexion（反思式规划）
Reflexion模式让Agent能够从失败中学习，通过反思改进策略：
class ReflexionAgent:
"""基于Reflexion的自我反思Agent"""
def __init__(self, llm, tools, memory, max_retries=3):
self.llm = llm
self.tools = tools
self.memory = memory
self.max_retries = max_retries
self.reflections: List[str] = []
def run(self, task: str) -> str:
for attempt in range(self.max_retries):
# 尝试执行任务
result, success = self._attempt(task)
if success:
return result
# 反思失败原因
reflection = self._reflect(task, result)
self.reflections.append(reflection)
self.memory.remember(reflection, memory_type="reflection")
print(f"尝试 {attempt + 1} 失败，反思: {reflection}")
return f"经过 {self.max_retries} 次尝试后未能完成任务"
def _attempt(self, task: str) -> tuple:
"""尝试执行任务"""
# 构建包含反思的提示
reflections_text = ""
if self.reflections:
reflections_text = "\n之前的反思（请避免重复这些错误）:\n"
for i, r in enumerate(self.reflections, 1):
reflections_text += f"{i}. {r}\n"
prompt = f"""任务: {task}
{reflections_text}
请完成任务。如果遇到问题，请明确说明。"""
# 使用ReAct执行
agent = ReActAgent(self.llm, self.tools, ShortTermMemory())
result = agent.run(prompt)
# 判断是否成功
success = self._evaluate_success(task, result)
return result, success
def _reflect(self, task: str, failed_result: str) -> str:
"""反思失败原因"""
prompt = f"""你尝试完成以下任务但失败了。
任务: {task}
你的回答: {failed_result}
请反思：
1. 为什么这个回答不正确或不完整？
2. 你犯了什么错误？
3. 下次应该如何改进？
反思："""
return self.llm.generate(prompt)
def _evaluate_success(self, task: str, result: str) -> bool:
"""评估任务是否成功完成"""
prompt = f"""请评估以下回答是否正确完成了任务。
任务: {task}
回答: {result}
回答是否正确且完整？请只回答 "是" 或 "否"。"""
evaluation = self.llm.generate(prompt).strip()
return "是" in evaluation
五、工具使用（Function Calling / Tool Use）
5.1 工具定义与注册
from typing import Callable, Dict, Any
from dataclasses import dataclass
import inspect
@dataclass
class Tool:
"""工具定义"""
name: str
description: str
function: Callable
parameters: Dict[str, Any]
def execute(self, **kwargs) -> str:
try:
result = self.function(**kwargs)
return str(result)
except Exception as e:
return f"执行错误: {str(e)}"
class ToolRegistry:
"""工具注册中心"""
def __init__(self):
self.tools: Dict[str, Tool] = {}
def register(self, name: str, description: str,
parameters: Dict = None):
"""装饰器方式注册工具"""
def decorator(func: Callable):
self.tools[name] = Tool(
name=name,
description=description,
function=func,
parameters=parameters or {}
)
return func
return decorator
def get_tool(self, name: str) -> Tool:
return self.tools.get(name)
def list_tools(self) -> str:
"""列出所有工具"""
descriptions = []
for name, tool in self.tools.items():
params = ", ".join(tool.parameters.keys())
descriptions.append(f"- {name}({params}): {tool.description}")
return "\n".join(descriptions)
def call(self, name: str, **kwargs) -> str:
"""调用工具"""
tool = self.tools.get(name)
if not tool:
return f"未知工具: {name}"
return tool.execute(**kwargs)
# 使用示例
registry = ToolRegistry()
@registry.register(
name="search_web",
description="搜索互联网获取信息",
parameters={"query": {"type": "string", "description": "搜索关键词"}}
)
def search_web(query: str) -> str:
# 实际实现会调用搜索API
return f"搜索 '{query}' 的结果: ..."
@registry.register(
name="calculate",
description="执行数学计算",
parameters={"expression": {"type": "string", "description": "数学表达式"}}
)
def calculate(expression: str) -> str:
try:
result = eval(expression)
return str(result)
except Exception as e:
return f"计算错误: {str(e)}"
@registry.register(
name="read_file",
description="读取文件内容",
parameters={"file_path": {"type": "string", "description": "文件路径"}}
)
def read_file(file_path: str) -> str:
try:
with open(file_path, 'r', encoding='utf-8') as f:
return f.read()
except Exception as e:
return f"读取错误: {str(e)}"
5.2 Function Calling实现
class FunctionCallingAgent:
"""基于Function Calling的Agent"""
def __init__(self, llm_client, tool_registry: ToolRegistry):
self.client = llm_client
self.tools = tool_registry
def run(self, task: str, max_iterations: int = 5) -> str:
messages = [{"role": "user", "content": task}]
for _ in range(max_iterations):
# 调用LLM（带工具定义）
response = self.client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
tools=self._get_tool_definitions(),
tool_choice="auto"
)
message = response.choices[0].message
messages.append(message)
# 如果没有工具调用，返回结果
if not message.tool_calls:
return message.content
# 处理工具调用
for tool_call in message.tool_calls:
function_name = tool_call.function.name
function_args = json.loads(tool_call.function.arguments)
# 执行工具
result = self.tools.call(function_name, **function_args)
# 添加工具结果到消息
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result
})
return "达到最大迭代次数"
def _get_tool_definitions(self) -> list:
"""转换为OpenAI工具格式"""
tools = []
for name, tool in self.tools.tools.items():
tools.append({
"type": "function",
"function": {
"name": name,
"description": tool.description,
"parameters": {
"type": "object",
"properties": {
param_name: {
"type": param_info.get("type", "string"),
"description": param_info.get("description", "")
}
for param_name, param_info in tool.parameters.items()
},
"required": list(tool.parameters.keys())
}
}
})
return tools
5.3 工具使用最佳实践
工具描述要清晰：LLM根据描述决定何时使用工具，描述不清会导致误用。
参数验证：在工具内部验证输入参数，防止错误输入。
错误处理：工具应该返回友好的错误信息，而不是抛出异常。
超时控制：对外部API调用设置超时。
import functools
import signal
def timeout_handler(signum, frame):
raise TimeoutError("工具执行超时")
def with_timeout(seconds=30):
"""工具超时装饰器"""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
old_handler = signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(seconds)
try:
result = func(*args, **kwargs)
finally:
signal.alarm(0)
signal.signal(signal.SIGALRM, old_handler)
return result
return wrapper
return decorator
@registry.register(
name="fetch_url",
description="获取网页内容",
parameters={"url": {"type": "string", "description": "网页URL"}}
)
@with_timeout(seconds=10)
def fetch_url(url: str) -> str:
import urllib.request
response = urllib.request.urlopen(url)
return response.read().decode('utf-8')[:5000] # 限制返回长度
六、环境感知与状态管理
6.1 环境接口设计
from abc import ABC, abstractmethod
from typing import Any, Dict
class Environment(ABC):
"""Agent环境接口"""
@abstractmethod
def get_state(self) -> Dict[str, Any]:
"""获取当前环境状态"""
pass
@abstractmethod
def execute_action(self, action: str, params: Dict) -> Any:
"""在环境中执行动作"""
pass
@abstractmethod
def observe(self) -> str:
"""观察环境变化"""
pass
class WebEnvironment(Environment):
"""Web环境实现"""
def __init__(self):
self.current_url = None
self.page_content = None
self.history = []
def get_state(self) -> Dict[str, Any]:
return {
"current_url": self.current_url,
"has_content": self.page_content is not None,
"history_length": len(self.history)
}
def execute_action(self, action: str, params: Dict) -> Any:
if action == "navigate":
self.current_url = params["url"]
# 实际会获取网页内容
self.page_content = f"页面内容: {params['url']}"
self.history.append(params["url"])
return f"已导航到 {params['url']}"
elif action == "extract":
# 提取页面信息
return f"从 {self.current_url} 提取的信息..."
return f"未知动作: {action}"
def observe(self) -> str:
state = self.get_state()
return f"当前URL: {state['current_url']}, 浏览历史: {state['history_length']}页"
class FileSystemEnvironment(Environment):
"""文件系统环境"""
def __init__(self, base_dir: str = "."):
self.base_dir = base_dir
self.current_dir = base_dir
def get_state(self) -> Dict[str, Any]:
import os
files = os.listdir(self.current_dir)
return {
"current_dir": self.current_dir,
"files": files[:20], # 限制返回数量
"file_count": len(files)
}
def execute_action(self, action: str, params: Dict) -> Any:
import os
if action == "read_file":
path = os.path.join(self.current_dir, params["path"])
with open(path, 'r', encoding='utf-8') as f:
return f.read()
elif action == "list_dir":
path = os.path.join(self.current_dir, params.get("path", "."))
return os.listdir(path)
elif action == "change_dir":
self.current_dir = os.path.join(self.current_dir, params["path"])
return f"当前目录: {self.current_dir}"
return f"未知动作: {action}"
def observe(self) -> str:
state = self.get_state()
return f"目录: {state['current_dir']}, 文件数: {state['file_count']}"
6.2 状态机管理
class AgentStateMachine:
"""Agent状态机"""
STATES = ["idle", "planning", "executing", "waiting", "error", "completed"]
def __init__(self):
self.current_state = "idle"
self.state_data = {}
self.transitions = {
"idle": ["planning"],
"planning": ["executing", "idle"],
"executing": ["waiting", "completed", "error"],
"waiting": ["executing", "error"],
"error": ["planning", "idle"],
"completed": ["idle"]
}
def transition(self, new_state: str, data: Dict = None) -> bool:
"""状态转换"""
if new_state in self.transitions.get(self.current_state, []):
self.current_state = new_state
if data:
self.state_data.update(data)
return True
return False
def get_state(self) -> Dict:
return {
"state": self.current_state,
"data": self.state_data
}
七、错误恢复与自纠正
7.1 错误处理策略
class ErrorRecoveryStrategy:
"""错误恢复策略"""
def __init__(self, llm):
self.llm = llm
self.error_history = []
def handle_error(self, error: Exception, context: Dict) -> str:
"""处理错误并返回恢复策略"""
error_info = {
"type": type(error).__name__,
"message": str(error),
"context": context
}
self.error_history.append(error_info)
# 分析错误类型
if isinstance(error, ToolExecutionError):
return self._handle_tool_error(error, context)
elif isinstance(error, PlanningError):
return self._handle_planning_error(error, context)
else:
return self._handle_general_error(error, context)
def _handle_tool_error(self, error, context) -> str:
"""处理工具执行错误"""
prompt = f"""工具执行失败：
工具: {context.get('tool_name')}
输入: {context.get('tool_input')}
错误: {str(error)}
请分析原因并建议恢复策略：
1. 是否应该重试？
2. 是否应该使用不同的工具？
3. 是否应该修改输入参数？
建议："""
return self.llm.generate(prompt)
def _handle_planning_error(self, error, context) -> str:
"""处理规划错误"""
return "replan"
def _handle_general_error(self, error, context) -> str:
"""处理一般错误"""
prompt = f"""遇到错误: {str(error)}
当前状态: {context}
请建议恢复策略。"""
return self.llm.generate(prompt)
class ResilientAgent:
"""具有错误恢复能力的Agent"""
def __init__(self, llm, tools, memory):
self.llm = llm
self.tools = tools
self.memory = memory
self.error_handler = ErrorRecoveryStrategy(llm)
self.max_retries = 3
def run(self, task: str) -> str:
retry_count = 0
while retry_count < self.max_retries:
try:
result = self._execute(task)
return result
except Exception as e:
retry_count += 1
print(f"错误 (尝试 {retry_count}): {str(e)}")
# 获取恢复策略
context = {
"task": task,
"attempt": retry_count,
"memory": self.memory.get_context()
}
recovery = self.error_handler.handle_error(e, context)
print(f"恢复策略: {recovery}")
# 根据恢复策略调整
if "重试" in recovery:
continue
elif "修改" in recovery:
task = self._modify_task(task, recovery)
else:
break
return f"任务执行失败，已尝试 {retry_count} 次"
def _execute(self, task: str) -> str:
"""执行任务"""
agent = ReActAgent(self.llm, self.tools, self.memory.short_term)
return agent.run(task)
def _modify_task(self, task: str, recovery: str) -> str:
"""根据恢复策略修改任务"""
prompt = f"""原始任务: {task}
恢复建议: {recovery}
请根据建议，重新表述任务："""
return self.llm.generate(prompt)
7.2 自我验证
class SelfVerifyingAgent:
"""具有自我验证能力的Agent"""
def __init__(self, llm, tools):
self.llm = llm
self.tools = tools
def run(self, task: str) -> str:
# 执行任务
result = self._execute(task)
# 自我验证
is_valid, feedback = self._verify(task, result)
if is_valid:
return result
# 如果验证失败，尝试修正
corrected = self._correct(task, result, feedback)
return corrected
def _execute(self, task: str) -> str:
agent = ReActAgent(self.llm, self.tools, ShortTermMemory())
return agent.run(task)
def _verify(self, task: str, result: str) -> tuple:
"""验证结果"""
prompt = f"""请验证以下回答是否正确完成了任务。
任务: {task}
回答: {result}
检查清单：
1. 回答是否完整？
2. 信息是否准确？
3. 是否回答了问题的所有部分？
如果验证通过，回答 "VALID"。
如果验证失败，说明具体问题。
验证结果："""
verification = self.llm.generate(prompt)
if "VALID" in verification:
return True, ""
return False, verification
def _correct(self, task: str, result: str, feedback: str) -> str:
"""修正结果"""
prompt = f"""任务: {task}
之前的回答: {result}
验证反馈: {feedback}
请根据反馈修正回答："""
return self.llm.generate(prompt)
八、Agent评估与基准测试
8.1 评估维度
Agent的评估需要考虑多个维度：
任务完成率：Agent能否成功完成指定任务。
效率：完成任务所需的步骤数、时间、API调用次数。
准确性：结果的正确性和精确度。
鲁棒性：面对错误和异常情况的处理能力。
安全性：是否遵守安全约束。
8.2 评估框架
from dataclasses import dataclass
from typing import List, Callable
import time
import json
@dataclass
class TestCase:
"""测试用例"""
task: str
expected_output: str = None
evaluation_fn: Callable = None
max_steps: int = 10
timeout: int = 60
@dataclass
class EvalResult:
"""评估结果"""
test_case: str
success: bool
output: str
steps_taken: int
time_taken: float
error: str = None
class AgentEvaluator:
"""Agent评估器"""
def __init__(self, agent):
self.agent = agent
self.results: List[EvalResult] = []
def evaluate(self, test_cases: List[TestCase]) -> Dict:
"""运行评估"""
for test_case in test_cases:
result = self._run_test(test_case)
self.results.append(result)
return self._compute_metrics()
def _run_test(self, test_case: TestCase) -> EvalResult:
"""运行单个测试"""
start_time = time.time()
try:
output = self.agent.run(test_case.task)
time_taken = time.time() - start_time
# 评估成功性
success = True
if test_case.evaluation_fn:
success = test_case.evaluation_fn(output)
elif test_case.expected_output:
success = self._check_output(output, test_case.expected_output)
return EvalResult(
test_case=test_case.task,
success=success,
output=output,
steps_taken=getattr(self.agent, 'step_count', 0),
time_taken=time_taken
)
except Exception as e:
return EvalResult(
test_case=test_case.task,
success=False,
output="",
steps_taken=0,
time_taken=time.time() - start_time,
error=str(e)
)
def _check_output(self, output: str, expected: str) -> bool:
"""简单输出匹配"""
return expected.lower() in output.lower()
def _compute_metrics(self) -> Dict:
"""计算评估指标"""
total = len(self.results)
successes = sum(1 for r in self.results if r.success)
return {
"total_tests": total,
"successes": successes,
"success_rate": successes / total if total > 0 else 0,
"avg_time": sum(r.time_taken for r in self.results) / total,
"avg_steps": sum(r.steps_taken for r in self.results) / total,
"errors": [r for r in self.results if r.error]
}
def generate_report(self) -> str:
"""生成评估报告"""
metrics = self._compute_metrics()
report = f"""Agent评估报告
================
总测试数: {metrics['total_tests']}
成功数: {metrics['successes']}
成功率: {metrics['success_rate']:.2%}
平均耗时: {metrics['avg_time']:.2f}秒
平均步骤数: {metrics['avg_steps']:.1f}
详细结果:
"""
for result in self.results:
status = "✓" if result.success else "✗"
report += f"\n{status} [{result.time_taken:.1f}s] {result.test_case[:50]}..."
if result.error:
report += f"\n 错误: {result.error}"
return report
# 使用示例
test_cases = [
TestCase(
task="计算 15 * 23 + 47 的结果",
expected_output="392"
),
TestCase(
task="搜索Python最新版本号",
evaluation_fn=lambda x: "3." in x # 包含版本号
),
TestCase(
task="列出当前目录下的Python文件",
evaluation_fn=lambda x: ".py" in x
),
]
evaluator = AgentEvaluator(my_agent)
report = evaluator.evaluate(test_cases)
print(evaluator.generate_report())
九、主流Agent框架对比
9.1 LangGraph
LangGraph是LangChain团队推出的图状态Agent框架，支持复杂的有状态工作流。
# LangGraph核心概念示意
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
# 定义状态
class AgentState(TypedDict):
task: str
plan: list
results: dict
current_step: int
# 定义节点函数
def plan_node(state: AgentState) -> AgentState:
"""规划节点"""
plan = create_plan(state["task"])
return {"plan": plan, "current_step": 0}
def execute_node(state: AgentState) -> AgentState:
"""执行节点"""
step = state["plan"][state["current_step"]]
result = execute_step(step)
state["results"][state["current_step"]] = result
return {"current_step": state["current_step"] + 1, "results": state["results"]}
def should_continue(state: AgentState) -> str:
"""条件边：决定是否继续"""
if state["current_step"] >= len(state["plan"]):
return "end"
return "continue"
# 构建图
workflow = StateGraph(AgentState)
workflow.add_node("planner", plan_node)
workflow.add_node("executor", execute_node)
# 添加边
workflow.set_entry_point("planner")
workflow.add_edge("planner", "executor")
workflow.add_conditional_edges(
"executor",
should_continue,
{"continue": "executor", "end": END}
)
# 编译运行
app = workflow.compile()
result = app.invoke({"task": "分析销售数据", "plan": [], "results": {}, "current_step": 0})
优点：状态管理清晰、支持复杂流程、可视化调试 缺点：学习曲线较陡、概念较多
9.2 AutoGen
微软的多Agent对话框架，强调Agent之间的协作。
# AutoGen核心概念示意
from autogen import AssistantAgent, UserProxyAgent
# 创建助手Agent
assistant = AssistantAgent(
name="researcher",
system_message="你是一个研究助手，擅长分析数据和撰写报告。",
llm_config={"model": "gpt-4o-mini"}
)
# 创建用户代理（可执行代码）
user_proxy = UserProxyAgent(
name="user",
human_input_mode="NEVER", # 不需要人工输入
max_consecutive_auto_reply=10,
code_execution_config={"work_dir": "workspace"}
)
# 启动对话
user_proxy.initiate_chat(
assistant,
message="请分析当前目录下的sales.csv数据，并生成一份月度报告。"
)
优点：多Agent协作、代码执行、人机交互 缺点：对话管理复杂、调试困难
9.3 CrewAI
专注于角色扮演的多Agent框架，定义角色、目标和工具。
# CrewAI核心概念示意
from crewai import Agent, Task, Crew
# 定义Agent角色
researcher = Agent(
role="研究员",
goal="深入研究指定主题，收集全面信息",
backstory="你是一位资深研究员，擅长信息收集和分析。",
tools=[search_tool, read_tool],
verbose=True
)
writer = Agent(
role="技术作家",
goal="将研究结果转化为高质量的技术文档",
backstory="你是一位技术写作专家，擅长将复杂概念通俗化。",
tools=[write_tool],
verbose=True
)
# 定义任务
research_task = Task(
description="研究AI Agent的最新进展",
expected_output="一份详细的研究报告",
agent=researcher
)
writing_task = Task(
description="基于研究结果撰写技术教程",
expected_output="一篇完整的技术教程",
agent=writer,
context=[research_task] # 依赖研究任务的结果
)
# 组建团队并执行
crew = Crew(
agents=[researcher, writer],
tasks=[research_task, writing_task],
verbose=True
)
result = crew.kickoff()
优点：角色定义直观、协作模式清晰 缺点：灵活性有限、调试工具不足
9.4 OpenAI Agents SDK
OpenAI官方的轻量Agent框架。
# OpenAI Agents SDK核心概念示意
from openai import OpenAI
from agents import Agent, Runner, function_tool
# 定义工具
@function_tool
def get_weather(city: str) -> str:
"""获取指定城市的天气信息"""
return f"{city}今天晴朗，温度25°C"
@function_tool
def search_knowledge(query: str) -> str:
"""搜索知识库"""
return f"关于'{query}'的信息..."
# 定义Agent
agent = Agent(
name="助手",
instructions="你是一个有用的助手，可以查询天气和搜索知识。",
tools=[get_weather, search_knowledge],
model="gpt-4o-mini"
)
# 运行
result = Runner.run_sync(agent, "北京今天天气怎么样？")
print(result.final_output)
优点：简洁API、官方支持、与其他OpenAI服务集成 缺点：功能相对简单、依赖OpenAI生态
9.5 框架选型建议
| 场景 | 推荐框架 |
|---|---|
| 复杂有状态工作流 | LangGraph |
| 多Agent协作/代码执行 | AutoGen |
| 角色扮演/团队协作 | CrewAI |
| 快速原型/OpenAI生态 | OpenAI Agents SDK |
| 需要最大灵活性 | 自定义实现 |
十、Agent安全与人类监督
10.1 安全风险
Agent系统的安全风险包括：
工具滥用：Agent可能以不安全的方式使用工具（如删除文件、发送不当请求）。
权限提升：Agent可能尝试获取超出授权的权限。
信息泄露：Agent可能在不适当的场合泄露敏感信息。
幻觉与误导：Agent基于错误信息做出决策。
10.2 安全机制实现
from enum import Enum
from typing import List, Set
class RiskLevel(Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
class SecurityPolicy:
"""安全策略管理"""
def __init__(self):
self.tool_risk_levels: Dict[str, RiskLevel] = {}
self.forbidden_actions: Set[str] = set()
self.require_approval: Set[str] = set()
self.max_consecutive_actions: int = 20
def set_tool_risk(self, tool_name: str, level: RiskLevel):
self.tool_risk_levels[tool_name] = level
def add_forbidden_action(self, action: str):
self.forbidden_actions.add(action)
def require_human_approval(self, action: str):
self.require_approval.add(action)
def check_action(self, action: str, params: Dict) -> tuple:
"""检查动作是否安全"""
# 检查是否禁止
if action in self.forbidden_actions:
return False, "此操作已被禁止"
# 检查风险级别
risk = self.tool_risk_levels.get(action, RiskLevel.LOW)
if risk in (RiskLevel.HIGH, RiskLevel.CRITICAL):
if action in self.require_approval:
return None, f"需要人工审批 (风险级别: {risk.value})"
return True, "通过"
class HumanOversight:
"""人类监督机制"""
def __init__(self, security_policy: SecurityPolicy):
self.policy = security_policy
self.approval_log: List[Dict] = []
def request_approval(self, action: str, params: Dict,
reason: str) -> bool:
"""请求人工审批"""
print(f"\n{'='*50}")
print(f"⚠️ 需要人工审批")
print(f"动作: {action}")
print(f"参数: {params}")
print(f"原因: {reason}")
print(f"{'='*50}")
# 在实际系统中，这里会发送审批请求
response = input("是否批准？(y/n): ")
approved = response.lower() == 'y'
self.approval_log.append({
"action": action,
"params": params,
"approved": approved
})
return approved
class SafeAgent:
"""具有安全机制的Agent"""
def __init__(self, base_agent, security_policy: SecurityPolicy,
human_oversight: HumanOversight):
self.agent = base_agent
self.policy = security_policy
self.oversight = human_oversight
self.action_count = 0
def run(self, task: str) -> str:
# 任务安全检查
if not self._check_task_safety(task):
return "任务被安全策略拒绝"
# 包装工具调用
self._wrap_tools_with_security()
# 执行
return self.agent.run(task)
def _check_task_safety(self, task: str) -> bool:
"""检查任务本身是否安全"""
# 检查任务是否包含敏感操作
dangerous_keywords = ["删除所有", "格式化", "rm -rf", "drop table"]
for keyword in dangerous_keywords:
if keyword in task.lower():
return False
return True
def _wrap_tools_with_security(self):
"""包装工具添加安全检查"""
original_tools = self.agent.tools.copy()
for name, tool in original_tools.items():
original_execute = tool.execute
def make_safe_executor(tool_name, orig_exec):
def safe_execute(**kwargs):
self.action_count += 1
# 检查连续操作数
if self.action_count > self.policy.max_consecutive_actions:
return "超过最大连续操作数限制"
# 安全检查
allowed, message = self.policy.check_action(
tool_name, kwargs
)
if allowed is None:
# 需要人工审批
approved = self.oversight.request_approval(
tool_name, kwargs, message
)
if not approved:
return "操作被人工拒绝"
elif not allowed:
return f"安全拒绝: {message}"
return orig_exec(**kwargs)
return safe_execute
tool.execute = make_safe_executor(name, original_execute)
10.3 沙箱执行环境
class SandboxExecutor:
"""沙箱执行器：限制Agent的操作范围"""
def __init__(self, allowed_dirs: List[str] = None,
allowed_commands: List[str] = None,
network_access: bool = False):
self.allowed_dirs = allowed_dirs or []
self.allowed_commands = allowed_commands or []
self.network_access = network_access
def execute_code(self, code: str) -> str:
"""在沙箱中执行代码"""
import subprocess
import tempfile
# 写入临时文件
with tempfile.NamedTemporaryFile(mode='w', suffix='.py',
delete=False) as f:
f.write(code)
temp_path = f.name
# 构建受限执行命令
cmd = [
"python", temp_path
]
# 设置环境变量限制
env = {
"PYTHONPATH": "",
"SANDBOX_MODE": "1"
}
if not self.network_access:
env["NO_NETWORK"] = "1"
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=30,
env=env
)
return result.stdout if result.returncode == 0 else result.stderr
except subprocess.TimeoutExpired:
return "执行超时"
finally:
import os
os.unlink(temp_path)
十一、实战案例：构建自主研究Agent
11.1 项目概述
我们将构建一个自主研究Agent，它能够：
- 理解研究问题
- 制定研究计划
- 搜索和收集信息
- 分析和综合信息
- 生成研究报告
11.2 完整实现
# research_agent.py
import json
from typing import List, Dict, Optional
from dataclasses import dataclass, field
@dataclass
class ResearchSource:
"""研究来源"""
title: str
content: str
url: str = ""
relevance_score: float = 0.0
@dataclass
class ResearchState:
"""研究状态"""
question: str
plan: List[str] = field(default_factory=list)
sources: List[ResearchSource] = field(default_factory=list)
findings: Dict[str, str] = field(default_factory=dict)
current_step: int = 0
status: str = "initialized" # initialized, planning, researching, analyzing, completed
class ResearchAgent:
"""自主研究Agent"""
def __init__(self, llm, search_tool, memory):
self.llm = llm
self.search_tool = search_tool
self.memory = memory
self.state: Optional[ResearchState] = None
def research(self, question: str) -> str:
"""执行研究任务"""
self.state = ResearchState(question=question)
print(f"开始研究: {question}\n")
# 阶段1: 制定研究计划
self._create_plan()
# 阶段2: 收集信息
self._gather_information()
# 阶段3: 分析和综合
analysis = self._analyze_findings()
# 阶段4: 生成报告
report = self._generate_report()
self.state.status = "completed"
return report
def _create_plan(self):
"""制定研究计划"""
self.state.status = "planning"
print("📋 制定研究计划...")
prompt = f"""研究问题: {self.state.question}
请制定一个详细的研究计划，包括：
1. 需要搜索的关键主题
2. 需要回答的子问题
3. 信息收集的优先顺序
输出格式：
主题1: xxx
主题2: xxx
..."""
response = self.llm.generate(prompt)
# 解析计划
self.state.plan = []
for line in response.strip().split("\n"):
if line.strip().startswith("主题"):
parts = line.split(":", 1)
if len(parts) > 1:
self.state.plan.append(parts[1].strip())
print(f"研究计划: {self.state.plan}\n")
def _gather_information(self):
"""收集信息"""
self.state.status = "researching"
print("🔍 收集信息...\n")
for i, topic in enumerate(self.state.plan):
print(f" 搜索主题 {i+1}: {topic}")
# 生成搜索查询
queries = self._generate_search_queries(topic)
# 执行搜索
for query in queries:
results = self.search_tool.execute(query=query)
# 处理搜索结果
sources = self._process_search_results(query, results)
self.state.sources.extend(sources)
print(f" 查询 '{query}' 找到 {len(sources)} 个相关来源")
# 提取关键发现
finding = self._extract_findings(topic)
self.state.findings[topic] = finding
self.state.current_step = i + 1
print(f" 关键发现: {finding[:100]}...\n")
def _generate_search_queries(self, topic: str) -> List[str]:
"""为给定主题生成搜索查询"""
prompt = f"""研究主题: {topic}
相关问题: {self.state.question}
请生成2-3个不同的搜索查询，从不同角度搜索信息。
每行一个查询："""
response = self.llm.generate(prompt)
queries = [q.strip() for q in response.strip().split("\n") if q.strip()]
return queries[:3]
def _process_search_results(self, query: str, results: str) -> List[ResearchSource]:
"""处理搜索结果"""
# 简化处理：将结果拆分为来源
sources = []
sections = results.split("\n\n")
for i, section in enumerate(sections[:3]):
if len(section.strip()) > 50:
source = ResearchSource(
title=f"来源_{query}_{i}",
content=section.strip(),
relevance_score=1.0 - (i * 0.2)
)
sources.append(source)
return sources
def _extract_findings(self, topic: str) -> str:
"""从收集的来源中提取关键发现"""
# 获取与当前主题相关的来源
relevant_sources = [
s.content for s in self.state.sources
if topic.lower() in s.content.lower()
][:5]
if not relevant_sources:
return "未找到相关信息"
sources_text = "\n---\n".join(relevant_sources)
prompt = f"""研究主题: {topic}
研究问题: {self.state.question}
基于以下来源信息，提取关键发现和要点：
{sources_text}
请总结3-5个关键发现："""
return self.llm.generate(prompt)
def _analyze_findings(self) -> str:
"""分析和综合所有发现"""
self.state.status = "analyzing"
print("📊 分析研究发现...\n")
findings_text = "\n\n".join(
f"## {topic}\n{finding}"
for topic, finding in self.state.findings.items()
)
prompt = f"""研究问题: {self.state.question}
研究发现:
{findings_text}
请进行深入分析：
1. 这些发现之间有什么关联？
2. 是否存在矛盾或不一致？
3. 最重要的结论是什么？
4. 还有哪些未解答的问题？
分析结果："""
return self.llm.generate(prompt)
def _generate_report(self) -> str:
"""生成最终研究报告"""
print("📝 生成研究报告...\n")
analysis = self._analyze_findings()
prompt = f"""请基于以下研究内容，生成一份完整的研究报告。
研究问题: {self.state.question}
研究发现:
{json.dumps(self.state.findings, ensure_ascii=False, indent=2)}
分析:
{analysis}
请按照以下结构生成报告：
1. 摘要
2. 背景
3. 主要发现
4. 分析与讨论
5. 结论
6. 参考来源
报告："""
report = self.llm.generate(prompt)
# 保存到长期记忆
self.memory.remember(
f"研究报告: {self.state.question}\n{report[:500]}",
memory_type="research"
)
return report
# 使用示例
class MockLLM:
"""模拟LLM（用于演示）"""
def generate(self, prompt: str) -> str:
# 实际使用时替换为真实LLM调用
return "这是LLM的模拟响应..."
class MockSearchTool:
"""模拟搜索工具"""
def execute(self, query: str) -> str:
return f"搜索 '{query}' 的结果...\n相关内容片段1...\n相关内容片段2..."
# 运行研究Agent
if __name__ == "__main__":
llm = MockLLM()
search_tool = MockSearchTool()
memory = AgentMemory(None, None) # 简化示意
agent = ResearchAgent(llm, search_tool, memory)
report = agent.research("2024年AI Agent技术的发展趋势是什么？")
print(report)
11.3 增强功能
class EnhancedResearchAgent(ResearchAgent):
"""增强版研究Agent"""
def __init__(self, llm, tools, memory):
super().__init__(llm, tools.get("search"), memory)
self.tools = tools
self.verifier = SelfVerifyingAgent(llm, tools)
def research(self, question: str) -> str:
"""增强的研究流程"""
# 初始研究
initial_report = super().research(question)
# 自我验证
is_valid, feedback = self.verifier._verify(question, initial_report)
if not is_valid:
print(f"\n验证反馈: {feedback}")
print("进行补充研究...")
# 补充研究
additional_findings = self._supplemental_research(feedback)
# 修正报告
final_report = self._revise_report(
question, initial_report, additional_findings
)
return final_report
return initial_report
def _supplemental_research(self, feedback: str) -> str:
"""基于反馈进行补充研究"""
prompt = f"""基于以下反馈，确定需要补充研究的内容：
{feedback}
请列出需要进一步研究的问题："""
additional_questions = self.llm.generate(prompt)
# 执行补充搜索
findings = []
for question in additional_questions.split("\n"):
if question.strip():
result = self.search_tool.execute(query=question.strip())
findings.append(result)
return "\n".join(findings)
def _revise_report(self, question: str, original: str,
additional: str) -> str:
"""修正报告"""
prompt = f"""原始报告:
{original}
补充研究结果:
{additional}
请修正和完善报告，确保回答完整、准确。"""
return self.llm.generate(prompt)
十二、最佳实践
12.1 设计原则
- 单一职责：每个Agent只负责一个明确的任务
- 可组合性：Agent可以组合使用，形成更强大的系统
- 可观察性：Agent的每一步都应该可追踪、可调试
- 优雅降级：当工具或外部服务不可用时，有合理的回退方案
12.2 开发建议
- 从小开始：先实现核心功能，再逐步增加复杂性
- 充分测试：建立完善的评估数据集和测试流程
- 日志记录：记录Agent的每一步决策和行动
- 人工验证：关键决策点加入人工审核
12.3 性能优化
- 缓存：缓存重复的工具调用结果
- 并发：独立的子任务可以并发执行
- 提前终止：当结果已经足够好时，提前结束
- 工具选择：根据任务特点选择最合适的工具
十三、常见问题
Q1: Agent总是无限循环怎么办？
设置明确的终止条件：
- 最大迭代次数
- 重复检测（如果连续N次行动相同则终止）
- 进度检查（如果长时间没有进展则终止）
- 成本限制（API调用次数/金额上限）
Q2: 如何处理LLM的幻觉？
- 要求Agent引用来源
- 实施自我验证机制
- 使用工具验证关键信息
- 设置置信度阈值
Q3: 如何降低Agent的成本？
- 使用更小的模型处理简单任务
- 缓存重复查询
- 减少不必要的工具调用
- 优化Prompt减少Token使用
Q4: 多Agent系统如何处理冲突？
- 定义清晰的职责边界
- 实施仲裁机制
- 使用优先级系统
- 引入人类监督
Q5: 如何评估Agent的安全性？
- 建立安全测试用例集
- 红队测试（尝试让Agent做不安全的事）
- 监控异常行为模式
- 定期审计Agent的决策日志
十四、总结
AI Agent代表了人工智能从"工具"到"助手"再到"自主体"的演进方向。构建高质量的Agent系统需要综合考虑：
- 记忆系统：短期记忆保持对话连贯，长期记忆积累经验，工作记忆管理当前任务
- 规划能力：ReAct适合简单任务，Plan-and-Execute适合复杂任务，Reflexion适合需要迭代改进的任务
- 工具使用：清晰的工具定义、可靠的错误处理、安全的执行环境
- 安全机制：权限控制、人类监督、沙箱执行
技术栈推荐：
- 快速原型：OpenAI Agents SDK + Function Calling
- 复杂工作流：LangGraph
- 多Agent协作：AutoGen 或 CrewAI
- 生产环境：自定义实现 + 安全框架
未来的发展方向包括：
- 更强的自主规划能力
- 更好的多模态理解
- 更安全的Agent行为
- Agent之间的标准化协作协议
掌握本教程的核心技术和工程实践，你将能够构建出强大、可靠、安全的AI Agent系统，为用户提供真正有价值的智能服务。
