背景：上线3个月后开始答非所问
我们的客服机器人跑了3个月后，开始出现"答非所问"的问题。用户说"查一下我的订单"，AI 回的是"工单编号格式如下"。日志里单会话 token 已经突破 18 万，上下文窗口直接爆了。
事后复盘发现：不是模型变差了，是上下文累积没有控制。从 Round 25（约 46000 tokens）开始，模型就开始受影响，但没有告警机制，等到发现问题已经是 Round 50 了。
实测数据：
| 用户类型 | 开始告警阈值 | 崩溃临界点 |
|---|---|---|
| 普通用户（3个月） | ~46000 tokens（Round 25） | ~80000 tokens |
| 重度用户 | ~46000 tokens（Round 15） | ~180000 tokens（Round 50） |
解决方案：三层 Token 预算告警机制
核心思路：不等模型崩溃，先于用户感知告警。我们实现了一套三层告警：
- 信息池 — 关键字段（用户ID、订单号、工单号）永远置顶，不参与滑动窗口
- 预算控制 — Token 计数器 + 双阈值（warn 70%、critical 90%）
- 自动压缩 — 超过阈值时触发摘要压缩，保留关键信息
核心代码实现
第一步：Token 计数器
import tiktoken
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class TokenBudget:
model: str = "gpt-4o-mini"
warn_pct: float = 0.70
crit_pct: float = 0.90
max_tokens: int = 128000
# 关键信息池（永远不参与滑动窗口）
priority_pool: dict = field(default_factory=dict)
_history_tokens: int = 0
def count_messages(self, messages: list[dict]) -> int:
enc = tiktoken.encoding_for_model(self.model)
total = 0
for msg in messages:
total += len(enc.encode(msg.get("content", "")))
total += len(enc.encode(msg.get("role", ""))) + 4
return total
def check(self, messages: list[dict]) -> dict:
current = self.count_messages(messages)
pct = current / self.max_tokens
level = "normal"
if pct >= self.crit_pct:
level = "critical"
elif pct >= self.warn_pct:
level = "warning"
return {
"level": level,
"current_tokens": current,
"max_tokens": self.max_tokens,
"pct": round(pct * 100, 1),
"should_compress": pct >= self.warn_pct,
}
第二步：关键信息自动提取
import re
PRIORITY_PATTERNS = [
r"工单[号#]?\s*[:：]?\s*([A-Z0-9]{8,})",
r"订单[号#]?\s*[:：]?\s*([A-Z0-9]{8,})",
r"用户ID\s*[:：]?\s*([a-zA-Z0-9]{6,})",
]
def extract_priority_fields(messages: list[dict]) -> dict:
pool = {}
for msg in messages:
text = msg.get("content", "")
for pattern in PRIORITY_PATTERNS:
if match := re.search(pattern, text):
field_name = pattern.split(r"(")[1].rstrip(")")
pool[field_name] = match.group(1)
return pool
第三步：接入 Prometheus 告警
from prometheus_client import Counter, Gauge
token_warn_total = Counter("llm_token_warn_total", "Warning level triggered")
token_crit_total = Counter("llm_token_crit_total", "Critical level triggered")
token_pct_gauge = Gauge("llm_token_pct", "Current token usage pct", ["user_id"])
def on_token_check(budget: TokenBudget, messages: list[dict], user_id: str):
result = budget.check(messages)
token_pct_gauge.labels(user_id=user_id).set(result["pct"] / 100)
if result["level"] == "critical":
token_crit_total.inc()
compress_conversation(messages, budget)
elif result["level"] == "warning":
token_warn_total.inc()
log.warning(f"Token usage high for {user_id}: {result['pct']}%")
阈值怎么定：别猜，用数据
网上说"Token 超过 70% 就该压缩"，这是废话——不同用户、不同场景阈值完全不一样。
正确做法：先跑一个季度，用真实数据定阈值。我们的方法：
- 上线后记录每个会话的 Round 数和对应 Token 数
- 找到"模型开始答非所问"的 Round 作为 ground truth
- 往前取 20% 的 Round 数作为 warn 阈值
- 往前取 10% 的 Round 数作为 critical 阈值
Grafana 告警配置
groups:
- name: llm_token_alerts
rules:
- alert: LLMTokensWarning
expr: llm_token_pct > 0.70
for: 1m
labels:
severity: warning
annotations:
summary: "LLM Token 使用率超过 70%"
description: "用户 {{ $labels.user_id }} 的 Token 使用率已达 {{ $value | humanizePercentage }}"
- alert: LLMTokensCritical
expr: llm_token_pct > 0.85
for: 30s
labels:
severity: critical
annotations:
summary: "LLM Token 即将崩溃"
description: "用户 {{ $labels.user_id }} 的 Token 使用率已达 {{ $value | humanizePercentage }}，建议立即压缩"
推荐配两档：70% 预警发飞书通知客服，85% 自动触发上下文压缩。
现在你可以做什么
第一步：装依赖
pip install tiktoken prometheus_client --break-system-packages
第二步：把 TokenBudget 类复制到你的对话机器人代码里
第三步：在每轮对话结束后调用 budget.check(messages)，根据返回值决定是否压缩
第四步：Prometheus 上报后，在 Grafana 配两条告警规则：70% 预警 + 85% 严重
第五步：跑一个季度，用真实 Round 数据校准你的 warn/critical 阈值
不要抄网上的"80%就该压缩"——用你自己产品的真实数据，阈值差 5% 可能就是用户满意和不满意的分水岭。
相关踩坑记录：
© 版权声明
本站部分内容为网络收集，若侵犯到您的权益，请提供相关证明联系，即删。
更多交流点击入群
更多交流点击入群
THE END
