三级容灾：当辅助 LLM 挂了，你的摘要怎么办？

---

专栏信息

《从零到一构建跨平台 AI 助手：WeClaw 实战指南》专栏

本文是模块八第 4 篇，讲解摘要生成的可靠性工程设计。

---

作者与项目

作者简介：翁勇刚 WENG YONGGANG 新概念龙虾-WeClaw 开发团队负责人，一群专注于跨平台 AI 应用的实践者 理念："再复杂的技术，也能用代码讲清楚"

• 项目地址：https://github.com/wyg5208/weclaw.git
• 官网地址：https://weclaw.link
• 作者 CSDN：https://blog.csdn.net/yweng18

---

摘要

本文结构概览： 本文从"辅助模型 API 不可用"的真实故障场景出发，分析旧方案的致命缺陷，然后详解三级容灾架构的设计与实现，最后通过实测验证降级表现。

背景：上下文压缩通常由一个"辅助模型"执行（成本更低、速度更快）。但当辅助模型的 API 超时、限流或预算耗尽时，压缩流程会失败。

核心问题：压缩失败时，早期上下文怎么办？旧方案直接截断，等于丢失了全部历史记忆。

解决方案：三级容灾机制——Level 1 辅助模型、Level 2 主模型重试（30s 超时）、Level 3 静态回退（零 LLM 成本）。

关键成果：

• 任何情况下都不会丢失上下文（最坏情况保留结构化文件摘要）
• 冷却协调机制避免"每轮尝试两个模型"的浪费
• 静态回退方案零 LLM 成本仍能保留关键信息

适合读者：构建生产级 LLM Agent 的开发者，关注系统可靠性和容灾设计

阅读时长：约 10 分钟

关键词：三级容灾、摘要生成、降级策略、静态回退、可靠性工程

---

一、故障场景：辅助模型突然不工作了

1.1 正常流程

用户对话 → 上下文超限 → 调用辅助模型生成摘要 → 用摘要替换旧消息 → 继续对话

1.2 故障发生

用户对话 → 上下文超限 → 调用辅助模型 → API 超时/限流/余额不足 → ？？？

辅助模型不可用的常见原因：

故障类型	典型表现	发生频率
API 超时	请求 30s+ 无响应	中等（网络波动）
速率限制	HTTP 429 Too Many Requests	高（免费额度）
余额不足	HTTP 402/403	低（但致命）
服务不可用	HTTP 503 Service Unavailable	低（云厂商故障）
模型下线	HTTP 404 Not Found	极低（但不可恢复）

1.3 旧方案的致命缺陷

# 旧方案：压缩失败直接截断
async def compress_context(self, messages, threshold):
    try:
        summary = await self.auxiliary_model.summarize(messages[:-RECENT_N])
        return [summary_msg] + messages[-RECENT_N:]
    except Exception:
        # 压缩失败？那就直接截断吧
        logger.error("Compression failed, falling back to truncation")
        return messages[-RECENT_N:]  # 早期上下文全部丢失！

后果：用户在 30 分钟前讨论的所有内容——分析结论、搜索到的关键数据、达成的决策——全部被丢弃。AI 像失忆了一样重新开始。

---

二、三级容灾设计

[图片: 三级容灾流程图 | 生成方式: 文生图 PROMPT: "A three-level fallback waterfall diagram for LLM summary generation: Level 1 Auxiliary Model (green box, labeled 'cheap & fast'), Level 2 Main Model Retry with 30s timeout (yellow box, labeled 'expensive fallback'), Level 3 Static Fallback (red box, labeled 'zero cost, extract file paths and tool names'), with arrows showing degradation path from top to bottom, clean flowchart style, white background"]

2.1 架构概览

Level 1: 辅助模型 (AuxiliaryClient)
    │ 成功 → 返回 LLM 摘要 ✓
    │ 失败 ↓
Level 2: 主模型重试 (30s 超时, 429/503 快速跳过)
    │ 成功 → 返回 LLM 摘要 ✓
    │ 失败 ↓
Level 3: 静态回退 (提取文件路径+工具名, 零 LLM 成本)
    │ 必定成功 → 返回结构化摘要 ✓

2.2 Level 1：辅助模型

辅助模型是压缩的首选方案：

