企业官网建设流程全解析

简介：一份面向零基础的DeepSeek API监控实践指南，以28页篇幅系统讲解从调用日志采集到可视化看板落地的完整链路。文档从API监控基础概念切入，逐步演示申请DeepSeek API权限、获取调用日志，完成JSON/CSV格式日志的缺失值处理与错误值清洗，再提取响应时间、错误率、调用频率、吞吐量等核心指标，并通过时间序列分析与异常检测辅助排障。可视化部分基于ECharts设计了调用频率柱状图、平均响应时间折线图和鼠标悬停提示等交互功能，同时涵盖Flask后端、Nginx反向代理及云服务器部署步骤，便于读者在实际项目中直接复用。资源为单个PDF文档，大小1.89MB，目录完整、图表清晰，零基础读者可按章节跟随操作。已有115人学习下载，适合希望掌握AI服务可观测性、提升接口稳定性排查效率的开发者与运维人员。

1. 给 DeepSeek 调用做 API 监控：先看这份日志和看板方案解决什么问题

做 LLM 应用的人大概都经历过这种时刻：联调时请求跑得通、返回也正常，就以为万事大吉，结果上线第二天发现账单翻了三倍，或者半夜一个批量任务把上下文窗口塞满、连续重试把成本烧穿，等早上看到消息已经晚了。API 监控这件事，对普通 HTTP 接口是锦上添花，对 DeepSeek 这类大模型接口却是刚需——因为调用成本不只看次数，还看 token 消耗、模型档位、缓存命中和推理时长。这篇笔记要讲清楚的是：零基础怎么把 DeepSeek 调用日志从“埋点到入库再到看板”完整跑通，核心产物是一份可直接落地的日志结构、入库脚本和可视化看板代码。读完你会知道日志该记哪些字段、为什么第一版别上 ELK、成本为什么永远对不上账单，以及延迟变慢时怎么定位是网络问题还是模型本身慢。适合正在做 LLM 应用但还没建监控、或者建了但数据对不上的读者，也适合刚接触 DeepSeek API 想系统学习的人。

2. 数据从哪来：给 DeepSeek 调用日志设计一份能长期用的结构

2.1 先确认日志来源：业务代码、网关转发、开放平台后台，三条路选一条

做监控的第一步不是写代码，是搞清楚日志从哪条链路来。我见过不少人在这一步走弯路：跑去 DeepSeek 开放平台后台拉数据，发现只有账单汇总，根本没有逐请求明细，压根撑不起看板。实际上，能作为日志数据源的无非三条路。

第一条是自己写的业务代码里埋点。这是最可控的方式，请求参数、响应体、延迟、token 数全在手里，想记什么记什么。代价是得改代码，但改动量很小，一般加一个装饰器或一个日志函数就够了。

第二条是走网关转发。不少团队用 one-api、new-api 这类开源网关统一代理各家模型，或者用 ccswitch 这类工具在本地做模型路由切换。这类网关天然会记录每个请求的模型、token、耗时和费用，日志是现成的。如果你团队已经有网关，直接消费网关日志就好，比自己埋点省事。代价是字段由网关决定，不一定有你想要的 prompt 内容或业务标识。

第三条是 DeepSeek 开放平台后台，只有费用汇总和调用量曲线，没有请求明细，做不了下钻分析。它的价值只在于月底对账，不适合作为看板数据源。

零基础第一版我建议选第一条：自己有代码、有日志，理解整个链路后再决定要不要对接网关。如果你只是给 vscode 里的编码助手或某个现成工具接 DeepSeek，那就没有自己的业务代码可埋，这时候第二条路更实际——把你调用模型的那一层抽象出来，在统一出口记录。至于“deepseek api 如何调用”“本地部署 deepseek”这类搜索说明大家都在往这个方向走，不管你是用官方 API 还是把模型部署在自己服务器上，日志结构是同一套，后面讲字段。

2.2 统一字段设计：一条完整的 DeepSeek 调用日志该记什么

字段设计是整个监控系统的地基。地基歪了，后面看板、告警、成本分析全都会跟着歪。设计原则只有一条：把“这次调用是谁、在什么时间、花了多少 token、用了多久、结果如何”这几个问题一次性回答清楚。

