使用工具¶

工具是一种封装函数及其输入模式的方式，可以将其传递给支持工具调用的聊天模型。这允许模型请求使用特定输入执行此函数。本指南展示了如何在图中创建和使用工具。

创建工具¶

定义简单工具¶

要创建工具，您可以使用 @tool 装饰器或原生 Python 函数。

@tool 装饰器Python 函数

from langchain_core.tools import tool

@tool
def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b

这需要使用 LangGraph 预置的 ToolNode 或代理，它们会自动将函数转换为 LangChain 工具。

def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b

自定义工具¶

要更精细地控制工具行为，请使用 @tool 装饰器。

^{API 参考：tool}

from langchain_core.tools import tool

@tool("multiply_tool", parse_docstring=True)
def multiply(a: int, b: int) -> int:
    """Multiply two numbers.

    Args:
        a: First operand
        b: Second operand
    """
    return a * b

您还可以使用 Pydantic 定义自定义输入模式

from pydantic import BaseModel, Field

class MultiplyInputSchema(BaseModel):
    """Multiply two numbers"""
    a: int = Field(description="First operand")
    b: int = Field(description="Second operand")

@tool("multiply_tool", args_schema=MultiplyInputSchema)
def multiply(a: int, b: int) -> int:
    return a * b

如需其他自定义，请参阅自定义工具指南。

对模型隐藏参数¶

某些工具需要仅运行时参数（例如，用户 ID 或会话上下文），这些参数不应由模型控制。

您可以将这些参数放在代理的 state 或 config 中，并在工具内部访问这些信息。

^{API 参考：tool | RunnableConfig | InjectedState}

from langchain_core.tools import tool
from langchain_core.runnables import RunnableConfig
from langgraph.prebuilt import InjectedState
from langgraph.graph import MessagesState

@tool
def my_tool(
    # This will be populated by an LLM
    tool_arg: str,
    # access information that's dynamically updated inside the agent
    state: Annotated[MessagesState, InjectedState],
    # access static data that is passed at agent invocation
    config: RunnableConfig,
) -> str:
    """My tool."""
    do_something_with_state(state["messages"])
    do_something_with_config(config)
    ...

访问配置¶

您可以在运行时向图提供静态信息，例如 user_id 或 API 凭据。可以通过特殊的参数注释 — RunnableConfig — 在工具内部访问此信息

^{API 参考：RunnableConfig | tool}

from langchain_core.runnables import RunnableConfig
from langchain_core.tools import tool

@tool
def get_user_info(
    config: RunnableConfig,
) -> str:
    """Look up user info."""
    user_id = config["configurable"].get("user_id")
    return "User is John Smith" if user_id == "user_123" else "Unknown user"

在工具中访问配置

from langchain_core.runnables import RunnableConfig
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent

def get_user_info(
    config: RunnableConfig,
) -> str:
    """Look up user info."""
    user_id = config["configurable"].get("user_id")
    return "User is John Smith" if user_id == "user_123" else "Unknown user"

agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[get_user_info],
)

agent.invoke(
    {"messages": [{"role": "user", "content": "look up user information"}]},
    config={"configurable": {"user_id": "user_123"}}
)

短期记忆¶

LangGraph 允许代理在工具内部访问和更新其短期记忆（状态）。

读取状态¶

要访问工具内部的图状态，您可以使用特殊的参数注释 — InjectedState

^{API 参考：tool | InjectedState}

from typing import Annotated
from langchain_core.tools import tool
from langgraph.prebuilt import InjectedState

class CustomState(AgentState):
    user_id: str

@tool
def get_user_info(
    state: Annotated[CustomState, InjectedState]
) -> str:
    """Look up user info."""
    user_id = state["user_id"]
    return "User is John Smith" if user_id == "user_123" else "Unknown user"

在工具中访问状态

from typing import Annotated
from langchain_core.tools import tool
from langgraph.prebuilt import InjectedState, create_react_agent

class CustomState(AgentState):
    user_id: str

