Tools（工具）

Tools 扩展了 Agents（智能体） 的能力——让它们能够获取实时数据、执行代码、查询外部数据库并在现实世界中采取行动。

在底层，tools 是具有明确定义 inputs 和 outputs 的可调用函数，这些函数被传递给 chat model。Model 根据 conversation context 决定何时调用 tool，以及提供什么 input arguments。

创建工具

基本工具定义

:::python 创建 tool 最简单的方法是使用 @[@tool] 装饰器。默认情况下，函数的 docstring 成为 tool 的 description，帮助 model 理解何时使用它：

from langchain.tools import tool

@tool
def search_database(query: str, limit: int = 10) -> str:
    """在客户数据库中搜索与查询匹配的记录。

    Args:
        query: 要查找的搜索词
        limit: 要返回的最大结果数
    """
    return f"Found {limit} results for '{query}'"

Type hints 是 必需的，因为它们定义了 tool 的 input schema。Docstring 应该信息丰富且简洁，以帮助 model 理解 tool 的 purpose。 :::

:::js 创建 tool 最简单的方法是从 langchain 包导入 tool 函数。你可以使用 zod 来定义 tool 的 input schema：

import * as z from "zod"
import { tool } from "langchain"

const searchDatabase = tool(
  ({ query, limit }) => `Found ${limit} results for '${query}'`,
  {
    name: "search_database",
    description: "Search the customer database for records matching the query.",
    schema: z.object({
      query: z.string().describe("Search terms to look for"),
      limit: z.number().describe("Maximum number of results to return"),
    }),
  }
);

:::

:::python

自定义工具属性

自定义工具名称

默认情况下，tool 名称来自函数名。当你需要更具描述性的名称时覆盖它：

@tool("web_search")  # 自定义名称
def search(query: str) -> str:
    """搜索网络获取信息。"""
    return f"Results for: {query}"

print(search.name)  # web_search

自定义工具描述

覆盖自动生成的 tool description 以获得更清晰的 model guidance：

@tool("calculator", description="Performs arithmetic calculations. Use this for any math problems.")
def calc(expression: str) -> str:
    """评估数学表达式。"""
    return str(eval(expression))

高级 schema 定义

使用 Pydantic models 或 JSON schemas 定义复杂 inputs：

class WeatherInput(BaseModel):
    """天气查询的输入。"""
    location: str = Field(description="城市名称或坐标")
    units: Literal["celsius", "fahrenheit"] = Field(
        default="celsius",
        description="温度单位偏好"
    )
    include_forecast: bool = Field(
        default=False,
        description="包含 5 天预报"
    )

@tool(args_schema=WeatherInput)
def get_weather(location: str, units: str = "celsius", include_forecast: bool = False) -> str:
    """获取当前天气和可选预报。"""
    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
```

```python 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:
    """获取当前天气和可选预报。"""
    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 arguments。使用这些名称将导致 runtime errors。

要访问 runtime 信息，使用 @[ToolRuntime] 参数而不是将你自己的 arguments 命名为 config 或 runtime。 :::

访问上下文

当 tools 能够访问 runtime 信息（如 conversation history、user data 和 persistent memory）时，它们最强大。本节介绍如何从你的 tools 内部访问和更新这些信息。

:::python Tools 可以通过 @[ToolRuntime] 参数访问 runtime 信息，它提供：

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:#DCFCE7,stroke:#16A34A,stroke-width:2px,color:#14532D
    classDef process fill:#DBEAFE,stroke:#2563EB,stroke-width:2px,color:#1E3A8A
    classDef output fill:#F3E8FF,stroke:#9333EA,stroke-width:2px,color:#581C87
    classDef neutral fill:#F3F4F6,stroke:#9CA3AF,stroke-width:2px,color:#374151

    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 表示在 conversation 持续期间存在的 short-term memory。它包括 message history 和你在 graph state 中定义的任何 custom fields。

访问 state

Tools 可以使用 runtime.state 访问当前 conversation state：

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

@tool
def get_last_user_message(runtime: ToolRuntime) -> str:
    """获取用户的最近消息。"""
    messages = runtime.state["messages"]

    # 查找最后一条 human message
    for message in reversed(messages):
        if isinstance(message, HumanMessage):
            return message.content

    return "No user messages found"

# 访问 custom state fields
@tool
def get_user_preference(
    pref_name: str,
    runtime: ToolRuntime
) -> str:
    """获取用户偏好值。"""
    preferences = runtime.state.get("user_preferences", {})
    return preferences.get(pref_name, "Not set")

更新 state

使用 @[Command] 更新 agent 的 state。这对于需要更新 custom state fields 的 tools 很有用：

from langgraph.types import Command
from langchain.tools import tool

