工具调用
原文:Tools 工具调用
一句话
工具(tools)扩展了 agents 的能力,使其能够获取实时数据、执行代码、查询外部数据库并在世界中执行操作。
什么时候翻这页
当你需要为 agent 创建自定义工具,或者需要理解工具如何与模型交互时。 当你需要访问运行时信息(如对话历史、用户数据和持久化内存)时。 当你需要处理工具错误或进行动态工具选择时。
核心概念
- 工具(Tool): 可调用函数,具有明确定义的输入和输出,传递给 chat model
- 工具描述(Tool Description): 函数的文档字符串,帮助模型理解何时使用该工具
- 工具运行时(ToolRuntime): 提供对状态、上下文、存储等运行时信息的访问
- 中间件(Middleware): 用于处理工具调用、错误处理和动态工具选择的机制
- 直接返回(return_direct): 工具输出直接作为最终响应,绕过额外的模型调用
怎么做
创建基本工具
使用 @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}'"
自定义工具属性
自定义工具名称
@tool("web_search") # 自定义名称
def search(query: str) -> str:
"""Search the web for information."""
return f"Results for: {query}"
自定义工具描述
@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 模型定义复杂输入:
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 += "\nNext 5 days: Sunny"
return result
访问运行时信息
通过 ToolRuntime 参数访问运行时信息:
from langchain.tools import tool, ToolRuntime
@tool
def get_last_user_message(runtime: ToolRuntime) -> str:
"""Get the most recent message from the user."""
messages = runtime.state["messages"]
# Find the last human message
for message in reversed(messages):
if isinstance(message, HumanMessage):
return message.content
return "No user messages found"
更新状态
使用 Command 更新 agent 的状态:
from langchain.agents import AgentState
from langchain.messages import ToolMessage
from langchain.tools import ToolRuntime, tool
from langgraph.types import Command
class CustomState(AgentState):
user_name: str
@tool
def set_user_name(new_name: str, runtime: ToolRuntime[None, CustomState]) -> Command:
"""Set the user's name in the conversation state."""
return Command(
update={
"user_name": new_name,
"messages": [
ToolMessage(
content=f"User name set to {new_name}.",
tool_call_id=runtime.tool_call_id,
)
],
}
)
直接返回工具
使用 return_direct 让工具输出直接作为最终响应:
from langchain.agents import create_agent
from langchain.tools import tool
from langchain_openai import ChatOpenAI
@tool(return_direct=True)
def fetch_order_status(order_id: str) -> str:
"""Fetch the current status of a customer order."""
# In production, query your order management system here
return f"Order {order_id} is shipped and will arrive in 2 days."
agent = create_agent(
ChatOpenAI(model="google_genai:gemini-3.5-flash"),
tools=[fetch_order_status],
)
result = agent.invoke({
"messages": [{"role": "user", "content": "What is the status of order #12345?"}]
})
# The agent returns the tool output directly without another LLM call:
# "Order 12345 is shipped and will arrive in 2 days."
错误处理
使用中间件处理工具错误:
from collections.abc import Callable
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_tool_call
from langchain.messages import ToolMessage
from langchain.tools.tool_node import ToolCallRequest
@wrap_tool_call
def handle_tool_errors(
request: ToolCallRequest,
handler: Callable[[ToolCallRequest], ToolMessage],
) -> ToolMessage:
"""Convert tool exceptions into ToolMessages the model can handle."""
try:
return handler(request)
except Exception as e:
return ToolMessage(
content=f"Tool error: Please check your input and try again. ({e})",
tool_call_id=request.tool_call["id"],
)
agent = create_agent(
model="google_genai:gemini-3.5-flash",
tools=[],
middleware=[handle_tool_errors],
)
动态工具选择
基于状态过滤工具:
from langchain.agents import create_agent
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
from typing import Callable
@wrap_model_call
def state_based_tools(
request: ModelRequest,
handler: Callable[[ModelRequest], ModelResponse]
) -> ModelResponse:
"""Filter tools based on conversation State."""
# Read from State: check if user has authenticated
state = request.state
is_authenticated = state.get("authenticated", False)
message_count = len(state["messages"])
# Only enable sensitive tools after authentication
if not is_authenticated:
tools = [t for t in request.tools if t.name.startswith("public_")]
request = request.override(tools=tools)
elif message_count < 5:
# Limit tools early in conversation
tools = [t for t in request.tools if t.name != "advanced_search"]
request = request.override(tools=tools)
return handler(request)
agent = create_agent(
model="gpt-5.4",
tools=[public_search, private_search, advanced_search],
middleware=[state_based_tools]
)
命令 / API 速查
@tool: 创建基本工具的装饰器ToolRuntime: 提供对运行时信息的访问return_direct: 工具参数,使输出直接作为最终响应wrap_tool_call: 中间件装饰器,用于包装工具调用wrap_model_call: 中间件装饰器,用于包装模型调用args_schema: 指定工具输入模式的参数Command: 用于更新 agent 状态的类型
与 LangGraph / RAG 手册的联系
- LangGraph 提供了更底层的编排能力,而 LangChain 的工具系统是构建在 LangGraph 之上的高层抽象
- RAG 系统通常使用 retriever 工具来获取外部知识,工具系统可以与 RAG 集成以增强 agent 的能力
- 工具系统与 LangGraph 的状态管理紧密集成,允许工具访问和更新对话状态
初学者易错点
- 忽略类型提示: 类型提示是必需的,因为它们定义了工具的输入模式
- 使用保留参数名: 避免使用
config和runtime作为工具参数名,这些是保留的 - 过度使用 return_direct: 仅在工具输出是完整、用户就绪的答案时使用,否则模型无法进一步处理结果
- 忘记错误处理: 工具可能会失败,适当的错误处理对于健壮的 agent 至关重要
- 工具名称格式: 偏好使用
snake_case而不是包含空格或特殊字符的名称
相关词条
- create_agent: 创建 agent 的方法
- middleware: 处理工具调用和模型调用的中间件
- models: LangChain 中的模型系统
- RAG: 检索增强生成系统
- retriever: 用于检索外部知识的组件