跳转至

Tools

Abstract

Tools 扩展了 Agent 的能力,让 Agent 可以获取实时数据、执行代码、查询外部数据库,并在真实系统中采取行动。

从实现上看,tool 是一个带有明确输入和输出的可调用函数。它会被传给 chat model,模型根据当前对话上下文决定是否调用工具,以及应该提供哪些输入参数。关于模型如何产生 tool call,可以和 Models 里的 Tool Calling 一起理解。

创建 Tools

基本 Tools 定义

创建 tool 最简单的方式是使用 @tool 装饰器。默认情况下,函数名会成为 tool name,函数 docstring 会成为 tool description,帮助模型判断什么时候应该使用它。

from langchain.tools import tool

@tool
def search_database(query: str, limit: int = 10) -> str:
    """Search the customer database for records matching the query.

    Args:
        query: Search terms to look for
        limit: Maximum number of results to return
    """
    return f"Found {limit} results for '{query}'"

类型注解是必需的,因为它们会定义 tool 的输入 schema。docstring 应该简洁、明确,告诉模型这个工具的用途、输入含义和适用场景。

注意:一些 chat model 有内置的 server-side tools,例如 web search、code interpreter。这类工具由模型供应商服务端执行,不需要你自己定义或托管工具逻辑。

工具名建议使用 snake_case,例如 web_search,避免空格和特殊字符。部分 provider 会拒绝不兼容的 tool name。

自定义 Tools 属性

默认 tool name 来自函数名。如果需要更清晰的名称,可以手动覆盖:

@tool("web_search")
def search(query: str) -> str:
    """Search the web for information."""
    return f"Results for: {query}"

print(search.name)  # web_search

也可以覆盖自动生成的描述,让模型更准确地理解工具使用方式:

@tool("calculator", description="Performs arithmetic calculations. Use this for any math problems.")
def calc(expression: str) -> str:
    """Evaluate mathematical expressions."""
    return str(eval(expression))

工具描述会直接影响模型的工具选择。实践中应该写清楚“这个工具做什么”“什么时候使用”“参数代表什么”,而不是只写一个很泛的能力说明。

高级 Schema 定义

复杂输入可以用 Pydantic model 或 JSON Schema 定义。这样可以让字段描述、默认值、枚举约束更明确。

from pydantic import BaseModel, Field
from typing import Literal

class WeatherInput(BaseModel):
    """Input for weather queries."""

    location: str = Field(description="City name or coordinates")
    units: Literal["celsius", "fahrenheit"] = Field(
        default="celsius",
        description="Temperature unit preference",
    )
    include_forecast: bool = Field(
        default=False,
        description="Include 5-day forecast",
    )

@tool(args_schema=WeatherInput)
def get_weather(
    location: str,
    units: str = "celsius",
    include_forecast: bool = False,
) -> str:
    """Get current weather and optional forecast."""
    temp = 22 if units == "celsius" else 72
    result = f"Current weather in {location}: {temp} degrees {units[0].upper()}"
    if include_forecast:
        result += "\nNext 5 days: Sunny"
    return result

也可以使用 JSON Schema:

weather_schema = {
    "type": "object",
    "properties": {
        "location": {"type": "string"},
        "units": {"type": "string"},
        "include_forecast": {"type": "boolean"},
    },
    "required": ["location", "units", "include_forecast"],
}

@tool(args_schema=weather_schema)
def get_weather(
    location: str,
    units: str = "celsius",
    include_forecast: bool = False,
) -> str:
    """Get current weather and optional forecast."""
    temp = 22 if units == "celsius" else 72
    result = f"Current weather in {location}: {temp} degrees {units[0].upper()}"
    if include_forecast:
        result += "\nNext 5 days: Sunny"
    return result

保留参数名

下面两个参数名是保留字段,不能作为普通 tool argument 使用:

Parameter name Purpose
config 内部用于向 tool 传递 RunnableConfig
runtime 用于 ToolRuntime 参数,访问 state、context、store 等运行时信息

如果需要访问运行时信息,应该使用 ToolRuntime,而不是把 configruntime 当作模型可见的普通参数。

访问 Context

当 tools 能访问对话历史、用户数据、长期记忆和执行上下文时,它们会更有用。LangChain 通过 ToolRuntime 向 tool 注入这些信息。

