LangChain 设计与实现
第8章 工具系统
第8章 工具系统
在 AI Agent 架构中,工具是模型与外部世界交互的桥梁。语言模型擅长理解和推理,但它无法直接访问数据库、调用 API 或执行计算 -- 这些能力需要通过工具来提供。LangChain 的工具系统将 Python 函数、Runnable 对象乃至第三方服务统一抽象为标准化的工具接口,使得 Agent 能够在推理过程中动态选择和调用这些工具。
本章将从 BaseTool 的核心抽象开始,逐层展开 Tool、StructuredTool 的实现差异,深入剖析 @tool 装饰器的 schema 推导机制,探讨 InjectedToolArg 的运行时注入设计,以及工具系统与模型 function calling 之间的衔接方式。
本章要点
- 理解
BaseTool->Tool/StructuredTool的层级关系与设计差异 - 掌握
@tool装饰器的多种用法及其内部的 schema 推导流程 - 理解
create_schema_from_function如何从函数签名生成 Pydantic 模型 - 了解
ToolException的错误处理机制与handle_tool_error策略 - 掌握
InjectedToolArg的运行时参数注入设计 - 理解
convert_runnable_to_tool如何桥接 Runnable 与 Tool - 了解
render_text_description系列函数的工具描述渲染机制
8.1 工具的本质:从函数到 Runnable
在 LangChain 中,工具的本质是一个特殊的 Runnable -- 它接收字符串或字典输入,执行某种操作,返回结果。但与普通 Runnable 不同,工具携带了丰富的元信息:名称、描述、参数 schema,这些元信息用于告诉语言模型何时以及如何调用该工具。
工具系统的设计面临几个核心挑战:
- Schema 自动推导:开发者定义一个 Python 函数,系统需要自动提取其参数类型和描述,生成符合 OpenAI function calling 格式的 JSON Schema
- 执行安全性:工具执行可能失败,需要优雅地处理错误而不中断 Agent 循环
- 参数注入:某些参数(如回调管理器、运行时配置)不应暴露给模型,而是在执行时由系统注入
- 双模型支持:需要同时支持简单的单字符串输入工具和复杂的多参数结构化工具
8.2 BaseTool:工具的核心抽象
BaseTool 继承自 RunnableSerializable[str | dict | ToolCall, Any],这个类型签名清晰地表达了工具的输入输出契约:
# langchain_core/tools/base.py
class BaseTool(RunnableSerializable[str | dict | ToolCall, Any]):
name: str
description: str
args_schema: ArgsSchema | None = None
return_direct: bool = False
handle_tool_error: bool | str | Callable[[ToolException], str] | None = False
handle_validation_error: ...
response_format: Literal["content", "content_and_artifact"] = "content"
extras: dict[str, Any] | None = None
classDiagram
class RunnableSerializable {
+invoke(input, config) Any
+ainvoke(input, config) Any
+batch(inputs, config) list
+stream(input, config) Iterator
}
class BaseTool {
+name: str
+description: str
+args_schema: ArgsSchema
+return_direct: bool
+handle_tool_error: bool|str|Callable
+response_format: str
+extras: dict
+args: dict
+tool_call_schema: ArgsSchema
+is_single_input: bool
#_run(args, config, run_manager) Any*
#_arun(args, config, run_manager) Any*
}
class Tool {
+func: Callable
+coroutine: Callable
+from_function() Tool
}
class StructuredTool {
+func: Callable
+coroutine: Callable
+args_schema: ArgsSchema [required]
+from_function() StructuredTool
}
RunnableSerializable <|-- BaseTool
BaseTool <|-- Tool
BaseTool <|-- StructuredTool
8.2.1 核心属性解析
name 和 description:这两个属性不仅仅是标识符 -- 它们会被发送给语言模型,模型根据这些信息决定何时调用哪个工具。因此,description 的质量直接影响 Agent 的工具选择准确性。
args_schema:参数模式的类型是 ArgsSchema = TypeBaseModel | dict[str, Any],支持 Pydantic 模型和原始 JSON Schema 两种形式。这种灵活性允许开发者选择类型安全的 Pydantic 方式,也允许动态构建 schema。
response_format:控制工具输出如何被转换为 ToolMessage。当设为 "content_and_artifact" 时,工具返回值被期望是一个二元组 (content, artifact),其中 content 是发给模型的文本摘要,artifact 是完整的结构化数据(如文档列表)。
extras:这是为模型供应商的特定功能预留的扩展字段。例如,Anthropic 的 cache_control、defer_loading 等功能可以通过此字段传递。
8.2.2 tool_call_schema 与参数过滤
BaseTool 区分了两种 schema:args_schema 是完整的参数 schema,而 tool_call_schema 是去除了注入参数后的 schema -- 后者才是发送给语言模型的版本。
@property
def tool_call_schema(self) -> ArgsSchema:
"""Get the schema for tool calls, excluding injected arguments."""
if isinstance(self.args_schema, dict):
if self.description:
return {**self.args_schema, "description": self.description}
return self.args_schema
...
这个属性确保了标记为 InjectedToolArg 的参数不会出现在发送给模型的 schema 中,同时将工具描述注入到 schema 的 description 字段。
8.2.3 错误处理策略
handle_tool_error 提供了三种错误处理模式:
handle_tool_error: bool | str | Callable[[ToolException], str] | None = False
| 模式 | 行为 |
|---|---|
False(默认) |
异常直接上抛,中断 Agent |
True |
将异常消息作为工具输出返回给 Agent |
str |
使用指定的错误消息字符串 |
Callable |
调用自定义函数处理异常,返回错误消息 |
这种分级设计使得开发者可以根据场景灵活选择:对于关键工具,可能希望异常直接中断;对于非关键工具,可以让 Agent 看到错误信息后尝试其他策略。
8.3 Tool vs StructuredTool:两种工具范式
LangChain 提供了两个具体的 BaseTool 子类,代表了两种不同的工具范式。
8.3.1 Tool -- 单输入工具
Tool 类是为简单的单参数工具设计的。它在底层强制要求所有参数合并为单一输入:
# langchain_core/tools/simple.py
class Tool(BaseTool):
func: Callable[..., str] | None
coroutine: Callable[..., Awaitable[str]] | None = None
def _to_args_and_kwargs(self, tool_input, tool_call_id):
args, kwargs = super()._to_args_and_kwargs(tool_input, tool_call_id)
all_args = list(args) + list(kwargs.values())
if len(all_args) != 1:
msg = f"Too many arguments to single-input tool {self.name}."
raise ToolException(msg)
return tuple(all_args), {}
当参数超过一个时,Tool 会抛出 ToolException,提示开发者应该使用 StructuredTool。
8.3.2 StructuredTool -- 多参数工具
StructuredTool 是功能更强大的工具类,支持任意数量的命名参数。它要求必须提供 args_schema:
# langchain_core/tools/structured.py
class StructuredTool(BaseTool):
args_schema: ArgsSchema = Field(..., description="The tool schema.")
func: Callable[..., Any] | None = None
coroutine: Callable[..., Awaitable[Any]] | None = None
StructuredTool 的 from_function 类方法是创建工具的主要入口:
@classmethod
def from_function(
cls,
func=None, coroutine=None,
name=None, description=None,
args_schema=None, infer_schema=True,
response_format="content",
parse_docstring=False,
error_on_invalid_docstring=False,
**kwargs,
) -> StructuredTool:
source_function = func or coroutine
name = name or source_function.__name__
if args_schema is None and infer_schema:
args_schema = create_schema_from_function(name, source_function, ...)
if description is None and not parse_docstring:
description_ = source_function.__doc__ or None
...
这里的 infer_schema=True 默认行为是工具系统便利性的关键 -- 大多数情况下,开发者只需要定义一个带类型注解的函数,schema 就能自动生成。
flowchart TD
A[StructuredTool.from_function] --> B{args_schema 提供了?}
B -->|是| E[直接使用]
B -->|否| C{infer_schema?}
C -->|是| D[create_schema_from_function]
C -->|否| F[不推导 schema]
D --> G[检查函数签名]
G --> H[处理 Pydantic v1/v2 注解]
H --> I[validate_arguments 创建模型]
I --> J[过滤 run_manager/callbacks 等参数]
J --> K[提取 docstring 描述]
K --> L[创建子集 Pydantic 模型]
L --> E
E --> M[构建 StructuredTool 实例]
8.3.3 运行时参数注入:callbacks 和 config
两种工具在执行时都会自动注入回调管理器和运行时配置:
# Tool._run 和 StructuredTool._run 中的相同逻辑
def _run(self, *args, config, run_manager=None, **kwargs):
if self.func:
if run_manager and signature(self.func).parameters.get("callbacks"):
kwargs["callbacks"] = run_manager.get_child()
if config_param := _get_runnable_config_param(self.func):
kwargs[config_param] = config
return self.func(*args, **kwargs)
这段代码展现了一种优雅的"按需注入"策略:
- 只有当函数签名中包含
callbacks参数时,才注入回调子管理器 - 只有当函数签名中包含
RunnableConfig类型的参数时,才注入配置 - 注入的参数对模型不可见(被
FILTERED_ARGS过滤)
8.4 @tool 装饰器:最佳开发体验
@tool 装饰器是创建工具的推荐方式。它支持多种使用模式,通过函数重载(@overload)提供了清晰的类型提示。
8.4.1 四种使用模式
# 模式 1:无参装饰器
@tool
def search(query: str) -> str:
"""Search the web for a query."""
return "results..."
# 模式 2:自定义名称
@tool("web_search")
def search(query: str) -> str:
"""Search the web for a query."""
return "results..."
# 模式 3:带参数的装饰器
@tool(parse_docstring=True, response_format="content_and_artifact")
def search(query: str) -> tuple[str, dict]:
"""Search the web.
Args:
query: The search query string.
"""
return "summary", {"full": "results"}
# 模式 4:包装 Runnable
from langchain_core.runnables import RunnableLambda
runnable = RunnableLambda(lambda x: x["query"].upper())
search_tool = tool("uppercase", runnable)
8.4.2 内部分发逻辑
@tool 的实现通过分析参数类型来决定行为路径:
# langchain_core/tools/convert.py
def tool(name_or_callable=None, runnable=None, *, description=None, ...):
if runnable is not None:
# tool("name", runnable) -- 包装 Runnable
return _create_tool_factory(name_or_callable)(runnable)
if name_or_callable is not None:
if callable(name_or_callable) and hasattr(name_or_callable, "__name__"):
# @tool 无参装饰
return _create_tool_factory(name_or_callable.__name__)(name_or_callable)
if isinstance(name_or_callable, str):
# @tool("name") 或 @tool("name", parse_docstring=True)
return _create_tool_factory(name_or_callable)
# @tool(parse_docstring=True) 带参数装饰器
return _partial
flowchart TD
A["@tool(...)"] --> B{runnable 参数?}
B -->|是| C["tool(name, runnable)"]
C --> D[包装 Runnable 为 Tool]
B -->|否| E{name_or_callable?}
E -->|None| F["@tool(kwargs...) 带参装饰器"]
F --> G[返回 _partial 闭包]
E -->|Callable| H["@tool 无参装饰器"]
H --> I[提取 __name__ 作为工具名]
I --> J[_create_tool_factory]
E -->|str| K["@tool('name') 命名装饰器"]
K --> L[返回 _tool_factory 闭包]
J --> M{infer_schema?}
L --> M
M -->|是| N[StructuredTool.from_function]
M -->|否| O[Tool 简单工具]
8.4.3 _create_tool_factory 的内部工厂
def _create_tool_factory(tool_name: str):
def _tool_factory(dec_func: Callable | Runnable) -> BaseTool:
if isinstance(dec_func, Runnable):
# Runnable 包装逻辑
schema = runnable.input_schema
async def ainvoke_wrapper(callbacks=None, **kwargs):
return await runnable.ainvoke(kwargs, {"callbacks": callbacks})
def invoke_wrapper(callbacks=None, **kwargs):
return runnable.invoke(kwargs, {"callbacks": callbacks})
...
elif inspect.iscoroutinefunction(dec_func):
coroutine = dec_func; func = None
else:
coroutine = None; func = dec_func
if infer_schema or args_schema is not None:
return StructuredTool.from_function(func, coroutine, name=tool_name, ...)
else:
return Tool(name=tool_name, func=func, description=f"{tool_name} tool", ...)
return _tool_factory
注意 Runnable 包装的巧妙之处:它创建了两个 wrapper 函数(同步和异步),将 kwargs 转发给 Runnable 的 invoke/ainvoke,同时将 callbacks 传入配置。这使得 Runnable 在被包装为工具后,仍然能够正确传播回调。
8.4.4 四个 @overload 一个实现:为什么 @tool 的签名文件这么长
convert.py:17-76 连声明了四个 @overload——最后才是真正的 impl。四个 overload 分别对应四种调用形态——
| overload | 典型调用 | 返回类型 |
|---|---|---|
| overload 1(仅 kwargs) | @tool(description="...") |
Callable[[Callable], BaseTool] |
| overload 2(str + Runnable) | tool("name", my_runnable) |
BaseTool |
| overload 3(Callable first) | @tool 直接贴在函数上 |
BaseTool |
| overload 4(str first) | @tool("name") |
Callable[[Callable], BaseTool] |
两条非显然的类型学细节——
@overload不会真正执行、只为 IDE/mypy——Python runtime 只跑最后一个 impl;前 4 个纯粹是类型信息——这让@tool(name="x")的返回类型在 mypy 下能被精确推断为 decorator vs BaseTool、避免用户写result: BaseTool = tool("x")(f)这种丑陋的类型断言name_or_callable字段名的双关——同一个位置参数要么是工具名、要么是被装饰函数——这就是为什么 overload 2/3/4 第一参数名都叫name_or_callable——单一名字兼容两种语义,是 Python 重载的经典模式(对比 C# 的真重载不需要这么别扭)
8.5 create_schema_from_function:从函数签名到 Pydantic 模型
这是工具系统中最复杂的单个函数,它负责将任意 Python 函数的签名转换为一个 Pydantic 模型:
# langchain_core/tools/base.py
def create_schema_from_function(
model_name: str,
func: Callable,
*,
filter_args: Sequence[str] | None = None,
parse_docstring: bool = False,
error_on_invalid_docstring: bool = False,
include_injected: bool = True,
) -> type[BaseModel]:
sig = inspect.signature(func)
# 检测 Pydantic v1/v2 注解
if _function_annotations_are_pydantic_v1(sig, func):
validated = validate_arguments_v1(func, config=_SchemaConfig)
else:
validated = validate_arguments(func, config=_SchemaConfig)
inferred_model = validated.model
...
整个过程分为几个阶段:
flowchart TD
A[函数签名 inspect.signature] --> B{Pydantic v1 注解?}
B -->|是| C[validate_arguments_v1]
B -->|否| D[validate_arguments v2]
C --> E[获取 inferred_model]
D --> E
E --> F[确定 filter_args]
F --> G[过滤 self/cls/run_manager/callbacks]
G --> H{parse_docstring?}
H -->|是| I[解析 Google Style docstring]
H -->|否| J[提取 Annotated 描述]
I --> K[获取参数描述]
J --> K
K --> L[构建 valid_properties 列表]
L --> M[_create_subset_model]
M --> N[返回 Pydantic 模型]
8.5.1 参数过滤机制
默认的 FILTERED_ARGS 包含 ("run_manager", "callbacks"),这些是 LangChain 内部使用的参数。此外,还会过滤掉:
- 类方法的
self/cls参数 - 标记为
InjectedToolArg的参数(当include_injected=False时) - 类型为
RunnableConfig的参数
8.5.2 docstring 解析
当 parse_docstring=True 时,函数会解析 Google Style docstring 来提取参数描述:
@tool(parse_docstring=True)
def search(query: str, max_results: int = 10) -> str:
"""Search the web for information.
Args:
query: The search query string.
max_results: Maximum number of results to return.
"""
...
解析后,query 和 max_results 的描述会被注入到生成的 Pydantic 模型的字段描述中,最终出现在发送给模型的 JSON Schema 里。
8.5.3 Pydantic v1/v2 兼容性
create_schema_from_function 内部有一个关键的版本检测步骤:
def _function_annotations_are_pydantic_v1(signature, func):
any_v1_annotations = any(
_is_pydantic_annotation(param.annotation, pydantic_version="v1")
for param in signature.parameters.values()
)
any_v2_annotations = any(
_is_pydantic_annotation(param.annotation, pydantic_version="v2")
for param in signature.parameters.values()
)
if any_v1_annotations and any_v2_annotations:
raise NotImplementedError("Mixed v1 and v2 annotations not supported")
return any_v1_annotations and not any_v2_annotations
如果函数参数中包含 Pydantic v1 的模型类型,则使用 v1 的 validate_arguments;如果包含 v2 类型,则使用 v2 版本。混合使用两个版本的注解会直接报错。这种检测确保了在 Pydantic 版本迁移期间的平稳过渡。
8.5.4 langchain_core/tools/ 7 文件 2793 行的模块拆分
| 文件 | 行 | 角色 |
|---|---|---|
base.py |
1586 | 核心抽象——BaseTool(1138 行主体 + 一堆 helper)、ChildTool、InjectedToolArg 家族、create_schema_from_function、BaseToolkit |
convert.py |
476 | @tool 装饰器(4 个 overload + 1 个 impl)、convert_runnable_to_tool |
structured.py |
271 | StructuredTool——多参数工具 |
simple.py |
204 | Tool——单参数工具(向后兼容) |
retriever.py |
94 | create_retriever_tool——把 Retriever 包装成 Tool |
__init__.py |
95 | 导出名单——决定哪些是公共 API |
render.py |
67 | render_text_description / render_text_description_and_args——把工具列表渲染成 Prompt 文本 |
三条看得出设计意图的观察——
- base.py 占整个模块 57%——
BaseTool是 LCEL 的 Runnable 特化、承担 input validation / error handling / sync+async 双路径 / 回调穿透——所有工具类型都要继承它、逻辑必然集中在这里 - simple.py 比 structured.py 短(204 vs 271)——是因为
Tool单参数场景下、schema 构造退化为单字段 Pydantic 模型、多参数的 InjectedToolArg 过滤 / docstring 解析逻辑都不需要 - render.py 仅 67 行——工具渲染成 Prompt 这件事很简单——但它独立成文件、说明 LangChain 把"让模型看懂工具"和"让代码调用工具"清晰切开——这条分界线比很多 Agent 框架都清晰(ReAct / MRKL 论文里两者混在同一份 prompt 里)
8.6 InjectedToolArg:运行时参数注入
InjectedToolArg 是 LangChain 工具系统中一个精巧的设计。它允许某些参数在模型看来"不存在",但在工具实际执行时被自动注入。
# langchain_core/tools/base.py
class InjectedToolArg:
"""Annotation for tool arguments that are injected at runtime."""
class _DirectlyInjectedToolArg:
"""Annotation for args injected via direct type annotation."""
class InjectedToolCallId(InjectedToolArg):
"""Annotation for injecting the tool call ID."""
LangChain 定义了两种注入机制:
- 元数据注入(
InjectedToolArg):通过Annotated类型标记 - 直接注入(
_DirectlyInjectedToolArg):通过直接使用特定类型(如ToolRuntime)
使用方式:
from typing import Annotated
from langchain_core.tools import tool, InjectedToolCallId
from langchain_core.messages import ToolMessage
@tool
def create_report(
topic: str,
tool_call_id: Annotated[str, InjectedToolCallId]
) -> ToolMessage:
"""Generate a report on the given topic."""
content = f"Report on {topic}"
return ToolMessage(content, name="create_report", tool_call_id=tool_call_id)
在这个例子中,模型只会看到 topic 参数的 schema。tool_call_id 参数会在工具被 Agent 调用时自动注入当前 tool_call 的 ID。
检测逻辑:
def _is_injected_arg_type(type_, injected_type=None):
if _is_directly_injected_arg_type(type_):
return True
if injected_type is None:
injected_type = InjectedToolArg
return any(
isinstance(arg, injected_type)
or (isinstance(arg, type) and issubclass(arg, injected_type))
for arg in get_args(type_)[1:]
)
这个函数检查类型注解的 Annotated 元数据中是否包含 InjectedToolArg 实例或子类。StructuredTool 缓存了被注入参数的键名集合,在参数分发时将这些键排除在 schema 之外。
8.6.1 两种注入机制的语义差异与共存理由
上面把"元数据注入"和"直接注入"并列、但没讲清楚为什么要有两种机制。翻开 tools/base.py:1367-1394 两个 class 的原始 docstring:
InjectedToolArg(line 1367-1372)——装饰性注入:
class InjectedToolArg:
"""Annotation for tool arguments that are injected at runtime.
Tool arguments annotated with this class are not included in the tool
schema sent to language models and are instead injected during execution.
"""
用法:tool_call_id: Annotated[str, InjectedToolCallId]——str 仍然是真实的值类型,InjectedToolCallId 只是挂在 Annotated 元数据里的标记。LangChain 看到这个标记就不把 tool_call_id 暴露给模型的 schema、但运行时注入真实 str 值。
_DirectlyInjectedToolArg(line 1375-1394)——结构性注入:
class _DirectlyInjectedToolArg:
"""Annotation for tool arguments that are injected at runtime.
Injected via direct type annotation, rather than annotated metadata.
For example, `ToolRuntime` is a directly injected argument.
Note the direct annotation rather than the verbose alternative:
`Annotated[ToolRuntime, InjectedRuntime]`
```python
from langchain_core.tools import tool, ToolRuntime
@tool
def foo(x: int, runtime: ToolRuntime) -> str:
# use runtime.state, runtime.context, runtime.store, etc.
...
```
"""
用法:runtime: ToolRuntime——ToolRuntime 本身就是注入标记,参数类型直接使用它即可,不需要 Annotated 包装。docstring 原文直白说明理由:"Note the direct annotation rather than the verbose alternative: Annotated[ToolRuntime, InjectedRuntime]"——直接注入是为了人机工效。
两种机制的选择依据:
- 当注入目标是一个"基本值"时用
InjectedToolArg——被注入的内容是str/int/dict这类通用类型(tool_call_id本质是str、state_value本质是dict)、没法靠类型本身区分"这个 str 要注入"和"这个 str 是普通参数"。加个Annotated[str, InjectedToolCallId]既保持类型精确、又挂上注入标记。 - 当注入目标是一个"专用容器"时用
_DirectlyInjectedToolArg——ToolRuntime是 LangChain 自己定义的类、本身就是"运行时容器"语义、全世界没别的地方会用ToolRuntime作参数。这时候Annotated[ToolRuntime, InjectedRuntime]是冗余的——ToolRuntime就已经足够声明意图。直接类型标注省一层 verbosity。
_is_directly_injected_arg_type 的 generic-aware 实现(line 1423-1440):
def _is_directly_injected_arg_type(type_: Any) -> bool:
return (
isinstance(type_, type) and issubclass(type_, _DirectlyInjectedToolArg)
) or (
(origin := get_origin(type_)) is not None
and isinstance(origin, type)
and issubclass(origin, _DirectlyInjectedToolArg)
)
第二个分支用 get_origin 处理泛型特化:ToolRuntime[ContextT, StateT] 这种泛型形式里、type_ 本身不是一个普通 class(是 typing._GenericAlias)、issubclass 不能直接判断。get_origin(ToolRuntime[X, Y]) 返回 ToolRuntime 本身——然后对 origin 做 issubclass 检查。docstring 注释 "ToolRuntime or ToolRuntime[ContextT, StateT] would both return True" 精确描述了这个能力。
这种 "专用类型 = 隐式注入 + 泛型也认" 的设计让用户代码里泛型化的 ToolRuntime[MyCtx, MyState] 和裸 ToolRuntime 被一视同仁——用户不需要学习"什么时候该用泛型版",两种写法效果一致。
_is_injected_arg_type 的双路径组合(line 1443-1468):
if injected_type is None:
if _is_directly_injected_arg_type(type_):
return True
injected_type = InjectedToolArg
return any(
isinstance(arg, injected_type)
or (isinstance(arg, type) and issubclass(arg, injected_type))
for arg in get_args(type_)[1:]
)
当调用者不指定具体类型时(injected_type=None),先看结构性注入(_DirectlyInjectedToolArg 子类),再看装饰性注入(默认 InjectedToolArg 元数据)。两种机制在检测阶段被统一扫描——用户不管用哪种标记方式、_is_injected_arg_type 都能正确识别。这是 LangChain 扩展点的典型设计:两套用户 API / 一套内部统一处理。
flowchart LR
subgraph 模型视角
S1["{ topic: str }"]
end
subgraph 实际签名
S2["topic: str, tool_call_id: Annotated[str, InjectedToolCallId]"]
end
subgraph 运行时
A[Agent 调用] --> B[tool_call_id 从 ToolCall.id 注入]
B --> C["create_report(topic='AI', tool_call_id='call_123')"]
end
S2 -->|过滤 InjectedToolArg| S1
S1 -->|发送给模型| A
8.6.2 三种注入标记:InjectedToolArg / _DirectlyInjectedToolArg / InjectedToolCallId
base.py:1367-1420 并排定义了三个注入标记类——彼此关系是并列 + 继承——
| 类名 | 层 | 触发方式 | 典型用途 |
|---|---|---|---|
InjectedToolArg |
基类 | x: Annotated[T, InjectedToolArg] 或 Annotated[T, InjectedState(...)] |
所有元数据型注入的基类 |
_DirectlyInjectedToolArg |
独立、下划线前缀 | x: ToolRuntime(直接标注、不用 Annotated) |
ToolRuntime 一个类就够了 |
InjectedToolCallId(InjectedToolArg) |
继承 | x: Annotated[str, InjectedToolCallId] |
把 tool_call_id 注入参数 |
为什么要三套?——看 _is_directly_injected_arg_type(base.py:1423)的判断分支就知道——
return (
isinstance(type_, type) and issubclass(type_, _DirectlyInjectedToolArg)
) or (
(origin := get_origin(type_)) is not None
and isinstance(origin, type)
and issubclass(origin, _DirectlyInjectedToolArg)
)
三条关键差别——
InjectedToolArg走Annotated元数据通道——type_本身不是 InjectedToolArg 的子类、而是Annotated[T, InjectedToolArg()]里的第 2 个参数才是——所以判定要走get_args(type_)[1:]翻元数据(base.py:1443_is_injected_arg_type的末尾)_DirectlyInjectedToolArg走类型本身——ToolRuntime直接是这个类的子类、用户写runtime: ToolRuntime就够了、不用包Annotated——这是 LangGraph 引入ToolRuntime时新加的、因为 LangGraph 的场景(读 state / context)够常用、值得配专门的简洁语法InjectedToolCallId继承InjectedToolArg——因为注入 call_id 是同一种语义(从运行时捞出来塞给函数)、只是具体注入的值不同——所以能复用_is_injected_arg_type的整个判断链、只需换injected_type参数
一个容易误解的点——_DirectlyInjectedToolArg 带下划线前缀——是给框架内部用的、不是让用户继承——用户想做"裸类型注入"只有 ToolRuntime 一个官方入口。这和 InjectedToolArg(公开、鼓励扩展)的定位正好相反。
8.7 convert_runnable_to_tool:桥接 Runnable 与 Tool
convert_runnable_to_tool 函数将任意 Runnable 转换为 BaseTool,这是工具系统的一个重要扩展点。
# langchain_core/tools/convert.py
def convert_runnable_to_tool(
runnable: Runnable,
args_schema: type[BaseModel] | None = None,
*,
name: str | None = None,
description: str | None = None,
arg_types: dict[str, type] | None = None,
) -> BaseTool:
description = description or _get_description_from_runnable(runnable)
name = name or runnable.get_name()
schema = runnable.input_schema.model_json_schema()
if schema.get("type") == "string":
# 单字符串输入 -> Tool
return Tool(name=name, func=runnable.invoke,
coroutine=runnable.ainvoke, description=description)
# 多参数输入 -> StructuredTool
async def ainvoke_wrapper(callbacks=None, **kwargs):
return await runnable.ainvoke(kwargs, config={"callbacks": callbacks})
def invoke_wrapper(callbacks=None, **kwargs):
return runnable.invoke(kwargs, config={"callbacks": callbacks})
return StructuredTool.from_function(
name=name, func=invoke_wrapper, coroutine=ainvoke_wrapper,
description=description, args_schema=args_schema,
)
设计要点:
- 自动探测输入类型:通过检查 Runnable 的
input_schema来决定创建Tool还是StructuredTool - kwargs 转发:wrapper 函数将关键字参数打包为字典传给 Runnable,因为 Runnable 的
invoke接受单一输入 - 回调传播:wrapper 从参数中提取
callbacks并注入到配置中
当 Runnable 的输入类型无法自动推断时,可以通过 arg_types 参数手动指定:
def _get_schema_from_runnable_and_arg_types(runnable, name, arg_types=None):
if arg_types is None:
try:
arg_types = get_type_hints(runnable.InputType)
except TypeError as e:
msg = "Tool input must be str or dict. If dict, dict arguments must be typed."
raise TypeError(msg) from e
fields = {key: (key_type, Field(...)) for key, key_type in arg_types.items()}
return create_model(name, **fields)
8.8 render_text_description:工具描述渲染
在某些 Agent 实现(特别是基于 Prompt 的 ReAct Agent)中,需要将工具列表渲染为纯文本描述嵌入到 Prompt 中。LangChain 提供了两个渲染函数:
# langchain_core/tools/render.py
def render_text_description(tools: list[BaseTool]) -> str:
descriptions = []
for tool in tools:
if hasattr(tool, "func") and tool.func:
sig = signature(tool.func)
description = f"{tool.name}{sig} - {tool.description}"
else:
description = f"{tool.name} - {tool.description}"
descriptions.append(description)
return "\n".join(descriptions)
def render_text_description_and_args(tools: list[BaseTool]) -> str:
tool_strings = []
for tool in tools:
args_schema = str(tool.args)
description = f"{tool.name} - {tool.description}"
tool_strings.append(f"{description}, args: {args_schema}")
return "\n".join(tool_strings)
渲染结果示例:
# render_text_description
search(query: str, max_results: int = 10) - Search the web for information
calculator(expression: str) - Evaluate a mathematical expression
# render_text_description_and_args
search - Search the web, args: {"query": {"type": "string"}, "max_results": {"type": "integer"}}
calculator - Evaluate math, args: {"expression": {"type": "string"}}
第一种格式包含了完整的函数签名,对于模型理解参数类型很有帮助;第二种格式包含了 JSON Schema 形式的参数定义,更加精确。ToolsRenderer 类型别名 Callable[[list[BaseTool]], str] 则允许开发者自定义渲染函数。
8.9 create_retriever_tool:检索器工具化
LangChain 提供了一个专门的函数将检索器包装为工具,这是 RAG Agent 的核心组件:
# langchain_core/tools/retriever.py
class RetrieverInput(BaseModel):
query: str = Field(description="query to look up in retriever")
def create_retriever_tool(
retriever: BaseRetriever,
name: str,
description: str,
*,
document_prompt: BasePromptTemplate | None = None,
document_separator: str = "\n\n",
response_format: Literal["content", "content_and_artifact"] = "content",
) -> StructuredTool:
document_prompt_ = document_prompt or PromptTemplate.from_template("{page_content}")
def func(query: str, callbacks=None):
docs = retriever.invoke(query, config={"callbacks": callbacks})
content = document_separator.join(
format_document(doc, document_prompt_) for doc in docs
)
if response_format == "content_and_artifact":
return (content, docs)
return content
...
设计亮点:
- 固定 schema:使用
RetrieverInput(只有query字段),保持工具接口的简洁性 - 文档格式化:通过
document_prompt控制检索到的文档如何转换为文本 - artifact 模式:
"content_and_artifact"格式允许将原始Document对象作为 artifact 保留,便于后续处理
8.10 工具执行流程全景
从模型产生 tool_call 到工具执行完成返回 ToolMessage,完整的执行流程如下:
sequenceDiagram
participant M as Language Model
participant A as Agent
participant T as BaseTool
participant F as User Function
M->>A: AIMessage with tool_calls
A->>T: invoke(tool_call)
T->>T: _to_args_and_kwargs(tool_input)
T->>T: 验证 args_schema
T->>T: 注入 InjectedToolArg
T->>T: 创建 CallbackManager
T->>F: _run(*args, config, run_manager, **kwargs)
alt 执行成功
F-->>T: result
T->>T: 构造 ToolMessage(content=result)
else ToolException
T->>T: handle_tool_error 处理
T->>T: 构造 ToolMessage(content=error_msg)
else ValidationError
T->>T: handle_validation_error 处理
end
T-->>A: ToolMessage
A->>M: 将 ToolMessage 加入对话
8.11 设计决策分析
Tool vs StructuredTool 的历史演变
最初,LangChain 只有 Tool,它接受单个字符串输入。随着 Agent 架构的演进和 function calling 的出现,需要支持多参数工具,StructuredTool 应运而生。保留 Tool 是为了向后兼容,同时它在某些简单场景下确实更加方便。
为什么使用 validate_arguments 推导 schema?
create_schema_from_function 使用了 Pydantic 的 validate_arguments 装饰器来推导函数签名。这是一个巧妙但有些 hacky 的做法 -- 它利用了 validate_arguments 会自动从函数签名生成 Pydantic 模型这一特性。源码中的注释坦率地承认了这一点:
# This code should be re-written to simply construct a Pydantic model
# using inspect.signature and create_model.
这个注释暗示了未来可能的重构方向:直接使用 inspect.signature 和 create_model 来构建 schema,而不是依赖已被标记为废弃的 validate_arguments。
response_format 的双模式设计
"content_and_artifact" 格式的引入解决了一个实际问题:模型只需要看到工具输出的简洁摘要,但应用程序可能需要完整的结构化结果。这种分离使得 Agent 的对话上下文保持精简,同时不丢失完整数据。
extras 字段的开放性
extras 字段的设计体现了 LangChain 对供应商生态的务实态度。不同的模型供应商需要传递不同的工具元信息(如 Anthropic 的缓存控制),与其为每个供应商添加专门字段,不如提供一个通用的扩展点。这遵循了"开放-封闭"原则 -- 对扩展开放,对修改封闭。
8.12 小结
LangChain 的工具系统构建了一套从函数签名到 JSON Schema、从 Agent 调用到安全执行的完整链路。BaseTool 作为 Runnable 的特化,天然融入 LCEL 生态;Tool 和 StructuredTool 分别服务于简单和复杂的参数场景;@tool 装饰器通过 schema 自动推导和 docstring 解析,将创建工具的开发体验简化到了极致。
InjectedToolArg 的设计优雅地解决了"哪些参数给模型看、哪些参数由系统注入"的问题。create_schema_from_function 虽然实现上依赖了 Pydantic 的内部机制,但确实做到了从任意 Python 函数自动生成合规的 JSON Schema。render_text_description 系列函数和 create_retriever_tool 则将工具系统与 Prompt 工程和 RAG 流程紧密衔接。
8.12.1 工具源码里的三条边界:Schema、注入参数、回调
工具系统最容易被误解成"把函数包一下给模型调用"。源码显示它至少有三条边界。
第一条边界是 schema 推导。 langchain_core/tools/base.py:289-317 的 create_schema_from_function 从 inspect.signature(func) 开始,按 Pydantic v1/v2 选择不同的 validate_arguments 路径。base.py:323-329 的注释明确承认当前实现依赖已废弃能力,未来应改成 inspect.signature + create_model。这不是小细节:工具 schema 是发给模型看的契约,任何签名推导错误都会变成模型错误调用。
第二条边界是模型可见参数和运行时注入参数。 BaseTool 在 base.py:405-520 定义 name、description、args_schema、return_direct、callbacks、tags、metadata、response_format、extras 等字段;其中 description 给模型判断何时使用工具,args_schema 给模型约束输入,callbacks / metadata 给运行时追踪。InjectedToolArg 在 base.py:1367-1372 明确说明:这类参数不放进发送给模型的 schema,而是在执行时注入。InjectedToolCallId 在 base.py:1397-1419 则专门处理工具调用 ID。
第三条边界是执行路径。 BaseTool._parse_input 在 base.py:656-725 区分字符串输入、dict 输入、Pydantic v2、Pydantic v1、JSON schema dict,并在需要 InjectedToolCallId 但没有 tool_call_id 时抛错。BaseTool.run 在 base.py:878-951 先配置 CallbackManager,过滤 injected args 后触发 on_tool_start;base.py:958-967 再把 child callbacks 和 Runnable config 注入 _run;base.py:968-981 检查 content_and_artifact 是否真的返回二元组。
flowchart TD A[Python 函数或 Runnable] --> B[create_schema_from_function] B --> C[args_schema 给模型] A --> D[BaseTool.run] D --> E[_parse_input 验证输入] E --> F[注入 tool_call_id / config / run_manager] F --> G[_run 执行业务] G --> H[content 或 content_and_artifact] D --> I[CallbackManager 追踪]
@tool 装饰器只是把这些边界串起来。langchain_core/tools/convert.py:17-88 用多组 overload 覆盖不同调用形态;convert.py:303-316 在 infer_schema 或显式 args_schema 存在时创建 StructuredTool;convert.py:317-333 在不推导 schema 时退回简单 Tool,且要求函数必须有 docstring。也就是说,LangChain 不是无条件把函数变成复杂工具,而是按 schema 能力选择 StructuredTool 或简单 Tool。
create_retriever_tool 是一个工程化样本。langchain_core/tools/retriever.py:31-39 的签名固定接收 retriever、name、description、document_prompt、separator、response_format;retriever.py:65-74 的同步函数调用 retriever.invoke(query, config={"callbacks": callbacks}),把文档格式化后返回文本或 (content, docs);retriever.py:76-85 的异步函数同样保留 artifact。这个函数说明工具化不是把检索器直接暴露给模型,而是把"模型看到的内容"和"应用保留的原始文档"分开。
| 边界 | 源码位置 | 失败时的典型症状 |
|---|---|---|
| schema 推导 | base.py:289-345 |
模型参数名/类型理解错 |
| 注入参数 | base.py:656-725、base.py:1367-1468 |
模型看到不该看到的参数,或运行时缺 tool_call_id |
| response_format | base.py:968-981、retriever.py:72-85 |
工具返回值形状和 ToolMessage 不匹配 |
| 回调传播 | base.py:916-951、base.py:958-967 |
工具运行在 trace 树里断链 |
因此,设计工具时不能只写函数体。你要同时设计模型提示面、参数 schema、运行时注入、错误处理、artifact 保留和回调追踪。LangChain 的工具系统复杂,是因为它把这些生产约束集中在 BaseTool 这一层,而不是让每个 Agent 自己补。
这也解释了为什么工具描述不能只照搬函数注释。函数注释写给人类开发者,往往强调实现细节;工具描述写给模型,必须强调何时使用、何时不要使用、输入应该长什么样、输出是否可信。描述不清,模型会在错误场景调用工具;参数边界不清,模型会构造无效输入;返回格式不清,下游会把摘要和原始结果混在一起。
从安全角度看,注入参数尤其关键。认证信息、运行时状态、调用编号、用户上下文都不应该暴露给模型生成。模型只负责提出"要调用什么工具、传什么业务参数",系统负责补齐不可见的执行上下文。InjectedToolArg 把这条边界写进类型注解,避免开发者靠约定隐藏参数。
从可观测性角度看,工具也必须进入回调树。一次 Agent 运行里,模型为什么选择某工具、工具输入是什么、验证是否失败、返回是否包含 artifact,这些都应该被追踪。否则 Agent 失败时只能看到最终答案错误,却看不到是哪次工具调用把链路带偏。
一个成熟的工具定义,至少要回答六个问题:
| 问题 | 设计对象 |
|---|---|
| 模型什么时候该用它 | description 的场景说明 |
| 模型能传哪些参数 | args_schema |
| 哪些参数由系统补 | 注入参数 |
| 失败后是否转成消息 | handle_tool_error / handle_validation_error |
| 是否保留原始结果 | response_format |
| 是否能被追踪 | callbacks、tags、metadata |
这些问题都回答清楚,工具才是 Agent 协议的一等组件,而不是一个临时函数包装。
还有一个实践细节:工具越接近真实业务,越应该把模型可见内容压窄。模型不需要知道数据库连接、用户权限对象、缓存句柄、请求编号;它只需要知道能完成什么任务、需要哪些业务参数、失败时会返回什么。系统侧再把不可见上下文注入进去。这个分层能减少提示词泄露,也能减少模型构造无效参数的机会。
对调试来说,也要按这条链路排查:先看描述是否让模型选对工具,再看 schema 是否让模型填对字段,再看输入验证是否通过,再看注入参数是否齐全,最后看工具函数本身。很多工具调用失败,并不是函数逻辑错,而是前面的契约没有写清。
因此,工具质量评审应该同时看提示面和执行面。提示面决定模型会不会正确调用,执行面决定调用后能不能被验证、追踪和恢复。只看函数体,会漏掉 Agent 系统里最容易出错的部分,也会误判真实风险。