第 30 章：分层模型策略

"分层路由能节省 80% 成本"——但前提是你能正确判断任务复杂度，这比听起来难得多。

⏱️ 快速通道（5 分钟掌握核心）

1. 三层模型：Small（50%流量）→ Medium（40%）→ Large（10%）
2. 复杂度判断：规则优先（关键词/长度），LLM 兜底（模糊任务）
3. 升级机制：Small 输出质量差时自动升级到 Medium/Large
4. 成本监控：按层统计 Token 消耗，Large 占比超 20% 要排查
5. 缓存优先：相同 Query 命中缓存，比模型分层更省钱

10 分钟路径：30.1-30.3 → 30.5 → Shannon Lab

30.1 从一个账单说起

你的 AI Agent 系统上线三个月，老板找你谈话：

"这个月 LLM 账单 $15,000，比预算高了 10 倍。"

你拉出使用日志，发现问题：

[2025-01-10 09:15:22] 用户问: "今天周几?"
                       模型: claude-opus-4-1
                       消耗: 0.12 USD

[2025-01-10 09:15:45] 用户问: "帮我分析这份 200 页财报的核心风险点"
                       模型: claude-opus-4-1
                       消耗: 0.35 USD

两个任务用了同样昂贵的顶级模型。问"今天周几"花了 $0.12——这个价格，Haiku 能回答 200 次。

问题的本质：你的系统在"过度服务"——用 Rolls-Royce 接驾送菜，用火箭炮打蚊子。

这章我们解决一个问题：如何让对的任务用对的模型。

30.2 三层模型架构

为什么是三层？

三层的设计来自实践观察——大多数生产流量可以归类到这三个桶里：

目标分布：50% Small / 40% Medium / 10% Large。

这不是强制配额——是参考基准。实际分布取决于你的业务场景。但如果你的 Large 占比超过 20%，说明复杂度判断可能有问题。

配置结构

📦 实现参考 (Shannon): config/models.yaml - model_tiers 配置

# Shannon 三层模型配置
model_tiers:
  small:
    # 目标占比: 50% - 快速低成本，处理基础任务
    providers:
      - provider: anthropic
        model: claude-haiku-4-5-20251015
        priority: 1  # 首选
      - provider: openai
        model: gpt-5-nano-2025-08-07
        priority: 2  # 备选
      - provider: xai
        model: grok-3-mini
        priority: 3
      - provider: google
        model: gemini-2.5-flash-lite
        priority: 4

  medium:
    # 目标占比: 40% - 能力/成本平衡
    providers:
      - provider: anthropic
        model: claude-sonnet-4-5-20250929
        priority: 1
      - provider: openai
        model: gpt-5-mini-2025-08-07
        priority: 2
      - provider: xai
        model: grok-4-fast-non-reasoning
        priority: 3
      - provider: google
        model: gemini-2.5-flash
        priority: 4

  large:
    # 目标占比: 10% - 深度推理任务
    providers:
      - provider: openai
        model: gpt-5.1
        priority: 1
      - provider: anthropic
        model: claude-opus-4-1-20250805
        priority: 2
      - provider: xai
        model: grok-4-fast-reasoning
        priority: 3
      - provider: google
        model: gemini-2.5-pro
        priority: 4

优先级说明：数字越小优先级越高。同层级内按优先级依次尝试——首选不可用时自动切换到备选。

30.3 成本与能力的数学

实际定价对比

⚠️ 时效性提示 (2026-01): 模型定价和能力列表频繁变化。以下为示意配置，请查阅各厂商官网获取最新价格：OpenAI Pricing | Anthropic Pricing | Google AI Pricing

关键洞察：Opus 的单次调用成本是 Haiku 的 150 倍。

成本节省计算

假设每月 100 万次 API 调用，平均每次 1000 tokens：

场景 A：全用 Large 模型

1M * 1K tokens * ($0.015 + $0.075) / 1K = $90,000/月

