langchain_core.tools.BaseTool 类详解

langchain_core.tools.BaseTool 是 LangChain 框架中用于定义语言模型可调用工具的核心抽象基类。本文详细介绍其功能、属性、方法、参数、使用方法、实际应用及注意事项，确保内容条理清晰、易于理解。

1. 概述

1.1 什么是 BaseTool？

BaseTool 是 LangChain 框架中的抽象基类，用于创建语言模型（尤其在基于代理的系统中）可调用的工具。它提供标准接口，支持输入验证、错误处理、工件输出及与 LangChain 可运行对象的集成。开发者通过子类化 BaseTool 定义自定义工具，指定工具名称、描述、输入模式和执行逻辑。

BaseTool 继承自 RunnableSerializable[Union[str, dict, ToolCall], Any]，支持序列化并与 LangChain 的可运行对象生态系统无缝集成。它是工具定义的基础，广泛用于代理系统、LangGraph 工作流和自定义链。

1.2 与 @tool 装饰器的关系

前文讨论的 @tool 装饰器通过将 Python 函数转换为 BaseTool 实例简化工具创建。相比之下，直接子类化 BaseTool 提供更大控制权，适合复杂场景，但需要更多代码。两者共同支持工具调用机制，BaseTool 是底层的实现基础。

1.3 核心功能

• 输入验证：通过 Pydantic 模型验证输入参数。
• 执行方式：支持同步和异步执行、批量处理及流式输出。
• 工件支持：可返回内容和工件（如图片、数据框），工件不暴露给模型。
• 错误处理：提供灵活的工具和验证错误处理机制。
• 回调集成：支持执行期间的监控和日志记录。
• 版本要求：需 langchain-core>=0.2.14，工件（artifact）支持需 >=0.2.19。

2. 类定义与结构

• 模块：langchain_core.tools.base.BaseTool
• 继承：RunnableSerializable[Union[str, dict, ToolCall], Any]
• 输入类型：支持字符串、字典或 ToolCall 对象。
• 输出类型：任意类型，取决于工具实现。
• 引入版本：langchain-core 0.2.14。

BaseTool 实现了 LangChain 的可运行接口，支持 invoke、ainvoke、batch、stream 等方法，使其可作为链或代理的一部分。

3. 创建自定义工具

要创建自定义工具，需子类化 BaseTool，定义必要属性并实现核心方法。以下是关键步骤：

3.1 必要属性

属性	类型	描述
name	str	工具的唯一标识符，在代理工具集中必须唯一。
description	str	工具功能的描述，指导模型何时使用，可包含少样本示例。
args_schema	Optional[Type[BaseModel]]	Pydantic 模型，定义输入参数结构和验证规则，默认为 None。
return_direct	bool	若为 True，工具输出直接返回，代理停止处理，默认为 False。

3.2 必要方法

方法	签名	描述
_run	_run(self, *args, **kwargs) -> Any	同步执行工具的核心逻辑，必须实现。
_arun	_arun(self, *args, **kwargs) -> Any	异步执行工具的核心逻辑，可选实现，默认抛出 NotImplementedError。

3.3 示例：计算器工具

以下是一个实现乘法计算器的工具示例：

from langchain_core.tools import BaseTool
from pydantic import BaseModel, Field
from typing import Optional, Type
from langchain_core.callbacks import CallbackManagerForToolRun, AsyncCallbackManagerForToolRun

class CalculatorInput(BaseModel):
    a: int = Field(description="第一个数字")
    b: int = Field(description="第二个数字")

class CustomCalculatorTool(BaseTool):
    name = "Calculator"
    description = "用于回答数学相关问题"
    args_schema: Type[BaseModel] = CalculatorInput
    return_direct: bool = True

    def _run(
        self,
        a: int,
        b: int,
        run_manager: Optional[CallbackManagerForToolRun] = None
    ) -> int:
        """执行工具"""
        return a * b

    async def _arun(
        self,
        a: int,
        b: int,
        run_manager: Optional[AsyncCallbackManagerForToolRun] = None
    ) -> int:
        """异步执行工具"""
        return self._run(a, b, run_manager=run_manager)

使用方式：

tool = CustomCalculatorTool()
result = tool.invoke({"a": 2, "b": 3})  # 输出 6
async_result = await tool.ainvoke({"a": 2, "b": 3})  # 输出 6

3.4 工件支持示例

工具可返回内容和工件，需设置 response_format="content_and_artifact"：

class ImageGeneratorTool(BaseTool):
    name = "ImageGenerator"
    description = "根据提示生成图片"
    response_format = "content_and_artifact"

    def _run(self, prompt: str) -> tuple[str, bytes]:
        """生成图片"""
        image_data = generate_image(prompt)  # 假设的函数
        return f"图片生成：{prompt}", image_data

4. 初始化参数

BaseTool 的 __init__ 方法支持以下参数，用于自定义工具行为：

参数	类型	默认值	描述
args_schema	Annotated[ArgsSchema | None, SkipValidation()]	None	Pydantic 模型，定义输入参数结构和验证，可为 BaseModel 或 JSON 模式。
callback_manager	BaseCallbackManager | None	None	已废弃（0.1.7 起），建议使用 callbacks，将在 Pydantic 1.0 移除。
callbacks	List[BaseCallbackHandler] | BaseCallbackManager | None	None	执行期间的回调，用于监控和日志记录。
description	str	无默认	工具描述，指导模型使用，优先级高于文档字符串。
handle_tool_error	bool | str | Callable	False	处理 ToolException 的方式，可为布尔值、字符串或可调用对象。
handle_validation_error	bool | str | Callable	False	处理 ValidationError 或 ValidationErrorV1 的方式。
metadata	Dict[str, Any] | None	None	传递给回调的元数据。
response_format	Literal['content', 'content_and_artifact']	'content'	输出格式：仅内容或内容加工件。
return_direct	bool	False	若为 True，输出直接返回，代理停止处理。
tags	List[str] | None	None	传递给回调的标签，用于分类。
verbose	bool	False	若为 True，记录工具执行进度。

