AI API 成本监控：Token 消耗仪表盘搭建教程

去年我在做一个法律合同审查项目时，一晚上跑下来账单直接超了 1200 元。复盘的时候我发现，根本没有任何工具告诉我"到底哪一段 prompt 烧了最多 token"。从那天起，我就决定自己搭一套 Token 消耗仪表盘，把所有调用成本可视化、可追溯、可优化。

在动手之前，先看一眼 2026 年主流模型在 HolySheep AI 上的真实报价（output 价格，/MTok，¥1=$1 无损结算）：

模型	官方 $价	官方 ¥价（汇率7.3）	HolySheep ¥价	每月 100 万 token 节省
GPT-4.1	$8.00	¥58.40	¥8.00	¥50.40（-86.3%）
Claude Sonnet 4.5	$15.00	¥109.50	¥15.00	¥94.50（-86.3%）
Gemini 2.5 Flash	$2.50	¥18.25	¥2.50	¥15.75（-86.3%）
DeepSeek V3.2	$0.42	¥3.07	¥0.42	¥2.65（-86.3%）

单看 100 万 token 差距不大，可一旦业务量上来——比如每天 500 万 token、连续跑 30 天——Claude Sonnet 4.5 用官方渠道一个月要 ¥3285，HolySheep 只要 ¥450，单模型一年就能省下 ¥3.4 万。这就是为什么必须在用中转站的同时，自己再叠一层精细化的成本监控：既要"省钱"，也要"省得明白"。

一、为什么需要自建 Token 仪表盘？

官方控制台粒度太粗：只能看到当日总量，无法按"用户/业务线/prompt 模板"拆解
中转站只显示余额：你不知道哪个调用最烧钱、可以优化
多模型混用时无总账：GPT-4.1 + Claude + DeepSeek 混着调，月底对账对到崩溃
异常调用无法告警：某个 prompt 死循环调用，单日多出 ¥300 你也收不到通知

二、整体架构

我用三段式架构，部署在自己 2C2G 的小服务器上跑得很稳：

埋点 SDK（Python）：包装 OpenAI 兼容客户端，自动记录每次调用的 model、token、延迟、花费
数据接收服务（FastAPI + SQLite）：把埋点数据落库，提供 REST 查询接口
可视化仪表盘（Streamlit + Plotly）：实时刷新花费趋势、模型占比、TOP 10 烧钱 prompt

三、核心代码实现

3.1 埋点 SDK：自动拦截所有 OpenAI 兼容调用

这个类可以直接 import 到你现有业务代码里，无需修改调用逻辑：

import time
import sqlite3
from openai import OpenAI

HolySheep 中转配置，国内直连 < 50ms
client = OpenAI(
    base_url="https://api.holysheep.ai/v1",
    api_key="YOUR_HOLYSHEEP_API_KEY",
)

2026 主流模型 output 价格（/MTok），HolySheep ¥1=$1 结算
PRICE_MAP = {
    "gpt-4.1": 8.00,
    "claude-sonnet-4.5": 15.00,
    "gemini-2.5-flash": 2.50,
    "deepseek-v3.2": 0.42,
}

class TokenTracker:
    def __init__(self, db_path="token_usage.db"):
        self.db_path = db_path
        self._init_db()

    def _init_db(self):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS usage (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    ts REAL, model TEXT, user_tag TEXT,
                    prompt_tokens INTEGER, completion_tokens INTEGER,
                    cost_cny REAL, latency_ms REAL
                )
            """)

    def record(self, model, user_tag, pt, ct, latency_ms):
        price = PRICE_MAP.get(model, 2.50)
        cost = (ct / 1_000_000) * price  # 直接用 ¥ 结算
        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                "INSERT INTO usage (ts, model, user_tag, prompt_tokens, completion_tokens, cost_cny, latency_ms) VALUES (?,?,?,?,?,?,?)",
                (time.time(), model, user_tag, pt, ct, round(cost, 6), round(latency_ms, 2))
            )
        return cost

tracker = TokenTracker()

def chat_tracked(model, messages, user_tag="default", **kwargs):
    start = time.perf_counter()
    resp = client.chat.completions.create(
        model=model, messages=messages, **kwargs
    )
    latency_ms = (time.perf_counter() - start) * 1000
    cost = tracker.record(
        model=model, user_tag=user_tag,
        pt=resp.usage.prompt_tokens,
        ct=resp.usage.completion_tokens,
        latency_ms=latency_ms,
    )
    return resp, cost

业务里直接这样用，完全无感
resp, cost = chat_tracked("gpt-4.1", [{"role":"user","content":"你好"}], user_tag="contract-review")
print(f"本次 ¥{cost:.4f}，延迟 {resp.usage.completion_tokens} tokens")

3.2 数据接收服务：FastAPI + SQLite

如果你的服务部署在多台机器上，建议把埋点改成 HTTP 上报，统一进一个数据库：

from fastapi import FastAPI
from pydantic import BaseModel
import sqlite3, time

app = FastAPI(title="HolySheep Token 监控后端")

class UsageEvent(BaseModel):
    model: str
    user_tag: str
    prompt_tokens: int
    completion_tokens: int
    latency_ms: float

PRICE_MAP = {
    "gpt-4.1": 8.00,
    "claude-sonnet-4.5": 15.00,
    "gemini-2.5-flash": 2.50,
    "deepseek-v3.2": 0.42,
}

def init_db():
    with sqlite3.connect("usage.db") as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS usage (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                ts REAL, model TEXT, user_tag TEXT,
                prompt_tokens INTEGER, completion_tokens INTEGER,
                cost_cny REAL, latency_ms REAL
            )
        """)
