AI应用开发进阶（三）：成本优化LLM应用如何省钱50%

一、开场：LLM成本是个大问题

大家好，我是老金。

LLM应用最大的问题是什么？

成本。

GPT-4每千Token 0.03美元，一个复杂Agent一天跑下来可能几百美元。

今天聊聊成本优化的实战技巧。

二、成本构成分析

2.1 成本来源

┌─────────────────────────────────────────────────────────┐
│                  LLM应用成本构成                        │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  ┌─────────────────────────────────────────────────┐   │
│  │  Input Tokens（输入）                          │   │
│  │  • 用户问题                                     │   │
│  │  • System Prompt                                │   │
│  │  • Context/知识库                               │   │
│  └─────────────────────────────────────────────────┘   │
│                                                         │
│  ┌─────────────────────────────────────────────────┐   │
│  │  Output Tokens（输出）                          │   │
│  │  • 回答内容                                     │   │
│  │  • Tool参数                                     │   │
│  │  • 思考过程（ReAct）                           │   │
│  └─────────────────────────────────────────────────┘   │
│                                                         │
│  ┌─────────────────────────────────────────────────┐   │
│  │  API调用次数                                    │   │
│  │  • 多轮对话                                     │   │
│  │  • 工具调用                                     │   │
│  │  • 检索调用                                     │   │
│  └─────────────────────────────────────────────────┘   │
│                                                         │
└─────────────────────────────────────────────────────────┘

2.2 模型价格对比

# 2024年主流模型价格对比
MODEL_PRICING = {
    "GPT-4 Turbo": {
        "input": 0.01,    # $0.01 / 1K tokens
        "output": 0.03,  # $0.03 / 1K tokens
        "context": 128000,
        "speed": "medium"
    },
    "GPT-3.5 Turbo": {
        "input": 0.0005,
        "output": 0.0015,
        "context": 16000,
        "speed": "fast"
    },
    "Claude 3 Opus": {
        "input": 0.015,
        "output": 0.075,
        "context": 200000,
        "speed": "slow"
    },
    "Claude 3 Sonnet": {
        "input": 0.003,
        "output": 0.015,
        "context": 200000,
        "speed": "medium"
    },
    "Claude 3 Haiku": {
        "input": 0.00025,
        "output": 0.00125,
        "context": 200000,
        "speed": "fast"
    },
    "Gemini 1.5 Pro": {
        "input": 0.00125,
        "output": 0.005,
        "context": 1000000,
        "speed": "medium"
    }
}

# 计算示例
def calculate_cost(model: str, input_tokens: int, output_tokens: int) -> float:
    """计算单次请求成本"""
    pricing = MODEL_PRICING[model]
    input_cost = (input_tokens / 1000) * pricing["input"]
    output_cost = (output_tokens / 1000) * pricing["output"]
    return input_cost + output_cost

# 示例对比
print("简单问答（100 in + 200 out）:")
for model in ["GPT-4 Turbo", "GPT-3.5 Turbo", "Claude 3 Haiku"]:
    cost = calculate_cost(model, 100, 200)
    print(f"  {model}: ${cost:.6f}")

# 结果：
# GPT-4 Turbo: $0.000700
# GPT-3.5 Turbo: $0.000350
# Claude 3 Haiku: $0.000325

2.3 成本计算器

class CostCalculator:
    """成本计算器"""

    def __init__(self):
        self.pricing = MODEL_PRICING
        self.total_cost = 0
        self.total_tokens = {"input": 0, "output": 0}
        self.call_counts = {}

    def record(self, model: str, input_tokens: int, output_tokens: int):
        """记录一次调用"""
        cost = calculate_cost(model, input_tokens, output_tokens)
        self.total_cost += cost
        self.total_tokens["input"] += input_tokens
        self.total_tokens["output"] += output_tokens

        if model not in self.call_counts:
            self.call_counts[model] = 0
        self.call_counts[model] += 1

    def report(self) -> dict:
        """生成成本报告"""
        return {
            "total_cost": f"${self.total_cost:.4f}",
            "total_input_tokens": self.total_tokens["input"],
            "total_output_tokens": self.total_tokens["output"],
            "total_tokens": sum(self.total_tokens.values()),
            "call_counts": self.call_counts,
            "estimated_monthly": f"${self.total_cost * 30:.2f}" if self.total_cost > 0 else "$0"
        }

# 使用
calculator = CostCalculator()
calculator.record("gpt-4-turbo", 500, 300)
calculator.record("gpt-3.5-turbo", 200, 150)
print(calculator.report())

三、输入优化

3.1 Prompt压缩

