背景痛点:在线 Prompt 为什么总“卡脖子”
过去一年,我们把智能客服从规则引擎升级到 LLM 驱动,结果“秒回”变“转圈”,客服同学天天被用户吐槽。线上观察到的三大痛点:
- 响应延迟:高峰期平均首包时间 2.3 s,P99 飙到 5 s,直接击穿 SLA。
- 计算成本:每条对话都要把完整 Prompt 喂给模型,Token 量翻倍,GPU 利用率 90% 却打不满 QPS。
- 模型一致性:在线拼接 Prompt 容易因为字段缺失或顺序变化,导致同一问题给出不同答案,质检得分忽高忽低。
一句话:在线实时拼 Prompt,看起来灵活,其实“又慢又贵还不可控”。
技术对比:在线 vs 离线一张表看清
| 维度 | 在线实时计算 | 离线预处理 |
|---|---|---|
| RTFPS(Request To First Prompt Send) | 200-500 ms | 0 ms(直接命中缓存) |
| 冷启动/Cache Miss 时延 | 1-3 s | 一次预热,后续毫秒级 |
| 成本(Token/GPU) | 高,重复计算 | 低,复用缓存结果 |
| 灵活性 | 高,可动态拼接 | 中,需预定义模板+变量 |
| 数据安全 | 实时清洗压力大 | 可离线脱敏,审计友好 |
| 运维复杂度 | 低,无状态 | 高,需缓存、调度、版本控制 |
结论:离线不是“银弹”,但在 FAQ、工单摘要、标准问答等“高频低变”场景,能把 70% 请求从在线链路砍掉,性价比最高。
核心实现一:离线 Prompt 缓存机制(Python 版)
先给 Prompt 模板算个“指纹”,再把变量值排序后哈希,保证同一模板+同一变量集合只算一次。
# prompt_cache.py import hashlib, json, logging from functools import lru_cache from typing import Dict, Any logging.basicConfig(level=logging.INFO) class PromptCache: """离线 Prompt 缓存器:模板+变量 -> 最终 Prompt 文本""" def __init__(self, ttl_minutes: int = 60): self.ttl = ttl_minutes * 60 self._inner_cache: Dict[str, str] = {} @staticmethod def _make_key(template: str, variables: Dict[str, Any]) -> str: # 变量按 key 排序,保证顺序一致 var_str = json.dumps(variables, sort_keys=True, separators=(',', ':')) return hashlib.sha256(f"{template}|{var_str}".encode()).hexdigest()[:16] def get_or_build(self, template: str, variables: Dict[str, Any]) -> str: key = self._make_key(template, variables) if key in self._inner_cache: logging.info("Hit cache for key=%s", key) return self._inner_cache[key] prompt = template.format(**variables) self._inner_cache[key] = prompt logging.info("Miss cache, built prompt len=%s", len(prompt)) return prompt # LRU 装饰器,单机内存级兜底 @lru_cache(maxsize=1024) def cached_prompt(template: str, var_str: str) -> str: return template.format(**json.loads(var_str)) if __name__ == "__main__": pc = PromptCache() tpl = "请用一句话总结以下工单:{content}" print(pc.get_or_build(tpl, {"content": "用户无法登录,提示密码错误"}))运行效果:第二次相同内容 0 ms 返回,CPU 占用直接归零。
核心实现二:FastAPI 在线评估端点
离线缓存预热后,在线服务只需“读缓存 + 调模型”。下面给出一个最小可运行示例,含类型注解、日志、异常捕获。
# online_service.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from prompt_cache import PromptCache import uvicorn, logging app = FastAPI(title="PromptCache Online Eval") cache = PromptCache() class Query(BaseModel): template_id: str variables: dict @app.post("/eval") def eval_prompt(q: Query) -> dict: try: prompt = cache.get_or_build(q.template_id, q.variables) # TODO: 这里调用 LLM,返回假数据演示 return {"prompt": prompt, "answer": "这是模拟回答"} except Exception as e: logging.exception("eval error") raise HTTPException(status_code=500, detail=str(e)) if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000)压测前先用/warmup批量灌一批常用模板,能把命中率拉到 90% 以上。
避坑指南
1. 敏感数据清洗
离线阶段就把手机号、身份证等正则脱敏成***,在线不再处理,可大幅降低合规风险。示例:
def desensitize(text: str) -> str: import re text = re.sub(r'\d{11}', '***手机***', text) return text2. 分布式缓存一致性
采用“版本号 + 消息队列”双保险:
- 模板变更时,发一条
template.v2事件到 Kafka; - 各节点监听后,主动失效本地 LRU,同时把 Redis 对应 key 设置为
NX过期; - 读侧先查本地内存,miss 再查 Redis,再 miss 才回源构建,避免 Cache Penetration。
性能考量:压测数据说话
JMeter 200 并发、持续 10 min 结果(4C8G 容器,Redis 单节点):
| 指标 | 纯在线 | 离线缓存+在线 |
|---|---|---|
| 平均 RT | 2100 ms | 180 ms |
| P99 RT | 5100 ms | 350 ms |
| QPS | 90 | 1100 |
| GPU Token 消耗 | 100% | 30% |
缓存失效策略:TTL 设 45 min,同时带 5 min 随机 jitter,防止集中失效击穿。
代码规范小结
- 类型注解:所有函数签名带
->; - 异常处理:统一
try/except+HTTPException; - 日志:关键路径
info,异常exception,生产接入 ELK; - 单测:缓存命中率、TTL 过期、并发安全全覆盖。
互动环节:如何设计混合模式调度器?
思考题:业务里既有“标准问答”又有“开放式闲聊”,前者适合离线,后者必须在线动态拼 Prompt,如何在一个系统里智能路由?
参考要点:
- 模板打标签:
template_type = standard | freeform - 路由器根据标签分流:standard 走离线缓存,freeform 走在线拼接
- 监控缓存命中率,低于阈值时自动把 standard 降级到在线,防止体验受损
- 灰度发布:新模板先在线验证 24 h,稳定后一键预热进离线池
写在最后
把 Prompt Engineering 从在线搬到离线,看似只是“加了一层缓存”,实际救了 CPU 钱包,也救了用户耐心。我们目前 80% 的 FAQ 请求已进离线通道,GPU 成本直接砍掉一半。剩下的 20% 长尾场景,就留给在线链路继续“灵活”去吧。下一步想试试把缓存粒度再拆细到“变量差分”,欢迎一起踩坑交流。