★ FEATURED ARTICLE 搞LangChain链子的人应该都有过这种体验本地调得好好的一上生产就抽风——模型服务偶发超时、API返回个5xx、输出格式突然变成乱码整条链直接炸掉。我之前一个模拟项目X跑了个问答链挂着三四个工具调用高峰期一次都稳不住日志里全是Traceback。后来把异常处理从“写满try-except”改成依赖LangChain自带的with_retry()和with_fallbacks()才真正解放出来。这两个方法本质上都是给Runnable加容错能力的装饰器with_retry()让同一个调用失败后自动重试with_fallbacks()让你在主方案挂掉时直接切到备选方案。听起来简单但用的时候有非常多的细节和坑。这篇文章我把它们拆开讲透包括底层机制、参数怎么选、组合顺序怎么定以及我踩过的坑希望对正在搭LangChain应用的同学有参考价值。1. 为什么非得处理LangChain里的异常1.1 LLM调用的失败类型线上跑LLM应用失败的类型远比你想象得多。我一般把它们分成三类错误类型典型表现适合重试适合回退瞬时错误超时、连接断开、429限流、503临时不可用适合适合持续错误401鉴权失败、403拒绝、模型名不存在、请求参数非法不适合重试白搭适合换模型或换方案内容逻辑错误模型正常返回但输出格式不符合解析要求、JSON不合法通常不适合适合换修复策略瞬时错误是with_retry()的主场。因为服务端只是暂时性抖动你隔几秒再试一次往往就成功了。持续错误恰恰相反你重试一百次也是同样的结果白白浪费时间精力。内容逻辑错误最容易被忽略因为程序本身没有抛异常真正抛异常的是解析器——这时候with_fallbacks()能派上大用场。1.2 直接try-except的局限很多人上来就在链子外面套个try-except看起来解决了实际上有几个问题。第一个链子中间哪一步挂了你没法精确定位。LangChain链是多个Runnable串起来的一个超时发生在模型调用还是工具调用外面一个通用except根本分不清。第二个重试逻辑自己写太啰嗦。你要手动判断什么错误值得重试、重试几次、两次之间隔多久、要不要退避。这些代码会污染你的链子逻辑。第三个回退方案更难写。你想实现“模型A挂了用模型B顶上”“解析失败让修复模型重新输出”如果全部手写链子会变成一个巨大的switch-case。这时候你才理解LangChain提供with_retry()和with_fallbacks()的意图把容错逻辑变成链子的一部分而不是链子外面的补丁。它们返回的是新的Runnable可以和其他组件继续拼接这是手写异常处理很难做到的。2. with_retry()让同一个调用多试几次2.1 底层机制with_retry()的底层是tenacity这个库LangChain只是把它封装成了Runnable的方法。你调用model.with_retry(...)后得到的是一个新的Runnable原始模型对象没有变新对象在每次调用时执行原始逻辑捕获到异常后按tenacity的策略决定是否重新执行。这种设计有几个好处第一你不需要改原有链子的结构只需要在某个环节上加一层第二重试逻辑跟业务逻辑解耦第三你可以保留多个不同重试策略的副本比如一个接口重视稳定性、另一个接口重视延迟互不干扰。2.2 实操给模型调用接入重试用起来非常简单示例代码如下from langchain_openai import ChatOpenAI model ChatOpenAI( modelgpt-4o, timeout30.0, ) robust_model model.with_retry( stop_after_attempt3, retry_if_exception_type(TimeoutError, ConnectionError), ) response robust_model.invoke(解释一下依赖注入)这段代码的意思是如果模型调用抛出TimeoutError或ConnectionError最多重试3次。stop_after_attempt3表示总的尝试次数是3不是额外重试3次。很多人把这个参数理解成“失败后再试3次”写出来发现只调用了2次就是这个原因。如果你想对任何异常都重试可以传retry_if_exception_type(Exception,)但我不建议你这么做。因为有些错误如鉴权失败重试没有任何意义还拖慢响应。在线上环境我习惯先让所有错误信息走一遍日志再针对性配置重试类型。2.3 重试参数怎么定with_retry()有四个参数值得仔细调stop_after_attempt尝试次数、wait_exponential_jitter指数退避等待、retry_if_exception_type触发重试的异常类型、before_sleep重试前的回调。指数退避是这个方法内部默认的等待策略。它的核心逻辑是第一次等待时间较短之后每次等待时间翻倍直到达到上限。这样做是为了避免一个服务刚抖动所有客户端同时在几毫秒内涌入重试把服务打挂。用生活类比的话一群人挤一个门口有秩序地分批进比大家一起撞门强得多。我常用的设置是robust_model model.with_retry( stop_after_attempt4, retry_if_exception_type(TimeoutError, ConnectionError, ValueError), before_sleeplambda retry_state: print( f第{retry_state.attempt_number}次调用失败准备重试 ), )before_sleep是调试利器能告诉你当前发生在哪一次尝试、失败原因是什么。我在生产环境一般把print换成日志记录会把异常类型和错误信息写进去。注意这里的等待时间默认从2秒开始最大不超过60秒。如果你的场景是用户实时聊天4次尝试最大等待时间可能达到好几个数级的累计时长用户早就走了。实时场景建议stop_after_attempt2最多尝试两次就快速失败批处理场景可以放到4次甚至5次。2.4 重试的边界不是所有环节都适合加with_retry()。我踩过的坑有两个。第一个坑对整条链而不是单个模型加重试。如果链子里包含写数据库、发消息这类有副作用的步骤重试会导致数据重复写入、用户收到多条消息。只有幂等操作才适合重试。如果确认不了幂等性要么把副作用步骤拆出去要么在链子外面用全局唯一键做去重。第二个坑对格式解析失败加重试。模型第一次输出坏JSON你重试第二次同一个模型的输出大概率还是类似的坏结果。这时候重试的意义不大且每次都产生完整token费用。解析失败应该走修复策略对应的工具是with_fallbacks()。3. with_fallbacks()主方案挂了就换预案3.1 Fallback到底是什么with_fallbacks()比with_retry()更容易被误解成“换一个备用模型再试”实际上它更通用当主Runnable抛出异常时LangChain会自动执行备用Runnable把备用Runnable的执行结果作为最终结果返回。你可以给它传多个备用Runnable按顺序执行第一个成功就用第一个的结果前面的失败就继续尝试后面的。这相当于给链子加了一条“备用通道”。和with_retry()最大的区别是重试是同一个方案重来回退是换一套方案。举个例子你约了人去餐厅吃饭路上堵车迟到了with_retry()是换一条路再开过去with_fallbacks()是直接换一家餐厅。3.2 实操主模型失败自动切换备选模型最常见的用法就是模型层面做高可用切换。比如主模型服务偶尔抽风你可以准备一个备选模型接口from langchain_openai import ChatOpenAI from langchain_anthropic import ChatAnthropic main_model ChatOpenAI(modelgpt-4o, timeout30.0) backup_model ChatAnthropic(modelclaude-3-5-sonnet) chain main_model.with_fallbacks([backup_model]) response chain.invoke(推荐几个微服务拆分方案)当主模型抛出异常时备用模型会接管。注意备用模型同样会超时而且LangChain不会自动给备用模型加重试你需要手动给备用模型也加上重试backup_model ChatAnthropic(modelclaude-3-5-sonnet).with_retry( stop_after_attempt2, )细节决定成败。我只给主模型配了回退没给备用模型配重试结果主模型一挂备用模型也遇到限流直接抛异常等于回退白配。生产环境里每个模型都要做好自己的容错。备用模型的API密钥在初始化时要配好密钥应该走环境变量或密钥管理服务不要硬编码在代码里这点谁踩谁知道。3.3 用Fallback修复“半成品”输出除了换模型with_fallbacks()还能处理内容逻辑错误。这个场景非常常见模型正常返回了但它输出的JSON不合规解析器抛异常。你可以准备一个专门负责“修复”的备用Runnable让输出过一遍修复模型生成合规结果。但这里有个大坑with_fallbacks()触发时传给fallback的输入是原始输入而不是主模型产生的半成品输出。也就是说你没法在fallback里直接拿到主模型生成的“坏JSON”。很多人想用fallback做解析修复结果发现拿不到数据。我自己的做法是把“生成解析”包在一个自定义Runnable里如果解析失败就返回一个包含错误标记和原始文本的数据结构然后在链子的下一层判断是否需要走修复模型。这不算纯with_fallbacks()的功劳但它是LangChain里处理解析异常最稳的一种组合方式from langchain_core.runnables import RunnableLambda def generate_and_parse(state: dict, model, parser): raw model.invoke(state[prompt]) try: parsed parser.invoke(raw) return {ok: True, data: parsed} except Exception as e: return {ok: False, raw: raw, error: str(e)} def repair(state: dict, fix_model): if state.get(ok): return state[data] fixed_text fix_model.invoke(f下面这段输出不符合格式要求请修复成合法JSON\n{state[raw]}) return parser.invoke(fixed_text)这么写虽然多几步但每个阶段都在你控制范围内哪里出错一目了然。比把希望寄托在fallback自动传错信息上靠谱。3.4 多级Fallback可以给with_fallbacks()传多个备用方案LangChain会按顺序尝试。这种做法适合可靠性要求极高的场景。比如chain main_model.with_fallbacks([ backup_model_1, backup_model_2, RunnableLambda(last_resort), ])我在一个成本敏感的问答系统里用过这个模式主模型优先用强一点的模型备选模型用便宜一点的最后一个兜底用规则模板保证用户即使在大规模故障时也能收到一个不完美但可用的回答。这种“降级不降可用性”的思路特别适合线上产品。不过多级fallback也有代价每一次切换都是时间上的累加。如果主模型超时30秒备用模型又超时30秒用户等到的已经是一分钟后的结果。所以设计fallback时要考虑每个环节的响应时间上限甚至可以调低备用模型的超时时间宁可快速失败也不要无限等待。4. 重试与回退的组合4.1 组合的完整形态重试和回退不是互斥的生产环境的链子往往会同时用。问题是怎么组合。常见有两种写法# 写法一先回退再重试 chain_a main_model.with_fallbacks([backup_model]).with_retry( stop_after_attempt3, ) # 写法二先重试再回退 chain_b main_model.with_retry( stop_after_attempt3, ).with_fallbacks([backup_model])这两种写法执行逻辑完全不同。写法一执行的是“带回退的主模型”作为一个整体最多重试3次。每次尝试时如果主模型挂了会先切到备用模型如果备用模型也挂了这次尝试算失败然后整个重试流程从头再来。也就是说主模型和备用模型总共最多各自执行3次。写法二执行的是“主模型自身”先最多重试3次3次全部失败后才切到备用模型备用模型执行完后整个链子结束。备用模型不会跟着重试。从稳定性来说写法一更好因为它给备用模型也分配了足够的机会。但从成本角度写法一可能产生更多请求。我在实际项目里按照业务对成本的敏感度来选择高优业务用写法一成本敏感型后台任务用写法二。4.2 顺序决定行为这里值得一提包装顺序不是语言上的顺序是你调用方法的顺序。with_retry()和with_fallbacks()都返回新的Runnable外层的方法会包裹内层的结果。我建议你在加这些装饰时画一条执行线最外层的方法最后执行最内层的方法最先执行。比如model.with_fallbacks([b]).with_retry()执行顺序是先进入with_retry逻辑再进入with_fallbacks逻辑最后到model本身。这个细节不看文档很容易搞反。我之前写过一版代码本意是想让备用模型被重试结果因为顺序问题备用模型永远只执行一次排查了半天。4.3 成本控制与监控加了重试和回退之后链子的调用量会增加成本随之上涨必须要有监控手段。我自己的习惯是在每个关键环节打印或记录“当前使用哪个模型/第几次尝试”同时记录每次调用的token消耗。LangChain本身支持回调机制可以在自定义回调里拿到token使用信息。配合上before_sleep能清楚看到每一次重试的原因。还遇到过一种情况备用模型在功能上可以替代主模型但它的规范限制不同、返回格式略有差异导致下游解析器依然报错。这种问题不容易从日志里发现需要人工抽查备用模型触发的请求。所以建议给备用模型单独打标签方便事后分析。5. 常见问题与排查技巧实录5.1 常见问题速查表症状可能原因解决方案明明配了重试但只请求一次就失败stop_after_attempt理解错或异常类型不匹配检查异常类型是否在retry_if_exception_type里备用模型从不触发主模型没抛异常只是返回内容有问题检查是否只监听了异常内容校验请单独处理重试导致数据重复整条链包含非幂等操作把有副作用的步骤拆出只对模型调用加重试备用模型报鉴权错误备用模型未正确配置密钥检查环境变量或初始化参数总响应时间过长重试次数太多退避时间太长调低stop_after_attempt设置更短超时fallback拿不到主模型的坏输出with_fallbacks()传的是原始输入用自定义Runnable分别处理生成、解析、修复阶段5.2 排查心得我之前调试一个链子主模型和备用模型都配置了但生产环境依然有较高失败率。后来看日志发现备用模型确实被调用了只是备用模型在原生产环境里也被限流。那是用了前面说的写法二备用模型没有自己的重试策略等于把一切可靠性寄托在一个同样脆弱的模型上。另一个教训是不要盲目扩大重试范围。我把retry_if_exception_type配置为Exception表面上覆盖了所有异常结果某个模型接口参数配错了每次调用都抛异常系统就每次都重试不仅没有解决问题还拖垮了下游服务的负载均衡。重试前先判断这是瞬时错误还是持续错误再决定是否纳入重试。还有一个值得养成的习惯线上的链子一定要把“当前走的是主方案还是备用方案”记录下来。很多问题只有在你确定触发路径之后才查得出来。你可以在每次fallback触发的分支里加一行关键日志这样收到报警时能立刻判断是主模型故障还是备用模型自身问题。最后分享一个我自己的习惯现在我搭LangChain链默认就是在模型调用层做三件事单次调用设置合理的超时、可能瞬时失败的错误类型交给with_retry()、对持续失败或内容异常交给with_fallbacks()。重试次数一般控制在2到3次备用模型一定也配好独立的密钥和超时。没有银弹但把异常处理当成链子的一部分去设计而不是事后打补丁线上稳定性能上一个台阶。如果你也在生产里被LLM不稳定搞得头疼可以试着用这两个方法把你自己的容错策略搭出来。
阅读完成 · 觉得有帮助？
