1M 上下文意味着什么

DeepSeek V4 的 1M tokens 上下文窗口约等于:

  • 70 万中文字符——相当于一部《三体》全书的长度
  • 1000 页 PDF 文档
  • 10 万行代码
  • 数百轮对话历史

但窗口越大,越需要策略——不能无脑往里塞数据,否则 token 成本和延迟都会失控。

消息截断策略

对于多轮对话,最常见的需求是保留最近 N 轮,截断更早的消息:

def trim_conversation(messages, max_turns=20):
    """保留 system prompt + 最近 max_turns 轮对话"""
    system_msgs = [m for m in messages if m['role'] == 'system']
    chat_msgs = [m for m in messages if m['role'] != 'system']

    # 保留最近 max_turns 轮(每轮 = user + assistant)
    trimmed = chat_msgs[-(max_turns * 2):]

    return system_msgs + trimmed

# 使用
messages = [{"role": "system", "content": "你是一个编程助手"}]
# ... 多轮对话后
messages = trim_conversation(messages, max_turns=15)

response = client.chat.completions.create(
    model='deepseek-v4-flash',
    messages=messages
)

摘要压缩策略

更好的做法是将历史消息压缩为摘要,而不是直接丢弃:

async def compress_history(messages, threshold_turns=10):
    """当对话超过 threshold_turns 轮时,压缩早期消息为摘要"""
    system_msgs = [m for m in messages if m['role'] == 'system']
    chat_msgs = [m for m in messages if m['role'] != 'system']

    if len(chat_msgs) <= threshold_turns * 2:
        return messages  # 不需要压缩

    # 早期消息转文本
    early = chat_msgs[:-(threshold_turns * 2)]
    recent = chat_msgs[-(threshold_turns * 2):]

    history_text = '\n'.join([f"{m['role']}: {m['content'][:200]}" for m in early])

    # 生成摘要
    summary_response = await client.chat.completions.create(
        model='deepseek-v4-flash',
        messages=[{
            "role": "user",
            "content": f"用一段中文摘要总结以下对话的关键信息:\n{history_text}"
        }]
    )

    summary = summary_response.choices[0].message.content

    return system_msgs + [
        {"role": "system", "content": f"[历史对话摘要] {summary}"}
    ] + recent

长文档分段处理

对于超长文档(如整本书),分段处理是最实用的方法:

def chunk_document(text, chunk_size=30000):
    """将长文本分割为适合上下文窗口的块"""
    chunks = []
    for i in range(0, len(text), chunk_size):
        chunks.append(text[i:i+chunk_size])
    return chunks

async def analyze_long_doc(text, question):
    chunks = chunk_document(text)
    findings = []

    # 第一阶段:逐块分析
    for i, chunk in enumerate(chunks):
        response = await client.chat.completions.create(
            model='deepseek-v4-flash',
            messages=[{
                "role": "system",
                "content": "从文本块中提取与问题相关的关键信息。若无相关信息,回答'无'。"
            }, {
                "role": "user",
                "content": f"问题:{question}\n文本块 [{i+1}/{len(chunks)}]:{chunk}"
            }]
        )
        result = response.choices[0].message.content
        if result != '无':
            findings.append(f"[块 {i+1}] {result}")

    # 第二阶段:汇总分析
    context = '\n\n'.join(findings)
    final = await client.chat.completions.create(
        model='deepseek-v4-pro',  # 汇总用 Pro 保证质量
        messages=[{
            "role": "user",
            "content": f"基于以下分析结果回答:{question}\n\n{context}"
        }]
    )
    return final.choices[0].message.content

Token 监控

# 每次请求后检查 token 用量
response = client.chat.completions.create(...)
usage = response.usage
print(f"输入: {usage.prompt_tokens} tokens")
print(f"输出: {usage.completion_tokens} tokens")
print(f"总计: {usage.total_tokens} tokens")

# 估算上下文使用率
if usage.prompt_tokens > 800000:
    print("警告:上下文使用率超 80%,建议开启压缩!")

最佳实践总结

  • 日常对话保持消息数在 20 轮以内,使用截断策略
  • 需要保留历史时使用摘要压缩
  • 超长文档使用分段处理 + 汇总的两阶段法
  • 始终监控 token 用量,接近上限时提前处理