langchain tools源码解析以及扩展

提示：本章大量涉及到装饰器概念（有参装饰器），如有不清晰的地方，请看本人fluent python 闭包与装饰器章节。

一、概念提醒

@tool的任务：

把普通 Python 函数变成能被 Agent 理解与调用的信息化对象（Tool）

整个流程包含：

• 解析函数的名字、文档、参数类型等
• 包装成标准 Tool 对象
• 实现参数校验、序列化、调用分发

二、最小例子：一步步还原其实现

我们写一个最简单的 @tool 例子：

from langchain.tools import tool@tool
def add(a: int, b: int) -> int:"""相加"""return a + b

我们来追踪它背后的源码链路。

三、源码追踪与讲解

1、@tool 装饰器本身（源码解读）

langchain.tools.tool 的定义（大致代码结构和官方一致）：

def tool(_func=None, *, args_schema=None, return_direct=False, **kwargs):def decorator(func):return Tool.from_function(func,args_schema=args_schema,return_direct=return_direct,**kwargs,)if _func is None:return decoratorelse:return decorator(_func)

分析

• 支持无参数和有参数两种装饰用法
• 真正工作由 Tool.from_function 完成

2、Tool 对象创建流程

官方 Tool 定义路径：langchain/tools/base.py 及 langchain/tools/__init__.py

重点方法：Tool.from_function

class Tool(BaseTool):# ... 其他代码@classmethoddef from_function(cls, func: Callable, name: Optional[str]=None,description: Optional[str]=None,args_schema: Optional[Type[BaseModel]]=None,return_direct: bool=False,**kwargs,):# 自动获取函数名字和文档说明tool_name = name or func.__name__tool_description = description or func.__doc__ or ""# 关键点1：自动生成参数schemaif args_schema is None:args_schema = create_schema_from_function(func)# 关键点2：封装函数执行逻辑return cls(name=tool_name,func=func,args_schema=args_schema,description=tool_description,return_direct=return_direct,**kwargs,)

关键点解释：

• 参数定义自动生成

  默认情况下，会尝试从函数的参数签名和类型注解动态构造 pydantic schema。这点很重要，决定了 agent 能不能理解参数类型。

• 函数封装

  你的原始函数（例如 add）会直接挂载到 Tool 对象的 func 属性上，后续通过 Tool 对象统一调度。

  这一句 return cls(…) 就是标准的“封装与注册”，把你所有和Agent用工具相关的信息都集中了起来，变成LangChain Agent可用、可被AI推理时发现和交互的标准对象。

3、参数 schema 的自动构建

核心函数：create_schema_from_function

这段最难，但很本质。LangChain 依赖 pydantic 动态生成输入校验模型。

伪代码复现：

def create_schema_from_function(func):# 获取函数参数签名与注解sig = inspect.signature(func)fields = {}for name, param in sig.parameters.items():# 判断是否有注解类型，否则用 Anyanno = param.annotation if param.annotation is not inspect.Parameter.empty else Anydefault = param.default if param.default is not inspect.Parameter.empty else ...fields[name] = (anno, default)# 动态创建 Pydantic Model，用于参数解析/校验schema = pydantic.create_model(func.__name__ + "Schema",**fields)return schema

直观结果：
你如果写了 def add(a: int, b: int) -> int, 参数类型annotation会被抽出来，做出：

class AddSchema(BaseModel):a: intb: int

这样后面 LLM 或 Agent 只要用 json {a: 3, b: 7} 输入，就能用 Pydantic 安全校验并自动转换调用。

4、Tool 执行流程（用例演示）

假如 Agent 获得如下工具注册表：

tools = [add]  # 实际是 [Tool(...)]对象

假设要调用 add 工具，并校验参数类型，怎么用呢？

模拟内部调用过程：

# 1、从 tool 列表找到 Tool 对象
t = tools[0]# 2、让 pydantic 校验参数类型并实例化
args = {"a": 2, "b": "4"}  # 注意，类型不对也没关系，pydantic 会自动转换
validated = t.args_schema(**args)
print(validated)  # AddSchema(a=2, b=4)# 3、安全调用用户函数
result = t.func(**validated.dict())
print(result)  # 输出: 6

自动类型转换与校验机制：
假如用户输入的数据类型不对（如 b=“4” 是字符串），pydantic 会自动帮你转换成 int。如果语义混乱（如a不能为字符串），那会自动抛出参数校验异常。

5、结论汇总

1. @tool 实际会返回一个 Tool 类对象，而非原始函数本身！
2. Tool 内部自动生成 pydantic 参数Schema，提供标准化 json 参数解析/校验。
3. Tool 函数调度封装，LLM/Agent 调用时无需直接解包参数。
4. 可以指定返回形式，对构建复杂Agent流程很有帮助。

四、对比说明演示

不用 @tool，你只能这样写：

def add(a: int, b: int) -> int:...# Agent 不知道怎么自动传参

用了 @tool，你能获得：

t = add  # 实际是 Tool 对象args = {"a": 5, "b": 8}
validated = t.args_schema(**args)
output = t.func(**validated.dict())

而Agent内部正是这样调用你的工具！

五、扩展（自定义 args_schema）

如果你想干预参数校验过程，可以自定义 Pydantic schema 用作 Tool 的 args_schema：

from pydantic import BaseModelclass MyAddArgs(BaseModel):a: intb: int# 支持字段校验 或 复杂嵌套@tool(args_schema=MyAddArgs)
def add(a, b):"""加法定制"""return a + b

六、小结复盘

1. 包装过程：Python函数→@tool装饰→自动/手动Pydantic schema→Tool
2. 参数关键：标准化、注释和pydantic；为后续自动调用和安全校验铺路
3. 源码关键位置：tool装饰的from_function、create_schema_from_function的动态参数schema构造
4. 实用好处：代理智能推理能根据描述、参数schema自动用你的函数，无需再手写参数解析等无聊工作！**

