ChatGPT API报错上下文窗口优化实战：从原理到解决方案

1次阅读

共计 3110 个字符，预计需要花费 8 分钟才能阅读完成。

在使用 ChatGPT API 进行开发时，很多开发者都会遇到一个让人头疼的问题：上下文窗口报错。当对话历史超过 API 允许的最大 token 限制时（比如 GPT-3.5 的 4096token 限制），API 就会直接报错，导致对话中断或者响应不完整。这不仅影响用户体验，也给开发带来了不少麻烦。本文将深入分析这个问题的根源，并提出一套可立即落地的解决方案。

每个 API 请求中的文本都会被分割成 token 进行处理。对于英文来说，一个 token 大约相当于 4 个字符或 0.75 个单词；而对于中文，一个汉字通常对应 1 - 2 个 token。这个 token 计数包括了：

用户当前输入
系统指令
所有历史对话记录
API 返回的响应

主流模型的上下文窗口限制如下：

GPT-3.5: 4096 tokens
GPT-4: 8192 tokens（部分版本可达 32k）

当累计 token 数超过这个限制时，API 会返回类似 ”context length exceeded” 的错误。

我们采用 ” 动态上下文管理 ” 策略，主要包括：

实时 token 计数
智能上下文截断
关键信息保留

下面是完整的解决方案代码示例：

import tiktoken
from typing import List, Dict

class ContextManager:
    def __init__(self, model: str = "gpt-3.5-turbo"):
        self.encoder = tiktoken.encoding_for_model(model)
        self.max_tokens = 4096  # GPT-3.5 的默认限制
        self.safety_margin = 512  # 安全余量
        self.context: List[Dict] = []

    def count_tokens(self, text: str) -> int:
        """计算文本的 token 数量"""
        return len(self.encoder.encode(text))

    def add_message(self, role: str, content: str) -> None:
        """添加消息到上下文"""
        self.context.append({"role": role, "content": content})

    def truncate_context(self, new_message: str) -> None:
        """智能截断上下文"""
        # 计算新消息的 token 数
        new_msg_tokens = self.count_tokens(new_message)

        # 计算当前总 token 数
        total_tokens = sum(self.count_tokens(msg["content"]) for msg in self.context) + new_msg_tokens

        # 如果未超限则直接返回
        if total_tokens <= (self.max_tokens - self.safety_margin):
            return

        # 智能截断策略：保留系统指令 + 最近对话 + 关键信息
        truncated_context = []
        remaining_tokens = self.max_tokens - self.safety_margin - new_msg_tokens

        # 1. 优先保留系统指令（如果有）if self.context and self.context[0]["role"] == "system":
            sys_msg = self.context[0]
            sys_tokens = self.count_tokens(sys_msg["content"])
            if sys_tokens < remaining_tokens:
                truncated_context.append(sys_msg)
                remaining_tokens -= sys_tokens

        # 2. 从最新对话开始反向添加，直到填满剩余 token
        for msg in reversed(self.context[1:] if self.context and self.context[0]["role"] == "system" else self.context):
            msg_tokens = self.count_tokens(msg["content"])
            if msg_tokens <= remaining_tokens:
                truncated_context.insert(0, msg)  # 保持原始顺序
                remaining_tokens -= msg_tokens
            else:
                # 3. 对必须保留但过长的消息进行摘要处理
                if msg["role"] == "assistant" and "关键信息" in msg["content"]:  # 示例条件
                    summarized = self.summarize_message(msg["content"], remaining_tokens)
                    if summarized:
                        truncated_context.insert(0, {"role": msg["role"], "content": summarized})
                        break

        self.context = truncated_context

    def summarize_message(self, content: str, max_tokens: int) -> str:
        """生成消息摘要（简化实现，实际应使用更智能的摘要算法）"""
        words = content.split()
        estimated_tokens = 0
        summary = []

        for word in words:
            word_tokens = self.count_tokens(word)
            if estimated_tokens + word_tokens > max_tokens:
                break
            summary.append(word)
            estimated_tokens += word_tokens

        return ''.join(summary) +"..."if len(summary) < len(words) else' '.join(summary)

    def get_context(self, new_message: str) -> List[Dict]:
        """获取处理后的上下文"""
        self.truncate_context(new_message)
        return self.context.copy()