AI Engineering Academy · 课时

超时预算与平稳降级

Q: 「超时预算与平稳降级」这节课中我会学到什么？

在流程的每一层设置严格的超时预算，并在 LLM 超出预算时实现平稳降级，提供缓存响应或简化响应。 你通过在浏览器中直接运行的动手代码来练习 AI Engineering Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

在流程的每一层设置严格的超时预算，并在 LLM 超出预算时实现平稳降级，提供缓存响应或简化响应。

第 4 / 4 课13 个步骤

超时预算与平稳降级 是 CoddyKit 上的免费 AI Engineering Academy 课时。这是第 4 节课，共 4 节。你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。这是 AI Engineering Academy 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Engineering Academy 课程共包含 4 节课。

什么是超时预算

超时预算是为请求在整个流水线的所有阶段完成而分配的最大总时间。您不必为每个单独的 API 调用设置随意的超时时间，而是为面向用户的端到端操作定义总预算，并将其分配给检索、LLM 生成和后处理步骤。这样，即使某些阶段速度较慢，您也始终能在可接受的时间内返回响应。

在流水线各阶段之间分配预算

典型的 RAG 聊天流水线包含三个阶段：检索、LLM 生成和响应格式化。请根据每个阶段通常所需的时间，以及用户能够接受的延迟程度，为它们分别分配时间片。剩余的余量就是您的降级缓冲——如果某个阶段用完了全部分配时间，您就应在后续阶段开始减少处理，以确保不超出总体预算。

# Total user-facing SLA: 8000ms
BUDGET_TOTAL_MS = 8000

BUDGET_STAGES = {
    'retrieval':   1500,  # vector search + rerank
    'llm_call':    5500,  # token streaming
    'formatting':  500,   # post-processing
    'slack':       500,   # buffer for overhead
}

assert sum(BUDGET_STAGES.values()) == BUDGET_TOTAL_MS

跟踪预算消耗

使用 BudgetTracker 记录开始时间，并在每次阶段转换时检查剩余预算。开始某个阶段前，请确认剩余预算充足。这样，下游阶段就能进行调整——例如，检索步骤在分配的 1500ms 预算中耗时 1200ms，只剩下 300ms 的余量，此时应触发更简单的 LLM 提示词，或跳过重新排序步骤。

import time

class BudgetTracker:
    def __init__(self, total_ms: float):
        self.start = time.perf_counter()
        self.total_ms = total_ms

    def elapsed_ms(self) -> float:
        return (time.perf_counter() - self.start) * 1000

    def remaining_ms(self) -> float:
        return self.total_ms - self.elapsed_ms()

    def check(self, stage: str, required_ms: float = 0) -> bool:
        remaining = self.remaining_ms()
        if remaining < required_ms:
            print(f'Budget exhausted before {stage}: {remaining:.0f}ms left, need {required_ms}ms')
            return False
        return True

优雅降级的定义

优雅降级是指当完整流水线无法在预算内完成时，返回质量较低但仍有用的响应，而不是返回错误。示例包括：返回缓存的响应、跳过重新排序、截断上下文窗口、使用速度更快但准确性较低的模型，或返回预先编写的备用消息。目标始终是让用户有所收获，而不是一无所获。

# Degradation ladder for a RAG chat endpoint:
# Level 0 (normal):  retrieve 10 chunks + rerank + GPT-4o   -- 8000ms budget
# Level 1 (fast):    retrieve 5 chunks + skip rerank + GPT-4o -- 5000ms budget
# Level 2 (minimal): retrieve 3 chunks + GPT-4o-mini          -- 3000ms budget
# Level 3 (cached):  return semantic cache hit                 -- 100ms
# Level 4 (sorry):   return static 'Try again in a moment'    -- 1ms

实现降级阶梯

在每个流水线决策点检查剩余预算，并选择适当的质量级别。下面的代码会根据剩余预算选择检索深度和模型。这意味着在正常负载下，用户可以获得最佳质量；而在高延迟期间，用户仍能得到有用的响应，而不是遇到超时错误。