@tool
def set_user_name(new_name: str) -> Command:
    """在 conversation state 中设置用户的名称。"""
    return Command(update={"user_name": new_name})

Context（上下文）

Context 提供在 invocation time 传递的 immutable configuration data。将它用于 user IDs、session details 或在 conversation 期间不应更改的 application-specific settings。

:::python 通过 runtime.context 访问 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:
    """获取当前用户的账户信息。"""
    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-4.1")
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")
)

:::

:::js Tools 可以通过 config 参数访问 agent 的 runtime context：

import * as z from "zod"
import { ChatOpenAI } from "@langchain/openai"
import { createAgent } from "langchain"

const getUserName = tool(
  (_, config) => {
    return config.context.user_name
  },
  {
    name: "get_user_name",
    description: "Get the user's name.",
    schema: z.object({}),
  }
);

const contextSchema = z.object({
  user_name: z.string(),
});

const agent = createAgent({
  model: new ChatOpenAI({ model: "gpt-4.1" }),
  tools: [getUserName],
  contextSchema,
});

const result = await agent.invoke(
  {
    messages: [{ role: "user", content: "What is my name?" }]
  },
  {
    context: { user_name: "John Smith" }
  }
);

:::

长期记忆（Store）

@[BaseStore] 提供在 conversations 之间持续的 persistent storage。与 state（short-term memory）不同，保存到 store 的数据在未来 sessions 中仍然可用。

:::python 通过 runtime.store 访问 store。Store 使用 namespace/key pattern 来组织数据：

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


# 访问 memory
@tool
def get_user_info(user_id: str, runtime: ToolRuntime) -> str:
    """查找用户信息。"""
    store = runtime.store
    user_info = store.get(("users",), user_id)
    return str(user_info.value) if user_info else "Unknown user"

# 更新 memory
@tool
def save_user_info(user_id: str, user_info: dict[str, Any], runtime: ToolRuntime) -> str:
    """保存用户信息。"""
    store = runtime.store
    store.put(("users",), user_id, user_info)
    return "Successfully saved user info."

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

# 第一个 session：保存用户信息
agent.invoke({
    "messages": [{"role": "user", "content": "Save the following user: userid: abc123, name: Foo, age: 25, email: foo@langchain.dev"}]
})

# 第二个 session：获取用户信息
agent.invoke({
    "messages": [{"role": "user", "content": "Get user info for user with id 'abc123'"}]
})
# 这是 ID 为 "abc123" 的用户的用户信息：
# - Name: Foo
# - Age: 25
# - Email: foo@langchain.dev

:::

:::js 通过 config.store 访问 store。Store 使用 namespace/key pattern 来组织数据：

import * as z from "zod";
import { createAgent, tool } from "langchain";
import { InMemoryStore } from "@langchain/langgraph";
import { ChatOpenAI } from "@langchain/openai";

const store = new InMemoryStore();

// 访问 memory
const getUserInfo = tool(
  async ({ user_id }) => {
    const value = await store.get(["users"], user_id);
    console.log("get_user_info", user_id, value);
    return value;
  },
  {
    name: "get_user_info",
    description: "Look up user info.",
    schema: z.object({
      user_id: z.string(),
    }),
  }
);

// 更新 memory
const saveUserInfo = tool(
  async ({ user_id, name, age, email }) => {
    console.log("save_user_info", user_id, name, age, email);
    await store.put(["users"], user_id, { name, age, email });
    return "Successfully saved user info.";
  },
  {
    name: "save_user_info",
    description: "Save user info.",
    schema: z.object({
      user_id: z.string(),
      name: z.string(),
      age: z.number(),
      email: z.string(),
    }),
  }
);

const agent = createAgent({
  model: new ChatOpenAI({ model: "gpt-4.1" }),
  tools: [getUserInfo, saveUserInfo],
  store,
});

// 第一个 session：保存用户信息
await agent.invoke({
  messages: [
    {
      role: "user",
      content: "Save the following user: userid: abc123, name: Foo, age: 25, email: foo@langchain.dev",
    },
  ],
});

// 第二个 session：获取用户信息
const result = await agent.invoke({
  messages: [
    { role: "user", content: "Get user info for user with id 'abc123'" },
  ],
});

console.log(result);
// 这是 ID 为 "abc123" 的用户的用户信息：
// - Name: Foo
// - Age: 25
// - Email: foo@langchain.dev

:::

Stream writer（流式写入器）

在 execution 期间从 tools stream real-time updates。这对于在 long-running operations 期间向用户提供 progress feedback 很有用。

:::python 使用 runtime.stream_writer emit custom updates：

from langchain.tools import tool, ToolRuntime

@tool
def get_weather(city: str, runtime: ToolRuntime) -> str:
    """获取给定城市的天气。"""
    writer = runtime.stream_writer

    # 在 tool 执行时 stream custom updates
    writer(f"Looking up data for city: {city}")
    writer(f"Acquired data for city: {city}")

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

