企业官网建设流程全解析

15天学会AI应用开发（四）根据Token长度截断历史对话

在构建AI对话应用时，管理历史对话长度是一项关键挑战。无论是使用OpenAI的GPT模型、Claude还是其他大语言模型，每个模型都有固定的上下文窗口限制（如4096 token、8192 token等）。如果不加控制地累积历史对话，当总token数超过模型限制时，请求会被拒绝或截断，导致对话质量下降。本文将深入剖析Token截断的原理，并提供实用的代码实现方案。## Token与上下文窗口的基本原理### 什么是Token？Token是语言模型处理文本的基本单位。一个token可以是一个单词、子词或字符。例如，英文中"hello"通常是一个token，而中文中"你好"可能是1-2个token。不同模型的tokenizer（分词器）不同，但核心原则一致：模型通过token序列理解语义。### 上下文窗口的硬性限制每个模型都有最大上下文窗口，比如GPT-3.5-turbo支持4096 tokens，GPT-4支持8192 tokens。当输入的总token数（包括用户消息、系统提示、历史对话、当前问题）超过这个限制时，模型会拒绝处理或抛出错误。因此，我们需要一种策略来截断历史对话，确保总token数在安全范围内。### 截断策略的设计原则1.保留最新消息：通常最近的消息包含更相关的上下文。2.丢弃最旧消息：早期对话对当前回复贡献较小。3.考虑系统提示：系统提示（如角色设定）通常需要保留。4.动态调整：根据当前输入长度动态计算可保留的历史。## 基于Token长度的截断实现### 计算Token长度首先，我们需要一个函数来计算文本的token数。使用OpenAI的tiktoken库是最精准的方式，因为它与GPT模型的分词器一致。对于其他模型，可以使用transformers库或估算方法。pythonimport tiktokendef count_tokens(text: str, model: str = "gpt-3.5-turbo") -> int: """ 计算给定文本的token数 :param text: 输入文本 :param model: 模型名称，默认gpt-3.5-turbo :return: token数量 """ try: encoding = tiktoken.encoding_for_model(model) return len(encoding.encode(text)) except KeyError: # 如果模型不在tiktoken列表中，使用cl100k_base编码 encoding = tiktoken.get_encoding("cl100k_base") return len(encoding.encode(text))# 示例：计算字符串的token数example_text = "你好，今天天气怎么样？"token_count = count_tokens(example_text)print(f"文本: '{example_text}' 的token数: {token_count}")### 核心截断函数接下来，我们实现一个智能截断函数，它接收历史对话列表、最大token限制和系统提示，返回截断后的对话。pythonfrom typing import List, Dict, Tupledef truncate_conversation( messages: List[Dict[str, str]], max_tokens: int = 4096, system_prompt: str = "", model: str = "gpt-3.5-turbo") -> Tuple[List[Dict[str, str]], int]: """ 根据token限制截断历史对话，保留最新消息 :param messages: 对话消息列表，格式为[{"role": "user", "content": "..."}, ...] :param max_tokens: 最大token限制 :param system_prompt: 系统提示文本（可选） :param model: 模型名称 :return: 截断后的消息列表和总token数 """ # 计算系统提示的token数 system_tokens = count_tokens(system_prompt, model) if system_prompt else 0 # 预留空间给当前输入（假设当前输入最大为max_tokens的20%） input_buffer = int(max_tokens * 0.2) available_tokens = max_tokens - system_tokens - input_buffer # 从最新消息开始累积，直到达到限制 truncated_messages = [] total_tokens = system_tokens # 反转列表以便从最新消息开始处理 reversed_messages = list(reversed(messages)) for msg in reversed_messages: # 计算当前消息的token数 msg_tokens = count_tokens(msg["content"], model) # 加上角色标识的固定开销（如"user"或"assistant"） role_overhead = count_tokens(msg["role"], model) msg_total = msg_tokens + role_overhead if total_tokens + msg_total <= available_tokens: # 可以添加此消息 truncated_messages.append(msg) total_tokens += msg_total else: # 超出限制，停止添加 break # 恢复原始顺序（从旧到新） truncated_messages.reverse() # 如果系统提示存在，添加到开头 if system_prompt: truncated_messages.insert(0, {"role": "system", "content": system_prompt}) total_tokens += system_tokens return truncated_messages, total_tokens# 示例：模拟历史对话history_messages = [ {"role": "user", "content": "什么是机器学习？"}, {"role": "assistant", "content": "机器学习是人工智能的一个分支，它使计算机能够从数据中学习..."}, {"role": "user", "content": "能举个例子吗？"}, {"role": "assistant", "content": "当然，比如垃圾邮件分类器就是一个例子..."}, {"role": "user", "content": "深度学习与机器学习有什么关系？"}, {"role": "assistant", "content": "深度学习是机器学习的一个子集，使用神经网络..."},]# 假设模型最大4096 tokens，系统提示占用100 tokenssystem_prompt = "你是一个友好的AI助手，回答用户问题。"truncated, total = truncate_conversation( messages=history_messages, max_tokens=4096, system_prompt=system_prompt)print(f"原始消息数: {len(history_messages)}")print(f"截断后消息数: {len(truncated)}")print(f"总token数: {total}")for msg in truncated: print(f"[{msg['role']}]: {msg['content'][:50]}...")## 高级优化策略### 动态保留策略在实际应用中，我们可能需要更精细的控制。例如，保留系统提示和最后N条消息，同时丢弃中间部分。以下是一个增强版本：pythondef smart_truncate_conversation( messages: List[Dict[str, str]], max_tokens: int = 4096, system_prompt: str = "", min_recent_messages: int = 2, model: str = "gpt-3.5-turbo") -> Tuple[List[Dict[str, str]], int]: """ 智能截断：至少保留最近min_recent_messages条消息 """ # 计算系统提示和最少消息的token数 system_tokens = count_tokens(system_prompt, model) if system_prompt else 0 # 确保至少保留最近的几条消息 recent_messages = messages[-min_recent_messages:] if len(messages) >= min_recent_messages else messages recent_tokens = sum(count_tokens(msg["content"], model) + count_tokens(msg["role"], model) for msg in recent_messages) available_tokens = max_tokens - system_tokens - int(max_tokens * 0.2) # 保留20%给当前输入 # 如果最近消息已经超出限制，只保留最近消息（可能部分丢失） if recent_tokens > available_tokens: # 从最近消息中截断 reversed_recent = list(reversed(recent_messages)) truncated_recent = [] token_sum = 0 for msg in reversed_recent: msg_tokens = count_tokens(msg["content"], model) + count_tokens(msg["role"], model) if token_sum + msg_tokens <= available_tokens: truncated_recent.append(msg) token_sum += msg_tokens else: break truncated_recent.reverse() truncated_messages = truncated_recent total_tokens = token_sum else: # 从历史消息中逐步添加，从最新到最旧 truncated_messages = recent_messages.copy() total_tokens = recent_tokens for msg in reversed(messages[:-min_recent_messages]): msg_tokens = count_tokens(msg["content"], model) + count_tokens(msg["role"], model) if total_tokens + msg_tokens <= available_tokens: truncated_messages.insert(0, msg) total_tokens += msg_tokens else: break # 添加系统提示 if system_prompt: truncated_messages.insert(0, {"role": "system", "content": system_prompt}) total_tokens += system_tokens return truncated_messages, total_tokens# 测试智能截断truncated_smart, total_smart = smart_truncate_conversation( messages=history_messages, max_tokens=4096, system_prompt=system_prompt, min_recent_messages=3)print(f"智能截断 - 保留消息数: {len(truncated_smart)}, 总token数: {total_smart}")### 与API调用集成最后，将截断函数与OpenAI API调用结合：pythonimport openaidef chat_with_truncation( user_input: str, history: List[Dict[str, str]], system_prompt: str = "你是一个AI助手", max_tokens: int = 4096, model: str = "gpt-3.5-turbo") -> str: """ 带截断功能的对话API调用 """ # 将用户输入添加到历史 history.append({"role": "user", "content": user_input}) # 截断历史 truncated_history, total_tokens = smart_truncate_conversation( messages=history, max_tokens=max_tokens, system_prompt=system_prompt, model=model ) # 调用API response = openai.ChatCompletion.create( model=model, messages=truncated_history, max_tokens=500 # 生成回复的token限制 ) assistant_response = response.choices[0].message.content # 将回复添加到历史（全局变量或持久化存储） history.append({"role": "assistant", "content": assistant_response}) return assistant_response## 总结本文深入探讨了AI对话应用中Token截断的核心原理与实现。我们了解到：1. Token是模型处理文本的基本单位，每个模型有固定的上下文窗口限制。2. 截断策略应优先保留最新消息和系统提示，动态计算可用token空间。3. 使用tiktoken库可以精确计算token数，避免估算误差。4. 高级策略如智能保留最近消息能提升对话连续性。通过实现truncate_conversation和smart_truncate_conversation函数，我们可以有效管理对话历史，避免超出模型限制。在实际开发中，还需要考虑并发请求、持久化存储和错误处理等细节。掌握Token截断技术，是构建稳定、高效的AI应用的基础。