LangChain Core 源码学习路线：第 4 篇 Tool Calling 源码解剖#

1
from langchain.tools import tool
2

3
@tool
4
def search_attractions(city: str, preference: str) -> str:
5
    """Search attractions in a city based on user preference."""
6
    return f"{city} attractions for {preference}"
7

8
print(type(search_attractions))
9
print(search_attractions.name)
10
print(search_attractions.description)
11
print(search_attractions.args)
12
print(search_attractions.args_schema)

1
type:
2
StructuredTool
3

4
name:
5
search_attractions
6

7
description:
8
Search attractions in a city based on user preference.
9

10
args:
11
{
12
    "city": {"type": "string"},
13
    "preference": {"type": "string"}
14
}

1
print(type(search_attractions).__mro__)

1
StructuredTool
2
  ↓
3
BaseTool
4
  ↓
5
RunnableSerializable
6
  ↓
7
Serializable

1
search_attractions.invoke({
2
    "city": "大阪",
3
    "preference": "美食和城市漫步",
4
})

1
schema = search_attractions.args_schema.model_json_schema()
2
print(schema)

1
{
2
  "title": "search_attractions",
3
  "description": "Search attractions in a city based on user preference.",
4
  "type": "object",
5
  "properties": {
6
    "city": {
7
      "title": "City",
8
      "type": "string"
9
    },
10
    "preference": {
11
      "title": "Preference",
12
      "type": "string"
13
    }
14
  },
15
  "required": [
16
    "city",
17
    "preference"
18
  ]
19
}

1
Python 类型注解
2
    ↓
3
Pydantic Model
4
    ↓
5
JSON Schema
6
    ↓
7
模型供应商工具描述格式

1
result = search_attractions.invoke({
2
    "city": "大阪",
3
    "preference": "美食",
4
})
5

6
print(type(result))
7
print(result)

1
str
2
大阪 attractions for 美食

1
tool_call = {
2
    "name": "search_attractions",
3
    "args": {
4
        "city": "大阪",
5
        "preference": "美食",
6
    },
7
    "id": "call_001",
8
    "type": "tool_call",
9
}
10

11
result = search_attractions.invoke(tool_call)
12

13
print(type(result))
14
print(result)

1
ToolMessage(
2
    content="大阪 attractions for 美食",
3
    tool_call_id="call_001",
4
    name="search_attractions",
5
    status="success",
6
)

1
Python function
2
  ↓
3
@tool
4
  ↓
5
读取 __name__
6
  ↓
7
读取 inspect.signature()
8
  ↓
9
读取类型注解
10
  ↓
11
读取 docstring
12
  ↓
13
生成 Pydantic args_schema
14
  ↓
15
构造 StructuredTool

1
StructuredTool
2
  ↓
3
读取 name
4
  ↓
5
读取 description
6
  ↓
7
读取 tool_call_schema
8
  ↓
9
转换为供应商 Tool Schema
10
  ↓
11
model.bind_tools([...])
12
  ↓
13
返回已绑定工具的 ChatModel Runnable

1
bind_tools 不会执行工具。
2
bind_tools 只是把工具的“说明书”发送给模型。

1
messages
2
  ↓
3
model_with_tools.invoke(messages)
4
  ↓
5
供应商模型判断是否需要工具
6
  ↓
7
供应商返回原生 tool call
8
  ↓
9
LangChain 适配为标准 ToolCall
10
  ↓
11
AIMessage.tool_calls

1
{
2
    "name": "search_attractions",
3
    "args": {
4
        "city": "大阪",
5
        "preference": "美食",
6
    },
7
    "id": "call_001",
8
    "type": "tool_call",
9
}

1
ToolCall
2
  ↓
3
根据 name 查找 StructuredTool
4
  ↓
5
BaseTool.invoke(tool_call)
6
  ↓
7
_prep_run_args()
8
  ↓
9
BaseTool.run()
10
  ↓
11
_parse_input()
12
  ↓
13
_to_args_and_kwargs()
14
  ↓
15
StructuredTool._run()
16
  ↓
17
调用真实 Python function

1
Python function return value
2
  ↓
3
BaseTool.run()
4
  ↓
5
_format_output()
6
  ↓
7
ToolMessage(
8
    content=...,
9
    tool_call_id=...,
10
    status=...
11
)
12
  ↓
13
追加到 messages
14
  ↓
15
模型再次调用
16
  ↓
17
最终 AIMessage

1
User
2
 │
3
 │  HumanMessage
4
 ▼
5
ChatModel with tools
6
 │
7
 │  AIMessage(tool_calls=[...])
8
 ▼
9
Tool Executor
10
 │
11
 │  BaseTool.invoke(ToolCall)
