AI 推理成本治理：从模型量化到请求调度的全链路降本策略

一、AI 推理成本的冰山模型：显性支出与隐性浪费

AI 推理成本远不止 GPU 租赁费用这一项显性支出。完整的成本冰山分为三层。水面之上的显性层是 GPU 计算费用：按 A100 单价 25 元/小时计算，一个 7B 模型全年无休运行的成本约 22 万元。水面附近的半隐性层是冷启动和空闲浪费：GPU 实例在低峰期大量空闲，利用率通常不到 30%；扩容后的冷启动期间 GPU 已计费但未产出推理结果。水面之下的隐性层是冗余调用和过度推理：同一请求被重复推理、简单问题调用大模型而非小模型、长上下文请求消耗大量 KV Cache 但有效信息密度低。

三层成本中，显性层只占冰山的 40%，半隐性层和隐性层合计占 60%。只优化 GPU 单价而不治理半隐性和隐性浪费，降本效果有限。必须从模型层、调度层和请求层三个维度构建全链路降本策略。

二、全链路降本架构：模型量化、智能路由与请求优化

flowchart TD
    REQ[推理请求] --> ROUTER[智能路由层<br/>请求复杂度评估]

    subgraph 模型层降本
        QUANT[模型量化<br/>FP16 -> INT8/INT4]
        DISTILL[模型蒸馏<br/>大模型 -> 小模型]
        SPEC[投机解码<br/>小模型草拟 + 大模型校验]
    end

    subgraph 调度层降本
        ROUTER
        SCHED[动态调度<br/>按请求复杂度分配模型]
        BATCH[动态批处理<br/>合并相似请求]
    end

    subgraph 请求层降本
        CACHE[语义缓存<br/>相似问题复用结果]
        COMPRESS[Prompt 压缩<br/>去除冗余上下文]
        DEDUP[请求去重<br/>相同请求不重复推理]
    end

    ROUTER -->|简单问题| SMALL[小模型<br/>7B INT4<br/>成本: 0.5x]
    ROUTER -->|复杂问题| LARGE[大模型<br/>72B FP16<br/>成本: 1x]
    ROUTER -->|缓存命中| CACHE_HIT[直接返回<br/>成本: 0x]

    SMALL --> BATCH
    LARGE --> BATCH

    DEDUP --> ROUTER
    CACHE --> ROUTER
    COMPRESS --> ROUTER

    style ROUTER fill:#e74c3c,color:#fff
    style QUANT fill:#27ae60,color:#fff
    style CACHE fill:#3498db,color:#fff
    style DEDUP fill:#e67e22,color:#fff

模型层降本：量化是最直接的降本手段。将模型权重从 FP16 量化到 INT8，显存占用减半，推理速度提升约 40%，精度损失通常在 1% 以内。INT4 量化进一步将显存压缩到 1/4，但精度损失增大到 3-5%，需要根据业务容忍度选择。投机解码是一种更精巧的策略：先用小模型快速生成候选 Token，再用大模型并行校验，命中则直接接受，未命中则回退到大模型逐 Token 生成。在候选命中率较高时（如代码补全场景），推理速度可提升 2-3 倍。

调度层降本：智能路由根据请求的复杂度将请求分配到不同规模的模型。简单问题（如 FAQ 回答、格式转换）路由到 7B 小模型，复杂问题（如多步推理、代码生成）路由到 72B 大模型。通过请求分类，约 60% 的请求可以被小模型处理，大模型的 GPU 占用减少 60%。

请求层降本：语义缓存对相似问题复用已有推理结果。传统缓存基于精确匹配，但自然语言表述千变万化，精确匹配命中率极低。语义缓存通过 Embedding 向量相似度匹配，将语义相近的请求映射到同一缓存条目，命中率可从 5% 提升到 30%。

三、生产级降本实现

3.1 模型量化部署

