做 Canguo Science 这类要对接多个模型的工具时，绕不开一个现实：大模型调用是不可靠的。请求偶发超时、被限流打回 429、某个后端临时抽风、流式输出中途断开……在 demo 里裸调一次 create() 没问题，一上量就原形毕露。

这篇把"让 LLM 调用扛得住"拆开，用一段段能跑的 Python 讲清楚一整套客户端侧的高可用做法：超时分层、指数退避重试、熔断、多后端故障转移、请求对冲、流式兜底、用量核算与可观测性。全程只依赖 openai 官方 SDK（v1.x），不绑定任何具体服务。代码为便于阅读做了简化。

一、超时：一定要分层

超时分两层——连接超时和读取超时，得分开设，否则一个粗暴的总超时会误伤正常的长输出：

import os, httpx

from openai import OpenAI

client = OpenAI(

api_key=os.environ["LLM_API_KEY"],

base_url=os.environ["LLM_BASE_URL"],

timeout=httpx.Timeout(connect=5.0, read=30.0, write=10.0, pool=5.0),

max_retries=0, # 先关掉 SDK 内置重试，下面自己接管

)

连接 5 秒握不上手，基本是对端有问题，快速失败比干等强；读取给 30 秒，是因为长输出确实需要时间。把这两个值分开，才不会用一个总超时误杀正常的长响应。

二、重试：区分错误类型 + 退避 + 抖动

不是所有错误都该重试。400（参数错）重试一百次也没用；该重试的是瞬时故障：429 限流、5xx、连接错误、超时。

import random, time

from openai import (APIConnectionError, APITimeoutError,

RateLimitError, InternalServerError)

RETRYABLE = (APIConnectionError, APITimeoutError,

RateLimitError, InternalServerError)

def with_retry(fn, *, max_attempts=4, base=0.5, cap=8.0):

"""指数退避 + 满抖动（full jitter），只重试瞬时故障。"""

for attempt in range(1, max_attempts + 1):

try:

return fn()

except RETRYABLE:

if attempt == max_attempts:

raise

delay = min(cap, base * (2 ** (attempt - 1)))

time.sleep(random.uniform(0, delay))

# 其它异常（如 400 / 401）会自然向上抛出，不进入重试

满抖动（random.uniform(0, delay)）比固定退避更重要：一堆客户端同时失败、又同时重试，会把对端二次打垮；加了随机，重试请求就被摊平了，避免"重试风暴"。

三、熔断：别对着已经挂了的后端狂敲

如果一个后端已经挂了，继续对它重试只是浪费时间、还拖慢整体响应。熔断器的作用是：连续失败到一定次数就"跳闸"，在冷却期内直接拒绝请求，冷却后再放一个试探请求探路。

import time, threading

class CircuitBreaker:

"""三态熔断：closed 正常 / open 跳闸 / half-open 试探。"""

def __init__(self, fail_threshold=5, cooldown=30.0):

self.fail_threshold = fail_threshold

self.cooldown = cooldown

self._fails = 0

self._opened_at = None

self._lock = threading.Lock()

def allow(self) -> bool:

with self._lock:

if self._opened_at is None:

return True # closed：正常放行

if time.time() - self._opened_at >= self.cooldown:

return True # half-open：放一个试探

return False # open：冷却期内拒绝

def on_success(self):

with self._lock:

self._fails = 0

self._opened_at = None

def on_failure(self):

with self._lock:

self._fails += 1

if self._fails >= self.fail_threshold:

self._opened_at = time.time() # 跳闸

熔断和重试是互补的：重试应对"抖一下"，熔断应对"真的挂了"。没有熔断，一个彻底宕掉的后端会让你的每个请求都白白多等好几秒。

四、故障转移：消灭单点

高可用的核心是没有单点。给定一组后端（不同上游、不同模型、不同区域），按优先级尝试，主用不行就自动降级；配合熔断，跳闸的后端直接跳过：

from dataclasses import dataclass, field

from openai import OpenAI

@dataclass

class Backend:

name: str

base_url: str

api_key: str

model: str

breaker: CircuitBreaker = field(default_factory=CircuitBreaker)

def make_client(b: Backend) -> OpenAI:

return OpenAI(api_key=b.api_key, base_url=b.base_url,

timeout=20.0, max_retries=0)

def chat_ha(messages, backends: list[Backend], **kw) -> str:

last_err = None

for b in backends:

if not b.breaker.allow():

continue # 熔断中，直接跳过

client = make_client(b)

try:

resp = with_retry(lambda: client.chat.completions.create(

model=b.model, messages=messages, **kw))

b.breaker.on_success()

return resp.choices[0].message.content

except Exception as e:

last_err = e

b.breaker.on_failure()

print(f"[fallback] backend={b.name} failed: {type(e).__name__}")

continue

raise RuntimeError(f"all backends failed, last error: {last_err}")

配置从环境变量或配置文件读，扩容、切换都不用改代码。分层是关键：后端内部做重试（应对瞬时抖动），后端之间做故障转移（应对整体不可用），再用熔断避免在坏后端上浪费时间。 三者配合，才既不会因为一次抖动贸然切换、也不会在死掉的后端上死磕。

