文章目录with_retry()with_retry()示例with_fallbacks([fallback_function])with_fallbacks()在报错后回复维护中with_retry()和with_fallbacks()的链式操作这一块是异常处理重试和备用增强系统稳定性生产中非常实用面试的时候是加分项。with_retry()报错重试。核心有3个1、哪些异常才重试2、重试几次3、重试频次框架早就帮我们设计好了。1、stop_after_attempt (次数)含义总共尝试多少次包含第一次。建议通常设置为 3 到 5 次。次数太多会拖慢用户体验太少则无法解决偶发故障。2、retry_if_exception_type (异常类型)含义只有遇到特定的错误才重试。建议明确指定 (TimeoutError, ConnectionError) 等避免对逻辑错误进行无意义的重试。3、wait_exponential_jitter (等待策略)含义是否开启“指数退避 抖动”(简称指数抖动)。建议务必开启 (True)。原理第一次失败等 1 秒第二次等 2 秒第三次等 4 秒… 加上随机抖动是为了防止多个请求同时重试导致服务器瞬间崩溃惊群效应。with_retry()示例该代码用到的异常模拟工具类ExceptionMockUtils见其他笔记。代码fromlangchain_core.runnablesimportRunnableLambdafromexception_mock_utilsimportExceptionMockUtils# 引入你刚才写的工具类# 1. 定义一个“捣乱”的函数defunstable_task(input_data):print(f 正在尝试调用... (输入:{input_data}))# 故意抛出超时异常模拟网络故障# 注意LangChain 的重试默认主要针对 Exception 及其子类ExceptionMockUtils.mock_timeout()return成功结果# 2. 包装成 LangChain 的链并配置重试策略chain(RunnableLambda(unstable_task)).with_retry(stop_after_attempt3,# 最多重试 3 次retry_if_exception_type[TimeoutError],wait_exponential_jitterTrue# 开启指数退避这里会看到等待时间越来越长)# 3. 执行try:print(--- 开始测试 ---)chain.invoke(测试数据)exceptExceptionase:print(f❌ 重试耗尽最终报错:{e})输出结果--- 开始测试 --- 正在尝试调用...(输入: 测试数据) 正在尝试调用...(输入: 测试数据) 正在尝试调用...(输入: 测试数据)❌ 重试耗尽最终报错: Connection timed out after30.0seconds解读retry_if_exception_type需要重试的异常类型约定了TimeoutError异常需要重试。当发生该异常时触发重试机制次数为3次并开启指数抖动(指数退避抖动)。输出的结果和预期一致。当然这里如果用logger打印出具体的时间就更好了此处略。with_fallbacks([fallback_function])报错后使用备用服务。要求fallback_function指代的对象必须实现Runnable接口。例如本文示例RunnableLambda就是Runnable的实现类。with_fallbacks()在报错后回复维护中这里用千问plus故意写错base_url最后的-aaaa是额外加上的这样就可以触发异常。代码fromlangchain_openaiimportChatOpenAIfromlangchain_core.runnablesimportRunnableLambdaimportos API_KEYos.getenv(DASHSCOPE_API_KEY,YOUR API KEY)print(f 代码读取到的API_KEY{API_KEY})primary_modelChatOpenAI(api_keyAPI_KEY,# 替换为你的实际 Key (sk-...)base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1-aaaa,modelqwen-plus)# 2. 定义一个“假模型”函数# 这个函数接收输入虽然这里没用到返回一个固定的字符串defmock_fallback_logic(inputs):print(⚠️ 主模型挂了触发人工兜底逻辑...)# 这里可以写任何逻辑查数据库、返回固定话术、甚至调用另一个 APIreturn系统正在维护中请稍后再试或者您可以拨打客服热线 12345。# 3. 包装成 Runnable# 注意必须包装成 Runnable因为 with_fallbacks 只认这个fallback_functionRunnableLambda(mock_fallback_logic)# 4. 组合# 列表里放的是我们刚才包装的函数chainprimary_model.with_fallbacks([fallback_function])# 5. 测试# 如果 GPT-4 调用失败你会直接得到函数返回的字符串responsechain.invoke(帮我写首诗)print(response)# 输出: 系统正在维护中...输出结果 代码读取到的API_KEYsk-e49bdf8d737549ffb6269b63bcff8a7a ⚠️ 主模型挂了触发人工兜底逻辑...系统正在维护中请稍后再试或者您可以拨打客服热线12345。with_retry()和with_fallbacks()的链式操作重试和备用还可以链式操作简直完美。在定义Runnable实现类的时候直接链式操作即可。代码unstable_function()为自定义逻辑类。 retryable_chainRunnableLambda(unstable_function).with_retry(retry_if_exception_type(TimeoutError,),stop_after_attempt3,wait_exponential_jitterTrue)