Agent工具调用中的错误处理 重试机制降级处理

重试机制重试机制Retry是一种软件设计模式它允许系统在检测到某个操作失败时按照预定义的策略如次数、间隔时间等自动重新尝试执行该操作提高容错能力并保障系统的稳定性。详细内容可阅读tenacity文档https://tenacity.readthedocs.io/en/latest/为什么需要重试机制在设计agent时我们往往会调用外部api在这个过程中往往很出现很多暂时性错误网络抖动包括连接超时、路由器临时阻塞、服务器无响应等等api临时超时目标接口瞬间负载过高报错api调用的请求频率到达供应商限制等等llm错误llm输出格式不对导致解析失败以上这些错误都是暂时性的设计Retry的目的就是为了让模型自动重试处理这些错误而不需要人工处理如果没有重试机制那么任何微小的网络波动都会导致前端直接弹窗报错迫使任务中断每次中断都需要用户重新手动输入上一个问题大大影响用户体验。常见的重试策略有立即重试在发生错误后立即发起重试等间隔重试在发生错误后等待固定间隔如2s再发起重试指数退避每次错误发生后等待时间翻倍1s2s4s这样做能够缓解服务器压力最大次数限制设置重试上限比如5次防止死循环浪费tokenLangchain内置重试RetryWithErrorOutputParserRetryWithErrorOutputParser专门用于解析失败时的重试——当 LLM 输出格式不符合预期时自动将错误信息传回给模型让其重新生成。from langchain.output_parsers import PydanticOutputParser from langchain.output_parsers import RetryWithErrorOutputParser parser PydanticOutputParser(pydantic_objectMyModel) retry_parser RetryWithErrorOutputParser( parserparser, retry_chainllm, max_retries3 ) # 当解析失败时会自动将错误信息发给 LLM 要求重新生成 result retry_parser.parse_with_retry(bad_llm_output)参数parser解析器用于解析输出结果max_retries解析失败后尝试重新解析的最大次数retry_chain重试链让这个chain重新尝试生成补全内容retry_chain可以包含数据处理Prompt LLM OutputParser 的完整组合tenacity 库——重试策略tenacity 是 Python 生态中最流行的通用重试库。它通过装饰器Decorator的方式让你能以极简的代码为任何可能失败的函数增加强大的重试逻辑重试几次重试等待时间固定 / 指数退避什么错误才重试重试失败后做什么降级重试时打印日志停止条件安装tenacity库pip install tenacity基本格式retry(重试逻辑条件) def do_something_unreliable(参数): 函数逻辑 return基础重试retry def never_gonna_give_you_up(): print(Retry forever ignoring Exceptions, dont wait between retries) raise Exceptionraise Exception是一个触发器在程序没有出错时我们在这里主动抛出一个错误目的是观察重试策略。在没有任何重试逻辑参数的条件下retry默认进行无限次、无间隔的重试。重试逻辑参数stop重试停止条件stopstop_after_attempt(5)在尝试5次过后放弃stopstop_after_delay(10)最多重试10秒stop(stop_after_delay(10) | stop_after_attempt(5))组合条件最多尝试5次和最多重试10秒wait重试等待策略waitwait_fixed(2)每次重试都等待2s避免快速轮询waitwait_random(min1, max2)加入随机值重试时间在1到2s内取值waitwait_exponential(multiplier1, min4, max10)指数退避min规定了最小重试时间max规定了重试时间上限当多个进程争夺共享资源时指数级增长有助于减少碰撞。waitwait_random_exponential(multiplier1, max60)随机指数退避在0到2^{重试次数-1}之间选择一个随机数同时确保这个随机数不能大于60retry什么情况下需要重试retryretry_if_exception_type(IOError)只有当出现IOError这类错误时才会重试retryretry_if_exception_type((TimeoutError, ConnectionError))只重试这两类错误retryretry_if_not_exception_type(ClientError)除了ClientError其他错误都重试reraise错误处理reraiseTrue如果你想在控制台的最后一行看到你的报错信息Exception: 原始错误信息而不是在中间部分那么你就把reraise设置为Truebefore_sleep在报错过后打印日志记录before_sleepbefore_sleep_log(logging.getLogger(), logging.WARNING)在每次重试等待之前自动记录一条警告级别的日志让你能够追踪重试的发生时间和原因import logging import sys #日志配置 logging.basicConfig(streamsys.stderr, levellogging.DEBUG) logger logging.getLogger(__name__) retry(stopstop_after_attempt(3), before_sleepbefore_sleep_log(logger, logging.DEBUG)) def raise_my_exception(): raise MyException(Fail)中间件ToolRetryMiddlewareToolRetryMiddleware是Langchain中专门为工具调用创建的中间件ToolRetryMiddleware默认是指数退避范围只对 Agent 的 tools 生效不影响节点 / LLM 调用配置在创建 Agent 时全局注册作用于全部或指定工具能力指数退避、jitter、按工具过滤、重试后返回 ToolMessage 等具体参数max_retries最大重试次数tools可选工具列表应用于重试逻辑retry_on对某一组错误类型进行重试backoff_factor重试时间倍数跟multplier作用相同initial_delay第一次重试前的延迟max_delay两次重试之间的最大延迟jitter是否加入随机抖动代码示例from langchain.agents.middleware import ToolRetryMiddleware # 工具A网络API重试3次、指数退避 retry_api ToolRetryMiddleware( max_retries3, backoff_factor2.0, initial_delay1.0, tools[search_api] # 只给 search_api 用 ) # 工具B本地脚本只重试1次、固定延迟 retry_local ToolRetryMiddleware( max_retries1, backoff_factor0, # 固定延迟 initial_delay0.5, tools[local_script] # 只给 local_script 用 ) # 工具C支付类完全不重试不加入任何中间件 agent create_agent( tools[search_api, local_script, payment_tool], middleware[retry_api, retry_local] # 两个中间件并行 )降级策略降级策略是工具经过多次重试仍然失败不再继续报错、不再卡死流程主动启用兜底方案保证整个 Agent 任务能继续往下走的方法。它是重试多次不成功后的保险方案也是在主要功能无法使用时使用简要功能或者牺牲一部分性能的权宜之计。在agent中我们通常依赖很多外部服务api调用失败、搜索引擎超时、数据库限流每一个外部服务都存在相应的潜在问题路径上某一个部分出了问题如果写了降级策略我们可以让agent继续工作提高工作流的稳定性异常捕获异常捕获是降级的触发器只有捕获了异常系统才知道需要切换为降级方案# 降级的基本模式try:result primary_service() # 尝试主服务except Exception as e: # 捕获异常result fallback_service() # 触发降级try-except本质上是异常拦截器他可以阻止异常向其他地方传递但是它并不具备异常处理的能力在捕获了异常过后我们还需要写具体的处理逻辑为什么需要捕获异常如果不处理异常异常会逐层向上传播如果函数a调用了函数b函数b出现的异常会向上传递到函数a直至遇到异常捕获程序如果没有异常捕获在Langgraph中异常就会蔓延到整个图。如果异常被捕获了调度器就能决定是重试、跳过、还是使用降级策略。模型降级ModelFallbackMiddlewareModelFallbackMiddleware是Langchain的一个中间件在主模型中引用专门负责在错误时回归其他模型直至所有可用模型全部耗尽。#基本格式ModelFallbackMiddleware(self,first_model*additional_models)from langchain.agents import create_agent from langchain.agents.middleware import ModelFallbackMiddleware agent create_agent( modelgpt-4o, # 主模型 middleware[ ModelFallbackMiddleware( gpt-4o-mini, # 备用1 claude-3-5-sonnet # 备用2 ) ] )工具降级当agent依赖的主工具或者api失效时我们可以用平替工具进行替换或者使用模型自带的知识实在不行输出一个“该功能暂时不可用”的提醒而不是直接报错。from langchain.tools import tool tool async def robust_search(query: str) - str: 带降级的搜索工具 # 一级专业搜索 try: return await tavily_search(query) except Exception: pass # 二级通用搜索 try: return await google_search(query) except Exception: pass # 三级本地知识 return f无法搜索到{query}的实时结果请稍后重试数据降级实时数据获取失败时函数返回缓存数据或者默认数据import logging import time logger logging.getLogger(DataFallback) class DataService: def __init__(self): self.cache {user_1: {name: 张三, level: 10}} # 模拟本地缓存 async def get_user_data(self, user_id: str): try: # 1. 尝试从主数据库/实时 API 获取模拟可能失败 logger.info(f正在请求实时数据库: {user_id}) result await self._fetch_from_remote_db(user_id) # 成功后更新缓存 self.cache[user_id] result return result, REAL_TIME except Exception as e: # 2. 降级逻辑使用缓存数据 logger.warning(f实时数据获取失败: {e}。执行数据降级返回缓存数据。) fallback_data self.cache.get(user_id) if fallback_data: return fallback_data, CACHED (DEGRADED) else: # 3. 最终兜底返回静态默认值 return {name: Anonymous, level: 1}, DEFAULT_VALUE功能降级在一些非核心功能日志、埋点、通知失败时我们可以选择跳过当前工具环节而避免异常阻塞工作流的后续进程