ToolRuntime 提供的常见组件:

Component Description Use case
State 当前会话中的短期记忆,包括 messages、计数器、自定义字段 访问对话历史、记录 tool 调用次数
Context 调用时传入的不可变配置,例如 user id、session info 基于用户身份做个性化响应
Store 跨会话持久化的长期记忆 保存用户偏好、维护知识库
Stream Writer 工具执行过程中发出实时更新 为长任务展示进度
Execution Info 当前执行的 thread id、run id、attempt number 记录运行身份、根据重试状态调整行为
Server Info 在 LangGraph Server 上运行时的 assistant id、graph id、认证用户 访问服务端运行元信息
Config 当前执行的 RunnableConfig 访问 callbacks、tags、metadata
Tool Call ID 当前工具调用的唯一 ID 关联日志、生成对应 ToolMessage
graph LR
    %% Runtime Context
    subgraph "🔧 Tool Runtime Context"
        A[Tool Call] --> B[ToolRuntime]
        B --> C[State Access]
        B --> D[Context Access]
        B --> E[Store Access]
        B --> F[Stream Writer]
    end

    %% Available Resources
    subgraph "📊 Available Resources"
        C --> G[Messages]
        C --> H[Custom State]
        D --> I[User ID]
        D --> J[Session Info]
        E --> K[Long-term Memory]
        E --> L[User Preferences]
    end

    %% Tool Capabilities
    subgraph "⚡ Enhanced Tool Capabilities"
        M[Context-Aware Tools]
        N[Stateful Tools]
        O[Memory-Enabled Tools]
        P[Streaming Tools]
    end

    %% Connections
    G --> M
    H --> N
    I --> M
    J --> M
    K --> O
    L --> O
    F --> P

    classDef trigger fill:#F6FFDB,stroke:#6E8900,stroke-width:2px,color:#2E3900
    classDef process fill:#E5F4FF,stroke:#006DDD,stroke-width:2px,color:#030710
    classDef output fill:#EBD0F0,stroke:#885270,stroke-width:2px,color:#441E33
    classDef neutral fill:#F2FAFF,stroke:#40668D,stroke-width:2px,color:#2F4B68

    class A trigger
    class B,C,D,E,F process
    class G,H,I,J,K,L neutral
    class M,N,O,P output

短期记忆:State

State 表示一次会话中的短期记忆,包括消息历史和自定义 graph state 字段。只要在 tool 签名里加入 runtime: ToolRuntime,LangChain 就会自动注入 runtime;这个参数不会出现在模型可见的 tool schema 里。

读取 State

可以通过 runtime.state 访问当前 conversation state:

from langchain.tools import tool, ToolRuntime
from langchain.messages import HumanMessage

@tool
def get_last_user_message(runtime: ToolRuntime) -> str:
    """Get the most recent message from the user."""
    messages = runtime.state["messages"]

    for message in reversed(messages):
        if isinstance(message, HumanMessage):
            return message.content

    return "No user messages found"

@tool
def get_user_preference(pref_name: str, runtime: ToolRuntime) -> str:
    """Get a user preference value."""
    preferences = runtime.state.get("user_preferences", {})
    return preferences.get(pref_name, "Not set")

上面第二个工具里,模型只能看到 pref_nameruntime 是隐藏注入参数。

更新 State

如果工具需要更新 Agent 状态,可以返回 Command。如果模型需要看到工具结果,应该在 update 中包含一条 ToolMessage

from langchain.agents import AgentState
from langchain.messages import ToolMessage
from langchain.tools import ToolRuntime, tool
from langgraph.types import Command

class CustomState(AgentState):
    user_name: str

@tool
def set_user_name(new_name: str, runtime: ToolRuntime[None, CustomState]) -> Command:
    """Set the user's name in the conversation state."""
    return Command(
        update={
            "user_name": new_name,
            "messages": [
                ToolMessage(
                    content=f"User name set to {new_name}.",
                    tool_call_id=runtime.tool_call_id,
                )
            ],
        }
    )

如果多个工具可能并行更新同一个字段,需要考虑 reducer,避免并发更新互相覆盖。

调用上下文:Context

Context 是调用时传入的不可变配置,适合存放 user id、session details、租户信息或权限信息。工具通过 runtime.context 读取它。

