ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

上下文模式工程实践:从滑动窗口到分层检索的落地指南

上下文模式工程实践:从滑动窗口到分层检索的落地指南 1. 从“上下文模式”说起一个被低估的工程概念第一次看到“context-mode”这个词很多人会下意识地把它归到某个具体框架的配置项里觉得无非是个开关或者枚举值。但真正在项目里踩过坑的人会明白上下文模式本质上是一套关于“信息在什么时机、以什么粒度、被谁读取和改写”的约定。它决定了你的系统是清爽高效还是随着功能堆叠变成一团乱麻。我接触这个概念最早是在做多轮对话系统的时候。当时团队里有个争论用户的历史消息到底应该全量塞给模型还是只保留最近几轮全量塞进去token 消耗爆炸响应变慢只保留最近几轮用户前面说过的关键约束又丢了。后来我们引入了一套上下文模式的管理机制把“什么信息进入上下文”这件事从业务代码里抽离出来做成可配置、可观测的独立层问题才真正收敛。所以这篇内容我想聊的不是某个库的 API 怎么调而是context-mode 这套思路在实际工程里怎么落地。它适合谁看如果你正在做对话系统、Agent 编排、RAG 检索增强或者任何需要维护“会话状态”的服务那这套东西你迟早要面对。哪怕你只是写一个带多步表单的前端应用理解上下文模式也能帮你把状态管理写得更干净。接下来我会从设计思路、核心细节、实操落地到问题排查把我自己趟过的路完整讲一遍。2. 上下文模式的整体设计与选型逻辑2.1 为什么需要把“上下文”单独抽象出来在没有上下文模式这个概念之前大多数项目的做法是把会话状态直接挂在业务对象上。比如一个聊天服务每个 session 对象里存一个 messages 数组来一条消息就 push 进去要回复的时候把整个数组拼成 prompt。这种做法在 demo 阶段没问题但一旦上线就会暴露三个致命问题。第一个问题是耦合。业务逻辑和上下文拼装逻辑混在一起改一个 prompt 模板要翻遍半个代码库。第二个问题是不可控。上下文长度随着对话轮次线性增长某天突然发现接口超时排查半天才意识到是某个用户聊了两百轮。第三个问题是不可观测。你根本不知道每次请求实际送进模型的内容是什么出了问题只能靠猜。把上下文抽象成独立的一层本质上是把“状态管理”和“业务处理”解耦。这跟后端开发里把缓存层、数据访问层独立出来是一个道理。上下文模式要回答的核心问题就三个存什么、留多久、怎么取。这三个问题想清楚了剩下的都是实现细节。2.2 几种常见的上下文模式对比在实际项目里我见过也用过的上下文模式大致可以归为四类各有各的适用场景。下面这张表是我自己整理的对比参数是基于常见业务量级估算的你可以根据实际情况调整。模式类型核心思路适用场景上下文长度增长实现复杂度全量累积所有历史消息全部保留短对话、调试阶段线性增长低滑动窗口只保留最近 N 轮通用对话、客服恒定低摘要压缩旧消息压缩成摘要长对话、陪伴类缓慢增长中分层检索按相关性动态召回RAG、知识问答可控高滑动窗口是最常用的起点。我一般会把窗口大小设成 10 到 20 轮具体取决于单轮消息的平均长度。假设单轮平均 200 token20 轮就是 4000 token加上系统提示词和当前输入控制在 6000 token 以内对大多数模型来说都很轻松。这个数字不是拍脑袋来的你可以用tiktoken这类工具先统计一下自己业务里消息的 token 分布取中位数再乘以轮数。摘要压缩适合那种用户会聊很久的场景。做法是当消息数超过阈值时把最早的一批消息交给模型生成一段摘要用摘要替换原始消息。这里有个坑摘要本身也会消耗 token而且摘要质量不稳定。我的经验是摘要触发阈值不要设太低至少积累 30 轮以上再压缩否则频繁调用模型做摘要成本和延迟都受不了。分层检索是 RAG 场景的标配。它不按时间顺序保留上下文而是把历史消息存进向量库每次请求时根据当前问题检索最相关的若干条。这种模式的关键在于检索质量如果召回不准模型拿到的上下文就是噪音。我后面会专门讲怎么调这块。2.3 选型时最容易忽略的两个维度大多数人在选上下文模式时只看“对话轮次”这一个维度但实际工程里还有两个维度同样重要甚至更关键。第一个是上下文的时效性要求。有些场景里旧信息就是应该被遗忘的。比如一个订票助手用户三小时前问的航班信息现在早就没意义了留着反而干扰模型判断。这种场景下滑动窗口比摘要压缩更合适因为摘要会把过时信息也保留下来。反过来如果是长期陪伴类应用用户一个月前说过的偏好你希望它一直记得那就得用分层检索或者持久化存储。第二个是多用户隔离的粒度。上下文是按 session 隔离还是按 user 隔离还是按 tenant 隔离这直接决定了你的存储结构。我见过一个项目上下文按 session 存结果用户换个设备登录历史全丢了体验很差。后来改成按 user 维度存session 只是 user 上下文的一个视图问题才解决。这个设计决策要在项目早期定下来后期改代价很大。3. 核心细节解析与实操要点3.1 上下文的数据结构设计上下文模式落地第一步是设计数据结构。我推荐的结构是一个带元信息的消息列表而不是简单的字符串数组。每条消息至少包含这几个字段角色role、内容content、时间戳timestamp、token 数token_count、以及一个可选的标签tags。为什么要存 token 数因为你要做窗口裁剪每次现算太浪费。在消息写入的时候就统计好裁剪时直接累加效率高很多。标签字段是给分层检索用的比如你可以给消息打上“偏好”“约束”“事实”这样的标签检索时按标签过滤比纯向量相似度更准。时间戳的作用容易被低估。除了做时效性判断它还能帮你做上下文的重排序。有些场景下最近的消息不一定最重要但时间信息能帮你判断哪些是“当前有效”的。我一般会把时间戳存成 Unix 毫秒方便做区间查询。存储介质的选择上短上下文比如滑动窗口直接放内存或者 Redis 就够了读写快。长上下文分层检索需要持久化我一般用 PostgreSQL 存原始消息用向量库存 embedding。两者通过消息 ID 关联检索时先查向量库拿到 ID再回表取完整内容。3.2 窗口裁剪的边界条件处理滑动窗口看起来简单但边界条件处理不好会出大问题。最常见的坑是把工具调用和工具返回拆散了。比如模型发起了一个函数调用返回结果在下一轮如果你裁剪时正好把调用那轮删了只留下返回结果模型就会一脸懵。我的处理方式是按“对话回合”而不是“单条消息”来裁剪。一个回合包含用户输入、模型回复、以及可能的工具调用和返回它们作为一个整体要么全留要么全删。实现上可以给每条消息打一个 turn_id裁剪时按 turn_id 分组从最老的回合开始删直到总 token 数低于阈值。另一个坑是系统提示词被误删。系统提示词通常放在消息列表最前面裁剪时如果不做特殊处理很容易被当成最老的消息删掉。我的做法是把系统提示词单独存一个字段不放进消息列表每次拼装上下文时再合并进去。这样既避免了误删也方便动态调整系统提示词。还有一个细节是裁剪后的首条消息角色。如果裁剪完第一条是 assistant 消息有些模型会报错因为它期望对话以 user 开头。处理方式很简单裁剪后检查首条消息角色如果是 assistant就再往前删一条或者插入一条空的 user 消息占位。这个细节不处理线上会偶发 400 错误排查起来很费劲。3.3 摘要压缩的触发时机与质量控制摘要压缩的核心难点在于什么时候触发和摘要质量怎么保证。触发时机上我试过两种策略。一种是按 token 数触发当上下文总 token 超过阈值比如 8000时压缩。另一种是按消息数触发超过 50 条就压缩。实测下来按 token 数更合理因为消息长度差异很大按条数容易误判。但 token 数统计有延迟所以我会设一个软阈值和硬阈值软阈值到了就开始准备压缩硬阈值到了强制压缩。摘要质量的控制关键在于给模型的摘要指令要具体。不要只说“总结以下对话”而要明确告诉它保留什么、丢弃什么。比如“请总结以下对话保留用户提到的所有偏好、约束条件和已确认的事实丢弃寒暄和重复内容。输出格式为要点列表。”这样生成的摘要信息密度高后续用起来更可靠。摘要生成后我一般会做一次校验把摘要和最近几轮原始消息一起送给模型看它能不能正确回答一个依赖旧信息的问题。如果答不对说明摘要丢了关键信息需要调整指令重新生成。这个校验步骤会增加成本但在关键业务里值得做。3.4 分层检索的召回策略调优分层检索的效果八成取决于召回策略。我踩过的坑是只依赖向量相似度结果经常召回一些语义相近但实际无关的消息。后来改成混合召回效果明显提升。混合召回的做法是向量相似度占 70% 权重时间新鲜度占 20%标签匹配占 10%。时间新鲜度的计算方式是1 / (1 小时差)这样最近的消息得分高但旧消息也不会完全被忽略。标签匹配是精确匹配如果当前问题带了“偏好”标签就优先召回带同样标签的消息。召回数量上我一般取 top 5 到 top 8。太少覆盖不全太多引入噪音。这个数字可以用 A/B 测试调观察不同召回数量下模型回答的准确率。我做过一轮测试召回 6 条时准确率最高再多就开始下降因为噪音盖过了有效信息。还有一个技巧是对召回结果做去重。向量检索经常召回内容高度相似的多条消息白白占用上下文空间。我一般用简单的文本相似度比如 Jaccard 系数做去重相似度超过 0.8 的只保留最新一条。4. 实操过程与核心环节实现4.1 环境准备与依赖安装先把基础环境搭起来。我用 Python 做示例因为生态最全。需要装的包不多核心是 token 统计和向量检索两块。pip install tiktoken numpy pip install sentence-transformers pip install psycopg2-binarytiktoken用来统计 token 数sentence-transformers用来生成 embeddingpsycopg2用来连 PostgreSQL。如果你用别的向量库比如 FAISS 或者 Milvus把对应的客户端包换掉就行。数据库这边PostgreSQL 需要装一个向量扩展。我用的是 pgvector安装方式取决于你的系统这里不展开。建表语句大概长这样CREATE TABLE messages ( id BIGSERIAL PRIMARY KEY, session_id VARCHAR(64) NOT NULL, turn_id INT NOT NULL, role VARCHAR(16) NOT NULL, content TEXT NOT NULL, token_count INT NOT NULL, tags TEXT[], created_at TIMESTAMP DEFAULT NOW() ); CREATE INDEX idx_session_turn ON messages(session_id, turn_id);embedding 我单独存一张表通过 message_id 关联。这样原始消息和向量分开管理更新起来灵活。4.2 上下文管理器的核心实现上下文管理器是整个模式的核心它对外暴露的接口应该尽量简单。我设计的接口只有三个方法add_message、get_context、clear。内部逻辑全部封装起来业务代码不需要关心窗口怎么裁、摘要怎么压。import tiktoken from datetime import datetime class ContextManager: def __init__(self, session_id, max_tokens8000, window_turns20): self.session_id session_id self.max_tokens max_tokens self.window_turns window_turns self.encoder tiktoken.get_encoding(cl100k_base) self.system_prompt self.messages [] def add_message(self, role, content, turn_id, tagsNone): token_count len(self.encoder.encode(content)) self.messages.append({ role: role, content: content, turn_id: turn_id, token_count: token_count, tags: tags or [], timestamp: datetime.now().timestamp() }) def get_context(self): # 先按窗口裁剪 trimmed self._trim_by_window() # 再按 token 裁剪 trimmed self._trim_by_tokens(trimmed) # 拼装最终上下文 return [{role: system, content: self.system_prompt}] trimmed_trim_by_window的逻辑是按 turn_id 分组保留最近的 N 个回合。_trim_by_tokens是从最老的回合开始删直到总 token 数低于阈值。两个方法都返回新的消息列表不修改原始数据这样方便做回滚和调试。这里有个实现细节token 统计我用的是cl100k_base编码这是 GPT 系列常用的。如果你用的是别的模型比如 Claude 或者国产模型编码方式不一样需要换成对应的 tokenizer。统计误差控制在 5% 以内就够用了不用追求完全精确。4.3 摘要压缩的完整流程摘要压缩我单独写了一个类因为它涉及模型调用逻辑比较重。触发条件是上下文 token 数超过软阈值且消息数超过最小压缩条数。class Summarizer: def __init__(self, llm_client, soft_threshold6000, min_messages30): self.llm llm_client self.soft_threshold soft_threshold self.min_messages min_messages def should_compress(self, messages): total sum(m[token_count] for m in messages) return total self.soft_threshold and len(messages) self.min_messages def compress(self, messages): # 保留最近 10 条其余压缩 to_compress messages[:-10] recent messages[-10:] prompt self._build_summary_prompt(to_compress) summary self.llm.generate(prompt) summary_msg { role: system, content: f[历史摘要] {summary}, turn_id: -1, token_count: len(self.encoder.encode(summary)), tags: [summary], timestamp: datetime.now().timestamp() } return [summary_msg] recent_build_summary_prompt里我把摘要指令写得很具体明确要求保留偏好、约束、事实三类信息。实测下来这样生成的摘要比泛泛总结的信息密度高不少。摘要消息的 turn_id 设成 -1是为了在后续裁剪时能被识别出来避免被当成普通消息处理。压缩完成后我会把摘要消息和最近消息一起返回替换原来的消息列表。注意这里要更新存储否则下次请求又拿到旧数据。我一般用事务包起来保证一致性。4.4 分层检索的落地实现分层检索需要向量库支持我用 pgvector 做示例。先把消息的 embedding 存进去检索时按混合策略召回。class Retriever: def __init__(self, db_conn, embed_model): self.db db_conn self.embed embed_model def index_message(self, message_id, content): vec self.embed.encode(content) self.db.execute( INSERT INTO embeddings (message_id, vector) VALUES (%s, %s), (message_id, vec.tolist()) ) def retrieve(self, query, session_id, top_k6): query_vec self.embed.encode(query) # 向量召回 rows self.db.execute( SELECT m.id, m.content, m.tags, m.created_at, 1 - (e.vector %s) AS vec_score FROM messages m JOIN embeddings e ON m.id e.message_id WHERE m.session_id %s ORDER BY e.vector %s LIMIT 20 , (query_vec.tolist(), session_id, query_vec.tolist())).fetchall() # 混合打分 scored [] now datetime.now().timestamp() for row in rows: hours_ago (now - row[created_at].timestamp()) / 3600 time_score 1 / (1 hours_ago) tag_score 1.0 if preference in (row[tags] or []) else 0.0 final 0.7 * row[vec_score] 0.2 * time_score 0.1 * tag_score scored.append((final, row)) scored.sort(reverseTrue, keylambda x: x[0]) return [r for _, r in scored[:top_k]]这里先向量召回 20 条再用混合打分排序取前 6 条。为什么要两步因为纯向量排序只看语义相似度加上时间和标签后排序更符合业务直觉。是 pgvector 的余弦距离操作符1 - 距离就是相似度。去重逻辑我放在返回前做用简单的 Jaccard 系数比较内容。相似度超过 0.8 的只保留得分最高的那条。这个阈值可以调我试过 0.7 和 0.90.8 在大多数场景下比较平衡。5. 常见问题与排查技巧实录5.1 上下文相关问题的速查表下面这张表是我在实际项目里遇到过的典型问题按现象、可能原因、排查方法整理。遇到问题先查表能省不少时间。现象可能原因排查方法模型回答丢失早期信息窗口裁剪过度打印实际上下文检查旧消息是否被删响应突然变慢上下文 token 超限统计每次请求的 token 数看是否接近上限模型重复回答同一内容摘要消息和原始消息重复检查压缩后是否清理了原始消息工具调用报错裁剪拆散了调用和返回检查裁剪是否按回合分组检索召回无关内容向量模型不匹配换用业务语料微调过的 embedding 模型多用户上下文串了session_id 隔离失效检查查询条件是否带 session_id5.2 三个我踩过的坑和解决过程第一个坑是 token 统计不准导致裁剪失效。我一开始用字符数除以 4 来估算 token结果中文场景下严重低估实际 token 是估算的两倍多。上线后频繁触发模型长度限制。后来换成tiktoken精确统计问题解决。教训是token 统计不要估算用工具算误差控制在 5% 以内。第二个坑是摘要压缩后模型“失忆”。有次用户反馈聊了半小时后助手突然忘了用户之前说的预算约束。排查发现是摘要指令太笼统只说了“总结对话”模型把预算这种关键约束当成细节丢了。后来把指令改成明确列出要保留的信息类型并在摘要后加一句“以上约束仍然有效”问题才解决。这个坑让我意识到摘要不是越短越好关键信息必须显式保留。第三个坑是分层检索召回了跨会话的消息。有次测试发现A 用户的对话里出现了 B 用户提过的信息。排查发现是检索时忘了加 session_id 过滤向量库把全库消息都召回了。这个 bug 很危险涉及数据隔离。修复很简单加个 WHERE 条件就行但教训是任何检索操作都必须带隔离条件这是红线。5.3 性能优化的几个实用技巧上下文管理本身也会成为性能瓶颈尤其是消息量大、检索频繁的时候。我总结了几个优化点都是实测有效的。第一个是批量写入。消息不要一条一条插数据库攒够一批再插。我一般攒 10 条或者 1 秒取先到的条件。批量插入比单条插入快 5 到 10 倍。第二个是embedding 缓存。同样的内容不要重复算 embedding用内容哈希做 key 缓存起来。很多场景下用户会重复问类似的问题缓存命中率能到 30% 以上。第三个是异步压缩。摘要压缩涉及模型调用比较慢不要阻塞主请求。我的做法是压缩任务丢到后台队列主请求先用未压缩的上下文等压缩完成后再切换。这样用户感知不到延迟。第四个是上下文预热。对于活跃会话提前把上下文加载到内存缓存里请求来了直接用省去数据库查询。缓存过期时间设短一点比如 5 分钟避免数据不一致。5.4 监控与可观测性建设上下文模式上线后必须配套监控否则出了问题两眼一抹黑。我一般会埋这几个指标每次请求的上下文 token 数、裁剪触发次数、压缩触发次数、检索召回数量、检索命中率。token 数这个指标最重要它能提前预警容量问题。我设的告警阈值是平均 token 数超过上限的 70%或者单次请求超过 90%。裁剪和压缩的触发频率能反映窗口和阈值设置是否合理如果裁剪太频繁说明窗口设小了如果压缩太频繁说明阈值设低了。检索命中率是个业务指标定义是召回的消息里被模型实际引用的比例。这个指标低说明召回质量差需要调 embedding 模型或者打分权重。我一般会人工抽查一批请求看模型回答是否用到了召回内容用这个来校准。日志方面我建议把每次请求的完整上下文都记下来包括系统提示词、裁剪后的消息、召回的消息。存储成本不高但排查问题时价值巨大。注意脱敏用户隐私信息要处理掉。6. 上下文模式的扩展与组合玩法6.1 多模态上下文的处理思路现在很多场景不只是文本还有图片、音频。多模态上下文的管理思路和文本类似但有几个特殊点。图片的 token 消耗和分辨率相关不能简单按字符算。我的做法是给每种模态定义统一的“上下文单位”比如一张 512x512 的图片算 256 个 token一段 10 秒音频算 500 个 token。这样裁剪逻辑不用改只是 token 统计方式变了。存储上图片和音频存对象存储上下文里只存引用 ID。拼装上下文时再把引用换成实际内容。这样上下文列表本身很轻传输和存储都省事。6.2 上下文模式与 Agent 编排的结合做 Agent 编排时上下文模式会变得更复杂因为每个 Agent 可能有自己的上下文视图。我的做法是分层上下文全局上下文存所有 Agent 共享的信息比如用户画像、任务目标局部上下文存单个 Agent 的对话历史。Agent 执行时把全局上下文和局部上下文合并使用。合并时要注意去重和优先级。全局上下文里的信息优先级高局部上下文里如果和全局冲突以全局为准。这个规则要写清楚否则多个 Agent 协作时容易打架。6.3 上下文模式的测试策略上下文相关的 bug 很难通过单元测试发现因为涉及状态累积。我的做法是写场景测试模拟一段完整对话跑完之后断言上下文状态是否符合预期。比如模拟 50 轮对话断言裁剪后只剩最近 20 轮且摘要消息存在。场景测试的关键是可复现。我会把对话脚本写成 JSON 文件测试时按脚本回放。这样每次跑的结果一致方便定位问题。脚本里要覆盖边界情况刚好达到裁剪阈值、刚好触发压缩、工具调用跨裁剪边界等。另外我会做回归测试每次改上下文逻辑都跑一遍历史场景确保没破坏已有行为。这个习惯帮我避免了好几次线上事故。7. 我个人的一些实践体会上下文模式这个东西说起来是技术问题做久了会发现它更像产品问题。你得想清楚用户期望系统记住什么、忘记什么这直接决定了模式选型。我见过太多项目技术上用了最先进的检索方案但用户就是觉得“这助手怎么老忘事”根因是产品预期和技术实现没对齐。我的建议是项目早期先用最简单的滑动窗口跑起来把业务跑通同时把上下文相关的指标埋好。等有了真实数据再根据数据决定要不要上摘要、要不要上检索。不要一上来就搞复杂方案维护成本高而且没有数据支撑的优化都是瞎猜。还有一点上下文管理器的接口一定要稳定。业务代码只依赖add_message和get_context这两个方法内部实现随便换。这样你后期从滑动窗口升级到分层检索业务代码一行不用改。这个抽象带来的收益在项目迭代半年后你会深有体会。最后分享一个小技巧给上下文管理器加一个debug模式开启后每次get_context都打印实际拼装的内容和 token 数。开发阶段默认开启线上默认关闭。这个功能帮我省了无数排查时间强烈建议你也加上。
返回列表