AI Agent 智能体开发入门教程
面向零基础开发者的完整指南 最后更新：2026年5月 预计阅读时间：2-3小时
目录
- 什么是AI Agent
- Agent架构详解
- 主流Agent框架对比
- LangChain Agent开发实战
- 工具系统深入
- 记忆与上下文管理
- Agent规划能力
- 多Agent协作
- 生产部署
- 实战项目：构建智能客服Agent系统
第一章：什么是AI Agent
1.1 AI Agent的概念定义
AI Agent（人工智能智能体）是一种能够自主感知环境、进行推理决策、并采取行动来完成特定目标的软件系统。与传统的程序不同，AI Agent具备自主性和适应性，它可以根据环境的变化动态调整自己的行为策略。
简单来说，如果把传统的程序比作一条固定路线的火车，那么AI Agent就像一个会看地图、会问路、遇到堵车会自己绕道的出租车司机。它不仅仅执行预设的指令，更能在复杂多变的环境中自主做出合理的决策。
从技术角度定义，一个AI Agent包含以下核心要素：
- 感知能力（Perception）：能够接收和理解来自环境的信息，包括用户输入、API返回数据、传感器数据等
- 推理能力（Reasoning）：能够基于感知到的信息进行逻辑推理、分析和判断
- 行动能力（Action）：能够执行具体的操作来影响环境，如调用API、生成文本、控制设备等
- 目标导向（Goal-Oriented）：所有的感知、推理和行动都是围绕着完成特定目标展开的
- 自主性（Autonomy）：能够在无人干预的情况下自主决策和执行
1.2 AI Agent与传统Chatbot的区别
很多人会把AI Agent和传统的Chatbot（聊天机器人）混淆，但两者有着本质的区别。下面我们通过一个对比表格来理解：
| 特征 | 传统Chatbot | AI Agent |
|---|---|---|
| 交互模式 | 一问一答，被动响应 | 主动规划，自主执行 |
| 能力范围 | 仅限文本对话 | 可调用工具、访问网络、操作文件 |
| 任务复杂度 | 单轮简单问答 | 多步骤复杂任务 |
| 决策方式 | 基于规则或模板匹配 | 基于LLM推理的动态决策 |
| 上下文理解 | 有限的上下文窗口 | 配备记忆系统的长期上下文 |
| 错误处理 | 无法自我纠正 | 可以反思和修正错误 |
| 可扩展性 | 添加功能需要改代码 | 通过添加工具即可扩展能力 |
让我们通过一个具体的例子来理解这个区别：
场景：用户问"帮我查一下明天北京的天气，如果会下雨，就帮我在日历上创建一个'带伞'的提醒"
传统Chatbot的处理方式：
用户：帮我查一下明天北京的天气，如果会下雨，就帮我在日历上创建一个"带伞"的提醒
Bot：抱歉，我只能回答天气相关的问题，无法操作您的日历。
AI Agent的处理方式：
用户：帮我查一下明天北京的天气，如果会下雨，就帮我在日历上创建一个"带伞"的提醒
Agent内部执行流程：
1. 【思考】用户需要两步操作：先查天气，再根据结果决定是否创建日历提醒
2. 【行动】调用天气API查询明天北京的天气
3. 【观察】天气API返回：明天北京有小雨，温度15-22℃
4. 【思考】天气预报显示有雨，需要创建日历提醒
5. 【行动】调用日历API创建明天的"带伞"提醒
6. 【观察】日历事件创建成功
7. 【回复】明天北京有小雨（15-22℃），我已经帮你在日历上创建了"带伞"提醒，记得带伞哦！
1.3 Agent的核心能力：感知-推理-行动循环
AI Agent的核心工作原理可以用一个简洁的循环来描述：感知（Perceive）→ 推理（Reason）→ 行动（Act），简称PRA循环。
┌─────────────────────────────────────┐
│ AI Agent 循环 │
│ │
│ ┌──────────┐ │
│ │ 感知 │ ← 环境输入 │
│ │ Perceive │ (用户消息、API数据) │
│ └────┬─────┘ │
│ ↓ │
│ ┌──────────┐ │
│ │ 推理 │ ← 内部知识 │
│ │ Reason │ (LLM、记忆、规则) │
│ └────┬─────┘ │
│ ↓ │
│ ┌──────────┐ │
│ │ 行动 │ → 影响环境 │
│ │ Act │ (调用工具、回复) │
│ └────┬─────┘ │
│ │ │
│ └───── 反馈回感知 ──────────┘│
└─────────────────────────────────────┘
感知（Perceive）
感知是Agent获取信息的阶段。Agent通过多种渠道获取环境信息：
# 感知的几种常见形式
# 1. 用户直接输入
user_input = "帮我分析一下这个CSV文件的销售数据"
# 2. API返回的数据
api_response = {"temperature": 25, "humidity": 60, "weather": "晴"}
# 3. 文件系统信息
file_content = open("sales_data.csv").read()
# 4. 数据库查询结果
db_results = [{"product": "A", "sales": 100}, {"product": "B", "sales": 200}]
# 5. 传感器数据（IoT场景）
sensor_data = {"camera": "image_001.jpg", "motion": True}
推理（Reason）
推理是Agent的核心智能所在。Agent利用大语言模型（LLM）作为推理引擎，结合当前的感知信息和历史记忆，做出决策：
# 推理过程的简化表示
def agent_reason(perception, memory, tools):
"""
Agent推理函数
:param perception: 当前感知到的信息
:param memory: 历史记忆
:param available_tools: 可用工具列表
:return: 决策结果（下一步行动）
"""
# 构建推理上下文
context = build_context(perception, memory)
# 使用LLM进行推理
reasoning_prompt = f"""
当前情况：{context}
可用工具：{tools}
请分析当前情况，决定下一步应该采取什么行动。
"""
# LLM返回决策
decision = llm.generate(reasoning_prompt)
return decision
行动（Act）
行动是Agent执行决策的阶段。Agent可以通过多种方式影响环境：
# 行动的几种常见形式
# 1. 调用外部工具
result = weather_api.get_forecast("北京", "明天")
# 2. 生成回复文本
response = "明天北京天气晴朗，温度25℃"
# 3. 操作文件系统
with open("report.txt", "w") as f:
f.write(generated_report)
# 4. 发送消息通知
notification_service.send(user_id, "任务已完成")
# 5. 更新内部状态
memory.add("用户偏好：喜欢简洁的回复风格")
1.4 一个简单的Agent示例
让我们用Python写一个最简单的Agent，来直观感受Agent的工作方式：
"""
最简单的AI Agent示例
这个Agent可以回答问题，并在需要时使用计算器工具
"""
import re
class SimpleAgent:
def __init__(self):
self.name = "简易助手"
self.tools = {
"calculator": self.calculator,
"echo": self.echo,
}
self.conversation_history = []
def calculator(self, expression):
"""计算数学表达式"""
try:
# 安全地计算数学表达式
allowed_chars = set("0123456789+-*/.() ")
if all(c in allowed_chars for c in expression):
result = eval(expression)
return f"计算结果：{expression} = {result}"
else:
return "错误：表达式包含不允许的字符"
except Exception as e:
return f"计算错误：{str(e)}"
def echo(self, text):
"""回显文本"""
return f"你说的是：{text}"
def perceive(self, user_input):
"""感知：接收用户输入"""
self.conversation_history.append({"role": "user", "content": user_input})
return user_input
def reason(self, user_input):
"""推理：决定使用什么工具"""
# 检测是否包含数学计算请求
calc_pattern = r'计算\s*([\d+\-*/.()\s]+)'
match = re.search(calc_pattern, user_input)
if match:
expression = match.group(1).strip()
return "calculator", expression
# 默认使用echo工具
return "echo", user_input
def act(self, tool_name, tool_input):
"""行动：执行工具"""
if tool_name in self.tools:
result = self.tools[tool_name](tool_input)
self.conversation_history.append({"role": "assistant", "content": result})
return result
else:
return f"抱歉，我不认识工具 '{tool_name}'"
def run(self, user_input):
"""运行一个完整的感知-推理-行动循环"""
# 第一步：感知
perception = self.perceive(user_input)
# 第二步：推理
tool_name, tool_input = self.reason(perception)
# 第三步：行动
result = self.act(tool_name, tool_input)
return result
# 使用示例
agent = SimpleAgent()
# 测试计算功能
print(agent.run("请帮我计算 123 + 456 的结果"))
# 输出：计算结果：123 + 456 = 579
# 测试回显功能
print(agent.run("你好，今天天气怎么样"))
# 输出：你说的是：你好，今天天气怎么样
这个简单的例子展示了Agent的基本工作流程。虽然功能简单，但它完整地体现了感知-推理-行动的循环模式。在后续章节中，我们将学习如何使用大语言模型来增强Agent的推理能力，使其能够处理更加复杂的任务。
第二章：Agent架构详解
2.1 ReAct模式
ReAct（Reasoning + Acting）是目前最流行的Agent架构模式之一。它的核心思想是让Agent在执行任务时，交替进行推理（Reasoning）和行动（Acting），每一步行动之后都会观察结果，然后进行下一步推理。
ReAct的工作流程可以用以下伪代码表示：
def react_agent(task, tools, max_iterations=10):
"""
ReAct Agent 的核心循环
"""
context = task # 初始任务
for i in range(max_iterations):
# 第一步：思考（Thought）
thought = llm.think(f"""
任务：{context}
可用工具：{list(tools.keys())}
请思考下一步应该做什么？
- 分析当前进度
- 确定还需要哪些信息
- 选择合适的工具
""")
# 第二步：行动（Action）
action = llm.decide_action(thought, tools)
# 第三步：观察（Observation）
observation = execute_action(action, tools)
# 更新上下文
context += f"\n思考：{thought}\n行动：{action}\n观察：{observation}"
# 检查是否完成
if is_task_complete(observation):
return llm.generate_final_answer(context)
return "任务未能在限定步骤内完成"
让我们用一个具体的例子来理解ReAct模式。假设用户问："2024年诺贝尔物理学奖得主的年龄之和是多少？"
【第1轮】
思考(Thought)：我需要找到2024年诺贝尔物理学奖得主是谁。
行动(Action)：search("2024年诺贝尔物理学奖得主")
观察(Observation)：2024年诺贝尔物理学奖授予了John Hopfield和Geoffrey Hinton，
表彰他们在人工神经网络机器学习方面的基础性发现和发明。
【第2轮】
思考(Thought)：现在我知道了两位得主的名字，我需要查找他们的出生年份来计算年龄。
行动(Action)：search("John Hopfield 出生年份")
观察(Observation)：John Hopfield出生于1933年7月15日。
【第3轮】
思考(Thought)：John Hopfield出生于1933年，2024年时91岁。现在需要查Geoffrey Hinton的年龄。
行动(Action)：search("Geoffrey Hinton 出生年份")
观察(Observation)：Geoffrey Hinton出生于1947年12月6日。
【第4轮】
思考(Thought)：Geoffrey Hinton出生于1947年，2024年时77岁。年龄之和为91+77=168。
行动(Action)：finish("2024年诺贝尔物理学奖得主John Hopfield（91岁）和
Geoffrey Hinton（77岁）的年龄之和为168岁。")
下面是一个ReAct模式的完整Python实现：
"""
ReAct Agent 完整实现
"""
from typing import Dict, List, Callable, Any
import json
import re
class ReActAgent:
"""基于ReAct模式的Agent"""
def __init__(self, llm_client, tools: Dict[str, Callable]):
self.llm = llm_client
self.tools = tools
self.max_iterations = 10
self.system_prompt = """你是一个ReAct模式的AI助手。
在回答问题时，你需要按照以下格式交替进行思考和行动：
思考：分析当前情况，确定下一步需要做什么
行动：调用合适的工具
观察：记录工具返回的结果
当你获得足够信息后，用以下格式给出最终答案：
最终答案：[你的答案]
可用工具：{tools_description}
"""
def _build_tools_description(self) -> str:
"""构建工具描述"""
descriptions = []
for name, func in self.tools.items():
doc = func.__doc__ or "无描述"
descriptions.append(f"- {name}: {doc}")
return "\n".join(descriptions)
def _parse_action(self, text: str) -> tuple:
"""解析Agent输出中的行动"""
# 匹配 "行动：工具名(参数)" 的格式
action_pattern = r'行动[：:]\s*(\w+)\((.*?)\)'
match = re.search(action_pattern, text)
if match:
tool_name = match.group(1)
tool_input = match.group(2).strip()
return tool_name, tool_input
return None, None
def _execute_tool(self, tool_name: str, tool_input: str) -> str:
"""执行工具"""
if tool_name in self.tools:
try:
result = self.tools[tool_name](tool_input)
return str(result)
except Exception as e:
return f"工具执行错误：{str(e)}"
else:
return f"未知工具：{tool_name}"
def run(self, task: str) -> str:
"""运行ReAct循环"""
# 初始化上下文
context = self.system_prompt.format(
tools_description=self._build_tools_description()
)
context += f"\n\n任务：{task}\n"
for iteration in range(self.max_iterations):
# 让LLM进行思考和决定行动
response = self.llm.generate(context)
context += f"\n{response}\n"
# 检查是否包含最终答案
if "最终答案" in response:
answer_pattern = r'最终答案[：:]\s*(.*?)(?:\n|$)'
match = re.search(answer_pattern, response)
if match:
return match.group(1)
# 解析并执行行动
tool_name, tool_input = self._parse_action(response)
if tool_name:
observation = self._execute_tool(tool_name, tool_input)
context += f"观察：{observation}\n"
else:
# 如果没有识别到行动，让LLM继续
context += "请按照指定格式继续思考和行动。\n"
return "任务未能在限定步骤内完成。"
# 使用示例
def search_web(query: str) -> str:
"""搜索网络获取信息"""
mock_results = {
"2024年诺贝尔物理学奖": "John Hopfield和Geoffrey Hinton获得2024年诺贝尔物理学奖",
"John Hopfield年龄": "John Hopfield出生于1933年",
"Geoffrey Hinton年龄": "Geoffrey Hinton出生于1947年",
}
for key, value in mock_results.items():
if key in query:
return value
return f"未找到关于'{query}'的结果"
def calculator(expression: str) -> str:
"""计算数学表达式"""
try:
result = eval(expression)
return f"{expression} = {result}"
except Exception as e:
return f"计算错误：{e}"
# 创建Agent
agent = ReActAgent(
llm_client=your_llm_client, # 替换为实际的LLM客户端
tools={"search": search_web, "calculate": calculator}
)
# 运行
result = agent.run("2024年诺贝尔物理学奖得主的年龄之和是多少？")
print(result)
2.2 Plan-and-Execute模式
Plan-and-Execute（规划与执行）模式是另一种常见的Agent架构。与ReAct的逐步推理不同，Plan-and-Execute模式首先制定一个完整的计划，然后逐步执行。
这种模式的核心思想是先想清楚再动手，类似于人类在处理复杂任务时的做法——先列出步骤，然后按步骤执行。
┌─────────────────────────────────────┐
│ Plan-and-Execute 流程 │
│ │
│ ┌────────────┐ │
│ │ 制定计划 │ → [步骤1, 步骤2, ...] │
│ │ Plan │ │
│ └─────┬──────┘ │
│ ↓ │
│ ┌────────────┐ │
│ │ 执行步骤 │ ← 当前步骤 │
│ │ Execute │ │
│ └─────┬──────┘ │
│ ↓ │
│ ┌────────────┐ │
│ │ 检查结果 │ → 是否需要调整计划？ │
│ │ Evaluate │ │
│ └─────┬──────┘ │
│ ↓ │
│ ┌────────────┐ │
│ │ 调整计划 │ (如果需要) │
│ │ Replan │ │
│ └────────────┘ │
└─────────────────────────────────────┘
"""
Plan-and-Execute Agent 实现
"""
from dataclasses import dataclass, field
from typing import List, Optional
from enum import Enum
import json
class StepStatus(Enum):
PENDING = "pending"
IN_PROGRESS = "in_progress"
COMPLETED = "completed"
FAILED = "failed"
@dataclass
class PlanStep:
"""计划中的一个步骤"""
step_id: int
description: str
tool_name: Optional[str] = None
tool_input: Optional[str] = None
status: StepStatus = StepStatus.PENDING
result: Optional[str] = None
@dataclass
class Plan:
"""执行计划"""
goal: str
steps: List[PlanStep] = field(default_factory=list)
def get_next_pending_step(self) -> Optional[PlanStep]:
for step in self.steps:
if step.status == StepStatus.PENDING:
return step
return None
def is_complete(self) -> bool:
return all(s.status == StepStatus.COMPLETED for s in self.steps)
def has_failed(self) -> bool:
return any(s.status == StepStatus.FAILED for s in self.steps)
class PlanAndExecuteAgent:
"""Plan-and-Execute模式的Agent"""
def __init__(self, llm_client, tools: dict):
self.llm = llm_client
self.tools = tools
def create_plan(self, goal: str) -> Plan:
"""第一步：制定计划"""
planning_prompt = f"""
目标：{goal}
可用工具：{list(self.tools.keys())}
请将这个目标分解为具体的执行步骤。
每个步骤应该清晰明确，并指明使用哪个工具。
请以JSON格式输出步骤列表：
[
{{"step": 1, "description": "...", "tool": "...", "input": "..."}},
...
]
"""
response = self.llm.generate(planning_prompt)
steps_data = json.loads(response)
plan = Plan(goal=goal)
for i, step_data in enumerate(steps_data):
step = PlanStep(
step_id=i + 1,
description=step_data["description"],
tool_name=step_data.get("tool"),
tool_input=step_data.get("input")
)
plan.steps.append(step)
return plan
def execute_step(self, step: PlanStep) -> str:
"""执行单个步骤"""
step.status = StepStatus.IN_PROGRESS
if step.tool_name and step.tool_name in self.tools:
try:
result = self.tools[step.tool_name](step.tool_input)
step.status = StepStatus.COMPLETED
step.result = str(result)
except Exception as e:
step.status = StepStatus.FAILED
step.result = str(e)
else:
# 不需要工具的步骤，直接用LLM处理
result = self.llm.generate(step.description)
step.status = StepStatus.COMPLETED
step.result = result
return step.result
def replan_if_needed(self, plan: Plan) -> Plan:
"""根据执行结果决定是否需要调整计划"""
if plan.has_failed():
failed_steps = [s for s in plan.steps if s.status == StepStatus.FAILED]
replan_prompt = f"""
原计划：{[s.description for s in plan.steps]}
失败步骤：{[(s.description, s.result) for s in failed_steps]}
请制定一个新的计划来处理失败的步骤。
"""
new_steps = self.llm.generate(replan_prompt)
# 解析并替换失败的步骤
# ... (解析逻辑)
return plan
def run(self, goal: str) -> str:
"""运行完整的Plan-and-Execute流程"""
# 第一步：制定计划
plan = self.create_plan(goal)
print(f"计划已制定，共{len(plan.steps)}个步骤")
# 第二步：逐步执行
while not plan.is_complete():
current_step = plan.get_next_pending_step()
if current_step is None:
break
print(f"正在执行步骤{current_step.step_id}：{current_step.description}")
result = self.execute_step(current_step)
print(f"结果：{result}")
# 检查是否需要重新规划
if plan.has_failed():
plan = self.replan_if_needed(plan)
# 第三步：汇总结果
completed_results = [
f"步骤{s.step_id}：{s.description} → {s.result}"
for s in plan.steps
if s.status == StepStatus.COMPLETED
]
summary_prompt = f"""
目标：{goal}
执行结果：
{chr(10).join(completed_results)}
请根据以上执行结果，给出最终的总结回答。
"""
return self.llm.generate(summary_prompt)
2.3 工具调用链
工具调用链是Agent执行复杂任务的关键机制。一个任务可能需要多个工具协同工作，前一个工具的输出会成为后一个工具的输入。
"""
工具调用链示例
展示Agent如何将多个工具串联起来完成复杂任务
"""
class ToolChain:
"""工具调用链"""
def __init__(self):
self.chain = []
def add_step(self, tool_name: str, input_template: str):
"""添加一个步骤到调用链"""
self.chain.append({
"tool": tool_name,
"input_template": input_template,
"result_key": f"step_{len(self.chain)}_result"
})
return self # 支持链式调用
def execute(self, tools: dict, initial_context: dict) -> dict:
"""执行整个调用链"""
context = initial_context.copy()
for i, step in enumerate(self.chain):
tool_name = step["tool"]
input_template = step["input_template"]
# 使用模板生成实际输入
actual_input = input_template.format(**context)
# 执行工具
if tool_name in tools:
result = tools[tool_name](actual_input)
context[step["result_key"]] = result
print(f"步骤{i+1} [{tool_name}]: {actual_input} → {result}")
else:
raise ValueError(f"工具 {tool_name} 不存在")
return context
# 使用示例：构建一个"搜索并总结"的工具链
def search(query: str) -> str:
"""搜索网络"""
return f"搜索结果：关于'{query}'的前3条结果..."
def summarize(text: str) -> str:
"""总结文本"""
return f"总结：{text[:50]}..."
def translate(text: str, target_lang: str = "中文") -> str:
"""翻译文本"""
return f"翻译({target_lang})：{text}"
# 构建工具链
chain = ToolChain()
chain.add_step("search", "{query}") # 步骤1：搜索
chain.add_step("summarize", "{step_0_result}") # 步骤2：总结搜索结果
chain.add_step("translate", "{step_1_result}") # 步骤3：翻译总结
# 执行
tools = {"search": search, "summarize": summarize, "translate": translate}
result = chain.execute(tools, {"query": "AI Agent最新进展"})
print(f"最终结果：{result['step_2_result']}")
2.4 ReAct与Plan-and-Execute的对比与选型
两种模式各有优劣，选择哪种取决于具体的任务特性：
| 维度 | ReAct | Plan-and-Execute |
|---|---|---|
| 适用场景 | 信息查询、简单推理 | 复杂多步骤任务 |
| 灵活性 | 高，可随时调整 | 中，需要重新规划 |
| 可预测性 | 低，行为不确定 | 高，计划可审查 |
| 资源消耗 | 每步都需要LLM推理 | 只有规划和调整需要 |
| 错误恢复 | 自然地逐步修正 | 需要显式重新规划 |
| 开发复杂度 | 简单 | 较复杂 |
选择建议：
- 选择ReAct：当任务不太复杂，每一步都可能需要根据上一步的结果来决定时
- 选择Plan-and-Execute：当任务明确可以分解为多个独立步骤，且需要高效的执行时
- 混合使用：在Plan-and-Execute的每个执行步骤内部使用ReAct模式
第三章：主流Agent框架对比
3.1 LangChain Agents
LangChain是目前最流行的LLM应用开发框架之一，它提供了丰富的Agent实现和工具集成。
核心特点：
- 模块化设计，组件可自由组合
- 丰富的工具生态（数百个内置工具）
- 支持多种LLM（OpenAI、Anthropic、本地模型等）
- 活跃的社区和完善的文档
"""
LangChain Agent 基础示例
"""
from langchain.agents import AgentExecutor, create_react_agent
from langchain.tools import Tool
from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate
# 初始化LLM
llm = ChatOpenAI(model="gpt-4", temperature=0)
# 定义工具
def search_database(query: str) -> str:
"""搜索产品数据库"""
products = {
"手机": "iPhone 15 Pro: ¥7999, Samsung S24: ¥6999",
"电脑": "MacBook Pro: ¥14999, ThinkPad: ¥8999",
}
for key, value in products.items():
if key in query:
return value
return "未找到相关产品"
def get_weather(city: str) -> str:
"""获取城市天气"""
return f"{city}今天天气晴朗，温度25℃"
# 创建工具列表
tools = [
Tool(
name="ProductSearch",
func=search_database,
description="搜索产品数据库，输入应该是产品类别关键词"
),
Tool(
name="Weather",
func=get_weather,
description="获取指定城市的天气信息"
)
]
# 定义提示模板
prompt = PromptTemplate.from_template("""
Answer the following questions as best you can. You have access to the following tools:
{tools}
Use the following format:
Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question
Begin!
Question: {input}
Thought:{agent_scratchpad}
""")
# 创建Agent
agent = create_react_agent(llm, tools, prompt)
# 创建Agent执行器
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
max_iterations=5,
handle_parsing_errors=True
)
# 运行
result = agent_executor.invoke({"input": "帮我查一下手机的价格，顺便看看北京的天气"})
print(result["output"])
3.2 AutoGPT
AutoGPT是最早引起广泛关注的自主Agent项目之一，它展示了AI Agent自主完成复杂任务的能力。
核心特点：
- 完全自主的任务执行
- 内置网页浏览、文件操作等能力
- 支持长期记忆（通过向量数据库）
- 适合探索性和创意性任务
"""
AutoGPT风格的自主Agent简化实现
"""
class AutoGPTStyleAgent:
"""AutoGPT风格的自主Agent"""
def __init__(self, llm_client):
self.llm = llm_client
self.memory = [] # 短期记忆
self.long_term_memory = [] # 长期记忆
self.goals = [] # 目标列表
def set_goals(self, goals: list):
"""设置Agent的目标"""
self.goals = goals
print(f"目标已设定：{goals}")
def think(self) -> str:
"""自主思考下一步"""
prompt = f"""
我是一个自主AI助手，我的目标是：
{chr(10).join(f'- {g}' for g in self.goals)}
我目前的记忆：
{chr(10).join(self.memory[-5:])} # 只取最近5条
基于我的目标和当前状态，我下一步应该做什么？
请用以下格式回答：
思考：[你的分析]
计划：[具体行动计划]
执行：[需要执行的具体命令或操作]
"""
response = self.llm.generate(prompt)
self.memory.append(f"思考：{response}")
return response
def execute_plan(self, plan: str) -> str:
"""执行计划"""
result = f"执行结果：已完成 {plan}"
self.memory.append(result)
return result
def reflect(self) -> str:
"""反思当前进展"""
prompt = f"""
回顾我的执行过程：
{chr(10).join(self.memory)}
我是否在朝着目标前进？有什么需要调整的吗？
"""
reflection = self.llm.generate(prompt)
self.long_term_memory.append(reflection)
return reflection
def run(self, max_iterations: int = 10):
"""运行自主循环"""
for i in range(max_iterations):
print(f"\n=== 第{i+1}轮 ===")
# 思考
thought = self.think()
print(f"思考：{thought}")
# 执行
result = self.execute_plan(thought)
print(f"结果：{result}")
# 每5轮反思一次
if (i + 1) % 5 == 0:
reflection = self.reflect()
print(f"反思：{reflection}")
3.3 CrewAI
CrewAI专注于多Agent协作，它将Agent组织成一个"团队"（Crew），每个Agent扮演特定的角色。
核心特点：
- 角色驱动的Agent设计
- 支持Agent之间的任务委派
- 内置的任务管理和协作机制
- 适合模拟真实团队协作场景
"""
CrewAI 风格的多Agent协作示例
"""
from dataclasses import dataclass
from typing import List, Callable
@dataclass
class AgentRole:
"""Agent角色定义"""
name: str
goal: str
backstory: str
tools: List[Callable]
verbose: bool = True
class CrewAgent:
"""单个Agent"""
def __init__(self, role: AgentRole, llm_client):
self.role = role
self.llm = llm_client
self.assigned_tasks = []
def execute_task(self, task: str) -> str:
"""执行分配的任务"""
prompt = f"""
角色：{self.role.name}
目标：{self.role.goal}
背景：{self.role.backstory}
当前任务：{task}
请以你的专业角色身份完成这个任务。
"""
if self.role.verbose:
print(f"[{self.role.name}] 正在执行：{task}")
result = self.llm.generate(prompt)
self.assigned_tasks.append({"task": task, "result": result})
return result
class Crew:
"""Agent团队"""
def __init__(self, agents: List[CrewAgent], tasks: List[str]):
self.agents = agents
self.tasks = tasks
self.results = {}
def kickoff(self) -> dict:
"""启动团队协作"""
print("=" * 50)
print("团队任务开始执行")
print("=" * 50)
for task in self.tasks:
best_agent = self._select_agent(task)
print(f"\n分配任务给 [{best_agent.role.name}]：")
print(f"任务：{task}")
result = best_agent.execute_task(task)
self.results[task] = {
"agent": best_agent.role.name,
"result": result
}
return self.results
def _select_agent(self, task: str) -> CrewAgent:
"""根据任务选择最合适的Agent"""
task_lower = task.lower()
for agent in self.agents:
role_keywords = agent.role.goal.lower().split()
if any(kw in task_lower for kw in role_keywords):
return agent
return self.agents[0]
# 使用示例：创建一个内容创作团队
writer = CrewAgent(
role=AgentRole(
name="内容创作者",
goal="撰写高质量的文章内容",
backstory="你是一位资深的内容创作者，擅长将复杂概念用通俗易懂的语言表达",
tools=[]
),
llm_client=your_llm
)
editor = CrewAgent(
role=AgentRole(
name="编辑",
goal="审核和改进文章质量",
backstory="你是一位严格的标准编辑，注重内容的准确性和可读性",
tools=[]
),
llm_client=your_llm
)
seo_expert = CrewAgent(
role=AgentRole(
name="SEO专家",
goal="优化文章的搜索引擎排名",
backstory="你是一位SEO专家，擅长关键词优化和内容结构优化",
tools=[]
),
llm_client=your_llm
)
# 创建团队任务
tasks = [
"撰写一篇关于AI Agent技术趋势的博客文章，约1500字",
"审核文章内容，确保准确性和可读性",
"为文章添加SEO优化建议，包括关键词和标题优化"
]
crew = Crew(agents=[writer, editor, seo_expert], tasks=tasks)
results = crew.kickoff()
3.4 MetaGPT
MetaGPT是一个创新的多Agent框架，它模拟了软件公司的组织结构，让Agent扮演产品经理、架构师、工程师等不同角色来协作完成软件开发任务。
核心特点：
- 标准化操作程序（SOP）
- 结构化输出（PRD、技术设计文档等）
- 角色间的信息共享机制
- 专注于软件开发场景
"""
MetaGPT 风格的软件开发团队模拟
"""
from dataclasses import dataclass
from typing import Dict, List, Any
from enum import Enum
class Role(Enum):
PRODUCT_MANAGER = "产品经理"
ARCHITECT = "架构师"
ENGINEER = "工程师"
QA = "测试工程师"
@dataclass
class Document:
"""标准文档"""
title: str
content: str
author: Role
version: str = "1.0"
class MetaGPTAgent:
"""MetaGPT风格的Agent"""
def __init__(self, role: Role, llm_client):
self.role = role
self.llm = llm_client
self.shared_memory: Dict[str, Any] = {}
def produce_document(self, input_info: str) -> Document:
"""根据角色产出相应文档"""
templates = {
Role.PRODUCT_MANAGER: {
"title": "产品需求文档(PRD)",
"prompt": """作为产品经理，请根据以下需求信息生成PRD：
包含：需求背景、用户故事、功能列表、验收标准
需求信息：{input}"""
},
Role.ARCHITECT: {
"title": "技术设计文档",
"prompt": """作为架构师，请根据PRD生成技术设计文档：
包含：系统架构、模块划分、接口定义、技术选型
PRD内容：{input}"""
},
Role.ENGINEER: {
"title": "代码实现",
"prompt": """作为工程师，请根据技术设计文档实现代码：
要求：代码规范、注释完善、错误处理
设计文档：{input}"""
},
Role.QA: {
"title": "测试报告",
"prompt": """作为测试工程师，请根据需求和代码生成测试报告：
包含：测试用例、测试结果、bug列表
代码实现：{input}"""
}
}
template = templates[self.role]
content = self.llm.generate(template["prompt"].format(input=input_info))
return Document(
title=template["title"],
content=content,
author=self.role
)
class MetaGPTTeam:
"""MetaGPT软件开发团队"""
def __init__(self, llm_client):
self.team = {
Role.PRODUCT_MANAGER: MetaGPTAgent(Role.PRODUCT_MANAGER, llm_client),
Role.ARCHITECT: MetaGPTAgent(Role.ARCHITECT, llm_client),
Role.ENGINEER: MetaGPTAgent(Role.ENGINEER, llm_client),
Role.QA: MetaGPTAgent(Role.QA, llm_client),
}
self.documents: Dict[str, Document] = {}
def develop_software(self, requirement: str) -> Dict[str, Document]:
"""执行软件开发流程"""
print("开始软件开发流程...\n")
# 第1步：产品经理生成PRD
print("产品经理正在分析需求...")
prd = self.team[Role.PRODUCT_MANAGER].produce_document(requirement)
self.documents["prd"] = prd
print(f"✓ {prd.title} 已完成\n")
# 第2步：架构师生成技术设计
print("架构师正在设计系统架构...")
design = self.team[Role.ARCHITECT].produce_document(prd.content)
self.documents["design"] = design
print(f"✓ {design.title} 已完成\n")
# 第3步：工程师实现代码
print("工程师正在编写代码...")
code = self.team[Role.ENGINEER].produce_document(design.content)
self.documents["code"] = code
print(f"✓ {code.title} 已完成\n")
# 第4步：测试工程师进行测试
print("测试工程师正在执行测试...")
test_report = self.team[Role.QA].produce_document(code.content)
self.documents["test"] = test_report
print(f"✓ {test_report.title} 已完成\n")
return self.documents
3.5 框架选型建议
| 框架 | 最适合场景 | 学习曲线 | 社区活跃度 |
|---|---|---|---|
| LangChain | 通用Agent开发、工具集成 | 中等 | 非常活跃 |
| AutoGPT | 自主探索性任务 | 较低 | 活跃 |
| CrewAI | 多角色协作任务 | 低 | 快速增长 |
| MetaGPT | 软件开发自动化 | 较高 | 活跃 |
选型建议：
- 初学者入门：从LangChain开始，它的文档最完善，社区支持最好
- 多Agent项目：选择CrewAI，它的角色驱动设计很直观
- 软件开发自动化：选择MetaGPT，它有成熟的工作流
- 快速原型验证：直接用原生Python + LLM API，不依赖框架
第四章：LangChain Agent开发实战
4.1 环境搭建
在开始开发之前，我们需要搭建好开发环境。以下是详细的步骤：
# 第1步：创建项目目录
mkdir my-agent-project
cd my-agent-project
# 第2步：创建Python虚拟环境
python -m venv venv
# 激活虚拟环境（Linux/Mac）
source venv/bin/activate
# 激活虚拟环境（Windows）
venv\Scripts\activate
# 第3步：安装必要的依赖
pip install langchain langchain-openai langchain-community
pip install python-dotenv # 用于管理环境变量
pip install requests # 用于HTTP请求
# 第4步：创建环境变量文件
echo 'OPENAI_API_KEY=your_api_key_here' > .env
项目结构应该如下：
my-agent-project/
├── venv/ # Python虚拟环境
├── .env # 环境变量（不要提交到Git）
├── .gitignore # Git忽略文件
├── requirements.txt # 项目依赖
├── main.py # 主程序入口
├── tools/ # 自定义工具
│ ├── __init__.py
│ ├── search.py # 搜索工具
│ └── calculator.py # 计算器工具
└── utils/ # 工具函数
├── __init__.py
└── llm.py # LLM配置
requirements.txt 文件内容：
langchain>=0.2.0
langchain-openai>=0.1.0
langchain-community>=0.2.0
python-dotenv>=1.0.0
requests>=2.31.0
4.2 第一个Agent
让我们创建第一个能够使用工具的Agent：
"""
第一个LangChain Agent
功能：能够搜索天气和进行简单计算
"""
import os
from dotenv import load_dotenv
# 加载环境变量
load_dotenv()
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.tools import tool
# ============================================
# 第1步：定义工具
# ============================================
@tool
def get_weather(city: str) -> str:
"""获取指定城市的当前天气信息。
Args:
city: 城市名称，如"北京"、"上海"
"""
weather_data = {
"北京": "晴天，温度25℃，湿度40%",
"上海": "多云，温度28℃，湿度65%",
"广州": "小雨，温度30℃，湿度80%",
"深圳": "阴天，温度29℃，湿度70%",
}
return weather_data.get(city, f"暂无{city}的天气数据")
@tool
def calculator(expression: str) -> str:
"""计算数学表达式。
Args:
expression: 数学表达式，如 "2 + 3 * 4"
"""
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)}"
@tool
def word_count(text: str) -> str:
"""统计文本的字数和词数。
Args:
text: 需要统计的文本
"""
char_count = len(text)
word_count = len(text.split())
return f"字符数：{char_count}，词数：{word_count}"
# ============================================
# 第2步：配置LLM
# ============================================
llm = ChatOpenAI(
model="gpt-4",
temperature=0,
)
# ============================================
# 第3步：创建提示模板
# ============================================
prompt = ChatPromptTemplate.from_messages([
("system", """你是一个有用的AI助手。你可以使用工具来帮助回答问题。
请用中文回答所有问题。
在使用工具之前，请先思考需要使用哪个工具以及输入什么参数。"""),
MessagesPlaceholder(variable_name="chat_history", optional=True),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad"),
])
# ============================================
# 第4步：创建Agent
# ============================================
tools = [get_weather, calculator, word_count]
agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
max_iterations=5,
handle_parsing_errors=True,
)
# ============================================
# 第5步：运行Agent
# ============================================
def main():
"""主函数：运行Agent交互循环"""
print("=" * 50)
print("AI Agent 已启动！")
print("输入 'quit' 退出")
print("=" * 50)
while True:
user_input = input("\n你：")
if user_input.lower() in ['quit', 'exit', '退出']:
print("再见！")
break
try:
result = agent_executor.invoke({"input": user_input})
print(f"\nAI：{result['output']}")
except Exception as e:
print(f"\n错误：{str(e)}")
if __name__ == "__main__":
main()
4.3 工具定义与注册
LangChain提供了多种定义工具的方式，让我们详细了解每种方式：
"""
LangChain 工具定义的多种方式
"""
from langchain_core.tools import tool, StructuredTool
from langchain_core.pydantic_v1 import BaseModel, Field
from typing import Optional
# ============================================
# 方式1：使用 @tool 装饰器（最简单）
# ============================================
@tool
def search_web(query: str) -> str:
"""搜索互联网获取信息。
Args:
query: 搜索关键词
"""
return f"搜索 '{query}' 的结果：这里是模拟的搜索结果..."
# ============================================
# 方式2：使用 @tool 装饰器 + 类型注解（推荐）
# ============================================
@tool
def create_reminder(
content: str,
time: str,
priority: Optional[str] = "medium"
) -> str:
"""创建一个提醒事项。
Args:
content: 提醒内容
time: 提醒时间，格式为 "YYYY-MM-DD HH:MM"
priority: 优先级，可选值为 "low", "medium", "high"
"""
return f"已创建提醒：{content}，时间：{time}，优先级：{priority}"
# ============================================
# 方式3：使用 BaseModel 定义结构化输入
# ============================================
class SearchInput(BaseModel):
"""搜索工具的输入参数"""
query: str = Field(description="搜索关键词")
num_results: int = Field(default=5, description="返回结果数量")
language: str = Field(default="zh", description="搜索语言，zh表示中文")
def structured_search(query: str, num_results: int = 5, language: str = "zh") -> str:
"""执行结构化搜索"""
return f"搜索 '{query}'，返回{num_results}条{language}结果"
search_tool = StructuredTool.from_function(
func=structured_search,
name="advanced_search",
description="高级搜索工具，支持指定结果数量和语言",
args_schema=SearchInput,
)
# ============================================
# 方式4：从类继承创建工具
# ============================================
class DatabaseQueryTool(BaseModel):
"""数据库查询工具"""
name: str = "database_query"
description: str = "查询数据库中的数据"
def _run(self, query: str) -> str:
"""同步执行"""
return f"执行查询 '{query}'，返回10条记录"
async def _arun(self, query: str) -> str:
"""异步执行"""
return self._run(query)
# ============================================
# 工具注册示例
# ============================================
def create_agent_with_tools():
"""创建带有自定义工具的Agent"""
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
# 收集所有工具
tools = [
search_web,
create_reminder,
search_tool,
]
# 创建LLM
llm = ChatOpenAI(model="gpt-4", temperature=0)
# 创建提示模板
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个有用的助手，可以使用以下工具：{tools}"),
("human", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad"),
])
# 创建Agent
agent = create_tool_calling_agent(llm, tools, prompt)
return AgentExecutor(agent=agent, tools=tools, verbose=True)
4.4 使用本地模型
如果你不想使用OpenAI的API，也可以使用本地部署的模型。以下是使用Ollama的示例：
"""
使用本地Ollama模型的Agent
"""
from langchain_community.llms import Ollama
from langchain.agents import AgentExecutor, create_react_agent
from langchain_core.tools import tool
from langchain_core.prompts import PromptTemplate
# 使用本地Ollama模型
llm = Ollama(model="qwen2.5:7b") # 或其他本地模型
@tool
def get_current_time() -> str:
"""获取当前时间"""
from datetime import datetime
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
@tool
def read_file(filepath: str) -> str:
"""读取文件内容
Args:
filepath: 文件路径
"""
try:
with open(filepath, 'r', encoding='utf-8') as f:
return f.read()
except Exception as e:
return f"读取文件失败：{str(e)}"
# ReAct提示模板（适合本地模型）
template = """Answer the following questions as best you can. You have access to the following tools:
{tools}
Use the following format:
Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question
Begin!
Question: {input}
Thought:"""
prompt = PromptTemplate.from_template(template)
# 创建Agent
tools = [get_current_time, read_file]
agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
max_iterations=5,
handle_parsing_errors=True,
)
# 测试
result = agent_executor.invoke({"input": "现在几点了？"})
print(result["output"])
第五章：工具系统深入
5.1 自定义工具开发
工具是Agent与外部世界交互的桥梁。一个好的工具应该有清晰的定义、完善的错误处理和详细的文档。
"""
自定义工具开发最佳实践
"""
from langchain_core.tools import tool
from typing import Optional, List
import json
import os
# ============================================
# 示例1：文件操作工具
# ============================================
@tool
def read_file(filepath: str, encoding: str = "utf-8") -> str:
"""读取指定路径的文件内容。
支持读取文本文件，自动检测文件编码。
Args:
filepath: 文件的绝对路径或相对路径
encoding: 文件编码，默认utf-8
Returns:
文件内容的字符串，如果文件不存在返回错误信息
"""
try:
# 安全检查：限制访问路径
allowed_dirs = ["/tmp", os.path.expanduser("~/Documents")]
abs_path = os.path.abspath(filepath)
if not any(abs_path.startswith(d) for d in allowed_dirs):
return f"错误：不允许访问路径 {filepath}"
if not os.path.exists(filepath):
return f"错误：文件 {filepath} 不存在"
if os.path.getsize(filepath) > 10 * 1024 * 1024: # 10MB限制
return "错误：文件过大（超过10MB），请指定读取范围"
with open(filepath, 'r', encoding=encoding) as f:
content = f.read()
return f"文件内容（{len(content)}字符）：\n{content}"
except UnicodeDecodeError:
return f"错误：无法使用{encoding}编码读取文件，请尝试其他编码"
except Exception as e:
return f"错误：读取文件时发生异常 - {str(e)}"
@tool
def write_file(filepath: str, content: str, mode: str = "w") -> str:
"""将内容写入指定路径的文件。
Args:
filepath: 文件路径
content: 要写入的内容
mode: 写入模式，'w'覆盖写入，'a'追加写入
"""
try:
# 安全检查
allowed_dirs = ["/tmp", os.path.expanduser("~/Documents")]
abs_path = os.path.abspath(filepath)
if not any(abs_path.startswith(d) for d in allowed_dirs):
return f"错误：不允许写入路径 {filepath}"
# 确保目录存在
os.makedirs(os.path.dirname(abs_path), exist_ok=True)
with open(filepath, mode, encoding='utf-8') as f:
f.write(content)
return f"成功写入文件 {filepath}（{len(content)}字符）"
except Exception as e:
return f"错误：写入文件失败 - {str(e)}"
# ============================================
# 示例2：数据处理工具
# ============================================
@tool
def analyze_csv(filepath: str, column: Optional[str] = None) -> str:
"""分析CSV文件的数据。
可以计算数值列的基本统计信息（均值、中位数、标准差等）。
Args:
filepath: CSV文件路径
column: 要分析的列名（可选，默认分析所有数值列）
"""
try:
import csv
import statistics
with open(filepath, 'r', encoding='utf-8') as f:
reader = csv.DictReader(f)
rows = list(reader)
if not rows:
return "CSV文件为空"
columns = list(rows[0].keys())
result = f"CSV文件包含 {len(rows)} 行数据\n"
result += f"列名：{', '.join(columns)}\n\n"
for col in columns:
if column and col != column:
continue
try:
values = [float(row[col]) for row in rows if row[col]]
if values:
result += f"列 '{col}' 统计：\n"
result += f" 数量：{len(values)}\n"
result += f" 均值：{statistics.mean(values):.2f}\n"
result += f" 中位数：{statistics.median(values):.2f}\n"
if len(values) > 1:
result += f" 标准差：{statistics.stdev(values):.2f}\n"
result += f" 最小值：{min(values)}\n"
result += f" 最大值：{max(values)}\n\n"
except ValueError:
continue
return result
except Exception as e:
return f"错误：分析CSV失败 - {str(e)}"
5.2 工具组合
在实际应用中，我们经常需要将多个工具组合在一起使用：
"""
工具组合：将多个工具串联成工作流
"""
from langchain_core.tools import tool
from typing import List, Dict
import json
class ToolWorkflow:
"""工具工作流"""
def __init__(self):
self.steps = []
self.results = {}
def add_step(self, name: str, tool_func, input_template: str):
"""添加工作流步骤"""
self.steps.append({
"name": name,
"tool": tool_func,
"input_template": input_template
})
return self
def execute(self, initial_context: dict) -> dict:
"""执行工作流"""
context = initial_context.copy()
for step in self.steps:
print(f"执行步骤：{step['name']}")
try:
actual_input = step["input_template"].format(**context)
except KeyError as e:
return {"error": f"步骤 '{step['name']}' 缺少输入参数：{e}"}
try:
result = step["tool"](actual_input)
context[step["name"]] = result
self.results[step["name"]] = result
print(f" 结果：{result[:100]}...")
except Exception as e:
return {"error": f"步骤 '{step['name']}' 执行失败：{e}"}
return context
# 定义工具
@tool
def search(query: str) -> str:
"""搜索信息"""
return f"关于'{query}'的搜索结果：这是一些相关信息..."
@tool
def summarize(text: str) -> str:
"""总结文本"""
return f"总结：{text[:50]}..."
@tool
def translate_to_english(text: str) -> str:
"""翻译成英文"""
return f"English translation: {text}"
# 创建工作流
workflow = ToolWorkflow()
workflow.add_step("search_result", search, "{query}")
workflow.add_step("summary", summarize, "{search_result}")
workflow.add_step("english", translate_to_english, "{summary}")
# 执行工作流
result = workflow.execute({"query": "AI Agent最新进展"})
print(json.dumps(result, ensure_ascii=False, indent=2))
5.3 工具错误处理
健壮的错误处理是生产级Agent的关键：
"""
工具错误处理最佳实践
"""
from langchain_core.tools import tool
from functools import wraps
from typing import Any, Callable
import logging
import json
logger = logging.getLogger(__name__)
# ============================================
# 通用错误处理装饰器
# ============================================
def with_error_handling(max_retries: int = 3, fallback_value: Any = None):
"""工具错误处理装饰器
Args:
max_retries: 最大重试次数
fallback_value: 失败时的默认返回值
"""
def decorator(func: Callable) -> Callable:
@wraps(func)
def wrapper(*args, **kwargs):
last_error = None
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
last_error = e
logger.warning(
f"工具 {func.__name__} 第{attempt+1}次执行失败：{str(e)}"
)
if attempt < max_retries - 1:
import time
time.sleep(2 ** attempt) # 指数退避
error_msg = f"工具 {func.__name__} 执行失败（已重试{max_retries}次）：{str(last_error)}"
logger.error(error_msg)
if fallback_value is not None:
return fallback_value
return f"错误：{error_msg}"
return wrapper
return decorator
# ============================================
# 带错误处理的工具示例
# ============================================
@tool
@with_error_handling(max_retries=3)
def unreliable_api_call(endpoint: str) -> str:
"""调用可能不稳定的外部API
Args:
endpoint: API端点路径
"""
import requests
import random
if random.random() < 0.5:
raise ConnectionError("API连接超时")
response = requests.get(f"https://api.example.com/{endpoint}", timeout=10)
response.raise_for_status()
return response.text
@tool
def safe_calculator(expression: str) -> str:
"""安全的计算器，带有完善的错误处理
Args:
expression: 数学表达式
"""
try:
allowed_chars = set("0123456789+-*/.() ")
if not all(c in allowed_chars for c in expression):
return json.dumps({
"success": False,
"error": "表达式包含不允许的字符"
})
if len(expression) > 100:
return json.dumps({
"success": False,
"error": "表达式过长"
})
result = eval(expression)
return json.dumps({
"success": True,
"expression": expression,
"result": result
})
except ZeroDivisionError:
return json.dumps({"success": False, "error": "除零错误"})
except SyntaxError:
return json.dumps({"success": False, "error": "表达式语法错误"})
except Exception as e:
return json.dumps({"success": False, "error": f"未知错误：{str(e)}"})
5.4 MCP协议简介
MCP（Model Context Protocol，模型上下文协议）是一种新兴的开放标准，旨在为AI模型提供统一的工具和数据访问接口。
"""
MCP协议基础概念和客户端实现示例
MCP的核心思想：
- 标准化AI模型与外部工具/数据源的通信协议
- 类似于USB为外设提供统一接口，MCP为AI工具提供统一接口
- 支持工具发现、调用、结果返回的完整生命周期
"""
from dataclasses import dataclass
from typing import Any, Dict, List, Optional
import json
@dataclass
class MCPToolDefinition:
"""MCP工具定义"""
name: str
description: str
input_schema: Dict[str, Any]
def to_dict(self) -> dict:
return {
"name": self.name,
"description": self.description,
"inputSchema": self.input_schema
}
@dataclass
class MCPToolCall:
"""MCP工具调用"""
tool_name: str
arguments: Dict[str, Any]
@dataclass
class MCPToolResult:
"""MCP工具调用结果"""
success: bool
content: Any
error: Optional[str] = None
class MCPClient:
"""简化的MCP客户端"""
def __init__(self):
self.registered_tools: Dict[str, MCPToolDefinition] = {}
self.tool_handlers: Dict[str, callable] = {}
def register_tool(
self,
name: str,
description: str,
input_schema: dict,
handler: callable
):
"""注册一个MCP工具"""
tool_def = MCPToolDefinition(
name=name,
description=description,
input_schema=input_schema
)
self.registered_tools[name] = tool_def
self.tool_handlers[name] = handler
def list_tools(self) -> List[dict]:
"""列出所有可用工具"""
return [tool.to_dict() for tool in self.registered_tools.values()]
def call_tool(self, call: MCPToolCall) -> MCPToolResult:
"""调用指定工具"""
if call.tool_name not in self.tool_handlers:
return MCPToolResult(
success=False,
content=None,
error=f"工具 '{call.tool_name}' 未注册"
)
try:
handler = self.tool_handlers[call.tool_name]
result = handler(**call.arguments)
return MCPToolResult(success=True, content=result)
except Exception as e:
return MCPToolResult(
success=False,
content=None,
error=str(e)
)
# 使用示例
def create_mcp_server():
"""创建一个MCP服务器"""
client = MCPClient()
# 注册天气工具
client.register_tool(
name="get_weather",
description="获取指定城市的天气信息",
input_schema={
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名称"},
"unit": {"type": "string", "enum": ["celsius", "fahrenheit"], "default": "celsius"}
},
"required": ["city"]
},
handler=lambda city, unit="celsius": f"{city}：晴天，25{'℃' if unit == 'celsius' else '℉'}"
)
# 注册翻译工具
client.register_tool(
name="translate",
description="翻译文本",
input_schema={
"type": "object",
"properties": {
"text": {"type": "string", "description": "要翻译的文本"},
"target_lang": {"type": "string", "description": "目标语言"}
},
"required": ["text", "target_lang"]
},
handler=lambda text, target_lang: f"翻译成{target_lang}：{text}"
)
return client
# 测试MCP服务器
mcp = create_mcp_server()
print("可用工具：")
for tool in mcp.list_tools():
print(f" - {tool['name']}: {tool['description']}")
result = mcp.call_tool(MCPToolCall(
tool_name="get_weather",
arguments={"city": "北京"}
))
print(f"\n调用结果：{result.content}")
第六章：记忆与上下文管理
6.1 短期记忆
短期记忆是Agent在单次对话或任务执行过程中维护的上下文信息。它通常存储在内存中，生命周期与会话相同。
"""
短期记忆实现
"""
from dataclasses import dataclass, field
from typing import List, Dict, Any
from datetime import datetime
from collections import deque
@dataclass
class Message:
"""对话消息"""
role: str # "user", "assistant", "system"
content: str
timestamp: datetime = field(default_factory=datetime.now)
metadata: Dict[str, Any] = field(default_factory=dict)
class ShortTermMemory:
"""短期记忆管理器"""
def __init__(self, max_messages: int = 50, max_tokens: int = 4000):
"""
Args:
max_messages: 最大消息数量
max_tokens: 最大token数（近似值）
"""
self.messages: deque[Message] = deque(maxlen=max_messages)
self.max_tokens = max_tokens
self.current_tokens = 0
def add_message(self, role: str, content: str, **metadata):
"""添加消息到记忆"""
message = Message(role=role, content=content, metadata=metadata)
self.messages.append(message)
# 估算token数（简化：1个中文字符约等于2个token）
estimated_tokens = len(content) * 2
self.current_tokens += estimated_tokens
# 如果超过token限制，移除最早的消息
while self.current_tokens > self.max_tokens and len(self.messages) > 1:
removed = self.messages.popleft()
self.current_tokens -= len(removed.content) * 2
def get_context(self, last_n: int = None) -> List[Dict[str, str]]:
"""获取对话上下文"""
messages = list(self.messages)
if last_n:
messages = messages[-last_n:]
return [{"role": m.role, "content": m.content} for m in messages]
def get_formatted_context(self) -> str:
"""获取格式化的上下文字符串"""
formatted = []
for msg in self.messages:
role_name = {"user": "用户", "assistant": "助手", "system": "系统"}.get(msg.role, msg.role)
formatted.append(f"[{role_name}]: {msg.content}")
return "\n".join(formatted)
def search_messages(self, keyword: str) -> List[Message]:
"""搜索包含关键词的消息"""
return [m for m in self.messages if keyword.lower() in m.content.lower()]
def clear(self):
"""清空记忆"""
self.messages.clear()
self.current_tokens = 0
def summary(self) -> str:
"""生成记忆摘要"""
return f"记忆中有 {len(self.messages)} 条消息，约 {self.current_tokens} 个token"
# 使用示例
memory = ShortTermMemory(max_messages=20)
# 添加对话
memory.add_message("user", "你好，我想了解一下Python编程")
memory.add_message("assistant", "你好！Python是一门非常适合初学者的编程语言...")
memory.add_message("user", "能推荐一些学习资源吗？")
memory.add_message("assistant", "当然！以下是几个推荐的Python学习资源...")
# 获取上下文
context = memory.get_context()
print(context)
# 搜索
results = memory.search_messages("Python")
print(f"找到 {len(results)} 条相关消息")
6.2 长期记忆
长期记忆允许Agent跨会话保存和检索信息。通常使用向量数据库来实现语义搜索。
"""
长期记忆实现 - 基于向量存储
"""
from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional
from datetime import datetime
import json
import hashlib
@dataclass
class MemoryEntry:
"""记忆条目"""
id: str
content: str
embedding: List[float] = field(default_factory=list)
metadata: Dict[str, Any] = field(default_factory=dict)
created_at: datetime = field(default_factory=datetime.now)
access_count: int = 0
importance: float = 1.0
class SimpleVectorStore:
"""简单的向量存储实现（演示用，生产环境应使用FAISS/Chroma等）"""
def __init__(self):
self.vectors: Dict[str, List[float]] = {}
self.texts: Dict[str, str] = {}
def add(self, id: str, text: str, embedding: List[float]):
"""添加向量"""
self.vectors[id] = embedding
self.texts[id] = text
def search(self, query_embedding: List[float], top_k: int = 5) -> List[tuple]:
"""搜索最相似的向量"""
similarities = []
for id, vector in self.vectors.items():
similarity = self._cosine_similarity(query_embedding, vector)
similarities.append((id, similarity, self.texts[id]))
similarities.sort(key=lambda x: x[1], reverse=True)
return similarities[:top_k]
def _cosine_similarity(self, a: List[float], b: List[float]) -> float:
"""计算余弦相似度"""
import math
dot_product = sum(x * y for x, y in zip(a, b))
norm_a = math.sqrt(sum(x * x for x in a))
norm_b = math.sqrt(sum(x * x for x in b))
if norm_a == 0 or norm_b == 0:
return 0
return dot_product / (norm_a * norm_b)
class LongTermMemory:
"""长期记忆管理器"""
def __init__(self, embedding_func=None):
self.vector_store = SimpleVectorStore()
self.entries: Dict[str, MemoryEntry] = {}
self.embedding_func = embedding_func or self._simple_embedding
def _simple_embedding(self, text: str) -> List[float]:
"""简单的文本嵌入（演示用）"""
hash_obj = hashlib.md5(text.encode())
hash_bytes = hash_obj.digest()
return [float(b) / 255.0 for b in hash_bytes]
def _generate_id(self, content: str) -> str:
"""生成唯一ID"""
return hashlib.md5(content.encode()).hexdigest()[:12]
def store(self, content: str, metadata: Dict[str, Any] = None) -> str:
"""存储记忆"""
entry_id = self._generate_id(content)
embedding = self.embedding_func(content)
entry = MemoryEntry(
id=entry_id,
content=content,
embedding=embedding,
metadata=metadata or {}
)
self.entries[entry_id] = entry
self.vector_store.add(entry_id, content, embedding)
return entry_id
def recall(self, query: str, top_k: int = 5) -> List[MemoryEntry]:
"""根据查询召回相关记忆"""
query_embedding = self.embedding_func(query)
results = self.vector_store.search(query_embedding, top_k)
recalled = []
for id, similarity, text in results:
if id in self.entries:
entry = self.entries[id]
entry.access_count += 1
recalled.append(entry)
return recalled
def forget(self, entry_id: str):
"""删除记忆"""
if entry_id in self.entries:
del self.entries[entry_id]
def get_important_memories(self, top_k: int = 10) -> List[MemoryEntry]:
"""获取最重要的记忆"""
sorted_entries = sorted(
self.entries.values(),
key=lambda e: e.importance * (1 + e.access_count),
reverse=True
)
return sorted_entries[:top_k]
def export_memories(self) -> str:
"""导出所有记忆为JSON"""
data = []
for entry in self.entries.values():
data.append({
"id": entry.id,
"content": entry.content,
"metadata": entry.metadata,
"created_at": entry.created_at.isoformat(),
"access_count": entry.access_count,
"importance": entry.importance
})
return json.dumps(data, ensure_ascii=False, indent=2)
# 使用示例
ltm = LongTermMemory()
# 存储记忆
ltm.store("用户喜欢用Python编程", {"category": "preferences"})
ltm.store("用户是一名数据科学家", {"category": "user_info"})
ltm.store("上次讨论了机器学习的基础知识", {"category": "conversation"})
# 召回记忆
results = ltm.recall("Python编程")
for entry in results:
print(f"[{entry.metadata.get('category')}] {entry.content}")
6.3 对话窗口管理
对话窗口管理是控制发送给LLM的上下文长度的关键技术：
"""
对话窗口管理策略
"""
from typing import List, Dict
from dataclasses import dataclass
@dataclass
class ConversationWindow:
"""对话窗口配置"""
max_tokens: int = 4000
system_prompt_tokens: int = 500
reserved_tokens: int = 500 # 为模型回复预留
class WindowStrategy:
"""对话窗口管理策略"""
@staticmethod
def sliding_window(
messages: List[Dict],
max_messages: int = 20
) -> List[Dict]:
"""滑动窗口策略：保留最近的消息"""
return messages[-max_messages:]
@staticmethod
def summarize_older(
messages: List[Dict],
max_messages: int = 10,
summarizer=None
) -> List[Dict]:
"""总结旧消息策略：将较早的消息总结为一条"""
if len(messages) <= max_messages:
return messages
older_messages = messages[:-max_messages]
recent_messages = messages[-max_messages:]
if summarizer:
summary = summarizer(older_messages)
else:
summary = f"[之前对话摘要：共{len(older_messages)}条消息]"
return [{"role": "system", "content": summary}] + recent_messages
@staticmethod
def importance_based(
messages: List[Dict],
max_tokens: int = 4000
) -> List[Dict]:
"""基于重要性的策略：保留系统消息和重要消息"""
system_msgs = [m for m in messages if m["role"] == "system"]
other_msgs = [m for m in messages if m["role"] != "system"]
def estimate_tokens(msgs):
return sum(len(m["content"]) * 2 for m in msgs)
result = system_msgs.copy()
remaining_tokens = max_tokens - estimate_tokens(result)
for msg in reversed(other_msgs):
msg_tokens = len(msg["content"]) * 2
if msg_tokens <= remaining_tokens:
result.insert(-len(system_msgs) if system_msgs else 0, msg)
remaining_tokens -= msg_tokens
else:
break
return result
class ConversationManager:
"""对话管理器"""
def __init__(self, strategy: str = "sliding", max_tokens: int = 4000):
self.messages: List[Dict] = []
self.strategy = strategy
self.max_tokens = max_tokens
def add_message(self, role: str, content: str):
"""添加消息"""
self.messages.append({"role": role, "content": content})
def get_context(self) -> List[Dict]:
"""获取优化后的上下文"""
if self.strategy == "sliding":
return WindowStrategy.sliding_window(self.messages, max_messages=20)
elif self.strategy == "summarize":
return WindowStrategy.summarize_older(self.messages, max_messages=10)
elif self.strategy == "importance":
return WindowStrategy.importance_based(self.messages, max_tokens=self.max_tokens)
else:
return self.messages
def get_token_count(self) -> int:
"""估算当前token数"""
return sum(len(m["content"]) * 2 for m in self.messages)
# 使用示例
manager = ConversationManager(strategy="importance", max_tokens=2000)
manager.add_message("system", "你是一个专业的Python编程助手")
manager.add_message("user", "什么是装饰器？")
manager.add_message("assistant", "装饰器是Python中的一种设计模式...")
manager.add_message("user", "能给个例子吗？")
manager.add_message("assistant", "当然！这是一个简单的装饰器示例...")
context = manager.get_context()
print(f"上下文包含 {len(context)} 条消息")
第七章：Agent规划能力
7.1 任务分解
任务分解是Agent处理复杂问题的关键能力。Agent需要将一个大任务拆分成多个可执行的子任务。
"""
任务分解器实现
"""
from dataclasses import dataclass, field
from typing import List, Optional, Dict, Any
from enum import Enum
import json
class TaskStatus(Enum):
PENDING = "待执行"
IN_PROGRESS = "执行中"
COMPLETED = "已完成"
FAILED = "失败"
BLOCKED = "阻塞"
@dataclass
class SubTask:
"""子任务"""
id: str
title: str
description: str
dependencies: List[str] = field(default_factory=list)
status: TaskStatus = TaskStatus.PENDING
result: Optional[str] = None
estimated_complexity: int = 1 # 1-5
@dataclass
class TaskPlan:
"""任务计划"""
goal: str
subtasks: List[SubTask] = field(default_factory=list)
def get_ready_tasks(self) -> List[SubTask]:
"""获取可以执行的任务（依赖已完成）"""
completed_ids = {t.id for t in self.subtasks if t.status == TaskStatus.COMPLETED}
ready = []
for task in self.subtasks:
if task.status == TaskStatus.PENDING:
if all(dep in completed_ids for dep in task.dependencies):
ready.append(task)
return ready
def mark_complete(self, task_id: str, result: str):
"""标记任务完成"""
for task in self.subtasks:
if task.id == task_id:
task.status = TaskStatus.COMPLETED
task.result = result
break
def is_complete(self) -> bool:
"""检查是否所有任务都已完成"""
return all(t.status == TaskStatus.COMPLETED for t in self.subtasks)
def get_progress(self) -> Dict[str, int]:
"""获取进度统计"""
stats = {}
for status in TaskStatus:
stats[status.value] = sum(1 for t in self.subtasks if t.status == status)
return stats
class TaskDecomposer:
"""任务分解器"""
def __init__(self, llm_client):
self.llm = llm_client
def decompose(self, goal: str) -> TaskPlan:
"""将目标分解为子任务"""
prompt = f"""
请将以下目标分解为具体的执行步骤：
目标：{goal}
要求：
1. 每个步骤应该是可独立执行的
2. 明确步骤之间的依赖关系
3. 估计每个步骤的复杂度（1-5）
请以JSON格式输出：
{{
"subtasks": [
{{
"id": "task_1",
"title": "步骤标题",
"description": "详细描述",
"dependencies": [],
"complexity": 1
}},
...
]
}}
"""
response = self.llm.generate(prompt)
plan_data = json.loads(response)
plan = TaskPlan(goal=goal)
for task_data in plan_data["subtasks"]:
subtask = SubTask(
id=task_data["id"],
title=task_data["title"],
description=task_data["description"],
dependencies=task_data.get("dependencies", []),
estimated_complexity=task_data.get("complexity", 1)
)
plan.subtasks.append(subtask)
return plan
# 使用示例
"""
假设用户请求："帮我写一个博客网站，包括用户注册、文章发布、评论功能"
分解结果可能是：
task_1: 搭建项目基础框架 (无依赖)
task_2: 设计数据库模型 (依赖: task_1)
task_3: 实现用户注册功能 (依赖: task_2)
task_4: 实现文章发布功能 (依赖: task_2)
task_5: 实现评论功能 (依赖: task_2, task_4)
task_6: 前端页面开发 (依赖: task_3, task_4, task_5)
task_7: 集成测试 (依赖: task_6)
"""
7.2 多步推理
多步推理是Agent解决复杂问题时的核心能力：
"""
多步推理实现
"""
from typing import List, Dict, Optional
from dataclasses import dataclass
@dataclass
class ReasoningStep:
"""推理步骤"""
thought: str
action: Optional[str]
observation: Optional[str]
conclusion: Optional[str]
class MultiStepReasoner:
"""多步推理器"""
def __init__(self, llm_client, tools: Dict = None):
self.llm = llm_client
self.tools = tools or {}
self.reasoning_chain: List[ReasoningStep] = []
def reason(self, question: str, max_steps: int = 10) -> str:
"""进行多步推理"""
context = f"问题：{question}\n\n"
for step_num in range(max_steps):
step_prompt = f"""
{context}
当前推理进度：第{step_num + 1}步
请继续推理。如果已经有足够信息回答问题，请给出最终答案。
否则，请：
1. 思考下一步需要做什么
2. 如果需要，选择一个工具来获取信息
3. 基于信息得出中间结论
格式：
思考：[你的分析]
行动：[需要调用的工具，或"无需行动"]
输入：[工具输入，或空]
"""
response = self.llm.generate(step_prompt)
step = self._parse_reasoning_step(response)
self.reasoning_chain.append(step)
context += f"\n步骤{step_num + 1}：\n"
context += f" 思考：{step.thought}\n"
if step.action and step.action != "无需行动":
observation = self._execute_tool(step.action, step.observation)
step.observation = observation
context += f" 行动：{step.action}\n"
context += f" 观察：{observation}\n"
if step.conclusion:
context += f" 结论：{step.conclusion}\n"
if "最终答案" in response or "最终结论" in response:
return self._extract_final_answer(response)
return "推理过程未能得出确定结论"
def _parse_reasoning_step(self, response: str) -> ReasoningStep:
"""解析推理步骤"""
thought = ""
action = None
observation = None
conclusion = None
lines = response.split("\n")
for line in lines:
if line.startswith("思考：") or line.startswith("Thought:"):
thought = line.split("：", 1)[-1].strip()
elif line.startswith("行动：") or line.startswith("Action:"):
action = line.split("：", 1)[-1].strip()
elif line.startswith("输入：") or line.startswith("Input:"):
observation = line.split("：", 1)[-1].strip()
elif line.startswith("结论：") or line.startswith("Conclusion:"):
conclusion = line.split("：", 1)[-1].strip()
return ReasoningStep(
thought=thought,
action=action,
observation=observation,
conclusion=conclusion
)
def _execute_tool(self, tool_name: str, tool_input: str) -> str:
"""执行工具"""
if tool_name in self.tools:
return self.tools[tool_name](tool_input)
return f"未知工具：{tool_name}"
def _extract_final_answer(self, response: str) -> str:
"""提取最终答案"""
for marker in ["最终答案：", "最终答案:", "Final Answer:"]:
if marker in response:
return response.split(marker, 1)[1].strip()
return response
def get_reasoning_summary(self) -> str:
"""获取推理过程摘要"""
summary = "推理过程：\n"
for i, step in enumerate(self.reasoning_chain, 1):
summary += f"\n步骤{i}：\n"
summary += f" 思考：{step.thought}\n"
if step.observation:
summary += f" 观察：{step.observation}\n"
if step.conclusion:
summary += f" 结论：{step.conclusion}\n"
return summary
# 使用示例
"""
问题：小明比小红大3岁，小红比小华大2岁，小华10岁，小明几岁？
推理过程：
步骤1：
思考：已知小华10岁，小红比小华大2岁
结论：小红 = 10 + 2 = 12岁
步骤2：
思考：已知小红12岁，小明比小红大3岁
结论：小明 = 12 + 3 = 15岁
步骤3：
思考：我已经得出所有需要的信息
最终答案：小明15岁
"""
7.3 自我反思与纠错
自我反思是高级Agent的重要能力：
"""
自我反思与纠错机制
"""
from typing import List, Dict, Optional
from dataclasses import dataclass
import json
@dataclass
class ReflectionResult:
"""反思结果"""
has_errors: bool
errors_found: List[str]
corrections: List[str]
confidence: float # 0-1
class SelfReflectionAgent:
"""具有自我反思能力的Agent"""
def __init__(self, llm_client):
self.llm = llm_client
self.max_reflections = 3
def solve_with_reflection(self, problem: str) -> str:
"""带反思的问题求解"""
solution = None
for attempt in range(self.max_reflections):
# 第一步：尝试解决问题
if attempt == 0:
solution = self._attempt_solve(problem, None)
else:
solution = self._attempt_solve(problem, reflection)
# 第二步：反思解答
reflection = self._reflect(problem, solution)
if not reflection.has_errors:
return solution
print(f"发现错误（第{attempt + 1}次反思）：")
for error in reflection.errors_found:
print(f" - {error}")
return solution
def _attempt_solve(self, problem: str, previous_reflection: Optional[ReflectionResult]) -> str:
"""尝试解决问题"""
prompt = f"""
问题：{problem}
"""
if previous_reflection:
prompt += f"""
上一次解答中的错误：
{chr(10).join('- ' + e for e in previous_reflection.errors_found)}
需要修正的地方：
{chr(10).join('- ' + c for c in previous_reflection.corrections)}
请根据上述反馈重新解答。
"""
prompt += "\n请详细展示你的解题过程："
return self.llm.generate(prompt)
def _reflect(self, problem: str, solution: str) -> ReflectionResult:
"""反思解答"""
prompt = f"""
问题：{problem}
解答：{solution}
请仔细检查这个解答：
1. 计算是否正确？
2. 逻辑是否合理？
3. 是否有遗漏？
4. 假设是否正确？
如果有错误，请列出具体的错误和修正建议。
请以JSON格式输出：
{{
"has_errors": true/false,
"errors": ["错误1", "错误2"],
"corrections": ["修正1", "修正2"],
"confidence": 0.95
}}
"""
response = self.llm.generate(prompt)
try:
data = json.loads(response)
return ReflectionResult(
has_errors=data.get("has_errors", False),
errors_found=data.get("errors", []),
corrections=data.get("corrections", []),
confidence=data.get("confidence", 0.5)
)
except:
return ReflectionResult(
has_errors=False,
errors_found=[],
corrections=[],
confidence=0.5
)
# 使用示例
"""
问题：一个水池有两个水管，A管每小时注入3吨水，B管每小时排出1吨水。
水池容量20吨，从空池开始，多少小时能注满？
第一次尝试（可能出错）：
解：20 / 3 = 6.67小时
反思：
错误：忽略了B管在排水
修正：应该用净注入速度 = 3 - 1 = 2吨/小时
第二次尝试（修正后）：
解：20 / (3 - 1) = 20 / 2 = 10小时
反思：计算正确 ✓
"""
第八章：多Agent协作
8.1 主从模式
主从模式是一种常见的多Agent协作架构，其中一个Agent作为"主"负责协调，其他Agent作为"从"负责执行具体任务。
"""
主从模式的多Agent协作实现
"""
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
from enum import Enum
import json
class AgentCapability(Enum):
"""Agent能力类型"""
RESEARCH = "研究分析"
WRITING = "内容写作"
CODING = "代码编写"
REVIEW = "审核评审"
TRANSLATION = "翻译"
@dataclass
class WorkerAgent:
"""工作者Agent（从）"""
name: str
capabilities: List[AgentCapability]
llm_client: Any
def execute_task(self, task: str) -> str:
"""执行分配的任务"""
prompt = f"""
你是一个专门负责{[c.value for c in self.capabilities]}的助手。
请完成以下任务：
{task}
"""
return self.llm_client.generate(prompt)
class SupervisorAgent:
"""监督者Agent（主）"""
def __init__(self, llm_client, workers: List[WorkerAgent]):
self.llm = llm_client
self.workers = {w.name: w for w in workers}
self.task_queue = []
self.results = {}
def analyze_task(self, main_task: str) -> Dict[str, str]:
"""分析主任务，分解为子任务并分配给合适的工作者"""
prompt = f"""
主任务：{main_task}
可用的工作者：
{self._describe_workers()}
请将主任务分解为子任务，并分配给最合适的工作者。
请以JSON格式输出：
{{
"subtasks": [
{{
"worker": "工作者名称",
"task": "具体任务描述"
}}
]
}}
"""
response = self.llm.generate(prompt)
return json.loads(response)
def _describe_workers(self) -> str:
"""描述所有工作者的能力"""
descriptions = []
for name, worker in self.workers.items():
caps = ", ".join(c.value for c in worker.capabilities)
descriptions.append(f"- {name}：擅长{caps}")
return "\n".join(descriptions)
def execute(self, main_task: str) -> str:
"""执行主任务"""
print(f"主管收到任务：{main_task}\n")
plan = self.analyze_task(main_task)
print("任务分配计划：")
for subtask in plan["subtasks"]:
print(f" → {subtask['worker']}：{subtask['task']}")
print()
for subtask in plan["subtasks"]:
worker_name = subtask["worker"]
task_desc = subtask["task"]
if worker_name in self.workers:
print(f"[{worker_name}] 开始执行...")
result = self.workers[worker_name].execute_task(task_desc)
self.results[worker_name] = result
print(f"[{worker_name}] 完成！")
else:
print(f"警告：找不到工作者 '{worker_name}'")
print("\n主管正在汇总结果...")
summary = self._summarize_results(main_task)
return summary
def _summarize_results(self, main_task: str) -> str:
"""汇总所有工作者的结果"""
prompt = f"""
主任务：{main_task}
各工作者的执行结果：
{json.dumps(self.results, ensure_ascii=False, indent=2)}
请汇总以上结果，生成一份完整的最终报告。
"""
return self.llm.generate(prompt)
# 使用示例：创建一个研究写作团队
researcher = WorkerAgent(
name="研究员",
capabilities=[AgentCapability.RESEARCH],
llm_client=your_llm
)
writer = WorkerAgent(
name="写手",
capabilities=[AgentCapability.WRITING],
llm_client=your_llm
)
reviewer = WorkerAgent(
name="审稿人",
capabilities=[AgentCapability.REVIEW],
llm_client=your_llm
)
supervisor = SupervisorAgent(
llm_client=your_llm,
workers=[researcher, writer, reviewer]
)
result = supervisor.execute("写一篇关于AI Agent技术发展的研究报告")
print(result)
8.2 对等协作模式
对等协作模式中，所有Agent地位平等，通过协商和讨论来完成任务：
"""
对等协作模式的多Agent实现
"""
from typing import List, Dict
from dataclasses import dataclass, field
@dataclass
class PeerAgent:
"""对等Agent"""
name: str
expertise: str
llm_client: any
position: str = ""
class PeerDiscussion:
"""对等讨论"""
def __init__(self, agents: List[PeerAgent], topic: str, max_rounds: int = 3):
self.agents = agents
self.topic = topic
self.max_rounds = max_rounds
self.discussion_log: List[Dict] = []
def run(self) -> str:
"""运行讨论"""
print(f"讨论主题：{self.topic}\n")
print("=" * 50)
for round_num in range(self.max_rounds):
print(f"\n--- 第{round_num + 1}轮讨论 ---\n")
for agent in self.agents:
statement = self._generate_statement(agent, round_num)
agent.position = statement
self.discussion_log.append({
"round": round_num + 1,
"agent": agent.name,
"statement": statement
})
print(f"[{agent.name}]：{statement}\n")
if self._check_consensus():
print("达成共识！")
break
return self._generate_consensus()
def _generate_statement(self, agent: PeerAgent, round_num: int) -> str:
"""生成Agent的发言"""
others_positions = [
f"- {a.name}: {a.position}"
for a in self.agents
if a != agent and a.position
]
prompt = f"""
讨论主题：{self.topic}
你的专业领域：{agent.expertise}
{"其他参与者的观点：" + chr(10) + chr(10).join(others_positions) if others_positions else "这是第一轮讨论，请先发表你的观点。"}
请从你的专业角度发表看法。
"""
return agent.llm_client.generate(prompt)
def _check_consensus(self) -> bool:
"""检查是否达成共识"""
if len(self.discussion_log) < len(self.agents):
return False
last_statements = self.discussion_log[-len(self.agents):]
consensus_keywords = ["同意", "赞同", "consensus", "一致"]
return all(
any(kw in s["statement"] for kw in consensus_keywords)
for s in last_statements
)
def _generate_consensus(self) -> str:
"""生成共识总结"""
all_statements = [
f"[{s['agent']}] (第{s['round']}轮): {s['statement']}"
for s in self.discussion_log
]
prompt = f"""
讨论主题：{self.topic}
讨论记录：
{chr(10).join(all_statements)}
请总结讨论的共识和主要观点。
"""
return self.agents[0].llm_client.generate(prompt)
# 使用示例：技术选型讨论
tech_expert = PeerAgent(
name="技术专家",
expertise="软件架构和技术实现",
llm_client=your_llm
)
product_expert = PeerAgent(
name="产品专家",
expertise="用户需求和产品设计",
llm_client=your_llm
)
business_expert = PeerAgent(
name="商业专家",
expertise="商业模式和成本控制",
llm_client=your_llm
)
discussion = PeerDiscussion(
agents=[tech_expert, product_expert, business_expert],
topic="我们应该使用微服务架构还是单体架构来构建新系统？",
max_rounds=3
)
result = discussion.run()
print(f"\n最终共识：{result}")
8.3 任务分配策略
有效的任务分配是多Agent协作成功的关键：
"""
任务分配策略实现
"""
from typing import List, Dict, Callable
from dataclasses import dataclass
@dataclass
class AgentProfile:
"""Agent能力画像"""
name: str
skills: List[str]
current_load: int = 0
max_load: int = 5
success_rate: float = 1.0
class TaskAllocationStrategy:
"""任务分配策略"""
@staticmethod
def round_robin(agents: List[AgentProfile], tasks: List[str]) -> Dict[str, List[str]]:
"""轮询分配：依次分配给每个Agent"""
allocation = {agent.name: [] for agent in agents}
for i, task in enumerate(tasks):
agent = agents[i % len(agents)]
allocation[agent.name].append(task)
return allocation
@staticmethod
def skill_based(
agents: List[AgentProfile],
tasks: List[str],
task_analyzer: Callable
) -> Dict[str, List[str]]:
"""基于技能的分配：根据任务内容匹配最合适的Agent"""
allocation = {agent.name: [] for agent in agents}
for task in tasks:
required_skills = task_analyzer(task)
best_agent = None
best_score = -1
for agent in agents:
score = len(set(required_skills) & set(agent.skills))
load_factor = 1 - (agent.current_load / agent.max_load)
final_score = score * load_factor
if final_score > best_score:
best_score = final_score
best_agent = agent
if best_agent:
allocation[best_agent.name].append(task)
best_agent.current_load += 1
return allocation
@staticmethod
def load_balanced(agents: List[AgentProfile], tasks: List[str]) -> Dict[str, List[str]]:
"""负载均衡分配：优先分配给任务最少的Agent"""
allocation = {agent.name: [] for agent in agents}
sorted_agents = sorted(agents, key=lambda a: a.current_load)
for task in tasks:
target = sorted_agents[0]
allocation[target.name].append(task)
target.current_load += 1
sorted_agents.sort(key=lambda a: a.current_load)
return allocation
class TaskDispatcher:
"""任务调度器"""
def __init__(self, agents: List[AgentProfile]):
self.agents = agents
self.strategy = TaskAllocationStrategy()
def dispatch(
self,
tasks: List[str],
strategy: str = "skill_based",
task_analyzer: Callable = None
) -> Dict[str, List[str]]:
"""分配任务"""
if strategy == "round_robin":
return self.strategy.round_robin(self.agents, tasks)
elif strategy == "skill_based":
if not task_analyzer:
raise ValueError("skill_based策略需要提供task_analyzer")
return self.strategy.skill_based(self.agents, tasks, task_analyzer)
elif strategy == "load_balanced":
return self.strategy.load_balanced(self.agents, tasks)
else:
raise ValueError(f"未知策略：{strategy}")
# 使用示例
agents = [
AgentProfile(name="Python专家", skills=["python", "数据分析", "机器学习"]),
AgentProfile(name="前端专家", skills=["javascript", "react", "css"]),
AgentProfile(name="全栈工程师", skills=["python", "javascript", "数据库"]),
]
tasks = [
"用Python写一个数据处理脚本",
"设计一个React组件",
"优化数据库查询性能",
"实现一个机器学习模型",
"搭建REST API接口",
]
dispatcher = TaskDispatcher(agents)
allocation = dispatcher.dispatch(tasks, strategy="load_balanced")
for agent_name, agent_tasks in allocation.items():
print(f"\n{agent_name} 分配到的任务：")
for task in agent_tasks:
print(f" - {task}")
第九章：生产部署
9.1 安全沙箱
在生产环境中，Agent执行代码或调用外部工具时，必须在安全沙箱中运行：
"""
安全沙箱实现
"""
import subprocess
import tempfile
import os
from typing import Optional
class CodeSandbox:
"""代码执行沙箱"""
def __init__(
self,
timeout: int = 30,
max_memory_mb: int = 256,
allowed_modules: list = None
):
self.timeout = timeout
self.max_memory_mb = max_memory_mb
self.allowed_modules = allowed_modules or [
"math", "random", "datetime", "json", "re",
"collections", "itertools", "functools"
]
def execute(self, code: str, language: str = "python") -> dict:
"""在沙箱中执行代码"""
if language != "python":
return {"success": False, "error": f"不支持的语言：{language}"}
security_check = self._security_check(code)
if not security_check["safe"]:
return {"success": False, "error": security_check["reason"]}
with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
f.write(code)
temp_file = f.name
try:
result = subprocess.run(
["python3", temp_file],
capture_output=True,
text=True,
timeout=self.timeout,
env=self._create_restricted_env()
)
return {
"success": result.returncode == 0,
"stdout": result.stdout,
"stderr": result.stderr,
"return_code": result.returncode
}
except subprocess.TimeoutExpired:
return {"success": False, "error": f"执行超时（{self.timeout}秒）"}
except Exception as e:
return {"success": False, "error": str(e)}
finally:
os.unlink(temp_file)
def _security_check(self, code: str) -> dict:
"""代码安全检查"""
dangerous_patterns = [
"import os", "import sys", "import subprocess",
"import shutil", "__import__", "eval(", "exec(",
"open(", "os.system", "os.popen", "subprocess",
"shutil.rmtree", "os.remove", "os.unlink",
]
for pattern in dangerous_patterns:
if pattern in code:
return {
"safe": False,
"reason": f"检测到危险操作：{pattern}"
}
return {"safe": True}
def _create_restricted_env(self) -> dict:
"""创建受限的环境变量"""
env = os.environ.copy()
sensitive_keys = [
"AWS_SECRET_ACCESS_KEY", "API_KEY", "DATABASE_PASSWORD",
"SECRET_KEY", "TOKEN"
]
for key in sensitive_keys:
env.pop(key, None)
return env
# 使用示例
sandbox = CodeSandbox(timeout=10, max_memory_mb=128)
# 安全的代码
result = sandbox.execute("""
import math
print(f"圆周率：{math.pi}")
print(f"sqrt(2) = {math.sqrt(2)}")
""")
print(result)
# 不安全的代码（会被拦截）
result = sandbox.execute("""
import os
os.system("rm -rf /")
""")
print(result) # 会显示安全错误
9.2 速率限制
速率限制是防止Agent被滥用的重要机制：
"""
速率限制实现
"""
import time
from collections import defaultdict
from threading import Lock
from typing import Dict, Optional
from dataclasses import dataclass
@dataclass
class RateLimitConfig:
"""速率限制配置"""
requests_per_minute: int = 60
requests_per_hour: int = 1000
tokens_per_minute: int = 100000
burst_size: int = 10
class TokenBucket:
"""令牌桶算法实现"""
def __init__(self, rate: float, capacity: int):
self.rate = rate
self.capacity = capacity
self.tokens = capacity
self.last_time = time.time()
self.lock = Lock()
def consume(self, tokens: int = 1) -> bool:
"""尝试消费令牌"""
with self.lock:
now = time.time()
elapsed = now - self.last_time
self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
self.last_time = now
if self.tokens >= tokens:
self.tokens -= tokens
return True
return False
def wait_time(self, tokens: int = 1) -> float:
"""计算需要等待的时间"""
with self.lock:
if self.tokens >= tokens:
return 0
return (tokens - self.tokens) / self.rate
class RateLimiter:
"""速率限制器"""
def __init__(self, config: RateLimitConfig):
self.config = config
self.minute_bucket = TokenBucket(
rate=config.requests_per_minute / 60,
capacity=config.burst_size
)
self.hour_bucket = TokenBucket(
rate=config.requests_per_hour / 3600,
capacity=config.requests_per_hour
)
self.user_buckets: Dict[str, TokenBucket] = {}
self.lock = Lock()
def get_user_bucket(self, user_id: str) -> TokenBucket:
"""获取用户的令牌桶"""
with self.lock:
if user_id not in self.user_buckets:
self.user_buckets[user_id] = TokenBucket(
rate=self.config.requests_per_minute / 60,
capacity=self.config.burst_size
)
return self.user_buckets[user_id]
def check(self, user_id: str, tokens: int = 1) -> dict:
"""检查是否允许请求"""
if not self.minute_bucket.consume(tokens):
wait = self.minute_bucket.wait_time(tokens)
return {
"allowed": False,
"reason": "超出每分钟请求限制",
"retry_after": wait
}
if not self.hour_bucket.consume(tokens):
wait = self.hour_bucket.wait_time(tokens)
return {
"allowed": False,
"reason": "超出每小时请求限制",
"retry_after": wait
}
user_bucket = self.get_user_bucket(user_id)
if not user_bucket.consume(tokens):
wait = user_bucket.wait_time(tokens)
return {
"allowed": False,
"reason": "用户请求过于频繁",
"retry_after": wait
}
return {"allowed": True}
# 使用示例
config = RateLimitConfig(
requests_per_minute=60,
requests_per_hour=1000,
burst_size=10
)
limiter = RateLimiter(config)
for i in range(100):
result = limiter.check(user_id="user_123")
if result["allowed"]:
print(f"请求{i+1}: 允许")
else:
print(f"请求{i+1}: 拒绝 - {result['reason']}")
time.sleep(result["retry_after"])
9.3 监控日志
完善的监控和日志系统是生产级Agent的必备组件：
"""
Agent监控和日志系统
"""
import logging
import json
from datetime import datetime
from typing import Dict, Any, Optional
from dataclasses import dataclass, field
from collections import defaultdict
import time
@dataclass
class AgentMetrics:
"""Agent性能指标"""
total_requests: int = 0
successful_requests: int = 0
failed_requests: int = 0
total_tokens_used: int = 0
total_latency_ms: float = 0
tool_calls: Dict[str, int] = field(default_factory=lambda: defaultdict(int))
@property
def success_rate(self) -> float:
if self.total_requests == 0:
return 0
return self.successful_requests / self.total_requests
@property
def avg_latency_ms(self) -> float:
if self.total_requests == 0:
return 0
return self.total_latency_ms / self.total_requests
class AgentLogger:
"""Agent专用日志器"""
def __init__(self, log_file: str = "agent.log"):
self.logger = logging.getLogger("agent")
self.logger.setLevel(logging.DEBUG)
fh = logging.FileHandler(log_file, encoding='utf-8')
fh.setLevel(logging.DEBUG)
ch = logging.StreamHandler()
ch.setLevel(logging.INFO)
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
fh.setFormatter(formatter)
ch.setFormatter(formatter)
self.logger.addHandler(fh)
self.logger.addHandler(ch)
self.metrics = AgentMetrics()
def log_request(self, user_id: str, query: str):
"""记录请求"""
self.logger.info(f"请求 - 用户:{user_id} - 内容:{query[:100]}")
self.metrics.total_requests += 1
def log_tool_call(self, tool_name: str, input_data: Any, output_data: Any, latency_ms: float):
"""记录工具调用"""
self.logger.debug(
f"工具调用 - {tool_name} - 输入:{json.dumps(input_data, ensure_ascii=False)[:200]} - "
f"输出:{json.dumps(output_data, ensure_ascii=False)[:200]} - 耗时:{latency_ms:.2f}ms"
)
self.metrics.tool_calls[tool_name] += 1
def log_response(self, user_id: str, response: str, latency_ms: float, tokens_used: int):
"""记录响应"""
self.logger.info(
f"响应 - 用户:{user_id} - 长度:{len(response)} - "
f"耗时:{latency_ms:.2f}ms - tokens:{tokens_used}"
)
self.metrics.successful_requests += 1
self.metrics.total_latency_ms += latency_ms
self.metrics.total_tokens_used += tokens_used
def log_error(self, error: Exception, context: Dict = None):
"""记录错误"""
self.logger.error(
f"错误 - {type(error).__name__}: {str(error)} - 上下文:{context}"
)
self.metrics.failed_requests += 1
def get_metrics_summary(self) -> str:
"""获取指标摘要"""
return json.dumps({
"总请求数": self.metrics.total_requests,
"成功率": f"{self.metrics.success_rate:.2%}",
"平均延迟": f"{self.metrics.avg_latency_ms:.2f}ms",
"总tokens": self.metrics.total_tokens_used,
"工具调用统计": dict(self.metrics.tool_calls)
}, ensure_ascii=False, indent=2)
class AgentMonitor:
"""Agent监控器"""
def __init__(self, logger: AgentLogger):
self.logger = logger
def check_health(self) -> Dict[str, Any]:
"""检查Agent健康状态"""
metrics = self.logger.metrics
health = {
"status": "healthy",
"issues": []
}
if metrics.total_requests > 10 and metrics.success_rate < 0.9:
health["status"] = "degraded"
health["issues"].append(f"成功率过低：{metrics.success_rate:.2%}")
if metrics.avg_latency_ms > 5000:
health["status"] = "degraded"
health["issues"].append(f"平均延迟过高：{metrics.avg_latency_ms:.0f}ms")
return health
def generate_report(self) -> str:
"""生成监控报告"""
health = self.check_health()
metrics = self.logger.get_metrics_summary()
report = f"""
=== Agent 监控报告 ===
生成时间：{datetime.now().isoformat()}
健康状态：{health['status']}
问题列表：{health['issues'] if health['issues'] else '无'}
性能指标：
{metrics}
========================
"""
return report
# 使用示例
logger = AgentLogger("my_agent.log")
monitor = AgentMonitor(logger)
logger.log_request("user_123", "今天天气怎么样？")
logger.log_tool_call("get_weather", {"city": "北京"}, {"temp": 25}, 150.5)
logger.log_response("user_123", "今天北京天气晴朗，温度25℃", 500.0, 150)
print(monitor.generate_report())
9.4 成本控制
AI Agent的运行成本（主要是LLM API调用费用）是生产部署时必须考虑的重要因素：
"""
成本控制和优化
"""
from dataclasses import dataclass
from typing import Dict, List
from datetime import datetime
import json
@dataclass
class ModelPricing:
"""模型定价（每1000 tokens的价格，单位：美元）"""
input_price: float
output_price: float
# 常见模型定价
MODEL_PRICING = {
"gpt-4": ModelPricing(input_price=0.03, output_price=0.06),
"gpt-4-turbo": ModelPricing(input_price=0.01, output_price=0.03),
"gpt-3.5-turbo": ModelPricing(input_price=0.0015, output_price=0.002),
"claude-3-opus": ModelPricing(input_price=0.015, output_price=0.075),
"claude-3-sonnet": ModelPricing(input_price=0.003, output_price=0.015),
"qwen-7b": ModelPricing(input_price=0.001, output_price=0.002),
}
class CostTracker:
"""成本追踪器"""
def __init__(self, daily_budget: float = 100.0):
self.daily_budget = daily_budget
self.usage_records: List[Dict] = []
def record_usage(
self,
model: str,
input_tokens: int,
output_tokens: int,
task_type: str = "general"
):
"""记录使用量"""
pricing = MODEL_PRICING.get(model, MODEL_PRICING["gpt-3.5-turbo"])
cost = (input_tokens * pricing.input_price + output_tokens * pricing.output_price) / 1000
record = {
"timestamp": datetime.now().isoformat(),
"model": model,
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"cost_usd": cost,
"task_type": task_type
}
self.usage_records.append(record)
return cost
def get_daily_cost(self, date: str = None) -> float:
"""获取某日的总成本"""
if date is None:
date = datetime.now().strftime("%Y-%m-%d")
return sum(
r["cost_usd"]
for r in self.usage_records
if r["timestamp"].startswith(date)
)
def check_budget(self) -> Dict:
"""检查预算使用情况"""
daily_cost = self.get_daily_cost()
remaining = self.daily_budget - daily_cost
return {
"daily_budget": self.daily_budget,
"daily_cost": daily_cost,
"remaining": remaining,
"usage_percentage": (daily_cost / self.daily_budget) * 100,
"is_over_budget": daily_cost > self.daily_budget
}
def get_optimization_suggestions(self) -> List[str]:
"""获取成本优化建议"""
suggestions = []
model_usage = {}
for record in self.usage_records:
model = record["model"]
if model not in model_usage:
model_usage[model] = {"count": 0, "cost": 0}
model_usage[model]["count"] += 1
model_usage[model]["cost"] += record["cost_usd"]
for model, usage in model_usage.items():
if model in ["gpt-4", "claude-3-opus"] and usage["count"] > 10:
suggestions.append(
f"考虑将部分'{model}'调用替换为更经济的模型，"
f"当前该模型调用{usage['count']}次，花费${usage['cost']:.2f}"
)
long_conversations = sum(
1 for r in self.usage_records
if r["input_tokens"] > 2000
)
if long_conversations > len(self.usage_records) * 0.3:
suggestions.append("30%以上的请求输入超过2000 tokens，建议优化上下文管理")
return suggestions
class CostOptimizer:
"""成本优化器"""
@staticmethod
def select_model_by_task(task_complexity: str, budget_remaining: float) -> str:
"""根据任务复杂度和预算选择模型"""
if budget_remaining < 1.0:
return "gpt-3.5-turbo"
if task_complexity == "simple":
return "gpt-3.5-turbo"
elif task_complexity == "medium":
return "gpt-4-turbo"
else:
return "gpt-4"
@staticmethod
def optimize_prompt(prompt: str, max_tokens: int = 2000) -> str:
"""优化提示词以减少token使用"""
optimized = " ".join(prompt.split())
if len(optimized) > max_tokens * 4:
optimized = optimized[:max_tokens * 4] + "..."
return optimized
# 使用示例
tracker = CostTracker(daily_budget=50.0)
tracker.record_usage("gpt-4", 500, 200, "question_answering")
tracker.record_usage("gpt-3.5-turbo", 1000, 500, "summarization")
tracker.record_usage("gpt-4", 2000, 800, "code_generation")
budget = tracker.check_budget()
print(json.dumps(budget, indent=2))
suggestions = tracker.get_optimization_suggestions()
for s in suggestions:
print(f"建议：{s}")
第十章：实战项目：构建智能客服Agent系统
10.1 系统架构设计
在这一章中，我们将从零开始构建一个完整的智能客服Agent系统。这个系统将综合运用前面学到的所有知识。
┌─────────────────────────────────────────────────────┐
│ 智能客服Agent系统 │
├─────────────────────────────────────────────────────┤
│ │
│ 用户输入 → 意图识别 → 任务路由 → Agent执行 → 响应生成 │
│ │
│ ┌───────────┐ ┌───────────┐ ┌───────────┐ │
│ │ 订单Agent │ │ 退款Agent │ │ 咨询Agent │ │
│ └─────┬─────┘ └─────┬─────┘ └─────┬─────┘ │
│ │ │ │ │
│ ┌─────┴──────────────┴──────────────┴─────┐ │
│ │ 工具层 │ │
│ │ 订单查询 │ 退款处理 │ 知识库 │ CRM系统 │ │
│ └──────────────────────────────────────────┘ │
│ │
│ ┌──────────────────────────────────────────┐ │
│ │ 记忆层 │ │
│ │ 对话历史 │ 用户画像 │ 知识库 │ 工单记录 │ │
│ └──────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────┘
10.2 核心代码实现
"""
智能客服Agent系统 - 完整实现
"""
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import json
import uuid
# ============================================
# 第1部分：数据模型定义
# ============================================
class Intent(Enum):
"""用户意图类型"""
ORDER_QUERY = "订单查询"
REFUND_REQUEST = "退款申请"
PRODUCT_CONSULT = "产品咨询"
COMPLAINT = "投诉建议"
GENERAL_INQUIRY = "一般咨询"
HUMAN_AGENT = "转人工"
class ConversationState(Enum):
"""对话状态"""
ACTIVE = "进行中"
WAITING = "等待用户"
RESOLVED = "已解决"
ESCALATED = "已转人工"
@dataclass
class Customer:
"""客户信息"""
customer_id: str
name: str
vip_level: int = 0
order_history: List[str] = field(default_factory=list)
preferences: Dict[str, Any] = field(default_factory=dict)
@dataclass
class Order:
"""订单信息"""
order_id: str
customer_id: str
products: List[Dict]
total_amount: float
status: str
created_at: str
@dataclass
class Conversation:
"""对话记录"""
conversation_id: str
customer_id: str
messages: List[Dict] = field(default_factory=list)
intent: Optional[Intent] = None
state: ConversationState = ConversationState.ACTIVE
created_at: str = field(default_factory=lambda: datetime.now().isoformat())
# ============================================
# 第2部分：工具定义
# ============================================
class OrderTool:
"""订单相关工具"""
orders_db = {
"ORD001": {
"order_id": "ORD001",
"customer_id": "CUST001",
"products": [{"name": "iPhone 15", "price": 7999, "quantity": 1}],
"total_amount": 7999,
"status": "已发货",
"tracking": "SF1234567890"
},
"ORD002": {
"order_id": "ORD002",
"customer_id": "CUST001",
"products": [{"name": "AirPods Pro", "price": 1899, "quantity": 1}],
"total_amount": 1899,
"status": "待发货"
}
}
@staticmethod
def query_order(order_id: str) -> dict:
"""查询订单状态"""
order = OrderTool.orders_db.get(order_id)
if order:
return {"success": True, "data": order}
return {"success": False, "error": "订单不存在"}
@staticmethod
def get_customer_orders(customer_id: str) -> list:
"""获取客户的所有订单"""
return [
order for order in OrderTool.orders_db.values()
if order["customer_id"] == customer_id
]
class RefundTool:
"""退款相关工具"""
@staticmethod
def create_refund(order_id: str, reason: str, amount: float = None) -> dict:
"""创建退款申请"""
refund_id = f"REF{uuid.uuid4().hex[:8].upper()}"
return {
"success": True,
"refund_id": refund_id,
"order_id": order_id,
"reason": reason,
"amount": amount,
"status": "审核中",
"estimated_days": 3
}
@staticmethod
def check_refund_status(refund_id: str) -> dict:
"""查询退款状态"""
return {
"refund_id": refund_id,
"status": "审核中",
"message": "您的退款申请正在审核中，预计1-3个工作日完成"
}
class KnowledgeBaseTool:
"""知识库工具"""
knowledge_base = {
"退货政策": "自收到商品之日起7天内可无理由退货，商品需保持原包装完好。",
"配送时间": "标准配送：3-5个工作日；加急配送：1-2个工作日。",
"会员权益": "普通会员：9.5折；银卡会员：9折；金卡会员：8.5折。",
"支付方式": "支持微信支付、支付宝、银行卡、花呗分期。",
"发票开具": "订单完成后可申请电子发票，在订单详情页点击'申请发票'即可。"
}
@staticmethod
def search_knowledge(query: str) -> dict:
"""搜索知识库"""
results = []
for topic, content in KnowledgeBaseTool.knowledge_base.items():
if any(keyword in query for keyword in topic):
results.append({"topic": topic, "content": content})
if not results:
for topic, content in KnowledgeBaseTool.knowledge_base.items():
results.append({"topic": topic, "content": content})
return {"results": results[:3]}
# ============================================
# 第3部分：意图识别
# ============================================
class IntentClassifier:
"""意图分类器"""
def __init__(self, llm_client=None):
self.llm = llm_client
self.keyword_rules = {
Intent.ORDER_QUERY: ["订单", "物流", "发货", "配送", "快递", "到哪了"],
Intent.REFUND_REQUEST: ["退款", "退货", "退钱", "不想要了", "质量问题"],
Intent.PRODUCT_CONSULT: ["产品", "商品", "功能", "规格", "参数", "价格"],
Intent.COMPLAINT: ["投诉", "不满", "差评", "服务差", "态度"],
Intent.HUMAN_AGENT: ["人工", "转人工", "客服", "真人"],
}
def classify(self, text: str) -> Intent:
"""识别用户意图"""
text_lower = text.lower()
for intent, keywords in self.keyword_rules.items():
if any(keyword in text_lower for keyword in keywords):
return intent
if self.llm:
return self._llm_classify(text)
return Intent.GENERAL_INQUIRY
def _llm_classify(self, text: str) -> Intent:
"""使用LLM分类"""
prompt = f"""
请判断以下用户消息的意图，从以下选项中选择：
- 订单查询
- 退款申请
- 产品咨询
- 投诉建议
- 一般咨询
- 转人工
用户消息：{text}
请只返回意图类别，不要添加其他内容。
"""
response = self.llm.generate(prompt)
for intent in Intent:
if intent.value in response:
return intent
return Intent.GENERAL_INQUIRY
# ============================================
# 第4部分：专业Agent定义
# ============================================
class BaseAgent:
"""Agent基类"""
def __init__(self, name: str, llm_client):
self.name = name
self.llm = llm_client
def process(self, conversation: Conversation, customer: Customer) -> str:
raise NotImplementedError
class OrderAgent(BaseAgent):
"""订单查询Agent"""
def process(self, conversation: Conversation, customer: Customer) -> str:
last_message = conversation.messages[-1]["content"] if conversation.messages else ""
import re
order_match = re.search(r'[A-Z]{3}\d{3,}', last_message)
if order_match:
order_id = order_match.group()
result = OrderTool.query_order(order_id)
if result["success"]:
order = result["data"]
response = f"订单 {order_id} 的信息如下：\n"
response += f"- 商品：{', '.join(p['name'] for p in order['products'])}\n"
response += f"- 金额：¥{order['total_amount']}\n"
response += f"- 状态：{order['status']}\n"
if "tracking" in order:
response += f"- 物流单号：{order['tracking']}\n"
return response
else:
return f"抱歉，未找到订单 {order_id}，请确认订单号是否正确。"
else:
orders = OrderTool.get_customer_orders(customer.customer_id)
if orders:
response = "您最近的订单：\n"
for order in orders:
response += f"- {order['order_id']}: {order['products'][0]['name']} ({order['status']})\n"
response += "\n请问您想查询哪个订单的详情？"
return response
else:
return "您暂时没有订单记录。"
class RefundAgent(BaseAgent):
"""退款Agent"""
def process(self, conversation: Conversation, customer: Customer) -> str:
last_message = conversation.messages[-1]["content"] if conversation.messages else ""
if any(keyword in last_message for keyword in ["退款", "退货", "退钱"]):
import re
order_match = re.search(r'[A-Z]{3}\d{3,}', last_message)
if order_match:
order_id = order_match.group()
result = RefundTool.create_refund(
order_id=order_id,
reason=last_message
)
if result["success"]:
return f"""
退款申请已提交：
- 退款单号：{result['refund_id']}
- 订单号：{result['order_id']}
- 状态：{result['status']}
- 预计处理时间：{result['estimated_days']}个工作日
我们会尽快处理您的退款申请，请耐心等待。
"""
else:
return "退款申请提交失败，请稍后重试或联系人工客服。"
else:
return "请提供您的订单号，以便我们为您处理退款申请。订单号格式通常为ORD开头。"
return "请描述您的退款需求，包括订单号和退款原因。"
class ConsultAgent(BaseAgent):
"""产品咨询Agent"""
def process(self, conversation: Conversation, customer: Customer) -> str:
last_message = conversation.messages[-1]["content"] if conversation.messages else ""
# 搜索知识库
result = KnowledgeBaseTool.search_knowledge(last_message)
knowledge = result.get("results", [])
if knowledge:
response = "根据您的问题，以下信息可能对您有帮助：\n\n"
for item in knowledge:
response += f"【{item['topic']}】\n{item['content']}\n\n"
response += "如果以上信息没有解答您的疑问，请详细描述您的问题。"
return response
# 如果有LLM，使用LLM生成回答
if self.llm:
prompt = f"""
你是一个专业的客服代表，请回答以下客户问题：
客户问题：{last_message}
客户VIP等级：{customer.vip_level}
请用友好、专业的语气回答。
"""
return self.llm.generate(prompt)
return "感谢您的咨询。您的问题我已记录，稍后会有专人回复您。"
class ComplaintAgent(BaseAgent):
"""投诉处理Agent"""
def process(self, conversation: Conversation, customer: Customer) -> str:
last_message = conversation.messages[-1]["content"] if conversation.messages else ""
response = f"非常抱歉给您带来了不好的体验，{customer.name}。\n\n"
response += "我已经记录了您的投诉：\n"
response += f"投诉内容：{last_message[:100]}...\n\n"
response += "我们会认真对待您的反馈，并在24小时内给您回复。\n"
response += "如果您需要立即处理，我可以帮您转接人工客服。"
return response
# ============================================
# 第5部分：对话记忆管理
# ============================================
class ConversationMemory:
"""对话记忆管理"""
def __init__(self, max_history: int = 20):
self.conversations: Dict[str, Conversation] = {}
self.max_history = max_history
def get_or_create(self, conversation_id: str, customer_id: str) -> Conversation:
"""获取或创建对话"""
if conversation_id not in self.conversations:
self.conversations[conversation_id] = Conversation(
conversation_id=conversation_id,
customer_id=customer_id
)
return self.conversations[conversation_id]
def add_message(self, conversation_id: str, role: str, content: str):
"""添加消息到对话"""
if conversation_id in self.conversations:
self.conversations[conversation_id].messages.append({
"role": role,
"content": content,
"timestamp": datetime.now().isoformat()
})
# 限制历史长度
if len(self.conversations[conversation_id].messages) > self.max_history:
self.conversations[conversation_id].messages = \
self.conversations[conversation_id].messages[-self.max_history:]
def get_context(self, conversation_id: str) -> str:
"""获取对话上下文"""
if conversation_id not in self.conversations:
return ""
conversation = self.conversations[conversation_id]
context_parts = []
for msg in conversation.messages[-5:]: # 最近5条
role = "用户" if msg["role"] == "user" else "客服"
context_parts.append(f"{role}：{msg['content']}")
return "\n".join(context_parts)
# ============================================
# 第6部分：智能客服系统主类
# ============================================
class CustomerServiceAgent:
"""智能客服Agent系统"""
def __init__(self, llm_client=None):
self.llm = llm_client
# 初始化组件
self.intent_classifier = IntentClassifier(llm_client)
self.memory = ConversationMemory()
# 初始化专业Agent
self.agents = {
Intent.ORDER_QUERY: OrderAgent("订单Agent", llm_client),
Intent.REFUND_REQUEST: RefundAgent("退款Agent", llm_client),
Intent.PRODUCT_CONSULT: ConsultAgent("咨询Agent", llm_client),
Intent.COMPLAINT: ComplaintAgent("投诉Agent", llm_client),
}
# 客户数据库（模拟）
self.customers_db = {
"CUST001": Customer(
customer_id="CUST001",
name="张三",
vip_level=2,
order_history=["ORD001", "ORD002"]
),
}
def handle_message(
self,
customer_id: str,
message: str,
conversation_id: str = None
) -> str:
"""处理用户消息的主入口"""
# 1. 生成或获取会话ID
if not conversation_id:
conversation_id = f"conv_{uuid.uuid4().hex[:8]}"
# 2. 获取客户信息
customer = self.customers_db.get(
customer_id,
Customer(customer_id=customer_id, name="客户")
)
# 3. 获取或创建对话
conversation = self.memory.get_or_create(conversation_id, customer_id)
# 4. 记录用户消息
self.memory.add_message(conversation_id, "user", message)
# 5. 识别意图
intent = self.intent_classifier.classify(message)
conversation.intent = intent
print(f"[意图识别] {intent.value}")
# 6. 路由到专业Agent
if intent == Intent.HUMAN_AGENT:
response = self._transfer_to_human(customer)
conversation.state = ConversationState.ESCALATED
elif intent in self.agents:
agent = self.agents[intent]
response = agent.process(conversation, customer)
else:
response = self._general_response(message, customer)
# 7. 记录Agent回复
self.memory.add_message(conversation_id, "assistant", response)
return response
def _transfer_to_human(self, customer: Customer) -> str:
"""转接人工客服"""
return f"""
{customer.name}，正在为您转接人工客服...
预计等待时间：2-3分钟
当前排队人数：3人
在等待期间，您可以继续描述您的问题，我会记录下来转交给人工客服。
"""
def _general_response(self, message: str, customer: Customer) -> str:
"""通用回复"""
if self.llm:
context = self.memory.get_context("current")
prompt = f"""
你是一个专业的客服代表。
客户姓名：{customer.name}
VIP等级：{customer.vip_level}
对话历史：
{context}
客户最新消息：{message}
请用友好、专业的语气回复。
"""
return self.llm.generate(prompt)
return f"""
{customer.name}，感谢您的咨询。
您的问题我已记录，如果您需要：
- 查询订单 → 请提供订单号
- 申请退款 → 请说明退款原因
- 产品咨询 → 请描述您想了解的产品
- 转人工 → 请说"转人工"
请问还有什么可以帮助您的？
"""
# ============================================
# 第7部分：运行示例
# ============================================
def main():
"""运行智能客服系统示例"""
# 创建客服Agent
cs_agent = CustomerServiceAgent(llm_client=None)
print("=" * 60)
print("智能客服系统已启动")
print("=" * 60)
# 模拟对话
customer_id = "CUST001"
conversation_id = None
test_messages = [
"你好，我想查一下我的订单",
"订单号是ORD001，帮我看看到哪了",
"这个手机壳我不想要了，想退款",
"你们的退货政策是什么？",
"你们的服务太差了！我要投诉！",
"转人工",
]
for msg in test_messages:
print(f"\n客户：{msg}")
response = cs_agent.handle_message(customer_id, msg, conversation_id)
print(f"客服：{response}")
print("-" * 40)
if __name__ == "__main__":
main()
10.3 系统测试
"""
智能客服系统测试
"""
import unittest
class TestCustomerServiceAgent(unittest.TestCase):
"""智能客服Agent测试"""
def setUp(self):
"""测试初始化"""
self.agent = CustomerServiceAgent(llm_client=None)
self.customer_id = "CUST001"
def test_order_query(self):
"""测试订单查询"""
response = self.agent.handle_message(
self.customer_id,
"我想查一下订单ORD001"
)
self.assertIn("ORD001", response)
self.assertIn("iPhone", response)
def test_refund_request(self):
"""测试退款申请"""
response = self.agent.handle_message(
self.customer_id,
"我要退款，订单号ORD001"
)
self.assertIn("退款", response)
self.assertIn("REF", response)
def test_product_consult(self):
"""测试产品咨询"""
response = self.agent.handle_message(
self.customer_id,
"你们的退货政策是什么？"
)
self.assertIn("7天", response)
def test_complaint(self):
"""测试投诉处理"""
response = self.agent.handle_message(
self.customer_id,
"服务太差了，我要投诉"
)
self.assertIn("抱歉", response)
def test_human_transfer(self):
"""测试转人工"""
response = self.agent.handle_message(
self.customer_id,
"转人工"
)
self.assertIn("转接", response)
def test_intent_classification(self):
"""测试意图分类"""
classifier = IntentClassifier()
self.assertEqual(
classifier.classify("查一下我的订单"),
Intent.ORDER_QUERY
)
self.assertEqual(
classifier.classify("我要退货"),
Intent.REFUND_REQUEST
)
self.assertEqual(
classifier.classify("这个产品多少钱"),
Intent.PRODUCT_CONSULT
)
if __name__ == "__main__":
unittest.main()
10.4 扩展建议
在完成基础版本后，你可以从以下方向继续扩展这个客服系统：
1. 接入真实的LLM
# 替换模拟的LLM为真实的LLM客户端
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4", temperature=0.7)
cs_agent = CustomerServiceAgent(llm_client=llm)
2. 添加Web API接口
"""
使用FastAPI提供HTTP接口
"""
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
app = FastAPI()
cs_agent = CustomerServiceAgent()
class ChatRequest(BaseModel):
customer_id: str
message: str
conversation_id: str = None
class ChatResponse(BaseModel):
response: str
conversation_id: str
intent: str
@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
"""聊天接口"""
try:
response = cs_agent.handle_message(
request.customer_id,
request.message,
request.conversation_id
)
return ChatResponse(
response=response,
conversation_id=request.conversation_id or "new",
intent="auto"
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/health")
async def health_check():
"""健康检查"""
return {"status": "ok"}
3. 添加持久化存储
"""
使用SQLite存储对话历史
"""
import sqlite3
class PersistentMemory:
"""持久化记忆存储"""
def __init__(self, db_path: str = "conversations.db"):
self.conn = sqlite3.connect(db_path)
self._init_db()
def _init_db(self):
"""初始化数据库"""
self.conn.execute("""
CREATE TABLE IF NOT EXISTS conversations (
id TEXT PRIMARY KEY,
customer_id TEXT,
messages TEXT,
state TEXT,
created_at TEXT
)
""")
self.conn.commit()
def save_conversation(self, conversation: Conversation):
"""保存对话"""
self.conn.execute(
"INSERT OR REPLACE INTO conversations VALUES (?, ?, ?, ?, ?)",
(
conversation.conversation_id,
conversation.customer_id,
json.dumps(conversation.messages),
conversation.state.value,
conversation.created_at
)
)
self.conn.commit()
def load_conversation(self, conversation_id: str) -> Optional[Conversation]:
"""加载对话"""
cursor = self.conn.execute(
"SELECT * FROM conversations WHERE id = ?",
(conversation_id,)
)
row = cursor.fetchone()
if row:
return Conversation(
conversation_id=row[0],
customer_id=row[1],
messages=json.loads(row[2]),
state=ConversationState(row[3]),
created_at=row[4]
)
return None
4. 集成向量数据库进行知识库检索
"""
使用ChromaDB进行语义搜索
"""
# pip install chromadb
import chromadb
class VectorKnowledgeBase:
"""基于向量数据库的知识库"""
def __init__(self):
self.client = chromadb.Client()
self.collection = self.client.create_collection(
name="knowledge_base",
metadata={"hnsw:space": "cosine"}
)
def add_documents(self, documents: List[Dict]):
"""添加文档"""
ids = [f"doc_{i}" for i in range(len(documents))]
texts = [doc["content"] for doc in documents]
metadatas = [{"topic": doc["topic"]} for doc in documents]
self.collection.add(
ids=ids,
documents=texts,
metadatas=metadatas
)
def search(self, query: str, top_k: int = 3) -> List[Dict]:
"""语义搜索"""
results = self.collection.query(
query_texts=[query],
n_results=top_k
)
return [
{
"content": doc,
"topic": meta["topic"],
"distance": dist
}
for doc, meta, dist in zip(
results["documents"][0],
results["metadatas"][0],
results["distances"][0]
)
]
总结
恭喜你完成了这份AI Agent智能体开发入门教程！让我们回顾一下你学到的关键内容：
核心知识点回顾
- AI Agent基础概念：理解了Agent与传统Chatbot的区别，掌握了感知-推理-行动循环的核心原理
- Agent架构模式：学会了ReAct和Plan-and-Execute两种主流架构，了解了工具调用链的工作方式
- 主流框架对比：了解了LangChain、AutoGPT、CrewAI、MetaGPT的特点和适用场景
- LangChain实战：掌握了环境搭建、Agent创建、工具定义与注册的完整流程
- 工具系统：学会了自定义工具开发、工具组合、错误处理和MCP协议
- 记忆管理：掌握了短期记忆、长期记忆、对话窗口管理等关键技术
- 规划能力：学会了任务分解、多步推理、自我反思与纠错
- 多Agent协作：掌握了主从模式、对等协作、任务分配策略
- 生产部署：学会了安全沙箱、速率限制、监控日志、成本控制
- 实战项目：从零构建了一个完整的智能客服Agent系统
下一步学习建议
- 动手实践：选择一个你感兴趣的方向，动手实现一个Agent项目
- 深入框架：选择一个框架（推荐LangChain）深入学习其高级特性
- 关注前沿：持续关注Agent领域的最新研究和开源项目
- 社区参与：加入Agent开发者社区，参与讨论和贡献
推荐学习资源
- 官方文档：LangChain、OpenAI、Anthropic等官方文档
- 开源项目：AutoGPT、MetaGPT、CrewAI的GitHub仓库
- 学术论文：ReAct、Toolformer、HuggingGPT等经典论文
- 在线课程：各大平台的LLM和Agent相关课程
AI Agent是人工智能领域最具前景的方向之一。随着大语言模型能力的不断提升，Agent将会在更多场景中发挥重要作用。希望这份教程能够帮助你开启Agent开发之旅，在这个充满机遇的领域中探索和创新。
祝你学习愉快，编码顺利！ 🚀
