LangGraph Tool Calling:让AI拥有”超能力”

本文是 LangGraph 零基础入门系列的第4篇,我们将深入探讨 LangGraph 中最强大的功能之一——Tool Calling(工具调用)。通过工具调用,你的 AI Agent 将突破语言模型的知识边界,真正拥有与现实世界交互的能力。

引言:为什么需要工具?

想象你正在开发一个智能助手,用户问它:”北京今天的天气怎么样?”

如果你使用的是纯语言模型,它会这样回答:”作为AI,我无法获取实时天气信息,我的知识截止到训练数据的时间点…”

多么令人沮丧的回答!

这就是工具(Tool)存在的意义。工具是 AI Agent 与外部世界交互的桥梁,它们让语言模型能够:

  • 🌤️ 获取实时信息 — 天气、股价、新闻
  • 🔍 搜索知识库 — 查询数据库、检索文档
  • 🧮 执行精确计算 — 数学运算、代码执行
  • 📝 操作外部系统 — 发送邮件、创建日历事件、调用API

LangGraph 的 Tool Calling 机制,本质上是一种让 LLM 自主决策何时、如何使用工具的智能编排系统。它不是简单的函数调用,而是一个完整的决策-执行-反馈循环。

工具调用的核心流程

1
用户输入 → LLM 分析 → 决定调用工具 → 执行工具 → 返回结果 → LLM 整合 → 最终回答

这个流程的美妙之处在于:LLM 自己决定是否需要工具、需要哪个工具、传入什么参数。你不需要写一堆 if-else 来判断用户意图,LLM 会帮你做这件事。

在 LangGraph 中,工具是构建复杂 Agent 的基石。理解 Tool Calling,你就掌握了让 AI 从”聊天机器人”进化为”智能代理”的关键钥匙。

@tool 装饰器详解

在 LangGraph(以及底层的 LangChain)中,定义工具最简单的方式是使用 @tool 装饰器。它就像是给普通 Python 函数戴上一顶”工具帽”,让 LLM 能够发现和使用它。

基础用法

1
2
3
4
5
6
from langchain_core.tools import tool

@tool
def add(a: int, b: int) -> int:
"""Add two numbers together."""
return a + b

就这么简单!三行代码,你就创建了一个 LLM 可以调用的工具。

但这三行代码背后,隐藏着 LangGraph 的智能设计:

1. 类型注解是关键

1
2
3
4
@tool
def multiply(a: int, b: int) -> int:
"""Multiply two integers and return the result."""
return a * b

注意到 a: intb: int 了吗?这些类型注解不是给 Python 看的,而是给 LLM 看的。当你把这个工具绑定到模型时,LangGraph 会自动生成工具的模式(schema):

1
2
3
4
5
6
7
8
9
10
11
12
{
"name": "multiply",
"description": "Multiply two integers and return the result.",
"parameters": {
"type": "object",
"properties": {
"a": {"type": "integer", "description": "First integer"},
"b": {"type": "integer", "description": "Second integer"}
},
"required": ["a", "b"]
}
}

LLM 正是通过这套 schema 来理解:这个工具能做什么?需要什么参数?

2. 文档字符串就是说明书

函数的 docstring("""...""")会被提取为工具的 description。这个描述至关重要——它是 LLM 决定是否调用这个工具的主要依据。

糟糕的描述:

1
2
3
4
@tool
def func(x):
"""Does something."""
pass

优秀的描述:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
@tool
def calculate_compound_interest(principal: float, rate: float, years: int) -> float:
"""
Calculate compound interest for an investment.

Args:
principal: Initial investment amount (must be positive)
rate: Annual interest rate as decimal (e.g., 0.05 for 5%)
years: Number of years to compound

Returns:
Final amount after compound interest
"""
return principal * (1 + rate) ** years

3. 高级配置

@tool 装饰器支持更多配置参数:

1
2
3
4
5
6
7
8
9
10
11
from langchain_core.tools import tool

@tool(
name="weather_lookup", # 自定义工具名
description="Get current weather for a city", # 自定义描述
return_direct=False # 是否直接返回结果(不经过LLM处理)
)
def get_weather(city: str) -> str:
"""Fetch weather data from API."""
# ... 实现代码
return f"Weather in {city}: Sunny, 25°C"

4. 异步工具支持

现代应用大多是异步的,LangGraph 当然也支持:

1
2
3
4
5
6
7
8
9
import aiohttp
from langchain_core.tools import tool

