个人主页 for_ever_love__ 欢迎各位大佬莅临其他栏目: 大模型开发从0到1 其他栏目: iOS项目总结大全 其他栏目: 我想学python了 其他栏目: iOS UI 文章目录LangChain 降级、重试与超时让服务在模型抽风时还能用一、先分清三种故障二、第一层重试with_retry2.1 用 with_retry 更灵活2.2 重试的三个坑三、第二层降级with_fallbacks3.1 换模型3.2 换供应商更有价值3.3 降级到非模型方案3.4 什么时候会触发降级四、第三层超时4.1 设在哪里4.2 流式场景的超时五、三层的正确组合顺序六、监控别等用户告诉你挂了七、完整示例八、小结LangChain 降级、重试与超时让服务在模型抽风时还能用模型 API 会限流、会超时、会 5xx。你的服务不能跟着一起挂。这篇讲三层防护重试 → 降级 → 超时以及它们的组合顺序——顺序错了等于白做。一、先分清三种故障故障特征处理瞬时故障429 限流、503、网络抖动重试很快就好持续故障某个模型整体不可用降级换模型/换服务慢不报错但很慢超时别耗尽资源三种要分别处理混在一起会互相干扰。二、第一层重试with_retryfromlangchain_openaiimportChatOpenAI llmChatOpenAI(modelgpt-4o-mini,max_retries2,# 构造参数自动重试 2 次timeout30,)这是最省事的方式。LangChain内置指数退避只重试可重试错误限流/5xx/网络参数错误不会重试重试也没用。2.1 用 with_retry 更灵活chain(prompt|llm|parser).with_retry(stop_after_attempt3,wait_exponential_jitterTrue,# 指数退避 抖动retry_if_exception_type(TimeoutError,ConnectionError),)with_retry可以套在链的任意一层而不只是 LLM# 只给检索这一步加重试retrievalretriever.with_retry(stop_after_attempt3)chain{context:retrieval|format_docs,question:RunnablePassthrough()}|...2.2 重试的三个坑坑一重试次数相乘llmChatOpenAI(max_retries3)# 3 次chain(prompt|llm).with_retry(stop_after_attempt3)# 再 3 次# 最坏情况 3 × 3 9 次调用只在最外层加一层重试别叠。坑二重试放大成本一次请求重试 3 次 最多 4 倍费用。如果你的失败率是 20%平均成本就涨了 20%——这是隐形成本。坑三对不可重试的错误重试参数错误400、认证失败401重试一万次也没用。LangChain 已经帮你过滤了但自己写重试逻辑时要注意importasynciofromopenaiimportRateLimitError,APITimeoutErrorasyncdefsmart_retry(coro_factory,max_attempts3):foriinrange(max_attempts):try:returnawaitcoro_factory()except(RateLimitError,APITimeoutError)ase:ifimax_attempts-1:raiseawaitasyncio.sleep(2**i)# 1s, 2s, 4sexceptException:raise# 其他错误直接抛不重试三、第二层降级with_fallbacks重试解决偶发降级解决这个东西彻底不行了。3.1 换模型fromlangchain_openaiimportChatOpenAI primaryChatOpenAI(modelgpt-4o,timeout30,max_retries1)backupChatOpenAI(modelgpt-4o-mini,timeout30,max_retries1)lastChatOpenAI(modelgpt-3.5-turbo,timeout30,max_retries1)llmprimary.with_fallbacks([backup,last])chainprompt|llm|parser chain.invoke({q:...})# gpt-4o 挂了 → 自动换 gpt-4o-mini → 再换 3.5按顺序尝试第一个成功的就用。3.2 换供应商更有价值真正的生产级降级是跨供应商——OpenAI 挂了换 Anthropic都挂了换本地模型fromlangchain_openaiimportChatOpenAIfromlangchain_anthropicimportChatAnthropicfromlangchain_community.chat_modelsimportChatOllama llm(ChatOpenAI(modelgpt-4o-mini,timeout30).with_fallbacks([ChatAnthropic(modelclaude-3-haiku-20240307,timeout30),ChatOllama(modelqwen2.5,timeout60),# 最后兜底本地模型]))本地模型兜底很实用它可能效果差点但至少不会完全不可用。对客服这类场景回答质量下降远比直接报错可接受。3.3 降级到非模型方案最极端的兜底——不用模型fromlangchain_core.runnablesimportRunnableLambdadefkeyword_fallback(inputs):用关键词规则硬匹配一个答案returnKEYWORD_ANSWERS.get(inputs[question],抱歉服务暂时不可用请稍后再试。)llm_chainprompt|llm|parser finalllm_chain.with_fallbacks([RunnableLambda(keyword_fallback)])3.4 什么时候会触发降级关键点with_fallbacks只在抛异常时才切换。如果模型返回了一个质量很差但没报错的结果比如空字符串、答非所问它是不会降级的。要处理这种情况得自己加校验defvalidate_and_fallback(chain,fallbacks,inputs):try:resultchain.invoke(inputs)ifis_bad(result):# 自己定义什么是坏结果raiseValueError(结果质量不合格)returnresultexceptException:forfbinfallbacks:try:returnfb.invoke(inputs)exceptException:continuereturn抱歉服务暂时不可用。四、第三层超时4.1 设在哪里两处都要设# 1. 模型层llmChatOpenAI(modelgpt-4o-mini,timeout30)# 2. 应用层Web 服务importasyncio resultawaitasyncio.wait_for(chain.ainvoke(inputs),timeout45)应用层的超时要略大于模型层——否则可能出现应用层已经返回超时模型层还在跑的悬挂请求。4.2 流式场景的超时流式调用不能用简单的wait_for因为它是持续产出asyncdefstream_with_deadline(chain,inputs,first_token_timeout15,total_timeout120):startasyncio.get_event_loop().time()firstTrueasyncforchunkinchain.astream(inputs):nowasyncio.get_event_loop().time()iffirst:ifnow-startfirst_token_timeout:raiseTimeoutError(首字超时)firstFalseifnow-starttotal_timeout:raiseTimeoutError(总时长超时)yieldchunk首字超时TTFT和总时长要分开设——用户能容忍慢慢输出但不能容忍点了没反应。五、三层的正确组合顺序用户请求 ↓ 【超时】整体包一层 deadline ↓ 【降级】with_fallbacks([备用1, 备用2, 规则兜底]) ↓ 【重试】每个分支内部 max_retries / with_retry顺序反了会出问题如果重试包在降级外面 → 模型 A 失败重试 A 三次才降级到 B。浪费时间用户等太久正确是先降级快切每个降级分支内部自己重试。# ✅ 正确降级在外重试在内primaryChatOpenAI(modelgpt-4o,max_retries1)# 内部重试 1 次backupChatOpenAI(modelgpt-4o-mini,max_retries1)llmprimary.with_fallbacks([backup])chain(prompt|llm|parser).with_config(recursion_limit5)六、监控别等用户告诉你挂了fromlangchain_core.callbacksimportBaseCallbackHandlerimportloggingclassFallbackMonitor(BaseCallbackHandler):defon_chain_error(self,error,*,run_id,**kwargs):logging.error(fchain_error run{run_id}type{type(error).__name__})defon_llm_error(self,error,*,run_id,**kwargs):metrics.incr(fllm.error.{type(error).__name__})三个要看的指标指标告警线含义错误率 2%供应商出问题了降级触发率 5%主模型不可靠考虑换P95 延迟超 SLA有慢请求降级触发率特别重要——它告诉你主用方案还行不行。如果每天都大量走降级说明主模型该换了而不是继续凑合。七、完整示例fromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParserfromlangchain_core.runnablesimportRunnableLambda promptChatPromptTemplate.from_template({question})defmake_llm(model,**kw):returnChatOpenAI(modelmodel,timeoutkw.get(timeout,30),max_retrieskw.get(max_retries,1),temperature0)primarymake_llm(gpt-4o-mini)backupmake_llm(gpt-3.5-turbo)defrule_fallback(inputs):return抱歉AI 服务暂时不可用已为你转人工客服。chain(prompt|primary.with_fallbacks([backup])|StrOutputParser()).with_fallbacks([RunnableLambda(rule_fallback)])asyncdefsafe_answer(question:str)-str:try:returnawaitasyncio.wait_for(chain.ainvoke({question:question}),timeout45)exceptasyncio.TimeoutError:return处理超时请稍后再试。八、小结三种故障分开处理瞬时→重试、持续→降级、慢→超时重试只在最外层加一层多层叠加会让调用次数相乘、成本翻倍with_fallbacks只在抛异常时切换质量差但不报错的结果需要自己加校验跨供应商降级最有价值最后可以用本地模型或规则兜底正确顺序超时包最外层 → 降级 → 每个分支内部重试监控降级触发率持续偏高说明主方案该换而不是继续兜。下一篇讲批量与并发——batch的正确用法和速率控制。