参数说明

• args_schema：通过 Pydantic 模型定义输入结构，确保类型安全和验证。例如，CalculatorInput 定义了 a 和 b 两个整数字段。
• handle_tool_error：若为 True，将 ToolException 转为字符串；若为字符串，直接返回；若为可调用对象，调用以处理错误。
• response_format：设置为 content_and_artifact 时，工具返回元组 (content, artifact)，工件供下游使用。

5. 核心方法

BaseTool 提供了丰富的执行方法，继承自 RunnableSerializable：

方法	描述	输入	输出
__call__	已废弃（0.1.47 起），建议使用 invoke。	tool_input: str, callbacks: Callbacks	str
invoke	同步调用工具，转换输入。	input: str | dict | ToolCall, config: RunnableConfig, **kwargs	Any
ainvoke	异步调用工具。	input: str | dict | ToolCall, config: RunnableConfig, **kwargs	Any
run	同步运行工具，处理回调和错误。	input: str | dict, verbose: bool, callbacks: Callbacks, **kwargs	Any
arun	异步运行工具。	input: str | dict, verbose: bool, callbacks: Callbacks, **kwargs	Any
batch	同步批量处理多个输入。	inputs: List[Input], config: RunnableConfig | List[RunnableConfig], return_exceptions: bool	List[Output]
abatch	异步批量处理。	inputs: List[Input], config: RunnableConfig | List[RunnableConfig], return_exceptions: bool	List[Output]
batch_as_completed	同步批量处理，返回完成结果的迭代器。	inputs: Sequence[Input], config: RunnableConfig | Sequence[RunnableConfig], return_exceptions: bool	Iterator[tuple[int, Output | Exception]]
abatch_as_completed	异步批量处理，返回完成结果的异步迭代器。	inputs: Sequence[Input], config: RunnableConfig | Sequence[RunnableConfig], return_exceptions: bool	AsyncIterator[tuple[int, Output | Exception]]
stream	同步流式输出。	input: Input, config: RunnableConfig, **kwargs	Iterator[Output]
astream	异步流式输出。	input: Input, config: RunnableConfig, **kwargs	AsyncIterator[Output]
astream_events	异步流式事件。	input: Any, config: RunnableConfig, version: Literal['v1', 'v2'], **kwargs	AsyncIterator[StreamEvent]

6. 模式与工具调用

BaseTool 定义了两种模式，与前文讨论的 @tool 装饰器一致：

• 输入模式（get_input_schema().schema()）：包含所有参数，包括注入参数（如 InjectedToolArg）。
• 工具调用模式（tool_call_schema.schema()）：排除注入参数，确保模型仅生成非注入参数。

这些模式通过 args_schema 和工具配置生成，帮助模型理解工具的输入要求。

7. 实际应用

BaseTool 在以下场景中广泛使用：

• 代理系统：为代理提供工具执行任务，如计算、搜索或数据更新。例如，CalculatorTool 可用于数学运算。
• 自定义工作流：在 LangGraph 或链中实现模块化功能，输出反馈给模型或下游组件。
• 工件处理：生成供下游使用的工件，如图片、数据框或报告。
• 复杂输入处理：通过 args_schema 支持多参数输入，适合结构化数据处理。

示例：结合注入参数

结合前文 InjectedToolArg 的讨论，工具可隐藏敏感参数：

from langchain_core.tools import BaseTool, InjectedToolArg
from pydantic import BaseModel, Field
from typing import Annotated, List

class PetsInput(BaseModel):
    pets: List[str] = Field(description="收藏宠物列表")

class UpdatePetsTool(BaseTool):
    name = "UpdatePets"
    description = "更新用户的收藏宠物"
    args_schema = PetsInput

    def _run(self, pets: List[str], user_id: Annotated[str, InjectedToolArg]) -> None:
        """更新宠物列表"""
        user_to_pets[user_id] = pets  # 假设的存储

user_id 由运行时注入，模型仅生成 pets。

8. 最佳实践

• 清晰描述：description 应详细，包含少样本示例，帮助模型选择工具。
• 输入验证：使用 args_schema 定义和验证输入结构，确保类型安全。
• 异步实现：为高性能场景实现 _arun，简单场景可委托给 _run。
• 错误处理：配置 handle_tool_error 和 handle_validation_error 以优雅处理异常。
• 版本检查：通过 pip install -qU langchain 确保安装 langchain-core>=0.2.14，工件支持需 >=0.2.19。

9. 注意事项与限制

• 抽象方法：必须实现 _run，否则抛出 NotImplementedError。
• 异步方法：若不实现 _arun，异步调用会失败。
• 类方法支持：在类方法上使用可能需变通方法，建议在类外定义。
• 文档字符串格式：若使用回调，需正确格式化 Google 风格文档字符串。
• 版本依赖：工件支持等功能需特定版本。
• 工件格式：工件需为可序列化对象（如字典、字节）。

10. 结论

langchain_core.tools.BaseTool 是 LangChain 中创建自定义工具的基础类，提供输入验证、错误处理、工件输出和可运行接口集成等功能。它与 @tool 装饰器和 InjectedToolArg 配合使用，适合构建复杂代理系统和模块化工作流。开发者应遵循最佳实践，利用类型提示、清晰描述和最新版本以充分发挥其潜力。

11. 参考资料

• LangChain API 参考：BaseTool
• LangChain 文档：如何创建工具
• LangChain 文档：工具概述
• LangChain 文档：定义自定义工具