@tool
def get_user_info(
    state: Annotated[CustomState, InjectedState]
) -> str:
    """Look up user info."""
    user_id = state["user_id"]
    return "User is John Smith" if user_id == "user_123" else "Unknown user"

agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[get_user_info],
    state_schema=CustomState,
)

agent.invoke({
    "messages": "look up user information",
    "user_id": "user_123"
})

更新状态¶

您可以直接从工具返回状态更新。这对于持久化中间结果或使信息可供后续工具或提示访问非常有用。

^{API 参考：Command | tool | InjectedToolCallId}

from langgraph.graph import MessagesState
from langgraph.types import Command
from langchain_core.tools import tool, InjectedToolCallId

class CustomState(MessagesState):
    user_name: str

@tool
def update_user_info(
    tool_call_id: Annotated[str, InjectedToolCallId],
    config: RunnableConfig
) -> Command:
    """Look up and update user info."""
    user_id = config["configurable"].get("user_id")
    name = "John Smith" if user_id == "user_123" else "Unknown user"
    return Command(update={
        "user_name": name,
        # update the message history
        "messages": [
            ToolMessage(
                "Successfully looked up user information",
                tool_call_id=tool_call_id
            )
        ]
    })

从工具更新状态

这是一个使用预置代理与可以更新图状态的工具的示例。

from typing import Annotated
from langchain_core.tools import tool, InjectedToolCallId
from langchain_core.runnables import RunnableConfig
from langchain_core.messages import ToolMessage
from langgraph.prebuilt import InjectedState, create_react_agent
from langgraph.prebuilt.chat_agent_executor import AgentState
from langgraph.types import Command

class CustomState(AgentState):
    user_name: str

@tool
def update_user_info(
    tool_call_id: Annotated[str, InjectedToolCallId],
    config: RunnableConfig
) -> Command:
    """Look up and update user info."""
    user_id = config["configurable"].get("user_id")
    name = "John Smith" if user_id == "user_123" else "Unknown user"
    return Command(update={
        "user_name": name,
        # update the message history
        "messages": [
            ToolMessage(
                "Successfully looked up user information",
                tool_call_id=tool_call_id
            )
        ]
    })

def greet(
    state: Annotated[CustomState, InjectedState]
) -> str:
    """Use this to greet the user once you found their info."""
    user_name = state["user_name"]
    return f"Hello {user_name}!"

agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[get_user_info, greet],
    state_schema=CustomState
)

agent.invoke(
    {"messages": [{"role": "user", "content": "greet the user"}]},
    config={"configurable": {"user_id": "user_123"}}
)

重要

如果您想使用返回 Command 并更新图状态的工具，您可以使用预置的 create_react_agent / ToolNode 组件，或者实现您自己的工具执行节点，该节点收集工具返回的 Command 对象并返回它们的列表，例如：

def call_tools(state):
    ...
    commands = [tools_by_name[tool_call["name"]].invoke(tool_call) for tool_call in tool_calls]
    return commands

长期记忆¶

使用长期记忆在对话中存储用户特定或应用程序特定的数据。这对于聊天机器人等应用程序非常有用，您希望记住用户偏好或其他信息。

要使用长期记忆，您需要

配置存储以在多次调用中持久化数据。
使用 get_store 函数从工具或提示中访问存储。

读取¶

^{API 参考：RunnableConfig | tool | StateGraph | get_store}

from langchain_core.runnables import RunnableConfig
from langchain_core.tools import tool
from langgraph.graph import StateGraph
from langgraph.config import get_store

@tool
def get_user_info(config: RunnableConfig) -> str:
    """Look up user info."""
    # Same as that provided to `builder.compile(store=store)` 
    # or `create_react_agent`
    store = get_store()
    user_id = config["configurable"].get("user_id")
    user_info = store.get(("users",), user_id)
    return str(user_info.value) if user_info else "Unknown user"

builder = StateGraph(...)
...
graph = builder.compile(store=store)

访问长期记忆

from langchain_core.runnables import RunnableConfig
from langchain_core.tools import tool
from langgraph.config import get_store
from langgraph.prebuilt import create_react_agent
from langgraph.store.memory import InMemoryStore

