"我刚才说的那个方案,你直接给个结论吧。"用户这句话发过来的时候,我盯着后台日志看了半天。AI助手回复的是:"请问您指的是哪一个方案?我还需要更多上下文。"那一刻我就知道,这个助手又"失忆"了。
这已经是这周第三起同类客诉。翻看代码,原因一目了然:我们把最近20轮对话全部塞进系统提示词里,窗口一满就直接清空,没有任何分级、压缩、检索机制。用行业的话说,这个对话系统完全没有设计context-mode(上下文模式)。
过去一年我一直在做对话式AI应用,最大的体会是:模型选型决定能力上限,上下文管理决定体验下限。一个再强的模型,如果喂给它的上下文是乱七八糟的、过期的、超限的,它照样给你输出幻觉和错误结论。这篇文章我就把context-mode这件事彻底讲透,包括它是什么、有哪些实现思路、核心代码怎么写、以及我实际踩过的坑。
1. Context Mode到底是什么:先搞清楚我们为什么需要它
1.1 大模型的"金鱼记忆"与上下文窗口困境
如果你做过对话类产品,一定遇到过这种情况:用户连续聊了七八轮之后,助手开始答非所问,把之前的约定、偏好、中间结论全忘了。原因很简单,大模型本身是无状态的——它不记得任何历史对话,每一次请求都需要你把上下文完整塞进去,它才能"看起来记得"。
但塞多少是个问题。ChatGPT刚出来时上下文窗口是4K token,后来一路卷到8K、32K、128K甚至更大,但窗口再大也是有限的。更现实的问题是:窗口越长的模型,推理成本越高、首字延迟越大。你不可能无限制地往里塞东西。
context-mode就是解决这个问题的运行策略集合。它管的是三件事:哪些内容应该放进来、哪些内容应该压缩、哪些内容应该丢出去。
我见过很多团队做AI应用,第一版都是"无脑拼接":把聊天记录字符串拼一拼,拼到截断为止。这在Demo阶段完全没问题,用户聊三五轮也看不出毛病。但一旦真实上线,用户会长时间对话、会穿插新问题、会反复引用早期信息。上下文模式没做好,产品体验就是崩塌的。
1.2 三种主流上下文模式的思路对比
我把它分成三类,基本覆盖了市面上绝大多数对话系统的做法。
第一种:全量窗口模式(Full-Window Mode)
把最近N轮对话全部拼接进上下文,超过窗口就从最老的开始丢弃。实现最简单,但问题也很明显:丢弃即遗忘,早期关键信息可能被丢掉;而且大量低价值闲聊霸占窗口,token浪费严重。
第二种:滚动摘要模式(Rolling Summary Mode)
维护一份"历史摘要",随着对话推进不断调用模型把旧内容浓缩成摘要,每次请求把摘要+最近几轮原始对话一起塞进窗口。这是目前最主流的方案,LangChain的ConversationSummaryBufferMemory就是典型实现。
第三种:混合检索模式(Hybrid Retrieval Mode)
把历史对话向量化存储,每次请求根据当前用户问题,从历史中检索最相关的片段加入上下文。优点是不再局限于"最近",早期关键信息也能被翻出来;缺点是引入检索链路,有一堆自己的坑。
三种模式不是互斥的,成熟系统通常是"摘要+检索+窗口"三合一。我把它们的差异整理成了一张表,方便对比:
| 模式 | 优点 | 缺点 | 典型场景 |
|---|---|---|---|
| 全量窗口 | 实现快、零失真 | 窗口利用率低、早期信息必丢 | 短对话、评测Demo |
| 滚动摘要 | 信息密度高、成本可控 | 摘要会丢细节、必须调摘要触发点 | 客服助手、多轮任务 |
| 混合检索 | 能召回早期信息 | 检索噪声、延迟和成本增加 | 知识密集、长对话场景 |
无论选哪种,核心都在于一个东西:上下文管理器(Context Manager)。下面我重点讲怎么设计它。
2. 上下文模式的核心设计:从消息记录到可控记忆
2.1 消息数据结构:别把上下文当成普通日志
很多初学者写上下文管理,直接用List[String]存消息。一旦要做压缩、检索、按角色隔离,这种结构就废了。我用的是带元数据的消息对象。
每一个消息至少应该包含:角色(user/assistant/system)、内容、时间戳、消息ID。如果有条件,把"来源渠道"(比如来自哪个页面、哪个客户会话)也加上。为什么要这些字段?
有了时间戳,你才能做时间衰减或者按时间范围裁剪;有了消息ID,你才能做去重和引用关系链;有了来源字段,你在做检索时才能按业务维度过滤。我甚至见过有的团队在消息里加importance_score字段,人工标记某些关键节点消息(比如用户下订单、修改配置)的重要性权重,压缩时优先保留。这个思路很实用。
字段设计得多细致,直接决定了后面压缩策略和你能做出的检索效果。我建议从第一版就按结构化消息来规划,别等到系统上线后再重构。这个教训我是付过学费的——一版项目为了赶进度用了纯字符串数组,后来加摘要功能的时候,全量消息都要迁移,费了整整两天。
2.2 写入策略:什么东西值得占窗口
context-mode的另一个核心问题是"写入控制"。很多人只看淘汰,忽略了写入端的设计。实际上,你在把一条消息放入上下文之前,就应该判断它值不值得进。
我推荐做三级分层:
- 第一层:原始消息存储。所有对话都存,放在数据库里,不占上下文窗口。
- 第二层:热窗口。最近N轮原样保留在上下文中,保证模型对"当下正在聊的事"有完整感知。
- 第三层:摘要与索引层。旧消息浓缩成摘要,并按主题、实体、时间建立索引,方便后续检索。
写入策略举例:如果用户消息是"好的""继续""谢谢"这类短反馈,能不能直接并入上一轮而不单独占一条?可以直接把前一条消息追加标记,而不是新开一条记录。再比如系统主动推送的通知类消息,对后续对话没有实质影响,就应该直接进摘要层,不占用热窗口。
这套逻辑用一句话概括就是:上下文窗口应该只放"对当前决策有用"的信息,其余进数据库。你在写代码前先想清楚这个原则,后续压缩和检索都会轻松很多。
2.3 淘汰与压缩:窗口满了之后怎么办
窗口快满时,要有明确的处置顺序。我的经验是四步走,按优先级排列:
- 去重与合并:多轮中重复的表述合并成一条,比如用户反复说"帮我快点",不必在上下文里出现三遍。
- 细节降级:保留结论,删除过程。用户问"这个东西怎么安装",AI回复了详细步骤,后续对话里保留"已告知安装步骤"即可。
- 摘要化:把最老的若干轮调用模型生成摘要。
- 转移存储:如果摘要层也膨胀了,就把最早的摘要归档到数据库,只保留摘要的摘要(层级摘要)。
压缩之后,必须有一个压缩日志。我建议记录每次压缩的时间、原token量、压缩后token量、丢弃了哪些消息ID。为什么?因为用户投诉"你说过的又忘了"的时候,你得能追溯是哪个环节弄丢的。我遇到过一种情况:压缩策略每次都把上一轮总结直接丢掉了,把用户的关键指令(比如"不要用短信通知我")弄丢了,结果用户收到短信后炸毛。没有压缩日志,这类问题连查都无从查起。
3. 手写一个Context Manager:架构与关键代码
3.1 模块划分:存储、策略、生成
有了前面的设计,就可以落地代码了。我不建议一上来就套LangChain之类的框架,先用自己手写的版本跑通,你会对context-mode的每一步有更深的体感。整个管理器我拆成三个模块:
- 存储模块:负责追加消息、按条件读取历史。
- 策略模块:负责token估算、压缩触发决策、检索排序。
- 生成模块:负责把热窗口、摘要、检索结果组装成发给模型的最终上下文。
下面给出一份可运行的简化实现,覆盖核心链路。语言用Python,数据库先用list模拟,方便你理解主流程。
3.2 核心实现:ContextManager类
from dataclasses import dataclass, field from typing import List, Optional import time import hashlib # 消息类型 ROLE_USER = "user" ROLE_ASSISTANT = "assistant" ROLE_SYSTEM = "system" ROLE_SUMMARY = "summary" @dataclass class Message: role: str content: str ts: float = field(default_factory=time.time) msg_id: str = field( default_factory=lambda: hashlib.md5( str(time.time()).encode() ).hexdigest() ) importance: int = 1 # 重要性评分,默认1,越高越难被压缩丢弃 class ContextManager: def __init__( self, max_window_tokens: int = 4000, summary_trigger_tokens: int = 3000, summary_model=None, embed_model=None, ): # 热窗口大小上限 self.max_window_tokens = max_window_tokens # 触发摘要的阈值 self.summary_trigger_tokens = summary_trigger_tokens # 生成摘要用的模型,传入一个 callable: (messages) -> str self.summary_model = summary_model # 向量化模型,传入一个 callable: (str) -> list[float] self.embed_model = embed_model # 原始消息存储(模拟数据库) self.all_messages: List[Message] = [] # 当前热窗口 token 数 self.current_window_tokens = 0 # 摘要缓存 self.summary_cache: List[str] = [] def _estimate_tokens(self, text: str) -> int: # 粗略估算:中文约1.5字符/token,英文约4字符/token # 生产环境建议用 tiktoken 或对应模型的分词器 return max(1, int(len(text) / 1.5)) def add_message(self, role: str, content: str, importance: int = 1): msg = Message(role=role, content=content, importance=importance) self.all_messages.append(msg) # 只有 user/assistant/system 进热窗口,summary 直接放摘要缓存 if role != ROLE_SUMMARY: self.current_window_tokens += self._estimate_tokens(content) else: self.summary_cache.append(content) # 写完之后检查是否要压缩 self._maybe_compress() return msg def _maybe_compress(self): # 还没超阈值,不处理 if self.current_window_tokens <= self.summary_trigger_tokens: return # 按重要性排序,先处理 importance 低的消息 # 简化实现:把最老的若干轮合并摘要 messages = [m for m in self.all_messages if m.role != ROLE_SUMMARY] # 找出热窗口中最老的、且importance=1的消息 to_summarize = [] i = 0 while i < len(messages) and self.current_window_tokens > self.max_window_tokens: m = messages[i] if m.importance <= 1: to_summarize.append(m) self.current_window_tokens -= self._estimate_tokens(m.content) i += 1 if not to_summarize: # 所有消息都重要,就只能丢最老的(极端情况) m = messages[0] to_summarize.append(m) self.current_window_tokens -= self._estimate_tokens(m.content) # 调用摘要模型生成摘要 if self.summary_model: summary_text = self.summary_model([ {"role": m.role, "content": m.content} for m in to_summarize ]) self.add_message(ROLE_SUMMARY, summary_text) def retrieve(self, query: str, top_k: int = 3) -> List[Message]: # 如果没配置向量模型,就用简单的关键词重叠评分 if not self.embed_model: q_terms = set(query.lower().split()) scored = [] for m in self.all_messages: m_terms = set(m.content.lower().split()) score = len(q_terms & m_terms) if score > 0: scored.append((score, m)) scored.sort(key=lambda x: -x[0]) return [m for _, m in scored[:top_k]] # 配置了向量模型,做向量相似度检索 q_vec = self.embed_model(query) scored = [] for m in self.all_messages: m_vec = self.embed_model(m.content) sim = self._cosine_sim(q_vec, m_vec) scored.append((sim, m)) scored.sort(key=lambda x: -x[0]) return [m for _, m in scored[:top_k]] @staticmethod def _cosine_sim(a, b): import math if len(a) != len(b): return 0.0 dot = sum(x * y for x, y in zip(a, b)) norm_a = math.sqrt(sum(x * x for x in a)) norm_b = math.sqrt(sum(x * x for x in b)) if norm_a == 0 or norm_b == 0: return 0.0 return dot / (norm_a * norm_b) def build_context(self, current_query: str, include_summary: bool = True) -> List[dict]: # 组装最终上下文:摘要 + 检索片段 + 热窗口 + 当前问题 ctx_messages = [] if include_summary and self.summary_cache: combined_summary = "\n".join(self.summary_cache[-3:]) ctx_messages.append({ "role": "system", "content": f"以下是对历史对话的摘要:{combined_summary}", }) # 检索相关历史片段 retrieved = self.retrieve(current_query, top_k=3) if retrieved: hist_text = "\n".join( f"{m.role}: {m.content}" for m in retrieved if m.role != ROLE_SUMMARY ) ctx_messages.append({ "role": "system", "content": f"以下是与当前问题相关的历史片段:{hist_text}", }) # 热窗口消息 for m in self.all_messages: if m.role == ROLE_SUMMARY: continue if self._estimate_tokens(m.content) <= 0: continue ctx_messages.append({"role": m.role, "content": m.content}) # 当前问题 ctx_messages.append({"role": ROLE_USER, "content": current_query}) return ctx_messages这份代码就是context-mode的最小骨架。add_message负责写入并监控token水位;_maybe_compress负责在超阈值时触发压缩;retrieve负责从历史中召回相关片段;build_context负责最终组装。你可以直接跑通,再逐步加上数据库、向量存储和更细的压缩策略。
3.3 参数配置实测:max_window_tokens、summary_trigger_tokens、top_k怎么定
代码写出来只是第一步,参数才是真功夫。我调参的体感如下:
max_window_tokens:不要设成模型窗口的上限。比如GPT-4o有128K窗口,不等于你要用满128K。窗口越大,单次请求的推理耗时和费用越高,而且模型对超长上下文的"注意力分散"效应会放大。我的经验是:热窗口控制在总窗口的1/4到1/3。比如模型窗口是32K,热窗口设8K~10K比较舒服,剩下的留给摘要、检索结果、系统提示词和输出token。输出token也要留空间,很多模型把输出都算进上下文窗口里,不留够就会报错。
summary_trigger_tokens:这个值建议比max_window_tokens低20%~30%。比如max_window_tokens=8000时,trigger设6000。为什么留这个缓冲?因为压缩操作本身需要时间,而且压缩过程中还可能插入新消息,如果不设缓冲,很容易在压缩完成前窗口就爆掉。我第一版就是trigger=max_window_tokens,结果在高并发下频繁触顶,直接请求失败。
top_k:检索召回数量也不是越大越好。我试过top_k=5和top_k=3,前者在很多场景下反而效果更差——夹带了太多不相关片段,模型被噪声带偏。对于普通客服类场景,top_k=3够用;如果是复杂的研究型问答,可以分两次检索,一次查用户历史,一次查知识库,然后各取top_k=2,再让模型综合去噪。
这里说的所有参数,都不是一次性定死的。建议你把它们在配置中心里做成可动态调整的开关,上线后依据「关键词召回命中率」和「用户追问率」这两个指标持续调。追问率如果升高,往往说明上下文丢信息了,需要提高trigger阈值或加大top_k。
4. 实战中踩过的坑:Context Mode问题排查实录
4.1 窗口还是爆了:预压缩与事中压缩
我第一次上线context-mode时,是把压缩逻辑放在用户看到回复之后。结果高峰期还是频繁爆窗口。后来想明白了:对话是按轮次进行的,用户问一句,AI答一句,如果AI回答的内容特别长(比如生成了几千token的代码),这条回复会瞬间撑爆热窗口。
解决办法是加预压缩——在把当前对话写入热窗口之前,先检查"现有窗口+这条消息会不会超限",如果会,就先触发一次压缩再写入。另一个关键动作是对模型输出长度做上限约束,在system prompt里明确限制输出最长多少token。该限制的没限制,context-mode做得再细也扛不住。
我还发现一个反直觉的点:压缩不是压缩得越频繁越好。频繁压缩会带来两个问题,一是调用摘要模型的额外费用,二是每次摘要都会损失细节,压缩得越频繁,累积失真越严重。所以正确的做法是"能少压就少压",尽量通过写入端的过滤机制减少进窗口的消息量。
4.2 摘要后AI"变傻":关键信息恢复策略
有一次版本上线后,用户反馈AI突然记不住"用企业邮箱注册"这个前提条件了。查日志发现,这条信息出现在第6轮对话里,早就被摘要吞掉了,而摘要生成时模型没有强调它。
这就是摘要模式的通病:摘要模型认为不重要的信息,可能恰恰是用户的硬性约束。我的解决思路是加关键信息标记机制。在写入端检测特殊信号,比如用户说"注意""千万别""记住""不要",把这些消息的importance标记为5以上,在压缩时优先保留原文。同时,在生成摘要时,用prompt专门要求:"如原文包含用户的明确要求、禁止事项、偏好设置,必须原样保留而不要改写。"这两个动作加起来,基本能解决90%以上的"摘要丢关键信息"问题。
4.3 检索带上噪声对话:加角色隔离
混合检索模式上线测试的时候,又出现一个搞笑场景:用户在聊天里说"我要refer到你说的那个Java问题",结果检索系统把用户在某一条无关对话里的随口抱怨"这代码真像一坨Java屎"也召回进上下文了。模型一看,直接理解成用户在抱怨Java,回答就偏了。
问题出在角色隔离缺失:检索应该按角色限权。用户问题和用户历史对话是一类语义,助手回复和历史背景是另一类语义。我的修改方案是:在向量化存储时,把role作为一个过滤字段;检索用户问题时,只召回role=user的历史消息,在需要召回AI历史回复时,再单独带一组role=assistant的检索结果。此外,还要在检索时排除纯闲聊消息,判据是消息长度太短(低于5个字)或带大量表情符号。
4.4 成本与延迟的双重压力
context-mode做完整后,系统每次请求要调数据库、计算向量、可能还要调摘要模型,延迟比裸拼接版本高了不少。一开始我每轮都清点全部历史并做一次完整检索,成本直接翻了三四倍。
优化手段有这几个:
- Query改写:用户消息先经过一层轻量改写,把"它""那个方案"这类指代补充完整,再做检索,召回准确率高很多。
- 缓存复用:摘要不是每次请求都重新生成的,只要没有新消息加入,就复用上次的摘要结果。
- 检索降频:不是每一轮都需要检索。如果当前问题跟热窗口内已有内容高度重叠(用向量相似度判断),可以直接跳过检索,省一次嵌入计算。
- 批量嵌入:如果embed_model支持batch模式,把多条历史消息批量嵌入比逐条嵌入省一半时间。
延迟优化是一个持久战,不要指望一次到位。我用一个简单的"压测+监控"组合:每次发版前用固定脚本压100轮对话,统计平均首token延迟和超时率;另加日志监控摘要模型调用次数和检索耗时,一旦超过阈值就告警。
5. 工具链参考:框架内置的上下文模式能不能直接用
5.1 LangChain:ConversationSummaryBufferMemory的局限
很多团队做AI应用绕不开LangChain,它确实内置了ConversationSummaryBufferMemory,翻译过来就是"对话摘要缓冲记忆",本质上就是滚动摘要模式的一个封装。我试过,优点是开箱即用,但有三点局限:
第一,它只支持"最近轮次+摘要"的组合,不支持检索历史片段。第二,压缩触发时机只按token数阈值,不支持按重要性评分决定丢弃优先级。第三,它对消息结构有抽象约束,一旦你要自定义字段(比如来源渠道、情感标签)就得自己绕开memory接口。
所以我的判断是:**能用LangChain做demo,但没有必要在生产环境把上下文管理的命脉绑死在它上面。**如果你的业务只需要短对话、轻量场景,直接用没问题;如果你的系统要做精细控制,还是自己手写Context Manager更可控。
5.2 LlamaIndex:ChatMemoryBuffer与向量检索结合
LlamaIndex在索引和检索方面更强。它的ChatMemoryBuffer负责维护聊天记忆,也可以配合VectorStoreIndex把历史消息向量化,实现"聊天历史按需检索"。这种方式很适合已经上了LlamaIndex知识库体系的团队,可以减少自研工作量。
但要注意,LlamaIndex默认的聊天历史检索跟RAG检索并不是同一套体系。如果你既要做知识库RAG,又要做历史对话召回,需要把两个检索链路明确分离。否则模型会把"用户过去说过的话"和"知识库文档"混为一谈,回答时就会编造用户的历史信息。这我踩过,教训很深刻。
5.3 我的选型建议
我把这些年看到的团队做法总结一下,大致分三档:
- 第一档,刚起步/短对话:直接用框架内置memory,跑通流程再说。关键是先别过度设计。
- 第二档,有用户画像/较长会话:自研结构化消息存储+滚动摘要,用LangChain或LlamaIndex的组件辅助生成摘要。
- 第三档,知识密集/长会话/高复杂度:自研完整ContextManager,建议用Postgres+pgvector做向量存储,检索和摘要全部自控。
选型决策的核心标准就一句话:上下文管理对你的产品是核心壁垒还是边缘支撑?如果是核心壁垒,一定要掌握底层细节,不能只当一个框架调用者;如果是边缘支撑,用现成工具快速上线,然后把精力放在更重要的业务逻辑上。
6. 从Context Mode到更广阔的记忆体系
最后再分享一个我最近在思考的方向。Context Mode解决的是"这次对话怎么写上下文",但更长远的问题是:用户的跨会话记忆怎么办?比如用户周一在客服里说了"我习惯用邮箱收通知",周五又来了,新会话里AI助手是否还记得?
单靠context-mode解决不了跨会话问题。我的思路是增设一个用户级偏好库:在对话中提取用户偏好项,比如"通知方式=邮箱""时间段=工作日上午",存成结构化的profile。每次新会话开始时,把与当前场景相关的profile注入系统提示词,再配合context-mode正常工作。这套"偏好库+上下文模式"的组合,才真正接近人类的"记住你"。
再进一步,还可以做知识图谱化的记忆:把对话中提到的实体、关系、偏好抽出来,构建一张轻量级的知识图。这样用户在后续会话中问起"上次说的那个供应商后来又怎样了",系统可以借助知识图谱精准定位到对应历史片段,比纯向量检索的语义模糊性要强得多。
我这里给出一个个人偏好的架构参考:
- 会话级:Context Manager(热窗口+摘要+检索)
- 用户级:Profile Store(长期偏好、关键约束)
- 知识级:轻量知识图谱(实体、关系、跨会话引用)
这三层加起来,才算是一个相对完整的AI记忆体系。context-mode只是其中第一层,但也是最重要的一层——没有会话级上下文管好,后面两层的输入都不可靠。
在我个人的实践中,最让我受用的一句话是:上下文不是"存得越多越好",而是"该记住的恰好都在"。判断一个上下文系统好不好,不是看它的窗口多大,而是看它在关键时候有没有把最relevant的信息端到模型面前。
如果你正在做对话类AI应用,不管用的是闭源API还是开源模型,我建议你从第一轮对话就开始考虑上下文管理,不要等用户数量上来、客诉堆积了才返工。先把热窗口、摘要触发点、检索召回这三件事跑通,你已经超过了市面上大半的同类产品。
至于更高级的记忆体系,那是在context-mode稳定之后的事。先把地基打好,再谈高楼。