5个让LangChain调用更稳的小技巧：自动重试、限流、结构化输出、缓存、错误隔离

用LangChain调大模型API时，有一些小坑会让你线上服务不稳定——API超时、token耗尽、JSON解析失败、并发超限。这里分享5个我在生产环境里用到的技巧，代码可以直接拿去用。

技巧1：API调用自动重试 + 指数退避

大模型API调用不稳定是常态。OpenAI/DeepSeek经常偶发502或限流错误。加自动重试是最基本的防护。

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import openai
from openai import APIError, RateLimitError, Timeout

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    retry=retry_if_exception_type((APIError, RateLimitError, Timeout)),
    before_sleep=lambda retry_state: print(
        f"重试第{retry_state.attempt_number}次，等待{retry_state.next_action.sleep}秒"
    )
)
def safe_chat_completion(messages, model="deepseek-chat"):
    client = openai.OpenAI(
        api_key="your-api-key",
        base_url="https://api.deepseek.com/v1",
        timeout=30
    )
    response = client.chat.completions.create(
        model=model,
        messages=messages,
        temperature=0.7
    )
    return response.choices[0].message.content

3次重试 + 指数退避（2秒→4秒→8秒）+ 只重试特定异常类型。这样普通网络波动不会炸，但真正挂了也不会死循环。

技巧2：结构性输出 + 异常兜底

让LLM返回JSON是很常见的需求，但它偶尔会返回残缺的JSON。加一个兜底逻辑：

import json
import re

def safe_parse_json(llm_response: str) -> dict:
    """安全解析LLM返回的JSON，包含兜底逻辑"""
    # 尝试直接解析
    try:
        return json.loads(llm_response)
    except json.JSONDecodeError:
        pass
    
    # 尝试从markdown代码块中提取
    code_block = re.search(r'```(?:json)?\s*\n(.*?)\n```', llm_response, re.DOTALL)
    if code_block:
        try:
            return json.loads(code_block.group(1))
        except json.JSONDecodeError:
            pass
    
    # 尝试从花括号提取
    brace_content = re.search(r'\{.*\}', llm_response, re.DOTALL)
    if brace_content:
        try:
            return json.loads(brace_content.group())
        except json.JSONDecodeError:
            pass
    
    # 全部失败，返回默认值
    print(f"⚠️ JSON解析失败，返回默认值。原始输出: {llm_response[:200]}")
    return {"error": "parse_failed", "raw": llm_response[:500]}

这个函数有4层回退：直接解析→markdown代码块→花括号内容→默认值。线上跑了几个月，99%以上的非标准JSON输出都能兜住。

技巧3：多模型降级

主模型挂了自动切备用模型：

from functools import wraps

def model_fallback(func):
    """装饰器：主模型失败时自动降级到备用模型"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        primary_model = kwargs.get('model', 'deepseek-chat')
        fallback_models = ['gpt-4o-mini', 'qwen-plus']
        
        for i, model in enumerate([primary_model] + fallback_models):
            try:
                kwargs['model'] = model
                result = func(*args, **kwargs)
                if i > 0:
                    print(f"⚠️ 降级使用备用模型: {model}")
                return result
            except Exception as e:
                if i == len(fallback_models):
                    raise  # 全部失败才抛出
                print(f"❌ 模型{model}失败: {e}，尝试降级...")
                continue
    return wrapper

@model_fallback
def chat_with_fallback(messages, model="deepseek-chat"):
    # 你的调用代码
    pass

这个装饰器可以加在任何调用函数上。主模型挂了自动尝试备选模型，整个过程对调用方透明。

技巧4：请求级缓存

同样的Prompt重复请求大模型，浪费钱：

import hashlib
import json
from functools import lru_cache

def request_cache(ttl_seconds: int = 3600):
    """LRU缓存装饰器，带TTL"""
    cache = {}
    
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            key = hashlib.md5(
                json.dumps({"args": str(args), "kwargs": str(kwargs)}, 
                          sort_keys=True, default=str).encode()
            ).hexdigest()
            
            if key in cache:
                cached_time, cached_result = cache[key]
                if time.time() - cached_time < ttl_seconds:
                    return cached_result
            
            result = func(*args, **kwargs)
            cache[key] = (time.time(), result)
            return result
        return wrapper
    return decorator

@request_cache(ttl_seconds=300)
def classify_intent(user_message: str) -> str:
    """意图分类（相同的输入5分钟内只调一次API）"""
    # 调用LLM做分类
    pass

意图分类这种场景特别适合缓存——同样的问题问10遍，只有第1次真正调了API。

技巧5：请求级超时控制和并发限流

LangChain默认没有超时控制，一个卡住的请求可能拖垮整个服务：

import asyncio
import time

class APIRateLimiter:
    """简单的漏桶算法限流器"""
    def __init__(self, max_per_minute=30):
        self.max_per_minute = max_per_minute
        self.interval = 60.0 / max_per_minute
        self.last_call = 0
    
    def wait_if_needed(self):
        elapsed = time.time() - self.last_call
        if elapsed < self.interval:
            sleep_time = self.interval - elapsed
            time.sleep(sleep_time)
        self.last_call = time.time()

# 使用
limiter = APIRateLimiter(max_per_minute=30)

def rate_limited_call(messages):
    limiter.wait_if_needed()
    return client.chat.completions.create(
        model="deepseek-chat",
        messages=messages,
        timeout=25  # 单次请求超时25秒
    )

每分钟限制30次调用，每次请求超时25秒。这个组合在你同时跑多个Agent任务时特别有用。

---

以上5个技巧可以单独用也可以组合用。我目前在线上服务的核心流程是：限流 → 超时控制 → 自动重试 → 降级 → JSON兜底，跑了大半年没出过大的稳定性问题。

如果你的生产环境也遇到类似问题，可以直接把代码拿去用。