记录下学习agent应用开发的第四天（滑动窗口机制删除早期文本）

各位访问大哥能否留一些宝贵建议给小弟没人的话就当个人日常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_keyos.getenv(DEEPSEEK_API_KEY), base_urlos.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, encodingutf-8) as f: return json.load(f) return [{role: system, content: SYSTEM_PROMPT}] def save_history(history: list): with open(HISTORY_FILE, w, encodingutf-8) as f: json.dump(history, f, ensure_asciiFalse, indent2) def chat_stream(messages: list) - str: stream client.chat.completions.create( modeldeepseek-v3.2, messagesmessages, temperature0.7, max_tokens800, streamTrue, ) full_answer for chunk in stream: content chunk.choices[0].delta.content if content: print(content, end, flushTrue) 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, flushTrue) 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上限问题就把前面重要的内容删除了用户的心情会很糟糕甚至工作进度也会受影响。所以我认为直接使用滑动窗口机制去删除历史内容来控制文本体量是不可取的应该在考虑内存大小、维护难度及成本的前提下配合摘要压缩、按需检索等方式尽可能照顾用户体验来控制文本体量。总结今天就实现了利用滑动窗口机制来控制聊天文本体量的功能通过“删旧存新”的逻辑去避免记忆文本超出模型读取的上限。事实上这个方案并不完美后续我会进一步学习更好的方式来处理这个问题。今天健身晚了回来有点累写这篇博客时快要睡着了就没有精力多学点第四天就这样吧。