@tool
async def fetch_url(url: str) -> str:
"""Fetch content from a URL asynchronously."""
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return await response.text()

使用异步工具时,记得在调用处使用 await

工具的描述艺术:如何让 LLM 正确选择工具

这是 Tool Calling 中最容易被忽视,但也最关键的部分。工具描述的质量直接决定了 LLM 的选择准确性

反直觉的真相

你可能会想:”我写清楚工具是干什么的就行了吧?”

不够。你需要思考的是:LLM 在什么场景下应该选这个工具?

策略一:描述使用场景,而非功能

❌ 差的描述:

1
2
3
4
@tool
def search_wikipedia(query: str) -> str:
"""Search Wikipedia."""
pass

✅ 好的描述:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
@tool
def search_wikipedia(query: str) -> str:
"""
Search Wikipedia for general knowledge questions about
history, science, people, places, or concepts.

Use this when:
- The user asks about factual information
- You need to verify historical dates or events
- The question is about famous people or places

Do NOT use for:
- Real-time information (weather, news, stock prices)
- Personal or opinion-based questions
"""
pass

策略二:明确参数含义和格式

LLM 需要知道每个参数应该填什么:

1
2
3
4
5
6
7
8
9
10
11
@tool
def create_reminder(text: str, datetime_str: str, timezone: str = "UTC") -> str:
"""
Create a reminder for the user.

Args:
text: What to remind about (e.g., "Call mom")
datetime_str: When to remind, in ISO 8601 format (YYYY-MM-DDTHH:MM:SS)
timezone: Timezone for the reminder (default UTC, use "Asia/Shanghai" for China)
"""
pass

策略三:提供示例

对于复杂工具,示例能帮助 LLM 理解:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
@tool
def parse_date(date_string: str) -> str:
"""
Parse various date formats and return standardized ISO date.

Supports formats like:
- "tomorrow", "next Monday", "in 3 days" (relative)
- "2024-01-15", "Jan 15, 2024", "15/01/2024" (absolute)
- "2024-01-15 14:30", "3pm tomorrow" (with time)

Example:
Input: "next Friday at 3pm"
Output: "2024-01-19T15:00:00"
"""
pass

策略四:命名要有区分度

如果你有多个相似的工具,命名要让 LLM 一眼看出区别:

❌ 混淆的命名:

1
2
3
4
5
@tool
def search(query: str): ... # 搜索什么?

@tool
def search2(query: str): ... # ???

✅ 清晰的命名:

1
2
3
4
5
6
7
8
@tool
def search_company_database(query: str): ...

@tool
def search_web(query: str): ...

@tool
def search_internal_docs(query: str): ...

策略五:使用 return_direct 控制输出流

有时你希望工具结果直接返回给用户,不经过 LLM 再加工:

1
2
3
4
5
@tool(return_direct=True)
def get_current_time() -> str:
"""Get the current system time."""
from datetime import datetime
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

设置 return_direct=True 后,工具返回值会直接作为最终答案,节省一次 LLM 调用。

工具错误处理:容错与重试

现实世界充满不确定性。API 会超时,参数会无效,网络会抖动。一个健壮的 Agent 必须能优雅地处理这些情况。

基础错误处理

在工具函数内部处理异常:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from langchain_core.tools import tool
import requests

@tool
def get_weather(city: str) -> str:
"""
Get current weather for a city.

Args:
city: City name (e.g., "Beijing", "New York")
"""
try:
# 模拟 API 调用
response = requests.get(
f"https://api.weather.com/v1/current?city={city}",
timeout=5
)
response.raise_for_status()
data = response.json()
return f"Weather in {city}: {data['condition']}, {data['temp']}°C"
except requests.Timeout:
return f"Error: Weather service timed out. Please try again."
except requests.HTTPError as e:
return f"Error: Weather service returned {e.response.status_code}. City may not be found."
except Exception as e:
return f"Error fetching weather: {str(e)}"

参数验证

在工具内部验证参数,而不是依赖外部:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
@tool
def calculate_bmi(height_cm: float, weight_kg: float) -> str:
"""
Calculate BMI (Body Mass Index).

Args:
height_cm: Height in centimeters (must be > 0 and < 300)
weight_kg: Weight in kilograms (must be > 0 and < 500)
"""
# 参数验证
if height_cm <= 0 or height_cm >= 300:
return "Error: Height must be between 0 and 300 cm"
if weight_kg <= 0 or weight_kg >= 500:
return "Error: Weight must be between 0 and 500 kg"

