工具扩展了代理 的能力——让它们能够获取实时数据、执行代码、查询外部数据库以及在世界中执行操作。
在底层,工具是具有明确定义输入和输出的可调用函数,这些函数被传递给聊天模型 。模型根据对话上下文决定何时调用工具,以及提供什么输入参数。
有关模型如何处理工具调用的详细信息,请参阅工具调用 。
创建工具
基本工具定义
创建工具最简单的方式是使用 @tool 装饰器。默认情况下,函数的文档字符串成为工具的描述,帮助模型理解何时使用它:
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 } '"
类型提示是必需的 ,因为它们定义了工具的输入模式。文档字符串应该信息丰富且简洁,以帮助模型理解工具的目的。
服务器端工具使用: 某些聊天模型具有内置工具(网络搜索、代码解释器),这些工具在服务器端执行。请参阅服务器端工具使用 了解详细信息。
工具名称优先使用 snake_case(例如 web_search 而不是 Web Search)。某些模型提供商对包含空格或特殊字符的名称有问题或拒绝,并返回错误。坚持使用字母数字字符、下划线和连字符有助于提高跨提供商的兼容性。
自定义工具属性
自定义工具名称
默认情况下,工具名称来自函数名称。在需要更描述性的名称时覆盖它:
@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 ))
高级模式定义
使用 Pydantic 模型或 JSON 模式定义复杂输入:
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 += " \n Next 5 days: Sunny"
return result
保留的参数名称
以下参数名称是保留的,不能用作工具参数。使用这些名称将导致运行时错误。
参数名称 用途 config保留用于在内部向工具传递 RunnableConfig runtime保留用于 ToolRuntime 参数(访问状态、上下文、存储)
要访问运行时信息,请使用 ToolRuntime 参数,而不是命名您自己的参数为 config 或 runtime。
访问上下文
当工具能够访问对话历史、用户数据和持久记忆等运行时信息时,它们是最强大的。本节介绍如何从工具内部访问和更新这些信息。
工具可以通过 ToolRuntime 参数访问运行时信息,它提供:
组件 描述 用例 状态 短期记忆 - 存在于当前对话中的可变数据(消息、计数器、自定义字段) 访问对话历史、跟踪工具调用计数 上下文 在调用时传递的不可变配置(用户 ID、会话信息) 根据用户身份个性化响应 存储 长期记忆 - 跨对话持久存在的数据 保存用户偏好、维护知识库 流写入器 在工具执行期间发出实时更新 显示长时间运行操作的进度 执行信息 当前执行的标识和重试信息(线程 ID、运行 ID、尝试次数) 访问线程/运行 ID、根据重试状态调整行为 服务器信息 在 LangGraph Server 上运行时服务器特定的元数据(助手 ID、图形 ID、认证用户) 访问助手 ID、图形 ID 或认证用户信息 配置 执行的 RunnableConfig 访问回调、标签和元数据 工具调用 ID 当前工具调用的唯一标识符 关联日志和模型调用的工具调用
短期记忆(状态)
状态表示在对话持续时间内存在的短期记忆。它包括消息历史和您在图形状态 中定义的任何自定义字段。
将 runtime: ToolRuntime 添加到您的工具签名中以访问状态。此参数是自动注入的,对 LLM 隐藏——它不会出现在工具的模式中。
访问状态
工具可以使用 runtime.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" )
runtime 参数对模型隐藏。对于上面的示例,模型只在工具模式中看到 pref_name。
更新状态
使用 Command 更新代理的状态。这对于需要更新自定义状态字段的工具很有用:
from langgraph . types import Command
from langchain . tools import tool
@tool
def set_user_name ( new_name : str ) -> Command :
"""Set the user's name in the conversation state."""
return Command ( update = { "user_name" : new_name })
当工具更新状态变量时,考虑为这些字段定义一个 reducer 。由于 LLM 可以并行调用多个工具,reducer 决定了当同一状态字段被并发工具调用更新时如何解决冲突。
上下文
上下文提供在调用时传递的不可变配置数据。将其用于用户 ID、会话详细信息或应用程序特定设置,这些在对话期间不应更改。
通过 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 ' ] }\n Type: { user [ ' account_type ' ] }\n Balance: $ { 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" )
)
长期记忆(存储)
BaseStore 提供跨对话持久存在的持久存储。与状态(短期记忆)不同,保存到存储的数据在未来的会话中仍然可用。
通过 runtime.store 访问存储。存储使用命名空间/键模式来组织数据:
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-4.1" )
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'" }]
})
# 以下是 ID 为 "abc123" 的用户信息:
# - 姓名:小明
# - 年龄:25
# - 邮箱:xiaoming@langchain.dev
See all 44 lines
流写入器
在执行期间从工具流式传输实时更新。这对于在长时间运行操作期间向用户提供进度反馈很有用。
使用 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 } !"
如果您在工具中使用 runtime.stream_writer,则必须在 LangGraph 执行上下文中调用该工具。请参阅流式传输 了解更多详情。
执行信息
通过 runtime.execution_info 从工具内部访问线程 ID、运行 ID 和重试状态:
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"
需要 deepagents>=0.5.0(或 langgraph>=1.1.5)。
服务器信息
当您的工具在 LangGraph Server 上运行时,通过 runtime.server_info 访问助手 ID、图形 ID 和认证用户:
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"
当工具不在 LangGraph Server 上运行时(例如在本地开发或测试期间),server_info 为 None。
需要 deepagents>=0.5.0(或 langgraph>=1.1.5)。
ToolNode 是一个预构建的节点,在 LangGraph 工作流中执行工具。它自动处理并行工具执行、错误处理和状态注入。
基本用法
from langchain . tools import tool
from langgraph . prebuilt import ToolNode
from langgraph . graph import StateGraph , MessagesState , START , END
@tool
def search ( query : str ) -> str :
"""Search for information."""
return f "Results for: { query } "
@tool
def calculator ( expression : str ) -> str :
"""Evaluate a math expression."""
return str ( eval ( expression ))
# 使用您的工具创建 ToolNode
tool_node = ToolNode ([ search , calculator ])
# 在图形中使用
builder = StateGraph ( MessagesState )
builder . add_node ( "tools" , tool_node )
# ... 添加其他节点和边
工具返回值
您可以为工具选择不同的返回值:
返回 string 用于人类可读的结果。
返回 object 用于模型应该解析的结构化结果。
在需要写入状态时返回带有可选消息的 Command。
返回字符串
当工具应该为模型提供纯文本供其读取并在下一个响应中使用时,返回字符串。
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 } ."
行为:
返回值被转换为 ToolMessage。
模型看到该文本并决定下一步做什么。
除非模型或另一个工具稍后更改,否则不会更改代理状态字段。
当结果本质上是人类可读的文本时使用此方式。
返回对象
当您的工具产生模型应该检查的结构化数据时,返回一个对象(例如 dict)。
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" ,
}
行为:
对象被序列化并作为工具输出发送。
模型可以读取特定字段并对其进行推理。
与字符串返回一样,这不会直接更新图形状态。
当后续推理受益于显式字段而不是自由格式文本时使用此方式。
返回 Command
当工具需要更新图形状态时(例如设置用户偏好或应用程序状态),返回 Command 。
您可以返回带有或不带有 ToolMessage 的 Command。
如果模型需要看到工具成功(例如确认偏好更改),请在更新中包含一个 ToolMessage,使用 runtime.tool_call_id 作为 tool_call_id 参数。
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 ,
)
],
}
)
行为:
命令使用 update 更新状态。
更新的状态在同一运行的后续步骤中可用。
对于可能由并行工具调用更新的字段使用 reducer。
当工具不仅是返回数据,而且还改变代理状态时使用此方式。
错误处理
配置工具错误的处理方式。请参阅 ToolNode API 参考了解所有选项。
from langgraph . prebuilt import ToolNode
# 默认:捕获调用错误,重新抛出执行错误
tool_node = ToolNode ( tools )
# 捕获所有错误并返回错误消息给 LLM
tool_node = ToolNode ( tools , handle_tool_errors = True )
# 自定义错误消息
tool_node = ToolNode ( tools , handle_tool_errors = "Something went wrong, please try again." )
# 自定义错误处理器
def handle_error ( e : ValueError ) -> str :
return f "Invalid input: { e } "
tool_node = ToolNode ( tools , handle_tool_errors = handle_error )
# 仅捕获特定异常类型
tool_node = ToolNode ( tools , handle_tool_errors = ( ValueError , TypeError ))
使用 tools_condition 根据 LLM 是否发出工具调用进行条件路由:
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 ()
状态注入
工具可以通过 ToolRuntime 访问当前图形状态:
from langchain . tools import tool , ToolRuntime
from langgraph . prebuilt import ToolNode
@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."
tool_node = ToolNode ([ get_message_count ])
有关从工具访问状态、上下文和长期记忆的更多详细信息,请参阅访问上下文 。
预构建工具
LangChain 为常见任务(如网络搜索、代码解释、数据库访问等)提供了大量预构建工具和工具包。这些随时可用的工具可以直接集成到您的代理中,无需编写自定义代码。
请参阅工具和工具包 集成页面,获取按类别组织的可用工具完整列表。
服务器端工具使用
某些聊天模型具有内置工具,在模型提供商端服务器执行。这些包括网络搜索和代码解释器等功能,无需您定义或托管工具逻辑。
请参阅各个聊天模型集成页面 和工具调用文档 ,了解有关启用和使用这些内置工具的详细信息。