class AuxiliaryClient:
    """轻量级辅助模型客户端，专门用于上下文压缩"""

    def __init__(self, model_id, api_key, base_url=None):
        self.model_id = model_id
        self.api_key = api_key
        self.cooldown_until = 0  # 冷却期时间戳

    async def summarize(self, messages, focus_topic=None):
        """调用辅助模型生成摘要"""
        if time.time() < self.cooldown_until:
            raise CooldownError("Auxiliary model in cooldown")

        prompt = build_summary_prompt(messages, focus_topic)
        response = await self._call_api(prompt, timeout=45)
        return response.content

为什么用辅助模型而不是主模型？

维度	主模型	辅助模型
成本	高（按 token 计费）	低（廉价小模型）
延迟	中（排队+推理）	低（专用通道）
质量	高	中（摘要够用就行）
用途	核心推理	辅助任务

2.3 Level 2：主模型重试

当辅助模型失败时，用主模型做第二次尝试：

async def try_main_model_fallback(self, messages, focus_topic):
    """Level 2: 主模型重试"""
    try:
        # 关键：30s 超时，避免阻塞主循环
        prompt = build_summary_prompt(messages, focus_topic)
        response = await asyncio.wait_for(
            self.main_model.complete(prompt),
            timeout=30.0
        )
        return response.content
    except asyncio.TimeoutError:
        logger.warning("Main model timeout (30s)")
        return None
    except RateLimitError:
        # 429 限流：快速跳过，不等待
        logger.warning("Main model rate limited, skipping")
        return None

关键设计：

• 30 秒硬超时：主模型的首要职责是核心推理，不能为了压缩阻塞太久
• 429/503 快速跳过：限流和服务不可用时立即放弃，不做无谓等待

2.4 Level 3：静态回退（最精妙的设计）

当两个 LLM 都失败时，用纯代码提取关键信息，零 LLM 成本：

def generate_static_fallback(self, messages):
    """Level 3: 静态回退——零 LLM 成本的结构化摘要"""
    file_paths = []
    tool_names = []
    user_goals = []

    for msg in messages:
        # 提取文件路径
        content = msg.get("content", "")
        paths = extract_file_paths(content)
        file_paths.extend(paths)

        # 提取工具调用
        for tc in msg.get("tool_calls", []):
            name = tc.get("function", {}).get("name", "unknown")
            tool_names.append(name)

        # 提取用户请求（保留最后 3 条）
        if msg.get("role") == "user":
            user_goals.append(content[:200])

    # 构建结构化摘要
    summary = f"""[自动摘要 - 静态回退模式]
用户最近请求: {'; '.join(user_goals[-3:])}
操作过的文件: {', '.join(set(file_paths[-20:]))}
使用过的工具: {', '.join(set(tool_names))}
消息总数: {len(messages)} 条
"""
    return summary

静态回退的优势：

1. 零延迟：纯字符串操作，微秒级完成
2. 零成本：不调用任何 LLM API
3. 零失败：不可能出错（纯代码逻辑）
4. 保留关键信息：文件路径、工具名称、用户请求——足以让模型知道"之前做了什么"

2.5 静态回退的实际效果

[自动摘要 - 静态回退模式]
用户最近请求: 分析半导体板块走势; 帮我看腾讯的港股; 给出投资建议
操作过的文件: /data/reports/semiconductor.pdf, /data/stock/00700.HK.csv
使用过的工具: stock_query, web_search, read_file, write_file
消息总数: 47 条

虽然不如 LLM 摘要那样自然流畅，但关键信息都保留了：用户做了什么、操作了哪些文件、用了哪些工具。

---

三、冷却协调机制

3.1 问题：每轮都尝试两个模型

如果辅助模型持续不可用（比如余额耗尽），每轮压缩都会：

1. 尝试辅助模型 → 失败
2. 尝试主模型 → 成功

这意味着每轮都有两次 API 调用，其中第一次必定失败。

3.2 解决方案：冷却期

# 辅助模型进入冷却期后，直接跳到 Level 2
async def compress(self, messages, focus_topic=None):
    # Level 1: 辅助模型
    try:
        return await self.auxiliary.summarize(messages, focus_topic)
    except CooldownError:
        logger.debug("Auxiliary in cooldown, trying main model")
    except Exception as e:
        logger.warning(f"Auxiliary failed: {e}")
        # 辅助模型失败 → 进入 5 分钟冷却期
        self.auxiliary.cooldown_until = time.time() + 300

    # Level 2: 主模型
    result = await self.try_main_model_fallback(messages, focus_topic)
    if result:
        return result

    # Level 3: 静态回退
    return self.generate_static_fallback(messages)

