生产环境Token优化实战指南

当企业级 AI 应用规模化后，不可避免地会撞上一堵墙。对很多工程团队来说，这堵墙最初表现为账单问题——每月的 API 账单已经失控。然而，如果仅把 token 消耗当作财务指标来对待，就根本性地误解了 LLM 在生产环境中的运作方式。Token 优化不像其他优化手段，它不是一场财务核算练习，而是一个分布式系统与硬件利用率层面的挑战。

"Token 优化不像其他优化手段，它不是一场财务核算练习，而是一个分布式系统与硬件利用率层面的挑战。"

本指南将从 Concierge（一个对延迟敏感的同步客服 Agent）和 Pathfinder（一个异步多步自主 CI 调试 Agent）的视角出发，探讨这些系统如何在成长过程中陷入自回归瓶颈，以及我们如何修复这些问题。

你实际在为什么付费

Token 不等于单词：把它当单词来处理会让你的预算"模型"崩掉。每家主流 LLM 提供商都使用字节对编码（BPE）对文本进行分词，将单词拆解为子词单元。常见单词保持完整，罕见单词或标点符号则被拆分成碎片。作为经验值，1 token ≈ 4 个字符，或标准英文 prose 中约 0.75 个单词。

在生产环境做预算时，必须考虑结构性的定价差异；提供商对输入 token 和输出 token 的计费标准不同。输出 token 通常比输入 token 贵 4-5 倍。

作为基准，假设一个中高端前沿模型运行费用约为：输入 token 每百万 $3，输出 token 每百万 $15。

二次方历史税

LLM 提供商的 API 完全是无状态的；要让 LLM 表现得像是在"记住"过往事件，必须在每次 API 调用时重新发送整个会话历史和输入。

这意味着模型的既往输出在后续步骤中会不断被重新计费为输入。这触发了一种复利式成本，影响 Concierge 和 Pathfinder 两者，尽管它们的成本曲线 scaling 方式不同。

设 S 为静态系统上下文（指令和 schema），u 为每步的传入数据，r 为模型的响应载荷。则在每个回合 k 上的输入成本为：

当在 N 步的完整执行运行中求和时，总输入 token 量呈二次方增长。

这种 O(N²) 的历史累积正是导致成本和延迟双双爆炸的机制。

变量	Concierge（对话系统）	Pathfinder（自主 Agent）
静态上下文（S）	3,100 token（完整的退换货政策和品牌指南）	1,200 token（工具定义、系统约束、CI 环境数据）
传入数据（u）	80 token（客户简短聊天回复）	900 token（庞大的原始文本载荷：日志摘要、文件读取、shell 输出）
响应载荷（r）	220 token（礼貌的客用回复）	300 token（内部独白 + JSON 工具参数）
步数倍数（N）	10 轮（平均支持会话长度）	15 步（平均 Agent 问题排查循环长度）

当我们用二次公式计算单个会话消耗的总输入 token 时：

Concierge：每张 10 轮工单消耗 45,300 token

Pathfinder：每轮 15 步消耗 150,000 token

由于 Pathfinder 的步长增量是 Concierge 的 4 倍，其 token 成本曲线也陡峭得多。如果 Pathfinder 不慎陷入无限工具调用循环并达到 30 步，单次运行就可能消耗 570,000 token。

修复单个调用

Prompt 卫生：把静态参考文档硬编码进 system prompt 意味着每轮都要为解析相同文本付费。 所以我们把静态文本从 prompt 中剥离，改为动态注入。

对于 Concierge，我们实现了一个 RAG 步骤，只获取与该工单相关的 2-3 条策略片段。Prompt 从 3,100 token 降至 380 token。对于 10 轮会话减少了 60%。

对于 Pathfinder，我们使用 LLMLingua-2 进行自动化 prompt 压缩，在将冗长的 CI 日志文件发送给模型之前先压缩。通过过滤掉非必要的日志行，我们将传入工具观察数据的大小减少了 3 倍，同时不损失调试准确性。

from llmlingua import PromptCompressor

compressor = PromptCompressor(
model_name="microsoft/llmlingua-2-xlm-roberta-large-meetingbank",
use_llmlingua2=True
)

