""" ============================================================================= 前置讲 P2 · 代码 04：可靠地调用 —— 重试、退避、抖动、超时 ============================================================================= 对应讲义：lectures/prereq2_llm_as_a_service.md §5 【这个脚本要回答的问题】 1. 偶尔失败的服务，要不要重试？重试几次、等多久？（演示 1） 2. 哪些错误绝对不能重试？（演示 2） 3. 很多客户端同时被限速时，"统一等 1 秒再试"为什么是个坏主意？（演示 3：惊群效应） 4. 超时设得太短会怎样？（演示 4） 【怎么读输出】 演示 1–3 用虚拟时钟（VirtualClock）：等待不真的发生，所以几秒内就能模拟几分钟，而且每次结果一样。 演示 4 用真实的 HTTP 和真实的等待（约 2 秒）。 【运行】python3 04_reliability.py （约 3 秒） 【依赖】只用 Python 标准库；import 同目录的 llm_api.py、mock_server.py，第 1 讲的 console.py。 """ import heapq import os import random import sys sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "lecture01")) from console import pct, print_header, print_observations, print_table # noqa: E402 from llm_api import (BadRequestError, FlakyLLM, LLMError, LLMRequest, LLMTimeout, RetryingLLM, # noqa: E402 SimLLM, VirtualClock) from mock_server import HttpLLM, MockLLMServer # noqa: E402 def demo1_policies(): """演示 1：一个客户端、独立的偶发失败，几种重试策略。""" print_header("演示 1：偶尔失败的服务——几种重试策略（失败和负载无关）") sim = SimLLM() questions = [f"What is the capital of {c.name}?" for c in sim.world.countries] * 40 # 960 个请求 policies = [ ("不重试", dict(max_retries=0)), ("立刻重试，最多 4 次", dict(max_retries=4, base=0.0, jitter="none", honor_retry_after=False)), ("指数退避（0.5·2^k 秒），最多 4 次", dict(max_retries=4, base=0.5, jitter="none", honor_retry_after=False)), ("指数退避 + 全抖动，最多 4 次", dict(max_retries=4, base=0.5, jitter="full", honor_retry_after=False)), ("同上 + 尊重 retry-after", dict(max_retries=4, base=0.5, jitter="full", honor_retry_after=True)), ] rows, res = [], {} for name, kw in policies: clock = VirtualClock() flaky = FlakyLLM(sim, p_429=0.3, p_500=0.1, p_timeout=0.02, retry_after=2.0, timeout_s=30.0, seed=1, clock=clock) llm = RetryingLLM(flaky, clock=clock, seed=2, **kw) ok = 0 for q in questions: try: llm.chat(q, max_tokens=30) ok += 1 except LLMError: pass res[name] = ok rows.append([name, pct(ok, len(questions)), f"{flaky.calls / len(questions):.2f}", f"{llm.waited:.0f} 秒", f"{clock.now():.0f} 秒"]) print(f"{len(questions)} 个请求；每次调用 30% 被限速（429，retry-after = 2 秒）、10% 服务器错误、2% 超时（白等 30 秒）") print_table(["策略", "最终成功", "平均每个请求调用几次", "重试前的等待（合计）", "总耗时（虚拟时钟）"], rows) p_fail = 0.42 print(f" 理论值：每次调用失败的概率 = 0.42，重试 4 次仍失败的概率 = 0.42^5 = {p_fail ** 5:.3f}，" f"即成功率约 {1 - p_fail ** 5:.1%}") print_observations([ f"不重试时只有 {pct(res['不重试'], len(questions))} 的请求成功：一次实验跑几百道题，没有重试就会缺一大块数据。", f"重试 4 次后成功率是 {pct(res['立刻重试，最多 4 次'], len(questions))}，接近理论值 1 − 0.42⁵ ≈ 98.7%：" "失败是独立的，多试几次几乎总能成功；剩下的那一点点要靠更多次重试，或者在实验记录里标成'缺失'。", "在这个模拟里，'立刻重试'最快——因为这里的失败是**随机**的，和服务器忙不忙无关，等待没有任何好处。" "真实服务的 429 恰恰是因为**太忙**；这时立刻重试只会让它更忙（演示 3）。" "尊重 retry-after 让总耗时变长，但这是服务端明确要求的，不尊重它往往会被限得更狠。", ]) def demo2_dont_retry(): """演示 2：4xx 错误不应该重试。""" print_header("演示 2：哪些错误不能重试") clock = VirtualClock() flaky = FlakyLLM(SimLLM(), p_429=0.0, p_500=0.0, clock=clock) llm = RetryingLLM(flaky, max_retries=4, clock=clock) rows = [] for name, req in [("消息以 assistant 开头", LLMRequest(messages=[{"role": "assistant", "content": "hi"}])), ("max_tokens = 0", LLMRequest(messages=[{"role": "user", "content": "hi"}], max_tokens=0))]: before = flaky.calls try: llm.complete(req) rows.append([name, "成功", flaky.calls - before]) except BadRequestError as e: rows.append([name, f"BadRequestError：{e}", flaky.calls - before]) print_table(["错误的请求", "结果", "实际调用了几次"], rows, aligns=["<", "<", ">"]) print_observations([ "BadRequestError（400）的 retryable = False：RetryingLLM 第一次失败就直接抛出，不浪费任何重试。", "原则：**只重试'换个时间再来可能成功'的错误**（429、5xx、超时、连接中断）。" "400、401、404 说明请求或配置本身有问题，重试只会浪费时间和额度——还可能触发更严的限速。", ]) def herd(n_clients, policy, rate=20.0, burst=20, seed=0): """ 惊群模拟：n_clients 个客户端在 t=0 同时各发一个请求；服务器用令牌桶限速（每秒补充 rate 个令牌，最多存 burst 个）， 没有令牌就返回 429（retry-after = 1 秒）；被拒的客户端按 policy 等待后重试，直到成功。 【参数】policy："fixed1"（固定等 1 秒）/ "exp"（0.5·2^k 秒，不抖动）/ "exp_jitter"（全抖动）/ "exp_jitter_ra"（全抖动，但至少等 retry-after）；rate / burst：令牌桶的补充速度和容量 【返回】(最后一个客户端成功的时间, 一共发了多少个请求) 【思路】离散事件模拟：一个按时间排序的"待发请求"堆（heapq），每次取出最早的一个处理。 令牌桶：距离上次处理过去了 Δt 秒，就补充 rate·Δt 个令牌（但不超过 burst）；有令牌就放行并扣掉一个。 【Python 小知识】heapq.heappush / heappop 维护一个最小堆：堆顶永远是最小的元素（这里是最早的时间）。 """ rng = random.Random(seed) events = [(0.0, i, 0) for i in range(n_clients)] # (到达时间, 客户端编号, 第几次尝试) heapq.heapify(events) tokens, last_t, sent, done_t = float(burst), 0.0, 0, 0.0 while events: t, cid, k = heapq.heappop(events) tokens = min(burst, tokens + rate * (t - last_t)) last_t = t sent += 1 if tokens >= 1.0: tokens -= 1.0 done_t = max(done_t, t) continue d = min(8.0, 0.5 * 2 ** k) if policy == "fixed1": d = 1.0 elif policy.startswith("exp_jitter"): d = rng.uniform(0, d) if policy == "exp_jitter_ra": d = max(d, 1.0) heapq.heappush(events, (t + d, cid, k + 1)) return done_t, sent def demo3_herd(): """演示 3：惊群效应。""" print_header("演示 3：100 个客户端同时被限速——惊群效应（thundering herd）") n = 100 policies = [("固定等 1 秒", "fixed1"), ("指数退避，不抖动", "exp"), ("指数退避 + 全抖动", "exp_jitter"), ("全抖动，但至少等 retry-after（1 秒）", "exp_jitter_ra")] servers = [("宽松：每秒 20 个，最多攒 20 个", 20.0, 20), ("突发容量小：每秒 20 个，最多攒 5 个", 20.0, 5)] res = {} for sname, rate, burst in servers: rows = [] for name, pol in policies: done, sent = herd(n, pol, rate=rate, burst=burst) res[(burst, pol)] = (done, sent) rows.append([name, f"{done:.2f} 秒", sent, sent - n]) print(f"\n服务器（{sname}）；{n} 个客户端在同一时刻各发 1 个请求；被拒返回 429（retry-after = 1 秒）") print_table(["重试策略", "全部完成的时间", "一共发了多少请求", "其中被拒的"], rows) print(f"\n 下限：每秒 20 个，100 个请求最快也要 (100 − 攒着的令牌数) / 20 秒——宽松的服务器 4.00 秒，突发容量小的 4.75 秒") a, b = (lambda p: res[(20, p)]), (lambda p: res[(5, p)]) print_observations([ f"不抖动的指数退避在两种服务器上都是最差的（{a('exp')[0]:.1f} 秒 / {b('exp')[0]:.1f} 秒）：所有被拒的客户端**同步地**等待，" "等待的间隔越拉越长，服务器的令牌攒满了也没人来用，容量白白浪费。", f"'固定等 1 秒'在宽松的服务器上恰好最快（{a('fixed1')[0]:.2f} 秒，正好是下限）：每一波重试到达时，" f"令牌桶刚好补满 20 个。可一换成突发容量小的服务器，每一波只有 5 个能成功，其余的一起被拒、一起再来——" f"{b('fixed1')[0]:.1f} 秒、{b('fixed1')[1]} 个请求。它的'好'是碰巧和服务器的参数对上了。", f"全抖动把重试的时刻随机地摊开，到达的速度接近服务器的补充速度：在突发容量小的服务器上 {b('exp_jitter')[0]:.2f} 秒、" f"{b('exp_jitter')[1]} 个请求，远好于同步的策略；在宽松的服务器上也比不抖动的指数退避快一倍。" "它不是在每种情况下都最好，但在**你不知道服务器参数**的时候最稳——这就是 Brooker 那篇 'Exponential Backoff And Jitter' 的结论。", f"尊重 retry-after 在宽松的服务器上请求数最少（{a('exp_jitter_ra')[1]}）：服务器给的建议准确时，照做最省。" f"但在突发容量小的服务器上，'等 1 秒'这个建议并不准，照做反而比纯抖动慢（{b('exp_jitter_ra')[0]:.2f} vs " f"{b('exp_jitter')[0]:.2f} 秒）。跑大规模提示实验时，你的几十个并发请求就是这群客户端。", ]) def demo4_timeout(country): """演示 4：超时太短。""" print_header("演示 4：超时设得太短——白等，还白白让服务器干活") with MockLLMServer(api_key="sk-t", response_delay=0.5) as srv: clock = VirtualClock() llm = RetryingLLM(HttpLLM(srv.url, api_key="sk-t", timeout=0.2), max_retries=2, base=0.0, clock=clock) try: llm.chat(f"What is the capital of {country}?") result = "成功" except LLMTimeout as e: result = f"LLMTimeout：{e}" fast = HttpLLM(srv.url, api_key="sk-t", timeout=2.0).chat(f"What is the capital of {country}?") completed = srv.completed print_table(["设置", "结果"], [ ["服务器每个请求要 0.5 秒；客户端超时 0.2 秒，最多重试 2 次", result], ["服务器实际处理完的请求数（含上面 3 次尝试 + 下面 1 次）", completed], ["客户端超时改成 2 秒", repr(fast.text)], ], aligns=["<", "<"]) print_observations([ "超时短于服务器正常的处理时间时，每一次尝试都注定失败；重试只是把同样的失败重复三遍。", "更糟的是：客户端放弃了，服务器并不知道，它照样把 3 个回答都生成完了——真实 API 里这些工作可能照样计费。" "超时要设得比**正常情况下的最长耗时**宽裕；长回答用流式（03 号脚本），而不是把超时调到很大。", "超时之后的重试还可能造成**重复处理**：如果请求有副作用（发邮件、写数据库、下订单），就会做两遍。" "只读的生成请求重试无妨；有副作用的工具调用要设计成幂等的（第 8 讲）。", ]) def main(): demo1_policies() demo2_dont_retry() demo3_herd() demo4_timeout(SimLLM().world.countries[0].name) if __name__ == "__main__": main()