"""
模型量化部署脚本
核心设计：使用 AWQ 算法进行激活感知量化
相比朴素的 RTN 量化，AWQ 保护了对推理影响最大的权重通道
在 INT4 精度下保持更好的模型质量
"""
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
from awq import AutoAWQForCausalLM


class QuantizedModelDeployer:
    """量化模型部署器"""

    def __init__(self, model_path: str):
        self.model_path = model_path

    def quantize_to_int4(
        self,
        output_path: str,
        calibration_data: list
    ):
        """
        AWQ INT4 量化
        calibration_data: 校准数据集，用于识别重要的权重通道
        量化过程不是简单的截断，而是通过校准数据感知激活分布
        保护对输出影响最大的 1% 权重通道不进行量化
        """
        model = AutoAWQForCausalLM.from_pretrained(
            self.model_path
        )
        tokenizer = AutoTokenizer.from_pretrained(
            self.model_path, trust_remote_code=True
        )

        quant_config = {
            "zero_point": True,
            "q_group_size": 128,
            "w_bit": 4,
            # AWQ 的核心参数：保护比例
            # 保留 1% 最重要的权重通道为 FP16
            # 这 1% 的通道贡献了 90% 以上的激活值
            "version": "GEMM"
        }

        model.quantize(
            tokenizer,
            quant_config=quant_config,
            calib_data=calibration_data
        )

        model.save_quantized(output_path)
        tokenizer.save_pretrained(output_path)

        # 输出量化效果统计
        original_size = self._get_model_size(self.model_path)
        quantized_size = self._get_model_size(output_path)
        compression_ratio = original_size / quantized_size

        print(f"原始模型大小: {original_size / 1e9:.1f} GB")
        print(f"量化模型大小: {quantized_size / 1e9:.1f} GB")
        print(f"压缩比: {compression_ratio:.1f}x")

    def benchmark_quantized_model(
        self, model_path: str, test_prompts: list
    ):
        """
        量化模型基准测试
        对比量化前后的推理速度和输出质量
        """
        model = AutoAWQForCausalLM.from_quantized(
            model_path, fuse_layers=True
        )
        tokenizer = AutoTokenizer.from_pretrained(
            model_path, trust_remote_code=True
        )

        total_tokens = 0
        total_time = 0

        for prompt in test_prompts:
            inputs = tokenizer(prompt, return_tensors="pt").to(
                model.device
            )

            start = torch.cuda.Event(enable_timing=True)
            end = torch.cuda.Event(enable_timing=True)

            start.record()
            outputs = model.generate(
                **inputs,
                max_new_tokens=256,
                do_sample=False
            )
            end.record()

            torch.cuda.synchronize()
            elapsed_ms = start.elapsed_time(end)

            generated_tokens = outputs.shape[1] - inputs.shape[1]
            total_tokens += generated_tokens
            total_time += elapsed_ms / 1000

        throughput = total_tokens / total_time
        print(f"吞吐量: {throughput:.1f} tokens/s")
        print(f"平均延迟: {total_time / len(test_prompts) * 1000:.0f} ms")

    @staticmethod
    def _get_model_size(path: str) -> int:
        import os
        total = 0
        for dirpath, _, filenames in os.walk(path):
            for f in filenames:
                if f.endswith(('.bin', '.safetensors')):
                    total += os.path.getsize(
                        os.path.join(dirpath, f)
                    )
        return total

3.2 智能路由：请求复杂度评估与模型分配

"""
智能路由器
核心设计：根据请求复杂度将请求路由到不同规模的模型
简单问题用小模型，复杂问题用大模型
降低大模型的 GPU 占用，实现成本与质量的平衡
"""
import hashlib
import logging
from dataclasses import dataclass
from enum import Enum
from typing import Optional

logger = logging.getLogger(__name__)


class ModelTier(Enum):
    """模型层级"""
    SMALL = "7b-int4"       # 小模型：简单问答、格式转换
    MEDIUM = "14b-int8"     # 中模型：摘要、翻译
    LARGE = "72b-fp16"      # 大模型：多步推理、代码生成