冷却期逻辑：

• 辅助模型失败 → 进入 5 分钟冷却期
• 冷却期内直接跳到 Level 2，不做无谓尝试
• 冷却期过后自动恢复尝试

---

四、依赖注入：为什么不能直接调 LLM

4.1 反面案例

# 错误做法：直接依赖具体的 LLM 调用方式
class ContextEngine:
    async def summarize(self, messages):
        # 直接调用 litellm → 紧耦合
        from litellm import completion
        return await completion(model="gpt-4o-mini", messages=messages)

4.2 正确做法：通过 ModelRegistry 注入

# 正确做法：通过依赖注入获取模型客户端
class ContextEngine:
    def __init__(self, model_registry):
        self.registry = model_registry  # 注入模型注册表

    async def summarize(self, messages):
        # 通过注册表获取辅助模型
        aux_client = self.registry.get_auxiliary_client()
        main_client = self.registry.get_main_client()

        # Level 1
        try:
            return await aux_client.summarize(messages)
        except Exception:
            pass

        # Level 2
        try:
            return await main_client.complete(messages)
        except Exception:
            pass

        # Level 3
        return self.generate_static_fallback(messages)

好处：

1. 可测试性：Mock 注入，轻松模拟各种故障场景
2. 可配置性：不同场景使用不同的辅助模型
3. 可替换性：更换 LLM 提供商时只需修改注册表

---

五、实测：模拟辅助模型故障

5.1 测试场景

@pytest.mark.asyncio
async def test_fallback_to_main_model():
    """辅助模型失败时降级到主模型"""
    # Mock 辅助模型抛异常
    aux_client = MockAuxiliaryClient(side_effect=TimeoutError)
    main_client = MockMainClient(return_value="Summary from main model")

    engine = ContextEngine(model_registry=MockRegistry(aux_client, main_client))
    result = await engine.compress(test_messages)

    assert result == "Summary from main model"
    assert aux_client.call_count == 1
    assert main_client.call_count == 1

@pytest.mark.asyncio
async def test_fallback_to_static():
    """两个 LLM 都失败时降级到静态回退"""
    aux_client = MockAuxiliaryClient(side_effect=TimeoutError)
    main_client = MockMainClient(side_effect=RateLimitError)

    engine = ContextEngine(model_registry=MockRegistry(aux_client, main_client))
    result = await engine.compress(test_messages)

    assert "自动摘要" in result
    assert "静态回退" in result
    assert "stock_query" in result  # 工具名被提取

@pytest.mark.asyncio
async def test_cooldown_skip():
    """冷却期内跳过辅助模型"""
    aux_client = MockAuxiliaryClient(side_effect=TimeoutError)
    main_client = MockMainClient(return_value="Summary")

    engine = ContextEngine(model_registry=MockRegistry(aux_client, main_client))

    # 第一次调用：辅助失败，进入冷却
    await engine.compress(test_messages)
    assert aux_client.call_count == 1

    # 第二次调用：冷却中直接跳过辅助模型
    await engine.compress(test_messages)
    assert aux_client.call_count == 1  # 没有再次调用！
    assert main_client.call_count == 2

5.2 运行结果

$ pytest tests/test_fallback.py -v
test_fallback_to_main_model    PASSED
test_fallback_to_static        PASSED
test_cooldown_skip             PASSED

---

六、总结与展望

6.1 核心要点回顾

1. 压缩不能失败：三级容灾确保任何情况下都有摘要产出
2. 静态回退是兜底：零 LLM 成本也能保留文件路径和工具名称
3. 冷却期避免浪费：辅助模型故障后不做无谓重试

6.2 一个设计原则

"系统最脆弱的环节决定了整体的可靠性。在 LLM Agent 中，外部 API 是最不可靠的环节——因此你的容灾设计必须假设 API 随时可能挂掉。"

---

下期预告：《Token-Budget 尾部保护：别再按"轮次"保护上下文了》

• 为什么"保护最近 12 轮"是个糟糕的策略
• Token-budget 如何按实际 token 消耗精确保护
• 边界对齐和用户消息锚定的实现细节

敬请期待！

---

版权声明：本文为 CSDN 博主「翁勇刚」的原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接及本声明。