# 方案1：LLM压缩
async def compress_context(context: str, max_tokens: int = 2000) -> str:
    """用LLM压缩上下文"""
    prompt = f"""
请将以下文本压缩到约{max_tokens}个token：

{context}

要求：
1. 保留核心信息
2. 删除冗余表述
3. 保持逻辑连贯

压缩后的文本：
"""
    response = await openai.ChatCompletion.acreate(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

# 方案2：规则压缩
import re

def rule_based_compress(text: str, max_chars: int = 4000) -> str:
    """基于规则压缩"""
    # 移除多余空白
    text = re.sub(r's+', ' ', text)

    # 移除重复标点
    text = re.sub(r'([。！？])1+', r'1', text)

    # 如果还太长，按比例截取
    if len(text) > max_chars:
        # 保留开头和结尾
        head = text[:int(max_chars * 0.7)]
        tail = text[-int(max_chars * 0.3):]
        text = head + "n...n" + tail

    return text

# 方案3：智能摘要
def smart_truncate(chunks: List[str], max_chunks: int = 3) -> str:
    """智能截取最相关的chunks"""
    # 按chunk的元数据（如标题匹配度）排序
    scored_chunks = [(i, chunk) for i, chunk in enumerate(chunks)]

    # 选择最相关的max_chunks个
    selected = scored_chunks[:max_chunks]
    selected.sort(key=lambda x: x[0])  # 按原始顺序

    return "nn".join([chunk for _, chunk in selected])

3.2 System Prompt优化

# System Prompt优化
LONG_PROMPT = """
你是一个专业的Python开发工程师。

你具备以下能力：
1. 编写高质量Python代码
2. 代码审查和优化
3. 解决技术问题
4. 解释技术概念
5. 提供最佳实践建议

你的工作原则：
1. 代码优先：给出可运行的代码
2. 简洁明了：不废话，直接给答案
3. 注重可维护性：代码要易于理解和修改
4. 考虑性能：注意算法效率
5. 错误处理：做好异常处理

如果问题不清晰，你会追问。
如果信息不足，你会说明。
如果不确定，你会坦诚告知。
"""

# 优化后
OPTIMIZED_PROMPT = """角色：Python开发工程师
能力：代码编写、审查、优化
原则：代码优先、简洁、直接
"""

# 对比：LONG = 300 tokens, OPTIMIZED = 30 tokens
# 单次节省：270 tokens = $0.000135 (GPT-4)
# 如果每天100次调用，每月节省：$4.05

3.3 知识库精简

# 知识库检索优化
from llama_index import VectorStoreIndex
from llama_index.retrievers import VectorIndexRetriever

# 方案1：限制检索数量
retriever = VectorIndexRetriever(
    index=index,
    similarity_top_k=2,  # 只取Top2（原来是5）
)

# 方案2：限制每个chunk的长度
def truncate_chunks(nodes: List[Node], max_chars: int = 1000) -> List[Node]:
    """截断chunks"""
    for node in nodes:
        if len(node.text) > max_chars:
            node.text = node.text[:max_chars] + "..."
    return nodes

# 方案3：只检索必要字段
SELECTIVE_PROMPT = """
根据以下文档片段回答问题。

只使用与问题直接相关的信息。

问题：{query}

文档片段：
{context}

回答：
"""

四、输出优化

4.1 限制输出长度

# 限制输出长度
from openai import OpenAI

client = OpenAI()

# 设置max_tokens限制输出
def ask_with_limit(
    prompt: str,
    max_output_tokens: int = 200  # 限制最多200 token
) -> str:
    """限制输出的问答"""
    response = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=max_output_tokens
    )
    return response.choices[0].message.content

# 例子：简单问答
# 不限制：可能输出500 tokens
# 限制200：最多$0.0003（vs $0.00075）
# 节省：60%

4.2 结构化输出

# 结构化输出：减少token消耗
import json

# 非结构化输出
UNCODED_OUTPUT = """
让我来回答这个问题。

首先，我们需要...

然后，根据上述分析...

综上所述，答案是...

具体来说，有以下几个方面：
1. 第一点...
2. 第二点...
3. 第三点...

总结一下...
"""

# 结构化输出（JSON）
STRUCTURED_OUTPUT = {
    "answer": "简短答案",
    "key_points": ["要点1", "要点2", "要点3"],
    "conclusion": "结论"
}

# 对比：约500 tokens vs 100 tokens
# 节省：80%

4.3 流式输出

# 流式输出：让用户感知更快
async def stream_chat(prompt: str):
    """流式输出"""
    stream = await openai.ChatCompletion.acreate(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}],
        stream=True
    )

    for chunk in stream:
        if chunk.choices[0].delta.content:
            yield chunk.choices[0].delta.content

