推理成本核算与 FinOps

"""cost_model.py 推理成本核算与自建盈亏平衡，Python 3.12"""

from dataclasses import dataclass


@dataclass(frozen=True)
class ApiPricing:
    """按每百万 token 计价"""
    name: str
    input_per_mtok: float
    output_per_mtok: float

    def cost(self, in_tokens: int, out_tokens: int) -> float:
        return (in_tokens / 1_000_000) * self.input_per_mtok + \
               (out_tokens / 1_000_000) * self.output_per_mtok


@dataclass(frozen=True)
class GpuPricing:
    """自建按小时计价"""
    name: str
    usd_per_hour: float
    tokens_per_second: float          # 实测吞吐，含批处理
    utilization: float = 1.0          # 实际利用率

    def effective_hourly(self) -> float:
        # 利用率越低，等效每小时成本越高
        return self.usd_per_hour / max(self.utilization, 1e-6)

    def cost_per_mtok(self) -> float:
        # 一小时能产出的 token 数
        tokens_per_hour = self.tokens_per_second * 3600
        return self.effective_hourly() / (tokens_per_hour / 1_000_000)


GPT4O = ApiPricing("gpt-4o", 2.50, 10.00)
GPT4O_MINI = ApiPricing("gpt-4o-mini", 0.15, 0.60)
LLAMA70B_API = ApiPricing("llama-3.1-70b-managed", 0.52, 0.75)


def breakeven_tokens(api: ApiPricing, gpu: GpuPricing,
                     in_out_ratio: float = 6.86) -> float:
    """返回每月 token 盈亏平衡点：超过此值自建更省

    in_out_ratio 为输入 token 与输出 token 的比例，默认 6.86 对应
    上文客服场景（输入 2400，输出 350）。
    """
    in_share = in_out_ratio / (1 + in_out_ratio)
    out_share = 1 - in_share
    api_cost_per_mtok = (in_share * api.input_per_mtok +
                         out_share * api.output_per_mtok)
    gpu_cost_per_mtok = gpu.cost_per_mtok()
    if gpu_cost_per_mtok >= api_cost_per_mtok:
        return float("inf")   # 该 GPU 在任何规模下都不划算
    # 固定成本（人力、运维、冗余容量）按每月 8000 美元估算
    fixed_monthly = 8000.0
    # 求 x 使 fixed + x * gpu < x * api
    return fixed_monthly / (api_cost_per_mtok - gpu_cost_per_mtok) * 1_000_000


if __name__ == "__main__":
    a100 = GpuPricing("a100-80g", 1.10, 1800, utilization=0.6)
    h100 = GpuPricing("h100-80g", 2.10, 3400, utilization=0.6)

    print(f"A100 有效每百万 token: {a100.cost_per_mtok():.3f} 美元")
    print(f"H100 有效每百万 token: {h100.cost_per_mtok():.3f} 美元")
    print(f"对比 GPT-4o 的混合单价: "
          f"{GPT4O.cost(2_400_000, 350_000):.3f} 美元每百万混合 token")
    print(f"对比 Llama 托管混合单价: "
          f"{LLAMA70B_API.cost(2_400_000, 350_000):.3f} 美元每百万混合 token")

    for api in (GPT4O, LLAMA70B_API):
        for gpu in (a100, h100):
            be = breakeven_tokens(api, gpu)
            print(f"{api.name} vs {gpu.name}: 月盈亏平衡 {be / 1e9:.1f} B token")

-- cost_daily.sql 按租户与模型聚合日成本（PostgreSQL 16.4）
CREATE TABLE cost_daily (
    day           date        NOT NULL,
    tenant_id     text        NOT NULL,
    application   text        NOT NULL,
    model         text        NOT NULL,
    input_tokens  bigint      NOT NULL,
    output_tokens bigint      NOT NULL,
    cache_hits    bigint      NOT NULL,
    requests      bigint      NOT NULL,
    cost_usd      numeric(14,4) NOT NULL,
    PRIMARY KEY (day, tenant_id, application, model)
);

INSERT INTO cost_daily
SELECT
    date_trunc('day', ts)::date                       AS day,
    tenant_id,
    application,
    model,
    sum(input_tokens)                                 AS input_tokens,
    sum(output_tokens)                                AS output_tokens,
    count(*) FILTER (WHERE cache_status = 'hit')      AS cache_hits,
    count(*)                                          AS requests,
    round(sum(
        input_tokens  / 1e6 * in_price +
        output_tokens / 1e6 * out_price
    ), 4)                                             AS cost_usd
FROM llm_request_log
WHERE ts >= now() - interval '1 day'
GROUP BY 1, 2, 3, 4
ON CONFLICT (day, tenant_id, application, model) DO UPDATE
    SET input_tokens = EXCLUDED.input_tokens,
        cost_usd     = EXCLUDED.cost_usd;

"""cache_savings.py 缓存收益核算，Python 3.12"""

from dataclasses import dataclass


@dataclass
class Workload:
    requests_per_month: int
    input_tokens: int              # 平均输入 token
    output_tokens: int             # 平均输出 token
    prefix_tokens: int             # 可缓存的固定前缀长度
    in_price: float                # 每百万输入 token 价格
    out_price: float               # 每百万输出 token 价格
    cached_discount: float = 0.10  # 缓存命中后输入侧的折扣系数


def monthly_cost(w: Workload, prefix_hit: float, sem_hit: float) -> dict:
    """prefix_hit 与 sem_hit 分别为前缀缓存、语义缓存的命中率"""
    total_in = w.requests_per_month * w.input_tokens
    total_out = w.requests_per_month * w.output_tokens

    # 语义缓存命中：整次请求都不需要调用模型
    effective_requests = w.requests_per_month * (1 - sem_hit)

    # 前缀缓存命中：仅前缀部分按折扣计价
    cached_prefix = effective_requests * w.prefix_tokens * prefix_hit
    full_price_prefix = effective_requests * w.prefix_tokens * (1 - prefix_hit)
    rest_in = effective_requests * (w.input_tokens - w.prefix_tokens)

    in_cost = (
        cached_prefix * w.cached_discount +
        full_price_prefix +
        rest_in
    ) / 1_000_000 * w.in_price
    out_cost = effective_requests * w.output_tokens / 1_000_000 * w.out_price
    return {
        "input_usd": round(in_cost, 2),
        "output_usd": round(out_cost, 2),
        "total_usd": round(in_cost + out_cost, 2),
    }


if __name__ == "__main__":
    w = Workload(
        requests_per_month=21_600_000,
        input_tokens=2400,
        output_tokens=350,
        prefix_tokens=1500,
        in_price=2.50,
        out_price=10.00,
    )
    baseline = monthly_cost(w, 0.0, 0.0)["total_usd"]
    for prefix_hit in (0.0, 0.4, 0.7, 0.9):
        for sem_hit in (0.0, 0.2, 0.4):
            cost = monthly_cost(w, prefix_hit, sem_hit)["total_usd"]
            print(f"前缀命中 {prefix_hit:.0%} 语义命中 {sem_hit:.0%} "
                  f"月成本 {cost:,.0f} 美元 降幅 {(1 - cost / baseline):.1%}")