store = InMemoryStore() # (1)!

store.put(  # (2)!
    ("users",),  # (3)!
    "user_123",  # (4)!
    {
        "name": "John Smith",
        "language": "English",
    } # (5)!
)

@tool
def get_user_info(config: RunnableConfig) -> str:
    """Look up user info."""
    # Same as that provided to `create_react_agent`
    store = get_store() # (6)!
    user_id = config["configurable"].get("user_id")
    user_info = store.get(("users",), user_id) # (7)!
    return str(user_info.value) if user_info else "Unknown user"

agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[get_user_info],
    store=store # (8)!
)

# Run the agent
agent.invoke(
    {"messages": [{"role": "user", "content": "look up user information"}]},
    config={"configurable": {"user_id": "user_123"}}
)

InMemoryStore 是一个将数据存储在内存中的存储。在生产环境中，您通常会使用数据库或其他持久存储。请查阅存储文档以获取更多选项。如果您使用 LangGraph Platform 进行部署，该平台将为您提供生产就绪的存储。
对于此示例，我们使用 put 方法向存储写入一些示例数据。请参阅 BaseStore.put API 参考以获取更多详细信息。
第一个参数是命名空间。这用于将相关数据分组在一起。在此示例中，我们使用 users 命名空间来分组用户数据。
命名空间中的一个键。此示例使用用户 ID 作为键。
我们要为给定用户存储的数据。
get_store 函数用于访问存储。您可以从代码中的任何位置调用它，包括工具和提示。此函数返回在创建代理时传递给代理的存储。
get 方法用于从存储中检索数据。第一个参数是命名空间，第二个参数是键。这将返回一个 StoreValue 对象，其中包含值和有关值的元数据。
store 被传递给代理。这使得代理在运行工具时可以访问存储。您还可以使用 get_store 函数从代码中的任何位置访问存储。

更新¶

^{API 参考：RunnableConfig | tool | StateGraph | get_store}

from langchain_core.runnables import RunnableConfig
from langchain_core.tools import tool
from langgraph.graph import StateGraph
from langgraph.config import get_store

@tool
def save_user_info(user_info: str, config: RunnableConfig) -> str:
    """Save user info."""
    # Same as that provided to `builder.compile(store=store)` 
    # or `create_react_agent`
    store = get_store()
    user_id = config["configurable"].get("user_id")
    store.put(("users",), user_id, user_info)
    return "Successfully saved user info."

builder = StateGraph(...)
...
graph = builder.compile(store=store)

更新长期记忆

from typing_extensions import TypedDict

from langchain_core.tools import tool
from langgraph.config import get_store
from langgraph.prebuilt import create_react_agent
from langgraph.store.memory import InMemoryStore

store = InMemoryStore() # (1)!

class UserInfo(TypedDict): # (2)!
    name: str

@tool
def save_user_info(user_info: UserInfo, config: RunnableConfig) -> str: # (3)!
    """Save user info."""
    # Same as that provided to `create_react_agent`
    store = get_store() # (4)!
    user_id = config["configurable"].get("user_id")
    store.put(("users",), user_id, user_info) # (5)!
    return "Successfully saved user info."

agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet-latest",
    tools=[save_user_info],
    store=store
)

# Run the agent
agent.invoke(
    {"messages": [{"role": "user", "content": "My name is John Smith"}]},
    config={"configurable": {"user_id": "user_123"}} # (6)!
)

# You can access the store directly to get the value
store.get(("users",), "user_123").value

InMemoryStore 是一个将数据存储在内存中的存储。在生产环境中，您通常会使用数据库或其他持久存储。请查阅存储文档以获取更多选项。如果您使用 LangGraph Platform 进行部署，该平台将为您提供生产就绪的存储。
UserInfo 类是一个 TypedDict，它定义了用户信息结构。LLM 将使用它根据模式格式化响应。
save_user_info 函数是一个工具，允许代理更新用户信息。这对于聊天应用程序非常有用，用户希望更新其配置文件信息。
get_store 函数用于访问存储。您可以从代码中的任何位置调用它，包括工具和提示。此函数返回在创建代理时传递给代理的存储。
put 方法用于将数据存储在存储中。第一个参数是命名空间，第二个参数是键。这会将用户信息存储在存储中。
user_id 在配置中传递。这用于识别正在更新其信息的用户。