# 使用
async for text in stream_chat("什么是Python？"):
    print(text, end="", flush=True)

五、模型选择策略

5.1 路由策略

# 模型路由：根据任务选择最适合的模型
TASK_MODEL_MAPPING = {
    "simple_qa": {
        "model": "gpt-3.5-turbo",
        "condition": lambda x: len(x)  500 or "分析" in x
    },
    "quick_summary": {
        "model": "claude-3-haiku",
        "condition": lambda x: "总结" in x or "摘要" in x
    }
}

def route_model(task: str) -> str:
    """路由到合适的模型"""
    for task_type, config in TASK_MODEL_MAPPING.items():
        if config["condition"](task):
            return config["model"]
    return "gpt-3.5-turbo"  # 默认

# 实际案例
def smart_chat(user_input: str) -> str:
    """智能路由问答"""
    model = route_model(user_input)

    # 不同模型不同Prompt
    if model == "gpt-3.5-turbo":
        prompt = f"简洁回答：{user_input}"
    else:
        prompt = user_input

    return call_llm(model, prompt)

5.2 级联策略

# 级联策略：先用简单模型，复杂再用强模型
async def cascade_chat(query: str) -> str:
    """级联问答"""
    # Step 1：用Haiku快速判断
    quick_answer = await call_llm("claude-3-haiku", f"快速回答：{query}")

    # Step 2：如果Haiku不确定，上一级
    if "不确定" in quick_answer or "不知道" in quick_answer:
        better_answer = await call_llm("claude-3-sonnet", query)
        return better_answer

    # Step 3：如果Sonnet能确定，就用它
    check = await call_llm("claude-3-haiku", 
        f"这个回答正确吗？{quick_answer}n问题：{query}")

    if "不正确" in check or "错误" in check:
        return await call_llm("gpt-4-turbo", query)

    return quick_answer

5.3 模型对比选择

# 根据场景选模型
MODEL_SELECTION = {
    # 场景：简单问答
    "simple_qa": {
        "推荐": "gpt-3.5-turbo",
        "原因": "简单任务不需要贵模型",
        "节省": "90%"
    },

    # 场景：代码生成
    "code_gen": {
        "推荐": "gpt-4-turbo",
        "原因": "代码质量要求高",
        "注意": "可配合gpt-3.5生成、gpt-4审查"
    },

    # 场景：长文本处理
    "long_text": {
        "推荐": "claude-3-sonnet",
        "原因": "200K上下文，减少分段"
    },

    # 场景：实时对话
    "realtime": {
        "推荐": "gpt-3.5-turbo",
        "原因": "速度快，延迟低"
    },

    # 场景：创意写作
    "creative": {
        "推荐": "claude-3-opus",
        "原因": "创意能力强"
    }
}

六、缓存策略

6.1 语义缓存

from llama_index.storage.chat_store import SimpleChatStore
import hashlib

class SemanticCache:
    """语义缓存"""

    def __init__(self, similarity_threshold: float = 0.95):
        self.cache = {}  # embedding -> (query, response)
        self.embedding_model = OpenAIEmbedding()
        self.threshold = similarity_threshold

    def _get_embedding(self, text: str) -> List[float]:
        """获取文本embedding"""
        return self.embedding_model.get_text_embedding(text)

    def _cosine_similarity(self, a: List[float], b: List[float]) -> float:
        """计算余弦相似度"""
        import numpy as np
        return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

    def get(self, query: str) -> Optional[str]:
        """获取缓存"""
        query_emb = self._get_embedding(query)

        for cached_emb, (cached_query, response) in self.cache.items():
            sim = self._cosine_similarity(query_emb, cached_emb)
            if sim >= self.threshold:
                return response

        return None

    def set(self, query: str, response: str):
        """设置缓存"""
        query_emb = self._get_embedding(query)
        self.cache[query_emb] = (query, response)

    def stats(self) -> dict:
        """缓存统计"""
        return {
            "size": len(self.cache),
            "hit_rate_estimate": "根据query分布估算"
        }

# 使用
cache = SemanticCache()

async def cached_chat(query: str) -> str:
    """带缓存的问答"""
    # 检查缓存
    cached = cache.get(query)
    if cached:
        return cached

    # 没有缓存，调用LLM
    response = await call_llm("gpt-3.5-turbo", query)

    # 存入缓存
    cache.set(query, response)

    return response

6.2 精确缓存

# 精确缓存：基于hash
from hashlib import md5