12
 ▼
13
StructuredTool
14
 │
15
 │  Python function result
16
 ▼
17
ToolMessage
18
 │
19
 │  messages += [AIMessage, ToolMessage]
20
 ▼
21
ChatModel with tools
22
 │
23
 │  Final AIMessage
24
 ▼
25
User

1
class BaseTool(
2
    RunnableSerializable[
3
        str | dict[str, Any] | ToolCall,
4
        Any
5
    ]
6
):
7
    ...

1
str
2
dict[str, Any]
3
ToolCall

1
str
2
dict
3
list
4
ToolMessage
5
Command
6
自定义 ToolOutputMixin

1
tool.invoke(...)
2
await tool.ainvoke(...)
3
tool.batch(...)
4
await tool.abatch(...)
5
tool.with_config(...)
6
tool.with_retry(...)

1
BaseTool
2
=
3
Runnable 执行协议
4
+
5
工具能力描述
6
+
7
输入契约
8
+
9
执行生命周期
10
+
11
结果协议

1
class BaseTool(...):
2
    name: str
3
    description: str
4
    args_schema: BaseModel | dict | None
5
    return_direct: bool = False
6

7
    callbacks: Callbacks = None
8
    tags: list[str] | None = None
9
    metadata: dict[str, Any] | None = None
10

11
    handle_tool_error: bool | str | Callable | None = False
12
    handle_validation_error: bool | str | Callable | None = False
13

14
    response_format: Literal[
15
        "content",
16
        "content_and_artifact"
17
    ] = "content"

1
@tool
2
def search_attractions(...):
3
    ...

1
def search_attractions(...):
2
    ...
3

4
search_attractions = tool(search_attractions)

1
search_attractions

1
search_attractions.func

1
@tool
2
def search(...):
3
    ...

1
@tool("web_search")
2
def search(...):
3
    ...

1
@tool(
2
    "web_search",
3
    description="Search public web pages.",
4
    parse_docstring=True,
5
)
6
def search(...):
7
    ...

1
tool("my_tool", runnable)

1
description
2
return_direct
3
args_schema
4
infer_schema
5
response_format
6
parse_docstring
7
error_on_invalid_docstring
8
extras

1
def tool(
2
    name_or_callable=None,
3
    runnable=None,
4
    *,
5
    description=None,
6
    return_direct=False,
7
    args_schema=None,
8
    infer_schema=True,
9
    response_format="content",
10
    parse_docstring=False,
11
    error_on_invalid_docstring=True,
12
    extras=None,
13
):
14
    # 情况 1：tool("name", runnable)
15
    if runnable is not None:
16
        validate_name(name_or_callable)
17
        factory = _create_tool_factory(name_or_callable)
18
        return factory(runnable)
19

20
    # 情况 2：@tool
21
    # name_or_callable 直接就是被装饰函数
22
    if callable(name_or_callable):
23
        tool_name = name_or_callable.__name__
24
        factory = _create_tool_factory(tool_name)
25
        return factory(name_or_callable)
26

27
    # 情况 3：@tool("custom_name")
28
    if isinstance(name_or_callable, str):
29
        return _create_tool_factory(name_or_callable)
30

31
    # 情况 4：@tool(...)
32
    # 此时先返回一个等待接收函数的装饰器
33
    if name_or_callable is None:
34
        def partial(func):
35
            tool_name = (
36
                func.get_name()
37
                if isinstance(func, Runnable)
38
                else func.__name__
39
            )
40
            factory = _create_tool_factory(tool_name)
41
            return factory(func)
42

43
        return partial
44

45
    raise ValueError(...)

1
def _create_tool_factory(tool_name):
2

3
    def _tool_factory(dec_func):
4

5
        tool_description = description
6

7
        if isinstance(dec_func, Runnable):
8
            validate_runnable_input_schema(dec_func)
9

10
            def invoke_wrapper(callbacks=None, **kwargs):
11
                return dec_func.invoke(
12
                    kwargs,
13
                    {"callbacks": callbacks},
14
                )
15

16
            async def ainvoke_wrapper(callbacks=None, **kwargs):
17
                return await dec_func.ainvoke(
18
                    kwargs,
19
                    {"callbacks": callbacks},
20
                )
21

22
            func = invoke_wrapper
23
            coroutine = ainvoke_wrapper
24
            schema = dec_func.input_schema
25

26
            if tool_description is None:
27
                tool_description = repr(dec_func)
28

29
        elif inspect.iscoroutinefunction(dec_func):
30
            func = None
31
            coroutine = dec_func
32
            schema = args_schema
33

34
        else:
35
            func = dec_func
36
            coroutine = None
37
            schema = args_schema
38

39
        if infer_schema or args_schema is not None:
