简介API监控是保障大模型应用稳定性的基础能力尤其在调用量增长后完整记录每一次请求的调用日志成为排查问题、核算成本、优化服务的关键。通过Python在OpenAI兼容SDK外层包装日志埋点将请求时间、token用量、延迟、状态等核心字段写入SQLite即可实现轻量级的可观测性。在此基础上利用SQL聚合查询生成小时级趋势数据再通过ECharts或Grafana构建可视化看板能够直观呈现调用量、延迟波动和token消耗。这种方案无需引入重型组件适合个人开发者和中小团队快速落地。同时合理设计日志字段、避免重复计数、解决并发写入等常见坑位能让监控体系更健壮。本文基于DeepSeek API的实践展示了从零搭建日志监控、成本核算与告警的完整路径。1. 零基础 API 监控为什么你该给 DeepSeek 调用写日志DeepSeek 的调用量一旦涨起来API 监控就不再是可选项了。一次线上事故排查、一张给老板看的调用日报、一笔要分摊到业务线的成本全都依赖一份完整的调用日志而把日志变成可视化的趋势图才算真正把监控看板搭了起来。这条路零基础就能走通不用一开始就上企业级观测平台用 Python 包一层请求、把关键字段落在 SQLite 里、再画几张图就能解决你 80% 的监控需求。适合正在做 DeepSeek 应用的开发者、想管住 token 预算的技术负责人以及要给客户证明服务稳定性的交付团队。很多人在本地调通了一个对话接口就以为万事大吉。等到部署上去问题来了深夜某个时刻延迟从 1 秒飙到 8 秒是模型服务端抖动还是自己代码的锅某个业务方说下午调了几百次日志里却搜不到几条记录月底账单出来token 费用跟预期差了快一倍。没有调用日志这些问题全靠猜而猜是最贵的排查方式。2. 先想清楚再动手调用日志里抓哪些字段存储怎么选2.1 一份最小可用的调用日志要记什么给 DeepSeek 这类 OpenAI 兼容接口做日志最常见的误区是一上来就设计几十个字段结果埋点代码写了一整天最后有用的还是那几列。我一般先记一张九字段的明细表跑通之后再按需加字段如下字段示例值为什么重要ts2025-06-01T14:30:00时间戳是一切趋势分析的坐标轴modeldeepseek-chat不同模型价格、延迟完全不同必须区分endpointchat/completions定位是对话还是推理排查时有用prompt_tokens1240输入 token成本核算的左边completion_tokens85输出 token成本核算的右边total_tokens1325汇总值图表里最常被引用latency_ms1200延迟直接反映服务质量statusok / error失败率和错误码是稳定性核心指标request_id8a9f3e与业务侧请求 ID 关联追链路的锚点字段确定之后埋点位置要想清楚。最推荐的做法不是在业务代码里到处写print而是在客户端 SDK 外面包一层薄薄的封装所有 DeepSeek 调用都走同一个入口。这样日志逻辑只写一次业务代码不用塞满统计代码将来要加字段也只改一处。2.2 SQLite 还是时序库零基础选型与理由存储选型上很多人一上来就上 Prometheus、InfluxDB、ClickHouse对零基础项目来说这是过度设计。单机开发、日调用量在几万次以内、数据量每天几十 MBSQLite 完全够用而且有几个实打实的优势单文件部署拷贝一个.sqlite3文件就能迁移Python 标准库自带sqlite3不用装额外依赖事务和 SQL 支持完整统计查询写起来顺手。什么时候该换时序库两个信号一是调用量到每日百万级别SQLite 的聚合查询开始出现明显延迟二是你需要长时间保留原始明细做多维分析磁盘和查询压力都上来了。到那个阶段常见做法是把 Prometheus 作为指标存储、Grafana 做看板原始日志落到对象存储或 ES 里按需查。这个迁移是平滑的因为埋点字段设计不变变的只是写入目标。对于大多数个人开发者和中小企业我强烈建议先 SQLite不要为了“以后可能要扩容”提前上重型组件。技术债是欠给未来的不是欠给现在的现在把链路跑通、看到看板上有图了比什么都重要。2.3 DeepSeek 调用日志与通用 API 日志的差异点给 DeepSeek 写日志跟给普通 REST API 写日志有一个关键区别普通 API 日志记状态码和耗时就够了而大模型调用必须记 token 用量。token 直接对应账单金额是成本核算的唯一输入。另一个差异是错误形态模型接口的错误不只是 5xx还有rate limit、context length exceeded、invalid api key这类业务语义错误它们需要单独被统计因为它们各自对应不同的处理动作。还有一点容易被忽略DeepSeek 的接口虽然是 OpenAI 兼容的但返回的usage字段结构跟 OpenAI 有细微差别。兼容兼容不代表一模一样的坑这一点在后面的避坑章节里会展开。3. 包一层 Python 脚本把每一次 DeepSeek 调用记进 SQLite3.1 在 OpenAI SDK 外层加日志埋点这里用官方兼容 OpenAI 接口的 Python SDK在它外面包一层自定义的 Logger。核心思路是chat()方法替代你原来对 SDK 的直接调用业务侧只管传 messages日志由这个类全权负责import time import sqlite3 from datetime import datetime from openai import OpenAI class DeepSeekLogger: 在 OpenAI SDK 外层包装日志埋点记录每次调用的关键信息。 def __init__(self, api_key: str, db_path: str deepseek_calls.sqlite3, base_url: str https://api.deepseek.com): self.client OpenAI(api_keyapi_key, base_urlbase_url, max_retries0) self.db_path db_path self._init_db() def _init_db(self): conn sqlite3.connect(self.db_path) conn.execute( CREATE TABLE IF NOT EXISTS api_calls ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts TEXT NOT NULL, model TEXT, endpoint TEXT, prompt_tokens INTEGER DEFAULT 0, completion_tokens INTEGER DEFAULT 0, total_tokens INTEGER DEFAULT 0, latency_ms INTEGER DEFAULT 0, status TEXT, error_msg TEXT, request_id TEXT ) ) conn.execute(PRAGMA journal_modeWAL) conn.execute(PRAGMA busy_timeout5000) conn.commit() conn.close() def chat(self, messages, modeldeepseek-chat, **kwargs): start time.time() try: resp self.client.chat.completions.create( modelmodel, messagesmessages, **kwargs ) latency int((time.time() - start) * 1000) usage resp.usage self._log( tsdatetime.now().isoformat(timespecseconds), modelmodel, endpointchat/completions, prompt_tokensusage.prompt_tokens if usage else 0, completion_tokensusage.completion_tokens if usage else 0, total_tokensusage.total_tokens if usage else 0, latency_mslatency, statusok, error_msgNone, request_idresp.id, ) return resp except Exception as e: latency int((time.time() - start) * 1000) self._log( tsdatetime.now().isoformat(timespecseconds), modelmodel, endpointchat/completions, prompt_tokens0, completion_tokens0, total_tokens0, latency_mslatency, statuserror, error_msgstr(e), request_idNone, ) raise def _log(self, **fields): conn sqlite3.connect(self.db_path) conn.execute( INSERT INTO api_calls (ts, model, endpoint, prompt_tokens, completion_tokens, total_tokens, latency_ms, status, error_msg, request_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?), ( fields[ts], fields[model], fields[endpoint], fields[prompt_tokens], fields[completion_tokens], fields[total_tokens], fields[latency_ms], fields[status], fields[error_msg], fields[request_id], ), ) conn.commit() conn.close()逻辑说明类的chat()方法在调 SDK 前记录开始时间成功响应拿到后立即计算延迟并取usage里的 token 数据异常时也记录一条 status 为 error 的日志然后把异常原样抛出不影响业务代码的原有错误处理。_init_db()里建表并开了 WAL 模式这对多线程写入有实际帮助第 5 章会专门讲。参数说明max_retries0很关键SDK 默认会自动重试重试会导致重复计数这里显式关掉后重试逻辑由业务侧自己决定。base_url用 DeepSeek 官方兼容端点。PRAGMA busy_timeout5000的意思是遇到锁等待最多 5 秒超过才报错配合 WAL 能解决大部分写锁问题。调用方式跟原来几乎一样logger DeepSeekLogger(api_key你的key) resp logger.chat( [{role: user, content: 用一句话介绍你自己}], modeldeepseek-chat, max_tokens200, ) print(resp.choices[0].message.content)每个请求都会落一条明细到 SQLite。你不需要理解 SQL 也能先跑起来因为表结构已经在_init_db()里建好了。3.2 建表和常用统计查询调用量、延迟、tokens数据落库只是第一步监控的核心是聚合。以下几个查询是看板的最常用 SQL先存为.sql文件备用或者直接在 Python 脚本里执行-- 按小时聚合调用量与平均延迟 SELECT strftime(%Y-%m-%d %H:00, ts) AS hour, COUNT(*) AS calls, CAST(AVG(latency_ms) AS INTEGER) AS avg_latency, SUM(COALESCE(total_tokens, 0)) AS total_tokens FROM api_calls GROUP BY hour ORDER BY hour; -- 按模型统计调用次数与 token 消耗 SELECT model, COUNT(*) AS calls, SUM(COALESCE(total_tokens, 0)) AS tokens, SUM(CASE WHEN status error THEN 1 ELSE 0 END) AS errors FROM api_calls GROUP BY model; -- 最近 100 条错误记录 SELECT ts, model, latency_ms, error_msg, request_id FROM api_calls WHERE status error ORDER BY ts DESC LIMIT 100;逻辑说明第一个查询把 ISO 时间字符串按小时截断聚合得到时间序列这是折线图的数据源第二个查询按模型分组得到模型维度的横向对比第三个查询是排障入口直接看最近的错误明细。参数说明strftime(%Y-%m-%d %H:00, ts)把时间截断到小时%H:00的写法是为了在图表横轴上显示整点COALESCE(total_tokens, 0)防止 NULL 值导致聚合结果为 NULL错误统计用CASE WHEN加SUM比单独查两次更高效。如果你想看到分钟级粒度把%H:00换成%H:%M就行但分钟级聚合会带来更多噪声建议先看小时级。3.3 让监控脚本能长期运行systemd 守护与清理策略本地测试手动跑没问题但真要持续监控得让采集和清理进程长期跑。常见做法是写一个常驻小服务每间隔一段时间清理过期明细避免 SQLite 无限膨胀# monitor_service.py import sqlite3 import time from datetime import datetime, timedelta DB_PATH deepseek_calls.sqlite3 def cleanup(keep_days: int 30): conn sqlite3.connect(DB_PATH) cutoff (datetime.now() - timedelta(dayskeep_days)).isoformat(timespecseconds) conn.execute(DELETE FROM api_calls WHERE ts ?, (cutoff,)) conn.commit() conn.close() if __name__ __main__: while True: cleanup(keep_days30) time.sleep(3600) # 每小时清理一次逻辑说明这个脚本只做一个事——删除 30 天前的明细然后睡一小时再删。原始明细保留 30 天是个人项目和中小团队的主流做法要更长的历史趋势靠聚合表而不是保留全量明细。然后用 systemd 托管[Unit] DescriptionDeepSeek Monitor Service Afternetwork-online.target [Service] Userdeploy WorkingDirectory/opt/deepseek-monitor ExecStart/usr/bin/python3 /opt/deepseek-monitor/monitor_service.py Restartalways RestartSec10 [Install] WantedBymulti-user.target注意 systemd 文件里的WorkingDirectory要指向脚本所在目录否则 SQLite 的相对路径会跑到别处去。Restartalways保证进程挂掉后 10 秒自动拉起来这是长期跑的底线保障。4. 搭可视化看板从静态 HTML 到 Grafana 两条路4.1 最简路径用 ECharts 生成静态报告零基础最友好的看板方案不是搭服务而是定时生成一份静态 HTML。用 Python 从 SQLite 读聚合数据写成 JSON再用 ECharts 在浏览器里渲染成图。这个方案的优点是不用维护看板服务生成的文件放到任意静态目录甚至直接用 Python 的http.server就能在局域网里看。# build_report.py import sqlite3 import json conn sqlite3.connect(deepseek_calls.sqlite3) conn.row_factory sqlite3.Row hours conn.execute( SELECT strftime(%Y-%m-%d %H:00, ts) AS hour, COUNT(*) AS calls, ROUND(AVG(latency_ms), 0) AS avg_latency, SUM(COALESCE(total_tokens, 0)) AS tokens FROM api_calls GROUP BY hour ORDER BY hour ).fetchall() hour_data [dict(r) for r in hours] models conn.execute( SELECT model, COUNT(*) AS calls, SUM(COALESCE(total_tokens, 0)) AS tokens, SUM(CASE WHEN status error THEN 1 ELSE 0 END) AS errors FROM api_calls GROUP BY model ).fetchall() model_data [dict(r) for r in models] conn.close() with open(report_data.json, w, encodingutf-8) as f: json.dump({hours: hour_data, models: model_data}, f, ensure_asciiFalse)生成的数据文件由 HTML 读取后画图。核心逻辑只有一段 ECharts 配置fetch(report_data.json) .then(r r.json()) .then(data { const chart echarts.init(document.getElementById(chart)); chart.setOption({ tooltip: { trigger: axis }, xAxis: { type: category, data: data.hours.map(h h.hour) }, yAxis: [ { type: value, name: 调用量 }, { type: value, name: 平均延迟(ms) } ], series: [ { name: 调用量, type: bar, data: data.hours.map(h h.calls) }, { name: 平均延迟, type: line, yAxisIndex: 1, data: data.hours.map(h h.avg_latency) } ] }); });逻辑说明横轴是小时左轴是调用量柱状图右轴是平均延迟折线图一眼能看出调用高峰和延迟波动的关系。ECharts 的setOption是增量配置改数据源不需要改其他代码。部署时report_data.json、report.html、echarts.min.js 放同一目录跑python3 -m http.server 8080浏览器打开就是看板了。定时任务就交给 cron每 5 分钟生成一次 JSON浏览器刷新即更新。这个方案够很多小项目用几个月。4.2 生产路径用 Prometheus 暴露指标Grafana 拉取画图当不只一个人要看、要保留历史告警、要接入统一监控中心时就该切到 Prometheus 加 Grafana 组合了。这套方案的核心变化是不把明细写 SQLite而是把指标通过 HTTP 暴露给 Prometheus 周期性抓取。# metrics.py from prometheus_client import start_http_server, Counter, Histogram calls_total Counter( deepseek_calls_total, DeepSeek 调用总次数, [model, status], ) latency Histogram( deepseek_latency_seconds, DeepSeek 调用延迟, [model], buckets(0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10), ) tokens_used Counter( deepseek_tokens_total, 累计消耗 token 数, [model, type], # type: prompt / completion ) if __name__ __main__: start_http_server(8000) while True: time.sleep(60)在之前写的DeepSeekLogger里把成功和失败分别打点calls_total.labels(model, ok).inc() latency.labels(model).observe(latency_seconds) tokens_used.labels(model, prompt).inc(usage.prompt_tokens) tokens_used.labels(model, completion).inc(usage.completion_tokens)逻辑说明Counter 只增不减适合累计量Histogram 自动分桶记录延迟分布查询 p95 就不用自己排序了。start_http_server(8000)起一个独立端口Prometheus 配置抓取这个端点的/metrics即可。Grafana 里加 Prometheus 数据源之后两个最常用的查询语句sum(rate(deepseek_calls_total[5m])) by (model) histogram_quantile(0.95, sum(rate(deepseek_latency_seconds_bucket[5m])) by (le))第一条算每 5 分钟的调用速率第二条算延迟的 95 分位。这两句是看板的核心其他面板基本都是它们的变体。Grafana 的画图能力比静态 HTML 强得多告警规则也内置生产环境建议直接走这条路线。5. 避坑指南DeepSeek 日志监控最常见的 5 个坑5.1 请求被自动重试日志重复计数现象业务代码只触发了一次对话数据库里却多了两三条记录调用量看起来是实际的好几倍。原因OpenAI 兼容 SDK 默认开启自动重试遇到 429 或 5xx 会自动重发请求而你的日志埋在业务层每次重试都算了一次调用。解决初始化客户端时显式关闭重试OpenAI(api_key..., base_url..., max_retries0)。更保险的做法是在日志表里用request_id建唯一索引写入时做去重。把重试放到自己可控的中间层比 SDK 黑匣子里的自动重试可靠得多。5.2 流式响应拿不到 tokens现象调用时开了streamTrue结果日志里completion_tokens全是 0成本核算是错的。原因流式响应是一段一段返回的默认的 usage 字段只在最后一个 chunk 里才带很多场景下根本取不到。解决新版 SDK 支持在请求里加stream_options{include_usage: True}最后的 chunk 会携带完整 usage。但要注意兼容 OpenAI 接入的 DeepSeek 端点是否支持这个参数需要实测确认如果拿不到就只能按输出字符数粗略估算或者把流式改成非流式做精确记录。做成本核算时对流式调用要单独标注估算误差。5.3 SQLite 并发写入报 database is locked现象业务跑了多个线程日志表开始间歇性报database is locked服务看起来“卡一下又好了”。原因SQLite 默认允许一个写者多个线程同时写会互相阻塞超时就会抛异常。解决建库时执行PRAGMA journal_modeWAL和PRAGMA busy_timeout5000。WAL 模式下读和写不互斥写和写之间也排队等待而不是立即报错。这个改动对多线程场景有明显改善。如果还是冲突就单独用一个进程负责写库业务线程只往队列里丢日志。5.4 延迟统计虚高问题出在打点位置现象平均延迟显示 5 秒但用户实际感受没那么慢对不上账。原因日志埋点在 SDK 调用前后这个延迟包含了 DNS 解析、TCP 建连、排队时间还有上面提到的重试耗时。SDK 内部具体做了什么对外几乎是个黑匣子。解决明确自己想统计的是“用户感知延迟”还是“服务端处理延迟”。用户感知延迟就按 SDK 调用前后算接受它包含网络波动服务端处理延迟则要看响应头里的时间戳。更重要的是在看板标题上标注清楚口径不然自己过两周也会看晕。5.5 时区没统一按小时的图表对不上现象看板显示凌晨 4 点调用量最高但业务方坚称那会儿没人用或者深夜时段延迟异常高怎么查都查不到原因。原因SQLite 的datetime(now)返回 UTC 时间而业务里存的是本地时间两套混着用聚合结果自然错位。解决统一策略只有一个——存储一律用 UTC ISO 字符串只在展示层转成业务所在地时区。建表时就在插入逻辑里强制 UTC不要给业务侧留选择余地。排障时先看日志里时间有没有Z后缀或时区偏移量这能省掉大量无意义的排查时间。6. 进阶玩法成本核算、超额告警和调用链分析日志监控跑通之后下一个价值点是成本核算。DeepSeek 是按 token 计费不同模型的输入输出价格不同写一个估算函数每天跑一次就能把账单预判得八九不离十# 按官方价目表维护单位元/百万 token PRICE { deepseek-chat: {input: 0.0, output: 0.0}, deepseek-reasoner: {input: 0.0, output: 0.0}, } def estimate_cost(model, prompt_tokens, completion_tokens): p PRICE.get(model, {input: 0, output: 0}) return (prompt_tokens / 1_000_000) * p[input] \ (completion_tokens / 1_000_000) * p[output]注意价格的维护要设一个提醒模型厂商调价后及时更新不然估算失真。这个函数接入每日汇总脚本输出一张成本趋势表比月底看账单被吓一跳舒服得多。超额告警是另一个立竿见影的功能。每小时跑一次 SQL统计当日累计 token 和失败率超过阈值就推企业微信/钉钉的 webhook。阈值第一次设置容易翻车设得太松告警失去意义设得太紧每天半夜被叫醒。我的习惯是先观察两周正常水位再在正常值 1.5 倍的地方设告警线跑一个月再收紧。调用链分析是把日志价值放大的关键一步。在DeepSeekLogger.chat()里增加一个user_tag参数业务方调的时候传入自己的订单号或用户 ID日志就能做分账哪个业务线花了多少钱、哪个功能调用量异常一条 SQL 就查出来了。不要等到多个业务共用同一个 Key 才想到这点前期埋进去几乎零成本后期补就是折磨。我现在接任何一家大模型 API第一件事就是把日志器封装成内部工具库业务代码一进来就带埋点等要排查问题时从不怕没有数据。这个习惯帮我省掉了无数次“凭感觉猜问题”的折腾希望帮到你。本文还有配套的精品资源点击获取