try:
compressed_result = compressor.compress_prompt(
raw_ci_log_text,
rate=0.33,
force_tokens=["Error", "Exception", "Failed", "Traceback", "FATAL"]
)
# Pass high-density payload to the frontier model
compact_prompt = compressed_result["compressed_prompt"]
except Exception as e:
print(f"Compression failed, falling back to raw log text: {e}")
# Graceful degradation: pass the raw (or truncated) log if compression fails
compact_prompt = raw_ci_log_text

消除重试：依赖开放式 prose 指令"return JSON"会导致格式错误。 解析失败时，系统会发起同步重试，将整个累积上下文作为新尝试重新发送。我们用严格的结构化契约替换了整套自然语言格式请求，通过强制 schema 验证来实现。

在 Concierge 和 Pathfinder 中，我们都把输出格式转换为严格的 pydantic schema，用于工具调用模式和工具执行载荷。两套系统的格式错误输出均降至 0.5% 以下，消除了由级联队列导致的尾部延迟尖峰。

# Unified Schema Enforcement for Concierge Responses & Pathfinder Tool Execution
from pydantic import BaseModel
from typing import Literal

class TicketResponse(BaseModel):
    reply: str
    category: Literal["shipping", "returns", "billing", "product", "other"]
    escalate: bool
    confidence: float

# The API is structurally locked into emitting validated JSON matching the schema
response = client.messages.create(
    model="claude-opus-4",
    system=SYSTEM_PROMPT,
    messages=messages,
    tools=[
    {
    "name": "respond_to_ticket",
    "description": "Formulate a response and classify the support ticket.",
    "input_schema": TicketResponse.model_json_schema()
    }
    ],
    tool_choice={"type": "tool", "name": "respond_to_ticket"},
)

输出 token 边界： 模型自然会产生冗长的推理链和对话填充内容，从而膨胀昂贵的输出 token。在 LLM 提供商开放 logit bias 的地方，你可以在解码时直接压制有效集合之外的每个 token；在没有开放的地方，受限解码库（Outlines、Guidance）或带 enum 类型 schema 的强制工具调用也能实现相同保证。

class ClassifyOnly(BaseModel):
    category: Literal["shipping", "returns", "billing", "product", "other"]
    priority: Literal["low", "medium", "high", "urgent"]

缓存历史以控制成本

模型的无状态特性意味着我们必须在每轮都解析静态 prompt 前缀和历史步骤。我们引入了显式的缓存断点，允许推理引擎复用静态块的状态。我们修改了 Concierge 和 Pathfinder，让它们将稳定的历史段落标记为可缓存。在标准提供商定价下，缓存读取有 90% 的折扣。确认缓存是否已启用非常重要。

# Caching the stable history prefix for a multi-turn session
response = client.messages.create(
    model="claude-sonnet-4",
    max_tokens=4096,
    system=[{
        "type": "text",
        "text": SYSTEM_PROMPT,
        "cache_control": {"type": "ephemeral"} # Cache hits drop prefix costs by 90%
    }],
    tools=TOOL_SCHEMAS,
    messages=session_history + [{"role": "user", "content": current_step_input}],
)

对于 10 轮 Concierge 聊天，这使输入成本降低了约 70%。对于 15 步 Pathfinder 轨迹，成本降低了 76%。

语义缓存

跨独立会话的重复查询触发了冗余的前沿模型调用。我们在 LLM 上游使用 Redis 实现了一个向量相似度缓存层。我们的 Concierge 服务分析显示，34% 的客户支持工单在语义上是常见 FAQ 的重复。

拦截这些请求将延迟降至亚 50ms 级别。由于 CI pipeline 日志的特性，我们尚未为 Pathfinder 的输入找到合适的缓存方案。

import os
import json
import redis
from redis.commands.search.query import Query

# Configure connection via environment variable for environment portability
redis_url = os.environ.get("REDIS_URL", "redis://localhost:6379")
r = redis.Redis.from_url(redis_url)

def get_cached_response(tenant_id, query_text, threshold=0.92):
    try:
results = r.ft(f"cache_idx:{tenant_id}").search( # scoped by tenant -- see below
Query("*=>[KNN 1 @vector $vec AS score]").sort_by("score").dialect(2),
query_params={"vec": query_vec.tobytes()},
)
except redis.RedisError as e:
print(f"Redis cache error: {e}")
return None # Fail-open: gracefully fall back to a cache miss)
  query_vec = embed(query_text) # small, fast bi-encoder -- not the frontier model