场景 B：智能分层 (50/40/10)

Small:  500K * 1K * ($0.0001 + $0.0005) / 1K = $300
Medium: 400K * 1K * ($0.0003 + $0.0015) / 1K = $720
Large:  100K * 1K * ($0.015 + $0.075) / 1K   = $9,000
总计: $10,020/月

节省: $90,000 - $10,020 = $79,980/月 (89% 降低)

这就是分层策略的威力。但前提是——你得能准确判断哪些任务该用哪个层级。

30.4 复杂度分析：路由的核心

复杂度阈值配置

# 复杂度→层级映射
workflows:
  complexity:
    simple_threshold: 0.3   # complexity < 0.3 → small
    medium_threshold: 0.5   # 0.3 ≤ complexity < 0.5 → medium
                            # complexity ≥ 0.5 → large

启发式复杂度计算

📦 实现参考 (Shannon): llm_service/api/complexity.py - _heuristic_analysis 函数

def calculate_complexity(query: str, context: Dict) -> float:
    """
    计算任务复杂度分数 (0.0 - 1.0)

    启发式规则，不依赖 LLM 调用——用于快速路由决策。
    """
    score = 0.0
    query_lower = query.lower()

    # 1. 查询长度（长查询通常更复杂）
    word_count = len(query.split())
    if word_count > 100:
        score += 0.2
    elif word_count > 50:
        score += 0.1

    # 2. 关键词检测
    complex_keywords = [
        "analyze", "compare", "evaluate", "synthesize",
        "design", "architect", "optimize", "debug",
        "explain why", "trade-offs", "implications",
    ]
    simple_keywords = [
        "what is", "define", "list", "format",
        "convert", "translate", "summarize",
    ]

    for kw in complex_keywords:
        if kw in query_lower:
            score += 0.15

    for kw in simple_keywords:
        if kw in query_lower:
            score -= 0.1

    # 3. 上下文信息
    if context.get("requires_reasoning"):
        score += 0.3
    if context.get("requires_code_generation"):
        score += 0.2
    if context.get("multi_step"):
        score += 0.2
    if context.get("available_tools") and len(context["available_tools"]) > 5:
        score += 0.1  # 多工具场景通常更复杂

    return max(0.0, min(1.0, score))


def select_tier(complexity: float, config: Dict) -> str:
    """根据复杂度选择模型层级"""
    simple_threshold = config.get("simple_threshold", 0.3)
    medium_threshold = config.get("medium_threshold", 0.5)

    if complexity < simple_threshold:
        return "small"
    elif complexity < medium_threshold:
        return "medium"
    else:
        return "large"

基于模型的复杂度分析

启发式规则快但不够准确。对于重要决策，可以用小模型先做复杂度判断：

async def model_based_complexity(query: str, providers) -> Dict:
    """用小模型分析复杂度，成本约 $0.0001"""

    sys_prompt = (
        "You classify tasks into simple, standard, or complex. "
        "IMPORTANT: Tasks requiring calculations or tool usage "
        "must be 'standard' mode (not 'simple'). "
        "Simple mode is ONLY for direct Q&A without tools. "
        'Respond with JSON: {"recommended_mode": ..., '
        '"complexity_score": 0..1, "reasoning": ...}'
    )

    result = await providers.generate_completion(
        messages=[
            {"role": "system", "content": sys_prompt},
            {"role": "user", "content": f"Query: {query}"},
        ],
        tier=ModelTier.SMALL,  # 用小模型判断
        max_tokens=256,
        temperature=0.0,
        response_format={"type": "json_object"},
    )

    return json.loads(result.get("output_text", "{}"))

成本权衡：模型判断更准确，但每次多花 ~$0.0001。对于 Large 模型（$0.09/次）来说，避免一次误判就能覆盖 900 次判断成本。

30.5 LLM Manager：统一路由层

核心架构

