AI Day 59
AI Day 59: 实战(9)：性能调优与成本实战 — 从能跑到跑得好
AI Day 59: 实战(9)：性能调优与成本实战 — 从能跑到跑得好
2026-05-30
日期: 2026-05-30 | 阶段: 第五阶段 · 动手实战 (Day 51-60) | 主题: Performance Tuning & Cost Optimization in Practice
学习路径 / Learning Path
AI/LLM 深度技术学习 60天计划
├── 第一阶段：模型基础 (Day 1-15) ✅
│ ├── Day 1: Transformer与LLM基础 ✅
│ ├── Day 2: 量化与本地部署 ✅
│ ├── Day 3: 训练全流程 ✅
│ ├── Day 4: Prompt Engineering ✅
│ ├── Day 5: RAG架构 ✅
│ ├── Day 6: 向量数据库与Embedding ✅
│ ├── Day 7: 微调技术 ✅
│ ├── Day 8: 推理优化 ✅
│ ├── Day 9: 长上下文技术 ✅
│ ├── Day 10: 多模态模型 ✅
│ ├── Day 11: 推理模型 ✅
│ ├── Day 12: Agent框架 ✅
│ ├── Day 13: MCP协议 ✅
│ ├── Day 14: 模型评估 ✅
│ └── Day 15: 阶段一总结 ✅
├── 第二阶段：工程实践 (Day 16-30) ✅
│ ├── Day 16: LLM应用架构 ✅
│ ├── Day 17: 安全与护栏 ✅
│ ├── Day 18: 可观测性 ✅
│ ├── Day 19: 生产RAG·解析与分块 ✅
│ ├── Day 20: 生产RAG·检索与重排 ✅
│ ├── Day 21: 生产RAG·评估与迭代 ✅
│ ├── Day 22: Agent状态与恢复 ✅
│ ├── Day 23: Agent成本优化 ✅
│ ├── Day 24: 多Agent系统 ✅
│ ├── Day 25: Agent测试部署 ✅
│ ├── Day 26: LLM成本工程 ✅
│ ├── Day 27: 多模型编排 ✅
│ ├── Day 28: LLM应用测试 ✅
│ ├── Day 29: 企业LLM平台 ✅
│ └── Day 30: 阶段二总结 ✅
├── 第三阶段：金融零售AI应用 (Day 31-42) ✅
│ ├── Day 31: 金融AI风控 ✅
│ ├── Day 32: 智能投顾与量化 ✅
│ ├── Day 33: 合规与RegTech ✅
│ ├── Day 34: 信贷AI全链路 ✅
│ ├── Day 35: 金融AI总结 ✅
│ ├── Day 36: 零售AI推荐 ✅
│ ├── Day 37: 智能客服 ✅
│ ├── Day 38: 供应链AI ✅
│ ├── Day 39: 智能营销 ✅
│ ├── Day 40: 零售AI总结 ✅
│ ├── Day 41: CeFi×DeFi×AI融合 ✅
│ └── Day 42: AI融合案例与职业 ✅
├── 第四阶段：面试冲刺 (Day 43-50) ✅
│ ├── Day 43: 系统设计·LLM平台 ✅
│ ├── Day 44: 系统设计·RAG系统 ✅
│ ├── Day 45: 系统设计·Agent系统 ✅
│ ├── Day 46: 系统设计·推荐系统 ✅
│ ├── Day 47: 面试·产品AI ✅
│ ├── Day 48: 面试·架构AI ✅
│ ├── Day 49: 面试·行为AI ✅
│ └── Day 50: 学习总结 ✅
└── 第五阶段：动手实战 (Day 51-60)
├── Day 51: 本地大模型部署全流程 ✅
├── Day 52: RAG系统实战：从文档到问答 ✅
├── Day 53: RAG进阶：评估优化与生产化 ✅
├── Day 54: LoRA微调实战：训练你的专属模型 ✅
├── Day 55: Agent开发实战：构建工具调用Agent ✅
├── Day 56: MCP Server开发：扩展AI能力边界 ✅
├── Day 57: 多模态应用：图文理解与文档分析 ✅
├── Day 58: AI应用全栈开发：前后端集成 ✅
├── Day 59: 性能调优与成本实战 ← 你在这里
└── Day 60: 总结与作品集
核心概念 / Core Concepts
性能和成本是产品能否上线的最后一关 / Performance & Cost: The Final Gate
Day 51-58 解决了"能不能做"：
✅ 模型能跑
✅ RAG能检索
✅ Agent能调工具
✅ 前后端能联通
Day 59 解决"能不能上线"：
❓ 每次请求要等多久？（用户体验）
❓ 能同时服务多少人？（并发能力）
❓ 每月运行费用多少？（商业可行性）
❓ 出了问题怎么发现？（监控告警）
真实案例：
某公司做了一个AI客服，Demo阶段效果很好
上线后：
平均响应 8 秒 → 用户放弃率 60%
每月 API 费用 $12,000 → 预算只有 $3,000
并发 5 人就开始超时 → 实际需要支撑 50 人
结果：下线重做，浪费3个月
Day 59 = 避免这个悲剧的最后防线
Day 8 理论(推理优化) + Day 26 理论(成本工程) → Day 59 全部实战验证
知识点1：延迟分析 / Latency Analysis
端到端延迟拆解 / End-to-End Latency Breakdown
一次 AI 请求的完整延迟路径：
用户 ──→ 前端 ──→ 后端API ──→ Ollama ──→ 生成 ──→ 返回
│ │ │ │ │ │
├ 网络 ├ 渲染 ├ 路由 ├ Prefill ├ Decode ├ 后处理
│ ~50ms │ ~10ms │ ~5ms │ ~200ms │ ~2-5s │ ~10ms
│ │ │ │ │ │
└────────┴────────┴──────────┴──────────┴────────┘
总延迟: 2-6 秒
各阶段详解：
1. 网络延迟 (Network) ~50ms
客户端 → 服务器的RTT
本地部署几乎为0，云端取决于距离
2. API路由 (Routing) ~5ms
FastAPI路由匹配、中间件处理
通常可以忽略
3. Prefill (预填充) ~200-500ms
处理输入 prompt 的所有 token
= 理解"你问了什么"
与输入长度成正比
长 prompt → 长 prefill
4. Decode (解码/生成) ~2-5s ← 最大瓶颈
逐个生成输出 token
200 tokens × 15ms/token = 3s
这是用户等待的主要部分
5. 后处理 (Post-processing) ~10ms
格式化、过滤、缓存写入
瓶颈定位方法 / Bottleneck Identification
"""
latency_profiler.py — 延迟分析工具
"""
import time
import ollama
import json
from dataclasses import dataclass, asdict
@dataclass
class LatencyProfile:
"""延迟剖析结果"""
total_ms: float
prefill_ms: float
decode_ms: float
tokens_generated: int
tokens_per_second: float
input_tokens: int
time_to_first_token_ms: float # TTFT
def profile_request(
prompt: str,
model: str = "qwen2.5:7b",
max_tokens: int = 200,
) -> LatencyProfile:
"""
精确剖析一次LLM请求的延迟分布
"""
start = time.perf_counter()
first_token_time = None
token_count = 0
stream = ollama.chat(
model=model,
messages=[{"role": "user", "content": prompt}],
stream=True,
options={"num_predict": max_tokens},
)
for chunk in stream:
if first_token_time is None and chunk["message"]["content"]:
first_token_time = time.perf_counter()
token_count += 1
end = time.perf_counter()
total_ms = (end - start) * 1000
ttft_ms = (first_token_time - start) * 1000 if first_token_time else total_ms
decode_ms = (end - first_token_time) * 1000 if first_token_time else 0
tps = token_count / (decode_ms / 1000) if decode_ms > 0 else 0
return LatencyProfile(
total_ms=round(total_ms, 1),
prefill_ms=round(ttft_ms, 1),
decode_ms=round(decode_ms, 1),
tokens_generated=token_count,
tokens_per_second=round(tps, 1),
input_tokens=len(prompt.split()), # 粗略估计
time_to_first_token_ms=round(ttft_ms, 1),
)
def run_latency_benchmark():
"""运行不同场景的延迟基准测试"""
scenarios = [
{"name": "短问题", "prompt": "Hello", "max_tokens": 50},
{"name": "中等问题", "prompt": "请解释区块链的工作原理", "max_tokens": 200},
{"name": "长问题", "prompt": "请详细分析2024年DeFi市场的发展趋势，包括TVL变化、新协议、监管影响。" * 3, "max_tokens": 500},
{"name": "代码生成", "prompt": "用Python实现一个简单的HTTP服务器", "max_tokens": 300},
]
print("=" * 70)
print(f"{'场景':<12} {'总延迟':>8} {'TTFT':>8} {'Decode':>8} {'Token数':>8} {'TPS':>6}")
print("=" * 70)
for s in scenarios:
profile = profile_request(s["prompt"], max_tokens=s["max_tokens"])
print(
f"{s['name']:<12} "
f"{profile.total_ms:>7.0f}ms "
f"{profile.prefill_ms:>7.0f}ms "
f"{profile.decode_ms:>7.0f}ms "
f"{profile.tokens_generated:>7} "
f"{profile.tokens_per_second:>5.1f}"
)
print("=" * 70)
if __name__ == "__main__":
run_latency_benchmark()
# 预期输出：
# ====================================================================
# 场景 总延迟 TTFT Decode Token数 TPS
# ====================================================================
# 短问题 892ms 185ms 707ms 48 67.9
# 中等问题 3245ms 210ms 3035ms 198 65.2
# 长问题 8012ms 520ms 7492ms 487 65.0
# 代码生成 4567ms 198ms 4369ms 289 66.1
# ====================================================================
知识点2：吞吐量优化 / Throughput Optimization
Continuous Batching 配置 / Continuous Batching Setup
Ollama 默认是串行处理：
请求1 → 处理中... → 完成 → 请求2 → 处理中... → 完成
10个请求串行 = 10 × 3s = 30s 总等待
Continuous Batching（连续批处理）：
请求1 ─┐
请求2 ─┤──→ GPU 同时处理 ──→ 并行完成
请求3 ─┘
10个请求并行 ≈ 4-5s 总等待（取决于GPU显存）
Ollama 配置：
OLLAMA_NUM_PARALLEL=4 # 最大并行请求数
OLLAMA_MAX_LOADED_MODELS=2 # 最多加载的模型数
并发测试 / Concurrency Testing
"""
concurrency_test.py — 并发测试
"""
import asyncio
import aiohttp
import time
import statistics
async def single_request(session, url, payload):
"""发送单个请求并计时"""
start = time.perf_counter()
async with session.post(url, json=payload) as resp:
result = await resp.json()
elapsed = (time.perf_counter() - start) * 1000
return elapsed
async def concurrency_test(
url: str,
payload: dict,
concurrency: int,
total_requests: int,
):
"""并发压测"""
semaphore = asyncio.Semaphore(concurrency)
latencies = []
errors = 0
async def bounded_request(session):
nonlocal errors
async with semaphore:
try:
latency = await single_request(session, url, payload)
latencies.append(latency)
except Exception:
errors += 1
start = time.perf_counter()
async with aiohttp.ClientSession() as session:
tasks = [bounded_request(session) for _ in range(total_requests)]
await asyncio.gather(*tasks)
total_time = time.perf_counter() - start
# 统计
latencies.sort()
results = {
"concurrency": concurrency,
"total_requests": total_requests,
"total_time_s": round(total_time, 2),
"qps": round(total_requests / total_time, 2),
"errors": errors,
"latency_p50_ms": round(latencies[len(latencies)//2], 0),
"latency_p90_ms": round(latencies[int(len(latencies)*0.9)], 0),
"latency_p99_ms": round(latencies[int(len(latencies)*0.99)], 0),
"latency_avg_ms": round(statistics.mean(latencies), 0),
}
return results
async def run_qps_test():
"""QPS递增测试"""
url = "http://localhost:8000/chat/"
payload = {
"message": "What is DeFi?",
"model": "qwen2.5:7b",
"stream": False,
}
print("=" * 70)
print("QPS 递增测试 / QPS Ramp-Up Test")
print("=" * 70)
print(f"{'并发':>6} {'总请求':>8} {'QPS':>8} {'P50':>8} {'P90':>8} {'P99':>8} {'错误':>6}")
print("-" * 70)
for concurrency in [1, 2, 4, 8, 16]:
result = await concurrency_test(url, payload, concurrency, concurrency * 5)
print(
f"{result['concurrency']:>6} "
f"{result['total_requests']:>8} "
f"{result['qps']:>7.1f} "
f"{result['latency_p50_ms']:>7.0f}ms "
f"{result['latency_p90_ms']:>7.0f}ms "
f"{result['latency_p99_ms']:>7.0f}ms "
f"{result['errors']:>6}"
)
print("=" * 70)
# asyncio.run(run_qps_test())
# 预期输出:
# ======================================================================
# QPS 递增测试 / QPS Ramp-Up Test
# ======================================================================
# 并发 总请求 QPS P50 P90 P99 错误
# ----------------------------------------------------------------------
# 1 5 0.3 3102ms 3205ms 3205ms 0
# 2 10 0.6 3312ms 3580ms 3620ms 0
# 4 20 1.0 3890ms 4521ms 4800ms 0
# 8 40 1.5 5200ms 7100ms 8200ms 0
# 16 80 1.2 9800ms 14200ms 18500ms 3 ← 开始超时
# ======================================================================
QPS 提升策略 / QPS Improvement Strategies
提升策略（按效果排序）：
策略1: 缓存 (命中时: 延迟从3s→<10ms)
热门问题缓存 → 大部分请求不需要跑模型
详见知识点3
策略2: 更小的模型 (延迟降50%)
7B → 3B: 速度提升2倍，质量下降15%
策略: 简单问题用3B，复杂问题用7B
Day 27 学的"模型路由"在这里用上
策略3: 量化 (延迟降20-30%)
FP16 → INT4: 速度提升30%，质量下降5%
Day 2 学的量化知识在这里用上
策略4: 限制输出长度 (延迟可控)
max_tokens=200 而不是无限制
大部分场景不需要生成1000个token
策略5: GPU升级 (线性提升)
RTX 3060 → RTX 4090: 吞吐量提升3-4倍
但成本也提升3-4倍
投入产出比: 缓存 > 小模型路由 > 量化 > 限制长度 > GPU升级
知识点3：缓存实战 / Caching in Practice
三层缓存实现 / Three-Layer Cache Implementation
"""
cache_system.py — 三层缓存系统
Layer 1: 精确匹配 — 相同问题直接返回
Layer 2: 语义匹配 — 相似问题返回缓存
Layer 3: Prompt Cache — 相同前缀加速推理
"""
import hashlib
import json
import time
from typing import Optional
import chromadb
class ThreeLayerCache:
"""三层AI缓存系统"""
def __init__(self):
# Layer 1: 精确匹配（内存字典/Redis）
self.exact_cache: dict[str, dict] = {}
# Layer 2: 语义缓存（向量数据库）
self.semantic_client = chromadb.PersistentClient(path="./cache_db")
self.semantic_cache = self.semantic_client.get_or_create_collection(
name="semantic_cache",
metadata={"hnsw:space": "cosine"},
)
# 统计
self.stats = {
"total_requests": 0,
"exact_hits": 0,
"semantic_hits": 0,
"misses": 0,
}
def _hash_key(self, query: str, model: str) -> str:
"""生成精确匹配的key"""
raw = f"{model}:{query.strip().lower()}"
return hashlib.md5(raw.encode()).hexdigest()
def get(self, query: str, model: str = "qwen2.5:7b") -> Optional[dict]:
"""
查询缓存（三层依次查找）
Returns: 缓存结果或None
"""
self.stats["total_requests"] += 1
# === Layer 1: 精确匹配 ===
key = self._hash_key(query, model)
if key in self.exact_cache:
cached = self.exact_cache[key]
# 检查过期（默认1小时）
if time.time() - cached["timestamp"] < 3600:
self.stats["exact_hits"] += 1
return {**cached, "cache_layer": "exact"}
# === Layer 2: 语义匹配 ===
results = self.semantic_cache.query(
query_texts=[query],
n_results=1,
)
if results["distances"] and results["distances"][0]:
distance = results["distances"][0][0]
# 余弦相似度 > 0.92 认为是相似问题
if distance < 0.08: # 1 - 0.92 = 0.08
self.stats["semantic_hits"] += 1
cached_response = json.loads(results["metadatas"][0][0]["response"])
return {**cached_response, "cache_layer": "semantic"}
# === Layer 3: Prompt Cache ===
# Ollama 原生支持 KV Cache
# 相同前缀的请求会复用 prefill 结果
# 无需手动实现，但可以通过设计 prompt 模板来利用
self.stats["misses"] += 1
return None
def set(self, query: str, response: dict, model: str = "qwen2.5:7b"):
"""写入缓存"""
key = self._hash_key(query, model)
# Layer 1: 精确缓存
self.exact_cache[key] = {
**response,
"timestamp": time.time(),
}
# Layer 2: 语义缓存
self.semantic_cache.add(
ids=[key],
documents=[query],
metadatas=[{
"response": json.dumps(response, ensure_ascii=False),
"model": model,
"timestamp": str(time.time()),
}],
)
def get_stats(self) -> dict:
"""缓存统计"""
total = self.stats["total_requests"]
if total == 0:
return {**self.stats, "hit_rate": 0}
hits = self.stats["exact_hits"] + self.stats["semantic_hits"]
return {
**self.stats,
"hit_rate": round(hits / total * 100, 1),
"exact_hit_rate": round(self.stats["exact_hits"] / total * 100, 1),
"semantic_hit_rate": round(self.stats["semantic_hits"] / total * 100, 1),
}
# === 集成到API ===
cache = ThreeLayerCache()
async def cached_chat(message: str, model: str = "qwen2.5:7b") -> dict:
"""带缓存的Chat接口"""
# 查缓存
cached = cache.get(message, model)
if cached:
return {
**cached,
"from_cache": True,
"latency_ms": 1, # 缓存命中几乎无延迟
}
# 缓存未命中，调用模型
start = time.perf_counter()
import ollama
response = ollama.chat(
model=model,
messages=[{"role": "user", "content": message}],
)
latency = (time.perf_counter() - start) * 1000
result = {
"response": response["message"]["content"],
"model": model,
"latency_ms": round(latency, 0),
}
# 写入缓存
cache.set(message, result, model)
return {**result, "from_cache": False}
缓存命中率监控与ROI / Cache Hit Rate & ROI
缓存ROI计算：
假设每天1000次请求，无缓存：
每次请求延迟: ~3000ms
每次请求成本: ~$0.003 (本地GPU电费) 或 ~$0.01 (云API)
日成本: $3-10
加入三层缓存后（假设命中率60%）：
600次命中: ~5ms × 600 = 3s 总耗时, 接近$0成本
400次未命中: ~3000ms × 400 = 1200s 总耗时
延迟改善: 平均 3000ms → 1205ms (↓60%)
成本改善: $3-10/天 → $1.2-4/天 (↓60%)
缓存数据库成本: ~$0.01/天 (本地ChromaDB)
ROI: 缓存带来的收益远超投入
实际监控指标：
┌─────────────────────────────────────────┐
│ Cache Performance Dashboard │
├─────────────────────────────────────────┤
│ │
│ Total Requests: 1,247 │
│ Cache Hit Rate: 63.2% │
│ ├ Exact Hits: 45.1% │
│ └ Semantic: 18.1% │
│ │
│ Avg Latency: │
│ Cache Hit: 4ms │
│ Cache Miss: 3,124ms │
│ Overall: 1,154ms (↓62%) │
│ │
│ Est. Cost Saved: $6.21/day │
└─────────────────────────────────────────┘
知识点4：成本实战 / Cost Analysis in Practice
实际项目成本核算 / Real Project Cost Breakdown
场景: Day 58 构建的AI全栈应用，日活100人，每人每天10次请求
=== 方案A: 全部本地 (Ollama + 自有GPU) ===
硬件成本:
GPU: RTX 3060 12GB (已有) $0/月
电费: 200W × 24h × 30天 ~$15/月
网络: 家用宽带 (已有) $0/月
软件成本:
Ollama: 免费
ChromaDB: 免费
Next.js: 免费
总成本: ~$15/月
单次请求成本: $15 / (100×10×30) = $0.0005
限制: 无法24/7保证可用，单GPU并发有限
=== 方案B: 云端GPU实例 ===
基础设施:
AWS g5.xlarge (A10G GPU): $1.006/hr × 730h $734/月
或 按需开关: $1.006/hr × 12h × 30天 $362/月
存储 (100GB EBS): $10/月
网络: ~$5/月
软件: 全部免费/开源
总成本: $377-749/月
单次请求成本: $0.013-0.025
=== 方案C: 云API (OpenAI/Anthropic) ===
调用成本 (以 GPT-4o-mini 为例):
输入: 1000 requests × 500 tokens = 500K tokens/天
输出: 1000 requests × 200 tokens = 200K tokens/天
日成本: 500K × $0.15/1M + 200K × $0.60/1M = $0.195/天
月成本: $5.85
RAG Embedding:
月新增文档: 10万 tokens
$0.02/1M tokens ≈ $0.002/月
总成本: ~$6/月
单次请求成本: $0.0002
限制: 数据发送到第三方，延迟更高（网络RTT）
=== 方案D: 混合方案（推荐）===
热门/简单请求 → 本地 Ollama (70% 流量) $15/月
复杂/长文本请求 → 云API (30% 流量) $2/月
前端 → Vercel Free Tier $0/月
总成本: ~$17/月
最佳性价比！
优化前后成本对比表 / Before vs After Optimization
优化措施成本影响对比表:
措施 实施难度 延迟改善 成本改善 推荐度
──────────────────────────────────────────────────────────
三层缓存(知识点3) 中 ↓60% ↓60% ★★★★★
模型路由(小/大模型) 中 ↓30% ↓40% ★★★★★
INT4量化 低 ↓20% ↓15% ★★★★
限制max_tokens 低 ↓15% ↓20% ★★★★
Prompt优化(更短) 低 ↓10% ↓15% ★★★★
GPU升级 高 ↓50% ↑200% ★★
云API迁移 中 ↑50% ↓80% ★★★
Batch请求合并 高 ↓5% ↓30% ★★★
组合优化效果（全部实施）:
优化前: 平均延迟 3000ms, 月成本 $200
优化后: 平均延迟 800ms, 月成本 $40
改善: 延迟 ↓73%, 成本 ↓80%
最关键的3个优化（80/20法则）:
1. 缓存 — 大部分请求不需要跑模型
2. 模型路由 — 简单问题用小模型
3. Prompt优化 — 减少不必要的输入token
知识点5：压测与基准 / Load Testing & Benchmarking
Locust 压测配置 / Locust Load Testing
"""
locustfile.py — AI应用压测脚本
使用: locust -f locustfile.py --host=http://localhost:8000
"""
from locust import HttpUser, task, between, events
import json
import time
class AIAppUser(HttpUser):
"""模拟AI应用用户行为"""
# 请求间隔: 5-15秒（模拟真实用户思考时间）
wait_time = between(5, 15)
def on_start(self):
"""用户启动时"""
self.headers = {
"Content-Type": "application/json",
"X-API-Key": "demo-key-12345",
}
@task(5) # 权重5: 最常见的操作
def chat_simple(self):
"""简单对话"""
payload = {
"message": "什么是DeFi？",
"model": "qwen2.5:7b",
"stream": False,
}
with self.client.post(
"/chat/",
json=payload,
headers=self.headers,
catch_response=True,
name="/chat (simple)",
) as response:
if response.status_code == 200:
data = response.json()
if len(data.get("response", "")) < 10:
response.failure("Response too short")
else:
response.failure(f"Status {response.status_code}")
@task(3) # 权重3
def rag_query(self):
"""RAG问答"""
payload = {
"question": "如何设计空投方案？",
"n_results": 3,
}
self.client.post(
"/rag/query",
json=payload,
headers=self.headers,
name="/rag/query",
)
@task(1) # 权重1: 较少使用
def health_check(self):
"""健康检查"""
self.client.get("/health", name="/health")
# === 自定义统计输出 ===
@events.request.add_listener
def on_request(request_type, name, response_time, response_length, **kwargs):
"""记录每个请求的详细信息（可选）"""
pass # 可以写入文件或数据库
SLA 定义 / SLA Definition
AI应用的SLA标准（参考业界）:
指标 目标值 告警阈值 严重阈值
──────────────────────────────────────────────────────────
可用性 99.5% 99.0% 98.0%
P50 延迟 < 2s > 3s > 5s
P99 延迟 < 8s > 12s > 20s
TTFT < 500ms > 1s > 2s
错误率 < 1% > 2% > 5%
QPS (Chat) > 2 req/s < 1.5 < 1.0
缓存命中率 > 50% < 40% < 20%
不同场景的SLA差异:
内部工具:
可用性 99%, P50 < 5s, 错误率 < 5%
要求较低，但不能经常挂
B2C 产品:
可用性 99.9%, P50 < 2s, 错误率 < 1%
用户体验关键，超过3秒用户就流失
金融场景:
可用性 99.99%, P50 < 1s, 错误率 < 0.1%
合规要求高，错误可能导致资金损失
知识点6：监控仪表盘 / Monitoring Dashboard
Prometheus 指标收集 / Prometheus Metrics Collection
"""
backend/metrics.py — 监控指标收集
"""
from prometheus_client import (
Counter,
Histogram,
Gauge,
generate_latest,
CONTENT_TYPE_LATEST,
)
from fastapi import Response
import time
# === 定义指标 ===
# 请求计数
REQUEST_COUNT = Counter(
"ai_request_total",
"Total AI requests",
["endpoint", "model", "cache_hit"],
)
# 延迟直方图
REQUEST_LATENCY = Histogram(
"ai_request_duration_seconds",
"AI request latency in seconds",
["endpoint", "model"],
buckets=[0.1, 0.5, 1.0, 2.0, 3.0, 5.0, 8.0, 12.0, 20.0],
)
# Token使用量
TOKENS_USED = Counter(
"ai_tokens_total",
"Total tokens used",
["model", "direction"], # direction: input/output
)
# 活跃请求数
ACTIVE_REQUESTS = Gauge(
"ai_active_requests",
"Currently processing requests",
["endpoint"],
)
# 模型加载状态
MODEL_LOADED = Gauge(
"ai_model_loaded",
"Whether model is loaded",
["model"],
)
# 缓存命中率
CACHE_HIT_RATE = Gauge(
"ai_cache_hit_rate",
"Cache hit rate (percentage)",
)
# 估算成本
ESTIMATED_COST = Counter(
"ai_estimated_cost_usd",
"Estimated cost in USD",
["model"],
)
# === 使用示例 ===
def track_request(endpoint: str, model: str, cache_hit: bool, latency_s: float,
input_tokens: int, output_tokens: int):
"""记录一次请求的所有指标"""
REQUEST_COUNT.labels(
endpoint=endpoint,
model=model,
cache_hit=str(cache_hit),
).inc()
REQUEST_LATENCY.labels(
endpoint=endpoint,
model=model,
).observe(latency_s)
if not cache_hit:
TOKENS_USED.labels(model=model, direction="input").inc(input_tokens)
TOKENS_USED.labels(model=model, direction="output").inc(output_tokens)
# 估算成本（本地GPU电费）
cost_per_token = 0.000001 # $0.001 per 1K tokens
cost = (input_tokens + output_tokens) * cost_per_token
ESTIMATED_COST.labels(model=model).inc(cost)
# === Prometheus 端点 ===
async def metrics_endpoint():
"""Prometheus 指标端点"""
return Response(
content=generate_latest(),
media_type=CONTENT_TYPE_LATEST,
)
Grafana Dashboard 配置 / Grafana Dashboard
关键监控面板：
┌─────────────────────────────────────────────────────────────┐
│ AI Application Dashboard │
├─────────────────┬─────────────────┬─────────────────────────┤
│ QPS │ Error Rate │ Active Requests │
│ ████████ 2.3 │ ░░░░░░ 0.8% │ ██░░░░ 3/10 │
├─────────────────┴─────────────────┴─────────────────────────┤
│ │
│ Latency Distribution (P50 / P90 / P99) │
│ ┌──────────────────────────────────────────────┐ │
│ │ P50: 1.2s P90: 3.4s P99: 8.1s │ │
│ │ ■■■■■■■──────■■■──────────■ │ │
│ └──────────────────────────────────────────────┘ │
│ │
├─────────────────┬─────────────────┬─────────────────────────┤
│ Cache Hit Rate │ Token Usage │ Est. Daily Cost │
│ ████████ 63% │ In: 125K/hr │ $0.45 │
│ exact: 45% │ Out: 52K/hr │ Budget: $1.00 │
│ semantic: 18% │ │ ████████░░ │
├─────────────────┴─────────────────┴─────────────────────────┤
│ │
│ Request Latency Over Time (24h) │
│ 5s │ ╱╲ │
│ 4s │ ╱ ╲ ╱╲ │
│ 3s │───────╱────╲──╱──╲───────────────── │
│ 2s │ ╱╲ ╱ ╲╱ │
│ 1s │╱ ╲╱ │
│ 0s └────────────────────────────────────── │
│ 00:00 04:00 08:00 12:00 16:00 20:00 │
│ │
└─────────────────────────────────────────────────────────────┘
Grafana + Prometheus 配置:
prometheus.yml:
scrape_configs:
- job_name: 'ai-backend'
scrape_interval: 15s
static_configs:
- targets: ['backend:8000']
metrics_path: '/metrics'
告警规则:
- alert: HighLatency
expr: histogram_quantile(0.95, ai_request_duration_seconds) > 8
for: 5m
labels:
severity: warning
annotations:
summary: "AI请求P95延迟超过8秒"
- alert: HighErrorRate
expr: rate(ai_request_total{status="error"}[5m]) > 0.05
for: 2m
labels:
severity: critical
annotations:
summary: "AI请求错误率超过5%"
- alert: LowCacheHitRate
expr: ai_cache_hit_rate < 20
for: 15m
labels:
severity: warning
annotations:
summary: "缓存命中率低于20%，检查缓存策略"
今日思考 / Today's Reflections
思考1：性能优化的80/20法则 / The 80/20 Rule of Performance
今天测下来的结论：
3个优化覆盖了80%的收益：
1. 缓存 (命中率60% → 延迟↓60%, 成本↓60%)
2. 模型路由 (简单→小模型 → 延迟↓30%, 成本↓40%)
3. Prompt精简 (减少输入token → Prefill↓30%)
其余优化收益递减：
量化: ↓20%延迟（但已经在用INT4了）
GPU升级: 钱的问题
框架优化: 边际收益小
这给PM的启示：
不要追求"极致优化"
先把最有效的3个做了
剩下的看ROI再决定
就像产品功能：
20%的功能覆盖80%的用户需求
先做这20%，做好
思考2：缓存是AI应用的生命线 / Cache is AI's Lifeline
没有缓存的AI应用 = 没有CDN的网站
Day 26 学的缓存理论，今天实测验证：
相同问题不同用户问 → 完全不需要重新推理
相似问题 → 语义缓存命中，也不需要重新推理
只有真正新的问题 → 才需要跑模型
实际应用中，大量请求是重复的：
"什么是DeFi" "DeFi是什么" "defi的定义"
→ 这三个问题，只有第一个需要跑模型
→ 其他两个语义缓存命中
缓存的设计决策：
TTL多长？太短→命中率低，太长→答案过时
语义阈值多少？太低→误命中，太高→命中率低
我的经验值：
TTL=1小时（大多数知识不会1小时就变）
语义相似度阈值=0.92（实测最佳平衡点）
思考3：监控是AI应用的必需品而非可选项 / Monitoring is Non-Negotiable
传统应用监控：
知道请求是否成功就够了
HTTP 200 = 成功
HTTP 500 = 失败
AI应用监控更复杂：
HTTP 200 但答案是垃圾 = 质量问题
HTTP 200 但花了15秒 = 体验问题
HTTP 200 但花了$0.50 = 成本问题
HTTP 200 但缓存全部miss = 效率问题
必须监控的5个指标：
1. 延迟分布 (不是平均值，要看P50/P90/P99)
2. 缓存命中率 (低于40%就要调查)
3. Token消耗 (防止成本失控)
4. 错误率 (含超时和质量问题)
5. QPS (了解容量余量)
Day 18 学的可观测性三支柱(Logging/Tracing/Metrics)
在今天全部落地实现了
学习资源 / Resources
性能测试
监控
- Prometheus: https://prometheus.io/docs/
- Grafana: https://grafana.com/docs/
- prometheus_client (Python): https://github.com/prometheus/client_python
LLM 性能优化
- Ollama Performance Tips: https://github.com/ollama/ollama/blob/main/docs/gpu.md
- vLLM Benchmarking: https://docs.vllm.ai/en/latest/performance/benchmarks.html
- LLM Inference Optimization: https://lilianweng.github.io/posts/2023-01-10-inference-optimization/
成本优化
- LLM Cost Calculator: https://huggingface.co/spaces/philschmid/llm-pricing
- Cloud GPU Pricing: https://cloud-gpus.com/
- Token Counter: https://platform.openai.com/tokenizer
明日预告 / Tomorrow's Preview
Day 60: 60天总结 — AI深度学习完整复盘
明天是最后一天！
60天旅程即将画上句号：
Day 1: "Attention is All You Need" — 什么是Transformer？
Day 60: 构建了完整的AI应用并完成了性能调优
明天将回顾：
1. 60天全景回顾 — 每天一句话总结
2. 完整知识地图 — 从理论到实战
3. 实战产出统计 — 代码、笔记、系统
4. 能力自评 — 10个维度雷达图
5. 三大计划融合 — Web3 + 架构 + AI
6. Top 10 核心洞察
7. 下一步 — 持续学习和求职策略
准备工作：
回顾所有60篇笔记的标题
整理代码和产出清单
最后一天，画一个完美的句号！
Day 59 完成！ 从延迟分析、三层缓存、成本核算到压测和监控。 缓存命中率60%让平均延迟从3秒降到1.2秒，成本下降60%。 明天是最后一天，60天AI深度学习的完整复盘！