40
            return StructuredTool.from_function(
41
                func=func,
42
                coroutine=coroutine,
43
                name=tool_name,
44
                description=tool_description,
45
                args_schema=schema,
46
                infer_schema=infer_schema,
47
                response_format=response_format,
48
                parse_docstring=parse_docstring,
49
                error_on_invalid_docstring=(
50
                    error_on_invalid_docstring
51
                ),
52
                extras=extras,
53
            )
54

55
        return Tool(
56
            name=tool_name,
57
            func=func,
58
            coroutine=coroutine,
59
            description=f"{tool_name} tool",
60
            ...
61
        )
62

63
    return _tool_factory

1
infer_schema=True

1
StructuredTool.from_function(...)

1
@tool(infer_schema=False)

1
def invoke_wrapper(callbacks=None, **kwargs):
2
    return runnable.invoke(
3
        kwargs,
4
        {"callbacks": callbacks},
5
    )

1
Runnable
2
  ↓
3
包装为普通 callable
4
  ↓
5
StructuredTool

1
{
2
  "type": "object",
3
  "properties": {...}
4
}

1
func = dec_func
2
coroutine = None

1
func = None
2
coroutine = dec_func

1
invoke → _run → func
2
ainvoke → _arun → coroutine

1
if func is not None:
2
    source_function = func
3
elif coroutine is not None:
4
    source_function = coroutine
5
else:
6
    raise ValueError(
7
        "Function and/or coroutine must be provided"
8
    )

1
name = name or source_function.__name__

1
显式 name
2
  >
3
函数 __name__

1
@tool("attraction_search")
2
def search_attractions(...):
3
    ...

1
函数名：search_attractions
2
工具名：attraction_search

1
if args_schema is None and infer_schema:
2
    args_schema = create_schema_from_function(
3
        name,
4
        source_function,
5
        parse_docstring=parse_docstring,
6
        error_on_invalid_docstring=(
7
            error_on_invalid_docstring
8
        ),
9
        filter_args=_filter_schema_args(
10
            source_function
11
        ),
12
    )

1
未显式提供 args_schema
2
并且 infer_schema=True

1
description_ = explicit_description
2

3
if description_ is None and not parse_docstring:
4
    description_ = source_function.__doc__
5

6
if description_ is None and args_schema is not None:
7
    description_ = args_schema_description
8

9
if description_ is None:
10
    raise ValueError(
11
        "Function must have a docstring "
12
        "if description not provided."
13
    )
14

15
description_ = dedent(description_).strip()

1
显式 description
2
  >
3
函数 docstring
4
  >
5
args_schema 的 description/docstring

1
@tool
2
def search_attractions(city: str) -> str:
3
    return city

1
return StructuredTool(
2
    name=name,
3
    func=func,
4
    coroutine=coroutine,
5
    args_schema=args_schema,
6
    description=description_,
7
    return_direct=return_direct,
8
    response_format=response_format,
9
    **kwargs,
10
)

1
给模型看的能力契约
2
+
3
给执行器调用的真实函数

1
StructuredTool(
2
    model_interface={
3
        "name": ...,
4
        "description": ...,
5
        "args_schema": ...,
6
    },
7
    runtime_implementation={
8
        "func": ...,
9
        "coroutine": ...,
10
    },
11
)

1
def search_attractions(
2
    city: str,
3
    preference: str,
4
    limit: int = 5,
5
) -> list[str]:
6
    """Search attractions.
7

8
    Args:
9
        city: Destination city.
10
        preference: User interests.
11
        limit: Maximum result count.
12
    """

1
函数签名
2
参数名
3
参数类型
4
默认值
5
是否必填
6
docstring
7
参数描述
8
Annotated / Field 描述

1
def create_schema_from_function(
2
    model_name,
3
    func,
4
    *,
5
    filter_args=None,
6
    parse_docstring=False,
7
    error_on_invalid_docstring=False,
8
    include_injected=True,
9
):
10
    # 1. 读取 Python 函数签名
11
    sig = inspect.signature(func)
12

13
    # 2. 使用 Pydantic 包装并构造验证模型
14
    validated = validate_arguments(
15
        func,
16
        config=_SchemaConfig,
17
    )
18

19
    # 3. 获取 Pydantic 自动推断的完整模型
20
    inferred_model = validated.model
21

22
    # 4. 确定需要过滤的框架参数
23
    filter_args_ = resolve_filter_args(
24
        func,
25
        sig,
26
        filter_args,
27
    )
28

29
    # 5. 解析函数和参数说明
30
    description, arg_descriptions = (
31
        _infer_arg_descriptions(
32
            func,
33
            parse_docstring=parse_docstring,
34
            error_on_invalid_docstring=(
35
                error_on_invalid_docstring
36
            ),
37
        )
38
    )
