LangChain Tool 异常处理实战

输出结果：

工具返回结果： SearchTool tool failed with error: SearchTool: 搜索失败：未找到'不存在的关键词 12345'的相关结果（可能网络超时） 

适用场景：

快速开发、不需要自定义错误信息的场景，让工具自己搞定异常。

方式二：固定错误信息（handle_tool_error="提示文本"）——给用户'说人话'

如果觉得自动生成的错误信息太技术化，可以用 handle_tool_error 直接指定一句友好的提示（比如'网络有点卡，换个关键词试试？'），让用户更容易理解。

代码示例：

from langchain_core.tools import StructuredTool, ToolException

def weather_query(city: str) -> str:
    """查询指定城市的天气"""
    # 模拟异常：城市名不合法（比如'火星市'）
    if city not in ["北京", "上海", "广州"]:
        raise ToolException(f"无效城市：{city}（不在支持的城市列表中）")
    return f"{city}今天晴，25℃"

# 创建工具时指定固定错误提示
weather_tool = StructuredTool.from_function(
    func=weather_query,
    name="WeatherTool",
    description="查询天气的工具，参数为城市名",
    handle_tool_error="抱歉，暂时查不到这个城市的天气~ 试试北京、上海或广州吧！"  # 自定义友好提示
)

# 调用工具（传入无效城市）
result = weather_tool.invoke({"city": "火星市"})
print("工具返回结果：", result)

输出结果：

工具返回结果： 抱歉，暂时查不到这个城市的天气~ 试试北京、上海或广州吧！ 

适用场景：

需要给用户友好反馈的场景，隐藏技术细节，只说用户能懂的话。

方式三：自定义处理函数——灵活'降级'，智能应对

如果想更灵活（比如记录错误日志、自动重试、切换备用工具），可以定义一个异常处理函数，让它根据错误类型做不同操作——这是生产环境最常用的方式。

代码示例：

from langchain_core.tools import StructuredTool, ToolException
import time

# 1. 定义可能出错的工具（模拟 API 调用次数超限）
def stock_query(code: str) -> str:
    """查询股票价格，参数为股票代码（如 600036）"""
    # 模拟异常：每日调用次数超限
    raise ToolException(f"股票查询失败：{code}，今日调用次数已达上限")

# 2. 定义自定义异常处理函数（核心）
def handle_stock_error(error: ToolException) -> str:
    """
    处理股票查询工具的异常
    1. 记录错误日志（实际项目中可写入文件或日志系统）
    2. 根据错误类型返回降级方案
    """
    # 记录错误（模拟日志）
    error_msg = f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 错误：{str(error)}"
    print("【错误日志】", error_msg)
    
    # 根据错误内容返回不同的降级提示
    if "调用次数已达上限" in str(error):
        return "今日股票查询次数已用完~ 明天再试吧，或查看大盘指数替代"
    else:
        return f"查询失败：{str(error)}，请检查股票代码是否正确"

# 3. 创建工具时绑定自定义处理函数
stock_tool = StructuredTool.from_function(
    func=stock_query,
    name="StockTool",
    description="查询股票价格的工具",
    handle_tool_error=handle_stock_error  # 绑定自定义函数
)

# 4. 调用工具（触发次数超限异常）
result = stock_tool.invoke({"code": "600036"})
print("工具返回结果：", result)

输出结果：

【错误日志】 [2025-11-07 15:30:00] 错误：StockTool: 股票查询失败：600036，今日调用次数已达上限 
工具返回结果： 今日股票查询次数已用完~ 明天再试吧，或查看大盘指数替代 

适用场景：

复杂业务场景，需要根据错误类型做精细化处理（日志、重试、降级方案等）。

四、实战：给天气查询工具加'全链路异常防护'

结合之前学的天气查询工具，我们来做一个实战——给它加上完整的异常处理，应对'网络超时''城市不存在''API 密钥过期'三种常见错误。

4.1 完整代码（带详细注释）

import requests
from langchain_core.tools import StructuredTool, ToolException
from pydantic import BaseModel, Field

# 1. 定义参数模型（校验城市名格式）
class WeatherInput(BaseModel):
    city: str = Field(description="城市名称，如北京、上海")