📦 实现参考 (Shannon): llm_provider/manager.py - LLMManager 类

class LLMManager:
    """
    统一 LLM 管理：
    - Provider 注册和路由
    - 模型分层和选择
    - 缓存和限流
    - Token 预算控制
    - 使用量追踪
    """

    def __init__(self, config_path: Optional[str] = None):
        self.registry = LLMProviderRegistry()
        self.cache = CacheManager(max_size=1000)
        self.rate_limiters: Dict[str, RateLimiter] = {}
        self._breakers: Dict[str, _CircuitBreaker] = {}

        # Token 使用量追踪
        self.session_usage: Dict[str, TokenUsage] = {}
        self.task_usage: Dict[str, TokenUsage] = {}

        # Tier 路由偏好
        self.tier_preferences: Dict[str, List[str]] = {}

        # 加载配置
        if config_path:
            self.load_config(config_path)

Provider 选择逻辑

def _select_provider(self, request: CompletionRequest) -> tuple[str, LLMProvider]:
    """为请求选择最佳 provider"""

    # 1. 显式指定 provider（最高优先级）
    if request.provider_override:
        provider_name = request.provider_override
        if provider_name not in self.registry.providers:
            raise ValueError(f"Invalid provider_override: {provider_name}")
        if self._is_breaker_open(provider_name):
            raise RuntimeError(f"Provider '{provider_name}' circuit breaker is open")
        return provider_name, self.registry.providers[provider_name]

    # 2. 显式指定模型，找对应 provider
    if request.model:
        for pname, pprovider in self.registry.providers.items():
            if self._is_breaker_open(pname):
                continue
            if request.model in pprovider.models:
                return pname, pprovider
        # 找不到就清空，回退到层级选择
        request.model = None

    # 3. 按层级偏好选择
    tier_prefs = self.tier_preferences.get(request.model_tier.value, [])

    for pref in tier_prefs:
        if ":" in pref:
            provider_name, model_id = pref.split(":", 1)
            if provider_name in self.registry.providers:
                if self._is_breaker_open(provider_name):
                    continue
                provider = self.registry.providers[provider_name]
                if model_id in provider.models:
                    request.model = model_id  # 锁定模型
                    return provider_name, provider

    # 4. 回退到 registry 默认选择
    return self.registry.select_provider_for_request(request)

设计要点：

1. Override 最优先：允许调用方强制指定 provider/model
2. Circuit Breaker 感知：跳过不健康的 provider
3. 层级路由：按优先级尝试同层级内的多个选项
4. 优雅降级：找不到完美匹配时有 fallback

30.6 Fallback 与熔断

Circuit Breaker 模式

当某个 provider 连续失败，自动熔断并切换到备选：

class _CircuitBreaker:
    """
    状态机：closed → open → half-open → closed

    - closed: 正常工作，记录失败次数
    - open: 熔断状态，拒绝所有请求
    - half-open: 冷却后允许探测请求
    """

    def __init__(
        self,
        name: str,
        failure_threshold: int = 5,      # 连续失败 N 次触发熔断
        recovery_timeout: float = 60.0,  # 熔断冷却时间（秒）
    ):
        self.name = name
        self.failure_threshold = max(1, failure_threshold)
        self.recovery_timeout = recovery_timeout
        self.failures = 0
        self.state = "closed"
        self.opened_at = 0.0

    def allow(self) -> bool:
        if self.state == "closed":
            return True
        if self.state == "open":
            # 冷却后进入 half-open，加入抖动避免惊群
            jitter = self.recovery_timeout * random.uniform(-0.1, 0.1)
            if (time.time() - self.opened_at) >= (self.recovery_timeout + jitter):
                self.state = "half-open"
                return True  # 允许一次探测
            return False
        return True  # half-open 允许探测

    def on_success(self):
        if self.state in ("open", "half-open"):
            self._close()
        self.failures = 0

    def on_failure(self, transient: bool):
        if not transient:
            return  # 非瞬态错误不计入
        self.failures += 1
        if self.failures >= self.failure_threshold and self.state != "open":
            self._open()

