视频总结、思维导图、智能问答三大功能,都跑在同一套 API 上——DeepSeek 是产品真正”智能”起来的底座。选型阶段我把 GPT-4o、Claude 和 DeepSeek 都接进同一个中文视频场景对比跑了一周,最后 DeepSeek 留下来,主要是性价比和中文表现。下面把我从注册、基础对接、prompt 设计、流式优化到降级方案的完整落地过程写下来,包括线上踩过的坑和几个关键的优化点。
一、为什么选 DeepSeek
选型不是拍脑袋,三个模型放同一个中文视频总结场景实测了一周,关键维度如下:
| 维度 | DeepSeek | GPT-4o | Claude |
|---|---|---|---|
| 中文理解 | 优 | 良 | 良 |
| 价格 | 低(1/30 GPT) | 高 | 中 |
| 响应速度 | 快 | 中 | 中 |
| API 兼容 | OpenAI 兼容 | OpenAI | 自有 |
| 上下文 | 32k | 128k | 200k |
平台选 DeepSeek 关键理由:
- 中文视频总结场景,DeepSeek 表现优于 GPT-4;
- 价格低 30 倍,每日 1000 次调用仅 ¥30;
- 兼容 OpenAI SDK,集成简单。
结论很直接:中文视频场景 DeepSeek 的表现够用,价格是决定性因素,每天一千次调用的成本很低,GPT-4o 同量级至少高一个数量级。API 兼容 OpenAI 协议,迁移成本几乎为零,这也是我敢直接上生产的原因。
二、基础对接
对接本身没有学习成本,DeepSeek 接口完全兼容 OpenAI 协议,直接拿 OpenAI 的 Python SDK 换 base_url 就行:
pip install openai
from openai import AsyncOpenAI
client = AsyncOpenAI(
api_key=settings.DEEPSEEK_API_KEY,
base_url='https://api.deepseek.com'
)
密钥从环境变量读,不写进代码库。AsyncOpenAI 的异步客户端和我们的 FastAPI 异步链路能无缝配合,流式调用也原生支持。
三、AI 总结 Prompt
总结效果七分靠 prompt。第一版我只让模型”总结一下”,输出五花八门,前端解析逻辑写了一堆容错。后来我把输出格式完全固定,效果立刻稳下来:
SUMMARY_PROMPT = """你是一名专业的内容总结助手。请根据用户提供的视频字幕,生成结构化总结。
要求:
1. 用中文回答
2. 总结长度 200-500 字
3. 包含 3-5 个核心要点
4. 标注关键时间点(如「00:15 提到...」)
5. 保持客观,避免主观评价
输出格式(严格遵守):
【一句话总结】
用 30 字以内概括视频主旨。
【核心要点】
1. 要点 1
2. 要点 2
3. 要点 3
【关键时间点】
- 00:00 开场介绍
- 02:15 核心观点
- 05:30 总结陈词
"""
这个 prompt 的要点是格式约束放最前面,长度、要点数、时间点全部明确。输出稳定之后,前端解析基本不用容错,总结质量也一致了。
四、AI 总结实现
总结调用本身很常规,温度参数是唯一需要调试的地方:
async def summarize(self, subtitle: str) -> str:
response = await client.chat.completions.create(
model='deepseek-chat',
messages=[
{'role': 'system', 'content': SUMMARY_PROMPT},
{'role': 'user', 'content': subtitle[:4000]} # 限制 4k 字符
],
temperature=0.3, # 较低温度保证稳定性
max_tokens=1500
)
return response.choices[0].message.content
temperature 定在 0.3:总结类任务要稳定,压低随机性;生成思维导图这类偏创作的任务再单独调高。字幕限制在 4000 字符,是为了控制输入成本和首字延迟。
五、流式总结(首字延迟优化)
非流式接口要整块等 3-5 秒,页面干转圈。改成 stream 之后,首字几百毫秒就能到达:
async def stream_summarize(self, subtitle: str) -> AsyncIterator[str]:
stream = await client.chat.completions.create(
model='deepseek-chat',
messages=[
{'role': 'system', 'content': SUMMARY_PROMPT},
{'role': 'user', 'content': subtitle[:4000]}
],
stream=True,
temperature=0.3
)
async for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
yield delta
前端 EventSource / fetch SSE 逐字展示,打字机效果出来之后,用户等待焦虑明显下降,这个改动的性价比很高。
六、思维导图生成
总结之外,思维导图是第二个高频功能。让模型输出 Markdown 列表结构,前端用 markmap 渲染,实现路径比较直接:
MINDMAP_PROMPT = """根据用户提供的视频总结,生成 Markdown 思维导图。
要求:
1. 中心节点:视频主题
2. 二级节点:3-5 个主要话题
3. 三级节点:每个话题的关键点
4. 层级用 - 缩进表示
5. 用中文
示例输出格式:
# 视频主题
## 话题 1
- 要点 A
- 要点 B
## 话题 2
- 要点 C
- 要点 D
"""
生成逻辑和总结基本一样,换 prompt、温度调高一点:
async def generate_mindmap(self, summary: str) -> str:
response = await client.chat.completions.create(
model='deepseek-chat',
messages=[
{'role': 'system', 'content': MINDMAP_PROMPT},
{'role': 'user', 'content': summary}
],
temperature=0.5,
max_tokens=800
)
return response.choices[0].message.content
前端 markmap 渲染:
import { markmap } from 'markmap'
import { Transformer } from 'markmap-lib'
const transformer = new Transformer()
const { root } = transformer.transform(mindmapMarkdown)
markmap(document.getElementById('mindmap'), { data: root })
markmap 渲染有个坑:模型偶尔输出嵌套结构不规范的 Markdown,渲染会崩。我在入库前做了缩进规范化,之后没再出过问题。
七、智能问答
总结完之后用户一般会追问细节,这就是智能问答模块。核心原则是只答视频里有的内容,避免模型自由发挥:
QA_PROMPT = """你是一个视频内容问答助手。基于以下视频字幕回答用户问题。
要求:
1. 只基于视频内容回答
2. 不确定时回答「视频中未提到」
3. 引用具体时间点
4. 简洁,100 字以内
视频字幕:
{subtitle}
"""
async def qa(self, subtitle: str, question: str,
history: list = None) -> str:
messages = [
{'role': 'system', 'content': QA_PROMPT.format(subtitle=subtitle[:4000])}
]
# 加入历史
for h in (history or []):
messages.append({'role': h['role'], 'content': h['content']})
messages.append({'role': 'user', 'content': question})
response = await client.chat.completions.create(
model='deepseek-chat',
messages=messages,
temperature=0.3
)
return response.choices[0].message.content
问答模块带历史记录,支持多轮追问。prompt 明确要求”不确定就说视频中未提到”,实测能拦住大部分幻觉。
八、长视频分段总结
1 小时视频的字幕经常超过 8000 字符,单次请求塞不进去。我参考 map-reduce 的思路做分段总结:
async def summarize_long_video(self, subtitle_cues: list) -> str:
# 1) 分段(每段 4000 字符)
chunks = self._split_cues(subtitle_cues, max_chars=4000)
# 2) 每段并行总结
chunk_summaries = await asyncio.gather(*[
self.summarize(chunk) for chunk in chunks
])
# 3) 合并总结
combined = '\n\n'.join(chunk_summaries)
final = await self.summarize(combined)
return final
分段并行加合并总结,长视频的处理时间从线性增长变成可控。合并时如果分段总结有冲突,以最后一段为准,简单有效。
九、Token 控制与计费
线上跑起来之后,成本控制成了新问题。token 预估和计费逻辑我单独抽了一个模块:
import tiktoken
def count_tokens(text: str) -> int:
"""估算 token 数(DeepSeek 用类似 GPT 的分词)"""
encoding = tiktoken.encoding_for_model('gpt-3.5-turbo')
return len(encoding.encode(text))
# DeepSeek 计价示例(cache miss / cache hit):
# input: 1 元 / 1M tokens (cache miss)
# input: 0.1 元 / 1M tokens (cache hit)
# output: 2 元 / 1M tokens
PRICING = {
'input_miss': 1.0 / 1_000_000,
'input_hit': 0.1 / 1_000_000,
'output': 2.0 / 1_000_000
}
def calculate_cost(input_tokens: int, output_tokens: int,
cached: bool = False) -> float:
input_price = PRICING['input_hit' if cached else 'input_miss']
return (input_tokens * input_price +
output_tokens * PRICING['output'])
这个模块做两件事:调用前预估成本,调用后按实际用量记账。用户侧每日配额和成本侧统计都从它拿数据,prompt 缓存命中与否也在计费里体现。
十、限流与并发
DeepSeek 有 RPM/TPM 限制,免费额度很低,并发上来必须自己控速:
| 套餐 | RPM | TPM |
|---|---|---|
| 免费 | 60 | 1M |
| 付费 | 3000 | 5M |
付费档相对宽松,但我还是用信号量把并发压在 10 以内,留出余量给故障重试和突发流量:
deepseek_sem = asyncio.Semaphore(10) # 最多 10 并发
async def call_deepseek(self, messages):
async with deepseek_sem:
return await client.chat.completions.create(
model='deepseek-chat',
messages=messages
)
Semaphore 的写法有个注意点:要在调用函数内部获取信号量,而不是在循环外面包一层,否则并发控制形同虚设。
十一、错误重试
限流和网络抖动在线上是常态,接口调用必须带重试:
from openai import RateLimitError, APIError
async def call_with_retry(self, messages, max_retry=3):
for attempt in range(max_retry):
try:
return await client.chat.completions.create(
model='deepseek-chat', messages=messages
)
except RateLimitError:
wait = 2 ** attempt
log.warning(f'限流,等待 {wait}s')
await asyncio.sleep(wait)
except APIError as e:
if attempt == max_retry - 1:
raise
await asyncio.sleep(1)
raise Exception('DeepSeek 调用失败')
重试用指数退避,RateLimitError 和 APIError 分开处理。重试上限 3 次,再多就交给降级逻辑,避免雪崩。
十二、Prompt 缓存
DeepSeek 支持 prompt cache,同一个系统 prompt 第二次调用起,输入价格只有十分之一,这是白捡的省钱点:
# 同一个 system prompt 多次调用,第二次起成本 1/10
SUMMARY_PROMPT = "..."
# 第一次:cache miss,1 元/1M tokens
# 第二次起:cache hit,0.1 元/1M tokens
平台所有 AI 总结用同一 system prompt,自动享受 cache。注意 prompt 只要改一个字缓存就失效,所以改版要走灰度,别频繁动。
十三、监控与统计
成本要可控,前提是看得见。每次调用的成功失败、token 消耗、金额,全部上报指标:
metrics.counter('deepseek.call',
'type', 'summarize',
'result', 'success').increment()
metrics.histogram('deepseek.tokens',
'type', 'input').observe(input_tokens)
metrics.counter('deepseek.cost',
'amount', f'{cost:.4f}').increment(cost)
统计:
- 每日调用次数;
- 总 token 消耗;
- 总成本;
- 缓存命中率。
监控面板上重点盯三张图:调用成功率、每日成本、缓存命中率。缓存命中率掉下去,说明有人在改 prompt 或者换了 key。
十四、Token 限额保护
免费用户一天能消耗的 token 必须卡死,否则一个重度用户就能打穿成本预算:
class TokenLimitExceeded(Exception):
pass
async def summarize_with_limit(self, subtitle: str,
user: User) -> str:
# 检查用户配额
if user.daily_used >= user.daily_limit:
raise TokenLimitExceeded('已达每日限额')
# 预估 token
estimated_in = count_tokens(subtitle) + count_tokens(SUMMARY_PROMPT)
estimated_out = 1500
estimated_cost = calculate_cost(estimated_in, estimated_out)
if user.daily_used + estimated_cost > user.daily_limit:
raise TokenLimitExceeded('剩余额度不足')
# 调用
response = await self.call_with_retry([...])
# 实际计费
actual_cost = calculate_cost(
response.usage.prompt_tokens,
response.usage.completion_tokens,
response.usage.prompt_cache_hit_tokens > 0
)
user.daily_used += actual_cost
db.commit()
return response.choices[0].message.content
限额保护放在调用入口,预估加实际两步扣减。这个类的单测覆盖了所有边界:刚好够、超一点、完全不够,三种情况都要验证。
十五、Prompt 调优
prompt 会持续迭代,我把版本集中管理,方便回滚和 A/B:
PROMPTS = {
'summarize_v1': "你是一名内容总结...",
'summarize_v2': "你是专业总结助手,要求 1. ...",
'mindmap_v1': "...",
}
# 灰度切换
async def summarize_v2(self, subtitle):
return await self.call([
{'role': 'system', 'content': PROMPTS['summarize_v2']},
{'role': 'user', 'content': subtitle}
])
A/B 测试不同 prompt 效果。灰度切 prompt 我做了个简单开关:先放 10% 流量,看总结质量指标没掉再全量。对比时输出格式必须完全一致,否则没法量化。
十六、降级方案
再稳的供应商也有挂的时候,线上不能因为一家厂商抽风就全站不可用,我留了备用模型:
PRIMARY = 'deepseek-chat'
FALLBACK = 'gpt-4o-mini' # 备用
async def call_with_fallback(self, messages):
for model in [PRIMARY, FALLBACK]:
try:
client = self._get_client(model)
return await client.chat.completions.create(
model=model, messages=messages
)
except Exception as e:
log.warning(f'{model} 失败: {e}')
raise Exception('所有模型都失败')
主备切换是自动的,主模型连续失败两次就走备用。切换有日志和指标,事后能复盘是哪家的问题。
十七、与 Whisper 协作
字幕走 Whisper 转录时,ASR 的错别字会直接污染 prompt,这个问题我在问答模块的 prompt 里做了兜底:
QA_PROMPT = """你是一个视频内容问答助手。
注意:
- 字幕由 ASR 自动生成,可能有少量错别字
- 请根据上下文自动纠错
- 例如「基于」可能写作「机于」,「程序」可能写作「诚徐」
- 回答时使用正确的中文
视频字幕:
{subtitle}
"""
让模型”根据上下文自动纠错”,实测对”机于”这类同音错别字很有效,问答准确率回来不少。
十八、踩过的坑
整理一下项目里踩过的坑,基本覆盖 DeepSeek 对接的常见雷区:
- DeepSeek 限流:免费版 60 RPM,平台付费版 3000 RPM。
- System prompt 太长:每次都计费。系统 prompt 控制在 500 token 内。
- Context 限制:32k tokens 看似大,但视频字幕可能 10w+ 字符。分段处理。
- 温度选错:总结类用 0.3(稳定),创意类用 0.8(发散)。
- 中英混排:用户传英文视频,prompt 也要英文。用
language参数动态切换。 - 流式中断:客户端断开但 server 还在算。设超时 + 限 max_tokens。
- JSON 解析失败:要求输出 JSON 时模型偶有不规范,平台加 retry + 提取首个 JSON 块。
这里面最容易踩的是 system prompt 长度和流式中断,一条直接关系成本,一条直接浪费资源,上线前逐条核对。
十九、性能优化
几轮优化下来,几个关键指标的变化如下:
| 优化 | 效果 |
|---|---|
| Prompt 缓存 | 成本 -90% |
| 并发限流 | 稳定 10 RPS |
| 分段 map-reduce | 长视频 3x 加速 |
| 备用模型 | 故障 0 停机 |
优化优先级很明确:缓存省成本立竿见影,限流保稳定,map-reduce 撑长视频,备用模型兜底。四件事做完,线上基本没再出过大问题。
常见问题(FAQ)
Q1:DeepSeek 怎么申请?
deepseek.com 注册 → 申请 API Key → 充值。
Q2:DeepSeek 与 GPT 总结质量对比?
中文场景 DeepSeek 略胜,英文 GPT 略胜。
Q3:能换其他模型吗?
能。平台抽象了 LLMClient 接口,换 Claude/Gemini 改配置即可。