LLM Agent:从聊天机器人到智能助手的进化
LLM Agent是具有自主决策和执行能力的AI系统,它能够理解目标、制定计划、调用工具、 并根据反馈调整策略。本文将详细介绍如何构建强大的Agent系统。
Agent核心架构
智能Agent的关键组件
🧠 大脑(LLM)
- • 理解用户意图
- • 制定执行计划
- • 推理和决策
- • 生成响应
🔧 工具(Tools)
- • API调用
- • 数据库查询
- • 代码执行
- • 文件操作
💾 记忆(Memory)
- • 短期记忆(对话历史)
- • 长期记忆(知识库)
- • 工作记忆(任务状态)
- • 情景记忆(经验)
📋 规划(Planning)
- • 任务分解
- • 步骤排序
- • 资源调度
- • 策略优化
ReAct模式:思考与行动
Reasoning + Acting = ReAct
class ReActAgent:
"""ReAct模式的Agent实现"""
def __init__(self, llm, tools):
self.llm = llm
self.tools = tools
self.max_steps = 10
def run(self, task):
"""执行任务"""
prompt = f"""你是一个智能助手,需要完成以下任务:{task}
可用工具:
{self.get_tools_description()}
请按照以下格式回复:
Thought: 我需要思考下一步做什么
Action: 工具名称
Action Input: 工具输入参数
Observation: 工具返回结果
... (重复Thought/Action/Action Input/Observation)
Thought: 我已经知道最终答案
Final Answer: 最终答案
开始:"""
messages = [{"role": "user", "content": prompt}]
steps = 0
while steps < self.max_steps:
# LLM思考
response = self.llm.generate(messages)
# 解析响应
if "Final Answer:" in response:
return self.extract_final_answer(response)
# 提取动作
action, action_input = self.parse_action(response)
if action and action in self.tools:
# 执行工具
observation = self.tools[action](action_input)
# 添加观察结果
messages.append({
"role": "assistant",
"content": response
})
messages.append({
"role": "user",
"content": f"Observation: {observation}"
})
else:
# 处理错误
messages.append({
"role": "user",
"content": "无效的动作,请重新思考"
})
steps += 1
return "达到最大步骤限制"
def parse_action(self, text):
"""解析动作和输入"""
import re
action_match = re.search(r'Action:\s*(.+)', text)
input_match = re.search(r'Action Input:\s*(.+)', text)
if action_match and input_match:
return action_match.group(1), input_match.group(1)
return None, None
# 使用示例
agent = ReActAgent(
llm=llm_client,
tools={
"search": web_search_tool,
"calculator": calculator_tool,
"database": database_query_tool
}
)
result = agent.run("北京今天的天气怎么样,温度是多少摄氏度?")
print(result)Function Calling实现
让LLM调用外部函数
import json
from typing import List, Dict, Any
class FunctionCallingAgent:
"""基于Function Calling的Agent"""
def __init__(self, api_key: str):
self.client = OpenAI(api_key=api_key)
self.functions = []
self.function_map = {}
def register_function(self, func, description: str, parameters: Dict):
"""注册可调用函数"""
function_spec = {
"name": func.__name__,
"description": description,
"parameters": parameters
}
self.functions.append(function_spec)
self.function_map[func.__name__] = func
def run(self, user_input: str):
"""执行用户请求"""
messages = [
{"role": "system", "content": "你是一个智能助手"},
{"role": "user", "content": user_input}
]
# 第一次调用:决定是否需要使用函数
response = self.client.chat.completions.create(
model="gpt-4",
messages=messages,
functions=self.functions,
function_call="auto"
)
message = response.choices[0].message
# 检查是否需要调用函数
if message.function_call:
# 解析函数调用
function_name = message.function_call.name
function_args = json.loads(message.function_call.arguments)
# 执行函数
if function_name in self.function_map:
function_result = self.function_map[function_name](**function_args)
# 将函数结果添加到对话
messages.append(message)
messages.append({
"role": "function",
"name": function_name,
"content": str(function_result)
})
# 第二次调用:基于函数结果生成最终回复
final_response = self.client.chat.completions.create(
model="gpt-4",
messages=messages
)
return final_response.choices[0].message.content
return message.content
# 定义工具函数
def get_weather(location: str, unit: str = "celsius"):
"""获取天气信息"""
# 实际应调用天气API
return f"{location}的天气是晴天,温度是25{unit}"
def calculate(expression: str):
"""计算数学表达式"""
try:
result = eval(expression)
return f"计算结果:{result}"
except:
return "计算错误"
def search_database(query: str, table: str):
"""查询数据库"""
# 实际应执行数据库查询
return f"从{table}表查询'{query}'的结果:..."
# 创建Agent并注册函数
agent = FunctionCallingAgent(api_key="your-key")
agent.register_function(
get_weather,
"获取指定地点的天气信息",
{
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "地点名称"
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"description": "温度单位"
}
},
"required": ["location"]
}
)
agent.register_function(
calculate,
"计算数学表达式",
{
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "数学表达式"
}
},
"required": ["expression"]
}
)
# 使用Agent
result = agent.run("北京的天气怎么样?")
print(result) # 自动调用get_weather函数
result = agent.run("计算 123 * 456 + 789")
print(result) # 自动调用calculate函数多Agent协作系统
构建Agent团队完成复杂任务
class MultiAgentSystem:
"""多Agent协作系统"""
def __init__(self):
self.agents = {}
self.message_queue = []
def register_agent(self, name: str, agent):
"""注册Agent"""
self.agents[name] = agent
def run_task(self, task: str):
"""执行协作任务"""
# 主管Agent分解任务
manager = self.agents["manager"]
plan = manager.create_plan(task)
results = {}
for step in plan["steps"]:
agent_name = step["agent"]
subtask = step["task"]
# 分配任务给特定Agent
if agent_name in self.agents:
agent = self.agents[agent_name]
result = agent.execute(subtask, context=results)
results[step["id"]] = result
# 广播结果给其他Agent
self.broadcast_message({
"from": agent_name,
"type": "result",
"content": result
})
# 汇总结果
return manager.summarize(results)
def broadcast_message(self, message):
"""广播消息"""
for agent_name, agent in self.agents.items():
if hasattr(agent, "receive_message"):
agent.receive_message(message)
# 专门化的Agent类
class ResearchAgent:
"""研究Agent - 负责信息收集"""
def execute(self, task, context=None):
# 搜索和收集信息
return f"研究结果:{task}的相关信息..."
class WriterAgent:
"""写作Agent - 负责内容创作"""
def execute(self, task, context=None):
# 基于研究结果写作
research = context.get("research", "")
return f"基于{research}创作的内容..."
class ReviewerAgent:
"""审核Agent - 负责质量检查"""
def execute(self, task, context=None):
# 审核内容质量
content = context.get("content", "")
return f"审核意见:{content}需要改进的地方..."
class ManagerAgent:
"""管理Agent - 负责任务规划和协调"""
def create_plan(self, task):
return {
"steps": [
{"id": "research", "agent": "researcher", "task": f"研究{task}"},
{"id": "write", "agent": "writer", "task": f"撰写{task}"},
{"id": "review", "agent": "reviewer", "task": f"审核{task}"}
]
}
def summarize(self, results):
return f"任务完成,最终成果:{results}"
# 构建多Agent系统
system = MultiAgentSystem()
system.register_agent("manager", ManagerAgent())
system.register_agent("researcher", ResearchAgent())
system.register_agent("writer", WriterAgent())
system.register_agent("reviewer", ReviewerAgent())
# 执行复杂任务
result = system.run_task("写一篇关于人工智能的文章")
print(result)Agent记忆系统
让Agent记住和学习
短期记忆实现
class ShortTermMemory:
def __init__(self, max_size=10):
self.buffer = []
self.max_size = max_size
def add(self, item):
self.buffer.append(item)
if len(self.buffer) > self.max_size:
self.buffer.pop(0)
def get_context(self):
return "\n".join(self.buffer[-5:])长期记忆实现
class LongTermMemory:
def __init__(self, vector_db):
self.vector_db = vector_db
def store(self, content, metadata):
embedding = create_embedding(content)
self.vector_db.insert(
embedding, content, metadata
)
def recall(self, query, k=5):
results = self.vector_db.search(query, k)
return resultsAgent框架对比
| 框架 | 特点 | 易用性 | 扩展性 | 适用场景 |
|---|---|---|---|---|
| LangChain | 功能全面,生态丰富 | ⭐⭐⭐ | ⭐⭐⭐⭐⭐ | 通用开发 |
| AutoGPT | 完全自主 | ⭐⭐ | ⭐⭐⭐ | 自动化任务 |
| AutoGen | 多Agent协作 | ⭐⭐⭐⭐ | ⭐⭐⭐⭐ | 团队协作 |
| CrewAI | 角色定义清晰 | ⭐⭐⭐⭐⭐ | ⭐⭐⭐ | 流程自动化 |
| BabyAGI | 任务驱动 | ⭐⭐⭐ | ⭐⭐⭐ | 目标导向 |
Agent评估指标
如何评价Agent的性能
⚡ 效率指标
- • 任务完成时间
- • 步骤数量
- • 资源消耗
- • API调用次数
✅ 准确性指标
- • 任务成功率
- • 错误率
- • 结果质量
- • 一致性
🤝 交互指标
- • 响应时间
- • 理解准确度
- • 用户满意度
- • 澄清次数
实战案例:客服Agent
构建智能客服Agent
class CustomerServiceAgent:
"""智能客服Agent"""
def __init__(self, llm, knowledge_base, order_system):
self.llm = llm
self.knowledge_base = knowledge_base
self.order_system = order_system
self.conversation_history = []
def handle_query(self, user_query):
"""处理用户查询"""
# 意图识别
intent = self.identify_intent(user_query)
# 根据意图路由
if intent == "order_tracking":
return self.track_order(user_query)
elif intent == "product_inquiry":
return self.product_info(user_query)
elif intent == "complaint":
return self.handle_complaint(user_query)
else:
return self.general_response(user_query)
def identify_intent(self, query):
"""识别用户意图"""
prompt = f"""分析以下用户查询的意图:
查询:{query}
可能的意图:
- order_tracking: 订单查询
- product_inquiry: 产品咨询
- complaint: 投诉建议
- general: 一般问题
返回最匹配的意图类型:"""
response = self.llm.generate(prompt)
return response.strip().lower()
def track_order(self, query):
"""订单追踪"""
# 提取订单号
order_id = self.extract_order_id(query)
if order_id:
# 查询订单系统
order_info = self.order_system.get_order(order_id)
if order_info:
return f"""您的订单 {order_id} 当前状态:
📦 状态:{order_info['status']}
📍 位置:{order_info['location']}
📅 预计送达:{order_info['eta']}
如需其他帮助,请随时告诉我。"""
else:
return f"抱歉,未找到订单号 {order_id}。请确认订单号是否正确。"
else:
return "请提供您的订单号,我来帮您查询。"
def product_info(self, query):
"""产品信息查询"""
# 从知识库检索
relevant_info = self.knowledge_base.search(query, top_k=3)
if relevant_info:
response = self.llm.generate(f"""
基于以下产品信息回答用户问题:
信息:{relevant_info}
用户问题:{query}
请提供准确、有帮助的回答:""")
return response
else:
return "抱歉,暂时没有找到相关产品信息。我可以为您转接人工客服。"
def handle_complaint(self, query):
"""处理投诉"""
# 记录投诉
complaint_id = self.log_complaint(query)
# 生成同理心回复
response = self.llm.generate(f"""
用户投诉:{query}
请生成一个富有同理心的回复,包括:
1. 表达理解和歉意
2. 说明处理流程
3. 提供投诉编号:{complaint_id}
4. 承诺跟进
回复:""")
return response
def general_response(self, query):
"""一般问题回复"""
# 结合历史对话
context = "\n".join(self.conversation_history[-5:])
response = self.llm.generate(f"""
对话历史:
{context}
用户问题:{query}
请提供友好、专业的回答:""")
self.conversation_history.append(f"用户:{query}")
self.conversation_history.append(f"客服:{response}")
return response
# 使用示例
agent = CustomerServiceAgent(
llm=llm_client,
knowledge_base=vector_db,
order_system=order_api
)
# 处理各种查询
print(agent.handle_query("我的订单12345什么时候到?"))
print(agent.handle_query("这个产品的保修期是多久?"))
print(agent.handle_query("产品质量有问题,我要投诉!"))