作者：PySuper | 来源：zhengxingtao.com | 更新日期：2026-03-08
系列：Python/AI工程化 第43篇 | 上一篇：第42篇 LangGraph 快速上手 | 关联阅读：第23篇 Tool Calling 通用实践、第19篇 状态机设计、第39篇 人在回路
一、为什么 Agent 需要工具
纯 LLM 是个「嘴强王者」——能聊会写，但一动手就废：不能联网查资料，不能跑代码，不能操作数据库，连今天的天气都得瞎编。工具（Tool Calling）就是让 Agent 从「只会说」变成「能干活」的关键一跳。
LangGraph 中工具调用的完整生命周期：
plaintext
┌─────────────────────────────────────────────────────────────────────┐
│ LangGraph 工具调用完整生命周期 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ 用户输入 │
│ │ │
│ ▼ │
│ ┌──────────┐ ┌──────────────────┐ ┌──────────────────┐ │
│ │ Agent │───▶│ LLM 推理 │───▶│ 返回 AIMessage │ │
│ │ Node │ │ (带工具描述) │ │ 含 tool_calls? │ │
│ └──────────┘ └──────────────────┘ └────────┬─────────┘ │
│ ▲ │ │
│ │ ┌─────┴──────┐ │
│ │ │ 有tool_calls│ │
│ │ └─────┬──────┘ │
│ │ │ 是 │
│ │ ▼ │
│ │ ┌───────────────┐ │
│ │ │ ToolNode │ │
│ │ │ 解析并执行 │ │
│ │ │ tool_calls │ │
│ │ └───────┬───────┘ │
│ │ │ │
│ │ ▼ │
│ │ ┌───────────────┐ │
│ │ │ ToolMessage │ │
│ │ │ 工具结果回传 │ │
│ │ └───────┬───────┘ │
│ │ │ │
│ └──────────────────────────────────────────┘ │
│ (回到 Agent 继续推理) │
│ │
│ 无 tool_calls 时 → END（返回最终回答） │
│ │
└─────────────────────────────────────────────────────────────────────┘
关键点：LLM 不直接执行工具，它只决定"调什么、传什么参数"，真正的执行由 ToolNode 完成。 这个分离是 LangGraph 工具调用的核心设计。
二、@tool 装饰器：定义工具
2.1 基本用法
@tool 装饰器做三件事：
把 Python 函数包装成 LangChain 工具对象
从 类型注解 自动生成 JSON Schema（告诉 LLM 参数类型）
从 docstring 提取工具描述（LLM 靠这个决定调不调）
python
from langchain_core.tools import tool
@tool
def search_web(query: str) -> str:
"""搜索互联网获取实时信息。
当用户询问最新新闻、实时数据或需要联网查询的信息时使用此工具。
"""
# 实际实现中调用搜索 API
return f"搜索 '{query}' 的结果：..."
2.2 类型注解 = 自动 Schema
类型注解不是摆设，@tool 会解析它们生成 JSON Schema：
python
@tool
def calculate(expression: str, precision: int = 2) -> str:
"""计算数学表达式，支持基本四则运算。
Args:
expression: 数学表达式，如 "2+3*4"
precision: 结果保留小数位数，默认2位
"""
result = eval(expression) # 生产环境请用安全的方式
return f"{result:.{precision}f}"
生成的 Schema 大致如下（LLM 看到的）：
json
{
"name": "calculate",
"description": "计算数学表达式，支持基本四则运算。",
"parameters": {
"type": "object",
"properties": {
"expression": {"type": "string", "description": "数学表达式，如 \"2+3*4\""},
"precision": {"type": "integer", "description": "结果保留小数位数，默认2位", "default": 2}
},
"required": ["expression"]
}
}
2.3 docstring 是灵魂
LLM 全靠 docstring 判断"这个工具能不能解决当前问题" 。写好 docstring 是工具调用的第一要务：
python
# ❌ 烂写法：LLM 不知道什么时候该用
@tool
def search(query: str) -> str:
"""搜索"""
return "..."
# ✅ 好写法：明确说明用途、适用场景、参数含义
@tool
def search_web(query: str) -> str:
"""搜索互联网获取实时信息。
适用场景：
- 用户询问最新新闻、实时数据
- 需要查询当前日期之后的信息
- 需要验证某个事实的最新状态
不适用：纯知识类问题（直接回答即可）
Args:
query: 搜索关键词，应简洁精准
"""
return f"搜索结果：..."
2.4 多参数与可选参数
python
from typing import Optional
from pydantic import BaseModel, Field
# 方式1：直接用类型注解 + 默认值
@tool
def query_database(
sql: str,
database: str = "default",
limit: int = 100,
) -> str:
"""执行 SQL 查询。
Args:
sql: SQL 查询语句
database: 数据库名称，默认使用 default
limit: 返回结果上限，默认100条
"""
return f"在 {database} 执行: {sql[:50]}... (limit={limit})"
# 方式2：Pydantic Schema（复杂参数推荐）
class FileReadInput(BaseModel):
file_path: str = Field(description="文件路径，如 /data/report.csv")
encoding: str = Field(default="utf-8", description="文件编码")
max_lines: Optional[int] = Field(default=None, description="最大读取行数，None表示全部")
@tool("file_read", args_schema=FileReadInput)
def file_read(file_path: str, encoding: str = "utf-8", max_lines: Optional[int] = None) -> str:
"""读取文件内容。支持文本文件和CSV文件的读取。"""
return f"读取 {file_path} (encoding={encoding}, max_lines={max_lines})"
2.5 五个实用工具定义
python
import json
from typing import Optional
from datetime import datetime
from langchain_core.tools import tool
from pydantic import BaseModel, Field
# 1. 网络搜索
@tool
def web_search(query: str, max_results: int = 5) -> str:
"""搜索互联网获取实时信息。
适用于：最新新闻、实时数据、事实核查。
不适用于：纯知识问答、数学计算。
Args:
query: 搜索关键词
max_results: 返回结果数量上限
"""
# 实际调用搜索 API（如 Tavily、Serper）
return json.dumps([
{"title": f"搜索结果 {i+1}", "snippet": f"关于 {query} 的信息..."}
for i in range(max_results)
], ensure_ascii=False)
# 2. 代码执行
class CodeExecInput(BaseModel):
code: str = Field(description="要执行的 Python 代码")
timeout: int = Field(default=30, description="执行超时时间（秒）")
@tool("code_execute", args_schema=CodeExecInput)
def code_execute(code: str, timeout: int = 30) -> str:
"""在沙箱中执行 Python 代码并返回结果。
适用于：数据分析、数学计算、文本处理。
注意：代码在受限环境中执行，不支持网络访问和文件写入。
Args:
code: 要执行的 Python 代码
timeout: 执行超时时间（秒），默认30秒
"""
try:
local_vars = {}
exec(code, {"__builtins__": {}}, local_vars)
return str(local_vars.get("result", "代码执行成功，无返回值"))
except Exception as e:
return f"执行错误: {type(e).__name__}: {e}"
# 3. 文件读取
@tool
def file_read(file_path: str) -> str:
"""读取文件内容。
适用于：查看配置文件、读取数据文件、检查日志。
Args:
file_path: 文件路径
"""
try:
with open(file_path, "r", encoding="utf-8") as f:
content = f.read(10000) # 限制读取大小
return content if content else "文件为空"
except FileNotFoundError:
return f"错误：文件 {file_path} 不存在"
except Exception as e:
return f"读取错误: {e}"
# 4. 数据库查询
@tool
def db_query(sql: str, database: str = "main") -> str:
"""执行 SQL 查询并返回结果。
适用于：查询业务数据、统计分析。
仅支持 SELECT 语句，不支持修改操作。
Args:
sql: SQL 查询语句（仅 SELECT）
database: 数据库名称
"""
# 安全校验
if not sql.strip().upper().startswith("SELECT"):
return "错误：仅支持 SELECT 查询"
# 模拟查询结果
return json.dumps({
"columns": ["id", "name", "value"],
"rows": [[1, "sample", 42]],
"row_count": 1
}, ensure_ascii=False)
# 5. 获取当前时间
@tool
def get_current_time(timezone: str = "Asia/Shanghai") -> str:
"""获取当前日期和时间。
适用于：需要知道当前时间的场景，如计算截止日期、安排日程。
Args:
timezone: 时区，默认 Asia/Shanghai
"""
now = datetime.now()
return f"当前时间: {now.strftime('%Y-%m-%d %H:%M:%S')} (时区: {timezone})"
三、ToolNode：LangGraph 的工具执行器
3.1 ToolNode 是什么
ToolNode 是 LangGraph 预置的工具执行节点，它做了一件事：自动执行 LLM 返回的 tool_calls，并把结果包装成 ToolMessage 回传。
看起来简单，但如果你手动实现，要处理的边界情况一堆：
并行工具调用（LLM 一次返回多个 tool_call）
工具执行异常捕获和格式化
ToolMessage 的
tool_call_id必须与AIMessage.tool_calls中的id对应工具返回值的序列化（str / dict / list 都要处理）
Command对象的传播（工具内部更新 State）
3.2 基本用法
python
from langgraph.prebuilt import ToolNode
# 创建 ToolNode，传入工具列表
tool_node = ToolNode(tools=[web_search, code_execute, file_read, db_query])
ToolNode 要求 State 中有 messages 字段，且使用 add_messages reducer：
python
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages
class State(TypedDict):
messages: Annotated[list, add_messages]
3.3 工具调用循环
LangGraph 中工具调用的核心是 Agent → ToolNode 的循环：
plaintext
┌─────────────────────────────────────────────────────────────────────┐
│ ReAct 工具调用循环 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ START │
│ │ │
│ ▼ │
│ ┌──────────┐ │
│ │ agent │ call_model(): LLM 推理，决定是否调用工具 │
│ └────┬─────┘ │
│ │ │
│ ▼ │
│ should_continue() │
│ │ │
│ ├── 有 tool_calls ──▶ ┌──────────┐ │
│ │ │ tools │ ToolNode 执行工具 │
│ │ └────┬─────┘ │
│ │ │ │
│ │ └──────▶ 回到 agent（继续推理） │
│ │ │
│ └── 无 tool_calls ──▶ END（返回最终回答） │
│ │
└─────────────────────────────────────────────────────────────────────┘
3.4 should_continue 路由函数
这是整个循环的"方向盘"——判断 LLM 的输出是要调工具，还是直接结束：
python
from typing import Literal
from langgraph.graph import END, MessagesState
def should_continue(state: MessagesState) -> Literal["tools", "__end__"]:
"""判断是否需要继续调用工具。
检查最后一条消息（LLM 输出）是否包含 tool_calls：
- 有 → 路由到 "tools" 节点执行工具
- 无 → 路由到 END，返回最终回答
"""
messages = state["messages"]
last_message = messages[-1]
if last_message.tool_calls:
return "tools"
return END
也可以使用 LangGraph 预置的 tools_condition：
python
from langgraph.prebuilt import tools_condition
# tools_condition 功能与 should_continue 完全等价
workflow.add_conditional_edges("agent", tools_condition, ["tools", END])
3.5 完整的最小可运行示例
python
from typing import Annotated, Literal
from typing_extensions import TypedDict
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
# 1. 定义工具
@tool
def get_weather(location: str) -> str:
"""查询指定城市的天气信息。
Args:
location: 城市名称，如 "北京"、"上海"、"San Francisco"
"""
weather_data = {
"北京": "晴天，12°C，空气质量良好",
"上海": "多云，18°C，有轻微雾霾",
"san francisco": "雾天，15°C，典型SF天气",
}
return weather_data.get(location.lower(), f"{location}：暂无天气数据")
@tool
def calculator(expression: str) -> str:
"""计算数学表达式。
Args:
expression: 数学表达式，如 "2 + 3 * 4"
"""
try:
result = eval(expression)
return f"计算结果: {expression} = {result}"
except Exception as e:
return f"计算错误: {e}"
# 2. 定义状态
class State(TypedDict):
messages: Annotated[list, add_messages]
# 3. 创建工具节点和模型
tools = [get_weather, calculator]
tool_node = ToolNode(tools)
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
model_with_tools = model.bind_tools(tools)
# 4. 定义节点函数
def call_model(state: State):
"""Agent 节点：调用 LLM 推理"""
response = model_with_tools.invoke(state["messages"])
return {"messages": [response]}
def should_continue(state: State) -> Literal["tools", "__end__"]:
"""路由函数：判断是否调用工具"""
last_message = state["messages"][-1]
if last_message.tool_calls:
return "tools"
return END
# 5. 构建图
workflow = StateGraph(State)
# 添加节点
workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)
# 添加边
workflow.add_edge(START, "agent")
workflow.add_conditional_edges("agent", should_continue, ["tools", END])
workflow.add_edge("tools", "agent") # 工具执行后回到 agent
# 编译
app = workflow.compile()
# 6. 运行
if __name__ == "__main__":
result = app.invoke({
"messages": [{"role": "user", "content": "北京天气怎么样？顺便帮我算一下 (12 + 8) * 3"}]
})
for msg in result["messages"]:
print(f"[{msg.type.upper()}] {msg.content}")
运行输出类似：
plaintext
[HUMAN] 北京天气怎么样？顺便帮我算一下 (12 + 8) * 3
[AI] (tool_calls: get_weather, calculator)
[TOOL] 晴天，12°C，空气质量良好
[TOOL] 计算结果: (12 + 8) * 3 = 60
[AI] 北京现在是晴天，气温12°C，空气质量良好。另外，(12 + 8) × 3 = 60。
四、实战：多功能研究 Agent
把前面的工具组合起来，构建一个能搜索、能算、能读文件、能查库的研究 Agent：
python
import json
from typing import Annotated, Literal, Optional
from datetime import datetime
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.prebuilt import ToolNode
from pydantic import BaseModel, Field
# ==================== 工具定义 ====================
@tool
def web_search(query: str, max_results: int = 3) -> str:
"""搜索互联网获取实时信息。适用于最新新闻、实时数据、事实核查。
Args:
query: 搜索关键词，应简洁精准
max_results: 返回结果数量上限，默认3
"""
# 实际实现中调用 Tavily / Serper / Bing API
mock_results = [
{"title": f"关于 {query} 的最新报道", "url": "https://example.com/1",
"snippet": f"根据最新消息，{query} 的相关情况如下..."},
{"title": f"{query} 深度分析", "url": "https://example.com/2",
"snippet": f"专家对 {query} 的深度解读..."},
]
return json.dumps(mock_results[:max_results], ensure_ascii=False)
class CodeExecInput(BaseModel):
code: str = Field(description="要执行的 Python 代码，结果赋值给变量 result")
timeout: int = Field(default=30, description="超时秒数")
@tool("code_execute", args_schema=CodeExecInput)
def code_execute(code: str, timeout: int = 30) -> str:
"""在沙箱中执行 Python 代码。适用于数据分析、数学计算、文本处理。
Args:
code: Python 代码，执行结果需赋值给 result 变量
timeout: 执行超时时间（秒）
"""
try:
safe_builtins = {
"abs": abs, "len": len, "range": range,
"int": int, "float": float, "str": str,
"list": list, "dict": dict, "set": set,
"sum": sum, "min": min, "max": max,
"sorted": sorted, "enumerate": enumerate,
"zip": zip, "map": map, "filter": filter,
}
local_vars = {}
exec(code, {"__builtins__": safe_builtins}, local_vars)
result = local_vars.get("result", "代码执行成功，无返回值")
return str(result)
except Exception as e:
return f"执行错误: {type(e).__name__}: {e}"
@tool
def file_read(file_path: str) -> str:
"""读取文件内容。适用于查看配置、数据文件、日志。
Args:
file_path: 文件路径
"""
try:
with open(file_path, "r", encoding="utf-8") as f:
content = f.read(50000)
return content[:10000] if len(content) > 10000 else content
except FileNotFoundError:
return f"错误：文件 {file_path} 不存在"
except Exception as e:
return f"读取错误: {e}"
@tool
def db_query(sql: str) -> str:
"""执行只读 SQL 查询。适用于查询业务数据、统计分析。
仅支持 SELECT 语句，禁止 INSERT/UPDATE/DELETE。
Args:
sql: SQL 查询语句
"""
if not sql.strip().upper().startswith("SELECT"):
return "安全限制：仅支持 SELECT 查询"
# 模拟查询
return json.dumps({
"columns": ["id", "name", "value"],
"rows": [[1, "sample", 42], [2, "test", 88]],
"row_count": 2
}, ensure_ascii=False)
# ==================== 构建 Agent ====================
tools = [web_search, code_execute, file_read, db_query]
tool_node = ToolNode(tools)
model = ChatOpenAI(model="gpt-4o", temperature=0).bind_tools(tools)
def call_model(state: MessagesState):
"""Agent 节点"""
system_prompt = {
"role": "system",
"content": (
"你是一个研究助手。你可以搜索互联网、执行代码、读取文件、查询数据库。\n"
"请根据用户问题，选择合适的工具来获取信息，然后综合分析给出回答。\n"
"如果需要多步操作，请一步步来，每次调用一个或多个工具。"
)
}
messages = [system_prompt] + state["messages"]
response = model.invoke(messages)
return {"messages": [response]}
def should_continue(state: MessagesState) -> Literal["tools", "__end__"]:
"""路由函数"""
last_message = state["messages"][-1]
if last_message.tool_calls:
return "tools"
return END
# 构建图
workflow = StateGraph(MessagesState)
workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)
workflow.add_edge(START, "agent")
workflow.add_conditional_edges("agent", should_continue, ["tools", END])
workflow.add_edge("tools", "agent")
app = workflow.compile()
# ==================== 运行 ====================
if __name__ == "__main__":
result = app.invoke({
"messages": [{
"role": "user",
"content": "搜索一下 LangGraph 最新的版本号，然后帮我算一下 2^20 是多少"
}]
})
print("=" * 60)
print("最终回答：")
print(result["messages"][-1].content)
执行流程图：
plaintext
用户："搜索 LangGraph 最新版本号，算 2^20"
│
▼
Agent Node (LLM 推理)
│ → 返回 AIMessage，含 tool_calls:
│ 1. web_search(query="LangGraph latest version")
│ 2. code_execute(code="result = 2**20")
▼
ToolNode (并行执行两个工具)
│ → ToolMessage 1: 搜索结果...
│ → ToolMessage 2: 1048576
▼
Agent Node (LLM 综合分析)
│ → 返回 AIMessage，无 tool_calls
▼
END → "LangGraph 最新版本是 0.2.x，2^20 = 1,048,576"
五、错误处理
工具调用在真实场景中不可能一帆风顺。超时、参数错误、格式异常……都需要妥善处理。
5.1 ToolNode 内置错误处理
ToolNode 自带错误处理机制——当工具执行抛异常时，它不会让整个图崩溃，而是把错误信息包装成 ToolMessage 回传给 LLM，让 LLM 自行修正：
python
# ToolNode 内部大致逻辑（简化版）：
# try:
# result = tool.invoke(tool_call.args)
# return ToolMessage(content=str(result), tool_call_id=tool_call.id)
# except Exception as e:
# return ToolMessage(
# content=f"Error: {e}\n Please fix your mistakes.",
# tool_call_id=tool_call.id,
# status="error" # 标记为错误
# )
可以通过 handle_tool_errors 参数控制行为：
python
from langgraph.prebuilt import ToolNode
# 方式1：True — 捕获所有异常，回传错误信息给 LLM（默认行为）
tool_node = ToolNode(tools=tools, handle_tool_errors=True)
# 方式2：False — 异常直接抛出，由图级别处理
tool_node = ToolNode(tools=tools, handle_tool_errors=False)
# 方式3：指定只捕获某些异常类型
tool_node = ToolNode(tools=tools, handle_tool_errors=(ValueError, TypeError))
# 方式4：自定义错误消息
tool_node = ToolNode(
tools=tools,
handle_tool_errors="工具执行失败，请检查参数后重试"
)
# 方式5：用函数动态生成错误消息
def custom_error_handler(e: Exception) -> str:
if isinstance(e, TimeoutError):
return f"操作超时: {e}。请简化请求或增加超时时间。"
return f"执行出错: {e}。请修正后重试。"
tool_node = ToolNode(tools=tools, handle_tool_errors=custom_error_handler)
5.2 常见错误场景及处理
python
import time
from langchain_core.tools import tool
# 场景1：工具执行超时
@tool
def slow_api_call(query: str) -> str:
"""调用可能超时的外部 API。"""
import signal
def timeout_handler(signum, frame):
raise TimeoutError(f"API 调用超时: {query}")
# 设置超时
signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(5) # 5秒超时
try:
# 模拟慢 API
time.sleep(10)
return "API 结果"
finally:
signal.alarm(0) # 取消定时器
# 场景2：工具参数校验
@tool
def safe_db_query(sql: str) -> str:
"""执行安全的数据库查询。仅支持 SELECT。"""
# 参数校验
if not sql or not sql.strip():
raise ValueError("SQL 语句不能为空")
if not sql.strip().upper().startswith("SELECT"):
raise ValueError(f"仅支持 SELECT 查询，收到: {sql[:20]}...")
if len(sql) > 10000:
raise ValueError("SQL 语句过长，最大支持10000字符")
return "查询结果..."
# 场景3：LLM 返回了无法解析的 tool_call
# 这种情况 ToolNode 会自动处理：
# - 找不到对应工具名 → 返回错误 ToolMessage
# - 参数格式不匹配 → 返回错误 ToolMessage
5.3 自定义错误恢复策略
生产环境中，简单的"把错误回传 LLM"不够用。我们需要更精细的恢复策略：
python
from typing import Annotated, Literal
from langchain_core.messages import AIMessage, RemoveMessage, ToolMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.prebuilt import ToolNode
# 定义工具
@tool
def risky_operation(action: str) -> str:
"""执行可能有风险的操作。"""
import random
if random.random() < 0.5:
raise ConnectionError("网络连接失败")
if action == "delete":
raise PermissionError("不允许删除操作")
return f"操作成功: {action}"
tools = [risky_operation]
tool_node = ToolNode(tools, handle_tool_errors=True)
# 主模型和备用模型
primary_model = ChatOpenAI(model="gpt-4o-mini", temperature=0).bind_tools(tools)
fallback_model = ChatOpenAI(model="gpt-4o", temperature=0).bind_tools(tools)
def call_model(state: MessagesState):
"""Agent 节点"""
response = primary_model.invoke(state["messages"])
return {"messages": [response]}
def should_continue(state: MessagesState) -> Literal["tools", "__end__"]:
"""是否需要调用工具"""
last_message = state["messages"][-1]
if last_message.tool_calls:
return "tools"
return END
def should_fallback(state: MessagesState) -> Literal["agent", "remove_failed", "__end__"]:
"""工具执行后：检查是否有错误，决定恢复策略"""
messages = state["messages"]
# 检查最近的 ToolMessage 是否有错误
failed_tool_messages = [
msg for msg in messages
if isinstance(msg, ToolMessage)
and msg.status == "error"
]
if failed_tool_messages:
return "remove_failed" # 有错误 → 清理后用备用模型重试
return "agent" # 无错误 → 回到正常 Agent 流程
def remove_failed_tool_call_attempt(state: MessagesState):
"""移除失败的工具调用尝试，清理状态"""
messages = state["messages"]
# 找到最近的 AIMessage，移除它和后续的所有 ToolMessage
last_ai_idx = next(
i for i, msg in reversed(list(enumerate(messages)))
if isinstance(msg, AIMessage)
)
messages_to_remove = messages[last_ai_idx:]
return {"messages": [RemoveMessage(id=m.id) for m in messages_to_remove]}
def call_fallback_model(state: MessagesState):
"""使用更强大的模型重试"""
response = fallback_model.invoke(state["messages"])
return {"messages": [response]}
# 构建带错误恢复的图
workflow = StateGraph(MessagesState)
workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)
workflow.add_node("remove_failed", remove_failed_tool_call_attempt)
workflow.add_node("fallback_agent", call_fallback_model)
workflow.add_edge(START, "agent")
workflow.add_conditional_edges("agent", should_continue, ["tools", END])
workflow.add_conditional_edges("tools", should_fallback)
workflow.add_edge("remove_failed", "fallback_agent")
workflow.add_edge("fallback_agent", "tools")
app = workflow.compile()
错误恢复流程图：
plaintext
┌──────────────────────────────────────────────────────────────────┐
│ 带错误恢复的 Agent 流程 │
├──────────────────────────────────────────────────────────────────┤
│ │
│ START │
│ │ │
│ ▼ │
│ ┌──────────┐ │
│ │ agent │ (主模型 gpt-4o-mini) │
│ └────┬─────┘ │
│ │ │
│ should_continue() │
│ │ │
│ ├── 无 tool_calls ──▶ END │
│ │ │
│ └── 有 tool_calls ──▶ ┌──────────┐ │
│ │ tools │ │
│ └────┬─────┘ │
│ │ │
│ should_fallback() │
│ │ │
│ ┌─────────────┼─────────────┐ │
│ │ 成功 │ 有错误 │ │
│ ▼ ▼ │ │
│ ┌─────────┐ ┌──────────────┐ │ │
│ │ agent │ │ remove_failed│ │ │
│ │(继续) │ │ (清理状态) │ │ │
│ └─────────┘ └──────┬───────┘ │ │
│ │ │ │
│ ▼ │ │
│ ┌────────────────┐ │ │
│ │fallback_agent │ │ │
│ │(备用模型 gpt-4o)│ │ │
│ └───────┬────────┘ │ │
│ │ │ │
│ └──▶ tools ────┘ │
│ (重试工具调用) │
│ │
└──────────────────────────────────────────────────────────────────┘
5.4 带重试和降级的工具包装器
对于工具本身的执行，推荐用装饰器封装重试逻辑：
python
import time
import functools
from typing import Callable, Optional
def resilient_tool(
max_retries: int = 3,
retry_delay: float = 1.0,
fallback_value: Optional[str] = None,
timeout_seconds: int = 30,
):
"""让工具具备重试和降级能力的装饰器。
Args:
max_retries: 最大重试次数
retry_delay: 重试间隔（秒），指数退避
fallback_value: 所有重试失败后的降级返回值
timeout_seconds: 单次执行超时（秒）
"""
def decorator(func: Callable):
@functools.wraps(func)
def wrapper(*args, **kwargs):
last_error = None
for attempt in range(max_retries):
try:
# 这里可以加超时控制
return func(*args, **kwargs)
except Exception as e:
last_error = e
if attempt < max_retries - 1:
delay = retry_delay * (2 ** attempt) # 指数退避
time.sleep(delay)
# 所有重试都失败
if fallback_value is not None:
return fallback_value
raise last_error
return wrapper
return decorator
# 使用示例
@tool
@resilient_tool(max_retries=3, retry_delay=1.0, fallback_value="服务暂时不可用，请稍后重试")
def search_api(query: str) -> str:
"""搜索 API，自带重试和降级。"""
import random
if random.random() < 0.7:
raise ConnectionError("API 连接失败")
return f"搜索结果: {query}"
六、工具安全与权限控制
Agent 有了工具就像给了员工系统权限——不控制就出事。
6.1 工具白名单/黑名单
python
from langchain_core.tools import tool, BaseTool
from typing import Optional
class ToolAccessControl:
"""工具访问控制器：白名单 + 黑名单机制"""
def __init__(
self,
whitelist: Optional[list[str]] = None,
blacklist: Optional[list[str]] = None,
):
self.whitelist = whitelist # 白名单：只允许这些工具
self.blacklist = blacklist or [] # 黑名单：禁止这些工具
def filter_tools(self, tools: list[BaseTool], user_role: str = "user") -> list[BaseTool]:
"""根据用户角色过滤可用工具"""
filtered = []
for tool in tools:
tool_name = tool.name
# 黑名单优先
if tool_name in self.blacklist:
continue
# 白名单模式：只允许白名单中的工具
if self.whitelist and tool_name not in self.whitelist:
continue
filtered.append(tool)
return filtered
# 按角色定义权限
ROLE_PERMISSIONS = {
"admin": {
"whitelist": None, # None = 允许所有
"blacklist": [],
},
"analyst": {
"whitelist": ["web_search", "code_execute", "db_query", "file_read"],
"blacklist": [],
},
"viewer": {
"whitelist": ["web_search", "db_query"],
"blacklist": ["code_execute", "file_read"], # 不能执行代码和读文件
},
}
# 在 Agent 中使用
def create_agent_with_role(role: str):
"""根据角色创建 Agent"""
all_tools = [web_search, code_execute, file_read, db_query]
# 获取角色权限
perms = ROLE_PERMISSIONS.get(role, ROLE_PERMISSIONS["viewer"])
# 过滤工具
access_control = ToolAccessControl(
whitelist=perms["whitelist"],
blacklist=perms["blacklist"],
)
allowed_tools = access_control.filter_tools(all_tools, role)
tool_node = ToolNode(tools=allowed_tools)
model = ChatOpenAI(model="gpt-4o-mini", temperature=0).bind_tools(allowed_tools)
# ... 构建图
return tool_node, model
6.2 敏感操作需人工确认
对于危险操作（删除、执行不可信代码、发送消息等），应该结合 LangGraph 的 interrupt 机制（详见第39篇人在回路）：
python
from langgraph.types import interrupt, Command
from langchain_core.tools import tool
@tool
def delete_records(table: str, condition: str) -> str:
"""删除数据库记录。⚠️ 危险操作，需人工确认。
Args:
table: 表名
condition: 删除条件
"""
# 请求人工确认
human_decision = interrupt(
f"⚠️ 危险操作确认：是否删除 {table} 中满足 {condition} 的记录？\n"
f"请确认 (yes/no)"
)
if human_decision == "yes":
# 执行删除
return f"已删除 {table} 中满足 {condition} 的记录"
else:
return "操作已取消"
@tool
def send_email(to: str, subject: str, body: str) -> str:
"""发送邮件。⚠️ 需人工确认后发送。
Args:
to: 收件人邮箱
subject: 邮件主题
body: 邮件正文
"""
human_decision = interrupt(
f"📧 发送邮件确认：\n"
f"收件人: {to}\n"
f"主题: {subject}\n"
f"正文: {body[:100]}...\n"
f"确认发送？(yes/no)"
)
if human_decision == "yes":
return f"邮件已发送至 {to}"
return "邮件发送已取消"
6.3 工具调用审计日志
python
import json
import logging
from datetime import datetime
from langchain_core.tools import tool
from langchain_core.runnables import RunnableConfig
logger = logging.getLogger("tool_audit")
# 审计装饰器
def audit_tool(func):
"""为工具添加审计日志"""
@functools.wraps(func)
def wrapper(*args, **kwargs):
tool_name = func.__name__
timestamp = datetime.now().isoformat()
call_id = kwargs.get("config", {}).get("configurable", {}).get("run_id", "unknown")
# 记录调用
logger.info(json.dumps({
"event": "tool_call_start",
"tool": tool_name,
"call_id": call_id,
"timestamp": timestamp,
"args": str(kwargs)[:500], # 截断避免日志过大
}, ensure_ascii=False))
start_time = time.time()
try:
result = func(*args, **kwargs)
# 记录成功
logger.info(json.dumps({
"event": "tool_call_success",
"tool": tool_name,
"call_id": call_id,
"duration_ms": int((time.time() - start_time) * 1000),
"result_length": len(str(result)),
}, ensure_ascii=False))
return result
except Exception as e:
# 记录失败
logger.error(json.dumps({
"event": "tool_call_error",
"tool": tool_name,
"call_id": call_id,
"duration_ms": int((time.time() - start_time) * 1000),
"error_type": type(e).__name__,
"error_message": str(e)[:200],
}, ensure_ascii=False))
raise
return wrapper
# 使用
@tool
@audit_tool
def sensitive_query(sql: str) -> str:
"""查询敏感数据。所有调用都会被审计记录。"""
return "查询结果..."
6.4 速率限制
python
import time
from collections import defaultdict
class RateLimiter:
"""工具调用速率限制器"""
def __init__(self, max_calls: int = 10, window_seconds: int = 60):
self.max_calls = max_calls
self.window_seconds = window_seconds
self._call_times: dict[str, list[float]] = defaultdict(list)
def check(self, tool_name: str, user_id: str = "default") -> bool:
"""检查是否允许调用"""
key = f"{tool_name}:{user_id}"
now = time.time()
# 清理过期记录
self._call_times[key] = [
t for t in self._call_times[key]
if now - t < self.window_seconds
]
if len(self._call_times[key]) >= self.max_calls:
return False
self._call_times[key].append(now)
return True
rate_limiter = RateLimiter(max_calls=10, window_seconds=60)
def rate_limited_tool(func):
"""为工具添加速率限制"""
@functools.wraps(func)
def wrapper(*args, **kwargs):
if not rate_limiter.check(func.__name__):
raise RuntimeError(
f"工具 {func.__name__} 调用频率超限，"
f"请稍后重试（限制：10次/分钟）"
)
return func(*args, **kwargs)
return wrapper
# 使用
@tool
@rate_limited_tool
def expensive_api_call(query: str) -> str:
"""调用昂贵的外部 API（有速率限制）。"""
return f"API 结果: {query}"
6.5 安全工具包装器（综合示例）
把上面的安全措施整合到一个统一的包装器：
python
import functools
import json
import logging
import time
from datetime import datetime
from typing import Optional, Callable
from langchain_core.tools import tool
logger = logging.getLogger("tool_security")
def secure_tool(
*,
require_confirmation: bool = False,
max_retries: int = 2,
rate_limit: Optional[int] = None,
audit: bool = True,
allowed_roles: Optional[list[str]] = None,
max_result_length: int = 5000,
):
"""统一的安全工具包装器。
Args:
require_confirmation: 是否需要人工确认
max_retries: 最大重试次数
rate_limit: 速率限制（次/分钟）
audit: 是否记录审计日志
allowed_roles: 允许使用的角色列表
max_result_length: 返回结果最大长度
"""
def decorator(func: Callable):
@functools.wraps(func)
def wrapper(*args, **kwargs):
tool_name = func.__name__
start_time = time.time()
# 1. 审计日志
if audit:
logger.info(json.dumps({
"event": "tool_call",
"tool": tool_name,
"timestamp": datetime.now().isoformat(),
}, ensure_ascii=False))
# 2. 速率限制
if rate_limit:
if not rate_limiter.check(tool_name):
return f"错误：工具 {tool_name} 调用频率超限"
# 3. 人工确认
if require_confirmation:
decision = interrupt(
f"⚠️ 工具 {tool_name} 需要确认才能执行。确认？(yes/no)"
)
if decision != "yes":
return "操作已取消"
# 4. 带重试的执行
last_error = None
for attempt in range(max_retries + 1):
try:
result = func(*args, **kwargs)
# 5. 截断结果
if isinstance(result, str) and len(result) > max_result_length:
result = result[:max_result_length] + f"\n... (已截断，原始长度 {len(result)})"
# 审计成功
if audit:
logger.info(json.dumps({
"event": "tool_success",
"tool": tool_name,
"duration_ms": int((time.time() - start_time) * 1000),
}, ensure_ascii=False))
return result
except Exception as e:
last_error = e
if attempt < max_retries:
time.sleep(1 * (2 ** attempt))
# 所有重试失败
if audit:
logger.error(json.dumps({
"event": "tool_failed",
"tool": tool_name,
"error": str(last_error)[:200],
"duration_ms": int((time.time() - start_time) * 1000),
}, ensure_ascii=False))
return f"工具执行失败: {type(last_error).__name__}: {last_error}"
return wrapper
return decorator
# 使用示例
@tool
@secure_tool(
require_confirmation=True, # 需人工确认
max_retries=2, # 最多重试2次
rate_limit=5, # 5次/分钟
audit=True, # 记录审计日志
max_result_length=3000, # 结果最大3KB
)
def execute_maintenance(action: str, target: str) -> str:
"""执行系统维护操作。⚠️ 危险操作，需确认。
Args:
action: 操作类型 (restart/clear_cache/update_config)
target: 操作目标
"""
return f"维护操作完成: {action} on {target}"
七、动态工具注册
有时候工具不是固定的——不同用户、不同场景、不同租户可用的工具不同。
7.1 运行时根据 State 决定可用工具
python
from typing import Annotated
from langchain_core.tools import BaseTool
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.prebuilt import ToolNode
# 所有可用工具的注册表
TOOL_REGISTRY: dict[str, BaseTool] = {
"web_search": web_search,
"code_execute": code_execute,
"file_read": file_read,
"db_query": db_query,
}
# 按用户角色分配工具
ROLE_TOOL_MAP = {
"admin": list(TOOL_REGISTRY.keys()),
"analyst": ["web_search", "code_execute", "db_query"],
"viewer": ["web_search", "db_query"],
}
class DynamicAgentState(MessagesState):
"""扩展状态，包含用户角色信息"""
role: str # 用户角色
available_tools: list[str] # 当前可用工具名
def call_model_with_dynamic_tools(state: DynamicAgentState):
"""根据状态中的角色动态绑定工具"""
role = state.get("role", "viewer")
available_tool_names = ROLE_TOOL_MAP.get(role, ["web_search"])
# 获取工具对象
current_tools = [
TOOL_REGISTRY[name]
for name in available_tool_names
if name in TOOL_REGISTRY
]
# 动态绑定工具到模型
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
model_with_tools = model.bind_tools(current_tools)
response = model_with_tools.invoke(state["messages"])
return {"messages": [response]}
def dynamic_tool_node(state: DynamicAgentState):
"""动态选择工具执行"""
role = state.get("role", "viewer")
available_tool_names = ROLE_TOOL_MAP.get(role, ["web_search"])
current_tools = [
TOOL_REGISTRY[name]
for name in available_tool_names
if name in TOOL_REGISTRY
]
tool_node = ToolNode(tools=current_tools)
return tool_node.invoke(state)
def should_continue(state: DynamicAgentState):
last_message = state["messages"][-1]
if last_message.tool_calls:
return "tools"
return END
# 构建图
workflow = StateGraph(DynamicAgentState)
workflow.add_node("agent", call_model_with_dynamic_tools)
workflow.add_node("tools", dynamic_tool_node)
workflow.add_edge(START, "agent")
workflow.add_conditional_edges("agent", should_continue, ["tools", END])
workflow.add_edge("tools", "agent")
app = workflow.compile()
# 运行时指定角色
result = app.invoke({
"messages": [{"role": "user", "content": "帮我搜索一下"}],
"role": "viewer", # viewer 只能用 web_search 和 db_query
})
7.2 从配置文件加载工具
yaml
# tools_config.yaml
tools:
- name: web_search
enabled: true
roles: [admin, analyst, viewer]
- name: code_execute
enabled: true
roles: [admin, analyst]
- name: file_read
enabled: true
roles: [admin]
config:
allowed_paths: ["/data", "/tmp"]
- name: db_query
enabled: true
roles: [admin, analyst, viewer]
- name: send_email
enabled: false # 暂时禁用
roles: [admin]
python
import yaml
from pathlib import Path
def load_tools_from_config(config_path: str, role: str) -> list:
"""从 YAML 配置文件加载工具"""
config = yaml.safe_load(Path(config_path).read_text())
tool_names = []
for tool_cfg in config.get("tools", []):
if not tool_cfg.get("enabled", True):
continue
if role not in tool_cfg.get("roles", []):
continue
tool_names.append(tool_cfg["name"])
return [
TOOL_REGISTRY[name]
for name in tool_names
if name in TOOL_REGISTRY
]
7.3 插件式工具架构
python
import importlib
import importlib.util
from pathlib import Path
from langchain_core.tools import BaseTool
class ToolPluginManager:
"""插件式工具管理器：自动发现和加载工具"""
def __init__(self, plugins_dir: str = "plugins"):
self.plugins_dir = Path(plugins_dir)
self._registry: dict[str, BaseTool] = {}
def discover_tools(self) -> dict[str, BaseTool]:
"""扫描插件目录，自动发现和注册工具"""
if not self.plugins_dir.exists():
return self._registry
for file_path in self.plugins_dir.glob("*.py"):
if file_path.name.startswith("_"):
continue
try:
module_name = f"plugins.{file_path.stem}"
spec = importlib.util.spec_from_file_location(module_name, file_path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
# 查找模块中所有带 is_tool 属性的对象
for attr_name in dir(module):
obj = getattr(module, attr_name)
if hasattr(obj, "name") and hasattr(obj, "invoke"):
# 是 LangChain 工具对象
self._registry[obj.name] = obj
print(f" 发现工具: {obj.name}")
except Exception as e:
print(f" 加载失败 {file_path.name}: {e}")
return self._registry
def get_tools(self, names: list[str] | None = None) -> list[BaseTool]:
"""获取工具列表"""
if names is None:
return list(self._registry.values())
return [self._registry[n] for n in names if n in self._registry]
# 插件文件示例：plugins/weather.py
# -------------------------
# from langchain_core.tools import tool
#
# @tool
# def get_weather(city: str) -> str:
# """获取城市天气信息。"""
# return f"{city}: 晴天，25°C"
# -------------------------
# 使用
plugin_manager = ToolPluginManager("plugins")
plugin_manager.discover_tools()
all_tools = plugin_manager.get_tools()
tool_node = ToolNode(tools=all_tools)
八、Structured Output 与工具结合
有时候你需要的不是自由文本回复，而是结构化数据——比如提取实体、分类结果、表单数据。
8.1 with_structured_output 基础
python
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
class ResearchReport(BaseModel):
"""研究报告结构"""
title: str = Field(description="报告标题")
summary: str = Field(description="200字以内的摘要")
key_findings: list[str] = Field(description="关键发现列表")
confidence: float = Field(description="置信度 0-1")
sources: list[str] = Field(description="信息来源列表")
model = ChatOpenAI(model="gpt-4o", temperature=0)
# 方式1：纯结构化输出（不调用工具）
structured_model = model.with_structured_output(ResearchReport)
report = structured_model.invoke("分析一下 LangGraph 工具调用的优缺点")
print(report.title)
print(report.key_findings)
8.2 在 Agent 中同时使用工具和结构化输出
python
from typing import Annotated, Literal, Optional
from pydantic import BaseModel, Field
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.prebuilt import ToolNode
# 定义工具
@tool
def web_search(query: str) -> str:
"""搜索互联网获取实时信息。"""
return f"搜索结果: {query}"
@tool
def db_query(sql: str) -> str:
"""查询数据库。"""
return "查询结果: ..."
# 定义结构化输出 Schema
class AnalysisResult(BaseModel):
"""分析结果结构"""
conclusion: str = Field(description="分析结论")
evidence: list[str] = Field(description="支撑证据")
confidence_score: float = Field(description="置信度评分 0.0-1.0")
recommendation: Optional[str] = Field(default=None, description="行动建议")
# 构建 Agent
tools = [web_search, db_query]
tool_node = ToolNode(tools)
# 模型绑定工具
model_with_tools = ChatOpenAI(model="gpt-4o", temperature=0).bind_tools(tools)
# 结构化输出模型（用于最后一步）
structured_model = ChatOpenAI(model="gpt-4o", temperature=0).with_structured_output(AnalysisResult)
class AgentState(MessagesState):
"""Agent 状态，增加结构化结果字段"""
analysis_result: Optional[AnalysisResult] = None
def call_model(state: AgentState):
"""Agent 节点"""
response = model_with_tools.invoke(state["messages"])
return {"messages": [response]}
def should_continue(state: AgentState) -> Literal["tools", "structure", "__end__"]:
"""路由：需要工具→tools，最终回答→structure"""
last_message = state["messages"][-1]
if last_message.tool_calls:
return "tools"
# LLM 认为不需要工具了，进行结构化输出
return "structure"
def generate_structured_output(state: AgentState):
"""结构化输出节点"""
# 用对话历史生成结构化结果
result = structured_model.invoke(state["messages"])
return {"analysis_result": result}
# 构建图
workflow = StateGraph(AgentState)
workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)
workflow.add_node("structure", generate_structured_output)
workflow.add_edge(START, "agent")
workflow.add_conditional_edges("agent", should_continue, ["tools", "structure", END])
workflow.add_edge("tools", "agent")
workflow.add_edge("structure", END)
app = workflow.compile()
# 运行
result = app.invoke({
"messages": [{
"role": "user",
"content": "搜索 LangGraph 的最新功能，然后分析它的优势和不足"
}]
})
analysis = result["analysis_result"]
print(f"结论: {analysis.conclusion}")
print(f"置信度: {analysis.confidence_score}")
print(f"证据: {analysis.evidence}")
plaintext
┌─────────────────────────────────────────────────────────────────────┐
│ 工具调用 + 结构化输出的 Agent 流程 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ START │
│ │ │
│ ▼ │
│ ┌──────────┐ │
│ │ agent │ (LLM + tools，自由推理+调用工具) │
│ └────┬─────┘ │
│ │ │
│ should_continue() │
│ │ │
│ ├── 有 tool_calls ──▶ ┌──────────┐ ──▶ agent (循环) │
│ │ │ tools │ │
│ │ └──────────┘ │
│ │ │
│ └── 无 tool_calls ──▶ ┌──────────────┐ │
│ │ structure │ (with_structured_output) │
│ │ 生成结构化 │ │
│ └──────┬───────┘ │
│ │ │
│ ▼ │
│ END │
│ (返回 AnalysisResult) │
│ │
└─────────────────────────────────────────────────────────────────────┘
九、踩坑记录
坑1：docstring 不写 → LLM 不知道何时调用
python
# ❌ 灾难写法
@tool
def search(q: str) -> str:
"""search"""
return "..."
# 问题：LLM 看到 "search" 三个字母，不知道这是搜互联网还是搜数据库
# 结果：要么乱调用，要么该调的时候不调
# ✅ 正确写法
@tool
def search_web(query: str) -> str:
"""搜索互联网获取实时信息。适用于最新新闻、实时数据查询。
不适用于：纯知识问题（直接回答即可）、数学计算（用 calculator）。
Args:
query: 搜索关键词，应简洁精准
"""
return "..."
核心原则：docstring 就是写给 LLM 的 API 文档，写得越清晰，工具调用越准确。
坑2：工具返回值太大撑爆 State
python
# ❌ 返回整个数据库表
@tool
def query_all_users() -> str:
"""查询所有用户"""
return json.dumps(users) # 10万条记录，几十MB
# 结果：State 中的 messages 列表暴涨，内存溢出或 Token 超限
# ✅ 限制返回大小
@tool
def query_users(limit: int = 20, offset: int = 0) -> str:
"""分页查询用户数据。
Args:
limit: 每页数量，默认20，最大100
offset: 偏移量
"""
limit = min(limit, 100) # 强制上限
users = db.query("SELECT * FROM users LIMIT ? OFFSET ?", limit, offset)
result = json.dumps(users, ensure_ascii=False)
if len(result) > 50000: # 50KB 上限
return result[:50000] + "\n... 结果已截断，请使用分页查询"
return result
坑3：并行工具调用的顺序问题
LLM 可能一次返回多个 tool_calls，ToolNode 会并行执行它们。但**并行工具之间如果有依赖关系，结果可能不符合预期 **：
python
# 场景：LLM 同时调用 file_read 和 db_query
# file_read 的结果可能被 db_query 需要，但并行执行时 db_query 先完成了
# 解决方案1：在 docstring 中明确说明依赖
@tool
def db_query_with_context(sql: str, context: str = "") -> str:
"""查询数据库，可附带上下文信息。
如果需要先读取文件再查询，请先调用 file_read，
然后将读取的内容作为 context 参数传入。
Args:
sql: SQL 查询语句
context: 额外上下文信息（可选）
"""
return "查询结果..."
# 解决方案2：拆成多步（让 LLM 串行调用）
# 在 system prompt 中明确说明：
# "如果有工具间依赖，请逐个调用，不要并行。"
坑4：工具幂等性缺失导致重试重复执行
python
# ❌ 非幂等操作 + 自动重试 = 灾难
@tool
def send_notification(message: str) -> str:
"""发送通知"""
api.send(message) # 如果超时后重试，用户收到两条通知
return "已发送"
# ✅ 幂等设计
@tool
def send_notification(message: str, idempotency_key: str) -> str:
"""发送通知。使用 idempotency_key 防止重复发送。
Args:
message: 通知内容
idempotency_key: 幂等键，相同 key 的请求只执行一次
"""
if already_sent(idempotency_key):
return "通知已发送（跳过重复）"
api.send(message)
mark_sent(idempotency_key)
return "已发送"
坑5：模型不支持 Function Calling 的兼容处理
不是所有模型都支持 tool_calls，需要兼容处理：
python
from langchain_openai import ChatOpenAI
from langchain_core.messages import AIMessage
def safe_bind_tools(model, tools):
"""安全绑定工具，兼容不支持 function calling 的模型"""
try:
model_with_tools = model.bind_tools(tools)
# 测试一下是否能正常工作
test_response = model_with_tools.invoke("test")
return model_with_tools
except (NotImplementedError, ValueError) as e:
print(f"⚠️ 当前模型不支持 function calling: {e}")
print("回退到提示词模式：将工具描述写入 system prompt")
# 回退方案：把工具描述写入 system prompt，让模型用文本格式输出
tool_descriptions = []
for t in tools:
tool_descriptions.append(
f"- {t.name}: {t.description}\n 参数: {t.args_schema.schema()}"
)
system_prompt = (
"你可以使用以下工具。如需调用，请用以下格式回复：\n"
"ACTION: <工具名>\n"
"ARGS: <JSON参数>\n\n"
"可用工具：\n" + "\n".join(tool_descriptions)
)
return model, system_prompt # 返回原模型 + 提示词
# 使用
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
tools = [web_search, calculator]
model_with_tools, fallback_prompt = safe_bind_tools(model, tools)
总结
工具调用是 Agent 从"聊天机器人"变成"能干活的人"的关键。LangGraph 的 ToolNode 把工具执行的脏活累活（并行调用、错误处理、消息格式化）都封装好了，让我们能专注于业务逻辑。
核心要点回顾：
表格
关联阅读：
第23篇 Tool Calling 通用实践 — 工具调用的通用设计原则
第39篇 人在回路 — 危险操作的人工确认机制
第42篇 LangGraph 快速上手 — LangGraph 基础入门
第19篇 状态机设计 — LangGraph 状态机原理
评论区