LangChain 系列·(七):Tools——给大模型加上手脚
LangChain 系列 · 第七篇:Tools——给大模型加上手脚
🎯 适合人群:已掌握 LangChain 基础与 RAG,想让 LLM 能够调用外部函数、查询实时数据的工程师
⏱️ 阅读时间:约 25 分钟
💬 本文从"LLM 的能力边界"切入,介绍 Tool 的本质与定义方式,深入 Tool Calling 的底层机制,并处理结构化输入与错误场景
一、LLM 的能力边界
大语言模型本质上是一个文本变换函数:接收文本,输出文本。这个设计带来了两个天然局限:
局限一:知识截止日期
模型的训练数据有截止时间。问它"今天的天气"或"最新的 GPU 价格",它只能猜测或拒绝回答。
局限二:无法执行操作
模型无法直接查数据库、发邮件、调 API、执行代码。它能"说"如何做,但"做"本身超出了它的能力范围。
Tool(工具) 解决的正是这两个问题——它是一座桥梁,让 LLM 能够在生成回复的过程中,主动调用外部函数获取实时数据或触发实际操作。
Without Tools:
User ──► LLM ──► Answer
(limited to training data)
With Tools:
User ──► LLM ──► Tool Call ──► External System
▲ │
└──────── Result ◄─────────┘
LLM ──► Final Answer
📝 Tool Calling 与 Function Calling 是同一概念的不同称呼。OpenAI 早期文档称之为 Function Calling,后统一改名为 Tool Calling;LangChain 使用 Tool 作为统一抽象。本文统一使用 Tool Calling。
二、Tool 的本质
Tool 的本质是:函数 + 描述。
LLM 不会直接执行代码——它只是根据描述决定"调用哪个工具、传什么参数",实际执行由 Python 运行时完成。因此 Tool 的两个要素缺一不可:
| 要素 | 作用 | 影响 |
|---|---|---|
| 函数本体 | 实际执行逻辑 | 决定工具能做什么 |
| 描述(name + description) | 告知 LLM 工具的用途和参数含义 | 决定 LLM 是否选择调用该工具、如何构造参数 |
描述写得不清晰,LLM 就可能选错工具或传错参数——这是 Tool 开发中最常见的问题之一。
三、定义 Tool:三种方式
3.1 @tool 装饰器(最常用)
@tool 是最简洁的定义方式,直接将普通 Python 函数转换为 LangChain Tool:
from langchain_core.tools import tool
@tool
def get_weather(city: str) -> str:
"""获取指定城市的当前天气信息。
Args:
city: 城市名称,如"北京"、"上海"、"深圳"
"""
# 实际场景中这里会调用天气 API
# 示例中返回模拟数据
weather_data = {
"北京": "晴,气温 22°C,东南风 3 级",
"上海": "多云,气温 25°C,东风 2 级",
"深圳": "阵雨,气温 28°C,南风 4 级",
}
return weather_data.get(city, f"暂无 {city} 的天气数据")
@tool 从三个地方自动提取 Tool 的元数据:
print(get_weather.name) # "get_weather"
print(get_weather.description) # "获取指定城市的当前天气信息。\n\nArgs:\n city: ..."
print(get_weather.args) # {'city': {'title': 'City', 'type': 'string'}}
⚠️ docstring 是工具描述的来源,写给 LLM 看,不是给开发者看的。描述要准确反映工具的用途和参数含义,含糊的描述会直接导致 LLM 调用出错。
可以用 name 和 description 参数覆盖自动提取的值:
@tool(name="weather_query", description="查询实时天气,输入城市中文名,返回天气描述字符串")
def get_weather(city: str) -> str:
...
3.2 StructuredTool(多参数场景)
当工具需要多个参数时,推荐用 StructuredTool.from_function 显式管理参数结构:
from langchain_core.tools import StructuredTool
from pydantic import BaseModel, Field
class SearchInput(BaseModel):
query: str = Field(description="搜索关键词")
max_results: int = Field(default=5, description="返回结果数量,1~10 之间")
def search_documents(query: str, max_results: int = 5) -> list[str]:
"""在知识库中搜索相关文档片段。"""
# 实际逻辑
return [f"搜索结果 {i+1}: 关于 '{query}' 的内容..." for i in range(max_results)]
search_tool = StructuredTool.from_function(
func=search_documents,
name="search_documents",
description="在内部知识库中搜索相关文档,返回最相关的文档片段列表",
args_schema=SearchInput,
)
args_schema 中的 Field(description=...) 会被 LLM 读取,用于理解每个参数的含义——这比函数签名的类型注解提供了更多语义信息。
3.3 BaseTool 子类(需要状态或复杂初始化)
对于需要持有数据库连接、API 客户端等有状态对象的工具,继承 BaseTool:
from langchain_core.tools import BaseTool
from pydantic import BaseModel, Field
class DatabaseQueryInput(BaseModel):
sql: str = Field(description="要执行的 SQL 查询语句,只允许 SELECT")
class DatabaseQueryTool(BaseTool):
name: str = "database_query"
description: str = "执行 SQL SELECT 查询,返回结果列表。只支持只读查询。"
args_schema: type[BaseModel] = DatabaseQueryInput
db_connection: object = None # 持有数据库连接
def _run(self, sql: str) -> str:
if not sql.strip().upper().startswith("SELECT"):
return "Error: 只允许 SELECT 查询"
# self.db_connection.execute(sql)
return f"查询结果(模拟): [{sql}]"
# 初始化时注入连接
db_tool = DatabaseQueryTool(db_connection=None)
四、Tool Calling 的底层机制
理解 Tool Calling 的执行流程,是排查问题的关键。
4.1 完整执行流程
Step 1: Bind Tools
Developer binds tools to model
model_with_tools = model.bind_tools([tool1, tool2])
Step 2: LLM Decision
User message arrives → LLM decides:
┌─────────────────────────────────────┐
│ Should I call a tool? │
│ YES → output AIMessage with │
│ tool_calls field populated │
│ NO → output normal text reply │
└─────────────────────────────────────┘
Step 3: Tool Execution
Your code inspects tool_calls,
finds matching tool, executes it,
wraps result in ToolMessage
Step 4: Second LLM Call
Conversation history + ToolMessage
sent back to LLM → final answer
4.2 逐步拆解看消息流
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage
@tool
def multiply(a: int, b: int) -> int:
"""将两个整数相乘并返回结果。"""
return a * b
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# Step 1: 绑定工具
model_with_tools = model.bind_tools([multiply])
# Step 2: 发送请求
response = model_with_tools.invoke([HumanMessage(content="12 乘以 34 等于多少?")])
print(type(response)) # <class 'langchain_core.messages.ai.AIMessage'>
print(response.content) # "" (LLM 决定调工具时 content 通常为空)
print(response.tool_calls)
# [{'name': 'multiply', 'args': {'a': 12, 'b': 34}, 'id': 'call_abc123', 'type': 'tool_call'}]
此时 LLM 还没有给出最终答案,它只是发出了一个"工具调用请求"。接下来需要执行工具并把结果还给 LLM:
# Step 3: 执行工具,构造 ToolMessage
tool_call = response.tool_calls[0]
tool_result = multiply.invoke(tool_call["args"])
# ToolMessage 是工具执行结果的载体,tool_call_id 用于关联对应的请求
tool_message = ToolMessage(
content=str(tool_result),
tool_call_id=tool_call["id"],
)
# Step 4: 带上完整对话历史,再次调用 LLM 得到最终答案
messages = [
HumanMessage(content="12 乘以 34 等于多少?"),
response, # AIMessage with tool_calls
tool_message, # ToolMessage with result
]
final_response = model_with_tools.invoke(messages)
print(final_response.content)
# "12 乘以 34 等于 408。"
💡
ToolMessage在第一篇的消息类型表中有介绍(表示工具调用的返回结果,需关联tool_call_id)。tool_call_id是 LLM 在发出调用请求时自动生成的唯一 ID,用于在多工具并发调用时正确匹配请求与结果。
4.3 Tool Calling 的本质是"协议"
LLM 并不真正"调用"工具——它只是输出一段结构化数据,告诉调用方"我想调用哪个工具、传什么参数":
# LLM 实际输出的是这样的结构化信息(简化表示):
{
"tool_calls": [
{
"name": "multiply",
"arguments": {"a": 12, "b": 34}
}
]
}
LangChain 将这个协议的解析、执行、结果回传封装为统一的 API,开发者通常不需要手动处理消息流——这正是 Agent(下一篇)所做的事情。
五、结构化工具输入:用 Pydantic 约束参数
对于参数复杂的工具,直接在函数签名上写类型注解有时不够——LLM 可能传入不合法的值。Pydantic 的 BaseModel 提供了更严格的参数校验与描述能力。
📝 Pydantic 是 Python 的数据验证库,通过类型注解自动校验数据格式。LangChain 广泛使用 Pydantic 进行参数定义和校验。
Field是 Pydantic 提供的字段元数据描述器,description参数会被 LangChain 传递给 LLM 作为参数说明。
from pydantic import BaseModel, Field
from langchain_core.tools import tool
from typing import Literal
class StockQueryInput(BaseModel):
ticker: str = Field(
description="股票代码,如 'AAPL'、'TSLA'、'600519'(A 股加市场前缀如 'SSE:600519')"
)
metric: Literal["price", "pe_ratio", "market_cap"] = Field(
description="查询指标:price=当前股价,pe_ratio=市盈率,market_cap=市值"
)
currency: str = Field(
default="CNY",
description="返回货币单位,支持 CNY 和 USD,默认 CNY"
)
@tool(args_schema=StockQueryInput)
def query_stock(ticker: str, metric: str, currency: str = "CNY") -> str:
"""查询股票的实时财务数据,支持 A 股和美股。"""
# 实际场景调用金融数据 API
return f"{ticker} 的 {metric} 数据({currency}):模拟值 100.00"
使用 Literal 类型约束枚举值后,LLM 只会传入合法选项,避免了参数非法导致的运行时错误。
参数描述的质量直接影响 LLM 的调用准确率,对比以下两种写法:
# ❌ 描述不足,LLM 无法判断该传什么
ticker: str = Field(description="股票代码")
# ✅ 给出格式说明和示例,LLM 能正确构造参数
ticker: str = Field(
description="股票代码,如 'AAPL'、'TSLA'、'600519'(A 股加市场前缀如 'SSE:600519')"
)
六、内置工具
LangChain 提供了一批开箱即用的工具,无需自己实现:
| 工具类 | 包 | 用途 |
|---|---|---|
TavilySearchResults |
langchain-community |
Tavily 实时网络搜索,专为 LLM 场景优化 |
WikipediaQueryRun |
langchain-community |
查询 Wikipedia 摘要 |
PythonREPLTool |
langchain-experimental |
在沙箱中执行 Python 代码 |
DuckDuckGoSearchRun |
langchain-community |
DuckDuckGo 网络搜索(无需 API Key) |
ShellTool |
langchain-community |
执行 Shell 命令(⚠️ 生产环境慎用) |
HumanInputRun |
langchain-community |
请求人工输入(Human-in-the-loop) |
pip install langchain-community tavily-python wikipedia
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_community.tools import WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper
# Tavily 搜索(需要 TAVILY_API_KEY)
search_tool = TavilySearchResults(
max_results=3, # 返回最多 3 条结果
search_depth="advanced", # basic 或 advanced
include_answer=True, # 返回 Tavily 的直接答案摘要
)
# Wikipedia 查询
wiki_tool = WikipediaQueryRun(
api_wrapper=WikipediaAPIWrapper(
top_k_results=2, # 返回最相关的 2 篇文章
doc_content_chars_max=500, # 每篇摘要最多 500 字符
lang="zh", # 中文 Wikipedia
)
)
# 直接调用(不经过 LLM)
result = search_tool.invoke("LangChain 最新版本是什么")
print(result)
⚠️
PythonREPLTool和ShellTool会直接在服务器执行代码或命令,生产环境必须在完全隔离的沙箱中运行,否则存在严重的安全风险。
七、错误处理:ToolException
工具执行可能失败——API 超时、参数非法、权限不足……不处理这些错误,Agent 会陷入异常状态。
7.1 handle_tool_error:让 LLM 从错误中恢复
from langchain_core.tools import tool, ToolException
@tool
def query_database(table: str, id: int) -> str:
"""查询数据库中指定表的记录。
Args:
table: 表名,只允许查询 users 和 orders 表
id: 记录 ID,正整数
"""
allowed_tables = {"users", "orders"}
if table not in allowed_tables:
# 抛出 ToolException 而非普通异常
# ToolException 会被 LangChain 捕获并转换为 ToolMessage 返回给 LLM
raise ToolException(
f"表 '{table}' 不存在或无权访问。可查询的表:{', '.join(allowed_tables)}"
)
return f"查询结果:表 {table} 中 id={id} 的记录"
在绑定工具时,设置 handle_tool_error=True:
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# handle_tool_error=True:将 ToolException 的消息作为 ToolMessage 内容返回给 LLM
# LLM 看到错误信息后,可以自主调整参数重试,而不是直接崩溃
model_with_tools = model.bind_tools(
[query_database.with_config({"handle_tool_error": True})]
)
也可以传入自定义错误处理函数:
def handle_db_error(error: ToolException) -> str:
return f"数据库查询失败,请检查参数后重试。错误详情:{str(error)}"
db_tool = query_database.copy(update={"handle_tool_error": handle_db_error})
7.2 错误处理策略对比
handle_tool_error 的三种配置:
False(默认) → ToolException 直接抛出,程序崩溃
适用于:开发调试阶段
True → 将异常信息作为 ToolMessage 返回给 LLM
LLM 会读取错误信息并决定是否重试或换策略
适用于:生产环境大多数情况
callable → 自定义函数处理异常,返回字符串给 LLM
适用于:需要日志记录、脱敏错误信息等场景
八、完整示例:多工具助手
将所有概念整合,构建一个能查天气、搜索和计算的助手:
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool, ToolException
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
load_dotenv()
# --- 定义工具 ---
@tool
def get_weather(city: str) -> str:
"""查询指定城市的当前天气。支持国内主要城市,返回天气描述字符串。
Args:
city: 城市中文名,如"北京"、"上海"、"广州"
"""
data = {
"北京": "晴,22°C,东南风 3 级,空气质量良",
"上海": "多云,25°C,东风 2 级,空气质量优",
"广州": "阵雨,30°C,南风 4 级,空气质量良",
}
if city not in data:
raise ToolException(f"暂不支持城市:{city},支持的城市:{', '.join(data.keys())}")
return data[city]
@tool
def calculate(expression: str) -> str:
"""计算数学表达式,支持基本四则运算和幂运算。
Args:
expression: 数学表达式字符串,如 "2 ** 10"、"(3 + 5) * 12"
"""
try:
# 限制只允许数学运算,禁止任意代码执行
allowed_names = {"__builtins__": {}}
result = eval(expression, allowed_names)
return str(result)
except Exception as e:
raise ToolException(f"表达式计算失败:{e},请检查语法")
tools = [get_weather, calculate]
# --- 运行 Tool Calling 循环 ---
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
model_with_tools = model.bind_tools(tools)
# 构建工具名称到工具对象的映射,用于快速查找
tool_map = {t.name: t for t in tools}
def run_with_tools(user_input: str) -> str:
messages = [HumanMessage(content=user_input)]
while True:
response = model_with_tools.invoke(messages)
messages.append(response)
# 没有工具调用,直接返回文本答案
if not response.tool_calls:
return response.content
# 执行所有工具调用
for tool_call in response.tool_calls:
tool_name = tool_call["name"]
tool_args = tool_call["args"]
try:
tool_result = tool_map[tool_name].invoke(tool_args)
except ToolException as e:
tool_result = f"工具执行失败:{e}"
messages.append(ToolMessage(
content=str(tool_result),
tool_call_id=tool_call["id"],
))
# 继续循环,将工具结果还给 LLM
# 测试
print(run_with_tools("北京今天天气怎么样?2 的 10 次方是多少?"))
九、常见坑与最佳实践
坑一:工具描述写给开发者看,而非 LLM
# ❌ 描述对 LLM 无意义
@tool
def send_email(to: str, subject: str, body: str) -> str:
"""发邮件""" # LLM 不知道参数格式要求
# ✅ 描述说清楚参数格式和限制
@tool
def send_email(to: str, subject: str, body: str) -> str:
"""向指定邮箱发送一封邮件。
Args:
to: 收件人邮箱地址,标准 email 格式,如 "user@example.com"
subject: 邮件主题,不超过 100 个字符
body: 邮件正文,支持纯文本,不超过 5000 字符
"""
坑二:工具返回值过长或格式不统一
# ❌ 直接返回原始 API 响应,可能包含大量无关字段
def search_news(query: str) -> dict:
result = requests.get(f"https://api.example.com/news?q={query}")
return result.json() # 可能有几千字段,LLM 处理效率低
# ✅ 提取关键信息,返回简洁的字符串
def search_news(query: str) -> str:
result = requests.get(f"https://api.example.com/news?q={query}").json()
articles = result.get("articles", [])[:3] # 只取前 3 条
return "\n".join([f"- {a['title']}: {a['summary']}" for a in articles])
坑三:多工具同名或描述高度相似
# ❌ 两个工具功能高度重叠,LLM 会随机选择
@tool
def search_web(query: str) -> str:
"""搜索互联网上的信息"""
@tool
def search_internet(query: str) -> str:
"""在网上搜索内容"""
# ✅ 工具名和描述要明确区分用途
@tool
def search_recent_news(query: str) -> str:
"""搜索最近 7 天内的新闻资讯,适合查询时事、热点事件"""
@tool
def search_technical_docs(query: str) -> str:
"""搜索技术文档和 API 参考,适合查询框架用法、函数签名"""
坑四:没有限制工具的副作用
# ❌ 工具可以执行任意 SQL,包括 DELETE/UPDATE
@tool
def run_sql(sql: str) -> str:
"""执行 SQL 语句"""
return db.execute(sql)
# ✅ 明确限制只读操作,防止 LLM 误触发写操作
@tool
def run_sql(sql: str) -> str:
"""执行只读 SQL SELECT 查询,返回结果列表。不支持 INSERT/UPDATE/DELETE。"""
if not sql.strip().upper().startswith("SELECT"):
raise ToolException("只允许 SELECT 查询,拒绝执行写操作")
return db.execute(sql)
坑五:工具调用循环没有退出条件
手动实现 Tool Calling 循环时,如果 LLM 的工具调用结果触发了下一次调用,而下一次结果又触发了再下一次,就会产生无限循环:
# ❌ 没有最大迭代次数限制
while True:
response = model.invoke(messages)
if not response.tool_calls:
return response.content
# 执行工具...
# ✅ 加入最大迭代次数保护
MAX_ITERATIONS = 10
for _ in range(MAX_ITERATIONS):
response = model.invoke(messages)
if not response.tool_calls:
return response.content
# 执行工具...
return "超过最大工具调用次数,请简化问题"
十、总结
| 概念 | 核心要点 |
|---|---|
| Tool 的本质 | 函数 + 描述;LLM 选择工具并构造参数,Python 执行,结果通过 ToolMessage 返回 |
@tool 装饰器 |
最简定义方式,docstring 自动成为工具描述,是 LLM 选择调用的依据 |
Pydantic args_schema |
为复杂参数提供严格校验和精细描述,Field(description=...) 直接影响 LLM 的参数构造质量 |
bind_tools |
将工具列表附加到模型,模型输出的 AIMessage 将携带 tool_calls 字段 |
ToolMessage |
工具执行结果的载体,通过 tool_call_id 与对应请求关联 |
ToolException |
工具内部的受控错误,配合 handle_tool_error=True 让 LLM 能从错误中自主恢复 |
🎯 工具开发最容易忽视的环节是描述质量——模型能力再强,也无法从"发邮件"这三个字里知道参数格式是什么。把 docstring 写清楚,是让工具被正确调用的最低成本投入。
参考资料
下期预告
有了工具,下一步是让 LLM 自主决定何时调用哪个工具,并在调用结果的基础上继续推理,直到完成任务。
第八篇《Agent:让 AI 自己做决策》 将介绍 ReAct(Reasoning + Acting)模式的原理,以及 LangChain 的 create_tool_calling_agent + AgentExecutor 如何将工具调用循环封装为一个可复用的 Agent——这是从"会用工具"到"自主完成任务"的关键跃升。
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。
更多推荐


所有评论(0)