使用工具¶
工具 是一种将函数及其输入模式封装的方式,以便传递给支持工具调用的聊天模型。这使得模型可以请求使用特定输入执行该函数。本指南将展示如何创建工具并在你的图中使用它们。
创建工具¶
定义简单工具¶
要创建工具,你可以使用 @tool 装饰器或者普通的 Python 函数。
这需要使用 LangGraph 提供的预构建 ToolNode
或 代理,它们会自动将这些函数转换为 LangChain 工具。
自定义工具¶
为了更精细地控制工具的行为,请使用 @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 或会话上下文),这些参数不应由模型控制。
你可以将这些参数放入代理的 state
或 config
中,并在工具内部访问这些信息:
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 凭据。这些信息可以通过一个特殊的参数 annotation — RunnableConfig
在工具中访问:
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 允许代理在工具内部访问和更新其 短期记忆(状态)。
读取状态¶
要访问工具中的图状态,可以使用一个特殊的参数 annotation — InjectedState
:
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
对象,并返回它们的列表,例如:
长期记忆¶
使用长期记忆来跨会话存储用户特定或应用特定的数据。这对于聊天机器人等应用程序非常有用,您希望记住用户的偏好或其他信息。
要使用长期记忆,您需要:
读取¶
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"}}
)
InMemoryStore
是一种将数据存储在内存中的存储。在生产环境中,通常会使用数据库或其他持久存储。请查看 存储文档 获取更多选项。如果您使用的是 LangGraph 平台,平台将为您提供一个生产就绪的存储。- 在这个示例中,我们使用
put
方法向存储中写入一些示例数据。有关详细信息,请参阅 BaseStore.put API 参考。 - 第一个参数是命名空间。它用于将相关数据分组。在此示例中,我们使用
users
命名空间对用户数据进行分组。 - 命名空间内的一个键。此示例使用用户 ID 作为键。
- 要为指定用户存储的数据。
get_store
函数用于访问存储。您可以从代码的任何位置调用它,包括工具和提示。此函数返回创建代理时传递给它的存储。get
方法用于从存储中检索数据。第一个参数是命名空间,第二个参数是键。这将返回一个StoreValue
对象,其中包含值及其元数据。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
InMemoryStore
是一个在内存中存储数据的存储。在生产环境中,通常会使用数据库或其他持久化存储。请查看 存储文档 以了解更多信息。如果你使用的是 LangGraph 平台,平台将为你提供一个生产就绪的存储。UserInfo
类是一个TypedDict
,用于定义用户信息的结构。LLM 将使用它来根据模式格式化响应。save_user_info
函数是一个工具,允许代理更新用户信息。这在聊天应用程序中很有用,用户想要更新他们的个人资料信息。get_store
函数用于访问存储。你可以在代码的任何地方调用它,包括工具和提示。此函数返回创建代理时传递给它的存储。put
方法用于在存储中存储数据。第一个参数是命名空间,第二个参数是键。这将把用户信息存储在存储中。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})
如果你想让工具返回一个 ToolMessage,请使用工具调用进行调用:
tool_call = {
"type": "tool_call",
"id": "1",
"args": {"a": 42, "b": 7}
}
multiply.invoke(tool_call)
与聊天模型一起使用
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)
使用预构建代理¶
要创建一个工具调用代理,你可以使用预构建的 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
- 输出:包含
ToolMessage
的MessagesState
,即工具调用的结果
Tip
ToolNode
与 LangGraph 的预构建 agent 配合使用时效果非常好,但也可以与任何使用 MessagesState
的 StateGraph
一起使用。
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]})
多个工具调用
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)!
ToolNode
将并行执行这两个工具
与聊天模型一起使用
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]})
- 使用
.bind_tools()
将工具模式绑定到聊天模型
在工具调用代理中使用
这是一个使用 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
会捕获工具调用期间引发的所有异常,并将这些异常作为工具消息返回。要控制如何处理这些错误,可以使用 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])
# 使用错误处理(默认)
message = AIMessage(
content="",
tool_calls=[
{
"name": "multiply",
"args": {"a": 42, "b": 7},
"id": "tool_call_id",
"type": "tool_call",
}
],
)
tool_node.invoke({"messages": [message]})
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 参考。
有关不同工具错误处理选项的更多信息,请参见 API 参考。
处理大量工具¶
随着可用工具数量的增加,你可能希望限制LLM选择的范围,以减少token的消耗,并帮助管理LLM推理中的错误来源。
为了解决这个问题,你可以通过在运行时使用语义搜索来检索相关工具,从而动态调整模型可用的工具。
有关现成实现,请参阅 langgraph-bigtool
预构建库,以及这篇操作指南以获取更多详细信息。