大模型本身的知识是静态的,遇到实时数据、私有文档或企业内部接口时就会束手无策。LangChain的Tools机制正是为了解决这个问题而设计的:把一个函数或API封装成标准工具,让Agent在推理过程中自主决定何时调用、传什么参数。本文将以一个API检索工具为例,完整演示从工具定义、参数校验到接入Agent的全流程。

一、理解Tools的运行原理与接口约定
在LangChain中,一个Tool本质上是一个可以被大模型调用的函数,但光有函数还不够,模型需要知道这个工具是干什么的、需要哪些参数。所以每个Tool都由三部分构成:可执行的函数体、自然语言描述(description)以及参数结构定义(args_schema)。模型在决策时,只能看到后两者,函数体对它来说是黑盒。
这里有一个容易被忽视的关键点:工具描述的文案质量直接决定调用成功率。模型是根据描述文字来判断该不该用这个工具的,如果描述写得含糊,比如只写"用于检索数据",模型在多工具场景下就很难做出正确选择。好的描述应该说明工具的用途、适用场景,甚至可以举一个典型的输入例子。
参数校验通过Pydantic模型实现,也就是args_schema。它的作用有两个:一是把参数结构转换成模型能理解的功能调用规范(function calling schema),二是在实际调用前对入参做类型检查,避免模型生成的错误参数直接打崩你的业务代码。下面是一个最小的参数定义示例:
from pydantic import BaseModel, Field
class SearchInput(BaseModel):
"""API检索工具的输入参数"""
query: str = Field(description="检索关键词,例如:LangChain 工具调用")
top_k: int = Field(default=3, description="返回结果数量,默认3条")可以看到,Field中的description同样会被模型读到,字段写清楚能显著减少参数幻觉。默认值的设置也很讲究,给非核心参数加上合理默认值,模型调用时就只需要关注必填项,出错概率更低。
二、用类继承方式实现自定义检索工具
开发自定义工具有两条主流路径:@tool装饰器和继承BaseTool类。装饰器方式写起来快,适合简单场景;类继承方式结构清晰,支持异步、支持复用,适合正式项目。这里重点讲类继承方式。
继承BaseTool需要实现_run方法(同步)和可选的_arun方法(异步)。工具的name和description通过类属性声明,args_schema指向前面定义的Pydantic模型。下面是一个完整的API检索工具实现:
from langchain_core.tools import BaseTool
from typing import Type, Optional
import httpx
class ApiSearchInput(BaseModel):
query: str = Field(description="检索关键词")
top_k: int = Field(default=3, description="返回条数,默认3")
class ApiSearchTool(BaseTool):
name: str = "api_search"
description: str = (
"当需要查询实时资讯或外部知识库时使用该工具。"
"输入一个检索关键词,返回相关的结果列表。"
"例如用户询问最新技术动态时应优先调用本工具。"
)
args_schema: Type[BaseModel] = ApiSearchInput
def _run(self, query: str, top_k: int = 3) -> str:
"""同步执行:请求检索API并格式化结果"""
resp = httpx.get(
"https://api.ipipp.com/search",
params={"q": query, "k": top_k},
timeout=10
)
resp.raise_for_status()
items = resp.json().get("items", [])
if not items:
return "未检索到相关结果"
# 拼接成模型易读的文本格式
return "\n\n".join(
f"【{i+1}】{item['title']}\n{item['snippet']}"
for i, item in enumerate(items[:top_k])
)
async def _arun(self, query: str, top_k: int = 3) -> str:
"""异步版本,避免阻塞事件循环"""
async with httpx.AsyncClient() as client:
resp = await client.get(
"https://api.ipipp.com/search",
params={"q": query, "k": top_k},
timeout=10
)
resp.raise_for_status()
return resp.text有几个实现细节值得注意。第一,返回值务必是字符串,而且是格式化的纯文本,不要直接返回原始JSON,因为模型解析长JSON的效率不高且容易遗漏信息。第二,_run里一定要做异常兜底,网络超时、接口报错时返回一段友好提示,让模型知道调用失败的原因,而不是让异常直接抛出去中断整个Agent流程。第三,如果项目中用了异步框架,记得实现_arun,否则LangChain会报错提示不支持异步调用。
再简单对比一下装饰器写法。用@tool装饰一个函数,一行代码就能注册工具,参数描述通过docstring传递:
from langchain_core.tools import tool
@tool
def api_search(query: str, top_k: int = 3) -> str:
"""当需要查询实时资讯或外部知识库时使用该工具,输入检索关键词返回结果列表。"""
# 函数体与上面的_run一致
...这种方式的缺点是扩展性弱:无法维护内部状态、异步支持受限、参数描述只能靠docstring表达力有限。团队项目中建议统一用类继承,简单脚本用装饰器即可。
三、接入Agent并验证调用链路
工具写好后,接入Agent的过程比较标准。以create_tool_calling_agent为例,把工具列表传给Agent,再通过AgentExecutor执行。观察执行日志是这一步的重点,verbose=True能打印出完整的思考链路:
from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate
llm = ChatOpenAI(model="gpt-4o", temperature=0)
tools = [ApiSearchTool()]
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个知识检索助手,遇到不确定的问题请先调用工具查询。"),
("human", "{input}"),
("placeholder", "{agent_scratchpad}")
])
agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
result = executor.invoke({"input": "帮我查一下LangChain工具开发的最新实践"})
print(result["output"])temperature建议设为0或接近0,因为工具调用场景需要模型输出稳定的结构化决策,温度越高参数抖动越大。prompt中的agent_scratchpad是占位符,LangChain会把中间的工具调用记录填充进去,模型据此决定下一步动作,这个占位符不能漏。
四、高频问题排查与优化建议
实际调试中最常见的问题是模型不调用工具,直接凭自己的知识回答。原因通常是描述写得不够明确,或者问题本身模型觉得有把握。解决办法是在system提示词里明确规则,比如"涉及实时信息必须调用api_search工具",同时优化工具描述里的适用场景说明。
第二个常见问题是参数解析失败,模型传了不符合schema的参数。排查时先检查args_schema定义是否清晰,必填字段有没有加合理约束;其次可以考虑在_run内部对参数做二次清洗,比如strip空白字符、对top_k做范围钳制,不要完全信任模型生成的输入。
最后是性能层面。如果工具内部要请求多个接口,尽量用asyncio.gather并发执行;对高频重复的查询加一层缓存,能明显降低Agent的响应延迟。工具返回的内容如果太长,做截断或摘要,过长的上下文既费token也影响模型对关键信息的把握。把这些细节处理到位,一个自定义API检索工具就能稳定地跑在生产环境里了。