
一、Agent是什么?
打个比方:Agent就像你的智能助理团队
想象一下:
- 传统大模型:一个博学的学者,能回答各种问题,但只会说,不会做
- Agent:一个带手脚的学者,不仅能回答问题,还能实际操作各种工具完成任务
生活中的Agent例子
|
场景 |
传统AI |
AI Agent |
|
订外卖 |
告知你哪家餐厅好 |
直接帮你选餐厅、下单、支付 |
|
写报告 |
生成报告内容 |
收集数据→分析→写报告→排版→发送 |
|
旅行计划 |
提议去哪玩 |
查机票→订酒店→规划路线→提醒行程 |
技术定义
Agent = 大模型 + 思考能力 + 工具使用 + 记忆系统
二、Agent的核心原理
Agent的”大脑”结构
┌─────────────┐
│ 用户输入 │
└──────┬──────┘
▼
┌─────────────────┐
│ 思考/规划层 │ ← 大模型核心
│ (Thinking) │
└────────┬────────┘
│
┌─────────────┼─────────────┐
▼ ▼ ▼
┌─────────┐ ┌─────────┐ ┌─────────┐
│ 工具调用 │ │ 记忆检索 │ │ 决策判断 │
│ (Tools) │ │ (Memory) │ │(Decision) │
└────┬────┘ └────┬────┘ └────┬────┘
│ │ │
└────────── ┼───────── ─┘
▼
┌─────────────┐
│ 执行行动 │
└─────────────┘
1.思考过程:ReAct框架
这是Agent的”灵魂”,让AI先想后做
# ReAct模式的伪代码思考过程
用户: "北京的天气怎么样?"
Agent思考过程:
1. 【思考】用户想知道北京天气,我需要:
- 获取北京当前天气信息
- 需要调用天气API
2. 【行动】调用 weather_api("北京")
3. 【观察】API返回:北京,25°C,晴天
4. 【思考】需要把信息组织成友善回答
5. 【回答】"北京目前是25°C,晴天,适合外出。"
# 关键是:每个行动前都要"想一想"
2.工具使用:Agent的”超能力”
Agent自己不会做所有事,但会使用工具:
# Agent的工具箱
tools = {
"search_web": "搜索最新信息",
"calculator": "进行数学计算",
"code_interpreter": "执行代码",
"file_editor": "读写文件",
"api_caller": "调用外部API"
}
# Agent的思考:"用户让我算(12+34)*2"
# → 调用calculator工具 → 得到结果92 → 告知用户
3.记忆系统:Agent的”记事本”
- 短期记忆:当前对话的上下文
- 长期记忆:从过去经验中学习
- 工作记忆:正在处理的任务信息
# 记忆示例
memory = {
"user_preferences": { # 长期记忆
"name": "小明",
"likes_coffee": True,
"timezone": "UTC+8"
},
"conversation_history": [ # 短期记忆
{"role": "user", "content": "明天提醒我开会"},
{"role": "assistant", "content": "好的,已设置提醒"}
],
"current_task": { # 工作记忆
"goal": "帮用户订机票",
"step": 3,
"collected_info": {"destination": "上海", "date": "2024-12-10"}
}
}
4.规划能力:Agent的”路线图”
复杂任务拆解成小步骤:
任务: "帮我分析公司上周销售数据并写报告"
Agent的规划:
1. 【子任务1】连接到数据库
2. 【子任务2】查询上周销售数据
3. 【子任务3】计算关键指标(增长率、Top产品等)
4. 【子任务4】生成可视化图表
5. 【子任务5】撰写分析报告
6. 【子任务6】保存并发送报告
每个子任务都可能需要调用不同工具
三、Agent开发实战(从0到1)
开发环境准备
# 1. 安装Python(3.9+)
# 2. 安装必要库
pip install openai langchain crewai
# 或者使用全功能环境
pip install langchain langchain-openai langchain-community
pip install crewai crewai-tools
方案一:使用LangChain(最流行)
第1步:基础Agent – 计算助手
# basic_agent.py
from langchain.agents import initialize_agent, Tool
from langchain.llms import OpenAI
from langchain.chat_models import ChatOpenAI
import math
# 1. 定义工具函数
def calculator(query: str) -> str:
"""计算数学表达式"""
try:
# 安全评估数学表达式
allowed_chars = set("0123456789+-*/(). ")
if not all(c in allowed_chars for c in query):
return "只支持基本数学运算"
result = eval(query)
return f"计算结果: {result}"
except:
return "计算失败,请检查表达式"
def search_web(query: str) -> str:
"""模拟搜索功能(实际需接入搜索引擎API)"""
return f"搜索 '{query}' 的结果:[模拟数据]"
# 2. 创建工具列表
tools = [
Tool(
name="Calculator",
func=calculator,
description="用于数学计算,输入如 '12+34*2'"
),
Tool(
name="WebSearch",
func=search_web,
description="用于搜索最新信息"
)
]
# 3. 初始化Agent
llm = ChatOpenAI(temperature=0, model="gpt-3.5-turbo")
agent = initialize_agent(
tools,
llm,
agent="zero-shot-react-description", # ReAct模式
verbose=True # 显示思考过程
)
# 4. 运行Agent
result = agent.run("计算 (15 + 27) * 3 等于多少,然后搜索'AI发展现状'")
print(result)
第2步:增强Agent – 带记忆的对话助手
# memory_agent.py
from langchain.memory import ConversationBufferMemory
from langchain.agents import AgentExecutor
from langchain.agents import create_react_agent
from langchain.prompts import PromptTemplate
# 1. 创建记忆系统
memory = ConversationBufferMemory(
memory_key="chat_history",
return_messages=True
)
# 2. 定制提示模板
prompt_template = """你是一个有协助的助手。你可以使用工具。
之前的对话:
{chat_history}
问题:{input}
你可以使用以下工具:
{tools}
思考过程:{agent_scratchpad}"""
prompt = PromptTemplate.from_template(prompt_template)
# 3. 创建ReAct Agent
agent = create_react_agent(
llm=llm,
tools=tools,
prompt=prompt
)
# 4. 创建执行器
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
memory=memory,
verbose=True,
handle_parsing_errors=True
)
# 5. 运行多轮对话
agent_executor.invoke({"input": "我叫小明,记住我喜爱咖啡"})
agent_executor.invoke({"input": "我叫什么?我喜爱什么?"})
第3步:专业Agent – 数据分析助手
# data_analyst_agent.py
import pandas as pd
import matplotlib.pyplot as plt
from io import StringIO
class DataAnalystAgent:
def __init__(self):
self.data = None
def load_csv(self, file_path: str) -> str:
"""加载CSV数据"""
try:
self.data = pd.read_csv(file_path)
return f"数据加载成功,共{len(self.data)}行,{len(self.data.columns)}列"
except Exception as e:
return f"加载失败: {str(e)}"
def describe_data(self) -> str:
"""描述数据"""
if self.data is None:
return "请先加载数据"
buffer = StringIO()
self.data.info(buf=buffer)
return buffer.getvalue()
def calculate_statistics(self) -> str:
"""计算统计信息"""
if self.data is None:
return "请先加载数据"
numeric_cols = self.data.select_dtypes(include=['number']).columns
stats = self.data[numeric_cols].describe()
return str(stats)
def create_chart(self, x_col: str, y_col: str) -> str:
"""创建图表"""
if self.data is None:
return "请先加载数据"
plt.figure(figsize=(10, 6))
plt.plot(self.data[x_col], self.data[y_col])
plt.xlabel(x_col)
plt.ylabel(y_col)
plt.title(f"{y_col} vs {x_col}")
plt.savefig("chart.png")
return "图表已保存为 chart.png"
def query_data(self, query: str) -> str:
"""查询数据(自然语言转SQL)"""
# 这里简化处理,实际应用可以使用更复杂的NLP转SQL
if "最大值" in query and "销售额" in query:
max_sales = self.data['sales'].max()
return f"最大销售额是: {max_sales}"
return "请更明确地描述查询"
# 创建工具
analyst = DataAnalystAgent()
tools = [
Tool(
name="LoadData",
func=analyst.load_csv,
description="加载CSV文件,参数:文件路径"
),
Tool(
name="DescribeData",
func=analyst.describe_data,
description="描述数据基本信息"
),
Tool(
name="Statistics",
func=analyst.calculate_statistics,
description="计算数值列的统计信息"
),
Tool(
name="CreateChart",
func=analyst.create_chart,
description="创建图表,参数:x轴列名, y轴列名"
),
Tool(
name="QueryData",
func=analyst.query_data,
description="查询数据,可以用自然语言描述"
)
]
# 创建数据分析Agent
data_agent = initialize_agent(
tools,
llm,
agent="zero-shot-react-description",
verbose=True
)
# 使用
result = data_agent.run("加载sales.csv,然后告知我数据的统计信息")
方案二:使用CrewAI(团队协作)
# crewai_agent.py
from crewai import Agent, Task, Crew
from crewai_tools import SerperDevTool, FileReadTool
# 工具
search_tool = SerperDevTool()
file_read_tool = FileReadTool()
# 创建研究员Agent
researcher = Agent(
role='市场研究员',
goal='研究AI行业最新趋势',
backstory='你是顶尖的市场分析师,擅长发现行业趋势',
tools=[search_tool],
verbose=True,
allow_delegation=False
)
# 创建作家Agent
writer = Agent(
role='技术作家',
goal='撰写高质量的技术报告',
backstory='你是资深技术作家,擅长将复杂概念转化为易懂内容',
tools=[file_read_tool],
verbose=True,
allow_delegation=True
)
# 创建任务
research_task = Task(
description="研究2024年AI Agent的发展趋势,重点关注技术突破和商业应用",
expected_output="一份详细的研究笔记,包含关键发现和引用来源",
agent=researcher
)
writing_task = Task(
description="基于研究笔记,撰写一篇关于AI Agent发展趋势的博客文章",
expected_output="一篇1000字左右的博客文章,适合技术读者",
agent=writer
)
# 创建团队
crew = Crew(
agents=[researcher, writer],
tasks=[research_task, writing_task],
verbose=True
)
# 执行任务
result = crew.kickoff()
print(result)
方案三:自主开发
# custom_agent.py
import openai
import json
class SimpleAgent:
def __init__(self, api_key):
self.api_key = api_key
openai.api_key = api_key
self.conversation_history = []
self.tools = self._initialize_tools()
def _initialize_tools(self):
"""初始化工具集"""
return {
"get_weather": {
"func": self._get_weather,
"description": "获取城市天气,参数:城市名"
},
"calculate": {
"func": self._calculate,
"description": "数学计算,参数:数学表达式"
},
"search": {
"func": self._search,
"description": "搜索信息,参数:搜索词"
}
}
def _get_weather(self, city: str) -> str:
"""模拟获取天气"""
# 实际应该调用天气API
weather_data = {
"北京": "25°C, 晴天",
"上海": "28°C, 多云",
"广州": "30°C, 阵雨"
}
return weather_data.get(city, "未知城市")
def _calculate(self, expression: str) -> str:
"""计算"""
try:
return str(eval(expression))
except:
return "计算错误"
def _search(self, query: str) -> str:
"""搜索"""
return f"搜索 '{query}' 的结果:[模拟数据]"
def _think(self, user_input: str) -> dict:
"""思考:决定下一步做什么"""
prompt = f"""
你是一个智能助手,可以调用工具。
可用工具:
{json.dumps([{'name': k, 'desc': v['description']} for k, v in self.tools.items()], ensure_ascii=False)}
历史对话:
{self.conversation_history[-5:] if len(self.conversation_history) > 5 else self.conversation_history}
用户输入:{user_input}
分析用户需求,如果需要调用工具,返回:
{{"action": "use_tool", "tool": "工具名", "parameters": "参数"}}
如果可以直接回答,返回:
{{"action": "respond", "response": "你的回答"}}
只返回JSON格式。
"""
response = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": prompt}],
temperature=0
)
return json.loads(response.choices[0].message.content)
def process(self, user_input: str) -> str:
"""处理用户输入"""
# 1. 思考决策
decision = self._think(user_input)
# 2. 记录对话
self.conversation_history.append({"role": "user", "content": user_input})
# 3. 执行决策
if decision["action"] == "use_tool":
tool_name = decision["tool"]
params = decision["parameters"]
if tool_name in self.tools:
# 调用工具
result = self.tools[tool_name]["func"](params)
response = f"调用 {tool_name}({params}) 得到结果:{result}"
else:
response = f"工具 {tool_name} 不存在"
elif decision["action"] == "respond":
response = decision["response"]
# 4. 记录助手回复
self.conversation_history.append({"role": "assistant", "content": response})
return response
# 使用Agent
agent = SimpleAgent(api_key="your-api-key")
print(agent.process("北京天气怎么样?"))
print(agent.process("计算123+456"))
四、Agent开发最佳实践
1.工具设计原则
# 好的工具设计
class GoodTool:
def __init__(self):
self.name = "清晰的名字"
self.description = "明确的功能描述和参数说明"
def run(self, input_str: str) -> str:
"""
参数格式清晰
错误处理完善
返回结果结构化
"""
try:
# 解析参数
params = self._parse_input(input_str)
# 执行业务逻辑
result = self._execute(params)
# 格式化返回
return self._format_result(result)
except Exception as e:
return f"错误: {str(e)}"
2.错误处理策略
class RobustAgent:
def __init__(self):
self.max_retries = 3
self.fallback_responses = {
"tool_error": "抱歉,工具暂时不可用,我会手动帮您处理。",
"parse_error": "我没理解您的意思,能再说清楚点吗?",
"timeout": "处理超时,请稍后再试。"
}
def safe_execute(self, func, *args, **kwargs):
"""安全执行函数"""
for attempt in range(self.max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
if attempt == self.max_retries - 1:
return self.fallback_responses.get(
type(e).__name__,
"抱歉,出现了未知错误"
)
3.性能优化
class OptimizedAgent:
def __init__(self):
self.cache = {} # 缓存常用结果
self.batch_size = 10 # 批处理大小
def batch_process(self, tasks):
"""批处理优化"""
results = []
for i in range(0, len(tasks), self.batch_size):
batch = tasks[i:i+self.batch_size]
# 批量处理逻辑
batch_results = self._process_batch(batch)
results.extend(batch_results)
return results
五、实战项目:构建个人学习助手Agent
# learning_assistant.py
from datetime import datetime
import json
class LearningAssistant:
def __init__(self):
self.study_plan = {}
self.notes = {}
self.progress = {}
def create_study_plan(self, topic: str, hours: int, deadline: str):
"""创建学习计划"""
plan = {
"topic": topic,
"total_hours": hours,
"deadline": deadline,
"created_at": datetime.now().isoformat(),
"daily_hours": hours / 7, # 假设一周完成
"status": "active"
}
self.study_plan[topic] = plan
return f"已创建 '{topic}' 学习计划,每天学习{plan['daily_hours']:.1f}小时"
def add_note(self, topic: str, content: str):
"""添加学习笔记"""
if topic not in self.notes:
self.notes[topic] = []
self.notes[topic].append({
"content": content,
"timestamp": datetime.now().isoformat()
})
return f"已为 '{topic}' 添加笔记"
def track_progress(self, topic: str, hours_studied: float):
"""跟踪学习进度"""
if topic not in self.progress:
self.progress[topic] = 0
self.progress[topic] += hours_studied
if topic in self.study_plan:
total = self.study_plan[topic]["total_hours"]
percentage = (self.progress[topic] / total) * 100
return f"'{topic}' 学习进度:{percentage:.1f}% ({self.progress[topic]}/{total}小时)"
return f"已记录学习 {hours_studied} 小时"
def get_recommendations(self, topic: str):
"""获取学习提议"""
resources = {
"Python": [
"官方文档: https://docs.python.org",
"Python Crash Course 书籍",
"freeCodeCamp Python教程"
],
"机器学习": [
"吴恩达 Coursera 课程",
"Scikit-learn 官方教程",
"动手学深度学习"
],
"Agent开发": [
"LangChain 官方文档",
"OpenAI Cookbook",
"本教程的实战项目"
]
}
if topic in resources:
return f"推荐学习资源:
" + "
".join(f"- {r}" for r in resources[topic])
return "暂时没有该主题的推荐资源"
def review_notes(self, topic: str):
"""复习笔记"""
if topic in self.notes:
notes = self.notes[topic]
review = f"'{topic}' 的笔记复习:
"
for i, note in enumerate(notes, 1):
review += f"{i}. {note['content']}
"
return review
return f"'{topic}' 还没有笔记"
# 集成到Agent
assistant_tools = [
Tool(
name="CreateStudyPlan",
func=LearningAssistant().create_study_plan,
description="创建学习计划,参数:主题,总小时数,截止日期"
),
Tool(
name="AddNote",
func=LearningAssistant().add_note,
description="添加学习笔记,参数:主题,内容"
),
# ... 其他工具
]
六、常见问题解答
Q1: Agent需要多少数据才能工作?
A: Agent本身不需要训练数据,它依赖于底层大模型(如GPT)。你只需要:
- API密钥(或本地部署模型)
- 定义清晰的工具
- 设计好的提示词
Q2: 开发Agent难吗?
A: 入门不难,但深入需要:
- Python基础(能写函数)
- 理解API调用
- 基本的系统设计思维
Q3: Agent能做什么实际应用?
A: 许多!列如:
- 客服机器人(自动回答+查询订单)
- 数据分析助手(自动分析报表)
- 代码助手(写代码+调试+测试)
- 个人秘书(管理日程、提醒)
Q4: 需要多少计算资源?
A: 分两种情况:
- API模式:几乎零成本启动,按使用量付费
- 本地部署:需要较好的GPU(如RTX 3090+)
© 版权声明
文章版权归作者所有,未经允许请勿转载。
相关文章
暂无评论...



