Agent工具调用：让AI拥有超能力-开发者社区

🛠️Function Calling | 工具注册 + 智能选择 + 结果整合 | LangChain Tools实战 | 完整项目代码

📖 为什么需要工具调用？

LLM的局限性

# 纯LLM - 只能基于训练数据回答 llm = ChatOpenAI(model="gpt-4") response = llm.invoke("今天北京的天气怎么样？") # 输出: "抱歉，我无法获取实时天气信息" ❌ response = llm.invoke("计算 12345 * 67890") # 输出: 可能计算错误 ❌ response = llm.invoke("查询张三的数据库记录") # 输出: "我无法访问你的数据库" ❌

问题：

❌ 无法获取实时信息
❌ 数学计算可能出错
❌ 无法访问外部系统
❌ 无法执行实际操作

工具调用的优势

# Agent + 工具 - 可以调用外部能力 agent = create_agent(llm, tools=[ weather_tool, # 天气查询 calculator_tool, # 精确计算 database_tool # 数据库访问 ]) response = agent.run("今天北京的天气怎么样？") # 自动调用weather_tool → "北京今天晴，25°C" ✅ response = agent.run("计算 12345 * 67890") # 自动调用calculator_tool → "838102050" ✅ response = agent.run("查询张三的信息") # 自动调用database_tool → "张三，30岁，工程师" ✅

优势：

✅ 获取实时数据
✅ 精确计算
✅ 访问外部系统
✅ 执行实际操作

🏗️ 工具调用架构

核心流程

用户输入 │ ▼ ┌──────────────┐ │ LLM分析 │ ← 理解用户需求，判断是否需要工具 └──────┬───────┘ │ ├─ 需要工具 ──→ ┌──────────────┐ │ │ 工具选择 │ ← 从可用工具中选择 │ └──────┬───────┘ │ │ │ ▼ │ ┌──────────────┐ │ │ 工具执行 │ ← 调用工具函数 │ └──────┬───────┘ │ │ │ ▼ │ ┌──────────────┐ │ │ 结果整合 │ ← LLM整合工具结果 │ └──────┬───────┘ │ │ └─ 不需要工具 ────────┐│ ▼▼ ┌──────────────┐ │ 最终回复 │ └──────────────┘

工具调用模式

模式	说明	适用场景
单次调用	调用一个工具后直接回复	简单查询
多次调用	连续调用多个工具	复杂任务
并行调用	同时调用多个独立工具	提高效率
条件调用	根据前一个结果决定下一步	决策流程

🛠️ 核心技术实现

方案1：LangChain Tools基础用法

安装依赖

pip install langchain langchain-openai

1. 自定义工具

from langchain.tools import tool @tool def calculate(expression: str) -> str: """计算数学表达式 Args: expression: 数学表达式，如 "2 + 2" 或 "12345 * 67890" Returns: 计算结果 """ try: # 安全计算（只允许基本运算） result = eval(expression, {"__builtins__": {}}, {}) return f"计算结果: {result}" except Exception as e: return f"计算错误: {str(e)}" @tool def get_weather(city: str) -> str: """查询城市天气 Args: city: 城市名称，如 "北京" 或 "上海" Returns: 天气信息 """ # 模拟天气API weather_data = { "北京": "晴，25°C，空气质量优", "上海": "多云，28°C，湿度65%", "广州": "小雨，30°C，湿度80%" } return weather_data.get(city, f"未找到{city}的天气信息") @tool def search_database(query: str) -> str: """搜索数据库记录 Args: query: 搜索关键词 Returns: 搜索结果 """ # 模拟数据库 database = { "张三": {"age": 30, "occupation": "工程师", "city": "北京"}, "李四": {"age": 25, "occupation": "设计师", "city": "上海"}, "王五": {"age": 35, "occupation": "产品经理", "city": "广州"} } if query in database: info = database[query] return f"{query}的信息：年龄{info['age']}岁，职业{info['occupation']}，所在城市{info['city']}" else: return f"未找到'{query}'的记录" # 创建工具列表 tools = [calculate, get_weather, search_database]

2. 创建Agent

