开始学习
课程 › 0基础Agent开发课 / LLM基础 / LLM上下文窗口管理-长对话与长文档处理策略
— 15 min read

LLM上下文窗口管理-长对话与长文档处理策略

> **本文适合谁**

LLM 上下文窗口管理：长对话与长文档处理策略

本文适合谁

正在构建长对话系统或处理长文档的开发者。上下文窗口管理是 LLM 应用工程化的核心挑战之一，本文给出可直接复用的代码方案。

Transformer 架构的自注意力机制让 LLM 能关联上下文中的所有 token，但这个能力是有代价的：计算量随序列长度平方增长（O(n²)）。GPT-4o 的 128K token 窗口，单次推理的算力消耗是 4K token 的约 1000 倍。

这个限制产生了两类实际问题：超出窗口大小时如何处理，以及即使在窗口内，如何控制成本。

1.1 上下文窗口的本质限制

一次 LLM API 调用中，所有输入（System Prompt + 历史对话 + 当前消息）的 token 总数不能超过模型的上下文窗口。超出限制时 API 直接报错：

上下文窗口管理四种策略
图 6.11：上下文窗口管理四种策略对比——截断、摘要、滑动窗口与分层压缩

python 
from openai import OpenAI, BadRequestError
import tiktoken  # tiktoken：OpenAI官方的Token计数工具库

client  = OpenAI()
encoder = tiktoken.encoding_for_model("gpt-4o")

def count_tokens(messages: list[dict], model: str = "gpt-4o") -> int:
    """
    精确计算 messages 的 token 数
    每条消息有额外的 4 token 开销（角色标记等）
    整个请求还有额外 3 token 的固定开销
    """
    tokens_per_message = 4
    num_tokens = 3  # 固定开销

    for message in messages:
        num_tokens += tokens_per_message
        for key, value in message.items():
            num_tokens += len(encoder.encode(str(value)))
    return num_tokens

# 不同模型的上下文窗口限制（截至2026年3月，以官方文档为准）
CONTEXT_LIMITS = {
    "gpt-4o":              128_000,
    "gpt-4o-mini":         128_000,
    "claude-opus-4-6":     200_000,
    "deepseek-chat":       64_000,
    "gemini-2.0-flash":    1_000_000,
}

1.2 超出窗口的处理策略

1.2.1 策略一：截断（最简单，但损失信息）

最朴素的方案：保留最近的 N 条消息，丢弃更早的历史。

python 
def truncate_messages(
    messages: list[dict],
    max_tokens: int,
    model: str = "gpt-4o",
    keep_system: bool = True,
) -> list[dict]:
    """
    从最早的消息开始截断，保留最新的对话
    始终保留 system 消息（包含关键指令）
    """
    system_messages = [m for m in messages if m["role"] == "system"]
    other_messages  = [m for m in messages if m["role"] != "system"]

    if keep_system:
        # 先扣除 system 消息占用的 token
        system_tokens = count_tokens(system_messages, model)
        available     = max_tokens - system_tokens - 200  # 预留 200 token 给回复
    else:
        available = max_tokens - 200

    # 从最新消息开始向前保留，直到超出 token 限制
    kept = []
    total = 0
    for msg in reversed(other_messages):
        msg_tokens = count_tokens([msg], model)
        if total + msg_tokens > available:
            break
        kept.insert(0, msg)
        total += msg_tokens

    return system_messages + kept if keep_system else kept

截断的问题：丢失了早期对话里可能很重要的上下文（用户的偏好、约定的格式、前面分析的结论）。

1.2.2 策略二：滑动窗口

保留最近 N 轮完整对话，而不是按 token 数截断。适合对话内容前后关联不强的场景：

python 
def sliding_window_messages(
    messages: list[dict],
    window_size: int = 20,  # 保留最近 20 条消息
) -> list[dict]:
    """
    滑动窗口：保留固定数量的最新消息
    适合多轮问答、每轮相对独立的场景
    """
    system_msgs = [m for m in messages if m["role"] == "system"]
    other_msgs  = [m for m in messages if m["role"] != "system"]

    # 只保留最近 window_size 条
    recent = other_msgs[-window_size:]
    return system_msgs + recent

1.2.3 策略三：摘要压缩

把早期对话压缩成摘要，既保留重要信息，又大幅减少 token 消耗：

