🧠 图解记忆:围绕每个成功任务优化总成本,而不是只盯单次模型价格;点击图片可查看原图。
💡 STAR回答模板
S (Situation):
示例表达(仅在能用本人经历或可复现实验佐证时使用): "我们的AI客服系统每月OpenAI API成本高达$8000,主要是GPT-4调用。老板要求在不降低用户体验的前提下,成本降低50%。"
T (Task):
"目标:
- 月成本从$8000降到$4000 (50%)
- 用户满意度不低于当前4.2/5
- 响应速度不超过3秒"
A (Action):
优化路径1: Token使用优化 (-35%成本)
1. 上下文压缩
python
# Before: 直接把检索到的5个文档全传给LLM
docs = retriever.search(query, k=5)
context = "\n\n".join([doc.content for doc in docs])
# 平均输入: 3500 tokens
prompt = f"""
基于以下文档回答问题:
{context} # ← 很多无关内容
问题: {query}
"""
# After: LLMLingua压缩上下文
from llmlingua import PromptCompressor
compressor = PromptCompressor()
compressed = compressor.compress_prompt(
context,
instruction="保留回答问题所需的关键信息",
target_token=1000, # 目标压缩到1000 tokens
condition_compare=True,
reorder_context="sort" # 相关性排序
)
# 压缩后平均: 950 tokens
# Token减少: (3500-950)/3500 = 73%
# 成本节省: 输入token成本 -73%效果数据:
压缩前: 平均3500 input tokens
压缩后: 平均950 input tokens
准确率: 89% → 87% (下降2%,可接受)
成本节省: $2800/月 (-35%)2. 会话历史管理
展开 Python 代码示例(39 行)
python
class ConversationManager:
def __init__(self, max_history=3):
self.history = []
self.summary = ""
def add_turn(self, user_msg, assistant_msg):
self.history.append({
"user": user_msg,
"assistant": assistant_msg
})
# 超过3轮,摘要最早的一轮
if len(self.history) > 3:
old_turn = self.history.pop(0)
# 用DeepSeek-V4-Flash摘要(便宜)
turn_summary = gpt35_turbo.summarize(
f"用户: {old_turn['user']}\n助手: {old_turn['assistant']}"
)
self.summary += turn_summary + "\n"
def get_context(self):
# 上下文 = 历史摘要 + 最近3轮完整对话
context = []
if self.summary:
context.append(f"历史对话摘要:\n{self.summary}")
for turn in self.history:
context.append(f"用户: {turn['user']}")
context.append(f"助手: {turn['assistant']}")
return "\n".join(context)
# 效果
# Before: 10轮对话 = 10轮完整历史(约5000 tokens)
# After: 10轮对话 = 摘要(200 tokens) + 最近3轮(1500 tokens)
# Token节省: 70%优化路径2: 模型路由 (-25%成本)
原理: 简单问题用便宜模型,复杂问题用贵模型
展开 Python 代码示例(59 行)
python
class ModelRouter:
def __init__(self):
# 分类器: 判断问题复杂度
self.classifier = fasttext.load_model("complexity_classifier.bin")
self.cheap_model = "deepseek-v4-flash" # $0.5/1M tokens
self.expensive_model = "qwen3.5-plus" # $30/1M tokens
def route(self, query):
# 分类: simple / medium / complex
complexity = self.classifier.predict(query)[0][0]
if complexity == "simple":
# 60%的问题是简单问题
# 例: "营业时间?", "怎么退货?"
return self.cheap_model
elif complexity == "medium":
# 30%中等,先试DeepSeek V4-Flash
response = gpt35.generate(query)
# 如果DeepSeek V4-Flash不确定,升级到GPT-4
if self.is_uncertain(response):
return self.expensive_model
else:
return self.cheap_model
else: # complex
# 10%复杂问题,直接GPT-4
# 例: "对比3款产品的性价比"
return self.expensive_model
def is_uncertain(self, response):
# 检测不确定的标志
uncertain_phrases = [
"我不确定",
"可能是",
"这取决于",
"抱歉,我无法"
]
return any(phrase in response for phrase in uncertain_phrases)
# 使用
router = ModelRouter()
model = router.route(user_query)
response = llm_pool[model].generate(query)
# 效果数据:
# - 60%请求用DeepSeek V4-Flash (省$0.5 vs $30)
# - 30%请求先DeepSeek V4-Flash,20%升级到GPT-4
# - 10%请求直接GPT-4
#
# 平均成本 = 0.6*$0.5 + 0.3*(0.8*$0.5 + 0.2*$30) + 0.1*$30
# = $0.3 + $1.92 + $3 = $5.22
# vs 全用GPT-4: $30
# 成本降低: 83%
#
# 考虑质量损失,实际路由策略调整后:
# 成本降低: 25%优化路径3: 语义缓存 (-30%成本)
原理: 相似问题直接返回缓存,不调API
展开 Python 代码示例(66 行)
python
from langchain.cache import RedisSemanticCache
from langchain.embeddings import OpenAIEmbeddings
class SemanticCache:
def __init__(self):
self.redis_client = Redis(host='localhost', port=6379)
self.embedding_model = OpenAIEmbeddings()
self.similarity_threshold = 0.85
def get(self, query):
"""检查缓存"""
# 1. 计算query的embedding
query_embedding = self.embedding_model.embed_query(query)
# 2. 从Redis向量搜索相似query
similar = self.redis_client.search(
query_vector=query_embedding,
k=1,
similarity_threshold=self.similarity_threshold
)
if similar:
# 命中缓存
cached_query, cached_response = similar[0]
print(f"缓存命中: '{query}' ≈ '{cached_query}'")
return cached_response
return None
def set(self, query, response, ttl=86400):
"""存入缓存"""
query_embedding = self.embedding_model.embed_query(query)
self.redis_client.set(
key=query,
value=response,
embedding=query_embedding,
ttl=ttl # 24小时过期
)
# 使用
cache = SemanticCache()
def answer_question(query):
# 先查缓存
cached = cache.get(query)
if cached:
return cached
# 缓存未命中,调用LLM
response = llm.generate(query)
# 存入缓存
cache.set(query, response)
return response
# 示例缓存命中:
# Query1: "怎么退货?"
# Query2: "如何申请退款?" ← 相似度0.91,命中缓存
# Query3: "退货流程是什么?" ← 相似度0.88,命中缓存
# 效果数据:
# 缓存命中率: 35%
# 成本节省: 35% API调用
# 延迟降低: 2.5s → 0.1s (缓存响应)缓存策略优化:
展开 Python 代码示例(33 行)
python
# 问题: 不是所有问题都适合缓存
# 例: "今天天气怎么样?" - 答案每天变化
class SmartCache:
def should_cache(self, query, response):
"""判断是否应该缓存"""
# 1. 静态知识问题 → 缓存
static_patterns = [
r".*是什么",
r".*的定义",
r".*怎么.*", # 流程类
]
if any(re.match(p, query) for p in static_patterns):
return True, 86400*7 # 缓存7天
# 2. 实时数据问题 → 不缓存
dynamic_patterns = [
r".*今天.*",
r".*现在.*",
r".*最新.*",
]
if any(re.match(p, query) for p in dynamic_patterns):
return False, 0
# 3. 订单查询 → 短时缓存
if "订单" in query and re.search(r"\d{6,}", query):
return True, 300 # 缓存5分钟
# 4. 默认: 缓存24小时
return True, 86400R (Result):
"经过3个月优化:
- 月成本从$8000降到**$3600** (-55%,超过目标)
- 用户满意度4.2/5 → 4.4/5 (提升!)
- P95响应时间2.8s → 1.2s (更快!)
成本拆解:
- Token压缩: -$2800 (-35%)
- 模型路由: -$2000 (-25%)
- 语义缓存: -$2400 (-30%)
- 总节省: -$4400 (实际有重叠,最终-55%)
关键insight:
- 35%缓存命中率带来最大收益
- 60%简单问题用DeepSeek V4-Flash完全够用
- 上下文压缩73%,准确率只降2%"
面试追问应对:
Q: "语义缓存会不会返回过时信息?"
"我们做了3层防护:
- TTL分级: 静态知识7天,实时数据不缓存,订单5分钟
- 缓存失效机制: 产品价格变动时,主动清除相关缓存
- 用户反馈: 每个缓存答案底部显示'生成于X分钟前',用户可点击'重新生成'
示例表达(仅在能用本人经历或可复现实验佐证时使用): 实测过时率<1%,用户投诉0。"
Q: "模型路由的分类器怎么训练的?"
示例表达(仅在能用本人经历或可复现实验佐证时使用): "我们用FastText训练:
- 数据标注: 人工标注1000条历史问题为simple/medium/complex
- 特征: 问题长度、实体数量、是否有比较词('对比','哪个更')
- 训练: FastText 3分钟训练完,准确率89%
- 持续学习: 每周用新数据增量训练
如果分类错误,DeepSeek V4-Flash会有'不确定'标志,自动升级到GPT-4,兜底策略保证质量。"
Q: "Token压缩会不会丢失关键信息?"
"我们做了评估:
- AB测试: 200条问题,人工评分压缩前后答案质量
- 结果: 压缩到1000 token时,质量从4.5/5降到4.4/5,损失2% 示例表达(仅在能用本人经历或可复现实验佐证时使用): 3. 阈值选择: 我们用target_token=1000作为最优平衡点
LLMLingua的reorder_context='sort'会把最相关的内容放前面,LLM优先看到,所以质量损失小。"