from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate # 创建LLM llm = ChatOpenAI(model="gpt-4", temperature=0) # 创建提示词模板 prompt = ChatPromptTemplate.from_messages([ ("system", """你是一个有用的助手，可以使用各种工具来帮助用户。 可用的工具： - calculate: 计算数学表达式 - get_weather: 查询城市天气 - search_database: 搜索数据库记录 当用户的问题需要这些信息时，请使用相应的工具。"""), ("human", "{input}"), ("placeholder", "{agent_scratchpad}") ]) # 创建Agent agent = create_tool_calling_agent(llm, tools, prompt) # 创建执行器 agent_executor = AgentExecutor( agent=agent, tools=tools, verbose=True, # 显示详细过程 handle_parsing_errors=True ) # 使用Agent result = agent_executor.invoke({ "input": "今天北京的天气怎么样？" }) print(result["output"]) # 输出: 北京今天晴，25°C，空气质量优 result = agent_executor.invoke({ "input": "计算 12345 乘以 67890" }) print(result["output"]) # 输出: 计算结果: 838102050 result = agent_executor.invoke({ "input": "帮我查一下张三的信息" }) print(result["output"]) # 输出: 张三的信息：年龄30岁，职业工程师，所在城市北京

方案2：内置工具集

常用工具

from langchain_community.tools import ( DuckDuckGoSearchRun, # 网络搜索 WikipediaQueryRun, # 维基百科 ArxivQueryRun, # 学术论文 ShellTool, # 命令行执行 PythonREPLTool, # Python代码执行 ) # 网络搜索工具 search_tool = DuckDuckGoSearchRun() result = search_tool.run("Python最新版本") print(result) # 维基百科工具 wiki_tool = WikipediaQueryRun() result = wiki_tool.run("人工智能") print(result) # Python执行工具 python_tool = PythonREPLTool() result = python_tool.run("print(2 + 2)") print(result) # 输出: 4

组合使用

# 创建强大的Agent tools = [ DuckDuckGoSearchRun(), WikipediaQueryRun(), PythonREPLTool(), calculate, get_weather ] agent = create_tool_calling_agent(llm, tools, prompt) agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True) # 复杂查询 result = agent_executor.invoke({ "input": "搜索Python的最新版本，然后计算版本号中的数字之和" })

方案3：自定义工具类

高级工具实现

from langchain.tools import BaseTool from typing import Type from pydantic import BaseModel, Field class WeatherInput(BaseModel): """天气查询输入""" city: str = Field(description="城市名称，如'北京'、'上海'") date: str = Field(description="日期，格式YYYY-MM-DD，默认为今天", default="today") class WeatherTool(BaseTool): """天气查询工具""" name: str = "get_weather" description: str = "查询指定城市的天气信息" args_schema: Type[BaseModel] = WeatherInput def _run(self, city: str, date: str = "today") -> str: """同步执行""" # 这里可以调用真实的天气API weather_data = { "北京": {"temp": 25, "condition": "晴", "humidity": 40}, "上海": {"temp": 28, "condition": "多云", "humidity": 65}, "广州": {"temp": 30, "condition": "小雨", "humidity": 80} } if city in weather_data: data = weather_data[city] return f"{city}{date}天气：{data['condition']}，温度{data['temp']}°C，湿度{data['humidity']}%" else: return f"未找到{city}的天气信息" async def _arun(self, city: str, date: str = "today") -> str: """异步执行""" return self._run(city, date) class DatabaseTool(BaseTool): """数据库查询工具""" name: str = "query_database" description: str = "查询数据库中的用户信息" def _run(self, user_name: str) -> str: """查询用户信息""" # 模拟数据库连接 database = { "张三": {"age": 30, "occupation": "工程师", "salary": 20000}, "李四": {"age": 25, "occupation": "设计师", "salary": 15000}, } if user_name in database: info = database[user_name] return f"用户{user_name}：年龄{info['age']}，职业{info['occupation']}，薪资{info['salary']}元" else: return f"未找到用户'{user_name}'" async def _arun(self, user_name: str) -> str: return self._run(user_name) # 使用自定义工具 tools = [WeatherTool(), DatabaseTool()]

方案4：工具路由与优化

智能工具选择

from typing import List, Dict class ToolRouter: """工具路由器 - 智能选择最合适的工具""" def __init__(self, tools: List[BaseTool]): self.tools = tools self.tool_index = {tool.name: tool for tool in tools} # 工具关键词索引（用于快速匹配） self.keyword_index = {} for tool in tools: keywords = self._extract_keywords(tool.description) for keyword in keywords: if keyword not in self.keyword_index: self.keyword_index[keyword] = [] self.keyword_index[keyword].append(tool.name) def select_tools(self, query: str, top_k: int = 3) -> List[str]: """根据查询选择相关工具 Args: query: 用户查询 top_k: 返回工具数量 Returns: 工具名称列表 """ # 提取查询关键词 query_keywords = self._extract_keywords(query) # 计算工具相关性得分 tool_scores = {} for keyword in query_keywords: if keyword in self.keyword_index: for tool_name in self.keyword_index[keyword]: tool_scores[tool_name] = tool_scores.get(tool_name, 0) + 1 # 按得分排序 sorted_tools = sorted(tool_scores.items(), key=lambda x: x[1], reverse=True) return [tool_name for tool_name, score in sorted_tools[:top_k]] def _extract_keywords(self, text: str) -> List[str]: """提取关键词（简化版）""" # 实际应用中可以使用NLP技术 keywords = ["天气", "计算", "搜索", "查询", "数据库", "用户"] return [kw for kw in keywords if kw in text] # 使用示例 router = ToolRouter([WeatherTool(), DatabaseTool(), calculate]) relevant_tools = router.select_tools("北京今天的天气怎么样？") print(relevant_tools) # 输出: ['get_weather']

工具缓存

from functools import lru_cache import time class CachedTool(BaseTool): """带缓存的工具""" def __init__(self, base_tool: BaseTool, cache_ttl: int = 3600): self.base_tool = base_tool self.cache_ttl = cache_ttl self.cache = {} self.cache_timestamps = {} def _run(self, *args, **kwargs) -> str: """执行工具（带缓存）""" # 生成缓存键 cache_key = str((args, kwargs)) # 检查缓存 if cache_key in self.cache: timestamp = self.cache_timestamps[cache_key] if time.time() - timestamp < self.cache_ttl: print(f"✅ 使用缓存结果") return self.cache[cache_key] # 执行工具 result = self.base_tool._run(*args, **kwargs) # 更新缓存 self.cache[cache_key] = result self.cache_timestamps[cache_key] = time.time() return result

🎯 实战案例：智能研究助手

系统架构

用户提问 │ ▼ ┌──────────────┐ │ 意图识别 │ ← 判断需要什么类型的信息 └──────┬───────┘ │ ├─ 事实查询 ──→ 搜索引擎 ├─ 学术问题 ──→ 论文数据库 ├─ 计算问题 ──→ 计算器 ├─ 代码问题 ──→ Python执行器 └─ 其他 ─────→ LLM直接回答 │ ▼ ┌──────────────┐ │ 结果整合 │ ← 综合多个来源 └──────┬───────┘ │ ▼ 最终答案

完整实现

from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain_community.tools import DuckDuckGoSearchRun from langchain.tools import tool # ==================== 定义工具 ==================== @tool def calculate_expression(expression: str) -> str: """计算数学表达式""" try: result = eval(expression, {"__builtins__": {}}, {}) return str(result) except: return "计算错误" @tool def search_web(query: str) -> str: """搜索网络信息""" search = DuckDuckGoSearchRun() return search.run(query) @tool def summarize_text(text: str, max_length: int = 200) -> str: """总结文本内容""" # 简单总结（实际应用可以用LLM） words = text.split() if len(words) > max_length: return " ".join(words[:max_length]) + "..." return text # ==================== 创建Agent ==================== def create_research_assistant(): """创建智能研究助手""" llm = ChatOpenAI(model="gpt-4", temperature=0) tools = [ calculate_expression, search_web, summarize_text ] prompt = ChatPromptTemplate.from_messages([ ("system", """你是一个智能研究助手，可以帮助用户： 1. 搜索网络信息（使用search_web） 2. 进行数学计算（使用calculate_expression） 3. 总结长文本（使用summarize_text） 根据用户的需求选择合适的工具。如果一个问题需要多个步骤， 可以依次调用多个工具。 回答要准确、简洁、有用。"""), ("human", "{input}"), ("placeholder", "{agent_scratchpad}") ]) agent = create_tool_calling_agent(llm, tools, prompt) agent_executor = AgentExecutor( agent=agent, tools=tools, verbose=True, max_iterations=5, # 最大工具调用次数 handle_parsing_errors=True ) return agent_executor # ==================== 使用示例 ==================== def example_research_assistant(): """研究助手使用示例""" print("="*60) print("智能研究助手演示") print("="*60) assistant = create_research_assistant() # 测试用例 test_cases = [ "计算 2的10次方是多少？", "搜索Python的最新版本信息", "帮我总结一下：人工智能是计算机科学的一个分支，它企图了解智能的实质，并生产出一种新的能以人类智能相似的方式做出反应的智能机器。人工智能的研究包括机器人、语言识别、图像识别、自然语言处理和专家系统等。" ] for i, query in enumerate(test_cases, 1): print(f"\n[问题 {i}]") print(f"👤 用户: {query}") print("-" * 60) result = assistant.invoke({"input": query}) print(f"🤖 AI: {result['output']}") print("="*60) if __name__ == "__main__": example_research_assistant()

📊 性能优化技巧

1. 并行工具调用

import asyncio async def parallel_tool_calls(agent_executor, queries: List[str]): """并行执行多个查询""" tasks = [ agent_executor.ainvoke({"input": query}) for query in queries ] results = await asyncio.gather(*tasks) return results # 使用 queries = [ "北京的天气", "上海的天气", "广州的天气" ] results = asyncio.run(parallel_tool_calls(agent_executor, queries))

2. 工具预筛选

def prefilter_tools(query: str, all_tools: List[BaseTool]) -> List[BaseTool]: """预筛选相关工具，减少LLM负担""" # 基于关键词快速过滤 relevant_keywords = { "天气": ["get_weather"], "计算": ["calculate"], "搜索": ["search_web"], "代码": ["python_repl"] } selected_tools = set() for keyword, tool_names in relevant_keywords.items(): if keyword in query: selected_tools.update(tool_names) # 返回匹配的工具 return [t for t in all_tools if t.name in selected_tools]

3. 结果缓存

class ToolCache: """工具结果缓存""" def __init__(self, max_size: int = 1000): self.cache = {} self.max_size = max_size def get(self, tool_name: str, args: tuple) -> Optional[str]: """获取缓存""" key = (tool_name, args) return self.cache.get(key) def set(self, tool_name: str, args: tuple, result: str): """设置缓存""" key = (tool_name, args) # LRU策略 if len(self.cache) >= self.max_size: oldest_key = next(iter(self.cache)) del self.cache[oldest_key] self.cache[key] = result

💡 最佳实践总结

1. 工具设计原则

原则	说明	示例
单一职责	每个工具只做一件事	天气工具只查天气
清晰描述	description要详细说明用途	"查询城市天气，返回温度和状况"
类型安全	使用Pydantic定义输入	args_schema = WeatherInput
错误处理	优雅处理异常情况	返回友好错误信息
幂等性	相同输入产生相同输出	避免副作用

2. 工具选择策略

# 好的工具描述 @tool def get_stock_price(stock_symbol: str) -> str: """获取股票实时价格。 参数: stock_symbol: 股票代码，如'AAPL'、'GOOGL' 返回: 股票的当前价格和涨跌幅 """ pass # 不好的工具描述 @tool def stock(x): """股票""" pass

3. 调试技巧

# 启用verbose查看详细过程 agent_executor = AgentExecutor( agent=agent, tools=tools, verbose=True # 显示每一步 ) # 查看中间步骤 result = agent_executor.invoke({"input": query}) print(result["intermediate_steps"]) # 工具调用历史

🔍 常见问题解答

Q1: 如何防止工具被滥用？

# 1. 限制调用次数 agent_executor = AgentExecutor( agent=agent, tools=tools, max_iterations=3 # 最多调用3次 ) # 2. 权限控制 def safe_execute(tool, args): if not check_permission(tool.name): raise PermissionError("无权使用该工具") return tool.run(args) # 3. 输入验证 def validate_input(tool_name, args): if tool_name == "shell": dangerous_commands = ["rm", "sudo", "dd"] if any(cmd in args for cmd in dangerous_commands): raise ValueError("禁止执行危险命令")

Q2: 如何处理工具调用失败？

# 1. 重试机制 for attempt in range(3): try: result = tool.run(args) break except Exception as e: if attempt == 2: result = f"工具调用失败: {str(e)}" # 2. 降级策略 try: result = premium_api.query(args) except: result = basic_api.query(args) # 降级到基础API # 3. 错误传递给LLM agent_executor = AgentExecutor( handle_parsing_errors=True # 自动处理解析错误 )

Q3: 如何评估工具调用的效果？

# 关键指标 metrics = { "tool_selection_accuracy": 0.92, # 工具选择准确率 "success_rate": 0.95, # 调用成功率 "avg_response_time": 2.5, # 平均响应时间（秒） "cost_per_query": 0.02 # 每次查询成本（美元） } # A/B测试 variant_a = AgentExecutor(tools=basic_tools) variant_b = AgentExecutor(tools=enhanced_tools) compare_performance(variant_a, variant_b)