Skip to content
🔗 分享本题
查看我的学习进度 →

项目经验模块 Q4 教学图:按成功任务优化 AI 系统成本

🧠 图解记忆:围绕每个成功任务优化总成本,而不是只盯单次模型价格;点击图片可查看原图。

💡 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, 86400

R (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:

  1. 35%缓存命中率带来最大收益
  2. 60%简单问题用DeepSeek V4-Flash完全够用
  3. 上下文压缩73%,准确率只降2%"

面试追问应对:

Q: "语义缓存会不会返回过时信息?"

"我们做了3层防护:

  1. TTL分级: 静态知识7天,实时数据不缓存,订单5分钟
  2. 缓存失效机制: 产品价格变动时,主动清除相关缓存
  3. 用户反馈: 每个缓存答案底部显示'生成于X分钟前',用户可点击'重新生成'

示例表达(仅在能用本人经历或可复现实验佐证时使用): 实测过时率<1%,用户投诉0。"

Q: "模型路由的分类器怎么训练的?"

示例表达(仅在能用本人经历或可复现实验佐证时使用): "我们用FastText训练:

  1. 数据标注: 人工标注1000条历史问题为simple/medium/complex
  2. 特征: 问题长度、实体数量、是否有比较词('对比','哪个更')
  3. 训练: FastText 3分钟训练完,准确率89%
  4. 持续学习: 每周用新数据增量训练

如果分类错误,DeepSeek V4-Flash会有'不确定'标志,自动升级到GPT-4,兜底策略保证质量。"

Q: "Token压缩会不会丢失关键信息?"

"我们做了评估:

  1. AB测试: 200条问题,人工评分压缩前后答案质量
  2. 结果: 压缩到1000 token时,质量从4.5/5降到4.4/5,损失2% 示例表达(仅在能用本人经历或可复现实验佐证时使用): 3. 阈值选择: 我们用target_token=1000作为最优平衡点

LLMLingua的reorder_context='sort'会把最相关的内容放前面,LLM优先看到,所以质量损失小。"

📚 参考:OpenAI:Latency Optimization(降本提速实践)