from dataclasses import dataclass
from langchain_openai import ChatOpenAI
from langchain.agents import create_agent
from langchain.tools import tool, ToolRuntime

USER_DATABASE = {
    "user123": {
        "name": "Alice Johnson",
        "account_type": "Premium",
        "balance": 5000,
        "email": "alice@example.com",
    },
    "user456": {
        "name": "Bob Smith",
        "account_type": "Standard",
        "balance": 1200,
        "email": "bob@example.com",
    },
}

@dataclass
class UserContext:
    user_id: str

@tool
def get_account_info(runtime: ToolRuntime[UserContext]) -> str:
    """Get the current user's account information."""
    user_id = runtime.context.user_id

    if user_id in USER_DATABASE:
        user = USER_DATABASE[user_id]
        return f"Account holder: {user['name']}\nType: {user['account_type']}\nBalance: ${user['balance']}"
    return "User not found"

model = ChatOpenAI(model="gpt-5.4")
agent = create_agent(
    model,
    tools=[get_account_info],
    context_schema=UserContext,
    system_prompt="You are a financial assistant.",
)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "What's my current balance?"}]},
    context=UserContext(user_id="user123"),
)

长期记忆:Store

BaseStore 提供跨会话持久化存储。和 state 不同,store 中的数据可以在后续 session 中继续使用。访问方式是 runtime.store,通常使用 namespace/key 组织数据。

开发和示例可以使用 InMemoryStore;生产环境应使用持久化 store,例如 Postgres。

from typing import Any
from langgraph.store.memory import InMemoryStore
from langchain.agents import create_agent
from langchain.tools import tool, ToolRuntime
from langchain_openai import ChatOpenAI

@tool
def get_user_info(user_id: str, runtime: ToolRuntime) -> str:
    """Look up user info."""
    store = runtime.store
    user_info = store.get(("users",), user_id)
    return str(user_info.value) if user_info else "Unknown user"

@tool
def save_user_info(user_id: str, user_info: dict[str, Any], runtime: ToolRuntime) -> str:
    """Save user info."""
    store = runtime.store
    store.put(("users",), user_id, user_info)
    return "Successfully saved user info."

model = ChatOpenAI(model="gpt-5.4")
store = InMemoryStore()

agent = create_agent(
    model,
    tools=[get_user_info, save_user_info],
    store=store,
)

agent.invoke({
    "messages": [
        {
            "role": "user",
            "content": "Save the following user: userid: abc123, name: Foo, age: 25, email: foo@langchain.dev",
        }
    ]
})

agent.invoke({
    "messages": [{"role": "user", "content": "Get user info for user with id 'abc123'"}]
})

流式进度:Stream Writer

工具执行时间较长时,可以用 runtime.stream_writer 输出实时进度。

from langchain.tools import tool, ToolRuntime

@tool
def get_weather(city: str, runtime: ToolRuntime) -> str:
    """Get weather for a given city."""
    writer = runtime.stream_writer

    writer(f"Looking up data for city: {city}")
    writer(f"Acquired data for city: {city}")

    return f"It's always sunny in {city}!"

如果在 tool 中使用 runtime.stream_writer,tool 需要在 LangGraph execution context 中被调用。

执行信息:Execution Info

工具可以通过 runtime.execution_info 读取 thread id、run id 和 retry state。

from langchain.tools import tool, ToolRuntime

@tool
def log_execution_context(runtime: ToolRuntime) -> str:
    """Log execution identity information."""
    info = runtime.execution_info
    print(f"Thread: {info.thread_id}, Run: {info.run_id}")
    print(f"Attempt: {info.node_attempt}")
    return "done"

服务端信息:Server Info

当 tool 运行在 LangGraph Server 上时,可以通过 runtime.server_info 读取 assistant id、graph id 和 authenticated user。它在本地开发或测试中通常是 None

from langchain.tools import tool, ToolRuntime

@tool
def get_assistant_scoped_data(runtime: ToolRuntime) -> str:
    """Fetch data scoped to the current assistant."""
    server = runtime.server_info
    if server is not None:
        print(f"Assistant: {server.assistant_id}, Graph: {server.graph_id}")
        if server.user is not None:
            print(f"User: {server.user.identity}")
    return "done"

工具执行