Fallback 选择

def _get_fallback_provider(
    self, failed_provider: str, tier: ModelTier
) -> Optional[tuple[str, LLMProvider]]:
    """主 provider 失败时选择备选"""

    tier_prefs = self.tier_preferences.get(tier.value, [])

    for pref in tier_prefs:
        provider_name = pref.split(":")[0] if ":" in pref else pref
        if (
            provider_name != failed_provider
            and provider_name in self.registry.providers
            and not self._is_breaker_open(provider_name)
        ):
            return provider_name, self.registry.providers[provider_name]

    return None  # 没有可用备选

关键设计：

• 同层级内按优先级选择备选
• 跳过已熔断的 provider
• 返回 None 让上层决定是否降级到其他层

30.7 模型能力矩阵

不是所有模型都能做所有事。有些任务需要视觉理解，有些需要深度推理。

能力标记

📦 实现参考 (Shannon): config/models.yaml - model_capabilities 配置

model_capabilities:
  # 支持图片输入的模型
  multimodal_models:
    - gpt-5.1
    - gpt-5-pro-2025-08-07
    - claude-sonnet-4-5-20250929
    - gemini-2.5-flash
    - gemini-2.0-flash

  # 支持深度推理/thinking 的模型
  thinking_models:
    - gpt-5-pro-2025-08-07
    - gpt-5.1
    - claude-opus-4-1-20250805
    - gemini-2.5-pro
    - deepseek-r1
    - grok-4-fast-reasoning

  # 编程能力强的模型
  coding_specialists:
    - codestral-22b-v0.1
    - deepseek-v3.2
    - claude-sonnet-4-5-20250929
    - gpt-5.1

  # 支持超长上下文的模型
  long_context_models:
    - llama-4-scout-17b-16e-instruct  # 10M tokens
    - gemini-2.5-flash               # 1M tokens
    - claude-sonnet-4-5-20250929     # 200K tokens

能力感知路由

def select_model_by_capability(
    requirement: str,
    capabilities: Dict[str, List[str]],
    tier_preferences: Dict[str, List[str]],
) -> str:
    """根据任务需求选择合适的模型"""

    # 检测需求
    needs_vision = "image" in requirement.lower() or "screenshot" in requirement.lower()
    needs_reasoning = any(
        kw in requirement.lower()
        for kw in ["analyze", "evaluate", "trade-off", "why"]
    )
    needs_coding = any(
        kw in requirement.lower()
        for kw in ["code", "implement", "debug", "function"]
    )
    needs_long_context = len(requirement) > 50000

    # 筛选满足需求的模型
    candidates = set()

    if needs_vision:
        candidates.update(capabilities.get("multimodal_models", []))
    if needs_reasoning:
        candidates.update(capabilities.get("thinking_models", []))
    if needs_coding:
        candidates.update(capabilities.get("coding_specialists", []))
    if needs_long_context:
        candidates.update(capabilities.get("long_context_models", []))

    if not candidates:
        # 默认返回 medium 首选
        return tier_preferences.get("medium", [])[0].split(":")[1]

    # 从候选中选择最经济的（按层级从低到高）
    for tier in ["small", "medium", "large"]:
        for pref in tier_preferences.get(tier, []):
            model = pref.split(":")[1] if ":" in pref else pref
            if model in candidates:
                return model

    return list(candidates)[0]

核心逻辑：能力匹配 > 成本优化。先确保模型能做这件事，再考虑成本。

30.8 速率限制

按层级差异化限制

rate_limits:
  default_rpm: 60    # 默认每分钟请求数
  default_tpm: 100000  # 默认每分钟 token 数

  tier_overrides:
    small:
      rpm: 120        # 快速模型允许更高频率
      tpm: 200000
    medium:
      rpm: 60
      tpm: 100000
    large:
      rpm: 30         # 复杂模型限制频率
      tpm: 50000

