1. 项目概述:构建一个能调用工具的Agent
在AI技术快速发展的今天,Agent(智能代理)已经成为连接大语言模型(LLM)与现实世界的重要桥梁。不同于单纯的聊天机器人,一个真正的Agent能够主动调用外部工具、处理复杂任务并做出决策。这就像给一个知识渊博但"四肢不勤"的学者配上了一整套工具包,让它不仅能思考,还能动手操作。
我最近从零构建了一个具备工具调用能力的Agent系统,完整走通了从原理设计到代码实现的全部流程。这个项目最吸引人的地方在于,它完美展示了如何让AI突破纯文本交互的限制,真正具备"动手能力"——无论是查询天气、操作数据库,还是控制智能家居,只要提供合适的工具接口,Agent都能帮你完成。
2. 核心原理拆解
2.1 Agent系统的基本架构
一个完整的工具调用型Agent通常包含以下核心组件:
-
大脑(LLM核心):负责理解用户意图、规划任务步骤和生成决策。通常基于GPT-4、Claude或开源模型如Qwen等构建。
-
工具注册中心:维护所有可用工具的元信息,包括:
- 工具名称和描述(供LLM理解工具用途)
- 输入参数规范(JSON Schema格式)
- 调用方式(API端点、本地函数等)
-
执行引擎:负责:
- 解析LLM生成的工具调用请求
- 验证参数合法性
- 实际调用工具并获取结果
- 处理错误和重试逻辑
-
会话管理器:维护对话上下文,处理多轮交互和长期记忆。
2.2 工具调用的工作流程
当用户提出请求时,系统会经历以下典型流程:
- 意图识别:LLM判断是否需要调用工具(vs直接回答)
- 工具选择:从注册中心选择最合适的工具
- 参数提取:从用户输入中提取或推导所需参数
- 请求生成:构造符合工具要求的调用格式
- 执行验证:检查参数有效性后再实际调用
- 结果处理:将原始结果转换为用户友好的响应
这个过程中最关键的创新点是"工具使用"被建模为一种特殊的文本生成——当LLM需要调用工具时,它不会生成普通回复,而是输出结构化的工具调用指令。
3. 代码实现详解
3.1 基础环境搭建
我们使用Python 3.10+作为开发语言,主要依赖:
python复制# 核心依赖
openai>=1.0 # 新版SDK支持工具调用
langchain>=0.1 # 提供Agent框架
fastapi>=0.95 # 构建工具API
# 辅助工具
pydantic>=2.0 # 数据验证
tenacity>=8.0 # 重试逻辑
建议使用虚拟环境隔离依赖:
bash复制python -m venv agent_env
source agent_env/bin/activate # Linux/Mac
pip install -r requirements.txt
3.2 工具注册实现
首先定义工具接口规范:
python复制from pydantic import BaseModel
class Tool(BaseModel):
name: str
description: str
parameters: dict # JSON Schema格式
def execute(self, **kwargs):
raise NotImplementedError
然后实现具体工具。以天气查询为例:
python复制import requests
class WeatherTool(Tool):
def __init__(self):
super().__init__(
name="get_current_weather",
description="获取指定城市的当前天气情况",
parameters={
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "城市名称,如'北京'"
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"default": "celsius"
}
},
"required": ["location"]
}
)
def execute(self, location: str, unit: str = "celsius"):
# 实际调用天气API
api_key = os.getenv("WEATHER_API_KEY")
url = f"https://api.weatherapi.com/v1/current.json?key={api_key}&q={location}"
response = requests.get(url)
data = response.json()
return {
"temperature": data["current"]["temp_" + unit],
"condition": data["current"]["condition"]["text"]
}
3.3 Agent核心逻辑
构建能够理解工具调用的Agent:
python复制from openai import OpenAI
class ToolAgent:
def __init__(self, tools: list[Tool]):
self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
self.tools = {tool.name: tool for tool in tools}
def chat(self, message: str, history: list = None) -> str:
# 准备工具描述供LLM理解
tool_descriptions = [
{
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"parameters": tool.parameters
}
}
for tool in self.tools.values()
]
# 调用LLM并允许工具调用
response = self.client.chat.completions.create(
model="gpt-4-1106-preview",
messages=history or [{"role": "user", "content": message}],
tools=tool_descriptions,
tool_choice="auto"
)
# 处理响应
assistant_message = response.choices[0].message
if assistant_message.tool_calls:
# 需要执行工具调用
results = []
for tool_call in assistant_message.tool_calls:
tool = self.tools[tool_call.function.name]
kwargs = json.loads(tool_call.function.arguments)
result = tool.execute(**kwargs)
results.append(result)
# 将结果返回给LLM生成最终回复
return self._process_tool_results(message, results)
else:
# 直接文本回复
return assistant_message.content
def _process_tool_results(self, query: str, results: list):
# 将工具结果反馈给LLM生成友好回复
response = self.client.chat.completions.create(
model="gpt-4-1106-preview",
messages=[
{"role": "user", "content": query},
{"role": "assistant", "content": str(results)}
]
)
return response.choices[0].message.content
4. 实战中的关键问题与解决方案
4.1 工具描述的精确性
问题现象:LLM频繁错误调用工具或误解参数含义。
根本原因:工具名称和描述不够精确,导致LLM无法准确匹配用户意图。
解决方案:
- 使用动词开头的工具名(如"get_weather"而非"weather")
- 在描述中明确说明工具的适用场景和限制条件
- 为每个参数添加详细说明和示例
优化后的天气工具描述示例:
python复制description="""获取城市级别的实时天气数据。
适用场景:用户询问当前天气或出行建议时。
限制:仅支持地级市及以上城市,不支持具体街道。"""
4.2 参数验证与错误处理
典型错误:
- 用户提供"纽约市"但工具需要"New York"
- 温度单位参数缺失时未使用默认值
健壮性改进:
python复制from tenacity import retry, stop_after_attempt
class WeatherTool(Tool):
@retry(stop=stop_after_attempt(3))
def execute(self, location: str, unit: str = "celsius"):
try:
# 统一城市名称格式
normalized_loc = self._normalize_location(location)
# 参数验证
if unit not in ["celsius", "fahrenheit"]:
unit = "celsius"
# 实际API调用...
except Exception as e:
return {"error": str(e)}
def _normalize_location(self, location: str) -> str:
# 实现中文城市名到英文的转换逻辑
# 例如"北京" -> "Beijing"
...
4.3 工具组合与任务分解
复杂任务示例:
"帮我比较北京和上海下周的天气,选个更适合旅游的城市"
实现方案:
- 开发周天气预报工具
- 实现旅游建议生成工具
- 使用LLM自动分解任务:
- 调用周天气预报获取北京数据
- 调用周天气预报获取上海数据
- 将结果传递给旅游建议工具
对应的prompt工程技巧:
python复制system_prompt = """你是一个高级旅行助手,可以调用多种工具解决问题。
当用户请求涉及多步骤时,请按以下格式处理:
1. 分析请求中的子任务
2. 为每个子任务选择合适工具
3. 综合所有结果生成最终建议"""
5. 性能优化实战技巧
5.1 减少不必要的工具调用
问题:LLM有时会过度使用工具,增加延迟和成本。
优化策略:
- 设置工具调用置信度阈值
- 对简单查询直接回答而不调用工具
- 实现工具调用缓存
示例优化代码:
python复制def chat(self, message: str):
# 先判断是否真的需要工具
if self._is_simple_query(message):
return self._direct_response(message)
# 否则走正常工具调用流程
...
def _is_simple_query(self, message: str) -> bool:
# 使用小型分类模型判断
simple_keywords = ["你好", "谢谢", "你是谁"]
return any(kw in message for kw in simple_keywords)
5.2 并行工具调用
当多个工具调用没有依赖关系时,可以并行执行:
python复制import asyncio
async def execute_tools_parallel(tool_calls):
tasks = []
for call in tool_calls:
tool = self.tools[call.function.name]
kwargs = json.loads(call.function.arguments)
tasks.append(asyncio.create_task(
self._execute_tool(tool, kwargs)
))
return await asyncio.gather(*tasks)
async def _execute_tool(self, tool, kwargs):
loop = asyncio.get_event_loop()
return await loop.run_in_executor(
None, tool.execute, **kwargs
)
6. 安全与权限控制
6.1 工具访问权限分级
实现基于角色的工具访问控制:
python复制class Tool:
def __init__(self, required_role: str = "user"):
self.required_role = required_role
def chat(self, message: str, user_role: str):
available_tools = [
tool for tool in self.tools.values()
if tool.required_role <= user_role
]
# 仅传递有权限的工具给LLM
...
6.2 输入输出过滤
防止敏感信息泄露:
python复制import re
def sanitize_input(text: str) -> str:
# 移除信用卡号等敏感信息
text = re.sub(r"\d{4}-\d{4}-\d{4}-\d{4}", "[CARD]", text)
# 其他过滤规则...
return text
7. 部署与监控
7.1 容器化部署
使用Docker打包Agent服务:
dockerfile复制FROM python:3.10-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
7.2 监控指标收集
关键监控指标包括:
- 工具调用成功率
- 平均响应时间
- 错误类型分布
使用Prometheus客户端实现:
python复制from prometheus_client import Counter, Histogram
TOOL_CALLS = Counter(
'agent_tool_calls_total',
'Total tool calls',
['tool_name', 'status']
)
RESPONSE_TIME = Histogram(
'agent_response_time_seconds',
'Response time distribution',
['tool_name']
)
class InstrumentedTool(Tool):
def execute(self, **kwargs):
start_time = time.time()
try:
result = super().execute(**kwargs)
TOOL_CALLS.labels(
tool_name=self.name,
status="success"
).inc()
return result
except Exception:
TOOL_CALLS.labels(
tool_name=self.name,
status="failure"
).inc()
raise
finally:
RESPONSE_TIME.labels(
tool_name=self.name
).observe(time.time() - start_time)
8. 项目演进方向
8.1 动态工具加载
实现无需重启的热更新工具:
python复制def load_tool_from_path(self, path: str):
spec = importlib.util.spec_from_file_location(
"custom_tool", path
)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
for attr in dir(module):
if attr.endswith("Tool"):
tool_class = getattr(module, attr)
if issubclass(tool_class, Tool):
self.register_tool(tool_class())
8.2 工具学习能力
让Agent能够自动记录和优化工具使用模式:
python复制class LearningAgent(ToolAgent):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.usage_patterns = defaultdict(int)
def chat(self, message: str):
# 记录工具使用频率
for tool in self._select_tools(message):
self.usage_patterns[tool.name] += 1
# 根据使用频率优化工具选择
...
9. 避坑指南
9.1 工具注册的常见错误
- 描述不准确:避免使用"处理数据"这类模糊描述,改为"将CSV文件转换为JSON格式"
- 参数缺失默认值:为可选参数设置合理的默认值
- 忽略错误码:明确每个工具可能返回的错误类型
9.2 调试技巧
- 记录完整对话历史:保存LLM的中间思考过程
- 可视化工具调用流程:使用流程图展示工具选择逻辑
- 模拟测试框架:构建端到端的测试用例
示例测试代码:
python复制def test_weather_tool():
agent = ToolAgent(tools=[WeatherTool()])
# 测试正常查询
response = agent.chat("北京现在天气如何?")
assert "温度" in response
# 测试错误处理
response = agent.chat("查询火星的天气")
assert "不支持" in response or "抱歉" in response
10. 资源消耗优化
10.1 上下文长度控制
策略:
- 自动总结冗长的工具结果
- 选择性遗忘早期对话
- 使用向量检索只保留相关历史
实现示例:
python复制from langchain.text_splitter import RecursiveCharacterTextSplitter
def summarize_tool_result(result: dict) -> str:
text = str(result)
if len(text) > 1000:
splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200
)
chunks = splitter.split_text(text)
return self._summarize_chunks(chunks)
return text
10.2 模型选择策略
根据任务复杂度动态选择LLM:
python复制def select_model(self, message: str) -> str:
complexity = self._estimate_complexity(message)
if complexity < 3:
return "gpt-3.5-turbo"
elif complexity < 7:
return "gpt-4"
else:
return "gpt-4-32k"
def _estimate_complexity(self, text: str) -> int:
# 基于文本长度、问题类型等估算
...
构建一个功能完善的工具调用型Agent就像教一个聪明的助手使用各种专业工具——需要清晰的指令、严谨的接口设计和充分的错误处理。经过这个项目的实践,我发现最关键的不仅是技术实现,更是对用户意图的精准把握和工具的合理抽象。当看到Agent能够自动选择正确的工具并完美完成任务时,那种成就感绝对值得所有的调试和优化工作。
