第08章：中间件(6)

3.5 ToolRetryMiddleware中间件基于指数退避算法设置工具调用失败时的重试策略。指数退避Exponential Backoff 的核心思想就是当某个操作失败通常是网络请求、API 调 用或数据库连接时系统不会立刻重试也不会每次都等待相同的固定时间而是让每一次重试 的延迟时间按指数级增长。为什么不直接重试想象一下某个热门网站的服务器因为瞬间流量太大比如抢票或秒杀崩溃了。如果所有失败的 客户端都立刻或每隔1秒就重试一次这无异于对已经瘫痪的服务器进行了一场持续的 DDoS分 布式拒绝服务攻击服务器可能永远也缓不过来。jitter是为了避免大量工具的重试请求集中在固定的时间点引入抖动。 假设按照策略两次工具调用请求的时间间隔应为10秒加入抖动后可能为8.9秒也可能为10.2 秒。举例1带抖动from langchain.chat_models import init_chat_model from dotenv import load_dotenv import os # 从 .env 文件中加载环境变量 load_dotenv(overrideTrue) model init_chat_model( modeldeepseek-v4-flash, model_providerdeepseek, extra_body{thinking: {type: disabled}}, )from langchain.agents import create_agent from langchain.agents.middleware import ToolRetryMiddleware from langchain.messages import HumanMessage from langchain.tools import tool import datetime def write_times(s): 将每次工具调用的时间戳和间隔写入本地文件方便观察退避策略 with open(call_times_with_jitter.txt, a, encodingutf-8) as f: f.write(s \n) count 1 start_time None tool def get_weather(city: str): 查询指定城市天气 global count global start_time interval 0 current_time datetime.datetime.now() if not start_time: interval 0 else: # 计算当前调用与上一次调用之间的时间差秒 interval (current_time - start_time).total_seconds() start_time current_time res_str ( f第 {count} 次调用当前时间{start_time} f和上次调用间隔 {interval} 秒 ) count 1 # 记录日志 write_times(res_str) # 故意抛出 TimeoutError以此触发中间件的重试机制 raise TimeoutError(Not Implemented) agent create_agent( modelmodel, tools[get_weather], middleware[ # ToolRetryMiddleware 用于捕获工具执行中的异常并自动重试 ToolRetryMiddleware( max_retries6, # 最大重试次数不包含初始的那次调用一共最多调用 1 6 7 次 backoff_factor2.0, # 指数退避因子每次重试等待时间乘以 2 initial_delay1.0, # 第一次重试前的初始等待时间1 秒 max_delay10.0, # 最大等待延迟上限防止指数增长无限大限制在 10 秒 jitterTrue, # 开启抖动在等待时间中加入随机性防止并发请求时出现“惊群效应” retry_on(TimeoutError,), # 仅针对捕获到特定的 TimeoutError 异常时才触发重试 on_failurecontinue, # 达到最大重试次数依然失败时 # continue 表示将错误信息包装后塞回对话历史 # 让大模型知道失败了并继续决策 ), ], ) response agent.invoke( { messages: [ HumanMessage(今天北京天气如何) ] } ) # 1. 你的提问 # 2. AI 决定调用工具 # 3. 重试失败后的错误反馈 # 4. AI 最终给出的兜底回复 for msg in response[messages]: msg.pretty_print() Human Message 今天北京天气如何 Ai Message Tool Calls:get_weather (call_00_oF64KTaTyrWTEaqlkyzt1467)Call ID: call_00_oF64KTaTyrWTEaqlkyzt1467Args:city: 北京 Tool Message Name: get_weatherTool get_weather failed after 7 attempts with TimeoutError: Not Implemented. Please try again. Ai Message 很抱歉目前查询北京天气的服务暂时不可用无法获取到实时天气信息。建议您可以尝试以下方式获取天气1. **稍后重试** — 稍后再来问我一次可能服务会恢复。2. **使用天气类App** — 如中国天气网、墨迹天气、彩云天气等。3. **搜索引擎** — 直接在浏览器或搜索引擎中搜索北京天气。如果您有其他问题需要帮助随时告诉我call_times_with_jitter.txt 文件内容为了精确查看工具调用的时间间隔以理解各项参数的含义我们将工具调用时间写入本地文件举例2无抖动agent create_agent( modelmodel, tools[get_weather], middleware[ # ToolRetryMiddleware 用于捕获工具执行中的异常并自动重试 ToolRetryMiddleware( max_retries6, # 最大重试次数不包含初始的那次调用一共最多调用 1 6 7 次 backoff_factor2.0, # 指数退避因子每次重试等待时间乘以 2 initial_delay1.0, # 第一次重试前的初始等待时间1 秒 max_delay10.0, # 最大等待延迟上限防止指数增长无限大限制在 10 秒 jitterFalse, # 关闭抖动意味着重试机制从“随机化的指数退避”退化成了“严格固定的指数退避” retry_on(TimeoutError,), # 仅针对捕获到特定的 TimeoutError 异常时才触发重试 on_failurecontinue, # 达到最大重试次数依然失败时 # continue 表示将错误信息包装后塞回对话历史 # 让大模型知道失败了并继续决策 ), ], ) Human Message 今天北京天气如何 Ai Message Tool Calls:get_weather (call_00_4BjN5gaS4orCliF50RkM8452)Call ID: call_00_4BjN5gaS4orCliF50RkM8452Args:city: 北京 Tool Message Name: get_weatherTool get_weather failed after 7 attempts with TimeoutError: Not Implemented. Please try again. Ai Message 很抱歉查询北京的天气时出现了错误暂时无法获取到天气信息。这可能是服务暂时不可用或网络问题导致的。建议您可以1. **稍后重试** — 等一会儿再查询一次。2. **查看其他天气平台** — 比如中国天气网、手机自带天气应用等获取实时天气信息。3. **告诉我其他需求** — 如果您需要了解其他城市的天气我也可以再尝试帮您查询。很抱歉给您带来不便call_times_without_jitter.txt文件内容自动重试失败的工具调用。第 1 次重试retry_number1: 等待 ~ 1.0 * (2.0 ** 1) 2.0 秒第 2 次重试retry_number2: 等待 ~ 1.0 * (2.0 ** 2) 4.0 秒第 3 次重试retry_number3: 等待 ~ 1.0 * (2.0 ** 3) 8.0 秒也就是说等待时间以指数方式增长 —— 每失败一次下次再试之前等待更长时间。如果你把 backoff_factor 0就意味着不使用指数增长重试之间始终用固定的 initial_delay。将 jitter 从 True 改为 False 关闭抖动意味着重试机制从“随机化的指数退避”退化成了“严格固 定的指数退避”。为了更直观理解看一下这两种状态下的核心区别1. 理论上的等待时间对比在这段代码中你设置了 initial_delay1.0 初始延迟 1 秒、 backoff_factor2.0 倍数是 2以 及 max_delay10.0 最大延迟 10 秒。当工具持续报错时关闭抖动 jitterFalse 与开启抖动 jitterTrue 的等待延迟Interval对 比如下 现象结论关闭抖动后查看生成的 call_times_with_jitter.txt 日志你会发现输出的 interval 数字会极 其精准地趋近于 1.0 、 2.0 、 4.0 、 8.0 、 10.0 、 10.02. 为什么要引入 Jitter抖动关闭它会有什么问题在单用户、单并发的测试环境下关闭 jitter没有任何副作用甚至能让等待时间非常规律、可预测但在高并发的生产环境中关闭 jitter会引发灾难性的 “惊群效应Thundering Herd Problem” 没有 Jitter 的惨剧 jitterFalse 假设某刻天气 API 服务突然宕机了 1 秒。此时刚好有 1000 个用户同时发起了查询。因为这 1000 个请求同时失败并且它们都严格死板地等待 1 秒、2 秒、4 秒……意味着在第 1 秒、第 3 秒、第 7 秒的那个精准的时间点上这 1000 个请求会整整齐齐地再次 同时轰炸服务器。刚刚复活的服务器瞬间又被这波整齐的峰值流量压垮形成恶性循环。引入 Jitter 的优势 jitterTrue 通过给重试时间加上随机性这 1000 个请求会在 秒、 秒的区间内均匀地错开削峰 填谷。流量被平摊到了整条时间轴上服务器就能轻松地分批处理完这些请求。总结jitterFalse 你当前的代码重试间隔死板、精准、可预测。适合本地调试、测试重试逻辑是 否生效。jitterTrue 重试间隔随机、错开、更安全。适合线上生产环境防止把下游第三方 API 或数据 库冲垮。3.6 ModelRetryMiddleware中间件模型调用失败时重试策略和工具调用的重试一样都是基于指数退避算法。因此本节案例不再重点观察指数退避算法而是测试不同的退出模式。举例1继续运行from langchain.agents import create_agent from langchain.agents.middleware import ModelRetryMiddleware from langchain.messages import HumanMessage from dotenv import load_dotenv load_dotenv(overrideTrue) agent create_agent( modeldeepseek-cat, middleware[ ModelRetryMiddleware( max_retries6, backoff_factor2.0, initial_delay1.0, max_delay10.0, on_failurecontinue, jitterFalse, ), ], ) response agent.invoke( { messages: [ HumanMessage(你好) ] } ) for msg in response[messages]: msg.pretty_print() Human Message 你好 Ai Message Model call failed after 7 attempts with BadRequestError: Error code: 400 - {error: {message: The supported API model names are deepseek-v4-pro or deepseek-v4-flash, but you passed deepseek-cat., type: invalid_request_error, param: None, code: invalid_request_error}}举例2抛异常agent create_agent( modeldeepseek-cat, middleware[ ModelRetryMiddleware( max_retries6, backoff_factor2.0, initial_delay1.0, max_delay10.0, on_failureerror, jitterFalse, ), ], )BadRequestError: Error code: 400 - {error: {message: The supported API model names are deepseek-v4-pro or deepseek-v4-flash, but you passed deepseek-cat., type: invalid_request_error, param: None, code: invalid_request_error}}During task with name model and id 551a2150-76cf-282e-a922-033ebff3c1653.7 LLMToolEmulator中间件某些情况下工具尚未开发完成我们希望先测试工具调用可以用LLM tool emulator模拟工具。from langchain.chat_models import init_chat_model from dotenv import load_dotenv import os # 从 .env 文件中加载环境变量 load_dotenv(overrideTrue) model_out init_chat_model( modeldeepseek-v4-flash, model_providerdeepseek, extra_body{thinking: {type: disabled}}, )from langchain.chat_models import init_chat_model from dotenv import load_dotenv import os # 从 .env 文件中加载环境变量 load_dotenv(overrideTrue) model_in init_chat_model( modeldeepseek-v4-pro, model_providerdeepseek, extra_body{thinking: {type: disabled}}, )from langchain.agents import create_agent from langchain.agents.middleware import LLMToolEmulator from langchain.messages import HumanMessage from langchain.tools import tool tool def get_weather(city: str): 查询指定城市天气 return f{city}今天天气晴朗 agent create_agent( modelmodel_out, tools[get_weather], middleware[ LLMToolEmulator( modelmodel_in, ), ], ) response agent.invoke( { messages: [ HumanMessage(今天北京天气如何) ] } ) for msg in response[messages]: msg.pretty_print() Human Message 今天北京天气如何 Ai Message Tool Calls:get_weather (call_00_EZjPqBhtREz5IzCpVBwg7754)Call ID: call_00_EZjPqBhtREz5IzCpVBwg7754Args:city: 北京 Tool Message Name: get_weather{city: 北京, temperature: -3°C, weather: 晴, humidity: 28%, wind: 西北风 3级, update_time: 2025-03-19T08:00:00Z} Ai Message 北京今天的天气情况如下- **天气**晴 ☀️- **气温**-3°C- **湿度**28%- **风力**西北风 3级今天北京天气晴朗但气温较低需要注意保暖出门记得穿厚一些的衣服哦