先看一个标准的日志记录长什么样，这是 JSON 格式，方便后续解析：

{ "ts": "2025-06-11T08:23:15.482Z", "request_id": "a3f8c2e1-9b47-4d6e-8c1a-2f5e6d7a9c01", "app": "rag-service", "user_id": "u_10234", "model": "deepseek-chat", "prompt_tokens": 1523, "completion_tokens": 486, "total_tokens": 2009, "cache_hit_tokens": 930, "cache_miss_tokens": 593, "thinking_tokens": 0, "latency_ms": 2841, "ttft_ms": 620, "http_status": 200, "error_type": "", "cost_cny": 0.006922, "stream": true, "attempt": 1, "parent_request_id": "" }

字段说明不需要记完，但有几个必须理解，因为它们直接决定你后续能不能对上账。prompt_tokens是输入 token 数，completion_tokens是输出 token 数，total_tokens是两者之和。cache_hit_tokens和cache_miss_tokens是 DeepSeek 的上下文缓存计费字段，命中缓存的 token 单价远低于未命中的，这俩字段不记，成本一定算不准。thinking_tokens只有 deepseek-reasoner 这类推理模型会有，它算在输出费用里，但不能和普通输出 token 混着看，否则你没法回答“这个模型把预算花在思考还是生成上”这个问题。

ttft_ms是首 token 延迟，也就是从发起请求到收到第一个 token 的时间，它决定了用户体验的上限。latency_ms是完整请求总耗时，流式请求下它包含生成整段内容的时间。attempt和parent_request_id是给重试场景用的，一次业务请求可能因为超时重试了三次，没有这两个字段，你统计“调用次数”时会重复计数，统计“业务成功率”时会误判。

2.3 埋点怎么写：一个能直接抄的 Python 日志装饰器

字段定了，接下来就是埋点。常见做法是写一个装饰器或包装函数，把每次调用包起来，自动记录上述信息。我这里给一个最小可用的 Python 示例，基于 OpenAI SDK 兼容模式调用 DeepSeek，接口格式可以直接复用。

import json import time import uuid import logging from functools import wraps logger = logging.getLogger("deepseek_monitor") logger.setLevel(logging.INFO) handler = logging.FileHandler("deepseek_call.log", encoding="utf-8") handler.setFormatter(logging.Formatter("%(message)s")) logger.addHandler(handler) def monitor_deepseek_call(func): @wraps(func) def wrapper(*args, **kwargs): request_id = str(uuid.uuid4()) record = { "ts": time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime()) + ".000Z", "request_id": request_id, "app": kwargs.get("app_name", "default"), "model": kwargs.get("model", "deepseek-chat"), "stream": kwargs.get("stream", False), "attempt": kwargs.get("attempt", 1), "parent_request_id": kwargs.get("parent_request_id", ""), "http_status": 0, "error_type": "", "ttft_ms": 0, } start = time.time() try: resp = func(*args, **kwargs) record["latency_ms"] = int((time.time() - start) * 1000) record["ttft_ms"] = getattr(resp, "ttft_ms", 0) if hasattr(resp, "usage") and resp.usage: usage = resp.usage record["prompt_tokens"] = usage.prompt_tokens record["completion_tokens"] = usage.completion_tokens record["total_tokens"] = usage.total_tokens record["cache_hit_tokens"] = getattr(usage, "prompt_cache_hit_tokens", 0) record["cache_miss_tokens"] = getattr(usage, "prompt_cache_miss_tokens", 0) record["thinking_tokens"] = getattr(usage, "completion_tokens_details", None).thinking_tokens if getattr(usage, "completion_tokens_details", None) else 0 record["cost_cny"] = estimate_cost(record) record["http_status"] = 200 return resp except Exception as e: record["latency_ms"] = int((time.time() - start) * 1000) record["http_status"] = getattr(e, "status_code", 0) record["error_type"] = getattr(e, "type", type(e).__name__) logger.info(json.dumps(record, ensure_ascii=False)) raise finally: if record.get("total_tokens", 0) > 0 or record["http_status"] != 0: logger.info(json.dumps(record, ensure_ascii=False)) return wrapper

这个装饰器的逻辑是：调用前生成request_id和基础信息，调用成功后把 usage 里的 token 数据补齐，调用失败则记下 HTTP 状态码和错误类型，最后统一落一行 JSON 到日志文件。注意finally块里的判断是为了避免重复写日志——成功时在 try 里已经写了一次，失败时在 except 里写了一次，这个判断保证只写一次。

有两个参数必须解释。attempt和parent_request_id是给重试场景用的：如果外层有重试逻辑，每次重试都把 attempt 加 1，父请求 ID 保持一致，这样看板里就能区分“请求数”和“业务调用数”。cost_cny那行调用的estimate_cost函数需要你自己实现，按 DeepSeek 官方计价把 token 换算成金额，这部分后面第 4 章专门讲。

埋点层说完了，再补一句关于“deepseek harness”这类编排工具的提醒。如果你用 harness 这类多智能体编排框架，它不是单次调用，而是多个模型调用串成一条链路。这种情况下建议在 harness 的每一层调用入口都打同一条parent_request_id，否则你在看板里只能看到零散的单次调用，看不到整条链路的成本。

3. 存储与看板：用 SQLite + Streamlit 搭第一版可视化

3.1 为什么零基础第一版不推荐上 ELK 和 ClickHouse

日志有了，下一步是存起来并展示。这里我先泼一盆冷水：如果你不是每天几百万次调用，第一版别上 ELK，也别上 ClickHouse。ELK 三件套部署、调优、吃内存，一套下来小团队光是维护就够呛；ClickHouse 确实快，但列式存储的冷热分层、分区键设计、副本配置，每一项都是学习成本。

我的建议很朴素：SQLite + Streamlit。个人项目或小团队一天几千到几万次调用，SQLite 完全扛得住，单文件备份也方便；Streamlit 写一个看板脚本不到一百行，改完保存浏览器自动刷新，零前端成本。等哪天真的一天百万级调用了，再把 SQLite 换成 PostgreSQL，看板代码基本不用动，因为查询用的还是同样几条 SQL。这套“先轻后重”的路线，是无数小项目验证过的稳妥路径，别一上来就建大数据平台。

整条管道就是：业务代码写 JSON 日志文件 → 一个入库脚本定时把日志解析进 SQLite → Streamlit 读 SQLite 渲染看板。每个环节都可以单独验证，出了问题也容易定位。

3.2 初始化数据库：建表语句与入库脚本

先建表。注意时间字段统一存 UTC 时间戳的文本形式，展示时再转本地时区，不然不同机器、不同时区混在一起，看板上的时间线必乱。

CREATE TABLE IF NOT EXISTS deepseek_calls ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts TEXT NOT NULL, request_id TEXT UNIQUE NOT NULL, app TEXT DEFAULT 'default', model TEXT DEFAULT 'deepseek-chat', prompt_tokens INTEGER DEFAULT 0, completion_tokens INTEGER DEFAULT 0, total_tokens INTEGER DEFAULT 0, cache_hit_tokens INTEGER DEFAULT 0, cache_miss_tokens INTEGER DEFAULT 0, thinking_tokens INTEGER DEFAULT 0, latency_ms INTEGER DEFAULT 0, ttft_ms INTEGER DEFAULT 0, http_status INTEGER DEFAULT 0, error_type TEXT DEFAULT '', cost_cny REAL DEFAULT 0, stream INTEGER DEFAULT 0, attempt INTEGER DEFAULT 1, parent_request_id TEXT DEFAULT '' ); CREATE INDEX IF NOT EXISTS idx_ts ON deepseek_calls (ts); CREATE INDEX IF NOT EXISTS idx_app_ts ON deepseek_calls (app, ts);

request_id加 UNIQUE 约束是有意的，入库时重复的请求会被自动跳过，这在处理日志文件重读时非常好用。索引加在ts和(app, ts)上，覆盖了绝大多数查询场景——按时间范围看全量、按业务看某个应用的趋势。

入库脚本不需要复杂，逐行读日志文件，解析 JSON 后插入。核心逻辑如下：

import json import sqlite3 from pathlib import Path DB_PATH = "deepseek_monitor.db" LOG_PATH = Path("deepseek_call.log") def ingest_logs(log_file: str, db_path: str) -> int: conn = sqlite3.connect(db_path) inserted = 0 with open(log_file, "r", encoding="utf-8") as f: for line in f: line = line.strip() if not line: continue try: rec = json.loads(line) except json.JSONDecodeError: continue try: conn.execute( """ INSERT OR IGNORE INTO deepseek_calls (ts, request_id, app, model, prompt_tokens, completion_tokens, total_tokens, cache_hit_tokens, cache_miss_tokens, thinking_tokens, latency_ms, ttft_ms, http_status, error_type, cost_cny, stream, attempt, parent_request_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( rec.get("ts"), rec.get("request_id"), rec.get("app", "default"), rec.get("model"), rec.get("prompt_tokens", 0), rec.get("completion_tokens", 0), rec.get("total_tokens", 0), rec.get("cache_hit_tokens", 0), rec.get("cache_miss_tokens", 0), rec.get("thinking_tokens", 0), rec.get("latency_ms", 0), rec.get("ttft_ms", 0), rec.get("http_status", 0), rec.get("error_type", ""), rec.get("cost_cny", 0.0), 1 if rec.get("stream") else 0, rec.get("attempt", 1), rec.get("parent_request_id", "") ), ) inserted += 1 except sqlite3.Error as e: print(f"insert failed: {e}") conn.commit() conn.close() return inserted if __name__ == "__main__": count = ingest_logs(str(LOG_PATH), DB_PATH) print(f"ingested {count} new records")

INSERT OR IGNORE配合request_id的唯一约束，是幂等入库的关键。日志文件被重复读、脚本被 crontab 重复跑，都不会产生脏数据。参数说明：LOG_PATH和DB_PATH改成你自己的路径；脚本可以直接用python ingest.py跑，也可以加一个--rescan参数读取历史文件。

入库跑完后验证一下数据：sqlite3 deepseek_monitor.db "SELECT COUNT(*), SUM(cost_cny) FROM deepseek_calls;"。如果数字和你的心理预期差距很大，八成是埋点字段没对上，先回第 2 章查字段名。

3.3 看板脚本：请求量、延迟、Token 消耗三个核心面板

看板我用 Streamlit 写。核心思路是右侧放时间范围选择，左侧三块面板分别对应请求量趋势、延迟分位数、Token 消耗与成本。脚本可以直接复制运行。

import sqlite3 import pandas as pd import streamlit as st st.set_page_config(page_title="DeepSeek API 监控", layout="wide") DB_PATH = "deepseek_monitor.db" @st.cache_data(ttl=60) def load_data(ts_start: str, ts_end: str) -> pd.DataFrame: conn = sqlite3.connect(DB_PATH) query = """ SELECT * FROM deepseek_calls WHERE ts >= ? AND ts <= ? """ df = pd.read_sql_query(query, conn, params=[ts_start, ts_end]) conn.close() return df st.sidebar.header("筛选条件") ts_start = st.sidebar.text_input("开始时间 (UTC)", value="2025-06-01T00:00:00Z") ts_end = st.sidebar.text_input("结束时间 (UTC)", value="2025-06-11T23:59:59Z") df = load_data(ts_start, ts_end) if df.empty: st.warning("当前时间范围内没有数据") st.stop() df["ts"] = pd.to_datetime(df["ts"], utc=True).dt.tz_convert("Asia/Shanghai") df["hour"] = df["ts"].dt.floor("h") col1, col2, col3 = st.columns(3) col1.metric("总调用次数", len(df)) col2.metric("总成本 (元)", f"{df['cost_cny'].sum():.2f}") col3.metric("总 Token 数", f"{df['total_tokens'].sum():,}") st.subheader("按小时请求量与成本趋势") hourly = df.groupby("hour").agg( count=("request_id", "count"), cost=("cost_cny", "sum"), tokens=("total_tokens", "sum"), ).reset_index() st.line_chart(hourly.set_index("hour")[["count", "cost"]]) st.subheader("延迟分位数 (P50 / P95 / P99)") lat = df["latency_ms"].quantile([0.5, 0.95, 0.99]).round(0) st.dataframe(lat) st.subheader("各模型 Token 消耗与成本对比") model_stats = df.groupby("model").agg( calls=("request_id", "count"), total_tokens=("total_tokens", "sum"), cost=("cost_cny", "sum"), ).reset_index().sort_values("cost", ascending=False) st.dataframe(model_stats)

这个脚本的逻辑分四段：load_data带 60 秒缓存，避免每次交互都查库；时区统一在展示层转成北京时间，原始数据保持 UTC 不动；按小时聚合是趋势图的数据基础；延迟分位数直接调 pandas 的 quantile。参数说明：ttl=60表示结果缓存 60 秒，改小了看板刷新更快但查库更频繁；时间范围默认写死在侧边栏，后续可以换成日期组件。

运行方式是在终端执行streamlit run dashboard.py，浏览器会打开http://localhost:8501。日志文件十分钟前刚入库的数据，刷新后就能看到。

3.4 扩展替换：把 SQLite 换成 PostgreSQL 只需改一处连接串

很多读者看到这里会问：SQLite 够用，但公司要求统一用 PostgreSQL，怎么办？答案很简单：把连接和 SQL 方言换掉，视图逻辑不用动。

PostgreSQL 的建表语句基本可以直接拿来用，把AUTOINCREMENT换成SERIAL或IDENTITY即可。入库脚本里sqlite3.connect换成psycopg2.connect，参数占位符从?换成%s。Streamlit 看板里sqlite3.connect换成create_engine("postgresql://user:pass@host:5432/dbname")，其余代码原样跑通。

这个替换成本之所以低，是因为整套架构够简单。数据量上去之后，唯一要改的是在ts上加分区表，但那已经是另一个量级的话题了。现在先把看板跑起来，让数据说话。

4. 从看板到告警：成本与异常调用怎么主动暴露

4.1 成本估算：按缓存命中、thinking tokens 拆分计价

看板上有 Token 数不等于能对上成本，因为你还要搞清楚“这次调用到底花了多少钱”。DeepSeek 的计价有几个坑：缓存命中的输入 token 价格远低于未命中；deepseek-reasoner 的思考 token 虽然算在输出里，但和普通输出 token 是同一个单价，只是你需要单独记下来才能分析“这钱花在思考还是回答”；工具调用或结构化输出场景下，输出 token 数包含了这些附加内容，也要按照输出单价计费。

下面是一个成本估算函数，按 per-request 粒度计算，结果写入日志的cost_cny字段：

PRICING = { "deepseek-chat": { "input_miss": 0.002, # 元/千 token "input_hit": 0.0005, "output": 0.008, "thinking": 0.008, }, "deepseek-reasoner": { "input_miss": 0.004, "input_hit": 0.001, "output": 0.016, "thinking": 0.016, }, } def estimate_cost(record: dict) -> float: model = record.get("model", "deepseek-chat") pricing = PRICING.get(model, PRICING["deepseek-chat"]) miss = record.get("cache_miss_tokens", 0) hit = record.get("cache_hit_tokens", 0) completion = record.get("completion_tokens", 0) cost = (miss * pricing["input_miss"] + hit * pricing["input_hit"] + completion * pricing["output"]) / 1000.0 return round(cost, 6)

注意这里的单位换算：官方价格通常是“每百万 token X 元”，代码里换算成了“每千 token X 元”，所以input_miss的 0.002 元/千 token 等于 2 元/百万 token。以你账号后台看到的实时价格为准，这里只是演示结构。estimate_cost只算输入命中和未命中加输出，thinking_tokens包含在completion_tokens里，不需要额外加，但要用 SQL 单独统计它方便分析。

4.2 三条实用告警：成本突增、错误率、P95 延迟

看板是被动的，告警才是主动的。告警的粒度也不必太细，三条就够用，从成本、可用性、体验三个角度覆盖核心风险。

成本突增的判定逻辑：按小时统计cost_cny，和过去 7 天同一小时段的平均值比较，超过 2 倍就触发。这个规则能抓住 token 泄露、死循环重试、模型误配（把 deepseek-chat 换成 deepseek-reasoner）等大多数成本事故。

错误率告警：统计最近 15 分钟内http_status >= 400或error_type != ''的占比，超过 10% 就触发。这个规则覆盖模型过载、超时、鉴权失败等可用性问题。

P95 延迟告警：计算最近 15 分钟latency_ms的 P95，和过去 24 小时同期 P95 比较，超过 1.5 倍就触发。这能发现模型服务变慢但还没到报错的阶段。

4.3 告警通知怎么接：钉钉 Webhook 脚本

通知渠道我建议用钉钉群机器人，理由就一条：小团队里它是零成本、配置最快的方式。企业微信群机器人同理，逻辑完全一样，只改 webhook 地址。

import requests import json import sqlite3 import pandas as pd from datetime import datetime, timedelta DB_PATH = "deepseek_monitor.db" WEBHOOK_URL = "https://oapi.dingtalk.com/robot/send?access_token=YOUR_TOKEN" ALERT_KEYWORD = "DeepSeek告警" # 如果机器人设置了关键词校验，必须带上 def check_cost_surge(hours: int = 1, threshold: float = 2.0) -> str | None: conn = sqlite3.connect(DB_PATH) now = datetime.utcnow() recent_start = now - timedelta(hours=hours) baseline_start = now - timedelta(days=7, hours=hours) recent_cost = pd.read_sql_query( "SELECT COALESCE(SUM(cost_cny), 0) AS cost FROM deepseek_calls WHERE ts >= ? AND ts < ?", conn, params=[recent_start.isoformat() + "Z", now.isoformat() + "Z"], )["cost"].iloc[0] baseline_cost = pd.read_sql_query( "SELECT COALESCE(SUM(cost_cny), 0) AS cost FROM deepseek_calls WHERE ts >= ? AND ts < ?", conn, params=[baseline_start.isoformat() + "Z", recent_start.isoformat() + "Z"], )["cost"].iloc[0] / 7.0 conn.close() if baseline_cost > 0 and recent_cost > baseline_cost * threshold: return f"最近 {hours} 小时成本 {recent_cost:.2f} 元，是基线 {baseline_cost:.2f} 元的 {recent_cost / baseline_cost:.1f} 倍" return None def send_alert(message: str): payload = {"msgtype": "text", "text": {"content": f"{ALERT_KEYWORD}\n{message}"}} requests.post(WEBHOOK_URL, json=payload, timeout=5) if __name__ == "__main__": alert = check_cost_surge() if alert: send_alert(alert)

这个脚本逻辑很简单：check_cost_surge查最近 1 小时的成本和过去 7 天同一时段的日均成本做比较，超过阈值就返回告警文案；send_alert把文案发给钉钉机器人。WEBHOOK_URL里的YOUR_TOKEN替换成你自己的机器人 token，机器人安全设置里如果配了关键词，ALERT_KEYWORD就必须带上，否则消息被丢弃。延迟和错误率的检查函数结构完全相同，只是换 SQL 条件和阈值。把这三个函数放到同一个脚本里，用 crontab 每五分钟跑一次，告警链路就通了。

5. 避坑清单：成本对不上账、看板不刷新、统计重复，这类问题怎么排查

5.1 后台账单和你统计的成本差一大截

现象：看板里算出的成本和 DeepSeek 开放平台后台的账单对不上，差 30% 以上。原因基本出在三个地方：一是缓存命中 token 没单独计价，把命中和未命中的输入 token 都按全价算，成本偏高；二是 deepseek-reasoner 的思考 token 被忽略或重复计价；三是日志丢了请求，比如异步写日志时进程崩溃，数据没落盘。

解决：先对比“请求数”，再看“平均单次成本”。请求数对不上是采集链路问题，请求数对得上但金额不对是计价逻辑问题。把cache_hit_tokens和cache_miss_tokens分开统计，按官方计价规则拆开算，基本能解决。后台账单有延迟，一般 T+1 才稳定，当日对比有偏差是正常的。

5.2 日志文件里数据丢了一部分

现象：业务日志里明明每行都有记录，入库后统计的请求数比实际调用的少。原因多数是异步日志写入被进程 OOM 或其他异常打断，buffering设置为1行缓冲时，Python 进程异常退出会丢最后几行。另一种可能是多个进程同时写同一个日志文件，行被交叉写成了半行，JSON 解析失败被跳过。

解决：写日志改成同步写，或至少保证异常退出时有 flush。多进程写日志建议按进程号分文件，比如deepseek_call_{pid}.log，入库脚本统一读取。日志文件和数据库不在同一台机器时，先确认传输过程有没有丢行，我用wc -l对比源文件和入库脚本处理的记录数是常用手段。

5.3 看板数字过了十几分钟还不刷新

现象：日志文件里已经新增了数据，但 Streamlit 看板上的数字一直不变。原因多半是st.cache_data(ttl=60)的缓存生效，再加上浏览器端的 Streamlit 自动刷新不是实时的，两个因素叠加，看起来像“不刷新”。

解决：把ttl从 60 改成 10，确认代码里没有别的高级缓存。如果还是不行，按Shift + F5强制刷新浏览器缓存。另外确认入库脚本有没有跑成功，Streamlit 只负责展示，不会自动帮你入库——这个顺序很多人搞反。

5.4 入库脚本重跑一遍，统计数字翻倍

现象：日志文件没变，入库脚本手动跑了两次，看板上的请求数变成原来的两倍。原因是在没有request_id唯一约束的旧表结构里，INSERT变成了普通的追加插入。解决：重新建表并加上request_id TEXT UNIQUE，入库用INSERT OR IGNORE。已经脏掉的数据，按request_id去重后手工清理：

DELETE FROM deepseek_calls WHERE id NOT IN ( SELECT MIN(id) FROM deepseek_calls GROUP BY request_id );

这条 SQL 以request_id分组保留最小id，也就是最早入库的那条，其余重复记录删掉。注意先备份库文件再执行，request_id为空的记录会被归为一组，需要先用WHERE request_id != ''过滤。

5.5 重试机制导致“请求变大”

现象：上游因为超时自动重试了三次，看板里调用次数多了三倍，成本也虚高。原因是没有区分“业务请求”和“API 调用”。解决：在第 2 章的字段里，parent_request_id标识业务请求，attempt标识第几次重试。统计“业务请求数”时按parent_request_id去重：

SELECT COUNT(DISTINCT COALESCE(NULLIF(parent_request_id, ''), request_id)) FROM deepseek_calls WHERE ts >= ? AND ts <= ?;

这条 SQL 同时处理了“有父请求 ID”和“没有父请求 ID”的两种情况，没有父请求的请求本身就当成业务请求。成本统计不应该去重——每次重试都是真金白银的消耗，这正是需要在看板上暴露出来的事实。

6. 进阶：把延迟拆开来看——首 Token 时间与吞吐时间

看板搭好、告警通了之后，多数人会遇到一个更细的问题：看板上延迟变高了，但不知道是卡在哪一段。是网络到 DeepSeek 服务端的链路慢，还是模型生成太慢，还是流式返回的首包慢？这需要把一次调用的时间拆成两段：请求发起到收到第一个 token 的时间（首 Token 时间，TTFT），以及从第一个 token 到最后一个 token 的生成时间。前者主要反映网络和服务端排队压力，后者主要反映模型生成速度。

Streamlit 看板里可以加一个按小时的 TTFT 和生成耗时趋势图，但前提是你的日志里记录了两个字段：ttft_ms和latency_ms。生成耗时约等于latency_ms - ttft_ms。如果你用的是流式接口，ttft_ms在 SDK 里通常可以直接拿到；如果你是手工用 httpx 发请求，可以用一个计时器记录首字节时间：

import httpx import time start = time.perf_counter() with httpx.stream("POST", url, json=payload, timeout=None) as resp: first_byte = time.perf_counter() ttft_ms = (first_byte - start) * 1000 for chunk in resp.iter_bytes(): pass total_ms = (time.perf_counter() - start) * 1000

把这个ttft_ms塞进日志字段后，你就能回答“模型变慢了还是网络变慢了”这个经典问题。如果 TTFT 正常但生成时间变长，大概率模型负载高或 prompt 太长导致注意力计算量膨胀；如果 TTFT 变长，先查本机到服务端的网络延迟和重试率，再查是不是自己在请求头里加了多余的重试逻辑。

我现在的习惯是每次优化完 prompt 或切换模型档位，都会对比一下“成本/有效回答长度”这个比值。看板上的 P95 延迟和成本曲线帮我避过不少坑，但真正救我的往往是把一次请求拆开看的 TTFT 和生成耗时。如果你也遇到“接口偶尔有点慢但说不出慢在哪”，先把这个拆开，再做优化。希望帮到你。

本文还有配套的精品资源，点击获取