大模型成本治理：Token 用量优化与推理资源精细化管理

一、大模型调用成本的黑洞：Token 账单背后的资源浪费

大模型的调用成本正在成为 AI 应用的最大运营支出。以 GPT-4 级别模型为例，输入 Token 单价约 0.03 元/千 Token，输出 Token 单价约 0.06 元/千 Token。一个日活 10 万的 AI 对话应用，日均 Token 消耗约 5000 万，月成本超过 50 万元。更关键的是，其中约 30%-40% 的 Token 消耗是可避免的浪费：冗长的系统提示词重复发送、无效的上下文窗口填充、低价值请求占用高成本模型。

某 AI 客服平台在上线大模型对话功能后，月调用成本从预期的 20 万元飙升至 80 万元。分析发现：系统提示词占每次请求 Token 量的 40%，且每次对话都完整重发；大量简单咨询（如查快递、问营业时间）被路由到最贵的 GPT-4 模型，而这些请求用小模型即可满足；用户重复提问导致上下文窗口不断膨胀，单次对话的 Token 消耗从初始的 500 增长到 8000 以上。

二、大模型成本治理的核心机制：Token 全链路优化

大模型成本治理不是简单的"少调用"，而是在保证服务质量的前提下，通过精细化手段将每一分 Token 预算花在刀刃上。

flowchart TB
    A[用户请求] --> B{模型路由网关}
    B -->|简单请求| C[小模型 - 低成本]
    B -->|复杂请求| D[大模型 - 高质量]

    subgraph Token 优化层
        E[提示词压缩] --> F[系统提示词缓存]
        F --> G[上下文窗口裁剪]
        G --> H[语义去重]
    end

    C --> E
    D --> E

    subgraph 资源管理层
        I[Token 配额管理] --> J[租户级限额]
        I --> K[场景级预算]
        I --> L[弹性计费]
    end

    H --> I

    subgraph 监控分析
        M[Token 用量采集] --> N[成本归因分析]
        N --> O[优化建议引擎]
        O --> P[自动策略调整]
    end

    I --> M

模型路由网关是成本治理的第一道防线。根据请求复杂度将请求路由到不同规格的模型：简单问答路由到 7B 小模型（成本仅为大模型的 1/10），复杂推理路由到大模型。路由决策基于请求分类器的输出，分类器本身可以是一个轻量级的 BERT 模型。

Token 优化层从三个维度压缩 Token 消耗：提示词压缩将冗长的系统提示词精简为关键指令（通常可压缩 50% 以上）；上下文窗口裁剪移除历史对话中的冗余信息，保留关键语义；语义去重检测用户重复提问，直接引用之前的回答而非重新生成。

资源管理层实现 Token 的配额控制和成本归因：租户级限额防止单个租户消耗过多资源；场景级预算为不同业务场景设定 Token 预算上限；弹性计费根据实际用量动态调整配额。

三、生产级大模型成本治理的代码实现

3.1 智能模型路由网关

"""
模型路由网关——根据请求复杂度选择最优模型
为什么需要路由而非全部使用大模型？
大模型（如 GPT-4）单次推理成本约 0.15 元，
小模型（如 7B）自部署单次推理成本约 0.01 元，
约 60% 的客服咨询是简单问题，用小模型即可满足
"""
from dataclasses import dataclass
from enum import Enum
from typing import Optional

class ModelTier(Enum):
    """模型层级定义"""
    SMALL = "7b-chat"         # 小模型：简单问答
    MEDIUM = "13b-chat"       # 中模型：常规对话
    LARGE = "70b-chat"        # 大模型：复杂推理

@dataclass
class RoutingDecision:
    """路由决策结果"""
    tier: ModelTier
    confidence: float
    reason: str

class ModelRouter:
    """智能模型路由器"""

    # 复杂度判定的关键词映射
    COMPLEXITY_KEYWORDS = {
        ModelTier.LARGE: [
            "分析", "对比", "推理", "为什么", "原理",
            "evaluate", "analyze", "compare", "reasoning"
        ],
        ModelTier.MEDIUM: [
            "解释", "描述", "总结", "如何", "方法",
            "explain", "describe", "summarize", "how to"
        ],
    }

    def route(self, query: str,
              context: Optional[list] = None) -> RoutingDecision:
        """
        路由决策：基于规则 + 分类器双重判定
        为什么需要双重判定？
        规则匹配快速但覆盖不全，分类器准确但有延迟，
        双重判定以规则为快速路径，分类器兜底
        """
        # 快速路径：规则匹配
        rule_decision = self._rule_based_route(query)
        if rule_decision.confidence > 0.8:
            return rule_decision

        # 兜底路径：分类器判定
        return self._classifier_route(query, context)

    def _rule_based_route(self, query: str) -> RoutingDecision:
        """基于关键词的快速路由"""
        query_lower = query.lower()

        for tier, keywords in self.COMPLEXITY_KEYWORDS.items():
            if any(kw in query_lower for kw in keywords):
                return RoutingDecision(
                    tier=tier,
                    confidence=0.7,
                    reason=f"关键词匹配: {tier.value}"
                )

        # 默认路由到小模型（成本优先策略）
        # 为什么默认小模型？大部分查询是简单的，
        // 宁可偶尔降级也不要浪费大模型资源
        return RoutingDecision(
            tier=ModelTier.SMALL,
            confidence=0.6,
            reason="默认路由：无复杂度特征"
        )

    def _classifier_route(self, query: str,
                           context: Optional[list]) -> RoutingDecision:
        """基于分类器的路由"""
        # 构建分类特征
        features = self._extract_features(query, context)
        # 调用轻量级分类器（BERT-tiny，推理延迟 < 10ms）
        prediction = self.classifier.predict(features)

        tier = ModelTier(prediction.tier)
        return RoutingDecision(
            tier=tier,
            confidence=prediction.confidence,
            reason=f"分类器判定: confidence={prediction.confidence:.2f}"
        )

    def _extract_features(self, query: str,
                           context: Optional[list]) -> dict:
        """提取分类特征"""
        return {
            "query_length": len(query),
            "context_length": len(context) if context else 0,
            "has_question_mark": "?" in query or "？" in query,
            "sentence_count": query.count("。") + query.count(".") + 1,
            # 为什么用句子数？复杂问题通常包含多个子问题，
            # 简单问题通常只有一句话
        }

