AI Agent 入门完全指南
·
本文档面向 Agent 初学者，系统讲解 AI Agent 的核心概念、架构组成、关键组件、主流框架、实战案例与应用场景。每个知识点附带代码示例、代码含义说明和实际应用场景。
目录
1. 什么是 AI Agent
1.1 定义
AI Agent（人工智能智能体） 是一个以大语言模型（LLM）为大脑、能够自主感知环境、进行推理规划、调用工具执行行动、并从反馈中学习的自主系统。
简而言之：Agent = LLM + 记忆 + 规划 + 工具使用。
1.2 Agent 与传统 AI 的区别
| 维度 | 传统 AI / 聊天机器人 | AI Agent |
|---|---|---|
| 自主性 | 被动响应，无自主决策 | 能自主拆解任务、决策行动 |
| 工具使用 | 无外部工具能力 | 可调用搜索、代码执行、API 等 |
| 记忆能力 | 仅依赖上下文窗口 | 具备短期 + 长期记忆 |
| 推理深度 | 单轮问答 | 多步推理、迭代反思 |
| 目标导向 | 无明确目标 | 围绕目标自主行动直到完成 |
1.3 一个直观的例子
用户需求：「帮我查一下明天北京的天气，如果下雨就帮我给团队发一封提醒邮件。」
-
聊天机器人：只能告诉你「我无法访问实时天气，也无法发邮件」。
-
AI Agent：
-
调用天气 API 查询明天北京天气 → 发现有小雨
-
规划：需要发邮件提醒
-
调用邮件发送工具 → 撰写邮件内容 → 发送给团队成员
-
返回：「已查询到明天北京有小雨，已向团队成员发送提醒邮件。」
-
这就是 Agent 的核心价值：将语言理解转化为实际行动。
2. Agent 核心架构
2.1 架构全景图
┌──────────────────────────────────────────────────────┐ │ AI Agent 系统 │ │ │ │ ┌──────────┐ ┌──────────┐ ┌──────────────────┐ │ │ │ 用户输入 │──▶│ LLM │──▶│ 行动执行(Action) │ │ │ │ (Input) │ │ (大脑) │ │ - 工具调用 │ │ │ └──────────┘ └────┬─────┘ │ - 代码执行 │ │ │ │ └────────┬─────────┘ │ │ ┌──────┴──────┐ │ │ │ │ Planning │ ▼ │ │ │ (规划) │ ┌──────────────────┐ │ │ │ - 任务拆解 │◀──│ Observation │ │ │ │ - 推理链 │ │ (观察/反馈) │ │ │ │ - 反思 │ └──────────────────┘ │ │ └─────────────┘ │ │ │ │ ┌─────────────────────────────────────────────────┐ │ │ │ Memory (记忆) │ │ │ │ ┌──────────────┐ ┌──────────────────┐ │ │ │ │ │ 短期记忆 │ │ 长期记忆 │ │ │ │ │ │ (对话上下文) │ │ (向量数据库/文件) │ │ │ │ │ └──────────────┘ └──────────────────┘ │ │ │ └─────────────────────────────────────────────────┘ │ │ │ │ ┌─────────────────────────────────────────────────┐ │ │ │ Tools (工具集) │ │ │ │ 搜索 │ 代码执行 │ 数据库 │ API调用 │ 文件操作 │ │ │ └─────────────────────────────────────────────────┘ │ └──────────────────────────────────────────────────────┘
2.2 四大核心组件
| 组件 | 作用 | 类比 |
|---|---|---|
| LLM | 理解、推理、决策 | 大脑 |
| Memory | 存储上下文和历史经验 | 记忆 |
| Planning | 任务拆解、推理、反思 | 思考方式 |
| Tools | 与外部世界交互 | 双手和工具 |
2.3 Agent 的工作循环
Agent 的核心是一个 循环（Loop）：
思考（Thought）→ 行动（Action）→ 观察（Observation）→ 思考 → ... → 完成
这个循环会持续进行，直到 Agent 认为任务完成或达到最大迭代次数。
# Agent 工作循环的伪代码
def agent_loop(user_input, max_iterations=10):
messages = [{"role": "user", "content": user_input}]
for i in range(max_iterations):
# 1. LLM 思考并决定下一步行动
response = llm.chat(messages)
# 2. 判断是否完成
if response.is_final_answer:
return response.content
# 3. 执行工具调用
tool_result = execute_tool(response.tool_call)
# 4. 将结果加入上下文（观察）
messages.append({"role": "assistant", "content": response.thought})
messages.append({"role": "tool", "content": tool_result})
return "达到最大迭代次数，任务未完成。"
代码含义说明：
-
agent_loop函数实现了 Agent 的核心循环 -
每轮迭代中，LLM 先思考（
llm.chat），再决定是否调用工具 -
如果调用工具，执行后把结果作为"观察"反馈给 LLM
-
循环直到 LLM 给出最终答案或达到迭代上限
应用场景： 这是最基础的 Agent 循环模式，几乎所有 Agent 框架的底层都基于此模式。
3. LLM —— Agent 的大脑
3.1 大语言模型简介
LLM（Large Language Model）是 Agent 的核心推理引擎。常见的 LLM 包括：
-
OpenAI: GPT-4o, GPT-4, GPT-3.5
-
Anthropic: Claude 3.5 Sonnet, Claude 3 Opus
-
开源模型: Llama 3, Qwen, Mistral, DeepSeek
3.2 基础调用
from openai import OpenAI
# 初始化客户端
client = OpenAI(api_key="your-api-key")
# 基础对话调用
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": "你是一个有帮助的AI助手。"},
{"role": "user", "content": "什么是AI Agent？"}
],
temperature=0.7 # 控制随机性，0=确定性，1=高随机性
)
print(response.choices[0].message.content)
代码含义说明：
-
system消息：设定 Agent 的角色和行为准则 -
user消息：用户的实际输入 -
temperature：控制输出的随机性，Agent 场景通常用较低值（0~0.3）保证推理稳定性
3.3 System Prompt 设计
System Prompt 是定义 Agent 行为的关键：
SYSTEM_PROMPT = """你是一个数据分析助手Agent。 你的能力： 1. 可以执行Python代码进行数据分析 2. 可以查询数据库获取数据 3. 可以生成图表可视化数据 你的工作原则： - 先理解用户需求，再制定分析计划 - 每一步都要说明你的思考过程 - 用中文回复，结果要清晰易懂 可用工具： - python_execute: 执行Python代码 - query_database: 查询数据库(SQL) - generate_chart: 生成图表 """
应用场景： System Prompt 广泛用于各种 Agent 场景，是定制 Agent 角色和行为的第一步。好的 System Prompt 能显著提升 Agent 的表现。
3.4 流式输出
# 流式输出，适合实时展示 Agent 的思考过程
stream = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": "解释一下Agent的ReAct模式"}],
stream=True # 开启流式输出
)
for chunk in stream:
content = chunk.choices[0].delta.content
if content:
print(content, end="", flush=True)
代码含义说明： stream=True 使 API 逐 token 返回内容，用户能实时看到 Agent 的输出，提升交互体验。
4. Memory —— Agent 的记忆系统
4.1 记忆类型概述
┌─────────────────────────────────────────┐ │ Agent 记忆系统 │ │ │ │ ┌─────────────┐ ┌──────────────────┐ │ │ │ 短期记忆 │ │ 长期记忆 │ │ │ │ (Working │ │ (Long-term │ │ │ │ Memory) │ │ Memory) │ │ │ │ │ │ │ │ │ │ · 对话上下文 │ │ · 向量数据库 │ │ │ │ · 当前任务 │ │ · 知识图谱 │ │ │ │ · 临时变量 │ │ · 日志文件 │ │ │ └─────────────┘ └──────────────────┘ │ └─────────────────────────────────────────┘
| 记忆类型 | 存储方式 | 生命周期 | 类比 |
|---|---|---|---|
| 短期记忆 | 对话消息列表 | 单次会话 | 工作记忆 |
| 长期记忆 | 向量数据库/文件 | 跨会话持久 | 长期记忆 |
4.2 短期记忆实现
class ShortTermMemory:
"""短期记忆：维护对话上下文"""
def __init__(self, system_prompt: str, max_messages: int = 20):
self.messages = [{"role": "system", "content": system_prompt}]
self.max_messages = max_messages
def add_user_message(self, content: str):
"""添加用户消息"""
self.messages.append({"role": "user", "content": content})
def add_assistant_message(self, content: str):
"""添加助手消息"""
self.messages.append({"role": "assistant", "content": content})
def add_tool_result(self, tool_name: str, result: str):
"""添加工具执行结果"""
self.messages.append({
"role": "tool",
"name": tool_name,
"content": result
})
def get_messages(self):
"""获取当前对话上下文"""
return self.messages
def trim(self):
"""裁剪消息，防止超出上下文窗口"""
if len(self.messages) > self.max_messages:
# 保留 system 消息 + 最近的 N 条消息
system = self.messages[0]
recent = self.messages[-(self.max_messages - 1):]
self.messages = [system] + recent
# 使用示例
memory = ShortTermMemory(
system_prompt="你是一个旅行规划Agent。",
max_messages=20
)
memory.add_user_message("帮我规划一个三天的北京之旅")
memory.add_assistant_message("好的！我来帮你规划北京三日游...")
memory.add_user_message("预算大概5000元")
print(memory.get_messages())
# 输出包含完整的对话上下文
代码含义说明：
-
ShortTermMemory类封装了对话消息管理 -
add_*方法分别添加不同角色的消息 -
trim()方法在消息过多时裁剪旧消息，防止超出 LLM 上下文窗口 -
这种"滑动窗口"策略是短期记忆管理的基本方法
应用场景： 适用于所有需要多轮对话的 Agent，如客服机器人、旅行规划助手、编程助手等。
4.3 长期记忆实现（基于向量数据库）
from chromadb import Client
from chromadb.config import Settings
class LongTermMemory:
"""长期记忆：基于向量数据库的语义检索"""
def __init__(self, collection_name: str = "agent_memory"):
self.client = Client(Settings())
self.collection = self.client.create_collection(
name=collection_name,
metadata={"description": "Agent长期记忆存储"}
)
def store(self, content: str, metadata: dict = None):
"""存储一段记忆"""
import uuid
memory_id = str(uuid.uuid4())
self.collection.add(
ids=[memory_id],
documents=[content],
metadatas=[metadata or {}]
)
return memory_id
def search(self, query: str, top_k: int = 3):
"""语义搜索相关记忆"""
results = self.collection.query(
query_texts=[query],
n_results=top_k
)
return results
def delete(self, memory_id: str):
"""删除指定记忆"""
self.collection.delete(ids=[memory_id])
# 使用示例
memory = LongTermMemory()
# 存储经验
memory.store(
"用户偏好Python语言，代码风格遵循PEP8",
metadata={"type": "user_preference", "timestamp": "2024-01-01"}
)
memory.store(
"用户的项目使用FastAPI框架开发REST API",
metadata={"type": "project_info", "timestamp": "2024-01-02"}
)
# 检索相关记忆
results = memory.search("用户用什么编程语言？", top_k=2)
for doc in results['documents'][0]:
print(f"记忆: {doc}")
# 输出: 记忆: 用户偏好Python语言，代码风格遵循PEP8
代码含义说明：
-
使用 ChromaDB（向量数据库）存储和检索记忆
-
store()：将文本转为向量嵌入并存储，附带元数据 -
search()：将查询转为向量，通过语义相似度找到最相关的记忆 -
与关键词搜索不同，向量搜索能理解语义（如"编程语言"能匹配到"Python"）
应用场景：
-
个性化助手：记住用户的偏好和历史
-
学习型 Agent：从过去的成功/失败经验中学习
-
知识管理：存储和检索领域知识
4.4 记忆管理策略
import json
from datetime import datetime
class MemoryManager:
"""综合记忆管理器：结合短期和长期记忆"""
def __init__(self, llm_client):
self.llm = llm_client
self.short_term = [] # 短期对话记忆
self.long_term = [] # 长期记忆摘要
def add_interaction(self, user_input: str, agent_response: str):
"""记录一次交互"""
self.short_term.append({
"timestamp": datetime.now().isoformat(),
"user": user_input,
"agent": agent_response
})
# 当短期记忆超过阈值时，触发摘要压缩
if len(self.short_term) >= 10:
self._summarize_and_archive()
def _summarize_and_archive(self):
"""将短期记忆摘要后存入长期记忆"""
recent_interactions = self.short_term[-10:]
summary_prompt = f"""请将以下对话交互总结为关键信息：
{json.dumps(recent_interactions, ensure_ascii=False, indent=2)}
提取要点：用户偏好、关键决策、重要事实。
"""
summary = self.llm.chat(summary_prompt)
self.long_term.append({
"timestamp": datetime.now().isoformat(),
"summary": summary
})
# 保留最近2条短期记忆，其余清除
self.short_term = self.short_term[-2:]
def get_context(self, current_query: str) -> str:
"""获取当前对话所需的上下文"""
context_parts = []
# 添加长期记忆摘要
if self.long_term:
context_parts.append("=== 历史摘要 ===")
for mem in self.long_term[-3:]: # 最近3条摘要
context_parts.append(mem["summary"])
# 添加近期对话
if self.short_term:
context_parts.append("\n=== 近期对话 ===")
for interaction in self.short_term[-5:]:
context_parts.append(f"用户: {interaction['user']}")
context_parts.append(f"助手: {interaction['agent']}")
return "\n".join(context_parts)
代码含义说明：
-
_summarize_and_archive()：当短期记忆过多时，用 LLM 生成摘要存入长期记忆，避免上下文爆炸 -
get_context()：为当前查询组装上下文，包含长期摘要 + 近期对话 -
这是实际 Agent 系统中常用的记忆管理策略——摘要压缩 + 滑动窗口
应用场景： 长期运行的助手（如个人助理、学习伙伴），需要跨会话保持上下文连贯。
5. Tools —— Agent 的工具调用
5.1 什么是工具调用
工具调用（Tool Calling / Function Calling）让 Agent 能够：
-
查询实时信息（搜索引擎、天气 API）
-
执行代码和计算
-
操作数据库
-
调用外部服务（发邮件、创建日历事件等）
5.2 定义工具
import json
from datetime import datetime
# ===== 工具函数定义 =====
def search_web(query: str) -> str:
"""搜索网络获取信息"""
# 实际项目中使用 Google Search API、Tavily 等
return f"搜索结果: 关于 '{query}' 的最新信息..."
def get_current_time() -> str:
"""获取当前时间"""
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
def calculate(expression: str) -> str:
"""安全地计算数学表达式"""
try:
# 仅允许数字和基本运算符
allowed = set("0123456789+-*/.() ")
if not all(c in allowed for c in expression):
return "错误: 表达式包含非法字符"
result = eval(expression)
return f"计算结果: {expression} = {result}"
except Exception as e:
return f"计算错误: {str(e)}"
def send_email(to: str, subject: str, body: str) -> str:
"""发送邮件"""
# 实际项目中集成 SMTP 或邮件 API
return f"邮件已发送至 {to}，主题: {subject}"
# ===== 工具描述（供 LLM 理解如何使用）=====
TOOLS_SCHEMA = [
{
"type": "function",
"function": {
"name": "search_web",
"description": "搜索互联网获取最新信息。当需要查找实时数据、新闻或未知信息时使用。",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词"
}
},
"required": ["query"]
}
}
},
{
"type": "function",
"function": {
"name": "get_current_time",
"description": "获取当前的日期和时间。当需要知道当前时间时使用。",
"parameters": {
"type": "object",
"properties": {}
}
}
},
{
"type": "function",
"function": {
"name": "calculate",
"description": "计算数学表达式。支持加减乘除和括号。",
"parameters": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "数学表达式，如 '2 + 3 * 4'"
}
},
"required": ["expression"]
}
}
},
{
"type": "function",
"function": {
"name": "send_email",
"description": "发送电子邮件给指定收件人。",
"parameters": {
"type": "object",
"properties": {
"to": {"type": "string", "description": "收件人邮箱"},
"subject": {"type": "string", "description": "邮件主题"},
"body": {"type": "string", "description": "邮件正文"}
},
"required": ["to", "subject", "body"]
}
}
}
]
# 工具名到函数的映射
TOOL_MAP = {
"search_web": search_web,
"get_current_time": get_current_time,
"calculate": calculate,
"send_email": send_email
}
代码含义说明：
-
工具函数：实际执行逻辑的 Python 函数
-
TOOLS_SCHEMA：JSON Schema 格式的工具描述，让 LLM 知道有哪些工具可用、参数是什么
-
description字段至关重要——LLM 根据它判断何时使用该工具 -
TOOL_MAP：将工具名映射到实际函数，便于调度执行
5.3 工具调用流程
from openai import OpenAI
client = OpenAI(api_key="your-api-key")
def call_agent_with_tools(user_input: str, messages: list):
"""带工具调用的 Agent 单轮交互"""
messages.append({"role": "user", "content": user_input})
# 第一步：LLM 决定是否调用工具
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=TOOLS_SCHEMA,
tool_choice="auto" # auto: LLM自主决定; "none": 不调用; 指定函数名: 强制调用
)
message = response.choices[0].message
messages.append(message)
# 第二步：如果 LLM 决定调用工具，执行工具
if message.tool_calls:
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}({func_args})")
# 执行工具
func = TOOL_MAP[func_name]
result = func(**func_args)
print(f"[工具结果] {result}")
# 将工具结果返回给 LLM
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": str(result)
})
# 第三步：LLM 根据工具结果生成最终回复
final_response = client.chat.completions.create(
model="gpt-4o",
messages=messages
)
return final_response.choices[0].message.content
return message.content
# 使用示例
messages = [{"role": "system", "content": "你是一个能干的助手，可以使用工具帮助用户。"}]
# 测试：计算 + 查时间
result = call_agent_with_tools("现在几点了？另外帮我算一下 (15 + 27) * 3", messages)
print(result)
# [工具调用] get_current_time({})
# [工具结果] 2024-01-15 14:30:22
# [工具调用] calculate({"expression": "(15 + 27) * 3"})
# [工具结果] 计算结果: (15 + 27) * 3 = 126
# 现在是 2024年1月15日 14:30。计算结果：(15 + 27) * 3 = 126。
代码含义说明：
-
tools=TOOLS_SCHEMA：将工具描述传给 LLM -
tool_choice="auto"：让 LLM 自主决定是否调用工具 -
当
message.tool_calls存在时，解析工具名和参数，执行后把结果以role: "tool"消息回传 -
最后再调一次 LLM，让它根据工具结果生成自然语言回复
应用场景：
-
智能客服：查询订单、退款
-
数据分析助手：执行 SQL、生成图表
-
办公助手：发邮件、管理日历
5.4 自定义工具最佳实践
# 使用装饰器简化工具定义（推荐）
from functools import wraps
class ToolRegistry:
"""工具注册器：统一管理工具定义和调度"""
def __init__(self):
self.tools = {}
self.schemas = []
def register(self, description: str, parameters: dict):
"""装饰器：注册一个工具"""
def decorator(func):
@wraps(func)
def wrapper(**kwargs):
try:
result = func(**kwargs)
return {"success": True, "result": result}
except Exception as e:
return {"success": False, "error": str(e)}
self.tools[func.__name__] = wrapper
self.schemas.append({
"type": "function",
"function": {
"name": func.__name__,
"description": description,
"parameters": {
"type": "object",
"properties": parameters,
"required": [
k for k, v in parameters.items()
if not v.get("optional", False)
]
}
}
})
return wrapper
return decorator
def execute(self, name: str, args: dict):
"""执行指定工具"""
if name not in self.tools:
return {"success": False, "error": f"未知工具: {name}"}
return self.tools[name](**args)
# 使用示例
registry = ToolRegistry()
@registry.register(
description="查询股票实时价格",
parameters={
"symbol": {"type": "string", "description": "股票代码，如 'AAPL'"}
}
)
def get_stock_price(symbol: str) -> str:
# 模拟 API 调用
prices = {"AAPL": "189.50", "GOOGL": "141.80", "TSLA": "248.42"}
price = prices.get(symbol, "未找到")
return f"{symbol} 当前价格: ${price}"
@registry.register(
description="创建待办事项",
parameters={
"title": {"type": "string", "description": "待办标题"},
"priority": {"type": "string", "description": "优先级: high/medium/low", "optional": True}
}
)
def create_todo(title: str, priority: str = "medium") -> str:
return f"已创建待办: [{priority}] {title}"
# 执行工具
print(registry.execute("get_stock_price", {"symbol": "AAPL"}))
# {'success': True, 'result': "AAPL 当前价格: $189.50"}
print(registry.execute("create_todo", {"title": "买牛奶", "priority": "high"}))
# {'success': True, 'result': '已创建待办: [high] 买牛奶'}
代码含义说明：
-
ToolRegistry类提供了一种更优雅的工具管理方式 -
使用装饰器
@registry.register(...)注册工具，自动生成 Schema -
execute()方法统一调度，内置错误处理 -
这种模式在团队协作中易于维护和扩展
应用场景： 适合需要管理大量工具的复杂 Agent 系统，如企业级智能助手。
6. Planning —— Agent 的规划能力
6.1 规划策略概览
┌─────────────────────────────────────────────────────┐ │ Agent 规划策略 │ │ │ │ ┌──────────┐ ┌────────────┐ ┌───────────────────┐ │ │ │ ReAct │ │ Plan-and │ │ Reflexion │ │ │ │ (推理+ │ │ -Execute │ │ (反思+改进) │ │ │ │ 行动) │ │ (先规划 │ │ │ │ │ │ │ │ 再执行) │ │ │ │ │ └──────────┘ └────────────┘ └───────────────────┘ │ │ │ │ ┌──────────────┐ ┌──────────────────────────────┐ │ │ │ Tree of │ │ Graph of Thoughts │ │ │ │ Thoughts │ │ (思维图) │ │ │ │ (思维树) │ │ │ │ │ └──────────────┘ └──────────────────────────────┘ │ └─────────────────────────────────────────────────────┘
6.2 任务拆解（Task Decomposition）
将复杂任务分解为可执行的子任务：
def task_decomposition(user_request: str, llm_client) -> list:
"""使用 LLM 将复杂任务拆解为子任务"""
decomposition_prompt = f"""你是一个任务规划专家。
用户请求: {user_request}
请将这个请求分解为具体的、可执行的子任务步骤。
每个步骤应该清晰、明确，并且可以独立执行。
以 JSON 数组格式返回，每个元素包含:
- step: 步骤编号
- task: 任务描述
- tool_needed: 可能需要的工具
"""
response = llm_client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": decomposition_prompt}],
response_format={"type": "json_object"}
)
import json
plan = json.loads(response.choices[0].message.content)
return plan["steps"]
# 示例输出
"""
用户请求: "帮我分析特斯拉2023年的财务数据并生成一份投资建议报告"
拆解结果:
[
{"step": 1, "task": "搜索特斯拉2023年年度财报数据", "tool_needed": "search_web"},
{"step": 2, "task": "提取关键财务指标（营收、利润、增长率）", "tool_needed": "python_execute"},
{"step": 3, "task": "查找行业对比数据", "tool_needed": "search_web"},
{"step": 4, "task": "分析财务趋势和风险因素", "tool_needed": "python_execute"},
{"step": 5, "task": "生成投资建议报告", "tool_needed": "generate_document"},
{"step": 6, "task": "将报告发送给用户邮箱", "tool_needed": "send_email"}
]
"""
代码含义说明：
-
通过 Prompt 让 LLM 将复杂任务拆解为有序的子任务
-
每个子任务标注所需工具，便于后续调度
-
输出 JSON 格式，方便程序解析和执行
应用场景： 复杂任务自动化，如研究报告生成、项目规划、数据分析流程。
6.3 Plan-and-Execute 模式
class PlanAndExecuteAgent:
"""先规划全局，再逐步执行的 Agent 模式"""
def __init__(self, llm_client, tool_registry):
self.llm = llm_client
self.tools = tool_registry
def plan(self, task: str) -> list:
"""第一步：生成执行计划"""
prompt = f"""为以下任务制定详细的分步执行计划。
任务: {task}
返回JSON格式:
{{"steps": ["步骤1", "步骤2", ...]}}
注意：每个步骤应该是一个具体的、可执行的操作。"""
response = self.llm.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"}
)
import json
plan = json.loads(response.choices[0].message.content)
return plan["steps"]
def execute_step(self, step: str, context: str) -> str:
"""第二步：执行单个步骤"""
prompt = f"""你在执行一个多步计划中的一步。
已完成步骤的上下文:
{context}
当前步骤: {step}
请执行这个步骤。如果需要使用工具，请说明。
直接给出执行结果。"""
response = self.llm.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.content
def replan(self, original_plan: list, completed: list, results: list, remaining: list) -> list:
"""根据执行结果重新规划剩余步骤"""
prompt = f"""你在执行一个计划，部分已完成。请根据结果决定是否需要调整剩余计划。
原始计划: {original_plan}
已完成: {completed}
执行结果: {results}
剩余计划: {remaining}
如果之前的步骤执行结果良好，剩余计划可以不变。
如果需要调整，请返回新的剩余步骤。
返回JSON: {{"need_replan": true/false, "new_steps": [...]}}"""
response = self.llm.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"}
)
import json
result = json.loads(response.choices[0].message.content)
if result["need_replan"]:
return result["new_steps"]
return remaining
def run(self, task: str):
"""完整执行流程"""
print(f"任务: {task}\n")
# 1. 规划
plan = self.plan(task)
print(f"计划: {plan}\n")
# 2. 逐步执行
results = []
remaining = plan[:]
while remaining:
step = remaining.pop(0)
context = "\n".join(results)
print(f"执行: {step}")
result = self.execute_step(step, context)
results.append(f"步骤[{step}]: {result}")
print(f"结果: {result}\n")
# 3. 检查是否需要重新规划
if remaining:
completed = plan[:len(plan) - len(remaining) - 1]
remaining = self.replan(plan, completed, results, remaining)
# 4. 汇总
return "\n".join(results)
代码含义说明：
-
plan()：先让 LLM 生成完整的执行计划 -
execute_step()：逐步执行计划，传入之前步骤的结果作为上下文 -
replan()：每完成一步后检查是否需要调整后续计划（适应性规划） -
这种模式适合复杂、多步骤、可能需要中途调整的任务
应用场景：
-
研究报告撰写（先列大纲，再逐章撰写）
-
软件开发（先设计架构，再逐步实现）
-
旅行规划（先定路线，再逐步预订）
6.4 Reflexion 反思模式
class ReflexionAgent:
"""带有反思能力的 Agent：执行后自我评估并改进"""
def __init__(self, llm_client, max_attempts=3):
self.llm = llm_client
self.max_attempts = max_attempts
self.reflections = [] # 积累的反思经验
def execute(self, task: str) -> str:
"""执行任务"""
attempt = 0
while attempt < self.max_attempts:
attempt += 1
print(f"\n=== 第 {attempt} 次尝试 ===")
# 1. 生成解决方案（利用之前的反思经验）
solution = self._generate_solution(task)
print(f"解决方案: {solution[:100]}...")
# 2. 评估解决方案
evaluation = self._evaluate(task, solution)
print(f"评估: {evaluation}")
if evaluation["is_sufficient"]:
return solution
# 3. 反思失败原因
reflection = self._reflect(task, solution, evaluation)
self.reflections.append(reflection)
print(f"反思: {reflection}")
return solution # 返回最后一次尝试
def _generate_solution(self, task: str) -> str:
"""生成解决方案，融入之前的反思"""
reflection_context = ""
if self.reflections:
reflection_context = f"""
之前尝试的反思经验:
{chr(10).join(f'- {r}' for r in self.reflections)}
请避免之前犯过的错误。"""
prompt = f"""解决以下任务:
{task}{reflection_context}"""
response = self.llm.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.content
def _evaluate(self, task: str, solution: str) -> dict:
"""评估解决方案质量"""
import json
prompt = f"""评估以下解决方案是否充分解决了任务。
任务: {task}
解决方案: {solution}
返回JSON: {{"is_sufficient": true/false, "score": 1-10, "issues": ["问题1", ...]}}"""
response = self.llm.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"}
)
return json.loads(response.choices[0].message.content)
def _reflect(self, task: str, solution: str, evaluation: dict) -> str:
"""反思失败原因，生成改进建议"""
prompt = f"""你在尝试解决一个任务但结果不够好。请反思为什么。
任务: {task}
你的解决方案: {solution}
评估结果: {evaluation}
简要说明：
1. 哪里做得不好
2. 下次应该怎么改进"""
response = self.llm.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.content
代码含义说明：
-
Reflexion 的核心是 执行 → 评估 → 反思 → 改进 的循环
-
_generate_solution()：生成解决方案时融入之前的反思经验 -
_evaluate()：让 LLM 自我评估解决方案的质量 -
_reflect()：分析失败原因，积累经验教训 -
self.reflections列表存储所有反思，实现"从错误中学习"
应用场景：
-
代码生成与调试：生成的代码有 bug → 反思 → 修正
-
数学推理：答案错误 → 分析原因 → 重新推导
-
写作优化：初稿质量不够 → 反思 → 改进
6.5 思维树（Tree of Thoughts）
class TreeOfThoughts:
"""思维树：探索多条推理路径，选择最优解"""
def __init__(self, llm_client, branching_factor=3, max_depth=4):
self.llm = llm_client
self.branching_factor = branching_factor # 每个节点生成几个分支
self.max_depth = max_depth # 最大搜索深度
def solve(self, problem: str) -> str:
"""解决问题：生成思维树，搜索最优路径"""
# 初始想法
initial_thoughts = self._generate_thoughts(problem, "", n=self.branching_factor)
best_path = self._search(problem, initial_thoughts, depth=0)
# 基于最优路径生成最终答案
return self._generate_answer(problem, best_path)
def _generate_thoughts(self, problem: str, current_thought: str, n: int) -> list:
"""生成 n 个可能的下一步思考"""
prompt = f"""问题: {problem}
当前思考过程: {current_thought if current_thought else "（开始）"}
请生成 {n} 个不同的下一步思考方向，每个方向都是解决问题的一种可能途径。
以JSON数组格式返回: ["思考1", "思考2", ...]"""
response = self.llm.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"}
)
import json
result = json.loads(response.choices[0].message.content)
return result.get("thoughts", result.get("data", []))
def _evaluate_thought(self, problem: str, thought: str) -> float:
"""评估一个思考方向的前景（0-1分）"""
prompt = f"""评估以下思考方向解决该问题的前景。
问题: {problem}
思考方向: {thought}
评分标准:
- 0.0-0.3: 方向错误，不太可能解决问题
- 0.4-0.6: 有一定道理，但不确定
- 0.7-1.0: 很有前景的方向
只返回一个数字（0到1之间的分数）。"""
response = self.llm.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}]
)
try:
return float(response.choices[0].message.content.strip())
except ValueError:
return 0.5
def _search(self, problem: str, thoughts: list, depth: int) -> list:
"""搜索最优路径（简化版 BFS）"""
if depth >= self.max_depth:
return thoughts[:1] # 返回最优的一个
best_path = []
best_score = -1
for thought in thoughts:
score = self._evaluate_thought(problem, thought)
if score > 0.7: # 高分方向继续探索
next_thoughts = self._generate_thoughts(problem, thought, n=self.branching_factor)
sub_path = self._search(problem, next_thoughts, depth + 1)
if sub_path:
path_score = self._evaluate_thought(problem, thought + " " + " ".join(sub_path))
if path_score > best_score:
best_score = path_score
best_path = [thought] + sub_path
return best_path if best_path else thoughts[:1]
def _generate_answer(self, problem: str, thought_path: list) -> str:
"""基于最优思考路径生成最终答案"""
prompt = f"""问题: {problem}
思考路径: {' -> '.join(thought_path)}
基于以上思考路径，给出最终答案。"""
response = self.llm.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.content
代码含义说明：
-
思维树模拟人类"发散-评估-深入"的思考方式
-
_generate_thoughts()：在每个节点生成多个可能的思考方向 -
_evaluate_thought()：评估每个方向的前景（0-1 分） -
_search()：通过评分搜索最优路径，高分方向继续深入，低分方向剪枝 -
适合需要深度推理的复杂问题
应用场景：
-
数学竞赛题求解
-
策略游戏规划
-
复杂决策问题（如投资策略）
7. ReAct 模式详解
7.1 什么是 ReAct
ReAct = Reasoning + Acting（推理 + 行动），是 Agent 最经典的模式。
循环: Thought → Action → Observation → Thought → Action → ... → Final Answer
Thought: 我需要先搜索相关信息
Action: search_web("Python 异步编程教程")
Observation: 搜索结果: Python asyncio 是异步编程的标准库...
Thought: 现在我有了足够的信息来回答
Action: finish("Python异步编程使用asyncio库...")
7.2 ReAct Prompt 模板
REACT_PROMPT_TEMPLATE = """你是一个能够使用工具的AI助手。
请严格按照以下格式回答问题:
Question: 用户的输入问题
Thought: 你对当前情况的思考和推理
Action: 你要使用的工具名称
Action Input: 工具的输入参数（JSON格式）
Observation: 工具返回的结果
... (Thought/Action/Action Input/Observation 可以重复多次)
Thought: 我现在知道最终答案了
Final Answer: 最终答案
可用工具:
{tools_description}
重要规则:
1. 每次只执行一个 Action
2. 必须先 Thought 再 Action
3. 根据 Observation 决定下一步
4. 当你有足够信息回答时，使用 Final Answer 结束
Question: {question}
"""
7.3 ReAct Agent 简洁实现
import json
import re
from openai import OpenAI
class ReActAgent:
"""ReAct 模式的简洁实现"""
def __init__(self, tools: dict, tool_descriptions: str, max_steps: int = 8):
self.client = OpenAI(api_key="your-api-key")
self.tools = tools # 工具名 → 函数映射
self.tool_descriptions = tool_descriptions # 工具描述文本
self.max_steps = max_steps
def run(self, question: str) -> str:
"""运行 ReAct 循环"""
# 初始化对话历史
history = f"Question: {question}\n"
for step in range(self.max_steps):
print(f"\n--- Step {step + 1} ---")
# 1. LLM 生成 Thought + Action
prompt = self._build_prompt(history)
response = self.client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
temperature=0
)
output = response.choices[0].message.content
print(f"LLM输出:\n{output}")
# 2. 检查是否给出最终答案
final_answer = self._parse_final_answer(output)
if final_answer:
return final_answer
# 3. 解析并执行 Action
action_info = self._parse_action(output)
if action_info:
tool_name = action_info["action"]
tool_input = action_info["action_input"]
if tool_name in self.tools:
observation = str(self.tools[tool_name](**tool_input))
else:
observation = f"错误: 未知工具 '{tool_name}'"
print(f"Observation: {observation}")
# 4. 将结果加入历史
history += output + f"\nObservation: {observation}\n"
else:
history += output + "\n"
return "达到最大步数，未能完成任务。"
def _build_prompt(self, history: str) -> str:
return f"""你是一个能使用工具的AI助手。严格按照ReAct格式回答。
可用工具:
{self.tool_descriptions}
格式:
Thought: 你的推理
Action: 工具名称
Action Input: {{"参数名": "参数值"}}
Observation: (工具结果会自动填入)
...
Thought: 我知道答案了
Final Answer: 最终答案
{history}"""
def _parse_action(self, text: str) -> dict:
"""从LLM输出中解析Action和Action Input"""
action_match = re.search(r'Action:\s*(\w+)', text)
input_match = re.search(r'Action Input:\s*(\{.*?\})', text, re.DOTALL)
if action_match and input_match:
return {
"action": action_match.group(1),
"action_input": json.loads(input_match.group(1))
}
return None
def _parse_final_answer(self, text: str) -> str:
"""检查是否有Final Answer"""
match = re.search(r'Final Answer:\s*(.+)', text, re.DOTALL)
return match.group(1).strip() if match else None
# ===== 使用示例 =====
# 定义工具
def search(query: str) -> str:
"""模拟搜索"""
database = {
"python": "Python是一种编程语言，由Guido van Rossum创建于1991年。",
"agent": "AI Agent是能自主感知、决策和行动的智能系统。",
}
for key, val in database.items():
if key in query.lower():
return val
return f"未找到关于'{query}'的信息。"
def calculate(expression: str) -> str:
"""计算表达式"""
try:
return str(eval(expression))
except:
return "计算错误"
# 工具描述
tool_descriptions = """
1. search(query: str): 搜索信息。参数: query - 搜索关键词
2. calculate(expression: str): 数学计算。参数: expression - 数学表达式
"""
# 创建并运行 Agent
agent = ReActAgent(
tools={"search": search, "calculate": calculate},
tool_descriptions=tool_descriptions
)
# 运行
result = agent.run("Python是什么时候创建的？距离现在多少年？")
print(f"\n最终答案: {result}")
# === 输出示例 ===
# --- Step 1 ---
# Thought: 我需要先搜索Python的创建时间
# Action: search
# Action Input: {"query": "python"}
# Observation: Python是一种编程语言，由Guido van Rossum创建于1991年。
#
# --- Step 2 ---
# Thought: Python创建于1991年，现在需要计算距离当前年份的年数
# Action: calculate
# Action Input: {"expression": "2024 - 1991"}
# Observation: 33
#
# --- Step 3 ---
# Thought: 我现在知道了答案
# Final Answer: Python由Guido van Rossum于1991年创建，距今已有33年。
代码含义说明：
-
run()：核心循环，每轮先让 LLM 输出 Thought + Action，再执行工具 -
_parse_action()：用正则表达式从 LLM 输出中提取工具名和参数 -
_parse_final_answer()：检测是否到达最终答案 -
这是 ReAct 的纯文本实现，实际项目中推荐使用 LLM 的原生 Function Calling
应用场景： ReAct 适合需要信息检索 + 推理的任务，如问答系统、研究助手。
8. 主流 Agent 框架
8.1 框架对比
| 框架 | 语言 | 特点 | 适合场景 |
|---|---|---|---|
| LangChain | Python/JS | 生态丰富，组件全面 | 通用 Agent 开发 |
| LangGraph | Python | 图结构编排，状态管理 | 复杂多步工作流 |
| CrewAI | Python | 多 Agent 协作，角色扮演 | 团队协作场景 |
| AutoGen | Python | 微软出品，多 Agent 对话 | 多 Agent 对话 |
| LlamaIndex | Python | RAG 能力强 | 知识密集型 Agent |
| AutoGPT | Python | 自主性强，全自动 | 自动化任务 |
8.2 LangChain 基础
from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain.tools import Tool
from langchain_core.prompts import ChatPromptTemplate
# 1. 初始化 LLM
llm = ChatOpenAI(model="gpt-4o", temperature=0)
# 2. 定义工具
def search_tool(query: str) -> str:
"""搜索工具"""
return f"搜索结果: 关于 '{query}' 的信息"
def calculator_tool(expression: str) -> str:
"""计算器工具"""
try:
return str(eval(expression))
except:
return "计算错误"
tools = [
Tool(
name="search",
func=search_tool,
description="搜索互联网获取信息"
),
Tool(
name="calculator",
func=calculator_tool,
description="执行数学计算"
)
]
# 3. 创建 Prompt
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个能干的AI助手，可以使用工具帮助用户。"),
("user", "{input}"),
("assistant", "{agent_scratchpad}"),
])
# 4. 创建 Agent
agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
# 5. 运行
result = agent_executor.invoke({"input": "计算 25 * 4 + 10 的结果"})
print(result["output"])
代码含义说明：
-
create_tool_calling_agent()：创建基于 LLM 原生 Function Calling 的 Agent -
AgentExecutor：Agent 执行器，管理 Thought-Action-Observation 循环 -
verbose=True：打印 Agent 的推理过程，便于调试 -
LangChain 封装了底层细节，开发更快
8.3 LangGraph 基础
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
import operator
# 1. 定义状态
class AgentState(TypedDict):
messages: Annotated[list, operator.add]
next_step: str
# 2. 初始化 LLM
llm = ChatOpenAI(model="gpt-4o", temperature=0)
# 3. 定义节点（Node）
def research_node(state: AgentState) -> AgentState:
"""研究节点：搜索信息"""
response = llm.invoke([
HumanMessage(content=f"研究以下问题: {state['messages'][-1].content}")
])
return {"messages": [response]}
def draft_node(state: AgentState) -> AgentState:
"""起草节点：基于研究结果撰写内容"""
response = llm.invoke([
HumanMessage(content=f"基于以下信息撰写文章:\n{state['messages'][-1].content}")
])
return {"messages": [response]}
def review_node(state: AgentState) -> AgentState:
"""审核节点：检查内容质量"""
response = llm.invoke([
HumanMessage(content=f"审核以下内容并给出修改建议:\n{state['messages'][-1].content}")
])
return {"messages": [response], "next_step": "end"}
# 4. 定义路由
def should_continue(state: AgentState) -> str:
"""决定下一步去哪个节点"""
return state.get("next_step", "draft")
# 5. 构建图
workflow = StateGraph(AgentState)
# 添加节点
workflow.add_node("research", research_node)
workflow.add_node("draft", draft_node)
workflow.add_node("review", review_node)
# 设置边（连接关系）
workflow.set_entry_point("research")
workflow.add_edge("research", "draft")
workflow.add_edge("draft", "review")
workflow.add_conditional_edges(
"review",
should_continue,
{
"end": END,
"draft": "draft" # 如果需要修改，回到起草节点
}
)
# 6. 编译并运行
app = workflow.compile()
result = app.invoke({
"messages": [HumanMessage(content="写一篇关于AI Agent的科普文章")],
"next_step": ""
})
print(result["messages"][-1].content)
代码含义说明：
-
StateGraph：以图结构定义 Agent 工作流，每个节点是一个处理步骤 -
AgentState：定义在节点间传递的状态数据结构 -
add_edge()：定义节点间的顺序连接 -
add_conditional_edges()：根据条件决定下一个节点（如审核不通过则回到起草） -
LangGraph 适合复杂、有分支和循环的工作流
应用场景：
-
内容创作流水线（研究 → 写作 → 审核 → 发布）
-
多阶段数据处理管道
-
有审批环节的业务流程
8.4 CrewAI 多 Agent 协作
from crewai import Agent, Task, Crew, Process
# 1. 创建 Agent（角色）
researcher = Agent(
role="研究员",
goal="收集和分析关于AI Agent的最新信息",
backstory="你是一位经验丰富的AI研究员，擅长搜集和整理信息。",
verbose=True,
llm="gpt-4o"
)
writer = Agent(
role="技术作者",
goal="将研究结果转化为通俗易懂的文章",
backstory="你是一位技术写作专家，善于将复杂概念用简单语言解释。",
verbose=True,
llm="gpt-4o"
)
editor = Agent(
role="编辑",
goal="审核文章质量并提出修改建议",
backstory="你是一位严格的编辑，追求文章的准确性和可读性。",
verbose=True,
llm="gpt-4o"
)
# 2. 定义任务
research_task = Task(
description="研究AI Agent的核心概念、架构和最新发展。",
agent=researcher,
expected_output="一份包含AI Agent关键概念和发展的研究报告"
)
write_task = Task(
description="基于研究报告，撰写一篇面向初学者的AI Agent科普文章，约1000字。",
agent=writer,
expected_output="一篇1000字的AI Agent科普文章",
context=[research_task] # 依赖研究任务的输出
)
edit_task = Task(
description="审核文章，检查准确性、可读性，并给出最终版本。",
agent=editor,
expected_output="审核后的最终文章版本",
context=[write_task]
)
# 3. 组建团队并执行
crew = Crew(
agents=[researcher, writer, editor],
tasks=[research_task, write_task, edit_task],
process=Process.sequential # 顺序执行
)
result = crew.kickoff()
print(result)
代码含义说明：
-
Agent：定义角色，包括角色、目标、背景故事 -
Task：定义任务，分配给特定 Agent，可指定依赖关系（context） -
Crew：组建团队，管理任务执行顺序 -
每个角色有独特的
backstory，LLM 会根据角色设定调整行为风格
应用场景：
-
内容生产团队（研究 → 写作 → 编辑）
-
软件开发团队（产品经理 → 开发 → 测试）
-
市场调研（数据收集 → 分析 → 报告）
9. ReAct Agent 完整实现
下面是一个完整的、可运行的 ReAct Agent，结合了前面讲的所有概念：
"""
完整的 ReAct Agent 实现
功能：搜索信息 + 计算 + 文件操作
"""
import json
import os
from openai import OpenAI
from datetime import datetime
# ============================================================
# 第一部分：工具定义
# ============================================================
def web_search(query: str, num_results: int = 3) -> str:
"""搜索网络信息（模拟实现）"""
mock_data = {
"Python": "Python是一种高级编程语言，由Guido van Rossum于1991年创建。",
"AI Agent": "AI Agent是能自主感知环境、做出决策并采取行动的智能系统。",
"LangChain": "LangChain是一个用于开发LLM驱动应用的开源框架。",
}
results = []
for key, value in mock_data.items():
if key.lower() in query.lower() or query.lower() in key.lower():
results.append(value)
if not results:
results.append(f"未找到关于'{query}'的相关信息。")
return "\n".join(results[:num_results])
def calculate(expression: str) -> str:
"""数学计算"""
allowed = set("0123456789+-*/.() ")
if not all(c in allowed for c in expression):
return "错误：表达式包含非法字符"
try:
result = eval(expression)
return f"{expression} = {result}"
except Exception as e:
return f"计算错误：{str(e)}"
def write_file(filename: str, content: str) -> str:
"""写入文件"""
try:
with open(filename, 'w', encoding='utf-8') as f:
f.write(content)
return f"文件 {filename} 写入成功，共 {len(content)} 字符。"
except Exception as e:
return f"写入失败：{str(e)}"
def read_file(filename: str) -> str:
"""读取文件"""
try:
with open(filename, 'r', encoding='utf-8') as f:
content = f.read()
return content[:500] # 限制返回长度
except FileNotFoundError:
return f"文件 {filename} 不存在。"
except Exception as e:
return f"读取失败：{str(e)}"
def get_time() -> str:
"""获取当前时间"""
return f"当前时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
# 工具 Schema（OpenAI Function Calling 格式）
TOOLS = [
{
"type": "function",
"function": {
"name": "web_search",
"description": "搜索互联网获取信息",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜索关键词"},
"num_results": {"type": "integer", "description": "返回结果数量，默认3"}
},
"required": ["query"]
}
}
},
{
"type": "function",
"function": {
"name": "calculate",
"description": "执行数学计算",
"parameters": {
"type": "object",
"properties": {
"expression": {"type": "string", "description": "数学表达式"}
},
"required": ["expression"]
}
}
},
{
"type": "function",
"function": {
"name": "write_file",
"description": "将内容写入文件",
"parameters": {
"type": "object",
"properties": {
"filename": {"type": "string", "description": "文件名"},
"content": {"type": "string", "description": "文件内容"}
},
"required": ["filename", "content"]
}
}
},
{
"type": "function",
"function": {
"name": "read_file",
"description": "读取文件内容",
"parameters": {
"type": "object",
"properties": {
"filename": {"type": "string", "description": "文件名"}
},
"required": ["filename"]
}
}
},
{
"type": "function",
"function": {
"name": "get_time",
"description": "获取当前日期和时间",
"parameters": {"type": "object", "properties": {}}
}
}
]
# 工具映射
TOOL_FUNCTIONS = {
"web_search": web_search,
"calculate": calculate,
"write_file": write_file,
"read_file": read_file,
"get_time": get_time
}
# ============================================================
# 第二部分：Agent 核心实现
# ============================================================
class Agent:
"""完整的 ReAct Agent 实现"""
def __init__(
self,
api_key: str,
model: str = "gpt-4o",
system_prompt: str = None,
max_iterations: int = 10,
verbose: bool = True
):
self.client = OpenAI(api_key=api_key)
self.model = model
self.max_iterations = max_iterations
self.verbose = verbose
# 默认 System Prompt
if system_prompt is None:
system_prompt = """你是一个智能助手Agent。
你的工作方式：
1. 理解用户需求
2. 思考解决步骤
3. 使用工具执行操作
4. 根据工具返回结果继续推理
5. 当任务完成时给出最终答案
注意事项：
- 如果不确定信息，使用搜索工具查询
- 数学计算使用计算工具，不要自己算
- 每一步都要清晰说明你的思考过程
"""
self.messages = [{"role": "system", "content": system_prompt}]
def log(self, msg: str):
"""日志输出"""
if self.verbose:
print(msg)
def run(self, user_input: str) -> str:
"""运行 Agent"""
self.messages.append({"role": "user", "content": user_input})
self.log(f"\n{'='*60}")
self.log(f"用户: {user_input}")
self.log(f"{'='*60}\n")
for i in range(self.max_iterations):
self.log(f"--- 迭代 {i+1}/{self.max_iterations} ---")
# LLM 决策
response = self.client.chat.completions.create(
model=self.model,
messages=self.messages,
tools=TOOLS,
tool_choice="auto",
temperature=0
)
message = response.choices[0].message
self.messages.append(message)
# 如果有工具调用
if message.tool_calls:
for tool_call in message.tool_calls:
name = tool_call.function.name
args = json.loads(tool_call.function.arguments)
self.log(f" Thought: 调用工具 {name}")
self.log(f" Action: {name}({json.dumps(args, ensure_ascii=False)})")
# 执行工具
if name in TOOL_FUNCTIONS:
result = TOOL_FUNCTIONS[name](**args)
else:
result = f"错误: 未知工具 '{name}'"
self.log(f" Observation: {result[:200]}")
# 将结果加入对话
self.messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": str(result)
})
else:
# 没有工具调用，说明 LLM 给出了最终答案
self.log(f"\n{'='*60}")
self.log(f"Final Answer: {message.content}")
self.log(f"{'='*60}")
return message.content
return "Agent 达到最大迭代次数，未能完成任务。"
# ============================================================
# 第三部分：运行示例
# ============================================================
if __name__ == "__main__":
agent = Agent(
api_key="your-api-key",
model="gpt-4o",
verbose=True
)
# 示例1：搜索 + 计算
result = agent.run("Python是哪一年创建的？从创建到现在经过了多少年？")
# 示例2：多步骤任务
result = agent.run("搜索AI Agent的概念，然后将搜索结果保存到agent_info.txt文件中")
# 示例3：组合任务
result = agent.run("现在几点了？请计算从现在到今年年底还有多少天")
运行输出示例：
============================================================
用户: Python是哪一年创建的？从创建到现在经过了多少年？
============================================================
--- 迭代 1/10 ---
Thought: 调用工具 web_search
Action: web_search({"query": "Python 编程语言 创建时间"})
Observation: Python是一种高级编程语言，由Guido van Rossum于1991年创建。
--- 迭代 2/10 ---
Thought: 调用工具 calculate
Action: calculate({"expression": "2024 - 1991"})
Observation: 2024 - 1991 = 33
--- 迭代 3/10 ---
============================================================
Final Answer: Python由Guido van Rossum于1991年创建，距今已有33年。
============================================================
10. RAG + Agent 检索增强
10.1 什么是 RAG
RAG（Retrieval-Augmented Generation，检索增强生成） = 信息检索 + LLM 生成。
Agent 使用 RAG 的流程：
用户提问 → 从知识库检索相关文档 → 将文档作为上下文喂给 LLM → 生成答案
10.2 RAG 基础实现
from openai import OpenAI
import chromadb
import uuid
class RAGAgent:
"""结合 RAG 的 Agent：先检索知识库，再生成答案"""
def __init__(self, api_key: str):
self.client = OpenAI(api_key=api_key)
# 初始化向量数据库
self.chroma = chromadb.Client()
self.collection = self.chroma.create_collection("knowledge_base")
def add_documents(self, documents: list):
"""添加文档到知识库"""
for doc in documents:
self.collection.add(
ids=[str(uuid.uuid4())],
documents=[doc["content"]],
metadatas=[{"source": doc.get("source", "unknown")}]
)
print(f"已添加 {len(documents)} 篇文档到知识库")
def retrieve(self, query: str, top_k: int = 3) -> list:
"""检索相关文档"""
results = self.collection.query(
query_texts=[query],
n_results=top_k
)
return results['documents'][0]
def generate(self, query: str, retrieved_docs: list) -> str:
"""基于检索结果生成答案"""
context = "\n\n".join(retrieved_docs)
prompt = f"""你是一个知识助手。请基于以下检索到的资料回答用户问题。
参考资料:
{context}
用户问题: {query}
要求:
1. 只基于参考资料回答，不要编造信息
2. 如果资料中没有答案，请明确说明
3. 回答要清晰、准确"""
response = self.client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
temperature=0
)
return response.choices[0].message.content
def ask(self, question: str) -> str:
"""完整 RAG 流程"""
# 1. 检索
docs = self.retrieve(question)
print(f"检索到 {len(docs)} 篇相关文档")
# 2. 生成
answer = self.generate(question, docs)
return answer
# ===== 使用示例 =====
rag = RAGAgent(api_key="your-api-key")
# 添加知识库文档
rag.add_documents([
{
"content": "LangChain是一个用于开发LLM应用的开源框架，提供了链式调用、Agent、记忆等核心组件。",
"source": "langchain_docs.md"
},
{
"content": "ReAct模式结合了推理(Reasoning)和行动(Acting)，让Agent能够边思考边执行工具调用。",
"source": "react_paper.md"
},
{
"content": "向量数据库如ChromaDB、Pinecone、Weaviate可以存储文本的嵌入向量，支持语义搜索。",
"source": "vector_db_intro.md"
},
{
"content": "Function Calling是OpenAI提供的功能，允许LLM输出结构化的函数调用请求。",
"source": "openai_docs.md"
},
])
# 提问
answer = rag.ask("什么是ReAct模式？")
print(answer)
# 基于检索到的文档回答，而非凭空生成
代码含义说明：
-
add_documents()：将文档存入向量数据库（ChromaDB 自动生成嵌入向量） -
retrieve()：将用户问题转为向量，搜索语义最相似的文档 -
generate()：将检索到的文档作为上下文，让 LLM 基于文档内容回答 -
ask()：完整的 RAG 流程——检索 + 生成
应用场景：
-
企业知识库问答
-
文档助手（如 API 文档查询）
-
客服系统（基于 FAQ 回答）
10.3 进阶：Agent + RAG
class SmartRAGAgent:
"""智能 RAG Agent：能判断何时需要检索，何时直接回答"""
def __init__(self, api_key: str):
self.client = OpenAI(api_key=api_key)
self.chroma = chromadb.Client()
self.collection = self.chroma.create_collection("smart_kb")
self.conversation_history = []
def add_knowledge(self, documents: list):
"""添加知识"""
for doc in documents:
self.collection.add(
ids=[str(uuid.uuid4())],
documents=[doc],
)
def chat(self, user_input: str) -> str:
"""智能对话"""
# Step 1: 判断是否需要检索
decision = self._decide_retrieval(user_input)
context = ""
if decision["need_retrieval"]:
# Step 2: 生成搜索查询
search_query = decision.get("search_query", user_input)
results = self.collection.query(query_texts=[search_query], n_results=3)
context = "\n\n".join(results['documents'][0])
print(f"[检索] 查询: {search_query}, 找到 {len(results['documents'][0])} 篇文档")
# Step 3: 生成回答
messages = [
{"role": "system", "content": "你是一个智能助手。根据提供的上下文回答问题。"}
]
if context:
messages.append({
"role": "system",
"content": f"相关知识:\n{context}"
})
# 添加对话历史
messages.extend(self.conversation_history[-6:]) # 最近3轮对话
messages.append({"role": "user", "content": user_input})
response = self.client.chat.completions.create(
model="gpt-4o",
messages=messages,
temperature=0
)
answer = response.choices[0].message.content
# 更新对话历史
self.conversation_history.append({"role": "user", "content": user_input})
self.conversation_history.append({"role": "assistant", "content": answer})
return answer
def _decide_retrieval(self, user_input: str) -> dict:
"""让 LLM 判断是否需要检索知识库"""
import json
prompt = f"""判断回答以下问题是否需要从知识库检索信息。
用户输入: {user_input}
判断规则:
- 如果问题涉及具体知识、事实、文档内容 → 需要检索
- 如果是闲聊、打招呼、简单计算 → 不需要检索
返回JSON: {{"need_retrieval": true/false, "search_query": "检索关键词(如果需要)"}}"""
response = self.client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"},
temperature=0
)
return json.loads(response.choices[0].message.content)
# 使用示例
agent = SmartRAGAgent(api_key="your-api-key")
agent.add_knowledge([
"公司请假政策：年假15天，病假10天，事假5天。需提前3天申请。",
"报销流程：填写报销单 → 部门经理审批 → 财务审核 → 打款。",
])
print(agent.chat("你好")) # 不检索，直接回答
print(agent.chat("公司年假有几天？")) # 检索知识库后回答
print(agent.chat("怎么报销？")) # 检索知识库后回答
代码含义说明：
-
_decide_retrieval()：让 LLM 先判断是否需要检索，避免不必要的检索开销 -
检索时 LLM 会生成更好的搜索查询（而非直接用用户原话）
-
维护对话历史，支持多轮对话中的指代消解（如"它"指代什么）
11. Multi-Agent 多智能体系统
11.1 概念
Multi-Agent 系统（MAS）是多个 Agent 协作完成任务的系统。每个 Agent 有不同的角色和能力，通过通信和协作解决单个 Agent 难以处理的复杂问题。
┌─────────────────────────────────────────────────┐ │ Multi-Agent 系统 │ │ │ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │ │ Agent A │◀───▶│ Agent B │◀───▶│ Agent C │ │ │ │ (研究) │ │ (写作) │ │ (审核) │ │ │ └─────────┘ └─────────┘ └─────────┘ │ │ │ │ │ │ │ └──────────────┼──────────────┘ │ │ │ │ │ ┌─────▼─────┐ │ │ │ 协调者 │ │ │ │ (Orchestrator)│ │ │ └───────────┘ │ └─────────────────────────────────────────────────┘
11.2 多 Agent 协作实现
"""
多 Agent 协作系统：产品开发场景
角色：产品经理 + 开发者 + 测试员
"""
import json
from openai import OpenAI
class MultiAgentSystem:
"""多 Agent 协作系统"""
def __init__(self, api_key: str):
self.client = OpenAI(api_key=api_key)
self.agents = {}
self.shared_memory = [] # 共享记忆空间
def create_agent(self, name: str, role: str, goal: str, tools: list = None):
"""创建一个 Agent 角色"""
self.agents[name] = {
"role": role,
"goal": goal,
"tools": tools or [],
"system_prompt": self._build_system_prompt(role, goal)
}
print(f"[创建Agent] {name} - {role}")
def _build_system_prompt(self, role: str, goal: str) -> str:
return f"""你是{role}。
你的目标: {goal}
工作原则:
1. 专注于你的职责范围
2. 参考其他Agent的工作成果（在共享记忆中）
3. 输出要清晰、结构化
4. 如果发现问题，明确指出"""
def communicate(self, from_agent: str, to_agent: str, message: str):
"""Agent 间通信"""
comm = {
"from": from_agent,
"to": to_agent,
"message": message,
"timestamp": len(self.shared_memory)
}
self.shared_memory.append(comm)
print(f"\n[{from_agent} → {to_agent}]: {message[:100]}...")
def run_agent(self, agent_name: str, task: str) -> str:
"""让指定 Agent 执行任务"""
agent = self.agents[agent_name]
# 构建上下文（包含共享记忆）
context = "\n".join([
f"[{c['from']} → {c['to']}]: {c['message']}"
for c in self.shared_memory[-10:] # 最近10条通信
])
messages = [
{"role": "system", "content": agent["system_prompt"]},
]
if context:
messages.append({
"role": "system",
"content": f"团队上下文:\n{context}"
})
messages.append({"role": "user", "content": task})
response = self.client.chat.completions.create(
model="gpt-4o",
messages=messages,
temperature=0.3
)
result = response.choices[0].message.content
print(f"\n[{agent_name} 的输出]:\n{result[:200]}...")
return result
def collaborative_task(self, task: str):
"""多 Agent 协作完成任务"""
print(f"\n{'='*60}")
print(f"协作任务: {task}")
print(f"{'='*60}")
# 1. 产品经理：需求分析
pm_output = self.run_agent("product_manager", f"分析以下需求并制定产品方案:\n{task}")
self.communicate("product_manager", "developer", pm_output)
# 2. 开发者：技术实现
dev_output = self.run_agent("developer", f"根据产品经理的方案，设计技术实现方案:\n{pm_output}")
self.communicate("developer", "tester", dev_output)
# 3. 测试员：测试方案
test_output = self.run_agent("tester", f"根据技术方案，制定测试计划:\n{dev_output}")
self.communicate("tester", "product_manager", test_output)
# 4. 产品经理：最终审核
final_output = self.run_agent(
"product_manager",
f"审核团队方案并给出最终总结:\n方案: {pm_output}\n技术: {dev_output}\n测试: {test_output}"
)
return final_output
# ===== 使用示例 =====
system = MultiAgentSystem(api_key="your-api-key")
# 创建三个角色
system.create_agent(
name="product_manager",
role="产品经理",
goal="分析用户需求，制定产品方案和功能规格"
)
system.create_agent(
name="developer",
role="高级开发工程师",
goal="根据产品方案设计技术架构和实现方案"
)
system.create_agent(
name="tester",
role="测试工程师",
goal="设计全面的测试方案，确保产品质量"
)
# 协作完成任务
result = system.collaborative_task("开发一个AI Agent在线编程助手，能帮用户写代码、调试和解释代码")
代码含义说明：
-
create_agent()：创建不同角色的 Agent，每个角色有专属的 System Prompt -
communicate()：Agent 之间通过共享记忆通信 -
run_agent()：让特定 Agent 执行任务，自动注入团队上下文 -
collaborative_task()：编排多 Agent 协作流程——产品经理 → 开发 → 测试 → 审核 -
共享记忆
shared_memory让所有 Agent 能看到之前的交流内容
应用场景：
-
软件开发（需求 → 设计 → 编码 → 测试）
-
内容生产（选题 → 写作 → 编辑 → 发布）
-
投资决策（数据收集 → 分析 → 风险评估 → 决策）
11.3 Agent 间通信模式
"""
Agent 通信模式示例
"""
# 模式1：顺序传递（Pipeline）
def sequential_pipeline(agents: list, initial_input: str):
"""每个 Agent 处理后传给下一个"""
current = initial_input
for agent in agents:
current = agent.process(current)
return current
# 模式2：广播-汇总（Fan-out/Fan-in）
def fan_out_fan_in(agents: list, task: str, aggregator):
"""多个 Agent 并行处理同一任务，最后汇总"""
results = []
for agent in agents:
results.append(agent.process(task)) # 并行处理
return aggregator.combine(results) # 汇总结果
# 模式3：辩论（Debate）
def debate_mode(agents: list, topic: str, rounds: int = 3):
"""多 Agent 辩论，达成共识"""
positions = {agent.name: agent.initial_position(topic) for agent in agents}
for round_num in range(rounds):
for agent in agents:
# 每个Agent看到其他人的观点后更新自己的立场
others = {k: v for k, v in positions.items() if k != agent.name}
positions[agent.name] = agent.revise(topic, others)
# 最终共识
return aggregator.find_consensus(positions)
# 模式4：层级管理（Hierarchical）
def hierarchical_mode(manager_agent, worker_agents: list, task: str):
"""管理者分配任务给工人Agent，汇总结果"""
# 管理者拆解任务
subtasks = manager_agent.decompose(task)
# 分配给工人
results = {}
for subtask, worker in zip(subtasks, worker_agents):
results[subtask] = worker.execute(subtask)
# 管理者汇总
return manager_agent.synthesize(results)
代码含义说明：
-
顺序传递：流水线模式，每个 Agent 处理特定环节
-
广播-汇总：多个 Agent 并行处理同一问题，取长补短
-
辩论：Agent 互相挑战观点，逐步达成共识（适合决策类任务）
-
层级管理：一个管理者 Agent 拆解任务并分配，类似团队管理
12. Agent 应用场景与案例
12.1 应用场景总览
| 场景 | 描述 | 核心 Agent 能力 |
|---|---|---|
| 智能客服 | 自动回答用户问题、处理工单 | RAG + 工具调用 |
| 编程助手 | 代码生成、调试、解释 | 代码执行 + 文件操作 |
| 数据分析 | 数据查询、分析、可视化 | 代码执行 + 数据库查询 |
| 研究助手 | 文献检索、信息汇总、报告生成 | 搜索 + 规划 + 写作 |
| 个人助理 | 日程管理、邮件处理、提醒 | 多工具协调 |
| 自动化运维 | 监控告警、故障诊断、自动修复 | 系统操作 + 推理 |
| 内容创作 | 选题、写作、编辑、发布 | 多 Agent 协作 |
12.2 案例：智能数据分析 Agent
"""
智能数据分析 Agent
功能：接收自然语言查询 → 生成 SQL → 执行 → 分析 → 可视化
"""
import json
from openai import OpenAI
class DataAnalysisAgent:
def __init__(self, api_key: str):
self.client = OpenAI(api_key=api_key)
self.system_prompt = """你是一个数据分析Agent。
数据库表结构:
- sales(id, product, amount, date, region)
- customers(id, name, email, signup_date, region)
- products(id, name, category, price)
可用工具:
1. execute_sql: 执行SQL查询
2. analyze_data: 分析数据趋势
3. generate_chart: 生成图表
工作流程:
1. 理解用户的数据分析需求
2. 生成SQL查询获取数据
3. 分析数据
4. 生成可视化图表
5. 给出分析结论"""
def analyze(self, user_query: str) -> str:
messages = [
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": user_query}
]
# 工具定义
tools = [
{
"type": "function",
"function": {
"name": "execute_sql",
"description": "执行SQL查询语句",
"parameters": {
"type": "object",
"properties": {
"sql": {"type": "string", "description": "SQL查询语句"}
},
"required": ["sql"]
}
}
},
{
"type": "function",
"function": {
"name": "generate_chart",
"description": "生成数据图表",
"parameters": {
"type": "object",
"properties": {
"chart_type": {"type": "string", "description": "图表类型: bar/line/pie"},
"title": {"type": "string", "description": "图表标题"},
"data": {"type": "string", "description": "图表数据(JSON格式)"}
},
"required": ["chart_type", "title", "data"]
}
}
}
]
# 工具执行函数
def execute_sql(sql: str) -> str:
# 模拟数据库查询
mock_results = {
"SELECT region, SUM(amount) as total FROM sales GROUP BY region":
"[{'region': '华东', 'total': 150000}, {'region': '华北', 'total': 120000}, {'region': '华南', 'total': 180000}]",
}
return mock_results.get(sql, f"SQL执行结果: 查询返回5行数据...")
def generate_chart(chart_type: str, title: str, data: str) -> str:
return f"图表已生成: {title} ({chart_type}图)"
tool_map = {"execute_sql": execute_sql, "generate_chart": generate_chart}
# Agent 循环
for _ in range(5):
response = self.client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=tools,
temperature=0
)
msg = response.choices[0].message
messages.append(msg)
if not msg.tool_calls:
return msg.content
for tc in msg.tool_calls:
name = tc.function.name
args = json.loads(tc.function.arguments)
result = tool_map[name](**args)
messages.append({
"role": "tool",
"tool_call_id": tc.id,
"content": str(result)
})
return "分析完成"
# 使用
agent = DataAnalysisAgent(api_key="your-api-key")
result = agent.analyze("分析各区域的销售总额，并生成柱状图")
print(result)
# 输出: 各区域销售总额分析：华南(18万) > 华东(15万) > 华北(12万)...
# 已生成柱状图展示各区域对比...
12.3 案例：自动化邮件处理 Agent
"""
自动化邮件处理 Agent
功能：读取邮件 → 分类 → 根据类型自动处理（回复/转发/创建任务）
"""
class EmailAgent:
"""邮件处理 Agent"""
def __init__(self, api_key: str):
self.client = OpenAI(api_key=api_key)
def classify_email(self, email: dict) -> str:
"""分类邮件"""
prompt = f"""将以下邮件分类到一个类别中。
发件人: {email['sender']}
主题: {email['subject']}
内容: {email['body'][:500]}
类别:
- urgent: 紧急事项，需要立即处理
- inquiry: 询问类，需要回复
- notification: 通知类，仅供参考
- spam: 垃圾邮件
只返回类别名称。"""
response = self.client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
temperature=0
)
return response.choices[0].message.content.strip()
def draft_reply(self, email: dict) -> str:
"""起草回复"""
prompt = f"""为以下邮件起草一个专业的回复。
发件人: {email['sender']}
主题: {email['subject']}
内容: {email['body']}
要求:
- 语气专业、礼貌
- 直接回应邮件中的问题
- 不超过200字"""
response = self.client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.content
def process_emails(self, emails: list) -> list:
"""批量处理邮件"""
results = []
for email in emails:
category = self.classify_email(email)
action = {"email": email, "category": category}
if category == "urgent":
# 紧急邮件：创建任务 + 起草回复
action["action"] = "create_task"
action["reply"] = self.draft_reply(email)
elif category == "inquiry":
# 询问邮件：起草回复
action["action"] = "reply"
action["reply"] = self.draft_reply(email)
elif category == "notification":
# 通知邮件：标记已读
action["action"] = "mark_read"
else:
# 垃圾邮件：删除
action["action"] = "delete"
results.append(action)
return results
# 使用
agent = EmailAgent(api_key="your-api-key")
emails = [
{
"sender": "client@example.com",
"subject": "紧急：项目交付延迟",
"body": "我们的项目已经延迟一周了，请立即处理..."
},
{
"sender": "hr@company.com",
"subject": "公司年会通知",
"body": "公司年会将于下周五举行，请查收附件..."
}
]
results = agent.process_emails(emails)
for r in results:
print(f"邮件: {r['email']['subject']}")
print(f"分类: {r['category']}, 动作: {r['action']}")
if 'reply' in r:
print(f"回复草稿: {r['reply'][:100]}...")
print()
13. Agent 的挑战与未来
13.1 当前面临的挑战
| 挑战 | 描述 | 当前解决方案 |
|---|---|---|
| 可靠性 | LLM 可能产生幻觉或错误推理 | 多步验证、Reflexion 反思 |
| 成本 | 多轮 LLM 调用成本高 | 缓存、模型路由（简单任务用小模型） |
| 延迟 | 多步推理导致响应慢 | 并行化、流式输出 |
| 安全性 | 工具调用可能执行危险操作 | 权限控制、人工确认机制 |
| 上下文限制 | 上下文窗口有限 | 记忆压缩、RAG 检索 |
| 评估困难 | Agent 行为非确定性，难以测试 | 回归测试集、人工评估 |
13.2 安全性最佳实践
"""
Agent 安全机制示例
"""
class SafeAgent:
"""带安全机制的 Agent"""
# 危险操作白名单
SAFE_ACTIONS = {
"search_web", "get_time", "calculate", "read_file"
}
# 需要确认的操作
CONFIRM_REQUIRED = {
"send_email", "write_file", "delete_file", "execute_code"
}
# 禁止的操作
BLOCKED_ACTIONS = {
"execute_shell", "access_credentials", "modify_system"
}
def execute_action(self, action_name: str, params: dict) -> dict:
"""安全执行操作"""
# 1. 检查是否被禁止
if action_name in self.BLOCKED_ACTIONS:
return {
"success": False,
"error": f"操作 '{action_name}' 被安全策略禁止"
}
# 2. 检查是否需要人工确认
if action_name in self.CONFIRM_REQUIRED:
confirmed = self._ask_user_confirmation(action_name, params)
if not confirmed:
return {
"success": False,
"error": "用户取消了操作"
}
# 3. 参数验证
validated_params = self._validate_params(action_name, params)
# 4. 执行
try:
result = self._execute(action_name, validated_params)
return {"success": True, "result": result}
except Exception as e:
return {"success": False, "error": str(e)}
def _ask_user_confirmation(self, action: str, params: dict) -> bool:
"""请求用户确认"""
print(f"\n⚠️ 需要确认的操作:")
print(f" 操作: {action}")
print(f" 参数: {json.dumps(params, ensure_ascii=False)}")
# 实际项目中弹出确认对话框
return input(" 确认执行？(y/n): ").lower() == 'y'
def _validate_params(self, action: str, params: dict) -> dict:
"""参数验证和清洗"""
# 防止路径遍历攻击
if "filename" in params:
params["filename"] = params["filename"].replace("..", "").replace("/", "")
# 防止 SQL 注入
if "sql" in params:
dangerous_keywords = ["DROP", "DELETE", "TRUNCATE", "ALTER"]
sql_upper = params["sql"].upper()
for kw in dangerous_keywords:
if kw in sql_upper:
raise ValueError(f"SQL包含危险关键词: {kw}")
return params
代码含义说明：
-
三级安全策略：安全操作直接执行 → 需确认操作人工审批 → 危险操作直接禁止
-
_validate_params()：对参数进行安全清洗，防止路径遍历、SQL 注入等攻击 -
这是生产环境 Agent 系统必备的安全层
13.3 Agent 评估指标
"""
Agent 评估框架
"""
class AgentEvaluator:
"""Agent 性能评估"""
def __init__(self):
self.metrics = {
"task_success_rate": 0, # 任务成功率
"avg_steps": 0, # 平均步数
"avg_cost": 0, # 平均成本
"avg_latency": 0, # 平均延迟
"tool_accuracy": 0, # 工具调用准确率
}
self.test_cases = []
def add_test_case(self, input: str, expected_output: str, tools_expected: list = None):
"""添加测试用例"""
self.test_cases.append({
"input": input,
"expected": expected_output,
"tools_expected": tools_expected or []
})
def evaluate(self, agent) -> dict:
"""评估 Agent 表现"""
results = []
for case in self.test_cases:
# 运行 Agent
actual_output = agent.run(case["input"])
steps_taken = len(agent.messages) # 简化的步数统计
# 评估
success = self._check_success(actual_output, case["expected"])
results.append({
"input": case["input"],
"success": success,
"steps": steps_taken
})
# 计算指标
total = len(results)
self.metrics["task_success_rate"] = sum(r["success"] for r in results) / total
self.metrics["avg_steps"] = sum(r["steps"] for r in results) / total
return self.metrics
def _check_success(self, actual: str, expected: str) -> bool:
"""检查输出是否正确（简化版）"""
# 实际项目中可以用 LLM 做语义评估
return expected.lower() in actual.lower()
13.4 未来趋势
-
多模态 Agent：不仅能处理文本，还能理解和生成图像、视频、音频
-
Agent 操作系统：类似操作系统，统一管理多个 Agent 的调度、资源分配
-
自主进化：Agent 能从经验中持续学习和自我改进
-
Agent 互联网：Agent 之间能互相发现、协作、交易
-
具身智能：Agent 与物理世界交互（机器人）
-
更低成本：模型优化和硬件进步使 Agent 更普及
附录：术语表
| 术语 | 英文 | 解释 |
|---|---|---|
| Agent | Agent / 智能体 | 能自主感知、决策、行动的 AI 系统 |
| LLM | Large Language Model | 大语言模型，Agent 的推理引擎 |
| RAG | Retrieval-Augmented Generation | 检索增强生成，先检索再生成 |
| ReAct | Reasoning + Acting | 推理与行动结合的 Agent 模式 |
| Tool Calling | Function Calling | LLM 调用外部工具的能力 |
| Vector DB | Vector Database | 向量数据库，支持语义搜索 |
| Embedding | Embedding | 文本的向量表示 |
| Chain of Thought | CoT | 思维链，逐步推理 |
| Tree of Thoughts | ToT | 思维树，多路径探索 |
| Multi-Agent | Multi-Agent System | 多智能体系统 |
| Hallucination | Hallucination | 幻觉，LLM 生成不真实的内容 |
| Context Window | Context Window | 上下文窗口，LLM 单次能处理的最大文本 |
| Prompt | Prompt | 提示词，给 LLM 的输入指令 |
| Fine-tuning | Fine-tuning | 微调，在预训练模型上进一步训练 |
| Zero-shot | Zero-shot | 零样本，不给示例直接让 LLM 完成 |
| Few-shot | Few-shot | 少样本，给少量示例引导 LLM |
文档说明：本文档涵盖了 AI Agent 从基础概念到实战应用的完整知识体系。建议按照目录顺序学习，每读完一个章节后动手运行对应的代码示例，以加深理解。所有代码示例基于 Python，使用 OpenAI API 作为 LLM 接口，可根据实际需求替换为其他 LLM 或框架。
更多推荐
1
0- 0
已为社区贡献1条内容
所有评论(0)