try:
results = r.ft(f"cache_idx:{tenant_id}").search( # scoped by tenant -- see below
Query("*=>[KNN 1 @vector $vec AS score]").sort_by("score").dialect(2),
query_params={"vec": query_vec.tobytes()},
)
except redis.RedisError as e:
print(f"Redis cache error: {e}")
return None # Fail-open: gracefully fall back to a cache miss

语义缓存可能带来安全问题，因为如果你选择全局缓存，客户 A 的账户特定答案可能因为其措辞 embeddings 足够接近而被 served 给客户 B。为缓解此问题，我们将缓存分为两层：全局缓存用于租户无关内容，任何涉及账户状态的内容则使用以租户 ID 为前缀键控的每租户、每用户命名空间。

缓存污染是另一个风险：我们只从通过 schema 验证和注入模式分类器的响应中写入缓存，我们为每个缓存条目加盖其来源 traceId，并鼓励对未知缓存进行定期清理。

滑动窗口压缩

无上限的对话或 Agent 轨迹允许 N 持续增长，扩大成本曲线并导致延迟退化。我们通过实现一个滑动窗口来对 N 进行上限控制——用一个小型、超廉价的模型对历史上下文进行摘要。对于 Concierge，我们逐字保留最近 3 轮，而将更早的轮次压缩为一个滚动元数据块。

对于 Pathfinder，当调试步骤超过 4 轮时，我们将最旧的工具执行输出修剪并摘要为一个紧凑的时间线，将开放式的二次方成本爆炸转化为可预测的 bounded 窗口。

def compact_session_history(history_steps: List[Dict[str, Any]], keep_recent: int = 3) ->     List[Dict[str, Any]]:
    """Flattens older history into a cheap summary block, preserving recent context."""
    if len(history_steps) <= keep_recent:
        return history_steps
    old_steps = history_steps[:-keep_recent]
    recent_steps = history_steps[-keep_recent:]
    
# Compress the old history using a fast, low-cost utility model
try:
historical_summary = summarize_with_utility_model(old_steps)
# Note: Anthropic prohibits 'system' roles in the messages array.
# Using 'assistant' ensures cross-provider compatibility.
return [{"role": "assistant", "content": f"[System Context: Summary of prior steps: {historical_summary}]"}] + recent_steps
except Exception as e:
print(f"History compression failed: {e}")
# Fallback: Return the uncompressed history to gracefully degrade
return history_steps

模型级联

将每一个操作都定向到昂贵的前沿模型，对于简单任务来说是一种巨大的过度供给。我们集成了 LiteLLM 作为内部路由网关来实现模型级联，将每个请求路由到能够完成该任务的最低成本模型。

"将每一个操作都定向到昂贵的前沿模型，对于简单任务来说是一种巨大的过度供给。"

# litellm_config.yaml
model_list:
  - model_name: fast-path
    litellm_params:
      model: openai/mistral-support-ft
      api_base: http://vllm-internal:8000/v1
  - model_name: frontier-path
    litellm_params:
      model: anthropic/claude-opus-4

简单重复的任务被路由到较低层模型，使 70% 的 Concierge 对话从前沿模型中卸载。对于 Pathfinder，我们将 Agent 循环分解为多个子任务：高层次规划、工具选择和代码补丁合成保留在前沿模型，而机械性、文本密集型的操作（如日志解析、正则提取和错误字符串格式化）则卸载到较低层模型。这种混合编排使 Pathfinder 的 token 成本降低了 50% 以上。

结论

Concierge 和 Pathfinder 的蜕变证明了一个关于生产级 AI 的基本真理。你不能单纯依赖前沿模型的自然语言能力来实现规模化。你必须围绕它来设计系统。通过将关注点从朴素的 token 削减转向最大化系统资源利用率，我们重新获得了对基础设施的绝对控制。

"在生成式 AI 时代，效率不由你能以多低的成本运营来定义，而由你能以多高的密度打包信息、多快地提供信息、以及多可靠地解析输出来定义。"

本文详述的架构决策代表的不仅仅是一种 token 优化策略；它们是构建高吞吐、经实战考验、有韧性的规模化 AI 系统的必备基础。