1. 项目背景与问题定位
上周接手一个微信小助手开发需求，核心功能是日程管理和信息摘要。技术选型上采用了OpenClaw作为AI服务后端，这个基于阿里云算力的模型在推理速度和响应质量上表现优异。但上线三天后，后台数据让我惊出一身冷汗——Token消耗量达到预算的4倍，日均消耗280万Token，折合成本约40元/天。
深入分析日志发现两个核心问题：
- 每次请求平均携带4000-6000 Token的对话历史
- 大量语义重复的简单查询（如"明天天气"）都在调用高规格模型
这种粗放式的调用方式，就像用手术刀切水果——功能能实现，但成本完全不可持续。经过两周的优化实践，最终将Token消耗稳定在103万/天，降幅达63%。下面分享具体实施方案。
2. 核心优化策略解析
2.1 Prompt工程优化
2.1.1 System Prompt精简术
原始Prompt存在典型的三类问题：
- 冗余修饰词（"非常专业的"）
- 过度说明（"包括但不限于"）
- 格式要求重复声明
优化前后对比：
python复制# 优化前（178 tokens）
"""
你是一个非常专业的日程管理助手。你需要帮助用户管理他们的日程安排，
包括但不限于创建新的日程、修改已有的日程、删除日程、查询某个时间段的日程安排。
你应该用友好、专业的语气回复用户。当用户的请求不明确时，你应该主动询问更多细节。
你需要以 JSON 格式返回结构化的日程数据，包含 title、start_time、end_time、description 等字段。
请注意，所有时间都应该使用 ISO 8601 格式。
"""
# 优化后（62 tokens）
"""日程助手。操作：创建/修改/删除/查询。
输出JSON：{title,start_time,end_time,desc}，时间用ISO8601。
不明确时追问。"""
关键技巧：
- 使用电报式语言风格
- 用符号替代完整句子（{}表示JSON结构）
- 保留核心动词和名词
- 将格式要求压缩成单行说明
2.1.2 对话历史管理方案
微信场景下的长对话会带来两个挑战：
- Token消耗指数增长
- 早期重要信息丢失
解决方案采用三级策略：
python复制def manage_conversation(messages, max_tokens=2000):
# 第一级：基础裁剪
if count_tokens(messages) <= max_tokens:
return messages
# 第二级：滑动窗口保留最近对话
trimmed = trim_conversation(messages, max_tokens)
# 第三级：摘要压缩被裁减内容
if count_tokens(trimmed) >= max_tokens * 0.8: # 预留buffer
return smart_trim(messages, max_tokens)
return trimmed
其中smart_trim函数的实现要点：
- 使用DeepSeek-V3等低成本模型生成摘要
- 摘要指令明确限制字数（50字内）
- 将摘要以标记格式（[历史摘要]）插入system prompt
- 保留最近3-5轮完整对话
2.2 语义缓存层实现
2.2.1 缓存键设计演进
初版方案直接对用户消息MD5哈希，发现两个问题：
- 标点符号差异导致相同语义不同key
- 停用词影响缓存命中率
改进后的键生成逻辑：
python复制def normalize_text(text):
# 1. 统一转小写
text = text.lower()
# 2. 移除标点
text = re.sub(r'[^\w\s]', '', text)
# 3. 过滤停用词
stop_words = {"的","了","啊","请","帮"}
words = [w for w in text.split() if w not in stop_words]
return " ".join(words)
def get_cache_key(messages):
last_msg = extract_last_user_message(messages)
normalized = normalize_text(last_msg)
return f"llm_cache:{hashlib.md5(normalized.encode()).hexdigest()}"
2.2.2 缓存更新策略
采用分级TTL机制：
- 简单查询（天气/时间）：TTL 1小时
- 复杂查询（日程管理）：TTL 15分钟
- 涉及个人数据的查询：不缓存
通过Redis的sorted set实现自动清理：
python复制def update_cache(key, value, category):
ttl = get_ttl_by_category(category)
pipe = r.pipeline()
pipe.setex(key, ttl, json.dumps(value))
pipe.zadd("cache_index", {key: time.time()})
pipe.execute()
# 定期清理（每小时执行）
cutoff = time.time() - 86400 # 保留24小时记录
r.zremrangebyscore("cache_index", 0, cutoff)
2.3 动态模型路由系统
2.3.1 复杂度分类器设计
避免使用LLM进行复杂度判断（会产生额外开销），采用基于规则的轻量级方案：
python复制class ComplexityClassifier:
def __init__(self):
self.simple_patterns = {
"greeting": ["你好", "hi", "hello"],
"query": ["查询", "查看", "问一下"],
"time": ["几点", "时间", "什么时候"]
}
self.complex_patterns = {
"analysis": ["分析", "对比", "优缺点"],
"planning": ["规划", "方案", "建议"],
"multi_step": ["然后", "接着", "下一步"]
}
def classify(self, text):
text = text.lower()
word_count = len(text.split())
# 规则1：包含复杂关键词
for _, words in self.complex_patterns.items():
if any(w in text for w in words):
return "complex"
# 规则2：长文本且含多个动词
if word_count > 15 and len(detect_verbs(text)) >= 2:
return "medium"
# 规则3：简单问答模式
if word_count < 8 and any(w in text for w in ["吗","？","?"]):
return "simple"
return "medium"
2.3.2 模型路由表配置
通过YAML文件实现可配置化路由：
yaml复制model_routing:
simple:
model: deepseek-v3
max_tokens: 300
temperature: 0.3
medium:
model: openclaw
max_tokens: 500
temperature: 0.7
complex:
model: claude-opus-4.6
max_tokens: 1000
temperature: 0.9
动态加载配置实现热更新：
python复制class ModelRouter:
def __init__(self, config_path):
self.config = self._load_config(config_path)
self.last_modified = 0
def _load_config(self, path):
with open(path) as f:
return yaml.safe_load(f)
def check_update(self):
mtime = os.path.getmtime(self.config_path)
if mtime > self.last_modified:
self.config = self._load_config()
self.last_modified = mtime
def route(self, text):
self.check_update()
complexity = self.classifier.classify(text)
return self.config['model_routing'][complexity]
3. 系统架构与性能优化
3.1 整体架构设计
code复制用户请求 → 预处理层 → 缓存层 → 路由层 → 模型集群
│ │ │ │
│ │ │ └── OpenClaw/Claude/DeepSeek
│ │ └── Redis缓存池
│ └── 对话压缩/摘要生成
└── 微信/钉钉/Webhook接入
关键组件说明：
- 预处理层：负责Token计数、对话历史管理
- 缓存层：实现语义级缓存，支持多级过期策略
- 路由层：动态选择最优模型，支持A/B测试
- 监控模块：实时跟踪各模型消耗和响应质量
3.2 性能优化指标
优化前后关键指标对比：
| 指标 | 优化前 | 优化后 | 降幅 |
|---|---|---|---|
| 日均Token消耗 | 280万 | 103万 | 63% |
| 平均响应延迟 | 420ms | 380ms | 9.5% |
| 缓存命中率 | 0% | 38% | - |
| 小模型使用占比 | 0% | 45% | - |
| 日均成本 | ¥40 | ¥15 | 62.5% |
3.3 监控系统实现
使用Prometheus+Grafana搭建监控看板，关键metrics：
- token_usage_per_model：按模型统计消耗
- cache_hit_rate：缓存命中率
- response_time_bucket：响应时间分布
- complexity_distribution：请求复杂度分布
示例告警规则：
yaml复制groups:
- name: token-alerts
rules:
- alert: HighTokenUsage
expr: sum(rate(token_usage_per_model[5m])) by (model) > 100000
for: 10m
labels:
severity: warning
annotations:
summary: "High token usage on {{ $labels.model }}"
4. 避坑指南与经验总结
4.1 典型问题解决方案
问题1：缓存污染
现象：用户个性化查询被错误缓存
解决方案：
- 添加
no_cache标记位 - 实现敏感词过滤机制
- 对含个人数据的请求自动跳过缓存
问题2：模型路由震荡
现象：同类请求在不同时间被路由到不同模型
解决方案：
- 添加路由结果缓存（短期记忆）
- 实现基于会话ID的亲和性路由
- 设置最小切换间隔（如5分钟内不切换）
问题3：摘要失真
现象：关键信息在摘要过程中丢失
解决方案：
- 添加必保留关键词列表（如时间、地点）
- 实现摘要复核机制（用原模型快速校验）
- 保留原始消息的元数据（如创建时间）
4.2 成本控制心得
-
分级预算策略：
- 简单查询：<0.001元/次
- 中等查询：0.01-0.03元/次
- 复杂查询：0.05-0.1元/次
-
熔断机制：
python复制class TokenBudget:
def __init__(self, daily_limit):
self.limit = daily_limit
self.used = 0
def check(self, estimated_cost):
if self.used + estimated_cost > self.limit * 0.9: # 90%阈值
return "fallback_model"
return None
- 定期优化循环：
- 每周分析Top 20高消耗请求
- 每月评估模型路由效果
- 每季度更新缓存策略
4.3 扩展优化方向
-
渐进式响应：
- 先返回快速响应
- 后台继续完善内容
- 通过消息更新机制推送增强结果
-
客户端缓存：
- 在微信小程序端缓存常见回复
- 实现基于ETag的协商缓存
-
预测性预加载：
- 分析用户行为模式
- 预先执行可能需要的计算
- 结果暂存待用
这套优化方案实施后，不仅解决了OpenClaw的Token消耗问题，更形成了一套可持续的AI服务成本管控体系。核心经验可以总结为：该省省——对简单请求极致优化；该花花——对核心业务保持高质量投入。通过技术手段找到成本与效果的平衡点，才是工程实践的智慧所在。