简介:一份面向零基础开发者与运维人员的 DeepSeek API 监控实战文档,定位为可上手的实操指南,解决调用日志分散、缺少可视化手段等问题。内容从 API 监控的核心概念讲起,完整覆盖 DeepSeek 调用权限申请、日志格式读取、数据清洗与预处理、关键指标(响应时间、错误率、调用频率、吞吐量)提取与分析,再到基于 ECharts 可视化看板的搭建、交互设计、部署上云与持续运维,脉络清晰、案例贯穿,读者可参照目录按步骤复现一套可用的监控看板。资源为单个 PDF 文件,共 28 页,压缩包大小约 1.89MB,文档内文字、图表、目录均显示正常,适合直接打印或电子阅读。目前已有 115 人学习下载,对于刚接触 API 监控或希望提升 DeepSeek 应用稳定性的人来说,是一份值得参考的实操型资料。
1. 先从日志下手:零基础为什么也能把 DeepSeek 调用监控起来
上个月看到账单时我有点懵:一个内部工具项目,DeepSeek 的 API 消耗比预期多了一倍,但翻遍代码也不知道钱花在哪。问了一圈,有人建议上 Prometheus + Grafana 那套,我算了算学习成本,直接放弃了。后来换了个思路:既然所有调用都要走 API,那就在最外层把日志留下来,再找个轻量看板把数据画出来。这就是标题里“API 监控”的实际做法——不需要复杂的链路追踪,先有日志,后面所有分析、告警、对账都顺理成章。
这篇笔记写给谁?给那些用 DeepSeek API 写脚本、做产品原型、接进 IDE 或个人项目的人。你用不着是 SRE,也不用懂 Kubernetes,只要会一点 Python,能跑通一个 Flask 服务,就能在半天内搭出一套属于你自己的调用监控看板。它能回答三个问题:钱花在哪、哪些请求慢、哪天在报错。
2. 埋点采集:在 API 调用最外层记下每次请求的完整现场
2.1 日志字段清单:记录哪些内容才能支撑起对账、延迟与错误分析
埋点之前先想清楚要记什么字段。记少了后面补不了,记多了数据库撑不住。我的经验是“宁可多记几个冗余字段,也不要漏掉关键现场”,因为日志一旦丢了一部分,趋势线就断了。
一套最小的 DeepSeek 调用日志字段,我一般按下面这张表来定:
| 字段 | 类型 | 说明 |
|---|---|---|
| id | INTEGER | 自增主键,用来排序 |
| request_id | TEXT | 请求唯一标识,排查问题时按它去 API 平台侧对账 |
| model | TEXT | 模型名,比如 deepseek-chat 或 deepseek-reasoner |
| status | TEXT | ok 或 error,error 时看 error_type |
| error_type | TEXT | 错误类型,标记超时、限流、上下文超长、鉴权失败等 |
| prompt_tokens | INTEGER | 输入 token 数 |
| completion_tokens | INTEGER | 输出 token 数 |
| total_tokens | INTEGER | 合计 token 数 |
| cache_hit_tokens | INTEGER | 缓存命中的 token 数,对核对成本很有用 |
| latency_ms | INTEGER | 请求总耗时(毫秒),从发出到拿到完整响应 |
| req_ts_utc | TEXT | UTC 时间,作为基准时间 |
| req_ts_local | TEXT | 本地时间,按天分组画图时直接用它 |
| endpoint | TEXT | 调用的接口路径,如 /chat/completions |
| prompt_preview | TEXT | 输入内容的前 200 个字符,用于人工回溯 |
两个时间字段可能会让你觉得冗余,但实际用起来非常方便。UTC 时间用来保证排序和跨时区对账准确,本地时间用来按天分组、按小时画热力图。避免在 SQL 里反复做时区转换,这是踩过坑之后总结出来的习惯。
2.2 用 Python 封装一轮 DeepSeek 调用,顺手把日志写进 SQLite
既然 DeepSeek 的 API 兼容 OpenAI 协议,直接用openai这个 Python 库就能调。封装的核心思路是:把业务代码里原本直接调用client.chat.completions.create()的地方,改成调用我们自己的封装函数。这样埋点只写在一处,不会漏。
# deepseek_monitor.py import time import sqlite3 from datetime import datetime, timezone from openai import OpenAI DB_PATH = "deepseek_logs.db" class MonitoredDeepSeek: def __init__(self, api_key: str): # timeout 给足 60 秒,reasoner 模型思考时间长,30 秒经常不够 self.client = OpenAI( api_key=api_key, base_url="https://api.deepseek.com", timeout=60, max_retries=1, # 重试会让耗时和费用双倍膨胀,宁可失败一次也不要默默重试 ) def _save_record(self, record: dict): conn = sqlite3.connect(DB_PATH) conn.execute("PRAGMA journal_mode=WAL;") conn.execute( """ INSERT INTO deepseek_calls (request_id, model, status, error_type, prompt_tokens, completion_tokens, total_tokens, cache_hit_tokens, latency_ms, req_ts_utc, req_ts_local, endpoint, prompt_preview) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, (record["request_id"], record["model"], record["status"], record["error_type"], record["prompt_tokens"], record["completion_tokens"], record["total_tokens"], record["cache_hit_tokens"], record["latency_ms"], record["req_ts_utc"], record["req_ts_local"], record["endpoint"], record["prompt_preview"]) ) conn.commit() conn.close() def chat(self, model: str, messages: list, **kwargs): req_ts_utc = datetime.now(timezone.utc).isoformat() req_ts_local = datetime.now().astimezone().isoformat(timespec="seconds") t0 = time.perf_counter() try: resp = self.client.chat.completions.create( model=model, messages=messages, **kwargs ) latency_ms = int((time.perf_counter() - t0) * 1000) usage = resp.usage record = { "request_id": getattr(resp, "id", ""), "model": model, "status": "ok", "error_type": None, "prompt_tokens": usage.prompt_tokens if usage else 0, "completion_tokens": usage.completion_tokens if usage else 0, "total_tokens": usage.total_tokens if usage else 0, "cache_hit_tokens": getattr(usage, "prompt_cache_hit_tokens", 0) if usage else 0, "latency_ms": latency_ms, "req_ts_utc": req_ts_utc, "req_ts_local": req_ts_local, "endpoint": "/chat/completions", "prompt_preview": str(messages[-1].get("content", ""))[:200], } self._save_record(record) return resp except Exception as e: latency_ms = int((time.perf_counter() - t0) * 1000) record = { "request_id": "", "model": model, "status": "error", "error_type": type(e).__name__, "prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0, "cache_hit_tokens": 0, "latency_ms": latency_ms, "req_ts_utc": req_ts_utc, "req_ts_local": req_ts_local, "endpoint": "/chat/completions", "prompt_preview": str(messages[-1].get("content", ""))[:200], } self._save_record(record) raise重点说三个参数的选择逻辑。timeout=60是因为 DeepSeek 的 reasoner 模型面对复杂推理问题时,思考时间可能超过 20 秒,默认的 30 秒超时会在模型刚要返回结果时被掐断,日志里会留下一大批超时误报。max_retries=1的意思是 SDK 内部最多帮你重试一次,重试期间用户侧等待时间翻倍,而且第二次请求可能成功导致第一次请求的 token 费用照收,成本统计会虚高。prompt_preview截断到 200 字符,既能回溯调用场景,又不会让数据库体积失控。
用的时候,业务代码里把原来的client.chat.completions.create(...)换成monitor.chat(...)即可,其余参数完全不动。
2.3 批量任务与并发场景:埋点放在哪一层才不重复不遗漏
如果你只是写个单次调用的脚本,上面这个封装够用了。但如果你在跑批量任务,比如用循环处理几百个文件,或者接进了类似 harness 这种多智能体编排框架,埋点位置就要想清楚。
常见做法是把埋点约束在“所有请求必经的最外层出口”。以多智能体编排为例,每个 agent 内部可能调好几次模型,如果你在 agent 内部各调各的,日志会碎片化;更好的做法是在统一调用 DeepSeek API 的那一层上做埋点,也就是无论 agent 怎么编排,最后发出的 HTTP 请求都走同一个函数。这样日志里能看到一次完整任务产生了多少次调用、每次调用花了多久、哪一轮消耗了最多 token。
批量循环场景还有一个坑:不要在循环体内部手动写日志记录代码。把日志逻辑放进封装函数,循环里只调monitor.chat(),日志自然一条不落。并发场景下 SQLite 的写入锁值得留意,后面第 5 章专门展开。总之埋点层越靠外越稳,重试逻辑也越简单——每个请求只记一次,不要因为内部重试记出重复行。
3. 存储与分析:SQLite 单文件库加几张 SQL,把调用数据变成指标
3.1 为什么先用 SQLite:零部署、单文件、读写够用的边界在哪里
数据记下来之后要落库。对个人开发者和小团队来说,SQLite 是性价比最高的选择。它不需要单独装服务,就是一个文件,Python 标准库自带驱动,千行级别的调用记录查询基本是毫秒级。
用 MySQL 或者 PostgreSQL 当然显得更“正规”,但部署、账号、连接池这些事对零基础的人来说全是额外负担。Prometheus 那套生态更重,而且它的模型是拉取指标,不适合存带 prompt 内容的调用日志。等到日志量真正大到 SQLite 撑不住的时候,再迁移也不难,因为指标口径已经在 SQL 里写清楚了。
| 维度 | SQLite | MySQL | Prometheus |
|---|---|---|---|
| 部署 | 无,单文件 | 需要服务端 | 需要服务端和采集器 |
| 查询 | SQL | SQL | PromQL,有学习成本 |
| 适合量级 | 百万行以内 | 亿级 | 适合指标,不适合详细日志 |
| 扩展成本 | 直接换库 | 需要维护 | 组件多,运维重 |
我的判断标准很简单:日调用量在几万次以下,SQLite 完全够用;个人项目一般到不了这个量级。真正需要换库的时候,你已经有明确的迁移理由了。
3.2 建表与写入:WAL 模式、索引和批量插入的取舍
建表语句直接放在初始化脚本里,字段和 2.1 节保持一致。索引只加三个最常用的查询维度:时间、模型、状态。
CREATE TABLE IF NOT EXISTS deepseek_calls ( id INTEGER PRIMARY KEY AUTOINCREMENT, request_id TEXT, model TEXT, status TEXT, error_type TEXT, prompt_tokens INTEGER, completion_tokens INTEGER, total_tokens INTEGER, cache_hit_tokens INTEGER DEFAULT 0, latency_ms INTEGER, req_ts_utc TEXT, req_ts_local TEXT, endpoint TEXT, prompt_preview TEXT ); CREATE INDEX IF NOT EXISTS idx_ts ON deepseek_calls(req_ts_utc); CREATE INDEX IF NOT EXISTS idx_model ON deepseek_calls(model); CREATE INDEX IF NOT EXISTS idx_status ON deepseek_calls(status);写入时有三件事值得做。第一,每次连接后执行PRAGMA journal_mode=WAL;。WAL 模式让读操作不阻塞写操作,看板在刷新查询的同时,日志写入还能继续,不会出现“页面一转圈,日志就丢了”的情况。第二,批量场景下用executemany()代替单条execute(),比如攒满 100 条再写一次,能显著降低 SQLite 的锁竞争。第三,连接用完就关,不要持有长连接,SQLite 不是为高并发连接设计的。
# batch_write.py import sqlite3 def save_records(records: list[dict]): conn = sqlite3.connect("deepseek_logs.db") conn.execute("PRAGMA journal_mode=WAL;") conn.executemany( """ INSERT INTO deepseek_calls (request_id, model, status, error_type, prompt_tokens, completion_tokens, total_tokens, cache_hit_tokens, latency_ms, req_ts_utc, req_ts_local, endpoint, prompt_preview) VALUES (:request_id, :model, :status, :error_type, :prompt_tokens, :completion_tokens, :total_tokens, :cache_hit_tokens, :latency_ms, :req_ts_utc, :req_ts_local, :endpoint, :prompt_preview) """, records ) conn.commit() conn.close()用命名参数后字典的键名一目了然,字段再多也不容易写错位置。
3.3 核心指标 SQL:调用量、错误率、Token 消耗、成本与 P95 延迟
有了表和数据,后面的分析全靠 SQL。五个最常用的查询我直接列出来,照着抄就行。
按天统计调用量与错误率,用于看整体趋势和异常波动:
SELECT date(req_ts_local) AS day, COUNT(*) AS total_calls, SUM(CASE WHEN status = 'ok' THEN 1 ELSE 0 END) AS ok_calls, ROUND(SUM(CASE WHEN status = 'error' THEN 1 ELSE 0 END) * 100.0 / COUNT(*), 2) AS error_rate_pct, SUM(total_tokens) AS total_tokens FROM deepseek_calls WHERE req_ts_local >= datetime('now', '-14 days', 'localtime') GROUP BY day ORDER BY day;按模型统计最近 30 天的 Token 消耗和估算成本,用于回答“钱花在哪”:
SELECT model, SUM(prompt_tokens) AS sum_prompt_tokens, SUM(completion_tokens) AS sum_completion_tokens, ROUND(SUM(prompt_tokens) / 1000000.0 * :prompt_price, 2) AS prompt_cost, ROUND(SUM(completion_tokens) / 1000000.0 * :completion_price, 2) AS completion_cost, ROUND((SUM(prompt_tokens) / 1000000.0 * :prompt_price) + (SUM(completion_tokens) / 1000000.0 * :completion_price), 2) AS total_cost FROM deepseek_calls WHERE req_ts_local >= datetime('now', '-30 days', 'localtime') GROUP BY model ORDER BY total_cost DESC;价格参数:prompt_price和:completion_price按你账号所在平台的最新计价表填写,单位是“每百万 token 的价格”。把价格做成参数而不是写死在 SQL 里,是因为模型和价格都在变,写死在代码里后面对账会很痛苦。
统计最近 24 小时成功请求的 P95 延迟,用于定位“慢请求”:
SELECT latency_ms FROM deepseek_calls WHERE status = 'ok' AND req_ts_local >= datetime('now', '-24 hours', 'localtime') ORDER BY latency_ms LIMIT 1 OFFSET (SELECT CAST(COUNT(*) * 0.95 AS INT) FROM deepseek_calls WHERE status = 'ok' AND req_ts_local >= datetime('now', '-24 hours', 'localtime'));这个写法避开了一些数据库才有的PERCENTILE_CONT窗口函数,用排序后取 95% 位置的值来实现 P95,SQLite 直接能跑。查询先把耗时从小到大排序,再跳过前 95% 的行,取到的第一条就是 P95 延迟。按错误类型分组统计:
SELECT error_type, COUNT(*) AS cnt FROM deepseek_calls WHERE status = 'error' AND req_ts_local >= datetime('now', '-7 days', 'localtime') GROUP BY error_type ORDER BY cnt DESC;这四条 SQL 基本覆盖了监控看板所需的全部核心数据。剩下的工作就是把这些结果用图表画出来。
3.4 指标口径别含糊:缓存命中、重试与 tool call 对统计的影响
SQL 写完之后,指标口径必须提前定清楚,否则数据对不上会很头疼。第一件事是缓存命中:DeepSeek 对重复前缀有缓存机制,命中缓存的那部分输入 token 计费远低于未命中的。日志里的cache_hit_tokens字段就是为这个准备的。成本估算时如果把缓存 token 按全价算,月底对账会发现预估成本比实际账单高出一截,容易误以为账单错了。
第二件事是重试。SDK 的max_retries如果设置成大于 1,一次用户请求可能变成两三次 HTTP 请求,延迟统计会被严重拉高,费用也会重复计算。我的选择是max_retries=1,宁可让个别请求失败,也不要让统计数据失真。
第三件事是 tool call 场景。见过不少人在日志里发现“API 超时”,但实际原因不是 DeepSeek 服务出问题,而是工具调用的执行结果没有及时返回。搜索热词里经常能看到“deepseek messages tool calls need immediate results”这类报错,它的本质是:模型在等待工具执行结果时,如果消息流没有闭环,调用方会一直等下去。日志里这类请求的latency_ms会特别高,如果只按耗时判断 API 健康状态,会产生大量误报。区分方法很简单:看endpoint和请求参数里是否包含工具定义,这类请求的耗时字段参考价值要打折扣,真正需要关心的是 DeepSeek 侧返回首个 token 的时间,但那需要 SDK 层级的支持,暂时记总时长即可。
4. 可视化看板:Flask 加 ECharts,三十分钟跑起一个本地面板
4.1 看板该放哪几张图:从趋势判断异常,从分布定位根因
看板不是图表越多越好,而是每张图都要回答一个具体问题。
| 图表 | 指标 | 回答的问题 |
|---|---|---|
| 近 14 天调用量与错误率折线 | 每日请求数、错误率 | 整体服务健康度有没有波动 |
| 近 24 小时延迟分布 | P50 / P95 延迟 | 慢请求集中在什么时段 |
| 模型调用占比饼图 | 各模型调用次数 | 哪个模型用得最多 |
| 错误类型柱状图 | error_type 计数 | 当前报错主要是哪一类 |
| 近 30 天各模型成本排行 | 估算成本 | 钱花在了哪个模型上 |
五张图足够了。再多的图表就是给看板添乱,你盯不过来。
4.2 Flask 后端:四个 JSON 接口把 SQL 结果喂给前端
后端用 Flask 写,逻辑很薄:每个接口执行一条 SQL,把结果转成 JSON 返回。前端页面负责拉数据和绘图。
# app.py from flask import Flask, jsonify, render_template import sqlite3 app = Flask(__name__) DB_PATH = "deepseek_logs.db" def query(sql: str, args: tuple = ()) -> list: conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row rows = conn.execute(sql, args).fetchall() conn.close() return [dict(row) for row in rows] @app.route("/api/trend") def api_trend(): return jsonify(query(""" SELECT date(req_ts_local) AS day, COUNT(*) AS total_calls, ROUND(SUM(CASE WHEN status = 'error' THEN 1 ELSE 0 END) * 100.0 / COUNT(*), 2) AS error_rate FROM deepseek_calls WHERE req_ts_local >= datetime('now', '-14 days', 'localtime') GROUP BY day ORDER BY day; """)) @app.route("/api/model_share") def api_model_share(): return jsonify(query(""" SELECT model, COUNT(*) AS cnt FROM deepseek_calls WHERE req_ts_local >= datetime('now', '-7 days', 'localtime') GROUP BY model ORDER BY cnt DESC; """)) @app.route("/api/errors") def api_errors(): return jsonify(query(""" SELECT error_type, COUNT(*) AS cnt FROM deepseek_calls WHERE status = 'error' AND req_ts_local >= datetime('now', '-24 hours', 'localtime') GROUP BY error_type ORDER BY cnt DESC; """)) @app.route("/api/cost") def api_cost(): return jsonify(query(""" SELECT model, SUM(prompt_tokens) AS prompt_tokens, SUM(completion_tokens) AS completion_tokens FROM deepseek_calls WHERE req_ts_local >= datetime('now', '-30 days', 'localtime') GROUP BY model ORDER BY completion_tokens DESC; """)) @app.route("/") def index(): return render_template("index.html") if __name__ == "__main__": app.run(host="0.0.0.0", port=5000)每个接口对应一张图。/api/trend返回 14 天的调用量和错误率;/api/model_share返回近 7 天的模型分布;/api/errors返回最近 24 小时的错误类型统计;/api/cost返回各模型 token 消耗。前端通过这些接口拿数据,不用直接访问数据库,结构清晰,后面加鉴权也方便。
4.3 ECharts 前端:一个 HTML 页面把图表全部挂起来
前端用 ECharts,一个 HTML 页面就能装下全部图表。ECharts 是国产开源图表库,中文文档全,配置项直观,对零基础用户很友好。把下面的文件保存为templates/index.html,放在 Flask 项目的 templates 目录下。
<!DOCTYPE html> <html lang="zh-CN"> <head> <meta charset="UTF-8"> <title>DeepSeek 调用监控看板</title> <!-- 实际使用时可下载 echarts.min.js 到本地 static 目录,也可用公共 CDN --> <script src="/static/echarts.min.js"></script> <style> body { background: #f5f6fa; font-family: -apple-system, "PingFang SC", sans-serif; margin: 20px; } .chart-box { background: #fff; border-radius: 8px; padding: 16px; margin-bottom: 16px; } .chart { width: 100%; height: 320px; } .grid { display: grid; grid-template-columns: 1fr 1fr; gap: 16px; } @media (max-width: 800px) { .grid { grid-template-columns: 1fr; } } </style> </head> <body> <h2>DeepSeek 调用监控</h2> <div class="chart-box"><div id="trend" class="chart"></div></div> <div class="grid"> <div class="chart-box"><div id="modelShare" class="chart"></div></div> <div class="chart-box"><div id="errors" class="chart"></div></div> </div> <script> const charts = {}; function initChart(id) { charts[id] = echarts.init(document.getElementById(id)); return charts[id]; } function drawTrend(data) { const chart = charts.trend || initChart('trend'); chart.setOption({ title: { text: '近 14 天调用量与错误率' }, tooltip: { trigger: 'axis' }, legend: { data: ['调用量', '错误率'] }, xAxis: { type: 'category', data: data.map(d => d.day) }, yAxis: [ { type: 'value', name: '调用量' }, { type: 'value', name: '错误率(%)' } ], series: [ { name: '调用量', type: 'line', data: data.map(d => d.total_calls), smooth: true }, { name: '错误率', type: 'line', yAxisIndex: 1, data: data.map(d => d.error_rate), smooth: true, lineStyle: { color: '#e74c3c' } } ] }); } function drawModelShare(data) { const chart = charts.modelShare || initChart('modelShare'); chart.setOption({ title: { text: '近 7 天模型调用占比' }, tooltip: { trigger: 'item' }, series: [{ type: 'pie', radius: ['30%', '70%'], data: data.map(d => ({ name: d.model, value: d.cnt })) }] }); } function drawErrors(data) { const chart = charts.errors || initChart('errors'); chart.setOption({ title: { text: '最近 24 小时错误类型' }, xAxis: { type: 'category', data: data.map(d => d.error_type || 'unknown') }, yAxis: { type: 'value' }, series: [{ type: 'bar', data: data.map(d => d.cnt), itemStyle: { color: '#e74c3c' } }] }); } async function refresh() { const [trend, modelShare, errors] = await Promise.all([ fetch('/api/trend').then(r => r.json()), fetch('/api/model_share').then(r => r.json()), fetch('/api/errors').then(r => r.json()) ]); drawTrend(trend); drawModelShare(modelShare); drawErrors(errors); } window.addEventListener('resize', () => { Object.values(charts).forEach(c => c.resize()); }); refresh(); </script> </body> </html>前端代码的逻辑很直接:页面加载时同时请求三个接口,拿到数据后分别调用三个绘图函数。getOption里最常用到的两个配置是smooth: true让折线平滑,以及双 y 轴对应不同量纲的指标。窗口大小变化时,resize()事件让图表自动适配。
4.4 自动刷新与访问方式:看板开着就行,不用手动刷页面
看板图表的更新频率取决于你的调用量。个人项目一天几百次调用的话,每 30 秒刷新一次足够;如果接入的是实时业务,可以缩短到 10 秒。把refresh()改成一个定时任务:
async function refresh() { // ...原有代码 } refresh(); setInterval(refresh, 30000); // 每 30 秒自动拉取一次数据当脚本在后台跑着批量任务时,看板会自动更新,不必手动刷新页面。app.run(host="0.0.0.0")意味着局域网内的其他设备也能通过你的电脑 IP 访问看板,比如手机在工位上瞄一眼今天的情况。如果看板要暴露到公网或者多人共用一个入口,建议在前面加一层简单的 token 鉴权,至少不要让服务裸奔在公网端口上。
5. 避坑清单:DeepSeek 调用日志看板最容易翻车的五个地方
5.1 usage 字段没取到:缓存命中和异常返回都会让 Token 统计突然归零
现象:某天开始,看板上的 token 消耗曲线突然掉到零,但调用量曲线是正常的。逐条查日志,发现新记录的prompt_tokens、completion_tokens全是 0。
原因:响应对象里的usage字段在某些情况下不存在。最常见的是缓存命中的请求,返回的 usage 结构可能与正常请求不同;另外异常分支里你拿不到 usage。如果代码里直接访问usage.prompt_tokens而不做判空,就会把None写入数据库。
解决:取 token 字段之前先判空,代码里用usage.prompt_tokens if usage else 0这种写法。缓存命中时的 usage 可能只包含部分字段,用getattr(usage, "prompt_cache_hit_tokens", 0)安全取值,不要假设每个字段都存在。
5.2 高并发写 SQLite 卡死:journal_mode 没设置,读和写互相堵
现象:批量任务开到 20 个线程后,看板查询经常转圈,甚至报database is locked,日志写入也开始丢数据。
原因:SQLite 默认的 rollback journal 模式下,写操作会独占数据库,读操作在写入期间会被阻塞。并发线程同时写库,锁冲突频繁,积累到一定量就报错。
解决:在连接后执行PRAGMA journal_mode=WAL;。WAL 模式下读和写可以并发,写之间依然互斥,但阻塞窗口小很多。配合批量写入,攒 100 条再 commit,锁竞争会进一步降低。另外注意 WAL 模式会生成同名的-wal和-shm文件,备份数据库时记得把这三个文件一起复制,只拷主文件会发现数据缺一段。
5.3 时区错位:UTC 入库、本地展示,日期分组全乱
现象:每天 8 点前产生的调用,在“按天”统计时被算到了前一天。想看下午高峰时段的延迟分布,怎么都不对。
原因:日志入库用的是datetime.now(timezone.utc),SQL 里用date()按天分组时直接对 UTC 时间做日期截断,导致本地时间和 UTC 时间错位。对于东八区来说,早上 8 点前的数据全被算进前一天。
解决:入库时同时存req_ts_utc和req_ts_local两个字段。排序和跨时区对账用 UTC,按天分组和画图用本地时间,SQL 里不再做任何时区转换,也就不存在转换失误的问题。这是最简单也最不容易出错的方案。
5.4 把 prompt 全文落库:数据库膨胀与敏感信息泄露一起找上门
现象:跑了半个月后看数据库文件,已经涨到好几个 GB。想删掉某些记录释放空间,发现没法按内容检索。
原因:很多人在埋点时图省事,直接把整个 messages 数组序列化成 JSON 存进去。Prompt 越长,数据库膨胀越快,而且代码里出现过的业务数据、用户输入全留在库里,一旦看板暴露到公网,这就是泄露源头。
解决:存prompt_preview字段,只保留最后一条 message 的前 200 个字符,够用且安全。另外建议在代码里加一条脱敏规则,把疑似密钥、手机号、身份证号等敏感信息替换成占位符后再入库。日志的价值在于行为分析,不在于内容存档,Prompt 全文需要单独的场景才值得存。
5.5 tool call 流程误判 API 超时:等待工具执行的耗时不算网络故障
现象:日志里超时请求变多,每一条的latency_ms都在 50 秒以上。但去 DeepSeek 开放平台看,这些请求实际上都成功了,模型侧耗时并不高。
原因:在 function calling / tool call 场景下,模型生成完工具调用指令后会等待外部工具执行结果。如果你在业务代码里把工具执行过程也算在请求总耗时里,latency_ms就会被工具执行时间污染。发布后看到大量“超时”记录,其实不是 API 慢,而是你的工具慢。
解决:统计延迟时分两种口径。latency_ms记录从发出请求到拿到完整响应的总时长,这是用户实际感知的耗时;如果要评估 DeepSeek 服务本身的响应能力,应该单独统计“从发出请求到收到第一个响应字节”的时间,需要自定义 HTTP 传输层来实现。在日志里给 tool call 类请求加一个has_tool_calls标记,统计 P95 时把这类请求单独分组,就不会再被误报干扰判断。
6. 从看板到动作:阈值告警、成本预测与多模型横向对比
看板只是第一步,日志数据的价值在于驱动决策和行动。这套体系稳定跑起来后,值得加上三个进阶用法。
告警是最先值得做的。写一个独立脚本,每分钟查一次最近 5 分钟的错误率,超过阈值就通过企业微信或邮件推消息。注意窗口要留够样本量,调用量太小时错误率会被放大。
# alert.py import sqlite3 import requests def check_error_rate(): conn = sqlite3.connect("deepseek_logs.db") row = conn.execute(""" SELECT COUNT(*) AS total, SUM(CASE WHEN status = 'error' THEN 1 ELSE 0 END) AS errors FROM deepseek_calls WHERE req_ts_local >= datetime('now', '-5 minutes', 'localtime'); """).fetchone() conn.close() total, errors = row[0], row[1] or 0 if total >= 50 and errors / total > 0.1: # 企业微信机器人 webhook 地址填你自己的 requests.post("https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY", json={"msgtype": "text", "text": {"content": f"DeepSeek 最近 5 分钟错误率 {errors/total*100:.1f}%"}}) if __name__ == "__main__": check_error_rate()成本预测则用近 7 天的日均消耗乘上预估天数,再留 30% 的余量。当曲线显示某个模型消耗增长很快时,提前和业务方对齐预算。
多模型对比是我个人最推荐做的实验。选一组代表性 prompt,分别用 deepseek-chat 和 deepseek-reasoner 各跑多次,统计 P95 延迟、token 消耗、错误率和输出质量。结果经常能推翻“贵的一定更好”的印象。
这套日志体系的核心原则其实很简单:先落盘,再上报;先有数据,再谈分析。之前我吃过没落盘的亏,日志只打到标准输出,容器一重启一个月的趋势线全没了,重新攒数据费了不少时间。从那以后所有调用日志一律先落 SQLite,后续推送到任何分析系统都不是问题。做监控这事,最怕的不是技术难,而是数据在关键时候断档。希望这篇笔记帮到你。
本文还有配套的精品资源,点击获取