@dataclass
class RoutingDecision:
    """路由决策"""
    model_tier: ModelTier
    reason: str
    estimated_cost: float  # 预估推理成本（元）


class IntelligentRouter:
    """智能路由器"""

    # 复杂度评估规则
    # 这些阈值基于生产环境的 A/B 测试数据校准
    COMPLEXITY_RULES = {
        # 简单问题特征
        "simple_patterns": [
            "什么是", "解释一下", "翻译", "总结以下",
            "格式化", "提取", "分类"
        ],
        # 中等复杂度特征
        "medium_patterns": [
            "分析", "对比", "优化", "设计一个",
            "写一段", "评估"
        ],
        # 高复杂度特征
        "complex_patterns": [
            "多步推理", "代码生成", "架构设计",
            "调试", "重构", "性能优化方案"
        ]
    }

    def route(self, prompt: str, max_tokens: int = 512
              ) -> RoutingDecision:
        """
        根据请求内容评估复杂度，决定路由到哪个模型
        评估维度：问题类型、输入长度、输出长度要求
        """
        # 维度1：问题类型匹配
        complexity_score = self._evaluate_complexity(prompt)

        # 维度2：输入长度影响
        # 输入越长，KV Cache 占用越大，需要更强的模型处理
        input_length = len(prompt)
        if input_length > 4000:
            complexity_score += 1
        elif input_length > 2000:
            complexity_score += 0.5

        # 维度3：输出长度要求
        # 长输出通常需要更强的推理能力
        if max_tokens > 1024:
            complexity_score += 0.5

        # 根据综合评分决定路由
        if complexity_score <= 1.0:
            return RoutingDecision(
                model_tier=ModelTier.SMALL,
                reason=f"简单问题, 评分={complexity_score:.1f}",
                estimated_cost=0.002
            )
        elif complexity_score <= 2.0:
            return RoutingDecision(
                model_tier=ModelTier.MEDIUM,
                reason=f"中等复杂度, 评分={complexity_score:.1f}",
                estimated_cost=0.008
            )
        else:
            return RoutingDecision(
                model_tier=ModelTier.LARGE,
                reason=f"高复杂度, 评分={complexity_score:.1f}",
                estimated_cost=0.03
            )

    def _evaluate_complexity(self, prompt: str) -> float:
        """评估问题复杂度，返回 0-3 的评分"""
        score = 0.5  # 基础分

        for pattern in self.COMPLEXITY_RULES["simple_patterns"]:
            if pattern in prompt:
                score = min(score, 0.5)
                break

        for pattern in self.COMPLEXITY_RULES["medium_patterns"]:
            if pattern in prompt:
                score += 0.5
                break

        for pattern in self.COMPLEXITY_RULES["complex_patterns"]:
            if pattern in prompt:
                score += 1.5
                break

        return score

3.3 语义缓存：相似问题复用推理结果

"""
语义缓存
核心设计：基于 Embedding 向量相似度匹配缓存
解决自然语言表述多样化导致精确匹配命中率低的问题
"""
import hashlib
import json
import time
from dataclasses import dataclass
from typing import Optional, List

import numpy as np


@dataclass
class CacheEntry:
    """缓存条目"""
    query_embedding: np.ndarray
    response: str
    model_tier: str
    created_at: float
    hit_count: int = 0
    ttl_seconds: int = 3600  # 默认缓存1小时

    @property
    def is_expired(self) -> bool:
        return time.time() - self.created_at > self.ttl_seconds