将工具附加到模型¶

要将工具模式附加到聊天模型，您需要使用 model.bind_tools()

^{API 参考：tool | init_chat_model}

from langchain_core.tools import tool
from langchain.chat_models import init_chat_model

@tool
def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b

model = init_chat_model(model="claude-3-5-haiku-latest")
model_with_tools = model.bind_tools([multiply])

model_with_tools.invoke("what's 42 x 7?")

AIMessage(
    content=[{'text': "I'll help you calculate that by using the multiply function.", 'type': 'text'}, {'id': 'toolu_01GhULkqytMTFDsNv6FsXy3Y', 'input': {'a': 42, 'b': 7}, 'name': 'multiply', 'type': 'tool_use'}]
    tool_calls=[{'name': 'multiply', 'args': {'a': 42, 'b': 7}, 'id': 'toolu_01GhULkqytMTFDsNv6FsXy3Y', 'type': 'tool_call'}]
)

使用工具¶

LangChain 工具符合Runnable 接口，这意味着您可以使用 .invoke() / .ainvoke() 方法执行它们

^{API 参考：tool}

from langchain_core.tools import tool

@tool
def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b

multiply.invoke({"a": 42, "b": 7})

如果您希望工具返回 ToolMessage，请使用工具调用来调用它

tool_call = {
    "type": "tool_call",
    "id": "1",
    "args": {"a": 42, "b": 7}
}
multiply.invoke(tool_call)

ToolMessage(content='294', name='multiply', tool_call_id='1')

与聊天模型一起使用

from langchain_core.tools import tool
from langchain.chat_models import init_chat_model

@tool
def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b

model = init_chat_model(model="claude-3-5-haiku-latest")
model_with_tools = model.bind_tools([multiply])

response_message = model_with_tools.invoke("what's 42 x 7?")
tool_call = response_message.tool_calls[0]

multiply.invoke(tool_call)

ToolMessage(content='294', name='multiply', tool_call_id='toolu_0176DV4YKSD8FndkeuuLj36c')

使用预置代理¶

要创建工具调用代理，您可以使用预置的 create_react_agent

^{API 参考：tool | create_react_agent}

from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent

@tool
def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b

agent = create_react_agent(
    model="anthropic:claude-3-7-sonnet",
    tools=[multiply]
)
graph.invoke({"messages": [{"role": "user", "content": "what's 42 x 7?"}]})

请参阅本指南了解更多信息。

使用预置 `ToolNode`¶

ToolNode 是一个预置的 LangGraph 节点，用于执行工具调用。

为什么要使用 ToolNode？

支持同步和异步工具
并行执行工具
工具执行期间的错误处理。您可以通过设置 handle_tool_errors=True（默认为启用）来启用/禁用此功能。有关错误处理的更多详细信息，请参阅此部分

ToolNode 对 MessagesState 进行操作

输入：MessagesState，其中最后一条消息是带有 tool_calls 参数的 AIMessage
输出：MessagesState，其中包含工具调用结果的 ToolMessage

提示

ToolNode 旨在与 LangGraph 预置的代理无缝协作，但也可以与任何使用 MessagesState 的 StateGraph 协作。

^{API 参考：ToolNode}

from langgraph.prebuilt import ToolNode

def get_weather(location: str):
    """Call to get the current weather."""
    if location.lower() in ["sf", "san francisco"]:
        return "It's 60 degrees and foggy."
    else:
        return "It's 90 degrees and sunny."

def get_coolest_cities():
    """Get a list of coolest cities"""
    return "nyc, sf"

tool_node = ToolNode([get_weather, get_coolest_cities])
tool_node.invoke({"messages": [...]})

单工具调用

from langchain_core.messages import AIMessage
from langgraph.prebuilt import ToolNode