设计思路：

• Small 模型快且便宜，可以更激进
• Large 模型慢且贵，限制频率防止账单失控

Token Bucket 限流器

class RateLimiter:
    """Token bucket 限流器"""

    def __init__(self, requests_per_minute: int):
        self.requests_per_minute = requests_per_minute
        self.tokens = requests_per_minute
        self.last_refill = time.time()
        self._lock = asyncio.Lock()

    async def acquire(self):
        async with self._lock:
            now = time.time()
            elapsed = now - self.last_refill

            # 补充 token（按时间流逝）
            refill_amount = elapsed * (self.requests_per_minute / 60.0)
            self.tokens = min(self.requests_per_minute, self.tokens + refill_amount)
            self.last_refill = now

            if self.tokens >= 1:
                self.tokens -= 1
                return True

            # 等待足够 token
            wait_time = (1 - self.tokens) / (self.requests_per_minute / 60.0)
            await asyncio.sleep(wait_time)
            self.tokens = 0
            return True

30.9 集中式定价管理

定价配置

# 集中式模型定价（USD per 1K tokens）
# 用于成本追踪和预算控制
pricing:
  defaults:
    combined_per_1k: 0.005  # 未知模型的默认值

  models:
    openai:
      gpt-5-nano-2025-08-07:
        input_per_1k: 0.00005
        output_per_1k: 0.00040
      gpt-5-mini-2025-08-07:
        input_per_1k: 0.00025
        output_per_1k: 0.00200
      gpt-5.1:
        input_per_1k: 0.00125
        output_per_1k: 0.01000

    anthropic:
      claude-haiku-4-5-20251015:
        input_per_1k: 0.00010
        output_per_1k: 0.00050
      claude-sonnet-4-5-20250929:
        input_per_1k: 0.00030
        output_per_1k: 0.00150
      claude-opus-4-1-20250805:
        input_per_1k: 0.0150
        output_per_1k: 0.0750

成本追踪

def _update_usage_tracking(
    self, request: CompletionRequest, response: CompletionResponse
):
    """更新使用量追踪"""

    # 按会话追踪
    if request.session_id:
        if request.session_id not in self.session_usage:
            self.session_usage[request.session_id] = TokenUsage(0, 0, 0, 0.0)
        self.session_usage[request.session_id] += response.usage

    # 按任务追踪
    if request.task_id:
        if request.task_id not in self.task_usage:
            self.task_usage[request.task_id] = TokenUsage(0, 0, 0, 0.0)
        self.task_usage[request.task_id] += response.usage

可观测性集成

# Prometheus 指标
LLM_MANAGER_COST = Counter(
    "llm_manager_cost_usd_total",
    "Accumulated cost tracked by manager (USD)",
    labelnames=("provider", "model"),
)

# 每次调用后记录
if _METRICS_ENABLED:
    LLM_MANAGER_COST.labels(
        response.provider, response.model
    ).inc(max(0.0, float(response.usage.estimated_cost)))

监控层级分布：llm_requests_total{tier="small|medium|large"}

30.10 常见的坑

坑 1：过度依赖复杂度估计

复杂度估计不准导致模型选择错误：

# 错误：只用复杂度，不验证结果
tier = select_tier_by_complexity(query)
response = await llm.complete(query, tier=tier)

# 正确：加入验证和升级机制
tier = select_tier_by_complexity(query)
response = await llm.complete(query, tier=tier)
if response.confidence < 0.7 or response.quality_score < threshold:
    # 升级到更大模型重试
    tier = upgrade_tier(tier)
    response = await llm.complete(query, tier=tier)

经验值：对于重要任务，预留 10-20% 的"升级预算"。

坑 2：忽略模型能力差异

