1. 项目起源为什么我被 context-mode 逼到重新造轮子先交代一下背景。我最近在做一个面向开发者的 AI 辅助工具说直白点就是一个跑在终端里的编程助手用户问一句它答一句。功能原型三天就做出来了demo 演示也很顺可一旦进入真实使用场景问题立刻暴露用户聊到第 20 轮助手已经彻底忘了第 3 轮说过的技术栈约束甚至会把用户明确否掉的方案又推荐一遍。最初我以为是模型能力不行换了好几个大模型 API问题依旧。后来才想明白根子出在我自己写的调用逻辑上——我压根没做任何状态管理每次请求都是孤立的模型天然无状态你给它什么它就只能基于什么回答。这不是模型笨是我没把上下文这个东西当成一等公民来对待。context-mode 这个名字说穿了就是我给这套完整上下文管理方案起的代号它不是一个单一函数也不是某个配置开关而是一整套运行模式从对话记录的持久化到 Token 预算的分配再到窗口滑动和摘要压缩统一解决AI 记不住事的问题。这篇文章把我从设计到落地、再到踩坑排错的全过程原样写出来适合正在做 AI 应用开发、尤其是做聊天机器人和智能助手的同学参考也适合那些想给现有工具加记忆能力但不知道从哪下手的团队。2. 核心设计思路先把上下文拆成能落地的东西2.1 我踩的第一个坑把上下文简单等同于把所有聊天记录都塞进去新手最容易犯的错误就是把上下文管理做成拼字符串——把历史消息全部拼到一个大 prompt 里发给模型。我第一版就是这么干的。测试时发现三个问题第一Token 消耗爆炸。聊到 50 轮光携带历史消息就要吃掉上万 Token成本直线上升响应时间也明显变慢。第二模型注意力被稀释。我不是拿模型做实验我是被测对象——把 2 万字的闲聊记录全部塞给模型它连用户的核心诉求都聚焦不了回答质量反而下降。第三跨会话能力为零。程序一重启历史全没了用户昨天交代的重要背景今天得重新讲一遍。后来我翻阅了一些系统设计文档才算彻底明白上下文管理本质上做的是信息筛选 状态持久化目标是让模型在每一次请求时都能拿到当前最该知道的信息而不是所有曾经出现过的信息。这和人类的短期记忆很像大脑会自动遗忘无关细节只保留对当前决策有用的部分。2.2 三个核心模块记忆体、调度器、压缩器想清楚目标之后我把 context-mode 拆成了三个独立模块各管一摊互不干扰记忆体Memory Store负责把每一条对话消息结构化地存下来。我不建议用纯内存列表因为程序一重启就没了也不建议一上来就上重型数据库对个人项目和中小型工具来说SQLite 足够轻量、单文件、查询方便。窗口调度器Window Scheduler负责决定这一轮请求到底携带哪些消息。核心参数是 Token 预算上限超过上限就按策略丢弃或压缩最旧的消息。这个模块是整套方案的心脏后面我会详细讲参数的确定过程。压缩器Compressor负责把被挤掉的历史消息降维保存——生成摘要存到摘要区这样即使原始消息被丢弃核心信息仍然保留。相当于把短期记忆固化成长效记忆。这三个模块的划分不是拍脑袋定的我是按读、写、管三个职责来拆的。记忆体贴近数据层调度器贴近业务逻辑压缩器贴近模型能力。拆开的直接好处是任何一部分出了问题我只需要单独排查和替换不用牵一发动全身。2.3 数据结构设计一张表讲清楚我最终设计的消息模型包含以下字段字段名类型说明msg_idTEXT消息唯一 ID用 UUIDsession_idTEXT会话 ID标识属于哪一轮对话roleTEXTuser / assistant / systemcontentTEXT消息正文token_countINTEGER这条消息的 Token 数提前算好created_atINTEGER时间戳用于排序summaryTEXT摘要字段压缩后写入is_compressedINTEGER是否为压缩摘要消息其中 token_count 这个字段是我特意加上去的很多人在设计阶段会忽略它但它是调度器做预算分配的基础。如果每次调度时才临时计算 Token性能会差很多且无法做全局预算规划。session_id 用来区分不同对话主题这样同一个用户的不同需求可以各聊各的上下文互不串。可能有人问为什么不直接按 user_id 存一份全局记忆呢我后面遇到的实际问题是把两个完全不相干的任务混在一个上下文里模型会严重串戏。所以从第一天起就坚定按 session 隔离。3. 实操实现从零搭一个可运行的 context-mode3.1 技术选型和环境准备我的技术栈是 Python模型接口用 OpenAI 风格的 Chat Completions API这个格式是目前各家大模型厂商兼容度最高的换模型不用改业务代码。需要准备的环境如下# Python 3.10 以上即可 pip install openai tiktokentiktoken 是 OpenAI 官方的 Tokenizer 库用来精确计算消息的 Token 数。我踩过一个坑使用不同模型时Token 计算方法不一样比如 gpt-4 系列和 gpt-3.5 系列用的编码器名称就不同稍后会细说。3.2 记忆体实现SQLite 落地最小但够用我先写记忆体。它要提供的能力很简单插入消息、按会话读取消息、获取消息总数和总 Token 数、删除最旧的消息。import sqlite3 import uuid import json from datetime import datetime class MemoryStore: def __init__(self, db_pathcontext_mode.db): self.conn sqlite3.connect(db_path) self._init_table() def _init_table(self): self.conn.execute( CREATE TABLE IF NOT EXISTS messages ( msg_id TEXT PRIMARY KEY, session_id TEXT, role TEXT, content TEXT, token_count INTEGER, created_at INTEGER, summary TEXT, is_compressed INTEGER DEFAULT 0 ) ) self.conn.execute( CREATE INDEX IF NOT EXISTS idx_session_time ON messages(session_id, created_at) ) self.conn.commit() def add_message(self, session_id, role, content, token_count): msg_id str(uuid.uuid4()) now int(datetime.now().timestamp() * 1000) self.conn.execute( INSERT INTO messages VALUES (?, ?, ?, ?, ?, ?, NULL, 0), (msg_id, session_id, role, content, token_count, now), ) self.conn.commit() return msg_id def get_session_messages(self, session_id, limitNone): sql SELECT * FROM messages WHERE session_id ? ORDER BY created_at ASC if limit: sql f LIMIT {limit} rows self.conn.execute(sql, (session_id,)).fetchall() return rows def session_token_total(self, session_id): row self.conn.execute( SELECT COALESCE(SUM(token_count), 0) FROM messages WHERE session_id ?, (session_id,), ).fetchone() return row[0] def delete_oldest(self, session_id, keep_count1): self.conn.execute( DELETE FROM messages WHERE msg_id IN ( SELECT msg_id FROM messages WHERE session_id ? ORDER BY created_at ASC LIMIT ? ) , (session_id, keep_count)) self.conn.commit()这里有个细节created_at 我用毫秒时间戳而不是字符串日期原因是排序和比较都更快且不受时区影响。SQLite 没有专门的时间类型存 INTEGER 是最省心的。3.3 窗口调度器Token 预算到底怎么算窗口调度器是整套方案最核心、也是最容易出问题的地方。我采用的策略是经典的高低水位法High/Low Water Mark思路借鉴了操作系统内存管理的设计哲学——系统不会等资源耗尽才做清理而是设两个阈值达到高水位触发清理清理到低水位停止。高水位HWM和低水位LWM的确定直接决定了上下文质量。经过多轮实测我最终把参数定成这样子class WindowScheduler: def __init__(self, max_budget8000, high_water_mark0.8, low_water_mark0.5): self.max_budget max_budget self.hwm high_water_mark self.lwm low_water_mark def compute_high_water(self): return int(self.max_budget * self.hwm) def compute_low_water(self): return int(self.max_budget * self.lwm)参数含义用一个表说清楚参数取值含义max_budget8000 Token单次请求携带消息的 Token 上限high_water_mark0.8达到 80% 预算6400 Token时触发清理low_water_mark0.5清理到 50% 预算4000 Token时停止为什么高水位不设成 100%因为请求除了消息之外还要附带 system prompt、用户当前的问题、可能的工具返回结果这些都要占用 Token。如果消息区直接吃到上限真正任务的 Token 就不够了会报错或截断。留出 20% 的余量是必须的操作冗余。为什么低水位是 50% 而不是 0%因为频繁清理会带来两个问题一是性能开销每次清理都要调用模型生成摘要二是用户体验清理本身意味着信息丢失清理得越频繁丢失的信息就越多。留 50% 的缓冲区间可以让调度器在到达高水位后一次清理到位然后有充足的空间继续累积。实际调度逻辑如下def schedule(self, store, session_id, new_message_tokens): current_total store.session_token_total(session_id) estimated current_total new_message_tokens high_water self.compute_high_water() if estimated high_water: return None # 不需要清理 # 需要清理把超过低水位的部分压缩掉 low_water self.compute_low_water() overflow estimated - low_water return overflow # 返回需要释放的 Token 量调度器返回 overflow 后业务层会把最旧的消息交给压缩器处理压缩后的摘要保留在消息列表里原始消息从主列表中移除。这样模型依然能看到很久以前聊过某主题的摘要信息不会彻底失忆。3.4 压缩器用模型替你做记忆整理压缩器本质上就是一次专门的模型调用让它把一组旧消息浓缩成几句话。这里的关键是压缩要一次完成绝不能循环调用否则性能会非常难看。class Compressor: def __init__(self, client, modelgpt-4o-mini): self.client client self.model model def compress(self, messages): # 把消息序列化成文本交给模型总结 conversation_text \n.join( f{m[3]}: {m[4]} for m in messages ) prompt ( 请总结以下对话的核心信息包括用户的目标、偏好、 以及已经确定的结论。用简洁的中文输出不要超过150字。\n\n f{conversation_text} ) resp self.client.chat.completions.create( modelself.model, messages[{role: user, content: prompt}], temperature0.3, max_tokens200, ) return resp.choices[0].message.content把 temperature 设到 0.3 是个小讲究。压缩场景要的是稳定和忠实不是创造性发挥。设高了压缩结果会自由发挥甚至编造设 0 又有时过于保守0.3 是我测试下来准确率和召回率平衡最好的点。压缩完成后原消息并没有物理删除我选择把它们标记为 is_compressed 并存到 summary 字段同时在业务层面从随请求携带列表中拿掉。这样的好处是如果需要追溯随时可以从数据库里挖出原始消息如果不需要数据库文件也不会无限制膨胀。3.5 完整调用流程把三个模块串起来有了三个模块最后的工作就是把它们拼成一个完整的请求链路。我封装了一个 ContextEngine 类业务侧只需调用一个方法class ContextEngine: def __init__(self, store: MemoryStore, scheduler: WindowScheduler, compressor: Compressor, client, modelgpt-4o-mini): self.store store self.scheduler scheduler self.compressor compressor self.client client self.model model def send_message(self, session_id, user_text): # 1. 计算新消息 Token new_tokens count_tokens(user_text) # 2. 调用调度器判断是否需要清理 overflow self.scheduler.schedule(self.store, session_id, new_tokens) # 3. 需要清理时把最旧的消息交给压缩器 if overflow is not None: self._roll_compress(session_id, overflow) # 4. 保存用户消息 self.store.add_message(session_id, user, user_text, new_tokens) # 5. 组装消息列表 messages self._build_messages(session_id, user_text) # 6. 调用模型 resp self.client.chat.completions.create( modelself.model, messagesmessages, ) answer resp.choices[0].message.content # 7. 保存助手回复 answer_tokens count_tokens(answer) self.store.add_message(session_id, assistant, answer, answer_tokens) return answer def _roll_compress(self, session_id, overflow): messages self.store.get_session_messages(session_id) target [] used 0 for m in messages: if used overflow: break # 只压缩用户和助手消息不压缩摘要 if m[6] ! 1 and m[2] ! system: target.append(m) used m[5] if not target: return summary self.compressor.compress(target) # 把压缩后的摘要以 system 角色插入 self.store.add_message(session_id, system, f[历史摘要] {summary}, count_tokens(summary)) # 从主列表移除被压缩消息标记一下即可 ids [m[0] for m in target] self.store.conn.executemany( UPDATE messages SET is_compressed 1 WHERE msg_id ?, [(i,) for i in ids], ) self.store.conn.commit() def _build_messages(self, session_id, user_text): rows self.store.get_session_messages(session_id) messages [] for m in rows: if m[6] 1: continue # 跳过已压缩消息 messages.append({role: m[2], content: m[4]}) # 当前用户消息最后再追加一次避免时序问题 messages.append({role: user, content: user_text}) return messages注意 _build_messages 里我特意把当前用户消息追加了一次。实际有个容易忽略的坑如果用户消息已经通过 add_message 存入数据库再读取时时间排序没问题但如果你在组装时也把它们读出来就会把本次用户输入重复发送两遍。为了避免这个时序问题我用先存库 再手动追加当前消息的方式处理而组装函数里用 is_compressed 过滤掉压缩项。虽然简单粗暴但实测下来稳定可靠。3.6 Token 计算的细节不同模型编码器不一样Token 计算是整个系统里最容易出 bug、又最不容易被发现的环节。tiktoken 的使用方式如下import tiktoken def count_tokens(text, modelgpt-4o-mini): try: encoding tiktoken.encoding_for_model(model) except KeyError: # 新模型可能不在库中回退到 cl100k_base encoding tiktoken.get_encoding(cl100k_base) return len(encoding.encode(text))我踩过这样的坑gpt-4、gpt-4-turbo、gpt-4o 使用的编码器并不完全相同如果不小心把不同编码器的计数混用会导致调度的预算判断有偏差——偏差在几万 Token 的长对话里会被无限放大最直观的后果就是对话没聊多久API 就报超限错误。我的处理方式简单粗暴统一用 cl100k_base或者干脆在启动时按实际模型获取编码器并缓存。对于精度要求没那么高的场景cl100k_base 已经够用因为它与大部分 gpt 模型的编码高度接近。如果用的是开源模型比如 Qwen、Llama 系列tiktoken 就不适用了得用模型各自的分词器或者使用 HuggingFace 的 AutoTokenizer。4. 常见问题与排查技巧实录4.1 模型跑题严重压缩摘要误导模型我在测试压缩器时发现一个很有意思的现象当某段历史被压缩成摘要后模型有时会抓着摘要里的某个词大做文章反而忽略了用户当前的问题。排查下来根因在于摘要中包含了太多不重要的细节。比如原始对话里用户提到过一句我公司服务器在杭州模型在后面的回答里就反复围绕杭州的机房展开建议而用户当下的问题是这段代码为什么报错。我的解决方案是在压缩 prompt 里加上一条明确指令——忽略所有与地理位置、人名、具体时间等无关紧要的细节只保留任务目标、技术约束和结论。另外把摘要的 max_tokens 从 200 降到 100强制模型做减法。改完之后跑题问题大幅减少。4.2 清理触发频率过高低水位设计失误第二个问题出现在我把低水位从 50% 临时改成 80% 的时候。结果模型频繁触发压缩每聊两三轮就压缩一次既浪费 Token又导致对话体验极差用户会感觉到 AI 偶尔卡顿。这个问题的原理在前面提过高水位和低水位之间的缓冲区间越大清理频率越低。80% 的低水位让缓冲区只有 20%随便聊几句就触顶。调回 50% 之后缓冲区达到 30%大约每 8 到 10 轮才需要压缩一次体感流畅多了。4.3 会话隔离没做好A 任务污染 B 任务第三个问题来自于我自己的一个简化设计。最初我以为把 session_id 写上就万事大吉结果测试中发现用户在同一个 session 里会同时聊两个完全不相关的任务——一会儿问 Python 代码一会儿问菜谱。这种场景下模型的注意力会反复横跳摘要也变得越来越奇怪经常把菜谱内容和代码逻辑混在一起。后来我加了主题分裂检测的预处理逻辑如果当前消息与最近 5 条消息的语义相似度低于 0.4就自动开启新 session。虽然判断相似度需要额外调一次嵌入模型但效果好得明显值得这个开销。4.4 问题排查速查表现象可能原因排查方向模型回答内容飘忽不定历史消息带了太多无关内容检查窗口调度是否生效留意摘要是否过多API 频繁报 Token 超限max_budget 设置过高或 Token 计数不准检查编码器是否匹配模型预算余量是否充足压缩后信息丢失严重压缩 prompt 不够明确或 max_tokens 太小调整压缩指令适当增大输出限制数据库文件无限膨胀被压缩消息没有正确标记或清理检查 is_compressed 字段必要时定期归档旧 session启动后历史全部丢失忘记加载持久化或 db 路径错误检查 SQLite 文件路径和初始化逻辑5. 进阶优化把 context-mode 提升到好用级别5.1 多级记忆短期、中期、长期分层处理基础版本把历史消息分成原始消息和单层摘要聊到超级长的对话时单层摘要也会越积越多最终摘要本身变成负担。我的优化方案是引入分层摘要体系当摘要消息本身也超过预算的一半时把旧的摘要再压缩成更高层级的主题简报。这就形成了两层甚至三层的记忆金字塔——底层是原始消息中层是对话摘要顶层是跨会话的主题简报。实施起来不难压缩器复用了和之前一样的逻辑只是输入从原始消息变成摘要消息列表。效果上一个原本聊了 300 轮的长会话最终带给模型的上下文可以压缩到 2000 Token 以内且关键决策信息基本不丢。这对长时任务例如 AI 辅助项目管理帮助极大。5.2 语义检索增强只捞最相关的历史压缩器是被动遗忘语义检索是主动回忆。我给 context-mode 加了一个可选模块向量检索引擎。实现思路是每收到一条用户消息先用嵌入模型算出向量存在向量数据库里轻量场景完全可以用 sqlite-vss 或 Chroma。调度器组装消息时除了常规保留最近 N 条消息还会根据当前用户问题的向量去检索最相关的历史消息按相似度排序后插入上下文。这一组合让 context-mode 具备了该记的全记得该忘的全忘掉的能力。比如用户周二问了数据库索引优化周五又问了一个看似无关但对索引有依赖的问题系统能自动把周二那轮的关键结论捞回来而不是靠摘要里的模糊信息。不过要提醒的是向量检索会引入额外的延迟和成本。我的实际经验是只有当单个 session 的消息数超过 100 条时才触发检索短对话直接用窗口调度就足够了。5.3 成本控制结合 token_count 做调用前预估最后分享一个成本优化小技巧。我在 add_message 时保存了每条消息的 token_count这不仅是调度器预算判断的依据还可以用来做请求前的成本预估。每次构建请求前统计 messages 列表的总 Token 数再乘以单 Token 单价就能实时算出这一条 API 调用的预估成本。在开发调试阶段我还会把每一轮对话的 Token 使用情况打印出来方便自己观察调度器是否按预期工作。别小看这个日志它是我排查预算问题的最重要抓手。6. 结尾的一点私货我现在还在用什么文章写到这里最后再补充一些我个人的体会。context-mode 这个方案我从零写到现在前后迭代了差不多一个月最大的感触是AI 应用开发的复杂度大头不在模型选择也不在 prompt 技巧而是在于怎么组织信息和状态。谁把上下文管明白了谁的用户体验就上了一个台阶。目前这套逻辑我已经整合进我的终端助手项目里用 SQLite 做存储、tiktoken 做计数、gpt-4o-mini 做压缩整套方案开销很低单用户单会话的增量成本可以忽略不计。如果你也在做类似的东西我的建议是先别贪多把记忆体 窗口调度 压缩器这个最小闭环跑通再去考虑向量检索和多层摘要。核心链路稳定之后其他能力都是往里加插件的事。最后一个小技巧送给认真看到这里的读者调试上下文管理时务必把每一轮请求实际发出去的 messages 列表完整打出来哪怕打成 JSON 文件。你会非常直观地看到模型到底看到了什么这才是排查上下文问题的唯一正道。