# Define tools
@tool
def get_weather(location: str):
    """Call to get the current weather."""
    if location.lower() in ["sf", "san francisco"]:
        return "It's 60 degrees and foggy."
    else:
        return "It's 90 degrees and sunny."

tool_node = ToolNode([get_weather])

message_with_single_tool_call = AIMessage(
    content="",
    tool_calls=[
        {
            "name": "get_weather",
            "args": {"location": "sf"},
            "id": "tool_call_id",
            "type": "tool_call",
        }
    ],
)

tool_node.invoke({"messages": [message_with_single_tool_call]})

{'messages': [ToolMessage(content="It's 60 degrees and foggy.", name='get_weather', tool_call_id='tool_call_id')]}

多工具调用

from langchain_core.messages import AIMessage
from langgraph.prebuilt import ToolNode

# Define tools

def get_weather(location: str):
    """Call to get the current weather."""
    if location.lower() in ["sf", "san francisco"]:
        return "It's 60 degrees and foggy."
    else:
        return "It's 90 degrees and sunny."

def get_coolest_cities():
    """Get a list of coolest cities"""
    return "nyc, sf"

tool_node = ToolNode([get_weather, get_coolest_cities])

message_with_multiple_tool_calls = AIMessage(
    content="",
    tool_calls=[
        {
            "name": "get_coolest_cities",
            "args": {},
            "id": "tool_call_id_1",
            "type": "tool_call",
        },
        {
            "name": "get_weather",
            "args": {"location": "sf"},
            "id": "tool_call_id_2",
            "type": "tool_call",
        },
    ],
)

tool_node.invoke({"messages": [message_with_multiple_tool_calls]})  # (1)!

ToolNode 将并行执行这两个工具

{
    'messages': [
        ToolMessage(content='nyc, sf', name='get_coolest_cities', tool_call_id='tool_call_id_1'),
        ToolMessage(content="It's 60 degrees and foggy.", name='get_weather', tool_call_id='tool_call_id_2')
    ]
}

与聊天模型一起使用

from langchain.chat_models import init_chat_model
from langgraph.prebuilt import ToolNode

def get_weather(location: str):
    """Call to get the current weather."""
    if location.lower() in ["sf", "san francisco"]:
        return "It's 60 degrees and foggy."
    else:
        return "It's 90 degrees and sunny."

tool_node = ToolNode([get_weather])

model = init_chat_model(model="claude-3-5-haiku-latest")
model_with_tools = model.bind_tools([get_weather])  # (1)!


response_message = model_with_tools.invoke("what's the weather in sf?")
tool_node.invoke({"messages": [response_message]})

使用 .bind_tools() 将工具模式附加到聊天模型

{'messages': [ToolMessage(content="It's 60 degrees and foggy.", name='get_weather', tool_call_id='toolu_01Pnkgw5JeTRxXAU7tyHT4UW')]}

在工具调用代理中使用

这是一个使用 ToolNode 从头开始创建工具调用代理的示例。您也可以使用 LangGraph 预置的代理。

from langchain.chat_models import init_chat_model
from langgraph.prebuilt import ToolNode
from langgraph.graph import StateGraph, MessagesState, START, END

def get_weather(location: str):
    """Call to get the current weather."""
    if location.lower() in ["sf", "san francisco"]:
        return "It's 60 degrees and foggy."
    else:
        return "It's 90 degrees and sunny."

tool_node = ToolNode([get_weather])

model = init_chat_model(model="claude-3-5-haiku-latest")
model_with_tools = model.bind_tools([get_weather])

def should_continue(state: MessagesState):
    messages = state["messages"]
    last_message = messages[-1]
    if last_message.tool_calls:
        return "tools"
    return END

def call_model(state: MessagesState):
    messages = state["messages"]
    response = model_with_tools.invoke(messages)
    return {"messages": [response]}

builder = StateGraph(MessagesState)

# Define the two nodes we will cycle between
builder.add_node("call_model", call_model)
builder.add_node("tools", tool_node)

builder.add_edge(START, "call_model")
builder.add_conditional_edges("call_model", should_continue, ["tools", END])
builder.add_edge("tools", "call_model")

