AI Agent 智能体开发完全教程
前言
人工智能正在从"被动响应"走向"主动行动"。传统的AI应用是"你问我答"式的——用户提供输入，模型返回输出。而AI Agent（智能体）则代表了一种全新的范式：AI能够自主感知环境、制定计划、调用工具、执行任务，并在过程中不断反思和调整。
本教程将从零开始，系统性地讲解AI Agent的核心概念、架构设计、关键技术实现，并通过一个完整的实战项目帮助你掌握Agent开发的全流程。
第一章：AI Agent核心概念与发展历程
1.1 什么是AI Agent
AI Agent（智能体）是一个能够自主感知环境、进行推理、采取行动以实现特定目标的AI系统。与传统的LLM应用不同，Agent具有以下核心特征：
- 自主性（Autonomy）：能够在没有人类逐步指导的情况下独立完成任务
- 目标导向（Goal-oriented）：围绕明确的目标进行规划和执行
- 工具使用（Tool Use）：能够调用外部工具（搜索引擎、数据库、API等）来扩展自身能力
- 记忆能力（Memory）：能够记住历史交互，从过去的经验中学习
- 反思能力（Reflection）：能够评估自己的行为结果，发现错误并进行纠正
简单来说，传统LLM是一个"大脑"，而Agent是给这个大脑装上了"眼睛"（感知）、"手"（工具）和"记忆"（上下文管理）。
1.2 发展历程
AI Agent的概念并非新生事物，但大语言模型（LLM）的出现让它真正走向实用化。
第一阶段：传统AI Agent（1950s-2020）
早期的Agent研究主要集中在强化学习领域，如DeepMind的AlphaGo、OpenAI的Dota2 Agent等。这些Agent通常针对特定任务训练，泛化能力有限。
第二阶段：LLM驱动的Agent（2022-2023）
随着ChatGPT等大模型的爆发，研究者发现LLM具备强大的推理和规划能力，可以作为Agent的"大脑"。标志性的工作包括：
- ReAct（2022）：将推理（Reasoning）和行动（Acting）结合，Agent先思考再行动
- AutoGPT（2023）：第一个引起广泛关注的自主Agent，能够自动分解任务并执行
- BabyAGI（2023）：基于任务驱动的自主Agent系统
- Toolformer（2023）：Meta提出的让LLM自主学习使用工具的方法
第三阶段：多Agent与生产化（2024-至今）
当前阶段的特点是多Agent协作框架的成熟和生产级部署的探索：
- LangGraph、CrewAI、AutoGen等框架降低了Agent开发的门槛
- Function Calling成为工具调用的标准接口
- 企业开始将Agent应用于客服、代码生成、数据分析等实际业务场景
1.3 三种核心推理范式
ReAct（Reasoning + Acting）
ReAct是最经典也是目前最广泛使用的Agent推理范式。其核心思想是：Agent在每一步都先进行推理（Thought），再决定行动（Action），然后观察结果（Observation），如此循环。
用户输入 → Thought（思考）→ Action（行动）→ Observation（观察）→ Thought → ...
ReAct的优势：
- 推理过程透明，便于调试和理解
- 每一步都有明确的思考，减少盲目行动
- 与LLM的Chain-of-Thought能力天然契合
Plan-and-Execute（先规划后执行）
这种范式将任务分为两个阶段：
- 规划阶段：Agent先制定一个完整的执行计划，将复杂任务分解为多个子任务
- 执行阶段：按照计划逐步执行每个子任务
用户输入 → Planner（制定计划）→ [Task1, Task2, Task3, ...] → Executor（逐步执行）
Plan-and-Execute的优势：
- 适合复杂的多步骤任务
- 全局视角的规划可以避免局部最优
- 计划可以被人类审核和修改
Reflection（反思）
反思范式在ReAct基础上增加了自我评估环节。Agent在执行后会回顾自己的行为，评估效果，并据此调整策略。
用户输入 → Thought → Action → Observation → Reflection（反思）→ 调整策略 → 继续
Reflection的优势：
- 能够发现和纠正错误
- 从失败中学习，避免重复犯错
- 输出质量更高，但推理成本也更高
第二章：Agent架构设计——感知-推理-行动循环
2.1 Agent核心架构
一个完整的AI Agent系统由以下模块组成：
┌─────────────────────────────────────────────┐
│ 用户输入 │
└──────────────────┬──────────────────────────┘
▼
┌──────────────────────────────────────────────┐
│ 感知模块（Perception） │
│ - 解析用户意图 │
│ - 处理多模态输入（文本、图片、音频） │
│ - 提取关键信息 │
└──────────────────┬───────────────────────────┘
▼
┌──────────────────────────────────────────────┐
│ 推理模块（Reasoning） │
│ - 任务分解与规划 │
│ - 决定下一步行动 │
│ - 评估行动结果 │
└──────────────────┬───────────────────────────┘
▼
┌──────────────────────────────────────────────┐
│ 行动模块（Action） │
│ - 调用工具执行操作 │
│ - 与外部环境交互 │
│ - 生成最终响应 │
└──────────────────┬───────────────────────────┘
▼
┌──────────────────────────────────────────────┐
│ 记忆模块（Memory） │
│ - 短期记忆：当前对话上下文 │
│ - 长期记忆：历史交互和知识 │
│ - 工作记忆：当前任务的中间状态 │
└──────────────────────────────────────────────┘
2.2 感知-推理-行动循环的实现
下面是一个简化的Agent循环的Python实现：
import openai
import json
from typing import List, Dict, Any, Callable
class SimpleAgent:
def __init__(self, model: str = "gpt-4"):
self.client = openai.OpenAI()
self.model = model
self.tools: Dict[str, Callable] = {}
self.tool_descriptions: List[Dict] = []
self.conversation_history: List[Dict] = []
self.max_iterations = 10
def register_tool(self, name: str, description: str, parameters: dict, func: Callable):
"""注册一个工具"""
self.tools[name] = func
self.tool_descriptions.append({
"type": "function",
"function": {
"name": name,
"description": description,
"parameters": parameters
}
})
def perceive(self, user_input: str) -> str:
"""感知：解析用户输入"""
self.conversation_history.append({
"role": "user",
"content": user_input
})
return user_input
def reason(self) -> Dict[str, Any]:
"""推理：决定下一步行动"""
system_prompt = """你是一个智能助手，可以使用工具来帮助用户完成任务。
在每一步，你应该：
1. 思考当前的情况
2. 决定是直接回答还是调用工具
3. 如果需要调用工具，选择合适的工具和参数
请始终用中文回复。"""
messages = [
{"role": "system", "content": system_prompt}
] + self.conversation_history
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
tools=self.tool_descriptions if self.tool_descriptions else None,
)
return response.choices[0].message
def act(self, message) -> str:
"""行动：执行工具调用或返回最终响应"""
# 如果不需要调用工具，直接返回响应
if not message.tool_calls:
response_text = message.content
self.conversation_history.append({
"role": "assistant",
"content": response_text
})
return response_text
# 如果需要调用工具
self.conversation_history.append(message.model_dump())
for tool_call in message.tool_calls:
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
print(f" [Agent] 调用工具: {func_name}({func_args})")
# 执行工具
if func_name in self.tools:
try:
result = self.tools[func_name](**func_args)
except Exception as e:
result = f"工具执行错误: {str(e)}"
else:
result = f"未知工具: {func_name}"
# 将工具结果添加到对话历史
self.conversation_history.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": str(result)
})
return None # 需要继续循环
def run(self, user_input: str) -> str:
"""运行Agent的主循环"""
print(f"\n[用户] {user_input}")
# 感知
self.perceive(user_input)
# 推理-行动循环
for i in range(self.max_iterations):
print(f"\n--- 迭代 {i + 1} ---")
# 推理
message = self.reason()
# 行动
result = self.act(message)
if result is not None:
print(f"\n[Agent回复] {result}")
return result
return "达到最大迭代次数，任务未能完成。"
2.3 Agent循环的关键设计决策
在设计Agent循环时，有几个关键问题需要考虑：
1. 何时停止？
Agent需要一个明确的停止条件，否则可能陷入无限循环。常见的策略：
- 最大迭代次数限制
- LLM自主判断任务完成
- 用户确认机制
- 成本预算限制
2. 错误处理
当工具调用失败时，Agent应该能够：
- 记录错误信息
- 尝试替代方案
- 向用户报告问题
- 优雅地降级
3. 上下文管理
随着对话变长，上下文窗口可能不够用。需要：
- 压缩历史消息
- 保留关键信息
- 使用外部记忆存储
第三章：工具调用Function Calling机制详解
3.1 什么是Function Calling
Function Calling是大模型与外部世界交互的标准接口。它允许模型在需要时调用预定义的函数，获取实时信息或执行特定操作。
工作流程如下：
- 定义工具：在调用模型时，提供可用工具的描述（名称、功能、参数）
- 模型决策：模型根据用户请求，决定是否需要调用工具，以及调用哪个工具、传什么参数
- 执行工具：应用程序执行实际的工具调用
- 返回结果：将工具执行结果返回给模型
- 生成回复：模型根据工具结果生成最终回复
3.2 工具定义规范
每个工具需要定义以下信息：
tools = [
{
"type": "function",
"function": {
"name": "search_web",
"description": "搜索互联网获取最新信息。当你需要回答关于实时事件、最新数据或你不确定的事实时使用。",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词，应该简洁明了"
},
"num_results": {
"type": "integer",
"description": "返回结果数量，默认为5",
"default": 5
}
},
"required": ["query"]
}
}
},
{
"type": "function",
"function": {
"name": "read_file",
"description": "读取本地文件的内容。支持文本文件和代码文件。",
"parameters": {
"type": "object",
"properties": {
"file_path": {
"type": "string",
"description": "文件的完整路径"
},
"encoding": {
"type": "string",
"description": "文件编码，默认utf-8",
"default": "utf-8"
}
},
"required": ["file_path"]
}
}
}
]
工具描述的最佳实践：
- 名称清晰：使用动词+名词的格式，如
search_web、read_file、send_email - 描述精确：说明工具的功能和适用场景，帮助模型做出正确决策
- 参数完整：每个参数都要有类型、描述和是否必填
- 避免歧义：不同工具之间功能不要重叠，描述要区分清楚
3.3 并行工具调用
现代LLM支持在一次响应中返回多个工具调用，这在需要同时获取多个信息时非常高效：
# 模型可能在一次响应中返回多个工具调用
# 例如：同时查询天气和日程
# tool_call_1: get_weather(city="北京")
# tool_call_2: get_calendar(date="2024-01-15")
# 处理并行工具调用
for tool_call in message.tool_calls:
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
# 每个工具调用独立执行
result = execute_tool(func_name, func_args)
3.4 工具调用的错误处理
def safe_tool_execution(func_name: str, func_args: dict, tools: dict) -> str:
"""安全的工具执行包装器"""
if func_name not in tools:
return json.dumps({
"error": f"工具 '{func_name}' 不存在",
"available_tools": list(tools.keys())
})
try:
result = tools[func_name](**func_args)
return json.dumps({"result": result}, ensure_ascii=False)
except TypeError as e:
return json.dumps({
"error": f"参数错误: {str(e)}",
"expected_params": inspect.signature(tools[func_name]).parameters
})
except Exception as e:
return json.dumps({
"error": f"执行错误: {str(e)}",
"tool": func_name,
"args": func_args
})
3.5 构建可复用的工具注册系统
from typing import Callable, Dict, Any
from functools import wraps
import inspect
class ToolRegistry:
"""工具注册中心"""
def __init__(self):
self._tools: Dict[str, Dict] = {}
def tool(self, name: str = None, description: str = ""):
"""装饰器：注册一个工具"""
def decorator(func: Callable):
tool_name = name or func.__name__
# 从函数签名推断参数
sig = inspect.signature(func)
params = {}
required = []
for param_name, param in sig.parameters.items():
param_info = {
"type": "string", # 默认类型
"description": f"参数 {param_name}"
}
# 尝试从类型注解推断
if param.annotation != inspect.Parameter.empty:
type_map = {
str: "string",
int: "integer",
float: "number",
bool: "boolean",
list: "array",
dict: "object"
}
param_info["type"] = type_map.get(param.annotation, "string")
# 检查是否有默认值
if param.default == inspect.Parameter.empty:
required.append(param_name)
params[param_name] = param_info
self._tools[tool_name] = {
"function": func,
"schema": {
"type": "function",
"function": {
"name": tool_name,
"description": description or func.__doc__ or "",
"parameters": {
"type": "object",
"properties": params,
"required": required
}
}
}
}
@wraps(func)
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
return wrapper
return decorator
def get_schemas(self) -> list:
"""获取所有工具的schema，用于传递给LLM"""
return [tool["schema"] for tool in self._tools.values()]
def execute(self, name: str, args: dict) -> str:
"""执行指定工具"""
if name not in self._tools:
return f"错误：工具 '{name}' 不存在"
try:
result = self._tools[name]["function"](**args)
return str(result)
except Exception as e:
return f"执行错误：{str(e)}"
# 使用示例
registry = ToolRegistry()
@registry.tool(description="计算数学表达式的结果")
def calculate(expression: str) -> str:
"""计算数学表达式"""
try:
result = eval(expression) # 注意：生产环境应使用安全的表达式解析器
return str(result)
except Exception as e:
return f"计算错误: {e}"
@registry.tool(description="获取当前日期和时间")
def get_current_time() -> str:
"""获取当前时间"""
from datetime import datetime
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
第四章：记忆系统设计
4.1 记忆的重要性
人类的智能很大程度上依赖于记忆——我们记住过去的经历，从中学习，并将知识应用到新的情境中。同样，一个没有记忆的Agent每次对话都像"失忆"一样，无法积累经验，无法个性化服务。
Agent的记忆系统通常分为三种类型：
4.2 短期记忆（Short-term Memory）
短期记忆就是当前对话的上下文窗口。它是Agent"此刻正在思考的内容"。
class ShortTermMemory:
"""短期记忆：管理当前对话上下文"""
def __init__(self, max_tokens: int = 8000):
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) -> List[Dict]:
return self.messages.copy()
def _trim(self):
"""当消息过多时，保留最近的消息"""
while self._estimate_tokens() > self.max_tokens and len(self.messages) > 2:
# 保留系统消息和最新的消息
self.messages.pop(1) # 移除第二条消息（第一条通常是系统消息）
def _estimate_tokens(self) -> int:
"""粗略估计token数量"""
total_chars = sum(len(m["content"]) for m in self.messages)
return total_chars // 2 # 粗略估计：2个字符≈1个token
def clear(self):
self.messages.clear()
4.3 长期记忆（Long-term Memory）
长期记忆存储跨对话的信息，包括用户偏好、历史交互摘要、学到的知识等。通常使用向量数据库来实现语义检索。
import json
import hashlib
from datetime import datetime
from typing import List, Dict, Optional
class LongTermMemory:
"""长期记忆：基于向量检索的持久化记忆"""
def __init__(self, embedding_func=None):
self.memories: List[Dict] = []
self.embedding_func = embedding_func
def store(self, content: str, metadata: Dict = None):
"""存储一条记忆"""
memory = {
"id": hashlib.md5(content.encode()).hexdigest()[:8],
"content": content,
"timestamp": datetime.now().isoformat(),
"metadata": metadata or {},
"access_count": 0
}
self.memories.append(memory)
def retrieve(self, query: str, top_k: int = 5) -> List[Dict]:
"""根据查询检索相关记忆"""
if not self.memories:
return []
if self.embedding_func:
# 使用向量相似度检索
query_embedding = self.embedding_func(query)
scored = []
for mem in self.memories:
mem_embedding = self.embedding_func(mem["content"])
similarity = self._cosine_similarity(query_embedding, mem_embedding)
scored.append((similarity, mem))
scored.sort(reverse=True, key=lambda x: x[0])
results = [mem for _, mem in scored[:top_k]]
else:
# 简单的关键词匹配
query_lower = query.lower()
scored = []
for mem in self.memories:
content_lower = mem["content"].lower()
# 计算关键词重叠度
query_words = set(query_lower.split())
content_words = set(content_lower.split())
overlap = len(query_words & content_words)
scored.append((overlap, mem))
scored.sort(reverse=True, key=lambda x: x[0])
results = [mem for _, mem in scored[:top_k] if scored[0][0] > 0]
# 更新访问计数
for mem in results:
mem["access_count"] += 1
return results
def _cosine_similarity(self, a, b) -> float:
"""计算余弦相似度"""
import numpy as np
a, b = np.array(a), np.array(b)
return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))
def summarize_and_forget(self, max_memories: int = 100):
"""当记忆过多时，摘要并遗忘不重要的记忆"""
if len(self.memories) <= max_memories:
return
# 按访问频率和时间排序，保留重要的
self.memories.sort(key=lambda m: (m["access_count"], m["timestamp"]), reverse=True)
self.memories = self.memories[:max_memories]
4.4 工作记忆（Working Memory）
工作记忆是Agent在执行当前任务时的"白板"，存储任务分解结果、中间计算状态等。
class WorkingMemory:
"""工作记忆：管理当前任务的执行状态"""
def __init__(self):
self.current_task: Optional[str] = None
self.subtasks: List[Dict] = []
self.scratchpad: Dict[str, Any] = {} # 临时存储
self.execution_log: List[Dict] = []
def set_task(self, task: str):
"""设置当前任务"""
self.current_task = task
self.subtasks = []
self.scratchpad = {}
self.execution_log = []
def add_subtask(self, description: str, status: str = "pending"):
"""添加子任务"""
self.subtasks.append({
"description": description,
"status": status,
"result": None
})
def update_subtask(self, index: int, status: str, result: Any = None):
"""更新子任务状态"""
if 0 <= index < len(self.subtasks):
self.subtasks[index]["status"] = status
self.subtasks[index]["result"] = result
def set_scratchpad(self, key: str, value: Any):
"""在白板上记录信息"""
self.scratchpad[key] = value
def get_scratchpad(self, key: str) -> Any:
"""读取白板信息"""
return self.scratchpad.get(key)
def log(self, action: str, result: str):
"""记录执行日志"""
self.execution_log.append({
"timestamp": datetime.now().isoformat(),
"action": action,
"result": result
})
def get_summary(self) -> str:
"""获取当前工作状态摘要"""
completed = sum(1 for t in self.subtasks if t["status"] == "completed")
total = len(self.subtasks)
summary = f"当前任务: {self.current_task}\n"
summary += f"进度: {completed}/{total} 子任务完成\n"
if self.subtasks:
summary += "\n子任务列表:\n"
for i, task in enumerate(self.subtasks):
status_icon = {"pending": "⏳", "in_progress": "🔄", "completed": "✅", "failed": "❌"}
icon = status_icon.get(task["status"], "❓")
summary += f" {icon} {task['description']}\n"
return summary
def clear(self):
"""清空工作记忆"""
self.current_task = None
self.subtasks = []
self.scratchpad = {}
self.execution_log = []
4.5 三层记忆整合
class AgentMemorySystem:
"""整合三层记忆的统一接口"""
def __init__(self):
self.short_term = ShortTermMemory(max_tokens=8000)
self.long_term = LongTermMemory()
self.working = WorkingMemory()
def get_full_context(self, query: str) -> str:
"""获取完整的上下文信息"""
context_parts = []
# 1. 当前对话上下文
recent_messages = self.short_term.get_context()[-5:] # 最近5条
if recent_messages:
context_parts.append("=== 近期对话 ===")
for msg in recent_messages:
context_parts.append(f"[{msg['role']}]: {msg['content'][:200]}")
# 2. 相关的长期记忆
relevant_memories = self.long_term.retrieve(query, top_k=3)
if relevant_memories:
context_parts.append("\n=== 相关记忆 ===")
for mem in relevant_memories:
context_parts.append(f"- {mem['content'][:200]}")
# 3. 当前任务状态
if self.working.current_task:
context_parts.append(f"\n=== 任务状态 ===")
context_parts.append(self.working.get_summary())
return "\n".join(context_parts)
def after_interaction(self, user_input: str, agent_response: str):
"""交互后的记忆更新"""
# 添加到短期记忆
self.short_term.add("user", user_input)
self.short_term.add("assistant", agent_response)
# 如果是重要信息，存储到长期记忆
if self._is_important(user_input):
self.long_term.store(user_input, {"type": "user_input"})
def _is_important(self, content: str) -> bool:
"""判断内容是否值得存入长期记忆"""
important_keywords = ["记住", "重要", "下次", "偏好", "喜欢", "不喜欢", "密码", "地址"]
return any(kw in content for kw in important_keywords)
第五章：多Agent协作架构
5.1 为什么需要多Agent
单个Agent在处理复杂任务时面临诸多挑战：
- 上下文窗口限制：单个Agent的"工作记忆"有限
- 专业性不足：一个Agent难以精通所有领域
- 效率问题：串行执行多个子任务效率低下
- 可靠性：单点故障风险
多Agent协作通过分工合作来解决这些问题。
5.2 主从模式（Master-Slave）
主从模式中有一个"主Agent"负责任务分解和协调，多个"从Agent"负责具体执行。
class MasterAgent:
"""主Agent：负责任务分解和协调"""
def __init__(self, model: str = "gpt-4"):
self.client = openai.OpenAI()
self.model = model
self.worker_agents: Dict[str, 'WorkerAgent'] = {}
def register_worker(self, name: str, agent: 'WorkerAgent'):
"""注册从Agent"""
self.worker_agents[name] = agent
def decompose_task(self, task: str) -> List[Dict]:
"""将复杂任务分解为子任务"""
prompt = f"""请将以下任务分解为可执行的子任务：
任务：{task}
可用的工作者：
{chr(10).join(f'- {name}: {agent.description}' for name, agent in self.worker_agents.items())}
请以JSON格式返回子任务列表：
[{{"task": "子任务描述", "worker": "负责的工作者名称", "dependencies": ["依赖的子任务索引"]}}]
"""
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"}
)
result = json.loads(response.choices[0].message.content)
return result.get("subtasks", [])
def execute(self, task: str) -> str:
"""执行任务"""
print(f"[Master] 分析任务: {task}")
# 1. 分解任务
subtasks = self.decompose_task(task)
print(f"[Master] 分解为 {len(subtasks)} 个子任务")
# 2. 按依赖顺序执行
results = {}
for i, subtask in enumerate(subtasks):
worker_name = subtask["worker"]
task_desc = subtask["task"]
# 检查依赖是否完成
deps = subtask.get("dependencies", [])
dep_results = {d: results[d] for d in deps if d in results}
print(f"[Master] 执行子任务 {i}: {task_desc} (工作者: {worker_name})")
if worker_name in self.worker_agents:
result = self.worker_agents[worker_name].execute(task_desc, dep_results)
results[i] = result
else:
results[i] = f"错误：找不到工作者 '{worker_name}'"
# 3. 整合结果
return self._synthesize(task, results)
def _synthesize(self, original_task: str, results: Dict) -> str:
"""整合各子任务的结果"""
prompt = f"""请根据以下子任务结果，生成对原始任务的完整回答：
原始任务：{original_task}
子任务结果：
{json.dumps(results, ensure_ascii=False, indent=2)}
请生成一个连贯、完整的回答。"""
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.content
class WorkerAgent:
"""从Agent：执行具体的子任务"""
def __init__(self, name: str, description: str, model: str = "gpt-4"):
self.name = name
self.description = description
self.client = openai.OpenAI()
self.model = model
self.tools = []
def execute(self, task: str, context: Dict = None) -> str:
"""执行子任务"""
messages = [
{"role": "system", "content": f"你是{self.name}。{self.description}"},
{"role": "user", "content": f"任务：{task}\n\n上下文：{json.dumps(context or {}, ensure_ascii=False)}"}
]
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
tools=self.tools if self.tools else None
)
return response.choices[0].message.content
5.3 对等模式（Peer-to-Peer）
对等模式中所有Agent地位平等，通过消息传递进行协作。适合需要多角度分析的场景。
class PeerAgent:
"""对等Agent：可以与其他Agent通信"""
def __init__(self, name: str, expertise: str, model: str = "gpt-4"):
self.name = name
self.expertise = expertise
self.client = openai.OpenAI()
self.model = model
self.message_queue: List[Dict] = []
def receive_message(self, from_agent: str, message: str):
"""接收来自其他Agent的消息"""
self.message_queue.append({
"from": from_agent,
"content": message,
"timestamp": datetime.now().isoformat()
})
def think(self, topic: str, other_opinions: List[Dict]) -> str:
"""基于主题和其他Agent的观点进行思考"""
opinions_text = "\n".join([
f"[{op['from']}]: {op['content']}" for op in other_opinions
])
prompt = f"""你是一个{self.expertise}领域的专家。
讨论主题：{topic}
其他专家的观点：
{opinions_text}
请从你的专业角度给出你的分析和观点。如果同意其他专家，请补充细节。如果不同意，请指出问题并给出理由。"""
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.content
class PeerNetwork:
"""对等Agent网络"""
def __init__(self):
self.agents: Dict[str, PeerAgent] = {}
def add_agent(self, agent: PeerAgent):
self.agents[agent.name] = agent
def discuss(self, topic: str, rounds: int = 2) -> Dict[str, str]:
"""多轮讨论"""
all_opinions = []
final_opinions = {}
for round_num in range(rounds):
print(f"\n=== 第 {round_num + 1} 轮讨论 ===")
for name, agent in self.agents.items():
opinion = agent.think(topic, all_opinions)
print(f"[{name}]: {opinion[:100]}...")
all_opinions.append({
"from": name,
"content": opinion,
"round": round_num
})
if round_num == rounds - 1:
final_opinions[name] = opinion
return final_opinions
5.4 层级模式（Hierarchical）
层级模式适用于大型组织结构，有多层管理层级：
class HierarchicalOrchestrator:
"""层级式Agent编排器"""
def __init__(self):
self.agents: Dict[str, Dict] = {} # agent_id -> {agent, parent, children}
def add_agent(self, agent_id: str, agent, parent_id: str = None):
"""添加Agent到层级结构"""
self.agents[agent_id] = {
"agent": agent,
"parent": parent_id,
"children": []
}
if parent_id and parent_id in self.agents:
self.agents[parent_id]["children"].append(agent_id)
def execute(self, root_id: str, task: str) -> str:
"""从根节点开始执行任务"""
return self._execute_recursive(root_id, task)
def _execute_recursive(self, agent_id: str, task: str) -> str:
"""递归执行"""
agent_info = self.agents[agent_id]
agent = agent_info["agent"]
children = agent_info["children"]
if not children:
# 叶子节点，直接执行
return agent.execute(task)
# 非叶子节点，分解任务并分发给子节点
subtasks = agent.decompose_task(task, children)
results = {}
for child_id, subtask in subtasks.items():
results[child_id] = self._execute_recursive(child_id, subtask)
# 整合子节点结果
return agent.synthesize(task, results)
第六章：Agent安全与对齐
6.1 安全风险概览
AI Agent的安全问题比传统LLM应用更为严峻，因为Agent具有行动能力。主要风险包括：
- 越权操作：Agent可能执行超出预期范围的操作
- 数据泄露：Agent可能将敏感信息发送到外部
- 无限循环：Agent可能陷入死循环，消耗大量资源
- 误导信息：Agent可能基于错误信息做出有害决策
- 提示注入：恶意用户可能通过精心构造的输入操控Agent
6.2 沙箱执行环境
对于代码执行等高风险操作，必须在沙箱中运行：
import subprocess
import tempfile
import os
import resource
class SandboxExecutor:
"""安全的代码执行沙箱"""
def __init__(self, timeout: int = 30, max_memory_mb: int = 256):
self.timeout = timeout
self.max_memory_mb = max_memory_mb
def execute_python(self, code: str) -> Dict[str, str]:
"""在沙箱中执行Python代码"""
with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
# 添加安全限制
safe_code = f"""
import sys
import resource
# 限制内存使用
resource.setrlimit(resource.RLIMIT_AS, ({self.max_memory_mb * 1024 * 1024}, {self.max_memory_mb * 1024 * 1024}))
# 禁止危险操作
import builtins
_dangerous = ['eval', 'exec', 'compile', '__import__']
for name in _dangerous:
if hasattr(builtins, name):
delattr(builtins, name)
# 用户代码开始
{code}
"""
f.write(safe_code)
f.flush()
try:
result = subprocess.run(
['python3', f.name],
capture_output=True,
text=True,
timeout=self.timeout,
# 不传递环境变量，防止信息泄露
env={'PATH': '/usr/bin:/bin'}
)
return {
"stdout": result.stdout,
"stderr": result.stderr,
"returncode": str(result.returncode)
}
except subprocess.TimeoutExpired:
return {
"stdout": "",
"stderr": f"执行超时（超过{self.timeout}秒）",
"returncode": "-1"
}
finally:
os.unlink(f.name)
6.3 权限控制系统
from enum import Enum
from typing import Set
class Permission(Enum):
READ_FILE = "read_file"
WRITE_FILE = "write_file"
EXECUTE_CODE = "execute_code"
NETWORK_ACCESS = "network_access"
SEND_EMAIL = "send_email"
DELETE_DATA = "delete_data"
ADMIN_ACCESS = "admin_access"
class PermissionController:
"""权限控制器"""
def __init__(self):
self.default_permissions: Set[Permission] = {
Permission.READ_FILE,
Permission.NETWORK_ACCESS
}
self.tool_permissions: Dict[str, Set[Permission]] = {}
def register_tool_permissions(self, tool_name: str, permissions: Set[Permission]):
"""注册工具所需权限"""
self.tool_permissions[tool_name] = permissions
def check_permission(self, tool_name: str, user_permissions: Set[Permission]) -> bool:
"""检查是否有权限使用该工具"""
required = self.tool_permissions.get(tool_name, set())
return required.issubset(user_permissions | self.default_permissions)
def get_denial_reason(self, tool_name: str, user_permissions: Set[Permission]) -> str:
"""获取拒绝原因"""
required = self.tool_permissions.get(tool_name, set())
missing = required - user_permissions - self.default_permissions
return f"缺少权限: {', '.join(p.value for p in missing)}"
6.4 人类审核机制（Human-in-the-Loop）
对于高风险操作，需要人类确认：
class HumanApprovalGate:
"""人类审核关卡"""
def __init__(self):
self.pending_approvals: Dict[str, Dict] = {}
self.approved_actions: Set[str] = set()
self.rejected_actions: Set[str] = set()
def request_approval(self, action_id: str, action: Dict, risk_level: str = "medium") -> bool:
"""请求人类审核"""
if risk_level == "low":
return True # 低风险操作自动批准
self.pending_approvals[action_id] = {
"action": action,
"risk_level": risk_level,
"timestamp": datetime.now().isoformat()
}
print(f"\n⚠️ 需要人工审核:")
print(f" 操作: {action.get('description', '未知')}")
print(f" 风险等级: {risk_level}")
print(f" 详情: {json.dumps(action, ensure_ascii=False, indent=2)}")
# 在实际应用中，这里会暂停执行并等待人工审核
# 这里简化为交互式输入
response = input("批准此操作？(y/n): ").strip().lower()
if response == 'y':
self.approved_actions.add(action_id)
del self.pending_approvals[action_id]
return True
else:
self.rejected_actions.add(action_id)
del self.pending_approvals[action_id]
return False
6.5 输出内容过滤
class OutputFilter:
"""输出内容过滤器"""
def __init__(self):
self.sensitive_patterns = [
r'\b\d{16,19}\b', # 银行卡号
r'\b\d{3}-\d{2}-\d{4}\b', # SSN
r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', # 邮箱（可选）
]
self.blocked_topics = ["暴力", "色情", "违法"]
def filter(self, content: str) -> str:
"""过滤输出内容"""
import re
filtered = content
# 脱敏处理
for pattern in self.sensitive_patterns:
filtered = re.sub(pattern, '[已脱敏]', filtered)
# 检查是否包含不当内容
for topic in self.blocked_topics:
if topic in filtered:
return "抱歉，我无法提供此类内容。"
return filtered
第七章：实战项目——从零构建多工具自主Agent
7.1 项目概述
我们将构建一个功能完整的自主Agent，它具备以下能力：
- 搜索互联网获取实时信息
- 执行Python代码进行计算
- 读写本地文件
- 管理待办事项
- 多轮推理和自主决策
7.2 完整代码实现
"""
多工具自主Agent - 完整实现
功能：搜索、代码执行、文件操作、任务管理
"""
import openai
import json
import subprocess
import tempfile
import os
from datetime import datetime
from typing import List, Dict, Any, Optional, Callable
# ============ 工具实现 ============
class ToolBox:
"""工具箱：管理所有可用工具"""
def __init__(self):
self.tools: Dict[str, Dict] = {}
def register(self, name: str, description: str, parameters: dict, func: Callable,
risk_level: str = "low", requires_approval: bool = False):
"""注册工具"""
self.tools[name] = {
"function": func,
"schema": {
"type": "function",
"function": {
"name": name,
"description": description,
"parameters": parameters
}
},
"risk_level": risk_level,
"requires_approval": requires_approval
}
def get_schemas(self) -> list:
"""获取所有工具的schema"""
return [t["schema"] for t in self.tools.values()]
def execute(self, name: str, args: dict) -> str:
"""执行工具"""
if name not in self.tools:
return json.dumps({"error": f"未知工具: {name}"})
tool = self.tools[name]
# 高风险操作需要确认
if tool["requires_approval"]:
print(f"\n⚠️ 即将执行高风险操作: {name}")
print(f"参数: {json.dumps(args, ensure_ascii=False)}")
confirm = input("确认执行？(yes/no): ")
if confirm.lower() != "yes":
return json.dumps({"error": "用户取消操作"})
try:
result = tool["function"](**args)
return json.dumps({"result": result}, ensure_ascii=False)
except Exception as e:
return json.dumps({"error": str(e)})
# 具体工具实现
def search_web(query: str) -> str:
"""模拟网络搜索"""
# 实际实现中应调用搜索API
return f"搜索结果（模拟）：关于 '{query}' 的最新信息..."
def execute_code(code: str) -> str:
"""安全执行Python代码"""
with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
f.write(code)
f.flush()
try:
result = subprocess.run(
['python3', f.name],
capture_output=True,
text=True,
timeout=10
)
output = result.stdout
if result.stderr:
output += f"\n错误: {result.stderr}"
return output or "代码执行完成，无输出"
except subprocess.TimeoutExpired:
return "代码执行超时"
finally:
os.unlink(f.name)
def read_file(file_path: str) -> str:
"""读取文件内容"""
try:
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
return f"文件内容:\n{content[:2000]}" # 限制返回长度
except FileNotFoundError:
return f"文件不存在: {file_path}"
except Exception as e:
return f"读取错误: {str(e)}"
def write_file(file_path: str, content: str) -> str:
"""写入文件"""
try:
os.makedirs(os.path.dirname(file_path), exist_ok=True)
with open(file_path, 'w', encoding='utf-8') as f:
f.write(content)
return f"成功写入文件: {file_path}"
except Exception as e:
return f"写入错误: {str(e)}"
def get_current_time() -> str:
"""获取当前时间"""
return datetime.now().strftime("%Y年%m月%d日 %H:%M:%S")
def calculate(expression: str) -> str:
"""安全计算数学表达式"""
# 只允许基本数学运算
allowed_chars = set('0123456789+-*/.() ')
if not all(c in allowed_chars for c in expression):
return "错误：只允许基本数学运算"
try:
result = eval(expression)
return str(result)
except Exception as e:
return f"计算错误: {str(e)}"
# ============ 记忆系统 ============
class MemorySystem:
"""简化的记忆系统"""
def __init__(self):
self.short_term: List[Dict] = [] # 当前对话
self.long_term: List[Dict] = [] # 持久化记忆
self.max_short_term = 20
def add_message(self, role: str, content: str):
self.short_term.append({"role": role, "content": content})
if len(self.short_term) > self.max_short_term:
self.short_term = self.short_term[-self.max_short_term:]
def store_important(self, content: str):
self.long_term.append({
"content": content,
"timestamp": datetime.now().isoformat()
})
def get_context(self) -> List[Dict]:
return self.short_term.copy()
def get_relevant_memories(self, query: str) -> List[str]:
"""获取相关记忆（简单实现）"""
relevant = []
query_lower = query.lower()
for mem in self.long_term:
if any(word in mem["content"].lower() for word in query_lower.split()):
relevant.append(mem["content"])
return relevant[-5:] # 返回最近5条相关记忆
# ============ Agent核心 ============
class MultiToolAgent:
"""多工具自主Agent"""
def __init__(self, model: str = "gpt-4"):
self.client = openai.OpenAI()
self.model = model
self.toolbox = ToolBox()
self.memory = MemorySystem()
self.max_iterations = 15
self.total_cost = 0.0
# 注册工具
self._register_default_tools()
def _register_default_tools(self):
"""注册默认工具集"""
self.toolbox.register(
"search_web",
"搜索互联网获取最新信息",
{"type": "object", "properties": {"query": {"type": "string", "description": "搜索关键词"}}, "required": ["query"]},
search_web
)
self.toolbox.register(
"execute_code",
"执行Python代码进行计算或数据处理",
{"type": "object", "properties": {"code": {"type": "string", "description": "Python代码"}}, "required": ["code"]},
execute_code,
risk_level="medium"
)
self.toolbox.register(
"read_file",
"读取本地文件内容",
{"type": "object", "properties": {"file_path": {"type": "string", "description": "文件路径"}}, "required": ["file_path"]},
read_file
)
self.toolbox.register(
"write_file",
"写入内容到本地文件",
{"type": "object", "properties": {"file_path": {"type": "string", "description": "文件路径"}, "content": {"type": "string", "description": "写入内容"}}, "required": ["file_path", "content"]},
write_file,
risk_level="medium",
requires_approval=True
)
self.toolbox.register(
"get_current_time",
"获取当前日期和时间",
{"type": "object", "properties": {}},
get_current_time
)
self.toolbox.register(
"calculate",
"计算数学表达式",
{"type": "object", "properties": {"expression": {"type": "string", "description": "数学表达式"}}, "required": ["expression"]},
calculate
)
def _build_system_prompt(self) -> str:
"""构建系统提示词"""
return """你是一个智能助手，拥有多种工具能力。
你的工作流程：
1. 仔细分析用户的请求
2. 思考需要哪些步骤来完成任务
3. 选择合适的工具执行操作
4. 根据工具返回的结果继续推理
5. 当任务完成时，给出最终回答
重要原则：
- 每次只调用一个工具，等待结果后再决定下一步
- 如果工具返回错误，尝试理解原因并调整策略
- 如果任务过于复杂，将其分解为多个子步骤
- 始终用中文回复
- 在最终回答中，简要说明你的推理过程"""
def _estimate_cost(self, messages: list, response) -> float:
"""估算API调用成本"""
# 简化的成本估算
input_tokens = sum(len(m.get("content", "")) for m in messages) // 4
output_tokens = len(response.choices[0].message.content or "") // 4
cost = (input_tokens * 0.01 + output_tokens * 0.03) / 1000
return cost
def run(self, user_input: str) -> str:
"""运行Agent"""
print(f"\n{'='*60}")
print(f"用户: {user_input}")
print(f"{'='*60}")
# 添加用户消息到记忆
self.memory.add_message("user", user_input)
# 获取相关记忆
relevant_memories = self.memory.get_relevant_memories(user_input)
if relevant_memories:
memory_context = "\n相关记忆:\n" + "\n".join(f"- {m}" for m in relevant_memories)
else:
memory_context = ""
# 构建初始消息
system_message = self._build_system_prompt() + memory_context
for iteration in range(self.max_iterations):
print(f"\n--- 迭代 {iteration + 1} ---")
messages = [{"role": "system", "content": system_message}] + self.memory.get_context()
# 调用LLM
response = self.client.chat.completions.create(
model=self.model,
messages=messages,
tools=self.toolbox.get_schemas(),
)
message = response.choices[0].message
# 估算成本
cost = self._estimate_cost(messages, response)
self.total_cost += cost
# 如果没有工具调用，返回最终结果
if not message.tool_calls:
final_response = message.content
self.memory.add_message("assistant", final_response)
print(f"\n[Agent最终回复]\n{final_response}")
print(f"\n[本次会话成本: ${self.total_cost:.4f}]")
return final_response
# 执行工具调用
self.memory.add_message("assistant", message.model_dump())
for tool_call in message.tool_calls:
func_name = tool_call.function.name
func_args = json.loads(tool_call.function.arguments)
print(f" 🔧 调用工具: {func_name}")
print(f" 参数: {json.dumps(func_args, ensure_ascii=False)[:200]}")
result = self.toolbox.execute(func_name, func_args)
print(f" 结果: {result[:200]}...")
self.memory.add_message("tool", result)
return "达到最大迭代次数，任务未能完成。"
# ============ 使用示例 ============
if __name__ == "__main__":
agent = MultiToolAgent()
# 示例对话
print("多工具自主Agent已启动！输入 'quit' 退出。")
while True:
user_input = input("\n你: ").strip()
if user_input.lower() in ['quit', 'exit', 'q']:
break
if not user_input:
continue
response = agent.run(user_input)
7.3 运行效果示例
用户: 帮我计算一下，如果我每月存5000元，年利率3.5%，5年后能有多少钱？
--- 迭代 1 ---
🔧 调用工具: execute_code
参数: {"code": "monthly = 5000\nannual_rate = 0.035\nmonthly_rate = annual_rate / 12\nmonths = 5 * 12\ntotal = monthly * ((1 + monthly_rate)**months - 1) / monthly_rate\nprint(f'本金总额: {monthly * months}元')\nprint(f'利息总额: {total - monthly * months:.2f}元')\nprint(f'最终总额: {total:.2f}元')"}
结果: {"result": "本金总额: 300000元\n利息总额: 28056.29元\n最终总额: 328056.29元\n"}
[Agent最终回复]
根据计算结果：
- 每月存入：5,000元
- 存款期限：5年（60个月）
- 年利率：3.5%
- **本金总额：300,000元**
- **利息收入：28,056.29元**
- **5年后总额：328,056.29元**
这是一个复利计算，每月存入的资金都会从下个月开始产生利息，所以利息会逐月增加。
第八章：生产部署——监控、日志、成本控制
8.1 监控系统设计
生产环境中的Agent需要完善的监控体系：
import time
import logging
from dataclasses import dataclass, field
from typing import List
from collections import defaultdict
@dataclass
class AgentMetrics:
"""Agent运行指标"""
total_requests: int = 0
successful_requests: int = 0
failed_requests: int = 0
total_tokens_used: int = 0
total_cost: float = 0.0
avg_response_time: float = 0.0
tool_usage: dict = field(default_factory=lambda: defaultdict(int))
error_counts: dict = field(default_factory=lambda: defaultdict(int))
def record_request(self, success: bool, tokens: int, cost: float,
response_time: float, tools_used: List[str], error: str = None):
"""记录一次请求"""
self.total_requests += 1
if success:
self.successful_requests += 1
else:
self.failed_requests += 1
if error:
self.error_counts[error] += 1
self.total_tokens_used += tokens
self.total_cost += cost
# 更新平均响应时间
n = self.total_requests
self.avg_response_time = (self.avg_response_time * (n-1) + response_time) / n
for tool in tools_used:
self.tool_usage[tool] += 1
def get_report(self) -> str:
"""生成报告"""
return f"""
=== Agent运行报告 ===
总请求数: {self.total_requests}
成功率: {self.successful_requests/self.total_requests*100:.1f}%
总Token使用: {self.total_tokens_used}
总成本: ${self.total_cost:.4f}
平均响应时间: {self.avg_response_time:.2f}秒
工具使用统计: {dict(self.tool_usage)}
错误统计: {dict(self.error_counts)}
"""
8.2 结构化日志
import json
import logging
from datetime import datetime
class AgentLogger:
"""Agent结构化日志"""
def __init__(self, log_file: str = "agent.log"):
self.logger = logging.getLogger("agent")
self.logger.setLevel(logging.INFO)
handler = logging.FileHandler(log_file, encoding='utf-8')
handler.setFormatter(logging.Formatter('%(message)s'))
self.logger.addHandler(handler)
def log_interaction(self, session_id: str, user_input: str,
agent_response: str, tools_used: List[Dict],
tokens: int, cost: float, duration: float):
"""记录一次完整交互"""
log_entry = {
"timestamp": datetime.now().isoformat(),
"session_id": session_id,
"user_input": user_input[:500],
"agent_response": agent_response[:500],
"tools_used": tools_used,
"tokens": tokens,
"cost": cost,
"duration_seconds": duration,
"status": "success"
}
self.logger.info(json.dumps(log_entry, ensure_ascii=False))
def log_error(self, session_id: str, error: str, context: Dict):
"""记录错误"""
log_entry = {
"timestamp": datetime.now().isoformat(),
"session_id": session_id,
"error": error,
"context": context,
"status": "error"
}
self.logger.error(json.dumps(log_entry, ensure_ascii=False))
8.3 成本控制
class CostController:
"""成本控制器"""
def __init__(self, daily_budget: float = 10.0, monthly_budget: float = 200.0):
self.daily_budget = daily_budget
self.monthly_budget = monthly_budget
self.daily_cost: float = 0.0
self.monthly_cost: float = 0.0
self.last_reset_date = datetime.now().date()
def check_budget(self, estimated_cost: float) -> bool:
"""检查是否超出预算"""
self._reset_if_new_day()
if self.daily_cost + estimated_cost > self.daily_budget:
return False
if self.monthly_cost + estimated_cost > self.monthly_budget:
return False
return True
def record_cost(self, cost: float):
"""记录成本"""
self._reset_if_new_day()
self.daily_cost += cost
self.monthly_cost += cost
def _reset_if_new_day(self):
"""如果是新的一天，重置每日成本"""
today = datetime.now().date()
if today > self.last_reset_date:
self.daily_cost = 0.0
self.last_reset_date = today
def get_status(self) -> str:
return f"今日成本: ${self.daily_cost:.4f}/{self.daily_budget} | 本月成本: ${self.monthly_cost:.4f}/{self.monthly_budget}"
8.4 部署检查清单
在将Agent部署到生产环境前，确保完成以下检查：
安全性
- 所有工具调用都有适当的权限控制
- 高风险操作有人类审核机制
- 敏感信息已脱敏处理
- API密钥安全存储（使用环境变量）
可靠性
- 错误处理覆盖所有工具调用
- 有最大迭代次数限制
- 有请求超时机制
- 有重试和降级策略
可观测性
- 结构化日志记录所有交互
- 关键指标监控（成功率、延迟、成本）
- 告警机制（错误率过高、成本超支）
成本控制
- 设置每日/每月预算上限
- Token使用量监控
- 异常使用告警
第九章：最佳实践与进阶建议
9.1 提示工程最佳实践
- 系统提示词要清晰明确：定义Agent的角色、能力边界和行为准则
- 工具描述要精确：帮助模型做出正确的工具选择决策
- 提供示例：在系统提示词中给出典型的工具调用示例
- 限制输出格式：需要结构化输出时，明确定义格式要求
9.2 性能优化
- 并行工具调用：当多个工具调用之间没有依赖时，利用并行调用减少延迟
- 缓存机制：对于频繁查询的工具结果，实现缓存减少重复调用
- 上下文压缩：定期总结对话历史，避免上下文窗口溢出
- 模型选择：简单任务使用小模型，复杂推理使用大模型
9.3 测试策略
class AgentTester:
"""Agent测试框架"""
def __init__(self, agent):
self.agent = agent
self.test_cases = []
def add_test(self, name: str, input_text: str, expected_behavior: str,
validation_func: Callable = None):
"""添加测试用例"""
self.test_cases.append({
"name": name,
"input": input_text,
"expected": expected_behavior,
"validator": validation_func
})
def run_tests(self) -> Dict:
"""运行所有测试"""
results = {"passed": 0, "failed": 0, "errors": []}
for test in self.test_cases:
try:
response = self.agent.run(test["input"])
if test["validator"]:
passed = test["validator"](response)
else:
passed = True # 无验证函数时默认通过
if passed:
results["passed"] += 1
print(f"✅ {test['name']}")
else:
results["failed"] += 1
results["errors"].append({
"test": test["name"],
"expected": test["expected"],
"got": response[:200]
})
print(f"❌ {test['name']}")
except Exception as e:
results["failed"] += 1
results["errors"].append({
"test": test["name"],
"error": str(e)
})
print(f"💥 {test['name']}: {e}")
return results
9.4 未来展望
AI Agent领域正在快速发展，以下趋势值得关注：
- 多模态Agent：不仅能处理文本，还能理解图像、音频、视频
- 自主学习Agent：能够从经验中学习，持续改进自身能力
- Agent操作系统：Agent作为操作系统的核心组件，管理所有应用和服务
- 标准化协议：Agent之间的通信和协作协议标准化（如MCP）
- Agent安全框架：更完善的Agent安全评估和防护体系
总结
本教程从AI Agent的核心概念出发，系统性地讲解了Agent开发的各个方面：
- 核心概念：理解Agent与传统LLM应用的本质区别
- 推理范式：掌握ReAct、Plan-and-Execute、Reflection三大范式
- 架构设计：感知-推理-行动循环的完整实现
- 工具调用：Function Calling的原理和最佳实践
- 记忆系统：短期、长期、工作记忆的三层设计
- 多Agent协作：主从、对等、层级三种协作模式
- 安全防护：沙箱、权限、审核的多层保护
- 实战项目：完整的多工具Agent实现
- 生产部署：监控、日志、成本控制
AI Agent是一个快速发展的领域，建议持续关注最新的研究论文和开源项目，在实践中不断提升Agent的能力和可靠性。
本教程内容为原创撰写，如有疑问欢迎交流讨论。
