LangChain - 工具(Tools)
·
一、工具是什么?
工具就是可调用函数,有明确的输入和输出,传递给聊天模型。模型根据对话上下文决定何时调用工具、传什么参数。
简单说:模型是大脑,工具是手脚。模型思考后决定要不要用工具、用哪个工具、怎么调工具。
二、创建工具
2.1 最简单的方式:用 @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}'"
关键点:
- 类型提示是必需的(如 query: str),它们定义了工具的输入模式
- 文档字符串很重要,默认会成为工具描述,帮助模型理解何时使用
- 工具名建议用 snake_case(如 web_search),不要用空格或特殊字符
2.2 自定义工具属性
自定义名称
@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 定义复杂输入
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
2.3 保留参数名
这两个名字是保留的,不能用作工具参数:
- config —— 保留给 RunnableConfig
- runtime —— 保留给 ToolRuntime(访问状态、上下文、存储)
三、访问上下文
工具可以访问运行时信息,功能更强大。通过 ToolRuntime 参数访问。
3.1 短期记忆(状态)
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"
关键: runtime 参数会自动注入,对 LLM 隐藏——不会出现在工具模式里。
还可以用 Command 更新智能体的state(状态):
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,
)
],
}
)
注意: 如果工具更新状态变量,考虑为这些字段定义 reducer(解决并发工具调用时的冲突)。
3.2 上下文(不可变配置)
context:上下文是在调用时传递的不可变配置数据(用户 ID、会话信息等)。
from dataclasses import dataclass
from langchain_openai import ChatOpenAI
from langchain.agents import create_agent
from langchain.tools import tool, ToolRuntime
@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']}\nType: {user['account_type']}\nBalance: ${user['balance']}"
return "User not found"
创建代理时传入上下文模式
agent = create_agent(
ChatOpenAI(model="gpt-5.4"),
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")
)
3.3 长期记忆(存储)
存储(BaseStore)提供跨对话持久存在的存储。与状态不同,保存到存储的数据在未来会话仍可用。
from typing import Any
from langgraph.store.memory import InMemoryStore
from langchain.agents import create_agent
from langchain.tools import tool, ToolRuntime
@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."
store = InMemoryStore() # 生产环境用 PostgresStore
agent = create_agent(
ChatOpenAI(model="gpt-5.4"),
tools=[get_user_info, save_user_info],
store=store
)
# 第一次会话:保存用户信息
agent.invoke({
"messages": [{"role": "user", "content": "Save: user abc123, name: Foo, age: 25"}]
})
# 第二次会话:获取用户信息
agent.invoke({
"messages": [{"role": "user", "content": "Get user info for abc123"}]
})
# 能拿到之前保存的信息!
三种"记忆"的对比(核心)
| 概念 | 比喻 | 生命周期 | 能否修改 | 访问方式 |
|---|---|---|---|---|
| state(状态) | 本次通话的草稿纸 | 本次对话,结束即丢 | 可读可写 | runtime.state |
| context(上下文) | 你的身份工牌 | 本次调用期间不变 | 只读不可变 | runtime.context |
| store(存储) | 公司的档案柜 | 跨对话永久保存 | 可读可写 | runtime.store |
一句话区分:
- state:本次通话中的临时过程数据,“通话结束就忘”
- context:本次调用定死的配置(如 user_id),“通话期间不变”
- store:跨对话永久保存的数据,“永远都在”
四、工具节点(ToolNode)
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))
# 创建工具节点
tool_node = ToolNode([search, calculator])
# 在图中使用
builder = StateGraph(MessagesState)
builder.add_node("tools", tool_node)
# ... 添加其他节点和边
4.1 错误处理
配置工具错误的处理方式:
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))
4.2 路由(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()
五、工具返回值
5.1 返回字符串
提供纯文本供模型读取:
@tool
def get_weather(city: str) -> str:
"""Get weather for a city."""
return f"It is currently sunny in {city}."
行为:
- 返回值被转换为 ToolMessage
- 模型看到文本并决定下一步
- 智能体状态字段不会改变
5.2 返回对象
返回结构化数据(如 dict):
@tool
def get_weather_data(city: str) -> dict:
"""Get structured weather data for a city."""
return {
"city": city,
"temperature_c": 22,
"conditions": "sunny",
}
行为:
- 对象被序列化并作为工具输出返回
- 模型可以读取特定字段并推理
5.3 返回 Command
当工具需要更新图状态时:
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,
)
],
}
)
行为:
- Command 用 update 更新状态
- 更新后的状态可供后续步骤使用
- 如果模型需要看到工具成功,在更新中包含 ToolMessage
总结
- 工具 = 可调用函数,用 @tool 装饰器创建
- 三种访问上下文的方式:
- runtime.state —— 短期记忆(当前对话)
- runtime.context —— 不可变配置(用户 ID 等)
- runtime.store —— 长期记忆(跨对话持久化) - 三种返回值:
- 字符串 —— 纯文本结果
- 对象 —— 结构化数据
- Command —— 更新状态 - ToolNode —— 在 LangGraph 工作流中执行工具的预构建节点
工具是代理的"手脚",让模型不仅能思考,还能行动。通过 ToolRuntime,工具可以访问对话历史、用户信息、长期记忆等,功能更强大。
更多推荐


所有评论(0)