class ExactCache:
    """精确缓存"""

    def __init__(self):
        self.cache = {}

    def _hash(self, text: str) -> str:
        """计算hash"""
        return md5(text.encode()).hexdigest()

    def get(self, prompt: str) -> Optional[str]:
        """获取缓存"""
        key = self._hash(prompt)
        return self.cache.get(key)

    def set(self, prompt: str, response: str):
        """设置缓存"""
        key = self._hash(prompt)
        self.cache[key] = response

    def clear(self):
        """清空缓存"""
        self.cache.clear()

七、API调用优化

7.1 Batch处理

# Batch处理：批量API调用
async def batch_chat(queries: List[str], batch_size: int = 20) -> List[str]:
    """批量处理"""
    results = []

    for i in range(0, len(queries), batch_size):
        batch = queries[i:i+batch_size]

        # 批量调用
        # 注意：OpenAI的batch API还在测试中
        responses = await asyncio.gather(*[
            call_llm("gpt-3.5-turbo", q) for q in batch
        ])

        results.extend(responses)

    return results

7.2 并发控制

import asyncio
from semaphores import Semaphore

class RateLimiter:
    """速率限制器"""

    def __init__(self, max_rpm: int = 60):
        self.semaphore = Semaphore(max_rpm)
        self.last_call = 0

    async def acquire(self):
        """获取许可"""
        await self.semaphore.acquire()
        # 可以添加延迟控制

    def release(self):
        """释放许可"""
        self.semaphore.release()

# 使用
limiter = RateLimiter(max_rpm=30)  # 每分钟30次

async def rate_limited_call(prompt: str) -> str:
    """限流调用"""
    await limiter.acquire()
    try:
        return await call_llm("gpt-3.5-turbo", prompt)
    finally:
        limiter.release()

八、成本监控

8.1 实时监控

from datetime import datetime

class CostMonitor:
    """成本监控"""

    def __init__(self):
        self.daily_cost = {}
        self.alerts = []
        self.budget = 100  # 每日预算 $100

    def record(self, cost: float):
        """记录成本"""
        today = datetime.now().date().isoformat()

        if today not in self.daily_cost:
            self.daily_cost[today] = 0

        self.daily_cost[today] += cost

        # 检查预算
        if self.daily_cost[today] > self.budget:
            self.send_alert(f"今日成本 ${self.daily_cost[today]:.2f} 超过预算 $100")

    def send_alert(self, message: str):
        """发送告警"""
        print(f"⚠️ ALERT: {message}")
        self.alerts.append({
            "time": datetime.now(),
            "message": message
        })

    def report(self) -> dict:
        """生成报告"""
        return {
            "today_cost": self.daily_cost.get(datetime.now().date().isoformat(), 0),
            "budget": self.budget,
            "budget_remaining": self.budget - self.daily_cost.get(datetime.now().date().isoformat(), 0),
            "alert_count": len(self.alerts)
        }

8.2 成本优化报告

def generate_optimization_report(calculator: CostCalculator) -> str:
    """生成优化建议报告"""
    report = calculator.report()

    suggestions = []

    # 基于数据给出建议
    if report["total_tokens"]["input"] > report["total_tokens"]["output"] * 5:
        suggestions.append("📉 Input Token过多，考虑压缩Prompt")

    if calculator.call_counts.get("gpt-4-turbo", 0) > 10:
        suggestions.append("💡 简单任务可用gpt-3.5-turbo替代gpt-4")

    suggestions.append("🔄 启用缓存可节省30-50%成本")
    suggestions.append("📊 建议设置每日预算告警")

    return f"""
成本分析报告
================
总成本：{report['total_cost']}
总Token：{sum(report['total_tokens'].values())}
调用次数：{sum(calculator.call_counts.values())}

优化建议：
{chr(10).join(suggestions)}

预估月成本：{report['estimated_monthly']}
"""

九、总结

优化效果

优化方法	节省比例
模型降级（简单任务）	50-80%
Prompt压缩	20-40%
语义缓存	30-50%
输出限制	20-60%
综合优化	50-70%

实施路径

阶段1：立即生效（0成本）
- ✅ 限制输出长度
- ✅ 启用精确缓存
- ✅ 简单任务用gpt-3.5

阶段2：短期优化（1-2天）
- 🔄 语义缓存
- 🔄 Prompt压缩
- 🔄 模型路由

阶段3：长期优化（1周+）
- 📊 成本监控
- 📊 分析报告
- 📊 持续调优

---

相关阅读

• AI应用开发进阶（二）：RAG优化从60分到90分的实战技巧
• AI应用开发进阶（一）：提示工程从入门到精通
• Agent框架横评（七）：选型指南如何选对你的AI Agent框架