Skip to content

使用工具

工具 是一种将函数及其输入模式封装的方式,以便传递给支持工具调用的聊天模型。这使得模型可以请求使用特定输入执行该函数。本指南将展示如何创建工具并在你的图中使用它们。

创建工具

定义简单工具

要创建工具,你可以使用 @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 Reference: 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 或会话上下文),这些参数不应由模型控制。

你可以将这些参数放入代理的 stateconfig 中,并在工具内部访问这些信息:

API Reference: 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(
    # 这将由 LLM 填充
    tool_arg: str,
    # 访问代理内部动态更新的信息
    state: Annotated[MessagesState, InjectedState],
    # 访问在代理调用时传递的静态数据
    config: RunnableConfig,
) -> str:
    """My tool."""
    do_something_with_state(state["messages"])
    do_something_with_config(config)
    ...

访问配置

你可以在运行时向图提供静态信息,例如 user_id 或 API 凭据。这些信息可以通过一个特殊的参数 annotationRunnableConfig 在工具中访问:

API Reference: RunnableConfig | tool

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

@tool
def get_user_info(
    config: RunnableConfig,
) -> str:
    """查找用户信息."""
    user_id = config["configurable"].get("user_id")
    return "用户是 John Smith" if user_id == "user_123" else "未知用户"
在工具中访问配置
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:
    """查找用户信息."""
    user_id = config["configurable"].get("user_id")
    return "用户是 John Smith" if user_id == "user_123" else "未知用户"

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

agent.invoke(
    {"messages": [{"role": "user", "content": "查找用户信息"}]},
    config={"configurable": {"user_id": "user_123"}}
)

短期记忆

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

读取状态

要访问工具中的图状态,可以使用一个特殊的参数 annotationInjectedState