:::js 使用 config.writer emit custom updates：

import * as z from "zod";
import { tool, ToolRuntime } from "langchain";

const getWeather = tool(
  ({ city }, config: ToolRuntime) => {
    const writer = config.writer;

    // 在 tool 执行时 stream custom updates
    if (writer) {
      writer(`Looking up data for city: ${city}`);
      writer(`Acquired data for city: ${city}`);
    }

    return `It's always sunny in ${city}!`;
  },
  {
    name: "get_weather",
    description: "Get weather for a given city.",
    schema: z.object({
      city: z.string(),
    }),
  }
);

:::

ToolNode

@[ToolNode] 是一个预构建的 node，在 LangGraph workflows 中执行 tools。它自动处理 parallel tool execution、error handling 和 state injection。

基本用法

:::python

from langchain.tools import tool
from langgraph.prebuilt import ToolNode
from langgraph.graph import StateGraph, MessagesState, START, END

@tool
def search(query: str) -> str:
    """搜索信息。"""
    return f"Results for: {query}"

@tool
def calculator(expression: str) -> str:
    """评估数学表达式。"""
    return str(eval(expression))

# 用你的 tools 创建 ToolNode
tool_node = ToolNode([search, calculator])

# 在 graph 中使用
builder = StateGraph(MessagesState)
builder.add_node("tools", tool_node)
# ... 添加其他 nodes 和 edges

:::

:::js

import { ToolNode } from "@langchain/langgraph/prebuilt";
import { tool } from "@langchain/core/tools";
import * as z from "zod";

const search = tool(
  ({ query }) => `Results for: ${query}`,
  {
    name: "search",
    description: "Search for information.",
    schema: z.object({ query: z.string() }),
  }
);

const calculator = tool(
  ({ expression }) => String(eval(expression)),
  {
    name: "calculator",
    description: "Evaluate a math expression.",
    schema: z.object({ expression: z.string() }),
  }
);

// 用你的 tools 创建 ToolNode
const toolNode = new ToolNode([search, calculator]);

:::

工具返回值

你可以为 tools 选择不同的 return values：

• 返回 string 用于 human-readable results。
• 返回 object 用于 model 应该 parse 的 structured results。
• 当你需要写入 state 时返回带有 optional message 的 Command。

返回字符串

当 tool 应该提供 plain text 供 model 读取并在其 next response 中使用时，返回 string。

:::python

from langgraph.prebuilt import ToolNode

# 默认：catch invocation errors，re-raise execution errors
tool_node = ToolNode(tools)

# Catch all errors 并返回 error message 给 LLM
tool_node = ToolNode(tools, handle_tool_errors=True)

# 自定义 error message
tool_node = ToolNode(tools, handle_tool_errors="Something went wrong, please try again.")

# 自定义 error handler
def handle_error(e: ValueError) -> str:
    return f"Invalid input: {e}"

tool_node = ToolNode(tools, handle_tool_errors=handle_error)

# 仅 catch specific exception types
tool_node = ToolNode(tools, handle_tool_errors=(ValueError, TypeError))

import { ToolNode } from "@langchain/langgraph/prebuilt";

// 默认 behavior
const toolNode = new ToolNode(tools);

// Catch all errors
const toolNode = new ToolNode(tools, { handleToolErrors: true });

// 自定义 error message
const toolNode = new ToolNode(tools, {
  handleToolErrors: "Something went wrong, please try again."
});

from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.graph import StateGraph, MessagesState, START, END

builder = StateGraph(MessagesState)
builder.add_node("llm", call_llm)
builder.add_node("tools", ToolNode(tools))

builder.add_edge(START, "llm")
builder.add_conditional_edges("llm", tools_condition)  # 路由到 "tools" 或 END
builder.add_edge("tools", "llm")

graph = builder.compile()

import { ToolNode, toolsCondition } from "@langchain/langgraph/prebuilt";
import { StateGraph, MessagesAnnotation } from "@langchain/langgraph";

const builder = new StateGraph(MessagesAnnotation)
  .addNode("llm", callLlm)
  .addNode("tools", new ToolNode(tools))
  .addEdge("__start__", "llm")
  .addConditionalEdges("llm", toolsCondition)  # 路由到 "tools" 或 "__end__"
  .addEdge("tools", "llm");

const graph = builder.compile();

from langchain.tools import tool, ToolRuntime
from langgraph.prebuilt import ToolNode

@tool
def get_message_count(runtime: ToolRuntime) -> str:
    """获取 conversation 中的消息数。"""
    messages = runtime.state["messages"]
    return f"There are {len(messages)} messages."

tool_node = ToolNode([get_message_count])