init_db()

@app.post("/log")
async def log_usage(ev: UsageEvent):
    price = PRICE_MAP.get(ev.model, 2.50)
    cost = (ev.completion_tokens / 1_000_000) * price
    with sqlite3.connect("usage.db") as conn:
        conn.execute(
            "INSERT INTO usage VALUES (NULL,?,?,?,?,?,?,?)",
            (time.time(), ev.model, ev.user_tag,
             ev.prompt_tokens, ev.completion_tokens,
             round(cost, 6), round(ev.latency_ms, 2))
        )
    return {"cost_cny": round(cost, 6), "model": ev.model}

@app.get("/stats")
async def stats(days: int = 7, user_tag: str | None = None):
    since = time.time() - days * 86400
    sql = "SELECT model, SUM(completion_tokens), SUM(cost_cny), AVG(latency_ms), COUNT(*) FROM usage WHERE ts>?"
    params: list = [since]
    if user_tag:
        sql += " AND user_tag=?"
        params.append(user_tag)
    sql += " GROUP BY model"
    with sqlite3.connect("usage.db") as conn:
        rows = conn.execute(sql, params).fetchall()
    return {"data": [
        {"model": r[0], "tokens": r[1], "cost_cny": round(r[2], 4),
         "avg_latency_ms": round(r[3], 2), "calls": r[4]} for r in rows
    ]}

跑起来：uvicorn monitor:app --host 0.0.0.0 --port 8000，国内直连 HolySheep 整体往返基本在 80~120ms。

3.3 Streamlit 可视化仪表盘

一行命令启动：streamlit run dashboard.py

import streamlit as st
import sqlite3, pandas as pd
import plotly.express as px

st.set_page_config(page_title="HolySheep Token 仪表盘", layout="wide")
st.title("🐑 HolySheep Token 成本监控仪表盘")

conn = sqlite3.connect("usage.db")
df = pd.read_sql_query("SELECT * FROM usage ORDER BY ts DESC LIMIT 5000", conn)
conn.close()

if df.empty:
    st.info("暂无调用数据，去跑两轮 API 吧～")
    st.stop()

c1, c2, c3, c4 = st.columns(4)
c1.metric("总花费（¥）", f"{df['cost_cny'].sum():.2f}")
c2.metric("总 Completion Token", f"{int(df['completion_tokens'].sum()):,}")
c3.metric("平均延迟（ms）", f"{df['latency_ms'].mean():.1f}")
c4.metric("调用次数", f"{len(df):,}")

st.subheader("各模型花费占比")
fig1 = px.pie(df.groupby("model", as_index=False)["cost_cny"].sum(),
              values="cost_cny", names="model", hole=0.45)
st.plotly_chart(fig1, use_container_width=True)

st.subheader("近 30 天每日成本趋势")
df["date"] = pd.to_datetime(df["ts"], unit="s").dt.date
daily = df.groupby("date", as_index=False)["cost_cny"].sum()
fig2 = px.line(daily, x="date", y="cost_cny", markers=True)
st.plotly_chart(fig2, use_container_width=True)

st.subheader("按业务线 user_tag 排行")
tag_df = df.groupby("user_tag", as_index=False).agg(
    cost=("cost_cny","sum"), tokens=("completion_tokens","sum")
).sort_values("cost", ascending=False).head(10)
st.dataframe(tag_df, use_container_width=True)

四、几个落地小建议

给每个业务线打 user_tag：合同审查、客服、代码生成分别打不同 tag，月底直接拉账单
流式响应也要埋点：stream=True 时 usage 在最后一个 chunk，需要用累加器手动相加
异常告警：单次调用花费 > ¥0.5 或延迟 > 5000ms 直接发企业微信
价格表集中维护：HolySheep 更新价格时只需改 PRICE_MAP 一处

常见报错排查

openai.OpenAIError: Connection error：直连官方 api.openai.com 在国内极不稳定，请把 base_url 切到 https://api.holysheep.ai/v1，国内延迟稳定在 50ms 以内
KeyError: 'gpt-4.1' 出现在 PRICE_MAP：新模型没有登记价格，脚本会 fallback 到 ¥2.50/MTok，导致账单偏低。补全价格即可
sqlite3.OperationalError: database is locked：高并发下多线程写 SQLite 互锁。两种方案：①改用 check_same_thread=False + 单连接队列；②直接换 PostgreSQL
resp.usage 一直是 None：说明你开了 stream=True，usage 在最后一个 chunk 的 chunk.usage 里，需要累加
仪表盘 latency 全部显示为 0：用了 time.time() 精度不够，必须改用 time.perf_counter()