在 LangChain 中,tools 通常由 Agent 使用,例如通过 create_agent 注册。工具错误处理一般通过 middleware 配置。对于 LangGraph workflow,工具执行由 ToolNode 处理。

工具返回值

工具可以返回不同类型的值:

  • 返回 string:给模型看的自然语言结果。
  • 返回 object:给模型解析的结构化结果。
  • 返回 Command:需要写入 state 时使用,可以同时带上 ToolMessage

返回字符串

字符串返回值会转换为 ToolMessage。模型读取这段文本后,决定下一步如何回答或继续调用工具。

from langchain.tools import tool

@tool
def get_weather(city: str) -> str:
    """Get weather for a city."""
    return f"It is currently sunny in {city}."

适用于结果本身就是人类可读文本的场景。

返回对象

如果工具产生结构化数据,可以返回 dict 等对象。对象会被序列化为 tool output,模型可以读取其中字段继续推理。

from langchain.tools import tool

@tool
def get_weather_data(city: str) -> dict:
    """Get structured weather data for a city."""
    return {
        "city": city,
        "temperature_c": 22,
        "conditions": "sunny",
    }

这种返回方式不会直接更新 graph state。

返回 Command

当工具需要更新 graph state 时,返回 Command。如果模型需要看到工具执行成功的消息,需要在 update 中加入 ToolMessage,并使用 runtime.tool_call_id 关联当前 tool call。

from langchain.messages import ToolMessage
from langchain.tools import ToolRuntime, tool
from langgraph.types import Command

@tool
def set_language(language: str, runtime: ToolRuntime) -> Command:
    """Set the preferred response language."""
    return Command(
        update={
            "preferred_language": language,
            "messages": [
                ToolMessage(
                    content=f"Language set to {language}.",
                    tool_call_id=runtime.tool_call_id,
                )
            ],
        }
    )

错误处理

工具调用可能失败。可以用 LangChain Agent middleware 捕获异常,并返回模型可以理解的 ToolMessage

from collections.abc import Callable

from langchain.agents import create_agent
from langchain.agents.middleware import wrap_tool_call
from langchain.messages import ToolMessage
from langchain.tools.tool_node import ToolCallRequest

@wrap_tool_call
def handle_tool_errors(
    request: ToolCallRequest,
    handler: Callable[[ToolCallRequest], ToolMessage],
) -> ToolMessage:
    """Convert tool exceptions into ToolMessages the model can handle."""
    try:
        return handler(request)
    except Exception as e:
        return ToolMessage(
            content=f"Tool error: Please check your input and try again. ({e})",
            tool_call_id=request.tool_call["id"],
        )

agent = create_agent(
    model="openai:gpt-5.4",
    tools=[],
    middleware=[handle_tool_errors],
)

错误消息应该能帮助模型修正输入或调整策略,但不要暴露敏感内部信息。

State 注入

Tools 可以通过 ToolRuntime 访问当前 graph state:

from langchain.tools import tool, ToolRuntime

@tool
def get_message_count(runtime: ToolRuntime) -> str:
    """Get the number of messages in the conversation."""
    messages = runtime.state["messages"]
    return f"There are {len(messages)} messages."

如果需要更多 state、context 和 store 访问方式,可以参考上面的 Access Context 部分。

预构建工具

LangChain 提供了大量预构建 tools 和 toolkits,覆盖 web search、code interpretation、database access 等常见任务。这些工具可以直接集成到 Agent 中,减少重复实现。

使用预构建工具时,仍然要关注权限、错误处理、返回内容长度、限流和可观测性。工具能跑起来只是第一步,能被模型稳定、正确地使用才是关键。

服务端工具调用

部分 chat model 提供内置工具,这些工具由 provider 在服务端执行,例如 web search 或 code interpreter。使用 server-side tools 时,你不需要定义或托管工具函数,但能力范围、执行细节和可观测性会受到 provider 限制。

客户端工具更适合访问私有业务系统和自定义动作;server-side tools 更适合快速接入 provider 已经提供的通用能力。

小结

Tools 是 Agent 行动能力的来源。一个 tool 不只是一个 Python 函数,它还包括名称、描述、输入 schema、运行时上下文、返回值、错误处理和状态更新策略。设计良好的 tools 可以显著提升 Agent 的可靠性;设计不清晰的 tools 则会增加误调用、错误参数和上下文污染的概率。