LangChain Tool 异常处理实战

LangChain Tool 异常处理实战 | 极客日志

from langchain_core.tools import StructuredTool, ToolException

# 1. 定义一个可能出错的工具（模拟搜索失败）
def search(query: str) -> str:
    """执行搜索查询，返回结果"""
    # 模拟异常：比如搜索结果为空、网络超时
    raise ToolException(f"搜索失败：未找到'{query}'的相关结果（可能网络超时）")

# 2. 创建工具时开启自动异常处理
search_tool = StructuredTool.from_function(
    func=search,
    name="SearchTool",
    description="用于搜索信息的工具",
    handle_tool_error=True  # 关键：自动处理异常
)

# 3. 调用工具（故意触发异常）
try:
    result = search_tool.invoke({"query": "不存在的关键词 12345"})
    print("工具返回结果：", result)
except Exception as e:
    # 注意：开启 handle_tool_error 后，不会走到这里，异常已被工具内部处理
    print("捕获到异常：", e)

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

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，今日调用次数已达上限 
工具返回结果： 今日股票查询次数已用完~ 明天再试吧，或查看大盘指数替代

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": "上海"}))

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

LangChain Tool 异常处理实战

LangChain Tool 异常处理实战

一、工具调用也会'掉链子'：不处理异常，智能体就'罢工'

二、ToolException：给工具错误'定规矩'的关键

三、三种异常处理方式：从'简单兜底'到'自定义降级'

方式一：自动处理（handle_tool_error=True）——最简单的'兜底'

代码示例：

输出结果：

适用场景：

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

代码示例：

输出结果：

适用场景：

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

代码示例：

输出结果：

适用场景：

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

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

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

4.3 关键亮点：

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

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

更多推荐文章

相关免费在线工具

LangChain Tool 异常处理实战

LangChain Tool 异常处理实战

一、工具调用也会'掉链子'：不处理异常，智能体就'罢工'

二、ToolException：给工具错误'定规矩'的关键

三、三种异常处理方式：从'简单兜底'到'自定义降级'

方式一：自动处理（handle_tool_error=True）——最简单的'兜底'

代码示例：

输出结果：

适用场景：

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

代码示例：

输出结果：

适用场景：

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

代码示例：

输出结果：

适用场景：

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

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

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

4.3 关键亮点：

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

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

微信扫一扫，关注极客日志

更多推荐文章

相关免费在线工具