class SemanticCache:
    """语义缓存"""

    def __init__(
        self,
        similarity_threshold: float = 0.92,
        max_entries: int = 10000
    ):
        # 相似度阈值：0.92 表示语义高度相似
        # 阈值越高，缓存越精准但命中率越低
        # 阈值越低，命中率越高但可能返回不相关结果
        self.similarity_threshold = similarity_threshold
        self.max_entries = max_entries
        self.entries: List[CacheEntry] = []

    def get(
        self, query_embedding: np.ndarray
    ) -> Optional[CacheEntry]:
        """
        查询语义缓存
        使用余弦相似度匹配最相似的缓存条目
        """
        if not self.entries:
            return None

        # 过滤过期条目
        self.entries = [
            e for e in self.entries if not e.is_expired
        ]

        # 计算与所有缓存条目的余弦相似度
        best_entry = None
        best_similarity = 0.0

        for entry in self.entries:
            similarity = self._cosine_similarity(
                query_embedding, entry.query_embedding
            )
            if similarity > best_similarity:
                best_similarity = similarity
                best_entry = entry

        if best_similarity >= self.similarity_threshold:
            best_entry.hit_count += 1
            logger.info(
                f"语义缓存命中, 相似度={best_similarity:.3f}, "
                f"历史命中次数={best_entry.hit_count}"
            )
            return best_entry

        return None

    def put(
        self,
        query_embedding: np.ndarray,
        response: str,
        model_tier: str
    ):
        """写入语义缓存"""
        # LRU 淘汰：缓存满时移除最久未命中的条目
        if len(self.entries) >= self.max_entries:
            self.entries.sort(key=lambda e: e.hit_count)
            self.entries.pop(0)

        self.entries.append(CacheEntry(
            query_embedding=query_embedding,
            response=response,
            model_tier=model_tier,
            created_at=time.time()
        ))

    @staticmethod
    def _cosine_similarity(a: np.ndarray, b: np.ndarray) -> float:
        """计算余弦相似度"""
        norm_a = np.linalg.norm(a)
        norm_b = np.linalg.norm(b)
        if norm_a == 0 or norm_b == 0:
            return 0.0
        return float(np.dot(a, b) / (norm_a * norm_b))

四、降本策略的代价：精度损失、路由误判与缓存不一致

量化精度损失：INT4 量化在通用基准测试上精度损失约 3-5%，但在特定领域（如数学推理、代码生成）可能损失更大。对于精度敏感的业务（如金融风控、医疗诊断），INT4 量化的风险不可接受，只能使用 INT8 或 FP16。量化前必须在业务测试集上验证精度，而非仅依赖通用基准。

路由误判：智能路由基于关键词匹配评估复杂度，存在误判风险。简单问题被路由到大模型不会影响质量，但浪费成本；复杂问题被路由到小模型则直接影响输出质量。解决方案是设置置信度阈值——当路由器对复杂度评估不确定时，默认路由到大模型，宁可多花成本也不牺牲质量。

语义缓存不一致：缓存的结果可能已过时。当底层模型更新后，缓存中的旧结果与新模型的结果可能不同。解决方案是为缓存条目设置较短的 TTL（1 小时），并在模型版本更新时清空缓存。但 TTL 越短，命中率越低，降本效果越弱。

五、总结

AI 推理成本治理需要从模型层、调度层和请求层三个维度协同推进。模型层通过量化降低单次推理的显存和算力消耗；调度层通过智能路由将请求分配到最经济的模型；请求层通过语义缓存和请求去重避免冗余推理。三层策略叠加，可以将推理成本降低 50-70%。但量化精度损失、路由误判和缓存不一致是不可回避的代价，需要在成本和质量之间找到业务可接受的平衡点。

落地路线建议：第一步，统计当前推理服务的成本构成，识别显性、半隐性和隐性浪费的比例；第二步，对非精度敏感的业务模型实施 INT8 量化，验证精度后逐步替换 FP16 模型；第三步，实现智能路由器，根据请求复杂度将 60% 的流量路由到小模型；第四步，部署语义缓存，对高频相似问题复用推理结果；第五步，建立推理成本监控看板，追踪每百万 Token 的推理成本和各降本策略的 ROI。