height_m = height_cm / 100
bmi = weight_kg / (height_m ** 2)

# 分类
if bmi < 18.5:
category = "Underweight"
elif bmi < 25:
category = "Normal"
elif bmi < 30:
category = "Overweight"
else:
category = "Obese"

return f"BMI: {bmi:.1f} ({category})"

在 LangGraph 中实现重试机制

LangGraph 的图结构让重试逻辑变得优雅。你可以创建一个专门的错误处理节点:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
import operator

class AgentState(TypedDict):
messages: Annotated[list, operator.add]
retry_count: int # 追踪重试次数

def call_tool_with_retry(state: AgentState):
"""带重试逻辑的工具调用节点"""
messages = state["messages"]
retry_count = state.get("retry_count", 0)

# 调用工具...(简化示例)
result = tool_node.invoke(messages)

# 检查是否是错误
if "Error:" in str(result):
if retry_count < 3:
return {
"messages": [f"Attempt {retry_count + 1} failed, retrying..."],
"retry_count": retry_count + 1
}
else:
return {
"messages": ["Max retries reached. Unable to complete request."],
"retry_count": 0
}

return {"messages": [result], "retry_count": 0}

优雅降级

当工具失败时,提供备选方案:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
@tool
def get_detailed_weather(city: str) -> str:
"""Get detailed weather including hourly forecast."""
try:
# 尝试获取详细数据
detailed_data = fetch_from_primary_api(city)
return format_detailed_weather(detailed_data)
except Exception:
try:
# 降级到备用 API
basic_data = fetch_from_backup_api(city)
return format_basic_weather(basic_data) + "\n(Using backup data source)"
except Exception:
# 最终降级:返回通用建议
return f"Unable to fetch weather for {city}. Please check a weather website directly."

实战:多功能助手

现在,让我们把所学知识整合起来,构建一个真正的多功能助手。这个助手将具备:

  1. 天气查询 — 获取实时天气
  2. 网络搜索 — 搜索最新信息
  3. 计算器 — 执行数学运算

完整代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
"""
LangGraph Multi-Tool Agent Demo
多功能助手:天气 + 搜索 + 计算
"""

import os
import json
import requests
from typing import Annotated, TypedDict, Sequence
from langchain_core.tools import tool
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
import operator


# ============ 1. 定义工具 ============

@tool
def get_weather(city: str) -> str:
"""
Get current weather information for a specified city.

Use this tool when the user asks about weather conditions, temperature,
or forecasts for a specific location.

Args:
city: Name of the city (e.g., "Beijing", "Shanghai", "New York")

Returns:
Weather description including temperature and conditions
"""
# 模拟天气 API(实际项目中替换为真实 API)
try:
# 这里使用模拟数据演示
# 真实场景:调用 OpenWeatherMap、和风天气等 API
weather_db = {
"beijing": {"temp": 22, "condition": "Sunny", "humidity": 45},
"shanghai": {"temp": 25, "condition": "Cloudy", "humidity": 60},
"guangzhou": {"temp": 28, "condition": "Light Rain", "humidity": 75},
"shenzhen": {"temp": 29, "condition": "Clear", "humidity": 70},
"new york": {"temp": 15, "condition": "Partly Cloudy", "humidity": 55},
"london": {"temp": 12, "condition": "Rainy", "humidity": 80},
"tokyo": {"temp": 20, "condition": "Sunny", "humidity": 50},
}

city_key = city.lower().strip()

if city_key in weather_db:
data = weather_db[city_key]
return (
f"🌤️ Weather in {city}:\n"
f" Temperature: {data['temp']}°C\n"
f" Condition: {data['condition']}\n"
f" Humidity: {data['humidity']}%"
)
else:
# 模拟 API 调用失败
return f"⚠️ Weather data for '{city}' not available. Try: Beijing, Shanghai, New York, London, Tokyo"

except Exception as e:
return f"❌ Error fetching weather: {str(e)}"


@tool
def web_search(query: str, num_results: int = 3) -> str:
"""
Search the web for current information.

Use this tool when:
- The user asks about recent events, news, or current information
- You need to verify facts that may have changed
- The question requires information beyond your training data

Args:
query: Search query string
num_results: Number of results to return (1-5, default 3)

Returns:
Search results with titles and snippets
"""
try:
# 参数验证
num_results = max(1, min(5, num_results))

