第 28 篇
工程技术
新索引构建失败，怎样保住旧索引的查询能力
结合实际开发项目，讨论新索引构建失败，怎样保住旧索引的查询能力，说明设计理由、关键源码、验证方法及适用边界。
本文目录 · 8 节
本文属于 Memory Center：AI 编程上下文与记忆系统工程实践，第 28 篇，位于“可靠后台任务”模块。案例背景统一采用实际开发项目；复现步骤使用脱敏教学样例，不代表特定生产事故。
从实际问题出发
如果直接在当前索引上边删边建，一次解析失败就可能让整个项目无法查询。代际构建把准备新结果和切换可见结果分开，失败只影响新一轮工作。
把发布当作提交边界
新代际准备文件和分块，成功后在受控锁与数据库事务内验证任务状态并更新生效记录。任务成功、配置仍匹配、数据完整，都是切换前需要成立的条件。
图中展示的是本篇关注的职责和顺序。正常流程之外，还要结合下面的源码与实验检查失败分支；它不是对某次线上调用的运行时追踪。
原子切换有自己的成本
构建期间需要同时保留旧数据和新数据，增加空间占用。读者也必须一致地使用生效代际，不能分别从不同来源拿到新旧两套结果。代际隔离主要解决发布可见性，备份与灾难恢复仍需单独设计。
为发布边界设计可证伪检查
可以故意让新代际构建在最终提交前失败，然后检查三件事：生效索引标识是否未变；旧数据是否仍能查询；失败原因是否可定位。如果系统只在成功路径表现正确，这个实验会直接暴露缺口。
新代际目录、数据库中的生效记录与任务成功状态也要一起考虑。只做文件夹重命名，却让数据库提前指向新数据，仍会产生不一致。应追踪真正决定读者可见性的提交点。
关键源码与阅读方法
源码基线为 Memory Center 6e90553（文档标注 5.0.0b24）。定位：memory_center/platform/store.py，函数 publish，起始行 110。
以下为源码节选，省略后续分支；不能作为独立脚本执行。
def publish(self, workspace_id, job_id, chunks, workspace_file, config_version):
with self.generation_lock(workspace_id), self.db.connection(True) as connection:
row = connection.execute("SELECT payload FROM records WHERE kind='job' AND id=?", (job_id,)).fetchone()
job = json.loads(row[0])
if job['status'] != 'running':
raise CatalogError('JOB_CANCELLED', '任务已取消，未发布索引')
if 'embedding_settings_revision' in job:
from memory_center.platform.project_embedding import embedding_settings
if embedding_settings(self, workspace_id)['revision'] != job['embedding_settings_revision']:
raise CatalogError('EMBEDDING_SETTINGS_CHANGED', '项目向量设置已改变，未发布索引，请重新提交任务')
from memory_center.platform.retrieval import index_terms
index_terms(connection, workspace_id, job_id, chunks)
for chunk in chunks:
connection.execute('INSERT INTO chunks VALUES(?,?,?,?,?,?,?,?)' if not self.db.url else
'''INSERT INTO chunks(id,workspace_id,generation,source_id,content,metadata,profile,embedding)
VALUES(?,?,?,?,?,?,?,?)''',
(chunk['id'], workspace_id, job_id, chunk['source_id'], chunk['content'],
json.dumps(chunk['metadata'], ensure_ascii=False), chunk.get('profile', ''),
json.dumps(chunk.get('embedding', []))))
if self.db.url and chunk.get('embedding'):
connection.execute('UPDATE chunks SET vector_value=?::vector WHERE id=?',
(str(chunk['embedding']), chunk['id']))
if self.db.url:
from collections import Counter
profiles = Counter((c.get('profile'), len(c.get('embedding', []))) for c in chunks)
for (profile, dimension), _count in profiles.items():
if profile and 0 < dimension <= 2000:
name = 'mc_vec_' + profile + '_' + str(dimension)
connection.execute(f"CREATE INDEX IF NOT EXISTS {name} ON chunks USING hnsw ((vector_value::vector({dimension})) vector_cosine_ops) WHERE profile='{profile}'")
index = {'id': workspace_id, 'workspace_id': workspace_id, 'generation': job_id,
'workspace_file': workspace_file, 'config_version': config_version, 'revision': 1,
'wiki_signature': job.get('wiki_signature', ''),
'code_coverage': job.get('code_coverage', {}),
'embedding_profile': job.get('embedding_profile', next((c.get('profile', '') for c in chunks if c.get('profile')), '')),
'embedding_mode': job.get('embedding_mode', 'all' if any(c.get('profile') for c in chunks) else 'off'),
'embedding_settings_revision': job.get('embedding_settings_revision', 0),
阅读时先找输入由谁提供，再区分纯计算、读取和写入；遇到异常或提前返回，追问调用方会得到什么，以及已有状态是否仍然有效。上面的代码只证明展示范围内的实现，跨模块行为应继续查看调用处和相应测试。
用一个小实验核对设计
建议在独立临时目录和隔离数据中操作，使用虚构的接口、文档与项目名称。
- 先发布一个可查询的小索引。
- 让下一次隔离构建故意失败。
- 再成功构建并对比生效代际。
**要核对的结果：**失败前后旧查询应可用，成功后才更新生效记录，不出现半新半旧结果。
可把观察记录为下表；没有执行的项写“未执行”，不能用设计预期冒充实测结果。
| 核对项 | 应保留的证据 |
|---|---|
| 使用条件 | 源码版本、配置、输入与项目范围 |
| 正常行为 | 实际返回、来源位置或状态变化 |
| 关键失败分支 | 错误码、可见结果与原状态是否保留 |
| 结论范围 | 已覆盖、未覆盖和需要进一步验证的条件 |
适用边界与继续阅读
本篇以静态源码与项目设计为依据，给出可执行的核对方法，未据此宣称已经完成新的性能或生产环境测试。不同版本的入口、默认值和底层实现可能变化，应先核对当前合同，再把方法用于自己的实际开发项目。
理解这一节后，可以回到专题总导航，沿同一模块继续阅读；顺序学习时使用专栏内的上一篇和下一篇。涉及具体技术选择，应连同本篇的代价与限制一起采用。