# 2. 定义天气查询工具（可能抛出多种异常）
def get_weather(city: str) -> str:
    """调用天气 API 查询实时天气"""
    # 模拟 API 密钥（实际项目中用环境变量存储）
    api_key = "无效的密钥"
    # 故意用无效密钥模拟权限错误
    
    # 异常 1：城市名格式错误（非中文）
    if not all('\u4e00' <= c <= '\u9fff' for c in city):
        raise ToolException(f"参数错误：城市名'{city}'必须是中文（如北京）")
    
    # 调用天气 API（模拟）
    try:
        url = f"https://api.weather.com/{city}?key={api_key}"
        response = requests.get(url, timeout=5)  # 超时 5 秒
        
        # 异常 2：API 返回错误（如密钥无效）
        if response.status_code != 200:
            raise ToolException(f"API 错误：状态码{response.status_code}（可能密钥过期）")
        
        # 正常返回天气（模拟）
        return f"{city}实时天气：晴，22℃，微风"
        
    # 异常 3：网络超时
    except requests.exceptions.Timeout:
        raise ToolException(f"网络超时：查询'{city}'天气时连接超时，请检查网络")

# 3. 定义自定义异常处理函数（全链路防护）
def handle_weather_error(error: ToolException) -> str:
    """根据不同错误类型返回处理方案"""
    error_str = str(error)
    
    # 处理参数错误
    if "参数错误" in error_str:
        return f"输入有误：{error_str.split('：')[1]}，请重新输入中文城市名~"
    # 处理网络超时
    elif "网络超时" in error_str:
        return "网络有点慢呢~ 请稍后再试，或换个城市查询"
    # 处理 API 错误（如密钥问题）
    elif "API 错误" in error_str:
        return "系统暂时无法查询天气，工程师正在抢修中，抱歉啦~"
    # 其他未知错误
    else:
        return "查询遇到小问题，请稍后重试~"

# 4. 创建带异常处理的工具
weather_tool = StructuredTool.from_function(
    func=get_weather,
    name="WeatherTool",
    description="查询中文城市的实时天气",
    args_schema=WeatherInput,  # 绑定参数模型
    handle_tool_error=handle_weather_error  # 绑定异常处理函数
)

# 5. 测试三种异常场景
print("=== 测试 1：城市名非中文 ===")
print(weather_tool.invoke({"city": "beijing"}))
print("\n=== 测试 2：网络超时（模拟） ===")
# （代码中已模拟超时，实际可通过断网测试）
print(weather_tool.invoke({"city": "北京"}))
print("\n=== 测试 3：API 密钥无效 ===")
print(weather_tool.invoke({"city": "上海"}))

4.2 运行结果（模拟三种异常）

=== 测试 1：城市名非中文 === 
输入有误：城市名'beijing'必须是中文（如北京），请重新输入中文城市名~ 
=== 测试 2：网络超时（模拟） === 
网络有点慢呢~ 请稍后再试，或换个城市查询 
=== 测试 3：API 密钥无效 === 
系统暂时无法查询天气，工程师正在抢修中，抱歉啦~ 

4.3 关键亮点：

1. 分层异常捕获：从参数校验、网络请求到 API 响应，全链路覆盖；
2. 用户友好提示：不同错误返回不同的引导话术，避免技术术语；
3. 可扩展性：后续新增错误类型（如'城市未开通服务'），只需在处理函数里加分支。

五、生产环境最佳实践：异常处理的'三要素'

在实际项目中，处理工具异常不能只靠 ToolException，还要结合这三个要点：

1. 日志要全：异常发生时，记录'哪个工具、什么参数、什么时间、错误详情'，方便排查（比如用 logging 模块写入日志文件）；
2. 降级要柔：不能直接报错，要给用户可操作的替代方案（比如'查不了实时天气，要不要看明天的预报？'）；
3. 智能重试：对网络超时等临时错误，可在处理函数里自动重试 1-2 次（比如用 tenacity 库实现重试逻辑）。

六、小结：异常处理是智能体'成熟度'的试金石

这篇我们学了 LangChain Tool 工具的异常处理——用 ToolException 统一捕获错误，通过三种方式（自动处理、固定提示、自定义函数）实现兜底降级，让智能体在工具'掉链子'时也能优雅应对。

核心就是一句话：好的智能体不仅要'会干活'，更要'会处理麻烦'。就像优秀的员工，不仅能完成任务，遇到问题还能想办法解决，而不是直接摆烂。

结合之前学的 Agent，我们还能让智能体根据错误自动调整策略（比如天气 API 超时了，自动切换备用数据源），这就是更高级的'容错 Agent'。