常见错误与解决方案

错误 1：流式响应拿不到 token

症状：resp.usage 报 AttributeError。修复：

def chat_tracked_stream(model, messages, user_tag="default", **kwargs):
    start = time.perf_counter()
    stream = client.chat.completions.create(
        model=model, messages=messages, stream=True, **kwargs
    )
    pt = ct = 0
    content = []
    for chunk in stream:
        if chunk.usage:
            pt = chunk.usage.prompt_tokens
            ct = chunk.usage.completion_tokens
        if chunk.choices and chunk.choices[0].delta.content:
            content.append(chunk.choices[0].delta.content)
    latency_ms = (time.perf_counter() - start) * 1000
    cost = tracker.record(model, user_tag, pt, ct, latency_ms)
    return "".join(content), cost

错误 2：多进程写 SQLite 锁死

症状：database is locked 报错。修复：

import sqlite3, threading
from queue import Queue

_write_q = Queue()
def _writer_loop(db_path):
    conn = sqlite3.connect(db_path, check_same_thread=False)
    while True:
        sql, params = _write_q.get()
        try:
            conn.execute(sql, params); conn.commit()
        except Exception as e:
            print("write err:", e)

threading.Thread(target=_writer_loop, args=("usage.db",), daemon=True).start()

业务里只投递任务
def safe_record(model, user_tag, pt, ct, latency_ms):
    price = PRICE_MAP.get(model, 2.50)
    cost = (ct / 1_000_000) * price
    _write_q.put((
        "INSERT INTO usage (ts,model,user_tag,prompt_tokens,completion_tokens,cost_cny,latency_ms) VALUES (?,?,?,?,?,?,?)",
        (time.time(), model, user_tag, pt, ct, round(cost,6), round(latency_ms,2))
    ))
    return cost

错误 3：账单和实际余额对不上

症状：用官方价格算出来 ¥3285，用脚本算出来 ¥450，差距巨大——其实是把官方 $价直接当 ¥价算了。修复：

# ❌ 错误写法
cost = (completion_tokens / 1_000_000) * usd_price  # 当成 ¥，多算 7.3 倍

✅ 正确写法（HolySheep ¥1=$1）
cost = (completion_tokens / 1_000_000) * PRICE_MAP[model]
例如 gpt-4.1 100 万 token：¥8.00，绝不是 $8.00

五、总结

这套仪表盘我自己已经稳定跑了 4 个月，最直接的收益有三个：

砍掉了 3 个无效的 prompt 模板，每月省 ¥600+
发现 Claude Sonnet 4.5 在合同抽取任务上性价比不如 GPT-4.1，整体切到 GPT-4.1
用 HolySheep 中转，微信/支付宝就能充值，¥1=$1 实打实省下 85%+

👉 免费注册 HolySheep AI，获取首月赠额度

AI API 成本监控：Token 消耗仪表盘搭建教程

一、为什么需要自建 Token 仪表盘？

二、整体架构

三、核心代码实现

3.1 埋点 SDK：自动拦截所有 OpenAI 兼容调用

HolySheep 中转配置，国内直连 < 50ms

2026 主流模型 output 价格（/MTok），HolySheep ¥1=$1 结算

业务里直接这样用，完全无感

3.2 数据接收服务：FastAPI + SQLite

3.3 Streamlit 可视化仪表盘

四、几个落地小建议

常见报错排查

常见错误与解决方案

错误 1：流式响应拿不到 token

错误 2：多进程写 SQLite 锁死

业务里只投递任务

错误 3：账单和实际余额对不上

✅ 正确写法（HolySheep ¥1=$1）

例如 gpt-4.1 100 万 token：¥8.00，绝不是 $8.00

五、总结

相关资源

相关文章

一、为什么需要自建 Token 仪表盘？

二、整体架构

三、核心代码实现

3.1 埋点 SDK：自动拦截所有 OpenAI 兼容调用

HolySheep 中转配置，国内直连 < 50ms

2026 主流模型 output 价格（/MTok），HolySheep ¥1=$1 结算

业务里直接这样用，完全无感

3.2 数据接收服务：FastAPI + SQLite

3.3 Streamlit 可视化仪表盘

四、几个落地小建议

常见报错排查

常见错误与解决方案

错误 1：流式响应拿不到 token

错误 2：多进程写 SQLite 锁死

业务里只投递任务

错误 3：账单和实际余额对不上

✅ 正确写法（HolySheep ¥1=$1）

例如 gpt-4.1 100 万 token：¥8.00，绝不是 $8.00

五、总结

相关资源

相关文章

🔥 推荐使用 HolySheep AI