最近在重构一个AI客服项目时我又被“上下文管理”这件事反复按在地上摩擦。用户先问商品价格又问售后政策再切换到退款流程模型常常把三个场景的细节混在一起回答起来像喝多了。试过把系统提示词写得很长试过给会话加标签但代码越改越乱。后来我索性把整个会话管理层拆掉重写了一个叫 context-mode 的模块。这个模块把“上下文”从单一的消息列表变成了多个可切换、可持久化、各自独立的“模式”每个模式都有自己的系统指令、历史消息和组装策略。这篇文章是我从设计思路到落地实现的全过程记录包含完整代码和排错经验。如果你也在被多轮对话的状态混乱、token超限、场景串线折磨应该能从里面找到一些参考。1. 为什么AI对话总在串线context-mode的出发点1.1 模型上下文不是“聊天记录”那么简单传统方案里我们习惯以 session 维度维护一段对话历史把某个用户在一次会话里的所有消息按时间顺序存下来每次请求全量发给模型。这在单话题、短对话里没问题。但实际业务往往不是单话题的尤其是客服、保险、医疗咨询这类场景一段会话里会有多个子任务每个子任务需要不同的“背景信息”。如果用同一个消息列表很容易出现“上一单的售后政策被当成下一单的购买条款”这种串线事故。这里给个生活类比这就像一个人戴着同一副老花镜既想看清远方的路牌又想看清桌面的说明书视野里所有东西叠在一起自然容易看错。context-mode 的解决思路是准备多副眼镜看路牌时换一副看说明书时再换一副。在代码里就是为不同子任务建立独立的上下文容器。1.2 Session为什么不够用我们看看常见的 session 实现大致是这样的伪代码sessions[session_id] { messages: [] }每次对话就往 messages 里 append然后整体送给大模型。它能解决“不同用户之间的数据隔离”但解决不了“同一用户、同一会话内不同意图的数据隔离”。当用户在同一会话内切到另一个业务时session 不知道切换历史消息依然混在一起。尤其当用户回头说“我刚刚那个订单怎么了”时模型根本分不清是哪个订单因为上下文里夹杂着太多别的信息。还有一个更隐蔽的成本问题同一个用户聊得越久messages 越长每次请求的 token 消耗越高。很多项目在长对话后段模型开始胡言乱语不是因为模型变笨了而是因为早期信息一直没有被清理把真正重要的近期信息挤出了窗口。1.3 context-mode的建模方式context-mode 不把对话当成一条连续的线而是当成一组可以互相切换的卡片。每一张卡片就是一个 context mode包含四个要素mode_id、system_prompt、history、meta。切换业务时不重建 session而是切换 mode。比如用户从“商品咨询”切到“售后处理”我们保留前一个 mode 的全部状态新建或切换到另一个 mode。需要回到前一个时再切回来即可。单个 mode 内部的 token 长度可以单独控制互不影响。设计目标可以列成四条场景独立每个 mode 的状态互不干扰。长度可控每个 mode 有独立的上下文窗口限制。可持久化mode 可以序列化保存重启后恢复。可组合最终请求时允许按需合并部分 mode 的内容。2. 方案选型与核心数据结构2.1 为什么选用轻量Python模块实现选型时考虑过数据库、中间件、框架级解决方案。最后决定用 Python 写一个独立模块因为业务侧本来就是个 Python 服务模块化实现最直观而且不需要引入外部消息队列。Python 的 dataclass 天然适合定义数据结构配合 JSON/Pickle 就能完成持久化。但这里要提醒Python 的 GIL 和内存模型并不是高并发场景的最佳选择。如果你的服务单机 QPS 很高建议把 context 管理逻辑下沉到 Redis 或内存型 KV 存储中。我选择先实现内存版再在持久化层做适配这样逻辑可以复用。如果以后要换 Go 或 Node 实现核心数据结构保持一致迁移成本也不会太高。2.2 数据结构的核心定义一个 context mode 在代码里长这样from dataclasses import dataclass, field from typing import Any, Dict, List, Optional dataclass class Message: role: str # system | user | assistant content: str timestamp: float 0.0 dataclass class ContextMode: mode_id: str system_prompt: str history: List[Message] field(default_factorylist) metadata: Dict[str, Any] field(default_factorydict) max_tokens: int 4000 # 该模式的上下文上限 reserved_tokens: int 800 # 为模型回复预留的token compression: str truncate # truncate | summary其中metadata很关键。它可以放业务侧的各种参数例如用户ID、订单号、当前页面URL、偏好语言等。这些内容不一定进入模型但需要跟着 mode 走。比如用户切到“退款模式”时metadata 里要带上订单号组装消息时自动拼一句“当前处理订单XXX”。max_tokens和reserved_tokens是一对必须同时维护的参数。很多人在做上下文截断时只考虑输入上限忘了给模型输出留空间结果接口频繁报错。我见过太多生产事故本质上都是把窗口用满了模型一个字都吐不出来。2.3 组装策略truncate还是summary每个 mode 在发送给模型前要从原始历史生成一个“可见上下文”。常见策略有三种direct直接拼接所有消息适合短历史。truncate保留系统提示和最近 N 轮对话丢弃更早的内容适合长历史、但早期信息不重要的场景。summary把超出窗口的历史先做摘要再把摘要作为一条 system 消息放进上下文中适合关键信息分散在全程的场景。这三种策略在代码里可以通过一个函数模板统一暴露def build_context(mode: ContextMode) - List[Message]: # 根据mode.compression选择策略 if mode.compression summary: return _build_with_summary(mode) return _build_with_truncate(mode)_build_with_summary在真实产品里可以调用大模型本身来压缩历史也可以先用统计方法提取关键词。考虑到成本我通常只在整段历史的估算 token 数超过max_tokens的 65% 时才触发压缩。低于这个阈值时直接 truncate 就够了压缩反而损失信息。3. 从零实现context-mode详细代码3.1 上下文管理器先写一个最核心的内存版ContextManager。它负责创建、切换、读写 modeimport copy import time class ContextManager: def __init__(self): self.modes: Dict[str, ContextMode] {} self.current_mode_id: Optional[str] None def create_mode(self, mode_id: str, system_prompt: str, max_tokens: int 4000, compression: str truncate) - ContextMode: mode ContextMode( mode_idmode_id, system_promptsystem_prompt, max_tokensmax_tokens, compressioncompression, ) self.modes[mode_id] mode return mode def switch_mode(self, mode_id: str) - None: if mode_id not in self.modes: raise KeyError(fmode {mode_id} not found) self.current_mode_id mode_id def current_mode(self) - ContextMode: if self.current_mode_id is None: raise RuntimeError(no active mode) return self.modes[self.current_mode_id] def add_message(self, role: str, content: str, mode_id: Optional[str] None) - None: m self.modes[mode_id] if mode_id else self.current_mode() m.history.append(Message(rolerole, contentcontent, timestamptime.time())) def clear_history(self, mode_id: Optional[str] None) - None: m self.modes[mode_id] if mode_id else self.current_mode() m.history.clear()注意current_mode()返回的是引用外部可以直接修改 mode.history。生产环境下建议返回深拷贝避免不必要的副作用。但深拷贝每次都拷贝整个列表成本不低所以我在实际项目中用一个可选的readonly开关默认允许引用操作但通过代码规范约束。3.2 上下文组装与token控制build_request_messages的实现是这样的def estimate_tokens(text: str) - int: # 用字符数做粗略估算 return int(len(text) * 0.75) def truncate_history(mode: ContextMode) - List[Message]: # 保留最近若干轮按轮次裁剪 # 先保留System Prompt再尽量保留尾部消息 max_content_tokens mode.max_tokens - mode.reserved_tokens - estimate_tokens(mode.system_prompt) messages [] used 0 for msg in reversed(mode.history): cost estimate_tokens(msg.content) if used cost max_content_tokens: break messages.append(msg) used cost messages.reverse() return [Message(rolesystem, contentmode.system_prompt)] messages def build_request_messages(mode: ContextMode) - List[Message]: if mode.compression summary: return _build_with_summary(mode) return truncate_history(mode)这里reserved_tokens是给模型留出的回复空间不然把整个窗口都塞满输入模型没有输出预算接口会直接报错。按轮次裁剪是性价比比较高的做法因为一段对话里的相关性通常和位置强相关最近的几轮最相关。3.3 持久化层让mode不丢内存版重启就没了。我加了一个基于 JSON 文件的存储层序列化时把Message.timestamp转成数字恢复时再还原。如下import json import os class JsonStorage: def __init__(self, path: str context_modes.json): self.path path self._load() def _load(self) - None: if not os.path.exists(self.path): self.data: Dict[str, dict] {} return with open(self.path, r, encodingutf-8) as f: self.data json.load(f) def save(self, manager: ContextManager) - None: payload {} for mid, mode in manager.modes.items(): payload[mid] { mode_id: mode.mode_id, system_prompt: mode.system_prompt, metadata: mode.metadata, max_tokens: mode.max_tokens, reserved_tokens: mode.reserved_tokens, compression: mode.compression, history: [ {role: m.role, content: m.content, timestamp: m.timestamp} for m in mode.history ], } tmp self.path .tmp with open(tmp, w, encodingutf-8) as f: json.dump(payload, f, ensure_asciiFalse, indent2) os.replace(tmp, self.path)先写临时文件再 os.replace是为了避免进程中途崩溃导致 json 文件损坏。我用这种方式之后再没遇到过写一半文件变空的问题。3.4 接入大模型接口最终调用侧的逻辑很薄def chat_in_mode(manager: ContextManager, user_input: str): mode manager.current_mode() manager.add_message(user, user_input) messages build_request_messages(mode) # 调用大模型这里用伪代码 response call_llm(messages) manager.add_message(assistant, response) return responsecall_llm的内部就是常规的 HTTP 请求把 messages 列表序列化之后发出去再取回结果。有个细节有些平台的接口要求 system 消息必须在第一条这个在build_request_messages里已经保证了。如果将来要接入支持多 system 的接口也只需要调整这一步。4. 实战中踩过的坑与排查方法4.1 模式串场切换后还带着上一个模式的历史我最初犯过“类变量泄漏”的错误。某个版本里把 ContextManager 的 modes 定义成了类属性而不是实例属性。于是所有用户共享同一个模式字典A 用户的售后查询记录跑到 B 用户的商品咨询里了。排查方法很简单在切换 mode 后打 log输出当前 mode_id 和历史条数结果发现两个用户共用同一个 mode_id。修复方式是把self.modes {}放回__init__。这个坑很隐蔽因为本地单用户测试完全看不出来一上线多用户就乱套。建议所有上下文容器都做实例属性并且不要默认用模块级全局单例。4.2 token估算偏差导致的输入超限字符数乘以 0.75 的估算方法在英文场景下比较准中文场景会偏低。后来我在 estimate_tokens 里加了中文字符加权def estimate_tokens(text: str) - int: import re zh_chars re.findall(r[\u4e00-\u9fff], text) other_chars len(text) - len(zh_chars) return int(len(zh_chars) * 1.2 other_chars * 0.4)这个近似值足够用于截断判断比调用第三方 tokenizer 快得多。如果你对成本敏感可以在离线阶段用一个真实模型商用的 tokenizer 对典型语料做统计拟合出更准的系数。简单场景下中英文混合加权已经能避免绝大多数超限问题。4.3 并发修改同一个mode多线程下两个请求同时在 add_message 时会出现历史顺序错乱。我在写操作上加了一个threading.Lockimport threading class ContextManager: def __init__(self): self._lock threading.Lock() self.modes {} def add_message(self, role: str, content: str, mode_id: Optional[str] None) - None: with self._lock: m self.modes[mode_id] if mode_id else self.current_mode() m.history.append(Message(rolerole, contentcontent, timestamptime.time()))如果是多进程部署把锁换成 Redis 里的分布式锁或者直接把 mode 存到 Redis 里用原子操作。这里要特别注意不是每个请求都加锁就能解决一切问题组装上下文和写上下文应该也放在同一个锁内否则可能出现“读到半截写了一半”的历史。4.4 摘要压缩后上下文信息不够全summary 策略我踩过最大的一次坑把压缩结果当成普通 assistant 消息放回 history结果模型把摘要当成“自己说过的话”回答时经常引用不存在的内容。后来我把摘要放到一条独立的消息里role 固定为 “system”并且 content 前缀加了“以下是此前的对话摘要”模型的行为才正常。还有一点摘要本身要及时更新。早期版本只在每次消费完历史后生成一次摘要如果用户继续发了 20 轮新消息旧摘要已经和新消息对不上。后来我改成“摘要生成后若新增历史超过 token 上限的 50%则重新生成”。4.5 modal数据没有及时保存刚开始只在服务退出时 save结果服务被 kill 后数据全丢。改成每次 add_message 后异步延时保存例如每 10 秒落一次盘或者每隔 N 条消息 save 一次。对于要求不高的小项目直接每次 add_message 后 save 也行只是 IO 量大一点。可以按下面的速查表快速定位问题症状可能原因处理方案切换mode后还有上一个mode的内容类变量泄漏或同一个mode_id被复用检查modes是否为实例属性切换时打印mode_id接口报输入超限token估算不准、reserved_tokens太小更新估算函数调大reserved_tokens并发高峰历史错乱缺少锁或锁粒度过大在写和组装逻辑上加统一锁模型引用不存在的内容摘要被当成assistant消息把摘要role改为system并加明显前缀重启后上下文丢失未做持久化或保存时机过晚增加定时/计数保存机制5. 扩展思路与更多玩法5.1 做成角色卡片把 mode 定义成一组角色卡片客服、导购、技术支持。每个角色有自己的性格、话术模板和知识库引用方式。切换角色就是一次 context-mode 切换非常适配营销聊天、角色扮演类应用。实际使用时我会给每个角色卡片的 system_prompt 里带上角色背景和回复风格再把知识库的检索结果塞进 metadata。这样就不需要每次对话都在用户消息里重复“你现在是一个XXX”。用户感知到的就是每次进入不同业务助手自动换成对应的口吻和身份。5.2 和业务工单系统联动比如用户发一句“我要退款”业务层先创建一个退款 mode把订单号写入 metadata。后续几轮消息都在这个 mode 里处理即使中间用户又随口问了一句物流信息也不会污染退款上下文。这里有个实战细节不是所有消息都要进 refund mode。我设计了一个轻量的意图路由先用规则或小模型判断当前用户消息是否属于当前 mode 的主题如果完全无关可以挂到另一个临时 mode或者只放一条 user 消息到历史不给它改变当前 mode。这样可以避免“用户随口一句话整个上下文被打乱”的情况。5.3 做A/B测试可以给同一场景建多个 mode 副本只是 system_prompt 不同然后随机分配流量对比不同 prompt 下的满意率。由于 mode 之间完全隔离这样得到的对比数据比单纯改 prompt 更干净。常见做法是把 mode_id 作为分流维度比如prompt_a和prompt_b。线上日志里记录 mode_id后续分析时直接按 mode_id 分组。因为每个 mode 的历史和上下文都是独立维护的A/B 测试不会互相污染数据可信度很高。5.4 多模型混合调度每个 mode 还可以挂一个 model 字段比如普通闲聊用轻量模型复杂推理用重量模型。这样 context-mode 不只是上下文容器还能成为多模型路由的入口。我在实测中把成本降低了近 40%。实现上是在 ContextMode 里增加model_name: Optional[str]组装完 messages 后根据 mode 的 model_name 选择不同接口。对用户来说他们感知不到模型切换但因为每个 mode 的上下文是干净的模型切换后也不会出现“换了模型就失忆”的现象。6. 写在最后一点个人经验6.1 什么场景不要用context-mode如果只是做一个单话题的翻译助手、单轮问答工具或者测试玩具context-mode 确实有点重Session 方案足够。复杂度和收益要匹配。我的判断标准是一个会话里是否会出现“当前主题”的切换且切换后必须回到之前的主题。如果有context-mode 就是划算的。另外如果你的业务并发极高又不愿意引入 Redis单纯用内存锁方案会把性能压得很低。这种时候先把 context 存储外置到独立的缓存服务会更合理。模块本身的算法和数据结构不需要改只需要把存储层换掉。6.2 小型项目的极简接入方式即使不做完整框架你也可以把 context-mode 的核心思想简化应用到现有代码里只需要在原来 session 的 messages 前加一个current_scene字段每次组装消息时只截取current_scene对应时间段之后的消息。这个改动非常小但能立刻解决大多数“场景串线”问题。我在实际使用中发现context-mode 最有价值的地方不是那段代码而是逼着我去思考“每一轮对话到底属于哪个场景”。一旦你有了这个意识session 方案也能写出不错的上下文隔离。最后再分享一个小技巧给每个模式设置独立的开场白system prompt比全局调参管用得多。很多模型表现不稳定往往不是模型本身的问题而是它站在一个混乱的上下文里。把上下文理清楚模型自然就稳定了。