五、请求对冲：压住尾延迟（进阶）

有时候后端没挂，只是这一次特别慢。对延迟敏感的场景，可以用"对冲（hedging）"：先发一个请求，超过一个阈值还没回来，就并发补发一个，取最先返回的。

from concurrent.futures import ThreadPoolExecutor, FIRST_COMPLETED, wait

def hedged_call(fn, hedge_delay=2.0, max_parallel=2):

"""先发一个；hedge_delay 秒还没回来，再并发补发一个，取最先返回的。

注意：对冲会放大成本和上游压力，只对尾延迟敏感、且幂等的请求用。"""

with ThreadPoolExecutor(max_workers=max_parallel) as pool:

futures = [pool.submit(fn)]

done, _ = wait(futures, timeout=hedge_delay,

return_when=FIRST_COMPLETED)

if done:

return done.pop().result()

futures.append(pool.submit(fn)) # 补发对冲请求

done, _ = wait(futures, return_when=FIRST_COMPLETED)

return done.pop().result()

对冲是把"钱和算力"换"延迟"，务必克制——只对少数尾延迟敏感的关键请求开，别全局乱用。

六、流式输出与中断兜底

交互式场景要用流式（stream=True），首字更快、体验好很多。但流式的错误处理更麻烦——连接可能中途断，而且默认拿不到用量统计，得显式打开：

def stream_chat(client, model, messages):

stream = client.chat.completions.create(

model=model, messages=messages, stream=True,

stream_options={"include_usage": True}, # 让最后一帧带上用量

)

parts = []

try:

for chunk in stream:

if chunk.choices and chunk.choices[0].delta.content:

piece = chunk.choices[0].delta.content

parts.append(piece)

print(piece, end="", flush=True)

if chunk.usage:

print(f"\n[usage] {chunk.usage.total_tokens} tokens")

except Exception as e:

# 中途断流：把已收到的部分保住，再决定重试还是降级

print(f"\n[stream broken] {type(e).__name__}, got {len(parts)} pieces")

raise

return "".join(parts)

一个容易被忽略的点：stream_options={"include_usage": True} 会让流式响应在结尾多发一帧、带上 token 用量——否则流式模式下你是拿不到 usage 的，成本就成了黑洞。

七、幂等：重试别把事情做两遍

纯问答的 chat.completions 本身没有副作用，重试是安全的。但只要你的业务在拿到结果后有副作用（写库、扣费、发消息），重试 + 对冲就可能让同一件事被做两遍。

稳妥的做法是：给每个逻辑请求算一个内容指纹当幂等键，副作用落库前先按键去重。

import hashlib, json

def request_key(model, messages) -> str:

payload = json.dumps({"model": model, "messages": messages},

ensure_ascii=False, sort_keys=True)

return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:16]

把这个键带到你的副作用逻辑里做"处理过就跳过"，就能让重试和对冲变得安全。

八、用量与成本核算

非流式响应里直接带 usage，累计起来即可做预算控制：

from collections import defaultdict

class UsageMeter:

def __init__(self):

self.by_model = defaultdict(lambda: {"prompt": 0, "completion": 0})

def add(self, model, usage):

self.by_model[model]["prompt"] += usage.prompt_tokens

self.by_model[model]["completion"] += usage.completion_tokens

def report(self):

for m, u in self.by_model.items():

print(f"{m}: prompt={u['prompt']} completion={u['completion']}")

想在发请求前就估长度、避免超预算，用 tiktoken 离线数 token：

import tiktoken

enc = tiktoken.get_encoding("cl100k_base")

n_tokens = len(enc.encode("要发送的 prompt 文本"))

九、可观测性：每次调用都记清楚

出了问题能不能定位，取决于你有没有把每次调用记下来。用结构化日志，把模型、延迟、用量、结果都打出来：

import time, json, logging

logger = logging.getLogger("llm")

def logged(model, fn):

t0 = time.time()

try:

resp = fn()

logger.info(json.dumps({

"model": model, "ok": True,

"latency_ms": round((time.time() - t0) * 1000),

"tokens": resp.usage.total_tokens if resp.usage else None,

}, ensure_ascii=False))

return resp

except Exception as e:

logger.warning(json.dumps({

"model": model, "ok": False,

"latency_ms": round((time.time() - t0) * 1000),

"error": type(e).__name__,

}, ensure_ascii=False))

raise

有了这些结构化日志，你就能算出每个后端的成功率、P99 延迟、失败类型分布——熔断阈值、重试次数怎么调，才有数据支撑，而不是拍脑袋。

十、小结

生产环境里调 LLM，可靠性不是"要不要做"，而是"欠债迟早要还"。几条经验拎出来：

• 超时分层，别用一个总超时误杀长响应；
• 区分错误类型再重试，退避一定要加抖动；
• 熔断 + 故障转移 + 重试分三层，各管一类故障；
• 对冲压尾延迟但烧钱，务必克制；
• 有副作用的请求要做幂等，重试才安全；
• 流式记得开 include_usage，再配上结构化日志，调参才有依据。

这些模式跟你用哪家模型、哪个后端都无关，理解了就能落到自己的项目里。这也是我们在 Canguo Science 里对接多模型时，一点点踩出来的经验。