Runnable组件重试机制降低程序错误率
一、LangChain 重试机制深度解析
当构建生产级AI应用时，with_retry() 机制可有效提升系统容错性，典型应用场景包括：
- API调用频率限制时的自动恢复
- 模型服务临时不可用的故障转移
- 网络波动导致的瞬时异常处理
参数详解与配置策略
1. 参数配置表
| 参数名 | 类型 | 默认值 | 作用说明 | 示例值 |
|---|---|---|---|---|
retry_if_exception_type | Tuple[Exception] | None | 指定需要重试的异常类型 | (APIError, TimeoutError) |
wait_exponential_jitter | bool | True | 启用指数退避+随机抖动策略 | False |
stop_after_attempt | int | 3 | 最大重试次数（含初始调用） | 5 |
2. 代码示例
想要让一个 Runnable 组件最多重试 2 次，只需在 with_retry() 函数中传递 stop_after_attempt参数即可，代码：
from langchain_core.runnables import RunnableLambdacounter = -1def func(x):global countercounter += 1print(f"当前的值为 {counter=}")return x / counterchain = RunnableLambda(func).with_retry(stop_after_attempt=2)resp = chain.invoke(2)print(resp)
3. 资料推荐
- 🔗 官方文档参考
- 💡大模型中转API推荐
- ✨中转使用教程
二、with_retry() 函数的运行原理
构建一个新的 Runnable，在执行调用类的函数时，循环特定次数，直到组件能正常执行结束即暂停，并且在每次循环的过程中，休眠特定的时间，运行流程图如下：