工具使用 #
概述 #
工具（Tools）是 Agent 与外部世界交互的桥梁。通过工具，Agent 可以搜索网络、查询数据库、调用 API、执行代码等。LangGraph 提供了强大的工具集成能力。
text
┌─────────────────────────────────────────────────────────────┐
│ Agent 工具交互 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────┐ │
│ │ 用户 │ │
│ └────┬────┘ │
│ │ │
│ ▼ │
│ ┌─────────┐ 决定使用工具 ┌─────────┐ │
│ │ Agent │ ──────────────────> │ Tool │ │
│ │ LLM │ │ 搜索/DB │ │
│ └────┬────┘ └────┬────┘ │
│ │ │ │
│ │<──────────────────────────────┘ │
│ │ 获取结果 │
│ │ │
│ ▼ │
│ ┌─────────┐ │
│ │ 回答 │ │
│ └─────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘
定义工具 #
使用 @tool 装饰器 #
python
from langchain_core.tools import tool
@tool
def add(a: int, b: int) -> int:
"""Add two numbers together."""
return a + b
@tool
def multiply(a: int, b: int) -> int:
"""Multiply two numbers together."""
return a * b
tools = [add, multiply]
使用 Pydantic 定义参数 #
python
from langchain_core.tools import tool
from pydantic import BaseModel, Field
class SearchInput(BaseModel):
query: str = Field(description="Search query")
max_results: int = Field(default=5, description="Maximum results")
@tool(args_schema=SearchInput)
def search(query: str, max_results: int = 5) -> list[str]:
"""Search for information."""
return [f"Result {i} for {query}" for i in range(max_results)]
从函数创建工具 #
python
from langchain_core.tools import StructuredTool
def get_weather(location: str) -> str:
"""Get weather for a location."""
return f"Weather in {location}: Sunny, 25°C"
weather_tool = StructuredTool.from_function(get_weather)
异步工具 #
python
@tool
async def async_search(query: str) -> str:
"""Async search tool."""
result = await search_api(query)
return result
工具结构 #
text
┌─────────────────────────────────────────────────────────────┐
│ 工具结构 │
├─────────────────────────────────────────────────────────────┤
│ │
│ Tool 属性： │
│ ├── name: 工具名称 │
│ ├── description: 工具描述 │
│ ├── args_schema: 参数 Schema │
│ └── return_direct: 是否直接返回结果 │
│ │
│ 示例： │
│ { │
│ "name": "search", │
│ "description": "Search for information", │
│ "parameters": { │
│ "type": "object", │
│ "properties": { │
│ "query": {"type": "string", "description": "..."} │
│ }, │
│ "required": ["query"] │
│ } │
│ } │
│ │
└─────────────────────────────────────────────────────────────┘
ToolNode #
LangGraph 提供了预构建的 ToolNode 来执行工具。
基本使用 #
python
from langgraph.prebuilt import ToolNode
tools = [add, multiply, search]
tool_node = ToolNode(tools)
graph.add_node("tools", tool_node)
ToolNode 工作流程 #
text
┌─────────────────────────────────────────────────────────────┐
│ ToolNode 工作流程 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 输入： │
│ state["messages"][-1].tool_calls │
│ [ │
│ {"name": "search", "args": {"query": "weather"}}, │
│ {"name": "add", "args": {"a": 1, "b": 2}} │
│ ] │
│ │
│ 处理： │
│ 1. 解析 tool_calls │
│ 2. 查找对应的工具 │
│ 3. 执行工具调用 │
│ 4. 生成 ToolMessage │
│ │
│ 输出： │
│ {"messages": [ToolMessage(content="...", tool_call_id=..)]}│
│ │
└─────────────────────────────────────────────────────────────┘
自定义 ToolNode #
python
from langgraph.prebuilt import ToolNode
class CustomToolNode(ToolNode):
def _invoke(self, tool_call, tool):
try:
result = super()._invoke(tool_call, tool)
return result
except Exception as e:
return ToolMessage(
content=f"Error: {str(e)}",
tool_call_id=tool_call["id"]
)
将工具绑定到 LLM #
使用 bind_tools #
python
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini")
llm_with_tools = llm.bind_tools(tools)
response = llm_with_tools.invoke("What is 2 + 3?")
print(response.tool_calls)
工具调用格式 #
python
response = llm_with_tools.invoke("Search for Python tutorials")
print(response.tool_calls)
# [
# {
# "name": "search",
# "args": {"query": "Python tutorials"},
# "id": "call_123"
# }
# ]
构建 Tool Agent #
完整示例 #
python
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
class State(TypedDict):
messages: Annotated[list, add_messages]
@tool
def add(a: int, b: int) -> int:
"""Add two numbers."""
return a + b
@tool
def multiply(a: int, b: int) -> int:
"""Multiply two numbers."""
return a * b
tools = [add, multiply]
llm = ChatOpenAI(model="gpt-4o-mini")
llm_with_tools = llm.bind_tools(tools)
def agent_node(state: State):
return {"messages": [llm_with_tools.invoke(state["messages"])]}
def should_continue(state: State):
if state["messages"][-1].tool_calls:
return "tools"
return END
tool_node = ToolNode(tools)
graph = StateGraph(State)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue)
graph.add_edge("tools", "agent")
app = graph.compile()
使用预构建 Agent #
python
from langgraph.prebuilt import create_react_agent
agent = create_react_agent(llm, tools)
result = agent.invoke({"messages": [("user", "What is 2 + 3 * 4?")]})
工具类型 #
1. 计算工具 #
python
@tool
def calculate(expression: str) -> float:
"""Evaluate a mathematical expression."""
return eval(expression)
@tool
def convert_units(value: float, from_unit: str, to_unit: str) -> float:
"""Convert between units."""
conversions = {
("km", "miles"): 0.621371,
("miles", "km"): 1.60934,
}
factor = conversions.get((from_unit, to_unit), 1)
return value * factor
2. 搜索工具 #
python
@tool
def web_search(query: str) -> str:
"""Search the web for information."""
import requests
response = requests.get(f"https://api.search.com?q={query}")
return response.text
@tool
def wikipedia_search(query: str) -> str:
"""Search Wikipedia."""
import wikipedia
return wikipedia.summary(query)
3. 数据库工具 #
python
@tool
def query_database(sql: str) -> list:
"""Execute SQL query."""
import sqlite3
conn = sqlite3.connect("database.db")
cursor = conn.execute(sql)
return cursor.fetchall()
4. API 工具 #
python
@tool
def call_api(endpoint: str, method: str = "GET", data: dict = None) -> dict:
"""Call an external API."""
import requests
if method == "GET":
return requests.get(endpoint).json()
elif method == "POST":
return requests.post(endpoint, json=data).json()
5. 文件工具 #
python
@tool
def read_file(path: str) -> str:
"""Read file content."""
with open(path, "r") as f:
return f.read()
@tool
def write_file(path: str, content: str) -> str:
"""Write content to file."""
with open(path, "w") as f:
f.write(content)
return f"Written to {path}"
高级工具模式 #
工具返回 Command #
python
from langgraph.types import Command
@tool
def transfer_to_agent(agent_name: str) -> Command:
"""Transfer to another agent."""
return Command(goto=agent_name)
工具访问状态 #
python
from langchain_core.tools import InjectedToolArg
from typing import Annotated
@tool
def save_to_memory(
content: str,
state: Annotated[dict, InjectedToolArg]
) -> str:
"""Save content to memory."""
state["memory"].append(content)
return "Saved"
工具错误处理 #
python
@tool
def risky_operation(data: str) -> str:
"""Perform a risky operation."""
try:
result = process(data)
return result
except ValueError as e:
return f"Error: {e}"
except Exception as e:
return f"Unexpected error: {e}"
工具重试 #
python
from tenacity import retry, stop_after_attempt
@tool
@retry(stop=stop_after_attempt(3))
def fetch_with_retry(url: str) -> str:
"""Fetch URL with retry."""
import requests
return requests.get(url).text
工具选择控制 #
强制使用工具 #
python
llm_with_tools = llm.bind_tools(
tools,
tool_choice="search" # 强制使用 search 工具
)
自动选择 #
python
llm_with_tools = llm.bind_tools(
tools,
tool_choice="auto" # 让 LLM 决定
)
不使用工具 #
python
llm_with_tools = llm.bind_tools(
tools,
tool_choice="none" # 不使用工具
)
工具调用流程 #
text
┌─────────────────────────────────────────────────────────────┐
│ 工具调用流程 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 1. 用户输入 │
│ "What is the weather in Beijing?" │
│ │
│ 2. Agent 分析 │
│ LLM 决定需要调用 weather 工具 │
│ │
│ 3. 生成 tool_calls │
│ [{"name": "weather", "args": {"city": "Beijing"}}] │
│ │
│ 4. ToolNode 执行 │
│ 调用 weather("Beijing") │
│ │
│ 5. 返回结果 │
│ ToolMessage(content="Sunny, 25°C") │
│ │
│ 6. Agent 处理结果 │
│ LLM 生成最终回答 │
│ │
│ 7. 输出 │
│ "The weather in Beijing is sunny, 25°C." │
│ │
└─────────────────────────────────────────────────────────────┘
多工具并行调用 #
python
response = llm_with_tools.invoke(
"What is 2+3 and what is 4*5?"
)
print(response.tool_calls)
# [
# {"name": "add", "args": {"a": 2, "b": 3}, "id": "call_1"},
# {"name": "multiply", "args": {"a": 4, "b": 5}, "id": "call_2"}
# ]
ToolNode 会自动并行执行多个工具调用。
工具最佳实践 #
1. 清晰的描述 #
python
@tool
def search(query: str) -> str:
"""
Search for information on the web.
Args:
query: The search query string.
Returns:
A string containing search results.
"""
return search_api(query)
2. 参数验证 #
python
from pydantic import BaseModel, Field, validator
class EmailInput(BaseModel):
to: str = Field(description="Recipient email")
subject: str = Field(description="Email subject")
body: str = Field(description="Email body")
@validator("to")
def validate_email(cls, v):
if "@" not in v:
raise ValueError("Invalid email")
return v
@tool(args_schema=EmailInput)
def send_email(to: str, subject: str, body: str) -> str:
"""Send an email."""
return f"Email sent to {to}"
3. 安全考虑 #
python
import os
ALLOWED_PATHS = ["/data", "/tmp"]
@tool
def safe_read_file(path: str) -> str:
"""Read file safely."""
abs_path = os.path.abspath(path)
if not any(abs_path.startswith(p) for p in ALLOWED_PATHS):
return "Error: Access denied"
with open(abs_path, "r") as f:
return f.read()
4. 限制工具权限 #
python
@tool
def limited_query(sql: str) -> list:
"""Execute read-only SQL query."""
forbidden = ["DROP", "DELETE", "UPDATE", "INSERT"]
if any(word in sql.upper() for word in forbidden):
return "Error: Only SELECT queries allowed"
return execute_query(sql)
调试工具调用 #
查看工具调用 #
python
for step in app.stream({"messages": [...]}):
if "tools" in step:
for msg in step["tools"]["messages"]:
print(f"Tool: {msg.name}")
print(f"Args: {msg.tool_call_kwargs}")
print(f"Result: {msg.content}")
使用 LangSmith #
python
import os
os.environ["LANGSMITH_TRACING"] = "true"
# 所有工具调用会自动追踪
下一步 #
现在你已经掌握了工具使用，接下来学习 记忆与持久化，了解如何让 Agent 记住历史！
最后更新：2026-03-30