七、LangChain @tool 与自己实现工具注册装饰器

class ToolExecutor:"""工具执行器"""tools = {}  # 存储工具的逻辑，映射工具名称 -> 工具类tool_metadata = {}  # 存储工具的元数据（包含描述和参数模型）@staticmethoddef register_tool(name: str, description: str):"""注册工具的装饰器"""def decorator(cls):# 确保类继承自 BaseModelif not issubclass(cls, BaseModel):raise ValueError(f"工具 {name} 必须继承自 BaseModel！")if not hasattr(cls, "run") or not callable(getattr(cls, "run")):raise ValueError(f"工具 {name} 缺少有效的 `run` 方法！")# 注册工具到工具列表中ToolExecutor.tools[name] = cls# 注册工具元数据，包括描述和参数结构ToolExecutor.tool_metadata[name] = {"介绍": description,"需要传入的参数": cls.model_json_schema()["properties"]}return cls  # 返回工具类return decorator@staticmethoddef execute_tool(func_name: str, params: Dict) -> Any:"""执行指定工具"""if func_name not in ToolExecutor.tools:return f"工具 {func_name} 未注册"# 获取工具类并执行tool_cls = ToolExecutor.tools[func_name]  # 获取工具类validated_params = tool_cls(**params)  # 验证参数（工具类本身作为参数模型）return tool_cls.run(**validated_params.model_dump())  # 调用工具逻辑@staticmethoddef export_metadata() -> Dict[str, Dict]:"""导出工具元数据"""return [e-mail]_tool(name="ip_is_external",description="查询IP是否是属于内网IP"
)
class IPIsInternalTool(BaseModel):"""IP查询工具"""ips: list[str] = Field(...,description="从输入内容与历史信息中提取出去重后要查询的IP地址列表。IP地址是由数字和点构成的格式，例如：192.168.1.1")@staticmethoddef run(ips: list[str]) -> dict:baidu_ips = []external_ips = []for ip in ips:try:# 将输入的IP地址转换为IPv4Address对象ip_obj = ipaddress.ip_address(ip)# 如果IP地址不是私有地址，则添加到public_ips列表if not ip_obj.is_private:external_ips.append(ip)else:baidu_ips.append(ip)except ValueError:# 如果IP地址无效，可以选择忽略或记录print(f"{ip} 是无效的IP地址")return {"baidu_ips": baidu_ips, "external_ips": external_ips}

1. return cls(…) 是什么？

它是【调用类构造器生成实例】的写法，可以理解为：

“把你传进来的参数，都按设计好的格式和流程，重新组合一下，打包成一个全新的‘标准件’（对象）。”

比如：

🚗“造车”类比

假设你要生产一辆小车：

class Car:def __init__(self, brand, color, horsepower):self.brand = brandself.color = colorself.hp = horsepowerdef make_car(brand, color, horsepower):# 这里就类似于 return cls(...)return Car(brand=brand, color=color, horsepower=horsepower)car1 = make_car("特斯拉", "红", 800)
print(car1.brand, car1.color, car1.hp)  # 特斯拉 红 800

• make_car传入一堆参数，内部通过Car(...)把原始的零碎信息组合“封装注册”为一个可用的Car对象。

所以，return cls(…) 是把“原始信息”→“规范对象”这一步的实现方式。

2. @tool装饰器 和 你自定义 ToolExecutor.register_tool 的对比

二者的核心区别

方面	LangChain @tool	自定义 ToolExecutor.register_tool
装饰对象	Python函数	工具类（通常继承BaseModel）
内部逻辑	封装：函数、元信息、参数schema → Tool对象 需显式传给Agent	注册：类进全局map，参数结构自动提取
调用入口	由Agent调度	由ToolExecutor.execute_tool统一入口调度
注册方式	没有全局注册（只生成对象实例）	有全局注册（map写入）
运行时扩展	适合分布式、组合	适合全局查找、集中执行调度
参数处理	自动抽取签名/注解/文档成Pydantic schema	直接以BaseModel类为参数模型

补图说明：

@tool                                        ToolExecutor.register_tool
def foo(x:int): ...                          ↓          ↓
↓                                          @register_tool                class MyTool(BaseModel):
返回 Tool(name,func,...)                   （全局map写入）                  ...     def run(...): ...
↓                                           ↓                                          ↓
显式传入Agent，列表工具                      ToolExecutor.tools[name]=cls              ToolExecutor.execute_tool名字调度运行
↓                                           ↓                                          ↓
agent自动分发、调用（schema校验）           统一调度，参数校验、元数据导出

实战例子：两者对比

LangChain风格

@tool
def multiply(x: int, y: int) -> int:"""相乘"""return x * yagent = initialize_agent([multiply], ...)
# 提供给Agent显式调度，没有全局注册

ToolExecutor风格

@ToolExecutor.register_tool(name="multiply", description="两个数相乘")
class MultiplyTool(BaseModel):x: inty: int@staticmethoddef run(x, y):return x * yres = ToolExecutor.execute_tool("multiply", {"x":3, "y":5})  # 结果: 15
# 注册后可统一按名字调度，无需显式传对象

总结

• return cls(...)：把各种属性参数打包成【标准对象/实例】，比如Tool、Car、Product等，“物理上的封装和标准化”
• LangChain @tool：封装为对象，灵活组合（哪怕跨文件），但注册是你自己传入Agent决定的
• ToolExecutor.register_tool：自动化全局注册，像插件表，所有工具类都集中在统一map、支持按名调用和批量导出元数据
• 选择方式取决于你的“调用组织模式”和系统需求