async def smart_rag_query(question: str, budget_ms: float = 8000) -> str:
    tracker = BudgetTracker(budget_ms)

    # Retrieval stage
    if tracker.remaining_ms() > 5000:
        chunks = await retrieve_and_rerank(question, top_k=10)
    elif tracker.remaining_ms() > 3000:
        chunks = await retrieve(question, top_k=5)  # skip rerank
    elif tracker.remaining_ms() > 1500:
        chunks = await retrieve(question, top_k=3)  # minimal retrieval
    else:
        return await get_cached_or_static(question)

    # LLM stage
    if tracker.remaining_ms() > 4000:
        model = 'gpt-4o'
    else:
        model = 'gpt-4o-mini'  # faster fallback

    timeout = tracker.remaining_ms() / 1000 - 0.5
    return await generate_answer(question, chunks, model, timeout)

在 API 调用级别设置超时

请始终为每个外部 API 调用设置明确的超时时间。OpenAI Python SDK 接受以秒为单位的 timeout 参数。请将其设置为略短于剩余预算的时间，这样您就有时间处理异常，并可能在总体响应截止时间前进行优雅降级。切勿依赖 SDK 的默认超时设置——对于面向用户的请求来说，它可能过长。

async def generate_answer(question: str, chunks: list, model: str, timeout_sec: float) -> str:
    context = '\n\n'.join(chunks)
    prompt = f'Answer using this context:\n{context}\n\nQuestion: {question}'
    try:
        resp = await client.chat.completions.create(
            model=model,
            messages=[{'role': 'user', 'content': prompt}],
            max_tokens=500,
            timeout=max(timeout_sec, 1.0)  # minimum 1 second
        )
        return resp.choices[0].message.content
    except openai.APITimeoutError:
        return 'I was unable to generate a response in time. Please try again.'

返回部分流式响应

使用流式传输时，您可以返回预算耗尽前已经生成的部分响应。如果流式传输过程中发生超时，请停止读取新令牌，追加省略号或简短的续写提示，然后关闭流。用户看到的是一个自然截断的响应，而不是空白的错误结果。这只有在流式传输中才可行——非流式调用要么全部成功，要么全部失败。

async def stream_with_budget(question: str, budget_ms: float):
    tracker = BudgetTracker(budget_ms)
    collected = []
    stream = await client.chat.completions.create(
        model='gpt-4o-mini',
        messages=[{'role': 'user', 'content': question}],
        stream=True
    )
    async for chunk in stream:
        if tracker.remaining_ms() < 200:  # 200ms safety margin
            collected.append(' [response truncated]')
            break
        delta = chunk.choices[0].delta.content or ''
        collected.append(delta)
        yield delta
    # Ensure stream is closed even if budget exceeded
    await stream.close()

将语义缓存作为降级层

语义缓存是出色的降级层，因为它几乎没有延迟。在调用 LLM 之前，请查询语义缓存，查找之前相似的问题。如果找到相似度较高的缓存命中结果（余弦相似度高于 0.92），请立即返回缓存的答案。这样既能加快响应速度，也能在 LLM 速度缓慢或不可用时提供即时备用方案。

async def query_with_cache_fallback(question: str, budget_ms: float = 8000) -> str:
    # Try semantic cache first (fast)
    cached = await semantic_cache.lookup(question, threshold=0.92)
    if cached:
        return cached.response

    tracker = BudgetTracker(budget_ms)
    # Try full pipeline
    if tracker.remaining_ms() > 3000:
        try:
            return await smart_rag_query(question, tracker.remaining_ms())
        except Exception:
            pass  # fall through to static response

    # Last resort
    return 'I am experiencing high load right now. Please try again in a moment.'

记录降级事件

