AI 应用的成本控制与优化

从一个月烧 $500 到 $50——手把手教你把 AI 应用的账单砍下来，同时不牺牲用户体验。

8. 监控与可观测性：看见你的钱花在哪

前面 7 章讲了怎么省钱。但有一个问题：你怎么知道省了多少？你怎么知道哪里还能省？

答案是监控。没有监控的成本优化就是盲人摸象——你不知道钱花在哪，就不知道该往哪砍。

8.1 如果你不监控，你就是在盲飞

你需要回答的 5 个问题

每个 AI 应用负责人都应该能回答：

  1. 今天花了多少钱？
     → 如果你不知道，你可能明天就超预算

  2. 哪个功能最烧钱？
     → 知识库问答？文档总结？代码生成？
     → 找到"吞金兽"，优先优化

  3. 哪些用户消耗最多？
     → 20% 的用户可能消耗了 80% 的 Token
     → 识别重度用户和异常用户

  4. 成本趋势是什么方向？
     → 在涨还是在跌？涨速多快？
     → 提前预警，而不是月底看账单

  5. 优化措施有没有效果？
     → 上了缓存后命中率多少？
     → 换了模型后质量有没有下降？

8.2 实战：构建 Token 用量监控仪表板

核心数据模型

python

from dataclasses import dataclass, asdict
from datetime import datetime
import json

@dataclass
class LLMUsageLog:
    """每次 LLM 调用的用量日志"""
    timestamp: str
    user_id: str
    feature: str          # 功能名称：chat / summary / code
    model: str            # gpt-4o / gpt-4o-mini
    input_tokens: int
    output_tokens: int
    total_tokens: int
    cost: float           # 美元
    cached: bool          # 是否缓存命中
    latency_ms: int       # 响应时间
    question_preview: str  # 问题前 50 字（脱敏）

    def to_dict(self) -> dict:
        return asdict(self)

用量记录器

python

class UsageTracker:
    """Token 用量追踪器"""

    def __init__(self, redis_client):
        self.redis = redis_client

    def log(self, entry: LLMUsageLog):
        """记录一次 API 调用"""
        data = json.dumps(entry.to_dict(), ensure_ascii=False)

        pipe = self.redis.pipeline()
        # 写入时间序列
        pipe.lpush("usage:logs", data)
        pipe.ltrim("usage:logs", 0, 99999)  # 保留最近 10 万条

        # 累加今日统计
        today = datetime.now().strftime("%Y%m%d")
        pipe.incrbyfloat(f"usage:daily:{today}:cost", entry.cost)
        pipe.hincrby(f"usage:daily:{today}:tokens", "input", entry.input_tokens)
        pipe.hincrby(f"usage:daily:{today}:tokens", "output", entry.output_tokens)
        pipe.hincrby(f"usage:daily:{today}:requests", "total", 1)
        if entry.cached:
            pipe.hincrby(f"usage:daily:{today}:requests", "cached", 1)

        # 按功能累加
        pipe.incrbyfloat(f"usage:feature:{today}:{entry.feature}", entry.cost)

        # 按用户累加
        pipe.incrbyfloat(f"usage:user:{today}:{entry.user_id}", entry.cost)

        # 按模型累加
        pipe.hincrby(f"usage:model:{today}", entry.model, entry.total_tokens)

        pipe.execute()

    def get_dashboard(self, date: str = None) -> dict:
        """获取仪表板数据"""
        if not date:
            date = datetime.now().strftime("%Y%m%d")

        cost = float(self.redis.get(f"usage:daily:{date}:cost") or 0)
        tokens = self.redis.hgetall(f"usage:daily:{date}:tokens")
        requests = self.redis.hgetall(f"usage:daily:{date}:requests")

        total_req = int(requests.get("total", 0))
        cached_req = int(requests.get("cached", 0))

        return {
            "date": date,
            "total_cost": f"${cost:.4f}",
            "input_tokens": int(tokens.get("input", 0)),
            "output_tokens": int(tokens.get("output", 0)),
            "total_requests": total_req,
            "cached_requests": cached_req,
            "cache_hit_rate": f"{cached_req/total_req*100:.1f}%" if total_req > 0 else "N/A",
            "avg_cost_per_request": f"${cost/total_req:.6f}" if total_req > 0 else "N/A",
        }

# 使用
tracker = UsageTracker(redis_client)
dashboard = tracker.get_dashboard()
print(dashboard)
# {
#   'total_cost': '$4.2300',
#   'cache_hit_rate': '38.5%',
#   'avg_cost_per_request': '$0.001230',
#   ...
# }

API 调用的自动记录

python

import time