3.2 提示词压缩与上下文裁剪

"""
Token 用量优化——提示词压缩与上下文裁剪
为什么需要压缩？系统提示词通常占每次请求 Token 量的 40%，
而其中大量内容是冗余的格式说明和示例，压缩后不影响模型输出质量
"""
import re
from dataclasses import dataclass

@dataclass
class TokenBudget:
    """Token 预算配置"""
    system_prompt_max: int = 500    # 系统提示词上限
    context_max: int = 2000         # 上下文窗口上限
    response_max: int = 1000        # 响应上限

class TokenOptimizer:

    def __init__(self, budget: TokenBudget):
        self.budget = budget

    def compress_system_prompt(self, prompt: str) -> str:
        """
        压缩系统提示词
        策略：移除冗余格式、精简指令、合并同类项
        """
        # 移除多余空白和格式标记
        compressed = re.sub(r'\n{3,}', '\n\n', prompt)
        compressed = re.sub(r' {2,}', ' ', compressed)

        # 移除示例（示例占大量 Token 但对指令遵循帮助有限）
        # 为什么可以移除示例？研究表明，清晰的指令比示例
        # 更能有效引导模型行为，示例主要在指令模糊时有用
        compressed = re.sub(
            r'例如：.*?(?=\n\n|\Z)',
            '', compressed, flags=re.DOTALL
        )

        # 精简冗余修饰词
        replacements = {
            "请你务必注意": "注意",
            "非常重要的一点是": "关键：",
            "请按照以下步骤操作": "步骤：",
        }
        for old, new in replacements.items():
            compressed = compressed.replace(old, new)

        # Token 数检查：超出预算则截断
        token_count = self._estimate_tokens(compressed)
        if token_count > self.budget.system_prompt_max:
            # 保留最关键的前 N 个 Token
            compressed = self._truncate_to_tokens(
                compressed, self.budget.system_prompt_max
            )

        return compressed

    def prune_context(self, messages: list[dict],
                       current_query: str) -> list[dict]:
        """
        上下文窗口裁剪——保留与当前查询相关的历史对话
        为什么需要裁剪？对话越长 Token 消耗越大，
        但大部分历史对话与当前问题无关，裁剪不影响回答质量
        """
        if not messages:
            return messages

        # 计算当前上下文 Token 数
        total_tokens = sum(
            self._estimate_tokens(m["content"]) for m in messages
        )

        if total_tokens <= self.budget.context_max:
            return messages

        # 策略一：语义相关性裁剪
        # 保留与当前查询语义最相关的历史消息
        scored_messages = []
        for i, msg in enumerate(messages):
            relevance = self._compute_relevance(
                msg["content"], current_query
            )
            recency = i / len(messages)  # 越新权重越高
            # 综合评分：相关性 60% + 时效性 40%
            score = 0.6 * relevance + 0.4 * recency
            scored_messages.append((score, i, msg))

        # 按评分排序，保留 Top-K 直到 Token 预算用尽
        scored_messages.sort(key=lambda x: x[0], reverse=True)
        selected = []
        used_tokens = 0

        for score, idx, msg in scored_messages:
            msg_tokens = self._estimate_tokens(msg["content"])
            if used_tokens + msg_tokens <= self.budget.context_max:
                selected.append((idx, msg))
                used_tokens += msg_tokens

        # 按原始顺序排列（对话的时序性很重要）
        selected.sort(key=lambda x: x[0])
        return [msg for _, msg in selected]

    def _compute_relevance(self, text: str, query: str) -> float:
        """计算文本与查询的语义相关性"""
        # 使用轻量级 Embedding 模型计算余弦相似度
        # 生产中可缓存 Embedding 结果避免重复计算
        text_emb = self.embedder.encode(text)
        query_emb = self.embedder.encode(query)
        similarity = cosine_similarity(text_emb, query_emb)
        return float(similarity)

    def _estimate_tokens(self, text: str) -> int:
        """估算 Token 数（中文约 1.5 字/Token）"""
        return int(len(text) / 1.5)