python 
async def summarize_history(
    messages: list[dict],
    client: OpenAI,
    summary_model: str = "gpt-4o-mini",  # 用便宜的小模型做摘要
) -> str:
    """
    用 LLM 把历史对话压缩成摘要
    summary_model 用 gpt-4o-mini 而不是 gpt-4o：
    - 摘要任务不需要最强模型
    - 成本约为 gpt-4o 的 1/15
    """
    history_text = "\n".join([
        f"{m['role'].upper()}: {m['content']}"
        for m in messages
        if m["role"] != "system"
    ])

    response = client.chat.completions.create(
        model=summary_model,
        messages=[{
            "role": "system",
            "content": "你是一个对话摘要助手。请将以下对话历史压缩成一段简洁的摘要，保留所有重要的事实、决策和约定。",
        }, {
            "role": "user",
            "content": f"请摘要以下对话：\n\n{history_text}",
        }],
        max_tokens=500,
    )
    return response.choices[0].message.content

class ConversationWithSummary:
    """
    带自动摘要的对话管理器
    当 token 数超过阈值时，自动压缩早期历史
    """

    def __init__(
        self,
        system_prompt: str,
        max_tokens: int = 100_000,
        compress_threshold: float = 0.7,  # 达到最大窗口的 70% 时触发压缩
    ):
        self.system_prompt  = system_prompt
        self.max_tokens     = max_tokens
        self.compress_at    = int(max_tokens * compress_threshold)
        self.messages: list[dict] = []
        self.summary: str | None  = None  # 历史摘要
        self.client = OpenAI()

    def _build_messages(self) -> list[dict]:
        """构建发给 API 的完整 messages 列表"""
        base = [{"role": "system", "content": self.system_prompt}]
        # 如果有历史摘要，把摘要作为 assistant 的第一条消息注入
        if self.summary:
            base.append({
                "role":    "assistant",
                "content": f"[对话历史摘要]\n{self.summary}",
            })
        return base + self.messages

    def chat(self, user_message: str) -> str:
        self.messages.append({"role": "user", "content": user_message})

        all_messages = self._build_messages()
        current_tokens = count_tokens(all_messages)

        # Token 超过阈值，触发摘要压缩
        if current_tokens > self.compress_at:
            # 压缩前半段历史，保留后半段（最近的对话）
            mid = len(self.messages) // 2
            to_compress = self.messages[:mid]
            self.messages = self.messages[mid:]

            new_summary = summarize_history(to_compress, self.client)
            # 如果已有摘要，把新旧摘要合并
            if self.summary:
                merge_prompt = f"旧摘要：{self.summary}\n\n新增对话摘要：{new_summary}"
                self.summary = summarize_history(
                    [{"role": "user", "content": merge_prompt}],
                    self.client
                )
            else:
                self.summary = new_summary

        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=self._build_messages(),
        )
        reply = response.choices[0].message.content
        self.messages.append({"role": "assistant", "content": reply})
        return reply

1.3 长文档处理策略

1.3.1 Map-Reduce

把长文档分块，每块独立处理，最后聚合结果：

python 
def map_reduce_summarize(
    document: str,
    chunk_size: int = 2000,  # 每块约 2000 token
) -> str:
    """
    Map-Reduce 文档摘要
    Map：分块并行处理
    Reduce：合并所有块的摘要
    """
    client = OpenAI()

    def chunk_text(text: str, size: int) -> list[str]:
        """按字符数分块（粗略估计，实际应按 token 数分块）"""
        return [text[i:i+size] for i in range(0, len(text), size)]

    # Map 阶段：每块独立摘要
    chunks   = chunk_text(document, chunk_size)
    summaries = []
    for i, chunk in enumerate(chunks):
        response = client.chat.completions.create(
            model="gpt-4o-mini",  # 分块摘要用小模型降低成本
            messages=[{
                "role":    "user",
                "content": f"请对以下文本（第 {i+1}/{len(chunks)} 部分）生成简洁摘要：\n\n{chunk}",
            }],
            max_tokens=300,
        )
        summaries.append(response.choices[0].message.content)

    # Reduce 阶段：合并所有分块摘要
    combined = "\n\n".join([f"第{i+1}部分：{s}" for i, s in enumerate(summaries)])
    final = client.chat.completions.create(
        model="gpt-4o",  # 最终整合用强模型
        messages=[{
            "role":    "user",
            "content": f"以下是一篇长文档的分段摘要，请整合成一份完整的最终摘要：\n\n{combined}",
        }],
        max_tokens=800,
    )
    return final.choices[0].message.content

### 1.3.2 Refine（迭代精炼）