def tracked_chat(
    question: str,
    user_id: str,
    feature: str = "chat",
    model: str = "gpt-4o-mini",
    system_prompt: str = "技术助手，简洁回答。"
) -> dict:
    """带监控追踪的 LLM 调用"""

    start = time.time()

    # 先查缓存
    cached_result = cache.get(question)
    if cached_result:
        tracker.log(LLMUsageLog(
            timestamp=datetime.now().isoformat(),
            user_id=user_id, feature=feature, model=model,
            input_tokens=0, output_tokens=0, total_tokens=0,
            cost=0, cached=True,
            latency_ms=int((time.time()-start)*1000),
            question_preview=question[:50]
        ))
        return cached_result

    # 调用 API
    response = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": question}
        ],
        max_tokens=500
    )

    usage = response.usage
    cost = (usage.prompt_tokens * 0.15 + usage.completion_tokens * 0.60) / 1_000_000

    # 记录用量
    tracker.log(LLMUsageLog(
        timestamp=datetime.now().isoformat(),
        user_id=user_id, feature=feature, model=model,
        input_tokens=usage.prompt_tokens,
        output_tokens=usage.completion_tokens,
        total_tokens=usage.total_tokens,
        cost=cost, cached=False,
        latency_ms=int((time.time()-start)*1000),
        question_preview=question[:50]
    ))

    result = {"answer": response.choices[0].message.content}
    cache.set(question, result)
    return result

8.3 成本归因：按功能、按用户、按时段

多维成本归因

python

class CostAttribution:
    """多维度成本归因分析"""

    def __init__(self, redis_client):
        self.redis = redis_client

    def by_feature(self, date: str = None) -> dict:
        """按功能归因"""
        if not date:
            date = datetime.now().strftime("%Y%m%d")

        features = {}
        for key in self.redis.scan_iter(f"usage:feature:{date}:*"):
            feature_name = key.split(":")[-1]
            cost = float(self.redis.get(key) or 0)
            features[feature_name] = f"${cost:.4f}"

        return dict(sorted(features.items(), key=lambda x: float(x[1][1:]), reverse=True))

    def by_user(self, date: str = None, top_n: int = 10) -> list:
        """按用户归因（Top N）"""
        if not date:
            date = datetime.now().strftime("%Y%m%d")

        users = []
        for key in self.redis.scan_iter(f"usage:user:{date}:*"):
            user_id = key.split(":")[-1]
            cost = float(self.redis.get(key) or 0)
            users.append({"user_id": user_id, "cost": f"${cost:.4f}"})

        return sorted(users, key=lambda x: float(x["cost"][1:]), reverse=True)[:top_n]

    def trend(self, days: int = 7) -> list:
        """成本趋势（近 N 天）"""
        trend = []
        for i in range(days - 1, -1, -1):
            date = (datetime.now() - timedelta(days=i)).strftime("%Y%m%d")
            cost = float(self.redis.get(f"usage:daily:{date}:cost") or 0)
            trend.append({"date": date, "cost": f"${cost:.4f}"})
        return trend

# 使用
attribution = CostAttribution(redis_client)

print("按功能：", attribution.by_feature())
# {'chat': '$2.8500', 'summary': '$0.9200', 'code': '$0.4600'}
# → chat 功能最烧钱，优先优化

print("按用户：", attribution.by_user(top_n=3))
# [{'user_id': 'u_789', 'cost': '$1.2000'},
#  {'user_id': 'u_123', 'cost': '$0.5600'}, ...]
# → u_789 消耗最多，检查是否正常使用

print("趋势：", attribution.trend(days=7))
# → 成本在涨还是在跌？

关键指标看板：每天早上花 30 秒看一眼：总成本、缓存命中率、Top 3 消耗用户。这 3 个数字就能让你对成本状况了如指掌。

8.4 ROI 评估：这个 AI 功能值不值得保留

不是每个 AI 功能都值得保留。有些功能成本高但使用率低，有些功能看似酷炫但对业务没有帮助。ROI 评估帮你做数据驱动的决策。

ROI 计算框架

AI 功能 ROI = (收益 - 成本) / 成本

  收益怎么算？
  ─────────────────────────────────────
  直接收益：
    → 付费用户因此功能而订阅？
    → 省了多少人工客服成本？
    → 提高了多少转化率？

  间接收益：
    → 用户满意度提升？
    → 减少了多少工单？
    → 用户留存率变化？

  成本怎么算？
  ─────────────────────────────────────
  直接成本：
    → LLM API 费用
    → 向量数据库费用
    → 基础设施费用

  间接成本：
    → 开发维护人力
    → 运维监控时间

功能 ROI 评估实战

python