某些任务只有特定模型能做好：

# 错误：只看成本
tier = "small"  # 最便宜

# 正确：检查能力匹配
if has_image_input(query):
    model = select_from_multimodal_models()
elif needs_deep_reasoning(query):
    model = select_from_thinking_models()
elif is_coding_task(query):
    model = select_from_coding_specialists()
else:
    tier = select_tier_by_complexity(query)

坑 3：缺少 Fallback

首选 provider 不可用时任务失败：

# 错误：只用一个 provider
response = await anthropic.complete(query)

# 正确：自动 fallback
try:
    response = await manager.complete(query, tier="medium")
    # 自动按优先级尝试多个 provider
except AllProvidersUnavailable:
    # 降级到其他层
    response = await manager.complete(query, tier="small")

坑 4：静态复杂度判断

用户问题经常超出预期：

# 错误：一次性判断
tier = classify_once(query)

# 正确：动态调整
tier = initial_classification(query)
response = await llm.complete(query, tier=tier)

# 基于结果质量调整
if needs_more_capability(response, query):
    tier = upgrade_tier(tier)
    response = await llm.complete(query, tier=tier)

坑 5：忽略提示词缓存

重复提示词浪费钱：

# 启用提示词缓存
prompt_cache:
  enabled: true
  similarity_threshold: 0.95
  ttl_seconds: 3600
  max_cache_size_mb: 2048

对于 System Prompt 不变的场景，缓存能节省 50%+ 的输入成本。

30.11 框架对比

回顾

1. 三层架构：Small (50%) / Medium (40%) / Large (10%) 是黄金分布
2. 复杂度路由：启发式快速判断 + 可选模型验证
3. 能力矩阵：先匹配能力，再优化成本
4. 韧性设计：Circuit Breaker + 自动 Fallback
5. 可观测性：追踪层级分布，发现成本异常

Shannon Lab（10 分钟上手）

本节帮你在 10 分钟内把本章概念对应到 Shannon 源码。

必读（1 个文件）

• config/models.yaml：三层模型配置 + 能力矩阵 + 定价，是"分层策略"真正落地的地方

选读深挖（2 个，按兴趣挑）

• python/llm-service/llm_provider/manager.py：LLMManager 怎么做路由、熔断、Fallback
• python/llm-service/llm_service/api/complexity.py：复杂度分析 API（你会发现"判断任务难度"比写路由更难）

练习

练习 1：复杂度分析器

实现一个复杂度分析器，能区分以下查询的层级：

• "今天周几？" → Small
• "总结这篇 2000 字文章" → Medium
• "分析这份财报的风险点并给出投资建议" → Large

要求：

1. 基于关键词和长度的启发式规则
2. 输出复杂度分数 (0-1) 和推荐层级
3. 写测试用例验证准确率

练习 2：成本追踪 Dashboard

基于 Prometheus 指标设计一个成本追踪仪表盘：

• 按层级显示请求分布
• 按 provider 显示成本趋势
• 设置成本超标告警 (daily_budget_usd)

练习 3：动态升级策略

实现一个"试探性路由"策略：

1. 首次用 Small 模型尝试
2. 如果响应质量不达标（如 confidence < 0.7），升级到 Medium
3. 如果仍不达标，升级到 Large
4. 记录升级次数，用于优化初始判断

进一步阅读

• Token 预算控制 - 参见第 23 章
• 可观测性与监控 - 参见第 22 章
• Provider 配置实践 - 参见 Shannon config/models.yaml

Part 9 总结

Part 9 探讨了 AI Agent 的前沿实践：

这些能力结合企业级基础设施（Part 7-8），构成了完整的生产级 AI Agent 平台。

从单体 Agent 到企业级多智能体系统，核心挑战不变：如何在能力、成本、可靠性之间找到平衡。分层模型策略是这个平衡的关键杠杆——用对的模型做对的事，既是技术问题，也是架构哲学。