3.3 Token 配额管理与成本归因

"""
Token 配额管理——租户级限额与成本归因
为什么需要配额管理？没有限额的 AI 服务会快速失控，
某租户的异常调用可能耗尽整个平台的 Token 预算
"""
from datetime import datetime, timedelta

class TokenQuotaManager:

    def __init__(self, redis_client):
        self.redis = redis_client

    def check_quota(self, tenant_id: str,
                     estimated_tokens: int) -> bool:
        """
        检查租户 Token 配额是否充足
        采用滑动窗口计数，避免固定窗口的边界突发问题
        """
        now = datetime.utcnow()
        window_key = f"quota:{tenant_id}:{now.strftime('%Y%m%d%H')}"

        used = int(self.redis.get(window_key) or 0)
        limit = self._get_tenant_limit(tenant_id)

        if used + estimated_tokens > limit:
            # 配额不足：记录告警
            self._alert_quota_exceeded(tenant_id, used, limit)
            return False

        # 预扣配额（防止并发超限）
        self.redis.incrby(window_key, estimated_tokens)
        # 设置过期时间（小时级窗口，2 小时后过期）
        self.redis.expire(window_key, 7200)
        return True

    def record_usage(self, tenant_id: str, model: str,
                      input_tokens: int, output_tokens: int,
                      scene: str):
        """记录 Token 用量，用于成本归因分析"""
        cost = self._calculate_cost(model, input_tokens, output_tokens)

        # 写入时序数据库，支持多维聚合分析
        self.tsdb.write({
            "measurement": "token_usage",
            "tags": {
                "tenant_id": tenant_id,
                "model": model,
                "scene": scene,
            },
            "fields": {
                "input_tokens": input_tokens,
                "output_tokens": output_tokens,
                "cost": cost,
            },
            "time": datetime.utcnow(),
        })

    def _calculate_cost(self, model: str,
                         input_tokens: int,
                         output_tokens: int) -> float:
        """计算单次调用成本"""
        pricing = {
            "7b-chat":  {"input": 0.002, "output": 0.004},
            "13b-chat": {"input": 0.008, "output": 0.016},
            "70b-chat": {"input": 0.030, "output": 0.060},
        }
        p = pricing.get(model, pricing["70b-chat"])
        return (input_tokens * p["input"]
                + output_tokens * p["output"]) / 1000

    def get_cost_report(self, tenant_id: str,
                         days: int = 7) -> dict:
        """生成成本归因报告"""
        result = self.tsdb.query(
            f"""
            SELECT
                model,
                scene,
                SUM(input_tokens) as total_input,
                SUM(output_tokens) as total_output,
                SUM(cost) as total_cost
            FROM token_usage
            WHERE tenant_id = '{tenant_id}'
              AND time > now() - {days}d
            GROUP BY model, scene
            ORDER BY total_cost DESC
            """
        )
        return result

四、大模型成本治理的架构权衡

模型路由的准确性代价：路由分类器的准确率直接影响服务质量和成本。误将复杂问题路由到小模型会降低回答质量，误将简单问题路由到大模型则浪费成本。生产中通常将分类器的阈值偏向大模型（宁可多花成本也不降低质量），但这会压缩成本优化空间。

提示词压缩的质量风险：过度压缩系统提示词可能导致模型行为偏离预期。例如，安全约束被压缩后，模型可能生成有害内容。压缩策略必须保留安全指令和关键约束，仅精简格式说明和示例。

上下文裁剪的信息丢失：语义相关性裁剪可能误删关键上下文。例如，用户说"他怎么样"，裁剪器可能认为前文提到的"他"与当前查询无关而删除，导致模型无法理解代词指代。需要保留代词解析所需的最近 N 轮对话。

适用边界：成本治理适用于 Token 消耗大、多模型共存、多租户共享的 AI 平台。对于单一模型、单一租户的小规模应用，治理成本可能超过节省的成本。

禁用场景：对回答质量有极高要求的场景（如医疗诊断、法律咨询），不应为节省成本而降低模型规格。成本优化的前提是服务质量不降级。

五、总结

大模型成本治理的核心是在服务质量与成本之间找到最优平衡点。模型路由网关根据请求复杂度选择合适规格的模型，Token 优化层从提示词压缩、上下文裁剪、语义去重三个维度压缩用量，配额管理层实现租户级限额和成本归因。三者协同构建了从请求入口到资源消耗的全链路成本控制体系。

落地路线建议：第一步，建立 Token 用量监控体系，按模型、场景、租户维度归因成本，找出浪费热点；第二步，实现模型路由网关，将简单请求路由到低成本模型；第三步，压缩系统提示词，设定 Token 预算上限；第四步，实现上下文窗口裁剪，控制长对话的 Token 增长；第五步，部署租户级配额管理，防止单租户消耗失控。成本治理不是一次性优化，而是持续的数据驱动过程，每一分 Token 的节省都应被量化追踪。