一、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,在执行调用类的函数时,循环特定次数,直到组件能正常执行结束即暂停,并且在每次循环的过程中,休眠特定的时间,运行流程图如下: