长 prompt 里藏着账单。把同一份代码库说明或同一篇文档反复塞进请求,如果每次都按“未命中”计费,成本会悄悄翻倍。这篇教程做三件事:把 DeepSeek V4.1 Flash 的 API 接通、把缓存命中率跑出来、用同一批编码与检索任务算出“命中前后”的延迟差和账单价差。
文中模型名、单价、缓存有效期这类会变化的信息,一律写成从环境变量或配置里读,具体数值以官方文档与价格页当前版本为准。
适用场景
适合正在把代码助手、内部知识库问答接进 IDE 插件或工单系统的团队:请求里有大段稳定的AI 词典:系统提示词">系统提示词、代码上下文或文档原文,重复部分多;想先量出缓存能不能命中、命中后延迟和账单差多少,再决定是否上线。也适合个人开发者做一次成本摸底。
环境与前置条件
- 操作系统:Linux、macOS、Windows(建议用 WSL)均可。
- 运行时:Python 3.9 及以上,具体以官方 SDK 要求为准;DeepSeek 提供 OpenAI 兼容接口,直接用
openai官方 SDK 即可。 - 硬件:纯云端调用,不需要 GPU。内存 4GB 起,磁盘预留 1GB 存放任务样本、请求日志和报表。
- 账号:已开通 API Key,账户余额充足。
- 网络:能访问官方文档给出的 API 域名,公司网络需放行。
- 物料:准备 10~30 条编码任务(改 bug、补单测、解释报错)和 5~10 条检索任务(基于同一份长文档提问),两轮实验要跑完全相同的集合。
分步骤部署
步骤 1:建虚拟环境并安装 SDK
```bash
python3 -m venv .venv
source .venv/bin/activate # Windows PowerShell: .venv\Scripts\Activate.ps1
pip install --upgrade openai
python -c "import openai; print(openai.__version__)"
```
这步把依赖隔离在项目内,避免污染系统 Python。能打印出版本号即成功,版本高低于本教程无影响。
步骤 2:用环境变量管理密钥与模型名
```bash
export DEEPSEEK_API_KEY="sk-你的密钥"
export DEEPSEEK_BASE_URL="https://api.deepseek.com" # 以官方文档给出的地址为准
export DEEPSEEK_MODEL="以官方文档当前模型名为准"
```
Windows PowerShell 用 $env:DEEPSEEK_API_KEY="sk-...",只对当前窗口生效。密钥不写进代码、不进 Git,项目根目录加一个 .gitignore 写入 .env 和 *.log。
步骤 3:最小连通性验证
新建 hello.py:
```python
import os
from openai import OpenAI
client = OpenAI(
api_key=os.environ["DEEPSEEK_API_KEY"],
base_url=os.environ["DEEPSEEK_BASE_URL"],
)
resp = client.chat.completions.create(
model=os.environ["DEEPSEEK_MODEL"],
messages=[{"role": "user", "content": "用一句话说明什么是缓存命中"}],
max_tokens=64,
temperature=0,
)
print(resp.choices[0].message.content)
print(resp.usage)
```
```bash
python hello.py
```
成功标志:先输出一段中文回答,再打印 usage 对象,里面能看到 prompt_tokens、completion_tokens,部分模型还会带回缓存相关字段。若 usage 里已有类似 prompt_cache_hit_tokens 的字段,说明该模型会回报缓存情况,后面的统计就靠它;字段名以官方文档为准。
步骤 4:准备两套任务样本
tasks/coding.jsonl,每行一条:
```json
{"id": "c1", "question": "把这段函数改成用字典统计词频,并保证空输入不报错:<粘贴代码>"}
{"id": "c2", "question": "这个 TypeError 出现在第 12 行,给出原因和最小修改:<粘贴报错>"}
{"id": "c3", "question": "为 calculate_total 补三个 pytest 用例,覆盖边界值"}
```
tasks/retrieval.jsonl 同理,问题都围绕同一份长文档。关键点:两套任务共享各自固定的长前缀,这是缓存能否命中的前提。
步骤 5:把“稳定前缀”做厚,这是命中率的关键
缓存按前缀匹配,前缀里任何一处字面差异都会让后面的内容全部变成未命中。做法:
1. 新建 prompts/system_prefix.txt,放固定的角色说明、编码规范、整段代码库上下文或整篇文档原文。
2. 每条请求的 messages 结构固定为:system = 上述长前缀,user = 本轮的问题。
3. 前缀里禁止出现时间戳、随机 ID、用户名、datetime.now()。
4. 同一前缀的换行、空格、标点必须逐字一致,不要用格式化工具重排。
5. 前缀不要太短,太碎的内容难以形成稳定匹配块;具体匹配粒度以官方文档为准。
步骤 6:写实测脚本,跑两轮同任务
新建 bench.py:
```python
import os, json, time
from openai import OpenAI
client = OpenAI(
api_key=os.environ["DEEPSEEK_API_KEY"],
base_url=os.environ["DEEPSEEK_BASE_URL"],
timeout=180,
max_retries=3,
)
MODEL = os.environ["DEEPSEEK_MODEL"]
PREFIX = open("prompts/system_prefix.txt", encoding="utf-8").read()
元 / 百万 tokens,填官方价格页当前值
P_MISS, P_HIT, P_OUT = 0.0, 0.0, 0.0
def cost(hit, miss, out):
return hit / 1e6 * P_HIT + miss / 1e6 * P_MISS + out / 1e6 * P_OUT
def one_call(question, max_tokens=800):
t0 = time.perf_counter()
ttft, parts, usage = None, [], None
stream = client.chat.completions.create(
model=MODEL,
messages=[
{"role": "system", "content": PREFIX},
{"role": "user", "content": question},
],
temperature=0,
max_tokens=max_tokens,
stream=True,
stream_options={"include_usage": True},
)
for chunk in stream:
if chunk.choices and chunk.choices[0].delta.content:
if ttft is None:
ttft = time.perf_counter() - t0
parts.append(chunk.choices[0].delta.content)
if getattr(chunk, "usage", None):
usage = chunk.usage
u = usage.model_dump() if usage is not None and hasattr(usage, "model_dump") else (usage or {})
hit = u.get("prompt_cache_hit_tokens", 0) or 0
miss = u.get("prompt_cache_miss_tokens", u.get("prompt_tokens", 0)) or 0
out = u.get("completion_tokens", 0) or 0
return {
"ttft": ttft,
"latency": time.perf_counter() - t0,
"hit": hit, "miss": miss, "out": out,
"cost": cost(hit, miss, out),
"answer": "".join(parts),
}
def run(path, tag):
rows = []
for line in open(path, encoding="utf-8"):
task = json.loads(line)
r = one_call(task["question"])
r["id"], r["tag"] = task["id"], tag
rows.append(r)
print(f'{tag} {task["id"]} ttft={r["ttft"]:.2f}s total={r["latency"]:.2f}s '
f'hit={r["hit"]} miss={r["miss"]} out={r["out"]}')
json.dump(rows, open(f"result_{tag}.json", "w", encoding="utf-8"), ensure_ascii=False, indent=2)
return rows
def summary(rows, name):
hit = sum(r["hit"] for r in rows)
miss = sum(r["miss"] for r in rows)
print(f'{name}: 命中率={hit / max(hit + miss, 1):.1%} '
f'总成本={sum(r["cost"] for r in rows):.4f} 元')
```
stream_options={"include_usage": True} 用于在流式模式下拿到完整 usage;若接口报参数不支持,去掉该参数并改用非流式调用统计,其余逻辑不变。
执行两轮:
```bash
python -c "from bench import run, summary; r=run('tasks/coding.jsonl','code_r1'); summary(r,'编码第1轮')"
python -c "from bench import run, summary; r=run('tasks/coding.jsonl','code_r2'); summary(r,'编码第2轮')"
```
第一轮负责把前缀写进缓存,第二轮观察复用效果。检索任务同样各跑两轮。生产环境里这个“第二轮”是持续发生的。
步骤 7:出对比表
把 result_code_r1.json、result_code_r2.json 读进一份表格,按任务 id 对齐,记录四列:首 token 延迟、总延迟、命中 tokens、单条成本。观察三件事:
- 第二轮
hit是否明显大于 0; - 第二轮首 token 延迟是否下降(省掉的正是处理未命中前缀的时间);
- 两轮成本差额是否等于“命中 tokens ×(未命中单价 − 命中单价)”,单位换算别忘了。
如果命中率一直是 0,优先怀疑前缀被改动,而不是接口有问题。
步骤 8:按实测结果做降本配置
- 前缀稳定化:把系统提示词、代码库摘要、文档原文固化成文件,版本改动才更新,改动即视为新前缀。
- 精简上下文:检索任务的 top-k 从大到小试,命中率不掉的前提下减少未命中 tokens。
- 收紧
max_tokens:按任务类型分档,解释类 300、生成代码类 800。 - 固定
temperature=0:便于横向对比,也减少无效输出。 - 本地结果缓存:完全相同的请求先在应用层查一遍,连 API 都不必调。
- 错峰与批处理:非实时任务放到调用低峰,配合重试队列。
验证部署是否成功
1. python hello.py 能返回中文回答且打印出 usage,说明鉴权与网络通。
2. python -c "from bench import run, summary; summary(run('tasks/coding.jsonl','check'),'校验轮')" 能完整跑完并输出“命中率”行,说明脚本、任务集、前缀文件三者都对。
3. 连续两次执行上述命令,第二次输出里 hit= 的值应大于 0;若为 0,检查 prompts/system_prefix.txt 是否被改动或长度过短。
4. 打开 result_check.json,逐条核对 hit + miss 与 prompt_tokens 是否吻合,吻合说明统计口径正确。
常见报错与解决
报错:AuthenticationError: 401 - Authentication Fails
原因:密钥未加载、复制时带了空格,或环境变量名拼错。
解决:echo ${DEEPSEEK_API_KEY:0:6} 确认前缀存在;重新 export 后在同一终端重跑;确认没有把 .env 里的引号一起写进变量值。
报错:402 - Insufficient Balance
原因:账户余额不足或额度未生效。
解决:登录平台核对余额与用量页,充值后重试;若刚创建 Key,确认它属于当前账户。
报错:429 - Rate Limit Reached
原因:并发或单位时间请求数超过限额。
解决:openai 客户端已内置重试,可显式设置 max_retries=5;同时把任务串行化,或加 1~2 秒间隔;批量任务改成队列消费。
报错:400 - Model Not Exist 或模型名报错
原因:DEEPSEEK_MODEL 填了控制台展示名而非接口模型名,或该模型未对当前账户开放。
解决:以官方文档当前模型名为准重新 export,重启终端后再跑 hello.py。
报错:APIConnectionError / 请求超时
原因:代理、DNS 或 base_url 写错(多了 /v1 或少了斜杠)。
解决:先 curl -I $DEEPSEEK_BASE_URL 看能否连通;校对 base_url 与官方文档一致;公司网络确认已放行。
现象:usage 里没有缓存字段,或命中率始终为 0
原因:模型未回报该字段,或前缀每次都在变。
解决:打印一次完整 usage 确认字段名;用 diff 比较两次请求实际发出的 system 内容是否逐字相同。
后续维护
备份:prompts/、tasks/、result_*.json 一并纳入 Git 或对象存储,前缀文件带版本号,改一次记一次,方便回溯“哪次改动导致命中率掉下来”。
升级:pip install --upgrade openai 前先在分支上跑一遍 bench.py,对比升级前后命中率与延迟;模型名、接口参数以官方文档当前版本为准,不要照抄半年前的示例。
日志与监控:每次调用落一条结构化日志,至少含 task_id、prompt_tokens、缓存命中/未命中 tokens、completion_tokens、首 token 延迟、总延迟、估算成本。按天汇总命中率与单位任务成本,命中率环比下滑超过阈值就告警。
账单核对:用脚本汇总的估算成本与平台账单页按周对账,差异持续偏大时检查单价配置是否过期。密钥按季度轮换,离职或项目结束时及时吊销。