# 模拟搜索结果(实际项目中使用 DuckDuckGo、SerpAPI 等)
mock_results = {
"langgraph": [
{"title": "LangGraph Documentation", "snippet": "Build stateful AI apps with LangGraph..."},
{"title": "LangGraph Tutorial 2024", "snippet": "Learn how to build multi-agent systems..."},
],
"ai news": [
{"title": "Latest AI Breakthroughs", "snippet": "New models show remarkable improvements..."},
{"title": "AI Industry Updates", "snippet": "Major companies announce new AI initiatives..."},
],
}

query_key = query.lower().strip()
results = mock_results.get(query_key, [
{"title": f"Search Result for: {query}", "snippet": "Simulated search result content..."},
{"title": "Related Information", "snippet": "Additional context about the query..."},
])

output = f"🔍 Search results for '{query}':\n\n"
for i, result in enumerate(results[:num_results], 1):
output += f"{i}. {result['title']}\n"
output += f" {result['snippet']}\n\n"

return output.strip()

except Exception as e:
return f"❌ Search error: {str(e)}"


@tool
def calculator(expression: str) -> str:
"""
Evaluate a mathematical expression safely.

Use this tool for:
- Any mathematical calculations (arithmetic, powers, etc.)
- Converting units or computing percentages
- Complex formulas that need precision

Args:
expression: Mathematical expression as a string
(e.g., "15 * 24 + 100", "2**10", "(100 - 20) / 4")

Returns:
Calculated result
"""
try:
# 安全计算:只允许数学运算
allowed_chars = set('0123456789+-*/.() **% ')
if not all(c in allowed_chars for c in expression):
return "❌ Error: Invalid characters in expression. Only numbers and operators allowed."

# 使用 eval 的安全限制版本
result = eval(expression, {"__builtins__": {}}, {})

return f"🧮 {expression} = {result}"

except ZeroDivisionError:
return "❌ Error: Cannot divide by zero"
except Exception as e:
return f"❌ Calculation error: {str(e)}"


# 工具列表
tools = [get_weather, web_search, calculator]


# ============ 2. 定义状态 ============

class AgentState(TypedDict):
"""Agent state definition"""
messages: Annotated[Sequence[BaseMessage], operator.add]


# ============ 3. 创建图节点 ============

# 初始化 LLM 并绑定工具
llm = ChatOpenAI(
model="gpt-4o-mini", # 或其他支持的模型
temperature=0,
api_key=os.getenv("OPENAI_API_KEY")
)
llm_with_tools = llm.bind_tools(tools)


def agent_node(state: AgentState) -> AgentState:
"""
Agent decision node.
The LLM decides whether to call tools or respond directly.
"""
messages = state["messages"]
response = llm_with_tools.invoke(messages)
return {"messages": [response]}


# 工具执行节点
tool_node = ToolNode(tools)


def should_continue(state: AgentState) -> str:
"""
Determine if we should continue to tools or end.
"""
messages = state["messages"]
last_message = messages[-1]

# 如果最后一条消息有 tool_calls,说明需要调用工具
if hasattr(last_message, 'tool_calls') and last_message.tool_calls:
return "continue"

# 否则结束
return "end"


# ============ 4. 构建图 ============

# 创建工作流
workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("agent", agent_node)
workflow.add_node("tools", tool_node)

# 设置入口
workflow.set_entry_point("agent")

# 添加条件边
workflow.add_conditional_edges(
"agent",
should_continue,
{
"continue": "tools",
"end": END
}
)

# 工具执行后返回给 Agent
workflow.add_edge("tools", "agent")

# 编译图
app = workflow.compile()


# ============ 5. 运行示例 ============

def run_agent(user_input: str):
"""Run the agent with user input"""
print(f"\n{'='*50}")
print(f"User: {user_input}")
print(f"{'='*50}\n")

# 初始化状态
initial_state = {
"messages": [HumanMessage(content=user_input)]
}

# 运行图
result = app.invoke(initial_state)

# 打印结果
for msg in result["messages"]:
if isinstance(msg, AIMessage):
if hasattr(msg, 'tool_calls') and msg.tool_calls:
print(f"🤖 Agent decided to call tools:")
for tc in msg.tool_calls:
print(f" - {tc['name']}({tc['args']})")
else:
print(f"🤖 Agent: {msg.content}")
elif isinstance(msg, ToolMessage):
print(f"🔧 Tool result:\n{msg.content}")

print()
return result