def evaluate_feature_roi(
    feature_name: str,
    monthly_cost: float,
    monthly_revenue: float = 0,
    human_hours_saved: float = 0,
    hourly_rate: float = 30,
    monthly_users: int = 0,
) -> dict:
    """评估一个 AI 功能的 ROI"""

    # 收益计算
    direct_revenue = monthly_revenue
    labor_savings = human_hours_saved * hourly_rate
    total_benefit = direct_revenue + labor_savings

    # ROI
    roi = (total_benefit - monthly_cost) / monthly_cost if monthly_cost > 0 else 0
    cost_per_user = monthly_cost / monthly_users if monthly_users > 0 else 0

    # 决策建议
    if roi > 2:
        decision = "🟢 高价值，继续投入"
    elif roi > 0.5:
        decision = "🟡 还行，可以优化成本后保留"
    elif roi > 0:
        decision = "🟠 勉强盈利，考虑优化或降级"
    else:
        decision = "🔴 亏损，考虑下线或大幅重构"

    return {
        "feature": feature_name,
        "monthly_cost": f"${monthly_cost:.2f}",
        "monthly_benefit": f"${total_benefit:.2f}",
        "roi": f"{roi:.1%}",
        "cost_per_user": f"${cost_per_user:.4f}",
        "decision": decision
    }

# 评估案例
print(evaluate_feature_roi(
    feature_name="知识库问答",
    monthly_cost=80,         # LLM + 向量数据库
    monthly_revenue=200,     # 因此功能带来的订阅收入
    human_hours_saved=40,    # 省了 40 小时人工答疑
    monthly_users=500
))
# {'roi': '400.0%', 'decision': '🟢 高价值，继续投入'}

print(evaluate_feature_roi(
    feature_name="AI 写诗",
    monthly_cost=50,
    monthly_revenue=0,
    human_hours_saved=0,
    monthly_users=12
))
# {'roi': '-100.0%', 'decision': '🔴 亏损，考虑下线'}

三种常见决策场景

场景 1：高价值功能（ROI > 200%）
─────────────────────────────────────
  例：智能客服每月省 $1200 人工，AI 成本 $100
  决策：继续投入，甚至可以增加预算提升质量

场景 2：鸡肋功能（ROI 0-50%）
─────────────────────────────────────
  例：文档自动总结，月成本 $60，用户 30 人
  决策：
    → 先尝试成本优化（换模型、加缓存）
    → 成本降到 $20 以下再保留
    → 否则考虑下线

场景 3：面子功能（ROI < 0）
─────────────────────────────────────
  例：AI 生成周报，月成本 $40，只有 5 人用
  决策：果断下线，$40 不多但没价值的钱一分也不该花

数据说话：每季度做一次 AI 功能的 ROI 盘点。把钱集中在真正有价值的功能上，砍掉"看起来酷但没人用"的功能。

本章小结

知识点	要点
监控必要性	不监控 = 盲飞，每天看 3 个数字就够
数据模型	每次调用记录 12 个字段（含成本、缓存、延迟）
用量追踪	Redis 多维度累加（按天/功能/用户/模型）
自动记录	封装 tracked_chat，所有调用自动埋点
成本归因	按功能找"吞金兽"，按用户找异常
趋势分析	7 天趋势判断成本走向
ROI 评估	ROI > 200% 继续投入，< 0% 考虑下线
季度盘点	每季度评估所有 AI 功能的 ROI

下一章预告：真实案例——从 $500 到 $50 的完整优化全流程。我们会把前 8 章的所有技术串联起来，用一个完整的 RAG 问答系统作为案例，展示每一步优化的前后对比数据。

AI 应用的成本控制与优化 ​

8. 监控与可观测性：看见你的钱花在哪 ​

8.1 如果你不监控，你就是在盲飞 ​

你需要回答的 5 个问题 ​

8.2 实战：构建 Token 用量监控仪表板 ​

核心数据模型 ​

用量记录器 ​

API 调用的自动记录 ​

8.3 成本归因：按功能、按用户、按时段 ​

多维成本归因 ​

8.4 ROI 评估：这个 AI 功能值不值得保留 ​

ROI 计算框架 ​

功能 ROI 评估实战 ​

三种常见决策场景 ​

本章小结 ​

AI 应用的成本控制与优化

8. 监控与可观测性：看见你的钱花在哪

8.1 如果你不监控，你就是在盲飞

你需要回答的 5 个问题

8.2 实战：构建 Token 用量监控仪表板

核心数据模型

用量记录器

API 调用的自动记录

8.3 成本归因：按功能、按用户、按时段

多维成本归因

8.4 ROI 评估：这个 AI 功能值不值得保留

ROI 计算框架

功能 ROI 评估实战

三种常见决策场景

本章小结