(各位访问大哥能否留一些宝贵建议给小弟,没人的话就当个人日常blog)虽然我在之前创建了一个json文件来保存聊天会话内容,但是模型读取的会话文本是有长度限制的,在达到极限后的对话机器人就会拒绝输出,影响使用体验。实际上开发人员并不会把用户和ai的聊天内容完全存储,因为这样占内存大、维护困难且成本高。为了维护用户体验和延长模型“寿命”,一般会采取滑动窗口、摘要压缩、按需检索等方式。我目前采取的是滑动窗口机制。
滑动窗口机制是删除早期的记忆内容,只保留最近 N 轮会话内容和System Prompt,好处是极大地节省算力和资源成本,保证最近会话的连贯和避免早期信息干扰。
其实还有个类似的机制叫固定窗口,同样是通过删除来避免内容超出文本限制,与滑动窗口不同的是滑动窗口机制规定会话达到一定轮数后就强制清除所有内容,处理更简单暴力,但问题也很明显:无法支持长期会话。所以一般是用来防止系统过载。
有句比喻我觉得很恰当:滑动窗口是“挤牙膏”,固定窗口是“倒垃圾”。
下面是添加了滑动窗口机制删除历史会话的第五代聊天机器人:
""" day4: 带 Token 感知的聊天机器人 新增: - 计算对话历史的 token 数量 - 超出上限自动裁剪最老的对话 """ import json import os from pathlib import Path import tiktoken from openai import OpenAI from dotenv import load_dotenv root = Path(__file__).parent.parent load_dotenv(root / ".env") client = OpenAI( api_key=os.getenv("DEEPSEEK_API_KEY"), base_url=os.getenv("DEEPSEEK_BASE_URL"), ) SYSTEM_PROMPT = """你是一个编程老师,名字叫"小码"。 规则: 1. 用中文回答,口语化,像和朋友聊天一样 2. 解释概念时用通俗的例子,不要堆术语 3. 如果用户让你写代码,要逐行解释每一行是干什么的 4. 回答控制在 200 字以内,除非用户明确要求详细""" HISTORY_FILE = Path(__file__).parent / "chat_history.json" # 分词器(和 DeepSeek 兼容) encoder = tiktoken.get_encoding("cl100k_base") # 安全阈值:模型上限 128K,留 28K 给 AI 回复用,100K 开始裁剪 MAX_TOKENS = 100_000 def count_tokens(messages: list) -> int: """计算 messages 列表的总 token 数""" total = 0 for msg in messages: total += len(encoder.encode(msg["content"])) return total def trim_history(history: list) -> list: """裁剪历史:删除最老的对话对,直到 token 数降到安全线以下""" # 必须保留 system 角色 system_msg = history[0] conversations = history[1:] # 后面的才是对话 while count_tokens([system_msg] + conversations) > MAX_TOKENS and len(conversations) >= 2: # 每次删掉最前面的一组对话(1 user + 1 assistant) conversations = conversations[2:] return [system_msg] + conversations def load_history() -> list: if HISTORY_FILE.exists(): with open(HISTORY_FILE, "r", encoding="utf-8") as f: return json.load(f) return [{"role": "system", "content": SYSTEM_PROMPT}] def save_history(history: list): with open(HISTORY_FILE, "w", encoding="utf-8") as f: json.dump(history, f, ensure_ascii=False, indent=2) def chat_stream(messages: list) -> str: stream = client.chat.completions.create( model="deepseek-v3.2", messages=messages, temperature=0.7, max_tokens=800, stream=True, ) full_answer = "" for chunk in stream: content = chunk.choices[0].delta.content if content: print(content, end="", flush=True) full_answer += content print() return full_answer def main(): history = load_history() if len(history) > 1: print(f"[恢复了 {len(history) // 2} 轮历史对话]") # 启动时也裁剪一次(防止历史文件里存的太长) old_len = len(history) history = trim_history(history) if len(history) < old_len: print(f"[自动裁剪了 {(old_len - len(history)) // 2} 轮旧对话]") token_count = count_tokens(history) print(f"[当前上下文: {token_count} / {MAX_TOKENS} tokens]") print("=" * 50) print(" 小码 AI 编程老师 0.5 (quit 退出, !clear 清空记忆)") print("=" * 50) while True: user_input = input("\n你: ").strip() if not user_input: continue if user_input.lower() in ("quit", "exit", "q"): print("小码: 再见! 有问题随时回来~") break if user_input == "!clear": history = [{"role": "system", "content": SYSTEM_PROMPT}] save_history(history) print("[记忆已清空]") continue history.append({"role": "user", "content": user_input}) # 发送前裁剪 old_len = len(history) history = trim_history(history) if len(history) < old_len: print(f"[自动裁剪了 {(old_len - len(history)) // 2} 轮旧对话]") print("小码: ", end="", flush=True) try: answer = chat_stream(history) history.append({"role": "assistant", "content": answer}) save_history(history) except Exception as e: import traceback print(f"\n出错了: {e}") traceback.print_exc() history.pop() if __name__ == "__main__": main()基于上述代码,聊天要达到文本上限起码得聊个几百轮,但为了演示滑动窗口删除效果,我把token上限常量从100_000改为200,这样只要聊几轮就能触发删除早期会话记录功能。效果如下所示(加了打印功能,可以知道记忆从什么时候开始丢失):
因为我设置的token上限常量很小,只要聊几轮就会丢失前面的记忆内容,和现实中的阿尔兹海默症患者比较相像(无歧视)。可以想象前几分钟还在和眼前熟悉的人谈笑风生,下一秒他/她突然问你是谁,站在患者的亲人朋友视角是很绝望的。类似地,站在用户视角,不久前还在和ai分享自己的生活或者使用ai办公,因为token上限问题就把前面重要的内容删除了,用户的心情会很糟糕,甚至工作进度也会受影响。所以我认为直接使用滑动窗口机制去删除历史内容来控制文本体量是不可取的,应该在考虑内存大小、维护难度及成本的前提下,配合摘要压缩、按需检索等方式尽可能照顾用户体验来控制文本体量。
总结:今天就实现了利用滑动窗口机制来控制聊天文本体量的功能,通过“删旧存新”的逻辑去避免记忆文本超出模型读取的上限。事实上这个方案并不完美,后续我会进一步学习更好的方式来处理这个问题。
(今天健身晚了回来有点累,写这篇博客时快要睡着了,就没有精力多学点)
第四天就这样吧。