if __name__ == "__main__":
# 测试场景 1: 天气查询
run_agent("北京今天天气怎么样?")

# 测试场景 2: 搜索 + 计算
run_agent("如果我有10000元,年化收益5%,复利计算10年后会变成多少钱?")

# 测试场景 3: 数学计算
run_agent("计算 (150 + 230) * 1.5 - 80")

# 测试场景 4: 需要搜索的信息
run_agent("帮我搜索一下 LangGraph 的最新教程")

代码解析

1. 工具定义层

三个工具分别对应三种能力:

  • get_weather: 模拟天气查询,实际项目中可接入真实天气 API
  • web_search: 模拟网络搜索,实际项目中可使用 DuckDuckGo、SerpAPI 等
  • calculator: 安全计算器,使用受限的 eval 执行数学表达式

2. 状态定义

1
2
class AgentState(TypedDict):
messages: Annotated[Sequence[BaseMessage], operator.add]

状态只包含消息列表,operator.add 表示新消息会追加到列表。

3. 节点逻辑

  • agent_node: LLM 决策节点,根据输入决定是调用工具还是直接回答
  • tool_node: 工具执行节点,LangGraph 预置的 ToolNode 会自动解析 tool_calls 并执行对应工具

4. 条件路由

1
2
3
4
5
def should_continue(state: AgentState) -> str:
last_message = state["messages"][-1]
if hasattr(last_message, 'tool_calls') and last_message.tool_calls:
return "continue" # 需要调用工具
return "end" # 直接结束

这是 Tool Calling 的核心逻辑:检查 LLM 的响应是否包含 tool_calls,如果有就路由到工具节点。

5. 执行流程

  1. 用户输入 → Agent 节点
  2. LLM 分析需求,决定调用哪个工具
  3. 生成 tool_calls → 路由到工具节点
  4. 工具执行,返回 ToolMessage
  5. 结果回到 Agent 节点
  6. LLM 整合工具结果,生成最终回答

实际运行效果

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
==================================================
User: 北京今天天气怎么样?
==================================================

🤖 Agent decided to call tools:
- get_weather({'city': '北京'})
🔧 Tool result:
🌤️ Weather in 北京:
Temperature: 22°C
Condition: Sunny
Humidity: 45%
🤖 Agent: 北京今天天气晴朗,气温22°C,湿度45%,是个出行的好天气!

==================================================
User: 如果我有10000元,年化收益5%,复利计算10年后会变成多少钱?
==================================================

🤖 Agent decided to call tools:
- calculator({'expression': '10000 * (1 + 0.05) ** 10'})
🔧 Tool result:
🧮 10000 * (1 + 0.05) ** 10 = 16288.946267774414
🤖 Agent: 根据复利计算,10年后你的10000元将增长到约 **16,289元**,总收益约 **6,289元**(收益率62.9%)。

总结与下篇预告

在本篇中,我们深入探讨了 LangGraph 的 Tool Calling 机制:

核心要点回顾

  1. @tool 装饰器 —— 最简单的工具定义方式,类型注解和文档字符串是关键
  2. 描述的艺术 —— 好的工具描述要明确使用场景、参数含义,提供示例
  3. 错误处理 —— 工具内部异常处理、参数验证、优雅降级策略
  4. 实战应用 —— 构建了一个支持天气、搜索、计算的多功能助手

Tool Calling 的设计哲学

LangGraph 的 Tool Calling 不只是”让 LLM 调用函数”,它体现了一个更深层的理念:将决策权交给模型

传统编程是命令式的:

1
2
3
4
if "weather" in user_input:
result = get_weather(...)
elif "search" in user_input:
result = web_search(...)

LangGraph 是声明式的:

1
2
# LLM 自己决定调用什么
response = llm_with_tools.invoke(messages)

这种范式转变让 Agent 具备了真正的”智能”——它能理解意图、规划步骤、选择工具、整合结果。你不需要为每个场景写 if-else,只需要定义好工具,LLM 会自己 figuring out。

下篇预告

在下一篇文章中,我们将探讨 LangGraph 的 Memory(记忆)系统

  • 为什么 Agent 需要记忆?
  • Short-term vs Long-term Memory
  • 使用 Redis/Postgres 持久化对话历史
  • 实现跨会话的用户画像记忆

记忆是让 Agent 从”聊天机器人”进化为”个人助理”的关键。敬请期待!


参考资源:

本文代码可在 GitHub 获取:github.com/yourusername/langgraph-demos