1. 这个项目到底在解决什么问题
我接手“context-mode”这个项目的时候,第一反应是:它到底是个什么东西?一个模式?一个工具?还是一种设计思想?后来随着需求逐步拆解,我意识到这个标题背后真正对应的是一个非常具体的痛点——上下文混乱。
不管你是做对话机器人、AI Agent、文档处理系统,还是写复杂业务逻辑的工程师,只要你需要在多个场景间切换,就一定会被“上下文”问题折磨过。举个例子,你在做一个人工智能助理,它既要在闲聊模式下回答用户“今天天气怎么样”,又要在编程模式下帮用户调试代码。如果你把所有对话历史一股脑喂给模型,结果就是:用户问你天气的时候,模型还在惦记着刚才的代码变量名。这种体验,用过的人都懂,太割裂了。
“context-mode”这个项目,本质上就是一个上下文管理模式的设计与实践。它做的事情非常简单:把不同类型、不同用途、不同生命周期的上下文分门别类,按需加载、按需清理、按需切换。听起来简单,做起来却牵扯到非常多的细节,包括上下文的结构化表达、模式切换时的状态迁移、历史数据的保留策略、不同模式之间的隔离机制等等。
这篇文章我打算用我自己实际开发这个项目的过程作为主线,把里面的设计思路、核心细节、踩坑实录和相关经验完整分享出来。适合正在做对话系统、Agent框架、多场景业务整合、以及任何被上下文管理折磨得头疼的工程师参考。内容会尽量说人话,尽量落到代码和配置层面,而不是一堆云里雾里的概念。
2. 整体设计思路与方案选型
2.1 先想清楚:Context 到底是什么
动手写代码之前,我先做了一件事:把所有业务场景里涉及的“上下文”全部列出来。不列不知道,一列才发现“上下文”这个词被用滥了。在我的项目里,至少存在三类上下文:
第一类是会话上下文(Session Context),指的是用户与系统之间多轮对话产生的历史信息,包括用户身份、偏好、之前的问题、之前的回复等。这类上下文的特征是会持续累积,并且越旧的对话对当前决策的影响通常越小,所以需要做截断或摘要处理。
第二类是任务上下文(Task Context),指的是当前正在进行的具体业务流程中的数据,比如订单状态、用户选择的配置项、中间计算结果、需要传递到下一步的参数等。这类上下文的生命周期与任务绑定,任务结束就必须清理,否则会泄漏到下一个任务里。
第三类是场景上下文(Scene Context),指的是系统当前运行在哪个大环境下,比如是白天模式还是夜间模式、是调试模式还是生产模式、是中文环境还是英文环境、是面向C端用户的简化交互还是面向B端客户的专业交互。这类上下文往往不会频繁变化,但对其他所有上下文的解析方式都有影响。
“context-mode”的核心思想,就是把这三类上下文明确分开,并且用一套统一的机制去管理它们的生命周期和访问方式。而不是像很多半路出家的项目那样,用一个大字典把什么都塞进去,最后改一行代码要担心会不会影响另一个功能。
2.2 为什么选模式驱动而不是直白堆变量
我在做技术选型的时候,很自然地想过一个问题:直接用全局变量或者 Context Manager 存不就行了?为什么还要搞出“模式”这个概念?
关键在于规模和隔离需求。当你的系统只有两三个上下文变量时,直接建全局字典完全没有任何问题。但当上下文种类超过十个,来源包括用户输入、传感器数据、第三方回调、数据库查询结果、模型中间输出时,你再想用全局变量堆,代码就会变成一团乱麻,改一个字段可能牵动十几个函数。
模式驱动的思路是:给每一类上下文定义一个独立的“模式”,每种模式下可以访问的字段、可以执行的指令、可以调用的工具都是显式声明好的。不同的模式之间互不干扰,切换模式时才允许发生上下文写入和清理。这样做的直接好处有两个:第一,上下文的数据流变得可预测,代码阅读者看一眼模式定义就知道这个环节里系统能感知什么、不能感知什么;第二,安全性明显提升,因为某个模式下无法访问其他模式的内部数据,天然就避免了越权访问。
我最终选择的设计方案,是做一个三层结构:底层是一个基于上下文的存储引擎,负责数据的持久化和过期管理;中间层是模式注册表,负责声明有哪些模式以及每个模式可以访问哪些上下文键;顶层是会话处理器,负责根据用户输入和当前模式状态决定如何调度上下文。这个结构的好处是每一层都能独立测试,坏了也能快速定位。
3. 核心细节解析与实操要点
3.1 上下文的数据结构与存储选型
在设计数据结构的时候,我踩了第一个坑:一开始图省事,用一个 JSON 结构把所有上下文塞进内存字典,结果跑了没一会儿就发现内存吃紧,而且重启进程后所有对话状态全丢了。后来我换成了 SQLite 做持久化,兼顾了轻量级和可靠性,对于大多数中小型项目足够了。
实际的表结构,我设计了三张表。第一张是 session_table,主要存会话级上下文,包括 session_id、user_id、created_at、updated_at、context_json。第二张是 task_table,主要存任务级上下文,包括 task_id、session_id、task_type、task_status、payload_json。第三张是 scene_table,主要存场景级配置,包括 scene_id、scene_name、scene_config_json。三张表之间通过 ID 关联,但业务上刻意不让它们互相直接访问,只能通过统一的网关。
在设计 JSON 结构时,我给自己定了一个规矩:所有上下文的值对象必须有 schema,不能今天存字符串、明天存数组、后天存嵌套对象。所谓 schema,就是在写入前声明这个字段的类型、长度限制、默认值、是否可空。比如会话上下文里的 user_preference 字段,我规定它必须是一个 object,里面至少包含 language 和 timezone 两个字段。如果外部传入的数据不符合 schema,直接拒绝写入并记录错误日志。
这个规矩看起来有点死板,但实际帮了大忙。有一次第三方服务回调返回的数据里把 age 字段传成了字符串,如果是无 schema 的设计,系统就会继续带病运行,用户的个性化推荐全乱套。正是因为有了 schema 校验,在我们接入早期就发现了问题,排查效率高了很多。
3.2 模式的声明与切换逻辑
模式的定义我用的是字典配置的方式,而不是写死在代码分支里。每定义一个模式,需要声明四样东西:
- mode_id:模式唯一标识,比如
chat、code_review、data_analysis - allowed_keys:该模式下允许访问的上下文键列表
- input_schema:该模式接受的外部输入结构
- output_schema:该模式输出给用户或下游系统的结构
这样做的好处是新增一种业务场景时,我只需要在配置中心加一段配置,然后写对应的处理函数,核心框架完全不用动。比如后来客户要求增加一个“营销文案生成模式”,我从定义配置到联调完成只花了半天,这在原来那种 if else 满天飞的写法下是不可想象的。
模式切换是整个系统里最容易出错的地方。我设计了一个状态机,把模式切换拆成五个阶段:请求切换、检查当前任务状态、执行旧模式清理、执行新模式初始化、确认切换完成。为什么要拆这么细?因为切换过程不是原子的,如果中途出现异常而没有处理,系统就会卡在一个“既不是旧模式也不是新模式”的中间态,所有请求都会异常。
具体的清理策略是:会话级上下文完全保留,场景级上下文按需热更新,任务级上下文一律清理并归档。这样确保用户在一个会话里从闲聊切换到编程辅助时,之前的闲聊内容不会干扰代码生成,但用户的身份信息和语言偏好仍然保留。
3.3 上下文过期与遗忘机制
很多做上下文管理的项目在“记忆”这件事上用力过猛,什么都要永久记住,结果就是上下文越来越臃肿,模型推理时间越来越长,最终用户体验反而下降。我在这套系统里加入了显式的过期机制,参考了人类遗忘曲线的思路。
每条上下文记录都带一个 ttl 字段,即过期时间。会话级上下文的 TTL 通常是 1 小时,如果 1 小时内没有新消息,就会自动触发摘要与归档流程。任务级上下文的 TTL 是 5 分钟,任务完成后立即归档。场景级配置的 TTL 是 12 小时,并且可以在系统启动时强制刷新。
这里我特别想强调一个容易被忽视的点:过期不意味着直接删除。过期的上下文会先进入一个归档区,里面保留完整的历史快照。只有在超过 7 天后,后台定时任务才真正物理删除。这样做的好处是:如果用户过两天回来继续之前的话题,我还能通过一键恢复把归档上下文重新加载回来;如果直接物理删了就彻底没了,用户会非常不满。
还有一个非常实用的小技巧:利用“上下文摘要”。当 TTL 到期时,系统会用当前已有的上下文生成一份不超过 200 字的摘要,写入会话表的 summary 字段。这样哪怕是几个月前的会话,用户回来后我们没法恢复完整对话,但至少能知道“这个用户当时正在准备考雅思、偏好简洁的回复风格、已经完成了词汇量测评”。这一招在留存和个性化推荐上效果显著。
4. 实操过程与核心环节实现
4.1 从零搭建最小可运行环境
我实际搭建项目时的环境大概是这样的:Python 3.11、FastAPI 作为 API 框架、SQLite 做存储、Redis 做缓存和状态同步,前端先不考虑,先把后端全链路打通。项目目录结构我按功能拆得比较清晰:
context-mode/ ├── app/ │ ├── context/ │ │ ├── store.py # 上下文存储引擎 │ │ ├── schema.py # schema 校验 │ │ ├── expiry.py # 过期与归档逻辑 │ │ └── registry.py # 模式注册表 │ ├── modes/ │ │ ├── chat.py │ │ ├── coding.py │ │ └── analysis.py │ └── api/ │ ├── session.py │ └── mode_switch.py ├── config/ │ └── modes.yaml ├── tests/ └── requirements.txt这里我提醒一句:目录结构千万别学网上那些“按层级架构”的复杂分法,项目刚开始时越简单越直接越好。我当时就是把所有和上下文相关的代码都放在同一个目录下,后续再根据需要拆分,避免一开始过度设计。
安装依赖也很简单,主要是 fastapi、uvicorn、redis、pydantic、sqlalchemy 这几个库。如果你用的是 Python 虚拟环境,一条命令就能搞定。我建议开发阶段直接把 FastAPI 和 Uvicorn 装在同一个环境里,因为 FastAPI 的热重载功能配合 Uvicorn 简直不要太爽,改完代码自动重启,调试效率直线上升。
4.2 核心类与关键接口代码拆解
我把最核心的上下文存储引擎代码精简后贴出来,方便你理解整体思路。这个类负责所有上下文数据的读写、过期判断和归档触发:
class ContextStore: def __init__(self, db_path: str, redis_client=None): self.db_path = db_path self.redis = redis_client self._init_db() def _init_db(self): conn = sqlite3.connect(self.db_path) conn.execute(""" CREATE TABLE IF NOT EXISTS session_table ( session_id TEXT PRIMARY KEY, user_id TEXT, context_json TEXT, summary TEXT, created_at TIMESTAMP, updated_at TIMESTAMP ) """) conn.commit() conn.close() def set_context(self, session_id: str, key: str, value: Any, ttl: int): # 先校验 schema if not self._validate_schema(key, value): raise ValueError(f"Schema validation failed for key {key}") # 从数据库读取旧数据 old_data = self.get_context(session_id) old_data[key] = value old_data["_ttl"] = ttl old_data["_updated_at"] = int(time.time()) # 写回数据库 conn = sqlite3.connect(self.db_path) conn.execute( "INSERT OR REPLACE INTO session_table (session_id, user_id, context_json, updated_at) VALUES (?, ?, ?, ?)", (session_id, "default", json.dumps(old_data), time.time()) ) conn.commit() conn.close() # 同步写 Redis 缓存 if self.redis: self.redis.set(f"ctx:{session_id}", json.dumps(old_data), ex=ttl) def get_context(self, session_id: str) -> dict: # 优先从 Redis 读取 if self.redis: cached = self.redis.get(f"ctx:{session_id}") if cached: return json.loads(cached) conn = sqlite3.connect(self.db_path) row = conn.execute( "SELECT context_json FROM session_table WHERE session_id = ?", (session_id,) ).fetchone() conn.close() if not row: return {} data = json.loads(row[0]) # 检查是否过期 ttl = data.get("_ttl", 3600) updated_at = data.get("_updated_at", 0) if int(time.time()) - updated_at > ttl: # 触发归档逻辑 self._archive_session(session_id, data) return {} return data def _archive_session(self, session_id: str, data: dict): # 生成摘要并归档 summary = self._generate_summary(data) conn = sqlite3.connect(self.db_path) conn.execute( "UPDATE session_table SET summary = ? WHERE session_id = ?", (summary, session_id) ) conn.commit() conn.close()这段代码最核心的点在于:写入前强制走 schema 校验,读取时先查缓存再查持久化,并且每次读取都会检查 TTL。如果你的项目并发量不高,完全可以先不加 Redis,直接用 SQLite 也能撑住中小场景。
模式注册表和切换器其实也不复杂,我贴一段核心逻辑:
class ModeSwitcher: def __init__(self, registry: dict): self.registry = registry self.current_mode = None self.mode_stack = [] def switch(self, new_mode: str, session_id: str): if new_mode not in self.registry: raise ValueError(f"Unknown mode: {new_mode}") if self.current_mode and len(self.mode_stack) > 0: # 检查当前任务是否完成 if self._has_unfinished_task(session_id): raise RuntimeError("Cannot switch mode with pending task") # 归档当前模式的任务上下文 self._archive_current_task(session_id) # 初始化新模式 self.current_mode = new_mode self.mode_stack.append(new_mode) self._load_mode_config(new_mode) return {"status": "ok", "current_mode": new_mode} def _load_mode_config(self, mode: str): # 从 config 中加载该模式允许的键和输入输出格式 mode_config = self.registry[mode] self.allowed_keys = mode_config.get("allowed_keys", []) self.current_mode = mode4.3 配置定义:如何用 YAML 管理多模式
我的模式配置全部放在config/modes.yaml里,用代码加载后注册到运行时。示例配置如下:
modes: - id: chat description: 日常闲聊模式 allowed_keys: - user_profile - chat_history - language_preference input_schema: type: object properties: message: type: string maxLength: 2000 output_schema: type: object properties: reply: type: string - id: coding description: 代码辅助模式 allowed_keys: - user_profile - repo_info - code_snippet - lint_result input_schema: type: object properties: code: type: string language: type: string output_schema: type: object properties: suggestion: type: string line_changes: type: array实际生产环境里,我还会在 YAML 配置里额外加一些字段,比如dangerous_actions,用来声明该模式下禁止执行的敏感操作;再比如fallback_mode,当新模式因为异常无法初始化时兜底切回哪个模式。这些配置看起来直白,但它们正是上下文管理系统的“边界感”所在,控制住边界才能谈得上安全和稳定。
4.4 模式切换的调用链路与状态迁移
实际运行中的完整调用链路是这样的:用户从前端页面发来一条消息,API 层先解析消息内容,通过语义分类或显式指令判断当前应该处于哪个模式。如果是新模式,先调用 ModeSwitcher 进行切换,再走正常的上下文读取和模型推理链路。如果仍是当前模式,则直接追加上下文并发起响应。
我做了一个很直观的状态迁移图来描述这个过程:从IDLE状态开始,收到消息后进入PENDING状态,解析后进入ACTIVE状态。在 ACTIVE 状态下,如果检测到模式切换触发条件,则进入SWITCHING状态,完成切换后回到ACTIVE。如果整个过程中出现任何异常,进入ERROR状态,并在 3 秒内自动回滚到切换前的模式。
状态迁移这一块我不建议你偷懒。刚开始时我想省事,直接用一个布尔变量记录当前模式,切换就改布尔值。结果有一次并发请求同时触发了两个切换指令,两个进程同时改布尔值,状态直接乱掉。后来改成状态机并加了锁,这才彻底解决。顺便吐槽一句:如果你做的是高并发系统,请务必给 mode_switch 接口加分布式锁,否则一定会死在并发上。
5. 常见问题与排查技巧实录
5.1 上下文串味:为什么用户A看到用户B的偏好
这个问题是我在联调阶段被测试小姐姐吐槽最多的。排查后发现,根因是 session_id 没有被正确传递。我一开始把 session_id 放在请求头里,但前端代码不知道是什么时候漏写了这个头,导致后端收到空 session_id,所有请求都落到同一个默认会话里,自然就串味了。
这类问题其实很隐蔽。我给读者的建议是:建立一套强制校验机制,在 API 网关层就拦截那些缺失 session_id 的请求,而不是到了业务逻辑里再判断。同时可以在数据库层给 session_id 加上唯一约束,这样就算代码写得再稀烂,也不会出现多用户共用一个桶的情况。如果你用的是 FastAPI,可以在依赖项里做校验,比如:
async def get_session_id(request: Request): session_id = request.headers.get("X-Session-Id") if not session_id: raise HTTPException(status_code=400, detail="Missing session id") return session_id这个小小的拦截器,能省掉你后面几十个小时的定位时间。我后来在项目里也养成了习惯:一切外部传入的标识符,都必须先经过合法性校验,再进入核心逻辑。
5.2 模式切换后旧数据残留
这个问题典型的症状是:用户从“聊天模式”切到“代码模式”之后,聊天气质还留着,系统回答问题的口吻偏口语化,甚至有时候会把聊天里的梗带到代码解释里。
本质上是因为在切换模式时,旧模式的任务上下文没有清理干净。我在最初版本里只清理了一部分键,比如清除了chat_history,但漏掉了pending_question之类的临时字段。这些漏网的字段被新模式读到了,自然就影响了行为。
解决思路也很直接:在每个模式声明里增加一个cleanup_keys字段,显式列出切换时必清的键。然后在 ModeSwitcher 的_archive_current_task里,把该模式配置的所有 cleanup_keys 一一删除。如果你希望更安全,可以在清理后做一次完整性校验,检查 data 字典里是否还有任何键名匹配旧模式的前缀。
这个排查过程让我养成了一个习惯:把模式的“输入面”和“输出面”都看作接口契约,切换即契约变更,契约变更就必须同步清理历史数据,而不是靠系统自动“猜”你应该清哪些。
5.3 上下文越长,响应越慢
做过大模型应用的朋友肯定有共鸣:只要上下文塞得多,模型响应就一定慢,而且成本直线上升。这个问题在 context-mode 项目里也真实发生过。有一次我把用户过去一整年的聊天记录全塞进上下文,结果接口响应时间从 0.8 秒直接飙升到 8 秒,用户直接投诉。
后来我做了两层优化。第一层是在写入时做限制:会话上下文最多保留最近 20 轮对话,超过部分触发摘要在议摘要机制,用一段概括性的文字替代旧对话,既节省 token 又保留核心记忆。第二层是在读取时做检索:使用简单的关键词命中或向量相似度计算,从历史上下文中挑出跟当前输入最相关的 5 到 8 条记录动态拼接,而不是全量读出。
这两层优化配合起来以后,响应时间降到了 1.2 秒左右。实测下来,效果比无脑塞全量历史要好很多,用户也没有觉得“失忆”。这里的经验是:上下文管理的目标不是尽量多记住,而是尽可能用最小的信息量完成当前决策。
5.4 并发场景下的上下文覆盖
最后一个高频问题出现在并发场景:同一用户同时发起两个请求,两个请求同时读旧上下文,各自加工后又同时写回,后写的数据覆盖了先写的数据,用户的某一段输入彻底丢失。这在对话系统中特别致命。
解决的思路有两种。第一种是乐观锁,在上下文数据里加一个版本号字段,写入前比对当前版本和读取时的版本,不一致就拒绝写入并触发重试。第二种是合并写入,对同一个上下文对象的不同字段做 merge,而不是整体覆盖。我实际项目里用的是版本号方案,简单直接,也方便后续审计。
示例代码如下:
def update_context_with_version(session_id, key, value, expected_version): current = get_context(session_id) if current.get("_version") != expected_version: raise ConflictError("Version mismatch, please retry") current[key] = value current["_version"] = expected_version + 1 save_context(session_id, current)5.5 常见问题速查表
| 问题 | 典型原因 | 快速定位方式 | 推荐解决动作 |
|---|---|---|---|
| 上下文串味 | session_id 缺失或未传递 | 查 API 网关日志里 x-session-id 是否为空 | 增加网关层强制校验,数据库加唯一约束 |
| 切换后旧数据残留 | 清理键列表不完整 | 打印切换前后上下文 key 的 diff | 为每个模式显式声明 cleanup_keys |
| 响应变慢 | 上下文无限制累积 | 统计单会话上下文平均 token 数 | 加入摘要机制和相关性检索 |
| 并发覆盖 | 缺少版本控制 | 复制用户操作步骤复现丢消息场景 | 引入乐观锁或按字段合并写入 |
| 上下文写入失败 | schema 校验不过 | 查看 schema 校验错误日志 | 拒绝写入并在日志中保留原始输入 |
6. 这块还能怎么用
context-mode 这套设计并不局限于对话机器人系统。实际上,凡是存在多场景切换和多来源信息的系统,都可以借鉴它的思路。比如自动化测试平台中,每个测试套件就是一类模式,套件之间必须隔离配置和状态;又比如电商领域的促销引擎,不同活动之间不能共享用户券包状态,否则并发会让用户领错券;再比如推荐系统里,用户处于“浏览模式”还是“结算模式”就决定了该推什么内容、该显示什么验证码。
我在实际使用中体会到,这套模式化上下文管理的最大价值不在于技术实现有多难,而在于它强制你建立边界意识,让你在写代码前就思考清楚数据的归属、流向和生命周期。边界清晰了,系统的复杂度才真正掌握在人的手里,而不是被需求推着走,在混乱中疲于奔命。
最后一招防线:无论你怎么设计,也别忘了给自己留一条后路,每次模式切换前都打个完整快照。等真出了问题时,你会发现这条后路救了你无数次命。