每当您的流水线降至较低的质量级别时，都应将其记录为结构化事件。请记录达到的降级级别、每个阶段的剩余预算以及最终延迟。分析这些日志可以告诉您每个降级级别被触发的频率，从而帮助您调整预算、找出哪些阶段持续超出预算，并为基础设施投入提供依据。

import structlog

log = structlog.get_logger()

def log_degradation(level: int, stage: str, remaining_ms: float, total_ms: float):
    log.warning(
        'pipeline_degradation',
        degradation_level=level,
        triggered_at_stage=stage,
        remaining_budget_ms=round(remaining_ms),
        total_budget_ms=total_ms,
        budget_consumed_pct=round((total_ms - remaining_ms) / total_ms * 100)
    )

通过界面信号管理用户预期

提供降级响应时，请向用户提示质量可能低于平时。对于聊天界面，可以显示类似“快速响应模式——部分细节可能受限”的低调提示。对于提取 API，请在 JSON 响应中加入 degraded: true 字段，以便下游使用者以不同方式处理降级结果。即使在服务中断期间保持透明，也有助于维护用户信任。

from pydantic import BaseModel
from typing import Optional

class ChatResponse(BaseModel):
    content: str
    degraded: bool = False
    degradation_level: Optional[int] = None  # 0=full, 1=fast, 2=minimal, 3=cached
    latency_ms: int

# API response when degraded:
# {
#   'content': 'Here is a brief answer...',
#   'degraded': true,
#   'degradation_level': 2,
#   'latency_ms': 2800
# }

随着时间推移调整预算分配

初始预算分配只是估算值。在生产环境运行一周后，请利用跟踪数据分析每个阶段耗时的分布。如果检索通常耗时 800ms，而不是预算中的 1500ms，就可以将这部分余量重新分配给 LLM 阶段，从而允许生成更多输出令牌或使用更大的上下文窗口。预算调整是一项持续的运营工作，而不是一次性的配置。

# Budget tuning based on production p95 data:
ACTUAL_P95 = {
    'retrieval': 780,    # vs budget 1500ms -> 720ms headroom
    'llm_call':  4200,   # vs budget 5500ms -> 1300ms headroom
    'formatting': 120,   # vs budget 500ms  -> 380ms headroom
}

TOTAL_HEADROOM = sum(
    BUDGET_STAGES[k] - ACTUAL_P95[k] for k in ACTUAL_P95
)
print(f'Total headroom: {TOTAL_HEADROOM}ms')
# Reallocate headroom to allow longer LLM responses

快速检查

测试您对超时预算和优雅降级的理解。

在交互式课程中回答

课程回顾

在本课中，您学习了：超时预算会在流水线各阶段之间分配时间，确保您始终能在可接受的截止时间内响应；降级阶梯会在时间耗尽时逐步提供质量较低的响应，而不是返回错误；记录降级事件有助于您发现并解决长期存在的瓶颈。接下来，我们将探索用于自动化质量评估的 LLM 评审模式。

免费开始

用 AI 导师学习 Python — 免费

在浏览器中编写并运行真实代码，获得全天候 AI 导师的即时帮助，并在网页或应用中继续学习。

课程: 30
课程: 120

免费开始学习查看完整课程

常见问题解答

「超时预算与平稳降级」课时是免费的吗？

是的 — 「超时预算与平稳降级」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Engineering Academy 课程的其余内容，请升级到 CoddyKit PRO。 AI Engineering Academy 课程共包含 4 节课。

「超时预算与平稳降级」这节课中我会学到什么？

在流程的每一层设置严格的超时预算，并在 LLM 超出预算时实现平稳降级，提供缓存响应或简化响应。你通过在浏览器中直接运行的动手代码来练习 AI Engineering Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Engineering Academy 需要有经验吗？

无需任何先前经验。CoddyKit 上的 AI Engineering Academy 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。这是第 4 节课，共 4 节。

「超时预算与平稳降级」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 AI Engineering Academy 课中编写并运行代码吗？

能。每节 AI Engineering Academy 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

← 返回 AI Engineering Academy