API Reference: 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:
    """查找用户信息。"""
    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:
    """查找用户信息。"""
    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 Reference: 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:
    """查找并更新用户信息。"""
    user_id = config["configurable"].get("user_id")
    name = "John Smith" if user_id == "user_123" else "Unknown user"
    return Command(update={
        "user_name": name,
        # 更新消息历史
        "messages": [
            ToolMessage(
                "成功查找用户信息",
                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:
    """查找并更新用户信息。"""
    user_id = config["configurable"].get("user_id")
    name = "John Smith" if user_id == "user_123" else "Unknown user"
    return Command(update={
        "user_name": name,
        # 更新消息历史
        "messages": [
            ToolMessage(
                "成功查找用户信息",
                tool_call_id=tool_call_id
            )
        ]
    })

def greet(
    state: Annotated[CustomState, InjectedState]
) -> str:
    """一旦找到用户信息,使用此方法向用户问好。"""
    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"}}
)

Important

如果你想使用返回 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

长期记忆

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

要使用长期记忆,您需要:

  1. 配置一个存储以在调用之间持久化数据。
  2. 使用 get_store 函数从工具或提示中访问存储。

读取

API Reference: 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:
    """查找用户信息。"""
    # 与提供给 `builder.compile(store=store)` 
    # 或 `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:
    """查找用户信息。"""
    # 与提供给 `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)!
)

# 运行代理
agent.invoke(
    {"messages": [{"role": "user", "content": "look up user information"}]},
    config={"configurable": {"user_id": "user_123"}}
)
  1. InMemoryStore 是一种将数据存储在内存中的存储。在生产环境中,通常会使用数据库或其他持久存储。请查看 存储文档 获取更多选项。如果您使用的是 LangGraph 平台,平台将为您提供一个生产就绪的存储。
  2. 在这个示例中,我们使用 put 方法向存储中写入一些示例数据。有关详细信息,请参阅 BaseStore.put API 参考。
  3. 第一个参数是命名空间。它用于将相关数据分组。在此示例中,我们使用 users 命名空间对用户数据进行分组。
  4. 命名空间内的一个键。此示例使用用户 ID 作为键。
  5. 要为指定用户存储的数据。
  6. get_store 函数用于访问存储。您可以从代码的任何位置调用它,包括工具和提示。此函数返回创建代理时传递给它的存储。
  7. get 方法用于从存储中检索数据。第一个参数是命名空间,第二个参数是键。这将返回一个 StoreValue 对象,其中包含值及其元数据。
  8. store 被传递给代理。这使得代理在运行工具时能够访问存储。您也可以使用 get_store 函数从代码的任何位置访问存储。

更新

API Reference: 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:
    """保存用户信息。"""
    # 与传递给 `builder.compile(store=store)` 
    # 或 `create_react_agent` 的相同
    store = get_store()
    user_id = config["configurable"].get("user_id")
    store.put(("users",), user_id, user_info)
    return "成功保存用户信息。"

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)!
    """保存用户信息。"""
    # 与传递给 `create_react_agent` 的相同
    store = get_store() # (4)!
    user_id = config["configurable"].get("user_id")
    store.put(("users",), user_id, user_info) # (5)!
    return "成功保存用户信息。"

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

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

# 你可以直接访问存储以获取值
store.get(("users",), "user_123").value
  1. InMemoryStore 是一个在内存中存储数据的存储。在生产环境中,通常会使用数据库或其他持久化存储。请查看 存储文档 以了解更多信息。如果你使用的是 LangGraph 平台,平台将为你提供一个生产就绪的存储。
  2. UserInfo 类是一个 TypedDict,用于定义用户信息的结构。LLM 将使用它来根据模式格式化响应。
  3. save_user_info 函数是一个工具,允许代理更新用户信息。这在聊天应用程序中很有用,用户想要更新他们的个人资料信息。
  4. get_store 函数用于访问存储。你可以在代码的任何地方调用它,包括工具和提示。此函数返回创建代理时传递给它的存储。
  5. put 方法用于在存储中存储数据。第一个参数是命名空间,第二个参数是键。这将把用户信息存储在存储中。
  6. user_id 在配置中传递。这用于标识正在更新信息的用户。

将工具附加到模型

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

API Reference: 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 Reference: 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})
294

如果你想让工具返回一个 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 Reference: 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
  • 输出:包含 ToolMessageMessagesState,即工具调用的结果

Tip

ToolNode 与 LangGraph 的预构建 agent 配合使用时效果非常好,但也可以与任何使用 MessagesStateStateGraph 一起使用。

API Reference: ToolNode

from langgraph.prebuilt import ToolNode

def get_weather(location: str):
    """获取当前天气的调用."""
    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():
    """获取最酷城市的列表"""
    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

# 定义工具
@tool
def get_weather(location: str):
    """获取当前天气的调用."""
    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

# 定义工具

def get_weather(location: str):
    """获取当前天气的调用."""
    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():
    """获取最酷城市的列表"""
    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)!
  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):
    """获取当前天气的调用."""
    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]})
  1. 使用 .bind_tools() 将工具模式绑定到聊天模型
{'messages': [ToolMessage(content="It's 60 degrees and foggy.", name='get_weather', tool_call_id='toolu_01Pnkgw5JeTRxXAU7tyHT4UW')]}
在工具调用代理中使用

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

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):
    """获取当前天气的调用."""
    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)

# 定义我们循环使用的两个节点
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 会捕获工具调用期间引发的所有异常,并将这些异常作为工具消息返回。要控制如何处理这些错误,可以使用 ToolNodehandle_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])

# 使用错误处理(默认)
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]})
  1. 此设置禁用了错误处理(默认是启用的)。有关所有可用策略,请参见 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]})
  1. 这提供了在发生异常时发送给 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选择的范围,以减少token的消耗,并帮助管理LLM推理中的错误来源。

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

有关现成实现,请参阅 langgraph-bigtool 预构建库,以及这篇操作指南以获取更多详细信息。