graph = builder.compile()

graph.invoke({"messages": [{"role": "user", "content": "what's the weather in sf?"}]})

{
    'messages': [
        HumanMessage(content="what's the weather in sf?"),
        AIMessage(
            content=[{'text': "I'll help you check the weather in San Francisco right now.", 'type': 'text'}, {'id': 'toolu_01A4vwUEgBKxfFVc5H3v1CNs', 'input': {'location': 'San Francisco'}, 'name': 'get_weather', 'type': 'tool_use'}],
            tool_calls=[{'name': 'get_weather', 'args': {'location': 'San Francisco'}, 'id': 'toolu_01A4vwUEgBKxfFVc5H3v1CNs', 'type': 'tool_call'}]
        ),
        ToolMessage(content="It's 60 degrees and foggy."),
        AIMessage(content="The current weather in San Francisco is 60 degrees and foggy. Typical San Francisco weather with its famous marine layer!")
    ]
}

错误处理¶

默认情况下，ToolNode 将捕获工具调用期间抛出的所有异常，并将它们作为工具消息返回。要控制如何处理错误，您可以使用 ToolNode 的 handle_tool_errors 参数。

启用错误处理（默认）禁用错误处理自定义错误处理

from langchain_core.messages import AIMessage
from langgraph.prebuilt import ToolNode

def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    if a == 42:
        raise ValueError("The ultimate error")
    return a * b

tool_node = ToolNode([multiply])

# Run with error handling (default)
message = AIMessage(
    content="",
    tool_calls=[
        {
            "name": "multiply",
            "args": {"a": 42, "b": 7},
            "id": "tool_call_id",
            "type": "tool_call",
        }
    ],
)

tool_node.invoke({"messages": [message]})

{'messages': [ToolMessage(content="Error: ValueError('The ultimate error')\n Please fix your mistakes.", name='multiply', tool_call_id='tool_call_id', status='error')]}

from langchain_core.messages import AIMessage
from langgraph.prebuilt import ToolNode

def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    if a == 42:
        raise ValueError("The ultimate error")
    return a * b

tool_node = ToolNode(
    [multiply],
    handle_tool_errors=False  # (1)!
)
message = AIMessage(
    content="",
    tool_calls=[
        {
            "name": "multiply",
            "args": {"a": 42, "b": 7},
            "id": "tool_call_id",
            "type": "tool_call",
        }
    ],
)
tool_node.invoke({"messages": [message]})

这将禁用错误处理（默认启用）。在 API 参考中查看所有可用的策略。

from langchain_core.messages import AIMessage
from langgraph.prebuilt import ToolNode

def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    if a == 42:
        raise ValueError("The ultimate error")
    return a * b

tool_node = ToolNode(
    [multiply],
    handle_tool_errors=(
        "Can't use 42 as a first operand, you must switch operands!"  # (1)!
    )
)
tool_node.invoke({"messages": [message]})

这提供了在发生异常时发送给 LLM 的自定义消息。在 API 参考中查看所有可用的策略。

{'messages': [ToolMessage(content="Can't use 42 as a first operand, you must switch operands!", name='multiply', tool_call_id='tool_call_id', status='error')]}

有关不同工具错误处理选项的更多信息，请参阅 API 参考。

处理大量工具¶

随着可用工具数量的增长，您可能希望限制 LLM 选择的范围，以减少令牌消耗并帮助管理 LLM 推理中的错误来源。

为了解决这个问题，您可以通过使用语义搜索在运行时检索相关工具来动态调整模型可用的工具。

有关即用型实现，请参阅 langgraph-bigtool 预置库；有关更多详细信息，请参阅本操作指南。

使用工具¶

创建工具¶

定义简单工具¶

自定义工具¶

对模型隐藏参数¶

访问配置¶

短期记忆¶

读取状态¶

更新状态¶

长期记忆¶

读取¶

更新¶

将工具附加到模型¶

使用工具¶

使用预置代理¶

使用预置 ToolNode¶

错误处理¶

处理大量工具¶

使用预置 `ToolNode`¶