```python
def refine_summarize(document: str, chunk_size: int = 2000) -> str:
    """
    Refine 策略：逐块处理，每次把之前的摘要和新块一起送给模型
    比 Map-Reduce 更好地保持上下文连贯性，但无法并行
    """
    client = OpenAI()
    chunks  = [document[i:i+chunk_size] for i in range(0, len(document), chunk_size)]
    current_summary = ""

    for i, chunk in enumerate(chunks):
        if i == 0:
            # 第一块：直接生成摘要
            prompt = f"请对以下文本生成摘要：\n\n{chunk}"
        else:
            # 后续块：在已有摘要的基础上精炼
            prompt = (
                f"当前摘要：\n{current_summary}\n\n"
                f"新增内容（第 {i+1} 部分）：\n{chunk}\n\n"
                f"请根据新增内容更新并完善摘要："
            )

        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            max_tokens=500,
        )
        current_summary = response.choices[0].message.content

    return current_summary

1.4 Prompt Caching：降低长上下文成本

Anthropic 和 OpenAI 都提供 Prompt Caching：当请求前缀和上次相同时，缓存的部分不重新计算，大幅降低长上下文的成本。

python 
import os
from openai import OpenAI

client = OpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com",
)

# DeepSeek / OpenAI 的 Prompt Caching：对相同前缀自动缓存，无需特殊标记
def query_with_cache(document: str, question: str) -> str:
    """
    把长文档放在 System Prompt 中，多次查询时前缀相同，命中缓存后大幅降低费用。
    DeepSeek 的缓存命中费率：缓存读取仅需原价的 10%（写入时正常计费）。
    缓存触发条件：System Prompt 内容完全相同且长度超过阈值（约 64 tokens）。
    """
    response = client.chat.completions.create(
        model="deepseek-chat",
        max_tokens=1024,
        messages=[
            {
                "role": "system",
                "content": f"你是一个文档问答助手。\n\n参考文档：\n{document}",
            },
            {"role": "user", "content": question},
        ],
    )
    return response.choices[0].message.content

# 第一次调用：缓存未命中，按正常费率计费
answer1 = query_with_cache(long_document, "文档的主要结论是什么？")

# 第二次调用（System Prompt 相同）：缓存命中，文档部分只需 0.1x 费率，节省 90% 成本
answer2 = query_with_cache(long_document, "文档中有哪些数据支持？")

Prompt Caching 适用场景：同一个 System Prompt 或同一份文档被多次查询，System Prompt 很长（包含大量工具定义、格式说明）。

1.5 各种策略的对比

长对话历史

不重要，最近内容为主

重要，早期内容有价值

长文档处理

否，各部分独立

是，需要上下文积累

重复查询同一文档

文档/对话超出上下文窗口

数据类型

保留完整性重要吗？

截断 / 滑动窗口
简单，有信息损失

摘要压缩
保留语义，压缩 token

需要保持连贯性？

Map-Reduce
可并行，速度快

Refine 迭代精炼
串行，质量更高

Prompt Caching
Anthropic/OpenAI 原生支持
降低 90% 重复 token 成本

各策略的量化对比：

策略 信息保留度 Token 消耗 实现复杂度 适用场景
截断 低（丢失早期信息） 最低 简单 短期任务，历史不重要
滑动窗口 中（保留最近 N 条） 低 简单 多轮问答，上下文局部
摘要压缩 高（语义保留） 中 中等 长会话，信息密度高
Map-Reduce 中（并行，可能遗漏跨块关联） 高 中等 长文档摘要，可并行
Refine 迭代 高（串行积累上下文） 高 中等 长文档，需要连贯性
Prompt Caching 完整（全量） 低（重复调用） 简单 重复查询同一文档

1.6 小结

上下文窗口管理是 LLM 应用工程化绕不开的问题。窗口大小不是无限的，即使是 Gemini 系列的 1M token 窗口，在高并发时也会带来巨大的算力成本。

核心选型原则：数据一次性处理用 Map-Reduce 或 Refine；多轮对话用摘要压缩；重复查询同一文档用 Prompt Caching；快速原型和简单场景用截断就够了。

LLM基础的内容到这里结束。接下来进入 Agent基础，学习如何把这些基础能力组合成能自主完成复杂任务的 AI Agent：ReAct 推理框架、工具链编排、多 Agent 协作。

←上一节Embedding向量化原理-语义搜索的数学基础下一节LLM训练三阶段-预训练-SFT-RLHF→
本页目录