AI 应用性能优化与成本控制完全教程
适用读者：AI 应用开发者、技术负责人、产品经理 预计阅读时间：25 分钟 最后更新：2026-05
目录
- LLM 成本构成分析
- Token 计费与优化策略
- Prompt 压缩技术
- 模型路由：大小模型分流
- 缓存策略：语义缓存
- 批处理优化
- 流式输出优化
- 模型量化降本
- A/B 测试与 ROI 分析
- 企业级成本监控平台搭建
- 最佳实践总结
1. LLM 成本构成分析
1.1 成本全景图
在构建 AI 应用时，成本远不止 API 调用费。完整的成本构成如下：
┌─────────────────────────────────────────────────────────┐
│ AI 应用总成本 │
├──────────────┬──────────────┬──────────────┬────────────┤
│ 模型推理成本 │ 基础设施成本 │ 开发运维成本 │ 隐性成本 │
├──────────────┼──────────────┼──────────────┼────────────┤
│ • API 调用费 │ • GPU 服务器 │ • 人员工资 │ • 重试浪费 │
│ • Token 消耗 │ • 存储费用 │ • 监控告警 │ • 低质量输出│
│ • 微调费用 │ • 网络带宽 │ • 测试环境 │ • 用户流失 │
│ • 训练成本 │ • CDN │ • CI/CD │ • 合规成本 │
└──────────────┴──────────────┴──────────────┴────────────┘
1.2 主流模型定价对比（2026 年参考）
| 模型 | 输入价格 (\(/1M tokens) | 输出价格 (\)/1M tokens) | 上下文窗口 | 适用场景 | |
|---|---|---|---|---|
| GPT-4o | $2.50 | $10.00 | 128K | 复杂推理 |
| GPT-4o-mini | $0.15 | $0.60 | 128K | 通用任务 |
| Claude 3.5 Sonnet | $3.00 | $15.00 | 200K | 长文档处理 |
| Claude 3.5 Haiku | $0.80 | $4.00 | 200K | 快速响应 |
| DeepSeek-V3 | $0.27 | $1.10 | 128K | 性价比之选 |
| Qwen2.5-72B | $0.35 | $1.40 | 128K | 中文优化 |
| Llama 3.1-8B (自部署) | ~$0.05 | ~$0.10 | 128K | 低成本大批量 |
注：价格为参考值，实际价格可能因地区、用量折扣等因素有所不同。
1.3 成本计算公式
def estimate_monthly_cost(
daily_requests: int,
avg_input_tokens: int,
avg_output_tokens: int,
input_price_per_million: float,
output_price_per_million: float,
retry_rate: float = 0.05
) -> dict:
"""估算月度 LLM 调用成本"""
# 考虑重试率
effective_requests = daily_requests * (1 + retry_rate)
# 日成本
daily_input_cost = (effective_requests * avg_input_tokens / 1_000_000) * input_price_per_million
daily_output_cost = (effective_requests * avg_output_tokens / 1_000_000) * output_price_per_million
daily_total = daily_input_cost + daily_output_cost
# 月成本（30天）
monthly_total = daily_total * 30
return {
'daily_requests': daily_requests,
'effective_daily_requests': int(effective_requests),
'daily_input_cost': round(daily_input_cost, 2),
'daily_output_cost': round(daily_output_cost, 2),
'daily_total': round(daily_total, 2),
'monthly_total': round(monthly_total, 2),
'annual_total': round(monthly_total * 12, 2),
}
# 示例：中等规模客服机器人
cost = estimate_monthly_cost(
daily_requests=10000,
avg_input_tokens=800,
avg_output_tokens=300,
input_price_per_million=2.50,
output_price_per_million=10.00,
retry_rate=0.05
)
for k, v in cost.items():
print(f" {k}: {v}")
输出：
daily_requests: 10000
effective_daily_requests: 10500
daily_input_cost: $21.0
daily_output_cost: $31.5
daily_total: $52.5
monthly_total: $1575.0
annual_total: $18900.0
2. Token 计费与优化策略
2.1 理解 Token
Token 是 LLM 处理文本的基本单位。不同模型的分词方式不同：
- 英文：约 1 token ≈ 4 个字符 ≈ 0.75 个单词
- 中文：约 1 token ≈ 1-2 个汉字（取决于模型和分词器）
import tiktoken
def count_tokens(text: str, model: str = "gpt-4o") -> int:
"""计算文本的 Token 数量"""
try:
encoding = tiktoken.encoding_for_model(model)
except KeyError:
encoding = tiktoken.get_encoding("cl100k_base")
return len(encoding.encode(text))
# 对比中英文 Token 效率
en_text = "What is the average salary of employees in the engineering department?"
zh_text = "工程部门员工的平均薪资是多少？"
print(f"英文 ({len(en_text)} chars): {count_tokens(en_text)} tokens")
print(f"中文 ({len(zh_text)} chars): {count_tokens(zh_text)} tokens")
# 英文 (71 chars): ~18 tokens
# 中文 (15 chars): ~12 tokens
2.2 Token 优化策略
策略一：精简系统 Prompt
# ❌ 冗余的系统 Prompt（~200 tokens）
BAD_SYSTEM_PROMPT = """
你是一个非常专业且经验丰富的客户服务代表。你需要以友好、耐心、专业的态度来帮助用户解决问题。
在回答问题时，请确保你的回答准确、清晰、易懂。如果遇到你不确定的问题，请坦诚告知用户。
你应该始终以用户为中心，提供高质量的服务体验。记住，用户体验是最重要的。
请用中文回答所有问题。
"""
# ✅ 精简的系统 Prompt（~60 tokens）
GOOD_SYSTEM_PROMPT = """你是客服代表。规则：
1. 用中文简洁准确地回答
2. 不确定时坦诚说明
3. 优先解决用户问题"""
# 节省 ~70% tokens
策略二：压缩历史对话
def compress_history(messages: list, max_turns: int = 10) -> list:
"""压缩对话历史，保留关键信息"""
if len(messages) <= max_turns:
return messages
# 保留系统消息
system_msgs = [m for m in messages if m['role'] == 'system']
other_msgs = [m for m in messages if m['role'] != 'system']
# 保留最近 N 轮
recent = other_msgs[-max_turns:]
# 中间部分压缩为摘要
middle = other_msgs[:-max_turns]
if middle:
summary = summarize_conversation(middle)
summary_msg = {
'role': 'system',
'content': f'[之前的对话摘要] {summary}'
}
return system_msgs + [summary_msg] + recent
return system_msgs + recent
def summarize_conversation(messages: list) -> str:
"""将多轮对话压缩为简短摘要"""
# 简单实现：提取关键信息
key_points = []
for msg in messages:
if msg['role'] == 'user':
# 提取用户的关键词
key_points.append(f"用户问了关于：{msg['content'][:50]}")
return "；".join(key_points[-3:]) # 只保留最近3个要点
策略三：结构化输出减少 Token
# ❌ 自由文本输出（token 多）
BAD_PROMPT = "请详细描述这个产品的所有特点和优势，包括各个方面..."
# ✅ 结构化输出（token 少且可控）
GOOD_PROMPT = """分析产品特点，用以下JSON格式返回：
{
"优点": ["点1", "点2"],
"缺点": ["点1"],
"评分": 8.5,
"一句话总结": "..."
}"""
3. Prompt 压缩技术
3.1 Selective Context（选择性上下文）
Selectively remove low-information tokens from the context.
import numpy as np
from sentence_transformers import SentenceTransformer
class SelectiveContextCompressor:
"""基于信息密度的选择性上下文压缩"""
def __init__(self, model_name: str = 'all-MiniLM-L6-v2'):
self.model = SentenceTransformer(model_name)
def compress(self, context: str, query: str, keep_ratio: float = 0.5) -> str:
"""
压缩上下文，保留与查询最相关的部分
Args:
context: 原始上下文文本
query: 用户查询
keep_ratio: 保留比例 (0-1)
"""
# 将上下文分句
sentences = self._split_sentences(context)
if len(sentences) <= 3:
return context
# 计算查询和每个句子的嵌入
query_emb = self.model.encode([query])[0]
sent_embs = self.model.encode(sentences)
# 计算每个句子与查询的相关性分数
scores = []
for emb in sent_embs:
score = np.dot(query_emb, emb) / (
np.linalg.norm(query_emb) * np.linalg.norm(emb)
)
scores.append(float(score))
# 按相关性排序，保留 top-k
k = max(2, int(len(sentences) * keep_ratio))
top_indices = np.argsort(scores)[-k:]
top_indices = sorted(top_indices) # 保持原始顺序
compressed = ' '.join(sentences[i] for i in top_indices)
compression_ratio = len(compressed) / len(context)
print(f"压缩率: {compression_ratio:.1%} ({len(context)} → {len(compressed)} chars)")
return compressed
def _split_sentences(self, text: str) -> list:
import re
sentences = re.split(r'[。！？\n.!?]+', text)
return [s.strip() for s in sentences if s.strip()]
# 使用示例
compressor = SelectiveContextCompressor()
long_context = """
人工智能（AI）是计算机科学的一个分支，致力于创建能够执行通常需要人类智能的系统。
机器学习是AI的一个子集，它使计算机能够从数据中学习而无需显式编程。
深度学习是机器学习的一个子集，使用多层神经网络来处理复杂模式。
自然语言处理（NLP）是AI的一个领域，专注于计算机与人类语言之间的交互。
计算机视觉是另一个重要的AI领域，使计算机能够理解和解释视觉信息。
强化学习是一种机器学习方法，通过与环境交互来学习最优策略。
大语言模型（LLM）是近年来最重要的AI突破之一，能够理解和生成自然语言。
"""
query = "什么是大语言模型？"
compressed = compressor.compress(long_context, query, keep_ratio=0.4)
print(compressed)
# 输出主要保留了关于 NLP 和 LLM 的句子
3.2 LLMLingua 压缩
LLMLingua 是微软开源的 Prompt 压缩框架，可以在不损失关键信息的情况下大幅减少 Token 数。
# LLMLingua 使用示例（需安装 pip install llmlingua）
from llmlingua import PromptCompressor
compressor = PromptCompressor(
model_name="microsoft/llmlingua-2-xlm-roberta-large-meetingbank",
device_map="cpu"
)
original_prompt = """
根据以下文档回答问题：
文档内容：
[此处可能有数千字的文档内容...]
问题：这个项目的预算是多少？
"""
compressed = compressor.compress_prompt(
original_prompt,
rate=0.5, # 压缩到原来的 50%
force_tokens=['?', '？', '问题', '预算'], # 强制保留的关键词
context_budget="+100" # 额外保留100个token的上下文
)
print(f"原始 Token 数: {compressed['origin_tokens']}")
print(f"压缩后 Token 数: {compressed['compressed_tokens']}")
print(f"压缩率: {compressed['compressed_tokens']/compressed['origin_tokens']:.1%}")
3.3 Prompt 压缩策略对比
| 方法 | 压缩率 | 质量损失 | 速度 | 适用场景 |
|---|---|---|---|---|
| 精简改写 | 30-50% | 低 | 快 | 系统 Prompt |
| Selective Context | 40-60% | 中 | 快 | 长文档 QA |
| LLMLingua | 50-70% | 低-中 | 中 | 通用压缩 |
| 滑动窗口 | 可控 | 低 | 快 | 流式对话 |
| 关键信息提取 | 60-80% | 中-高 | 中 | 摘要任务 |
4. 模型路由：大小模型分流
4.1 核心思想
并非所有请求都需要最强的模型。通过智能路由，将简单任务分配给小模型，复杂任务分配给大模型，可以在保持质量的同时大幅降低成本。
用户请求 ──▶ 路由分类器 ─┬─▶ 简单任务 ──▶ 小模型 (GPT-4o-mini) ──▶ 返回
│
├─▶ 中等任务 ──▶ 中模型 (DeepSeek-V3) ──▶ 返回
│
└─▶ 复杂任务 ──▶ 大模型 (GPT-4o) ──▶ 返回
4.2 路由分类器实现
from enum import Enum
from dataclasses import dataclass
class TaskComplexity(Enum):
SIMPLE = "simple" # 简单：问候、FAQ、格式转换
MODERATE = "moderate" # 中等：信息提取、简单分析
COMPLEX = "complex" # 复杂：推理、创作、代码生成
@dataclass
class ModelConfig:
name: str
cost_per_1k_input: float
cost_per_1k_output: float
max_context: int
speed_rating: str # fast, medium, slow
MODEL_REGISTRY = {
TaskComplexity.SIMPLE: ModelConfig("gpt-4o-mini", 0.00015, 0.0006, 128000, "fast"),
TaskComplexity.MODERATE: ModelConfig("deepseek-v3", 0.00027, 0.0011, 128000, "medium"),
TaskComplexity.COMPLEX: ModelConfig("gpt-4o", 0.0025, 0.01, 128000, "slow"),
}
class ModelRouter:
"""智能模型路由器"""
# 基于规则的快速分类
SIMPLE_PATTERNS = [
r'你好|hello|hi|谢谢|thank',
r'你是谁|你叫什么',
r'几点|天气|日期',
r'翻译.*单词|convert|换算',
]
COMPLEX_PATTERNS = [
r'分析|analyze|比较|compare',
r'代码|code|实现|implement',
r'推理|reason|证明|prove',
r'创作|写.*文章|write.*essay',
r'策略|方案|设计|design',
]
def __init__(self, classifier_model=None):
self.classifier = classifier_model
def route(self, user_message: str, context: dict = None) -> ModelConfig:
"""路由请求到合适的模型"""
import re
message_lower = user_message.lower()
# 第一层：基于规则的快速分类
for pattern in self.SIMPLE_PATTERNS:
if re.search(pattern, message_lower):
return MODEL_REGISTRY[TaskComplexity.SIMPLE]
for pattern in self.COMPLEX_PATTERNS:
if re.search(pattern, message_lower):
return MODEL_REGISTRY[TaskComplexity.COMPLEX]
# 第二层：基于特征的分类
features = self._extract_features(user_message, context)
complexity = self._classify_by_features(features)
return MODEL_REGISTRY[complexity]
def _extract_features(self, message: str, context: dict = None) -> dict:
"""提取消息特征"""
return {
'length': len(message),
'question_marks': message.count('?') + message.count('？'),
'has_code': '```' in message or 'def ' in message,
'conversation_turns': context.get('turns', 0) if context else 0,
'word_count': len(message.split()),
}
def _classify_by_features(self, features: dict) -> TaskComplexity:
"""基于特征判断复杂度"""
score = 0
# 消息长度
if features['length'] > 500:
score += 2
elif features['length'] > 100:
score += 1
# 包含代码
if features['has_code']:
score += 2
# 对话轮次（多轮对话通常更复杂）
if features['conversation_turns'] > 5:
score += 1
# 多个问题
if features['question_marks'] > 2:
score += 1
if score >= 3:
return TaskComplexity.COMPLEX
elif score >= 1:
return TaskComplexity.MODERATE
else:
return TaskComplexity.SIMPLE
# 使用示例
router = ModelRouter()
test_messages = [
"你好",
"帮我分析一下这个季度的销售数据，找出下滑的原因并给出改进建议",
"今天天气怎么样？",
"请用Python实现一个快速排序算法，并解释时间复杂度",
]
for msg in test_messages:
config = router.route(msg)
print(f"[{config.name:15s}] {msg[:40]}...")
4.3 基于 LLM 的路由（更准确）
class LLMRouter:
"""使用小模型作为路由分类器"""
ROUTER_PROMPT = """将以下用户请求分类为 simple、moderate 或 complex。
分类标准：
- simple: 问候、简单问答、格式转换、翻译单词
- moderate: 信息查找、简单总结、基本分析
- complex: 深度推理、代码生成、创作写作、多步骤分析
只返回分类标签，不要解释。
用户请求: {message}
分类:"""
def __init__(self, llm_client):
# 使用最小最快的模型作为分类器
self.classifier = llm_client
async def route(self, message: str) -> TaskComplexity:
prompt = self.ROUTER_PROMPT.format(message=message)
result = await self.classifier.generate(prompt, max_tokens=10)
result = result.strip().lower()
mapping = {
'simple': TaskComplexity.SIMPLE,
'moderate': TaskComplexity.MODERATE,
'complex': TaskComplexity.COMPLEX,
}
return mapping.get(result, TaskComplexity.MODERATE)
4.4 路由效果评估
def calculate_routing_savings(
test_cases: list,
router: ModelRouter,
without_routing_model: str = "gpt-4o"
) -> dict:
"""计算路由策略的成本节省"""
total_with_routing = 0
total_without_routing = 0
for case in test_cases:
message = case['message']
input_tokens = case['input_tokens']
output_tokens = case['output_tokens']
# 有路由的情况
model_config = router.route(message)
routed_cost = (
input_tokens * model_config.cost_per_1k_input / 1000 +
output_tokens * model_config.cost_per_1k_output / 1000
)
total_with_routing += routed_cost
# 无路由的情况（全部使用大模型）
big_model = MODEL_REGISTRY[TaskComplexity.COMPLEX]
no_route_cost = (
input_tokens * big_model.cost_per_1k_input / 1000 +
output_tokens * big_model.cost_per_1k_output / 1000
)
total_without_routing += no_route_cost
savings = total_without_routing - total_with_routing
savings_pct = savings / total_without_routing * 100
return {
'total_with_routing': round(total_with_routing, 4),
'total_without_routing': round(total_without_routing, 4),
'savings': round(savings, 4),
'savings_percentage': f"{savings_pct:.1f}%",
}
5. 缓存策略：语义缓存
5.1 为什么需要语义缓存
传统的精确匹配缓存对 LLM 应用几乎无用——用户每次提问的措辞都不同。语义缓存通过语义相似度匹配，即使问题措辞不同也能命中缓存。
传统缓存：
"北京天气如何？" → 缓存命中 ✅
"今天北京天气怎么样？" → 缓存未命中 ❌（措辞不同）
语义缓存：
"北京天气如何？" → 缓存命中 ✅
"今天北京天气怎么样？" → 缓存命中 ✅（语义相同）
5.2 语义缓存实现
import numpy as np
from sentence_transformers import SentenceTransformer
import hashlib
import json
import time
from typing import Optional
class SemanticCache:
"""基于语义相似度的 LLM 响应缓存"""
def __init__(
self,
similarity_threshold: float = 0.92,
ttl_seconds: int = 3600,
max_size: int = 10000,
model_name: str = 'all-MiniLM-L6-v2'
):
self.threshold = similarity_threshold
self.ttl = ttl_seconds
self.max_size = max_size
self.model = SentenceTransformer(model_name)
# 缓存结构：{embedding_key: (embedding, response, timestamp, metadata)}
self.cache: dict = {}
self.hit_count = 0
self.miss_count = 0
def _get_embedding(self, text: str) -> np.ndarray:
"""获取文本嵌入"""
return self.model.encode([text])[0]
def _cosine_similarity(self, a: np.ndarray, b: np.ndarray) -> float:
"""计算余弦相似度"""
return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))
def get(self, query: str) -> Optional[dict]:
"""查找语义相似的缓存"""
if not self.cache:
self.miss_count += 1
return None
query_emb = self._get_embedding(query)
best_match = None
best_score = 0
for key, (cached_emb, response, timestamp, metadata) in self.cache.items():
# 检查 TTL
if time.time() - timestamp > self.ttl:
continue
score = self._cosine_similarity(query_emb, cached_emb)
if score > best_score:
best_score = score
best_match = (key, response, metadata)
if best_match and best_score >= self.threshold:
self.hit_count += 1
key, response, metadata = best_match
return {
'response': response,
'similarity': best_score,
'cached_query': metadata.get('original_query', ''),
'cache_age_seconds': int(time.time() - metadata.get('timestamp', 0)),
}
self.miss_count += 1
return None
def set(self, query: str, response: str, metadata: dict = None):
"""缓存查询结果"""
# 清理过期缓存
self._evict_expired()
# 如果缓存满了，删除最旧的
if len(self.cache) >= self.max_size:
oldest_key = min(self.cache, key=lambda k: self.cache[k][2])
del self.cache[oldest_key]
embedding = self._get_embedding(query)
key = hashlib.md5(query.encode()).hexdigest()
self.cache[key] = (
embedding,
response,
time.time(),
{**(metadata or {}), 'original_query': query, 'timestamp': time.time()}
)
def _evict_expired(self):
"""清理过期缓存"""
now = time.time()
expired_keys = [
k for k, (_, _, ts, _) in self.cache.items()
if now - ts > self.ttl
]
for k in expired_keys:
del self.cache[k]
@property
def hit_rate(self) -> float:
total = self.hit_count + self.miss_count
return self.hit_count / total if total > 0 else 0
def stats(self) -> dict:
return {
'size': len(self.cache),
'hit_count': self.hit_count,
'miss_count': self.miss_count,
'hit_rate': f"{self.hit_rate:.1%}",
'max_size': self.max_size,
'ttl_seconds': self.ttl,
}
# 使用示例
cache = SemanticCache(similarity_threshold=0.90, ttl_seconds=3600)
# 模拟 LLM 调用
def mock_llm_call(query: str) -> str:
# 实际中这里调用 LLM API
return f"这是对'{query}'的回答"
# 第一次查询
query1 = "Python怎么读取CSV文件？"
cached = cache.get(query1)
if cached:
print(f"缓存命中: {cached['response']}")
else:
response = mock_llm_call(query1)
cache.set(query1, response)
print(f"缓存未命中，已缓存: {response}")
# 语义相似的查询
query2 = "如何用Python打开CSV文件？"
cached = cache.get(query2)
if cached:
print(f"缓存命中 (相似度: {cached['similarity']:.2f}): {cached['response']}")
else:
print("缓存未命中")
print(f"\n缓存统计: {cache.stats()}")
5.3 向量数据库加速
当缓存量超过万级时，使用向量数据库替代内存缓存：
class VectorDBSemanticCache:
"""基于向量数据库的语义缓存（以 Qdrant 为例）"""
def __init__(self, collection_name: str = "llm_cache"):
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams
self.client = QdrantClient(host="localhost", port=6333)
self.collection = collection_name
self.model = SentenceTransformer('all-MiniLM-L6-v2')
# 创建集合（如果不存在）
try:
self.client.get_collection(collection_name)
except Exception:
self.client.create_collection(
collection_name=collection_name,
vectors_config=VectorParams(
size=384, # all-MiniLM-L6-v2 的维度
distance=Distance.COSINE
)
)
def get(self, query: str, threshold: float = 0.92):
embedding = self.model.encode([query])[0].tolist()
results = self.client.search(
collection_name=self.collection,
query_vector=embedding,
limit=1,
score_threshold=threshold
)
if results:
hit = results[0]
return {
'response': hit.payload.get('response'),
'similarity': hit.score,
'cached_query': hit.payload.get('query'),
}
return None
def set(self, query: str, response: str):
from qdrant_client.models import PointStruct
import uuid
embedding = self.model.encode([query])[0].tolist()
self.client.upsert(
collection_name=self.collection,
points=[
PointStruct(
id=str(uuid.uuid4()),
vector=embedding,
payload={'query': query, 'response': response}
)
]
)
6. 批处理优化
6.1 为什么批处理能降本
大多数 LLM API 对批处理（Batch）请求提供折扣，同时批量推理在 GPU 利用率上更高效。
| 处理方式 | 价格折扣 | 延迟 | 适用场景 |
|---|---|---|---|
| 实时 API | 基准价 | 低 | 用户交互 |
| 批处理 API | 50% 折扣 | 高（数小时） | 数据处理、报告生成 |
| 异步队列 | 自部署优化 | 中 | 后台任务 |
6.2 批处理队列实现
import asyncio
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Callable, Any
import time
@dataclass
class BatchRequest:
"""单个批处理请求"""
request_id: str
prompt: str
future: asyncio.Future = field(default_factory=asyncio.Future)
created_at: float = field(default_factory=time.time)
class BatchProcessor:
"""LLM 批处理器：自动攒批，减少 API 调用次数"""
def __init__(
self,
llm_client,
max_batch_size: int = 20,
max_wait_seconds: float = 2.0,
on_batch_complete: Callable = None
):
self.llm = llm_client
self.max_batch_size = max_batch_size
self.max_wait = max_wait_seconds
self.on_complete = on_batch_complete
self._queue: list[BatchRequest] = []
self._processing = False
self._stats = defaultdict(int)
async def submit(self, prompt: str) -> str:
"""提交请求到批处理队列"""
request = BatchRequest(
request_id=f"req_{self._stats['total_requests']}",
prompt=prompt
)
self._queue.append(request)
self._stats['total_requests'] += 1
# 触发批处理
if not self._processing:
asyncio.create_task(self._process_batch())
# 等待结果
return await request.future
async def _process_batch(self):
"""处理一批请求"""
self._processing = True
# 等待凑批或超时
start = time.time()
while (
len(self._queue) < self.max_batch_size and
time.time() - start < self.max_wait
):
await asyncio.sleep(0.1)
if not self._queue:
self._processing = False
return
# 取出当前批次
batch = self._queue[:self.max_batch_size]
self._queue = self._queue[self.max_batch_size:]
self._stats['batch_count'] += 1
self._stats['avg_batch_size'] = (
self._stats.get('avg_batch_size', 0) * (self._stats['batch_count'] - 1) +
len(batch)
) / self._stats['batch_count']
try:
# 批量调用 LLM
prompts = [req.prompt for req in batch]
responses = await self.llm.batch_generate(prompts)
# 返回结果
for req, response in zip(batch, responses):
if not req.future.done():
req.future.set_result(response)
if self.on_complete:
self.on_complete(len(batch), True)
except Exception as e:
for req in batch:
if not req.future.done():
req.future.set_exception(e)
if self.on_complete:
self.on_complete(len(batch), False)
self._processing = False
# 如果还有待处理的，继续
if self._queue:
asyncio.create_task(self._process_batch())
def get_stats(self) -> dict:
return dict(self._stats)
# 使用示例
async def main():
# 假设的 LLM 客户端
class MockLLM:
async def batch_generate(self, prompts):
print(f" 批量处理 {len(prompts)} 个请求")
await asyncio.sleep(0.5) # 模拟 API 调用
return [f"回答: {p[:20]}..." for p in prompts]
processor = BatchProcessor(
llm_client=MockLLM(),
max_batch_size=10,
max_wait_seconds=1.0
)
# 并发提交多个请求
tasks = []
for i in range(25):
task = processor.submit(f"问题 {i}: 这是一个测试问题")
tasks.append(task)
results = await asyncio.gather(*tasks)
print(f"\n处理完成，共 {len(results)} 个结果")
print(f"统计: {processor.get_stats()}")
# asyncio.run(main())
6.3 OpenAI Batch API 使用
import openai
import json
def create_batch_file(requests: list, output_path: str):
"""创建 OpenAI Batch API 所需的 JSONL 文件"""
with open(output_path, 'w') as f:
for i, req in enumerate(requests):
line = {
"custom_id": f"request-{i}",
"method": "POST",
"url": "/v1/chat/completions",
"body": {
"model": "gpt-4o-mini",
"messages": req['messages'],
"max_tokens": req.get('max_tokens', 500),
}
}
f.write(json.dumps(line) + '\n')
def submit_batch(jsonl_path: str):
"""提交批处理任务"""
client = openai.OpenAI()
# 上传文件
with open(jsonl_path, 'rb') as f:
file_obj = client.files.create(file=f, purpose='batch')
# 创建批处理任务
batch = client.batches.create(
input_file_id=file_obj.id,
endpoint="/v1/chat/completions",
completion_window="24h", # 24小时内完成
metadata={"description": "批量数据处理"}
)
print(f"批处理任务已创建: {batch.id}")
print(f"状态: {batch.status}")
print(f"预计完成时间: 24小时内")
print(f"价格折扣: 约50%")
return batch.id
def check_batch_status(batch_id: str):
"""检查批处理状态"""
client = openai.OpenAI()
batch = client.batches.retrieve(batch_id)
print(f"状态: {batch.status}")
print(f"已处理: {batch.request_counts.completed}/{batch.request_counts.total}")
if batch.status == 'completed':
# 下载结果
result = client.files.content(batch.output_file_id)
return json.loads(result.read())
return None
7. 流式输出优化
7.1 流式输出的价值
流式输出不直接降低成本，但显著改善用户感知延迟，减少用户等待导致的取消请求（减少浪费）。
# 非流式：用户等待完整响应生成
# 延迟 = TTFB + 完整生成时间（可能 5-30 秒）
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": "详细解释量子计算"}],
stream=False
)
# 流式：用户立即看到输出
# 感知延迟 = TTFB（通常 < 1 秒）
stream = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": "详细解释量子计算"}],
stream=True
)
for chunk in stream:
if chunk.choices[0].delta.content:
print(chunk.choices[0].delta.content, end="", flush=True)
7.2 流式处理中间件
import asyncio
from typing import AsyncGenerator
class StreamingMiddleware:
"""流式输出中间件：处理、过滤、优化流式响应"""
def __init__(self, llm_client):
self.llm = llm_client
self.buffer = ""
self.token_count = 0
async def stream_with_fallback(
self,
messages: list,
model: str = "gpt-4o",
fallback_model: str = "gpt-4o-mini",
timeout_seconds: float = 5.0
) -> AsyncGenerator[str, None]:
"""带超时降级的流式输出"""
try:
# 尝试主模型，设置超时
async for chunk in self._stream_with_timeout(
messages, model, timeout_seconds
):
yield chunk
except asyncio.TimeoutError:
# 主模型超时，降级到小模型
yield "[切换到备用模型...]\n"
async for chunk in self._stream(messages, fallback_model):
yield chunk
async def _stream_with_timeout(
self, messages: list, model: str, timeout: float
) -> AsyncGenerator[str, None]:
"""带超时的流式调用"""
stream = self.llm.chat.completions.create(
model=model, messages=messages, stream=True
)
start = time.time()
first_token = True
for chunk in stream:
if first_token:
ttfb = time.time() - start
if ttfb > timeout:
raise asyncio.TimeoutError()
first_token = False
content = chunk.choices[0].delta.content
if content:
self.token_count += 1
yield content
async def _stream(self, messages: list, model: str) -> AsyncGenerator[str, None]:
"""普通流式调用"""
stream = self.llm.chat.completions.create(
model=model, messages=messages, stream=True
)
for chunk in stream:
content = chunk.choices[0].delta.content
if content:
yield content
def get_stats(self) -> dict:
return {'tokens_streamed': self.token_count}
7.3 Server-Sent Events (SSE) 接口
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import json
app = FastAPI()
@app.post("/chat/stream")
async def chat_stream(request: dict):
"""SSE 流式聊天接口"""
async def generate():
messages = request.get('messages', [])
# 发送开始事件
yield f"data: {json.dumps({'type': 'start'})}\n\n"
stream = await llm_client.chat.completions.create(
model="gpt-4o",
messages=messages,
stream=True
)
full_response = ""
async for chunk in stream:
content = chunk.choices[0].delta.content
if content:
full_response += content
yield f"data: {json.dumps({'type': 'content', 'text': content})}\n\n"
# 发送完成事件（包含用量统计）
yield f"data: {json.dumps({'type': 'done', 'full_response': full_response})}\n\n"
return StreamingResponse(
generate(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
}
)
8. 模型量化降本
8.1 量化概述
模型量化通过降低权重精度（FP32 → INT8/INT4）减少显存占用和推理成本，是自部署模型降本的核心手段。
| 精度 | 显存占用（7B 模型） | 速度 | 质量损失 |
|---|---|---|---|
| FP32 | ~28 GB | 基准 | 无 |
| FP16 | ~14 GB | 1.5x | 可忽略 |
| INT8 | ~7 GB | 2x | 轻微 |
| INT4 | ~3.5 GB | 3x | 中等 |
| GPTQ-4bit | ~3.5 GB | 2.5x | 低 |
| AWQ-4bit | ~3.5 GB | 3x | 低 |
8.2 使用 bitsandbytes 量化
from transformers import AutoModelForCausalLM, AutoTokenizer
import torch
def load_quantized_model(model_name: str, bits: int = 4):
"""加载量化模型"""
from transformers import BitsAndBytesConfig
if bits == 4:
quantization_config = BitsAndBytesConfig(
load_in_4bit=True,
bnb_4bit_compute_dtype=torch.float16,
bnb_4bit_use_double_quant=True, # 双重量化，进一步压缩
bnb_4bit_quant_type="nf4", # NF4 比 FP4 更好
)
elif bits == 8:
quantization_config = BitsAndBytesConfig(
load_in_8bit=True,
)
else:
raise ValueError(f"不支持的量化位数: {bits}")
model = AutoModelForCausalLM.from_pretrained(
model_name,
quantization_config=quantization_config,
device_map="auto",
)
tokenizer = AutoTokenizer.from_pretrained(model_name)
# 打印显存使用
if torch.cuda.is_available():
memory_gb = torch.cuda.memory_allocated() / (1024**3)
print(f"GPU 显存使用: {memory_gb:.2f} GB")
return model, tokenizer
# 加载 Qwen2.5-72B 的 4-bit 量化版本
# 原始需要 ~140 GB 显存，4-bit 只需 ~35 GB
model, tokenizer = load_quantized_model("Qwen/Qwen2.5-72B-Instruct", bits=4)
8.3 成本对比计算
def compare_deployment_cost(
model_params_b: float,
requests_per_day: int,
avg_input_tokens: int,
avg_output_tokens: int
) -> dict:
"""对比不同精度的部署成本"""
# 每个参数的显存占用（字节）
memory_per_param = {
'fp32': 4, 'fp16': 2, 'int8': 1, 'int4': 0.5
}
# GPU 租赁价格（$/hour，参考）
gpu_prices = {
'A100_80GB': 2.0,
'A100_40GB': 1.5,
'A6000_48GB': 1.2,
'RTX4090_24GB': 0.8,
'RTX3090_24GB': 0.5,
}
results = {}
for precision, bytes_per_param in memory_per_param.items():
# 计算显存需求
memory_gb = model_params_b * 1e9 * bytes_per_param / (1024**3)
# 选择合适的 GPU
gpu_choice = None
for gpu, capacity in [
('A100_80GB', 80), ('A100_40GB', 40),
('A6000_48GB', 48), ('RTX4090_24GB', 24)
]:
if memory_gb < capacity * 0.85: # 留 15% 余量
gpu_choice = gpu
break
if not gpu_choice:
results[precision] = {'error': '显存不足，需要多卡'}
continue
# 计算月成本
monthly_cost = gpu_prices[gpu_choice] * 24 * 30
results[precision] = {
'memory_gb': round(memory_gb, 1),
'gpu': gpu_choice,
'monthly_cost_usd': round(monthly_cost, 0),
'cost_per_1k_requests': round(monthly_cost / (requests_per_day * 30 / 1000), 2),
}
return results
# 对比 Qwen2.5-72B 不同精度
costs = compare_deployment_cost(
model_params_b=72,
requests_per_day=10000,
avg_input_tokens=500,
avg_output_tokens=200
)
for precision, info in costs.items():
if 'error' in info:
print(f"{precision:6s}: {info['error']}")
else:
print(f"{precision:6s}: {info['memory_gb']}GB → {info['gpu']} → ${info['monthly_cost_usd']}/月")
9. A/B 测试与 ROI 分析
9.1 A/B 测试框架
import random
import hashlib
from dataclasses import dataclass
from typing import Any
import json
import time
@dataclass
class ABTestConfig:
test_name: str
variants: dict # {'control': {...}, 'treatment': {...}}
traffic_split: dict # {'control': 0.5, 'treatment': 0.5}
metrics: list # 要追踪的指标
class ABTestManager:
"""A/B 测试管理器"""
def __init__(self):
self.active_tests: dict[str, ABTestConfig] = {}
self.results: dict[str, list] = {}
def create_test(self, config: ABTestConfig):
"""创建 A/B 测试"""
self.active_tests[config.test_name] = config
self.results[config.test_name] = []
print(f"测试 '{config.test_name}' 已创建，变体: {list(config.variants.keys())}")
def assign_variant(self, test_name: str, user_id: str) -> str:
"""为用户分配变体（基于用户 ID 的确定性哈希）"""
config = self.active_tests[test_name]
# 使用哈希确保同一用户始终看到同一变体
hash_val = int(hashlib.md5(f"{test_name}:{user_id}".encode()).hexdigest(), 16)
normalized = (hash_val % 10000) / 10000
cumulative = 0
for variant, ratio in config.traffic_split.items():
cumulative += ratio
if normalized < cumulative:
return variant
return list(config.variants.keys())[-1]
def get_variant_config(self, test_name: str, variant: str) -> dict:
"""获取变体配置"""
return self.active_tests[test_name].variants[variant]
def log_result(self, test_name: str, variant: str, metrics: dict):
"""记录测试结果"""
self.results[test_name].append({
'variant': variant,
'timestamp': time.time(),
**metrics
})
def analyze(self, test_name: str) -> dict:
"""分析测试结果"""
results = self.results[test_name]
variant_data = {}
for r in results:
v = r['variant']
if v not in variant_data:
variant_data[v] = []
variant_data[v].append(r)
analysis = {}
for variant, data in variant_data.items():
analysis[variant] = {
'sample_size': len(data),
'avg_latency_ms': sum(d.get('latency_ms', 0) for d in data) / len(data),
'avg_cost': sum(d.get('cost', 0) for d in data) / len(data),
'avg_quality_score': sum(d.get('quality_score', 0) for d in data) / len(data),
'avg_user_rating': sum(d.get('user_rating', 0) for d in data) / len(data),
}
return analysis
# 使用示例：测试模型路由策略
ab = ABTestManager()
ab.create_test(ABTestConfig(
test_name="model_routing_v1",
variants={
'control': {'model': 'gpt-4o', 'description': '全部使用 GPT-4o'},
'treatment': {'model': 'routed', 'description': '智能路由分流'},
},
traffic_split={'control': 0.5, 'treatment': 0.5},
metrics=['latency_ms', 'cost', 'quality_score', 'user_rating']
))
# 模拟请求
for i in range(100):
user_id = f"user_{i}"
variant = ab.assign_variant("model_routing_v1", user_id)
config = ab.get_variant_config("model_routing_v1", variant)
# 模拟结果
if variant == 'control':
metrics = {'latency_ms': 2000, 'cost': 0.05, 'quality_score': 9.0, 'user_rating': 4.5}
else:
metrics = {'latency_ms': 1500, 'cost': 0.02, 'quality_score': 8.5, 'user_rating': 4.3}
ab.log_result("model_routing_v1", variant, metrics)
# 分析结果
analysis = ab.analyze("model_routing_v1")
for variant, stats in analysis.items():
print(f"\n{variant}:")
for metric, value in stats.items():
print(f" {metric}: {value}")
9.2 ROI 分析模型
def calculate_ai_roi(
monthly_llm_cost: float,
monthly_infra_cost: float,
monthly_dev_cost: float,
# 收益侧
monthly_revenue_increase: float,
monthly_cost_savings: float,
# 优化前数据
baseline_monthly_cost: float = None
) -> dict:
"""计算 AI 应用的 ROI"""
total_monthly_cost = monthly_llm_cost + monthly_infra_cost + monthly_dev_cost
total_monthly_benefit = monthly_revenue_increase + monthly_cost_savings
monthly_roi = (total_monthly_benefit - total_monthly_cost) / total_monthly_cost
annual_roi = monthly_roi # 年化（简化）
payback_months = total_monthly_cost / (total_monthly_benefit - total_monthly_cost) \
if total_monthly_benefit > total_monthly_cost else float('inf')
result = {
'monthly_cost': {
'llm': monthly_llm_cost,
'infrastructure': monthly_infra_cost,
'development': monthly_dev_cost,
'total': total_monthly_cost,
},
'monthly_benefit': {
'revenue_increase': monthly_revenue_increase,
'cost_savings': monthly_cost_savings,
'total': total_monthly_benefit,
},
'roi': {
'monthly_roi': f"{monthly_roi:.1%}",
'payback_months': round(payback_months, 1) if payback_months != float('inf') else "N/A",
}
}
if baseline_monthly_cost:
savings = baseline_monthly_cost - total_monthly_cost
result['optimization'] = {
'baseline_cost': baseline_monthly_cost,
'current_cost': total_monthly_cost,
'savings': savings,
'savings_percentage': f"{savings/baseline_monthly_cost:.1%}"
}
return result
# 示例分析
roi = calculate_ai_roi(
monthly_llm_cost=1500,
monthly_infra_cost=800,
monthly_dev_cost=5000, # 摊销
monthly_revenue_increase=15000,
monthly_cost_savings=8000,
baseline_monthly_cost=5000 # 优化前的成本
)
print(json.dumps(roi, indent=2, ensure_ascii=False))
10. 企业级成本监控平台搭建
10.1 数据模型设计
-- 成本追踪数据库 Schema
-- API 调用日志
CREATE TABLE api_calls (
id BIGSERIAL PRIMARY KEY,
timestamp TIMESTAMPTZ NOT NULL DEFAULT NOW(),
user_id VARCHAR(64),
team_id VARCHAR(64),
model VARCHAR(64) NOT NULL,
provider VARCHAR(32) NOT NULL,
input_tokens INT NOT NULL,
output_tokens INT NOT NULL,
cost_usd DECIMAL(10, 6) NOT NULL,
latency_ms INT,
status VARCHAR(16), -- success, error, timeout
endpoint VARCHAR(128),
cached BOOLEAN DEFAULT FALSE
);
-- 每日汇总
CREATE TABLE daily_cost_summary (
date DATE NOT NULL,
team_id VARCHAR(64),
model VARCHAR(64),
total_requests INT,
total_input_tokens BIGINT,
total_output_tokens BIGINT,
total_cost_usd DECIMAL(12, 4),
avg_latency_ms INT,
error_count INT,
cache_hit_rate DECIMAL(5, 4),
PRIMARY KEY (date, team_id, model)
);
-- 预算配置
CREATE TABLE budgets (
id SERIAL PRIMARY KEY,
team_id VARCHAR(64) UNIQUE,
monthly_limit_usd DECIMAL(10, 2),
alert_threshold_pct DECIMAL(5, 2) DEFAULT 80.0,
current_month_usage DECIMAL(10, 2) DEFAULT 0,
is_active BOOLEAN DEFAULT TRUE
);
-- 创建索引
CREATE INDEX idx_api_calls_timestamp ON api_calls(timestamp);
CREATE INDEX idx_api_calls_team ON api_calls(team_id, timestamp);
CREATE INDEX idx_api_calls_model ON api_calls(model, timestamp);
10.2 监控服务实现
from fastapi import FastAPI, HTTPException
from datetime import datetime, timedelta
import asyncpg
app = FastAPI(title="AI Cost Monitor")
class CostMonitor:
"""AI 成本监控服务"""
def __init__(self, db_pool):
self.db = db_pool
async def record_api_call(self, call_data: dict):
"""记录一次 API 调用"""
async with self.db.acquire() as conn:
await conn.execute("""
INSERT INTO api_calls
(user_id, team_id, model, provider, input_tokens, output_tokens,
cost_usd, latency_ms, status, endpoint, cached)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
""",
call_data['user_id'], call_data['team_id'],
call_data['model'], call_data['provider'],
call_data['input_tokens'], call_data['output_tokens'],
call_data['cost_usd'], call_data['latency_ms'],
call_data['status'], call_data['endpoint'],
call_data.get('cached', False)
)
# 检查预算告警
await self._check_budget_alert(call_data['team_id'])
async def get_team_costs(self, team_id: str, days: int = 30) -> dict:
"""获取团队成本统计"""
since = datetime.utcnow() - timedelta(days=days)
async with self.db.acquire() as conn:
rows = await conn.fetch("""
SELECT
DATE(timestamp) as date,
model,
COUNT(*) as requests,
SUM(input_tokens) as input_tokens,
SUM(output_tokens) as output_tokens,
SUM(cost_usd) as cost,
AVG(latency_ms) as avg_latency,
SUM(CASE WHEN status = 'error' THEN 1 ELSE 0 END) as errors
FROM api_calls
WHERE team_id = $1 AND timestamp >= $2
GROUP BY DATE(timestamp), model
ORDER BY date DESC, cost DESC
""", team_id, since)
# 按模型汇总
by_model = {}
daily = {}
for row in rows:
model = row['model']
date = str(row['date'])
if model not in by_model:
by_model[model] = {'requests': 0, 'cost': 0, 'input_tokens': 0, 'output_tokens': 0}
by_model[model]['requests'] += row['requests']
by_model[model]['cost'] += float(row['cost'])
by_model[model]['input_tokens'] += row['input_tokens']
by_model[model]['output_tokens'] += row['output_tokens']
if date not in daily:
daily[date] = 0
daily[date] += float(row['cost'])
total_cost = sum(v['cost'] for v in by_model.values())
return {
'team_id': team_id,
'period_days': days,
'total_cost_usd': round(total_cost, 2),
'daily_average': round(total_cost / days, 2),
'projected_monthly': round(total_cost / days * 30, 2),
'by_model': {k: {**v, 'cost': round(v['cost'], 2)} for k, v in by_model.items()},
'daily_trend': daily,
}
async def get_cost_leaderboard(self, days: int = 7) -> list:
"""获取成本排行榜"""
since = datetime.utcnow() - timedelta(days=days)
async with self.db.acquire() as conn:
rows = await conn.fetch("""
SELECT
team_id,
SUM(cost_usd) as total_cost,
COUNT(*) as total_requests,
SUM(input_tokens + output_tokens) as total_tokens
FROM api_calls
WHERE timestamp >= $1
GROUP BY team_id
ORDER BY total_cost DESC
LIMIT 20
""", since)
return [
{
'rank': i + 1,
'team_id': row['team_id'],
'total_cost': round(float(row['total_cost']), 2),
'total_requests': row['total_requests'],
'total_tokens': row['total_tokens'],
}
for i, row in enumerate(rows)
]
async def _check_budget_alert(self, team_id: str):
"""检查预算告警"""
async with self.db.acquire() as conn:
budget = await conn.fetchrow("""
SELECT monthly_limit_usd, alert_threshold_pct, current_month_usage
FROM budgets WHERE team_id = $1 AND is_active = TRUE
""", team_id)
if not budget:
return
usage_pct = budget['current_month_usage'] / budget['monthly_limit_usd'] * 100
if usage_pct >= budget['alert_threshold_pct']:
# 发送告警
await self._send_alert(team_id, usage_pct, budget['monthly_limit_usd'])
async def _send_alert(self, team_id: str, usage_pct: float, limit: float):
"""发送预算告警"""
print(f"⚠️ 预算告警: 团队 {team_id} 已使用 {usage_pct:.1f}% (限额 ${limit})")
# 实际中发送到 Slack/邮件/钉钉
# API 接口
@app.get("/api/costs/{team_id}")
async def get_costs(team_id: str, days: int = 30):
monitor = CostMonitor(db_pool)
return await monitor.get_team_costs(team_id, days)
@app.get("/api/leaderboard")
async def get_leaderboard(days: int = 7):
monitor = CostMonitor(db_pool)
return await monitor.get_cost_leaderboard(days)
@app.post("/api/record")
async def record_call(call_data: dict):
monitor = CostMonitor(db_pool)
await monitor.record_api_call(call_data)
return {"status": "ok"}
10.3 Grafana 可视化看板
# grafana-dashboard.yaml (简化示例)
dashboard:
title: "AI 成本监控看板"
panels:
- title: "每日成本趋势"
type: timeseries
query: |
SELECT date, SUM(total_cost_usd)
FROM daily_cost_summary
WHERE date >= NOW() - INTERVAL '30 days'
GROUP BY date ORDER BY date
- title: "模型成本占比"
type: piechart
query: |
SELECT model, SUM(cost_usd) as cost
FROM api_calls
WHERE timestamp >= DATE_TRUNC('month', NOW())
GROUP BY model
- title: "团队成本排行"
type: table
query: |
SELECT team_id, SUM(cost_usd) as cost, COUNT(*) as requests
FROM api_calls
WHERE timestamp >= NOW() - INTERVAL '7 days'
GROUP BY team_id ORDER BY cost DESC LIMIT 10
- title: "缓存命中率"
type: gauge
query: |
SELECT
SUM(CASE WHEN cached THEN 1 ELSE 0 END)::float / COUNT(*) * 100
FROM api_calls
WHERE timestamp >= NOW() - INTERVAL '1 day'
- title: "平均响应延迟"
type: timeseries
query: |
SELECT
DATE_TRUNC('hour', timestamp) as time,
model,
AVG(latency_ms) as avg_latency
FROM api_calls
WHERE timestamp >= NOW() - INTERVAL '24 hours'
GROUP BY time, model
10.4 告警规则配置
# 告警规则引擎
ALERT_RULES = [
{
'name': '日成本超限',
'condition': lambda stats: stats['daily_cost'] > stats['daily_budget'] * 1.2,
'severity': 'critical',
'message': '今日 LLM 成本已超过预算的 120%',
},
{
'name': '异常调用量',
'condition': lambda stats: stats['hourly_requests'] > stats['avg_hourly'] * 3,
'severity': 'warning',
'message': '当前小时调用量超过平均值的 3 倍',
},
{
'name': '错误率飙升',
'condition': lambda stats: stats['error_rate'] > 0.1,
'severity': 'critical',
'message': f"API 错误率超过 10%",
},
{
'name': '缓存命中率下降',
'condition': lambda stats: stats['cache_hit_rate'] < 0.3,
'severity': 'warning',
'message': '缓存命中率低于 30%，可能需要调整缓存策略',
},
{
'name': 'Token 浪费',
'condition': lambda stats: stats['avg_output_tokens'] > 2000,
'severity': 'info',
'message': '平均输出 Token 数过高，建议优化 Prompt 或设置 max_tokens',
},
]
async def check_alerts(stats: dict):
"""检查所有告警规则"""
triggered = []
for rule in ALERT_RULES:
try:
if rule['condition'](stats):
triggered.append({
'name': rule['name'],
'severity': rule['severity'],
'message': rule['message'],
'timestamp': datetime.utcnow().isoformat(),
})
except Exception as e:
print(f"告警规则 {rule['name']} 执行失败: {e}")
return triggered
11. 最佳实践总结
11.1 成本优化检查清单
- Prompt 优化：系统 Prompt 控制在 200 tokens 以内
- 模型路由：简单任务用小模型，复杂任务用大模型
- 语义缓存：相似问题直接返回缓存结果
- 批量处理：非实时任务使用 Batch API
- Token 限制：合理设置 max_tokens，避免输出过长
- 量化部署：自部署模型使用 INT4/INT8 量化
- 监控告警：建立完整的成本监控和告警体系
- 定期审计：每周审查成本报告，找出优化空间
11.2 不同阶段的优化策略
| 阶段 | 日请求量 | 重点策略 | 预期节省 |
|---|---|---|---|
| 初创期 | < 1,000 | Prompt 优化 + 模型选择 | 30-50% |
| 成长期 | 1K-100K | + 缓存 + 路由 | 50-70% |
| 成熟期 | > 100K | + 量化部署 + 批处理 | 60-80% |
11.3 常见误区
| 误区 | 真相 |
|---|---|
| "用最便宜的模型就行" | 模型选择要看任务匹配度，小模型错误率高会导致重试成本 |
| "缓存万能" | 语义缓存有相似度阈值，低阈值会导致错误命中 |
| "量化没有损失" | INT4 量化在复杂推理任务上有明显质量下降 |
| "成本只看 API 费用" | 人力成本、基础设施成本、重试浪费都是大头 |
| "优化一次就够了" | 模型定价、用户行为、业务量都在变化，需要持续优化 |
11.4 优化路线图
Phase 1（第 1-2 周）：基础优化
├── 精简系统 Prompt
├── 设置合理的 max_tokens
├── 建立基础监控
└── 预期节省：30%
Phase 2（第 3-4 周）：智能路由
├── 实现大小模型分流
├── 部署语义缓存
├── A/B 测试验证效果
└── 预期节省：50%
Phase 3（第 5-8 周）：深度优化
├── Prompt 压缩技术
├── 批处理非实时任务
├── 模型量化部署
└── 预期节省：65%
Phase 4（持续）：监控与迭代
├── 完善监控看板
├── 定期成本审计
├── 跟踪新模型定价
└── 持续优化
参考资源
- OpenAI Tokenizer
- LLMLingua - Prompt 压缩框架
- LangChain - LLM 应用框架
- bitsandbytes - 模型量化
- vLLM - 高性能推理引擎
- Helicone - LLM 可观测平台