39

40
    # 6. 移除 Pydantic 生成的虚拟字段
41
    valid_properties = []
42
    for field in get_fields(inferred_model):
43
        if field in {"args", "kwargs",
44
                     "v__duplicate_kwargs"}:
45
            continue
46
        if field in filter_args_:
47
            continue
48
        valid_properties.append(field)
49

50
    # 7. 只保留模型可见字段
51
    return _create_subset_model(
52
        model_name,
53
        inferred_model,
54
        valid_properties,
55
        descriptions=arg_descriptions,
56
        fn_description=description,
57
    )

1
(city: str, preference: str, limit: int = 5)

1
city
2
  annotation = str
3
  default = empty
4
  required = True
5

6
preference
7
  annotation = str
8
  default = empty
9
  required = True
10

11
limit
12
  annotation = int
13
  default = 5
14
  required = False

1
Python 函数签名
2
  ↓
3
Pydantic 字段模型
4
  ↓
5
可校验输入
6
  ↓
7
可输出 JSON Schema

1
未来可直接使用 inspect.signature + create_model
2
重写这一部分。

1
def my_tool(
2
    query: str,
3
    config: RunnableConfig,
4
    runtime: ToolRuntime,
5
    callbacks=None,
6
):
7
    ...

1
query

1
config
2
runtime
3
callbacks
4
run_manager

1
config
2
runtime

1
parse_docstring=False

1
@tool(parse_docstring=True)

1
def search_attractions(
2
    city: str,
3
    preference: str,
4
) -> str:
5
    """Search attractions for a traveler.
6

7
    Args:
8
        city: Destination city name.
9
        preference: User interest, such as food or art.
10
    """

1
{
2
  "properties": {
3
    "city": {
4
      "type": "string",
5
      "description": "Destination city name."
6
    },
7
    "preference": {
8
      "type": "string",
9
      "description": "User interest, such as food or art."
10
    }
11
  }
12
}

1
Args:
2
    country: ...

1
from typing import Annotated
2

3
@tool
4
def search_attractions(
5
    city: Annotated[str, "Destination city name"],
6
    preference: Annotated[
7
        str,
8
        "Traveler interests such as food, art, or history",
9
    ],
10
) -> str:
11
    """Search attractions matching traveler preferences."""
12
    ...

1
@tool(parse_docstring=True)
2
def search_attractions(
3
    city: str,
4
    preference: str,
5
    limit: int = 5,
6
) -> list[str]:
7
    """Search attractions matching user interests.
8

9
    Args:
10
        city: Destination city.
11
        preference: Traveler preference.
12
        limit: Maximum number of attractions.
13
    """

1
{
2
  "title": "search_attractions",
3
  "description": "Search attractions matching user interests.",
4
  "type": "object",
5
  "properties": {
6
    "city": {
7
      "type": "string",
8
      "description": "Destination city."
9
    },
10
    "preference": {
11
      "type": "string",
12
      "description": "Traveler preference."
13
    },
14
    "limit": {
15
      "type": "integer",
16
      "default": 5,
17
      "description": "Maximum number of attractions."
18
    }
19
  },
20
  "required": [
21
    "city",
22
    "preference"
23
  ]
24
}

1
from typing import Literal
2
from pydantic import BaseModel, Field
3
from langchain.tools import tool
4

5
class AttractionSearchInput(BaseModel):
6
    """Input schema for attraction search."""
7

8
    city: str = Field(
9
        description="Destination city name, such as Osaka"
10
    )
11
    preference: str = Field(
12
        description="Traveler interest, such as food or architecture"
13
    )
14
    limit: int = Field(
15
        default=5,
16
        ge=1,
17
        le=20,
18
        description="Maximum number of results"
19
    )
20
    language: Literal["zh", "en", "ja"] = Field(
21
        default="zh",
22
        description="Preferred result language"
23
    )
24

25
@tool(args_schema=AttractionSearchInput)
26
def search_attractions(
27
    city: str,
28
    preference: str,
29
    limit: int = 5,
30
    language: str = "zh",
31
) -> list[str]:
32
    """Search attractions matching traveler preferences."""
33
    return []

1
ATTRACTION_SCHEMA = {
2
    "title": "search_attractions",
3
    "description": "Search attractions matching traveler interests.",
4
    "type": "object",
5
    "properties": {
6
        "city": {
7
            "type": "string",
8
            "description": "Destination city",
9
        },
10
        "preference": {
11
            "type": "string",
12
            "description": "Traveler preference",
13
        },
14
    },
15
    "required": ["city", "preference"],
16
    "additionalProperties": False,
17
}