ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

Ollama+RAG代码知识库增量更新:让本地索引自动同步

Ollama+RAG代码知识库增量更新:让本地索引自动同步

我最初搭好 Ollama + RAG 代码知识库的那几天,心情确实不错。本地模型回答仓库问题,不用把代码送给任何外部服务,私密性和可玩性都拉满。但用了大概一周就发现了痛点:代码库不是静态的,每当我改了函数、新增模块、删掉旧文件之后,知识库里的答案还是旧版本。我甚至问它"刚才新增的某某接口怎么调用",它一本正经地给我编出了一个不存在的签名。

问题不在检索增强生成(RAG)本身,而在索引的"保鲜"。大多数 RAG 教程只教你如何全量扫描一次仓库,却很少讨论代码持续变化时怎么办。这篇文章我就完整记录一下,我如何把"全量索引一次"升级成"增量更新 + 自动同步",让本地代码知识库真正活起来。如果你也在用 Ollama 跑本地模型做代码问答,正被"索引老得快、重建等半天"折磨,这篇应该对你有用。

1. 全量重建不可持续:从一次"索引老化"事故说起

先复盘一下当时那个让我开始重构的现场。当时我搭好的流程很朴素:扫描仓库全部代码文件,按固定 token 窗口切块,用 Ollama 里的嵌入模型逐块生成向量,存入向量数据库,之后问答时检索最相关的块喂给对话模型。这套流程跑通之后效果不错,代码定位挺准。直到有一天我连续改了十几个文件,还新增了一个核心模块,然后向知识库提问时,它完全没意识到这些变化。

1.1 事故现场:改了 3 个文件,问答却引用旧代码

那次我改了三个文件:一个工具函数从递归改成了迭代,一个配置文件的字段改名,还有一个新增了日志模块。按理说增量很小,但我当时的知识库根本没有自动更新机制,唯一的刷新方式就是手动重跑全量索引脚本。而我那会儿仓库已经有几千个文件,全量跑一次要十几分钟,我懒得跑,就一直拖着。结果就是问答系统变成了"活在过去的助手"。

它回答时会把旧实现甚至已经删除的函数讲得头头是道。这不是模型能力问题,而是检索源的时效性问题。更麻烦的是,错误信息一旦被检索出来,模型会非常自信地基于它继续推理,你很难察觉哪里不对。要等到你亲自翻代码才发现知识库讲错了。那一刻我意识到,RAG 系统的质量下限不是模型决定的,而是索引新鲜度决定的。

1.2 为什么代码场景对索引时效性比普通文档敏感得多

代码知识库和文档知识库有个本质区别:文档内容相对稳定,你今天看和下周看差别不大;但代码是活物,一个文件一天可能被提交好几次。而且代码里存在大量强耦合关系——函数签名一变,所有调用点都受影响;接口字段一改名,上游全部失效。哪怕知识库只滞后了十几个提交,回答就会开始出现过时甚至错误的调用方式。

全量重建在代码场景下的代价也被放大了。普通文档可以按文档粒度切块,一个文档几十块;代码文件数量多、单文件又经常很短,动辄数万块。每块都要过一遍本地嵌入模型,在 CPU 环境下那是非常痛苦的等待。我手头这台机器跑 Ollama 嵌入模型,全量嵌入一个约有 3000 个文件、120MB 的代码仓库,耗时基本在 15 到 20 分钟。这个时长意味着你根本不想频繁重建,而不频繁重建又会让知识库迅速变旧,陷入两难。

1.3 把"全量"拆成"可追踪的增量":一个朴素的可行性判断

事后我冷静算了一笔账:假设一个仓库有 3000 个文件,日常迭代中一次变更通常只涉及 10 到 30 个文件。如果全量索引耗时 15 分钟,那么精心处理这 20 个文件理论上只要几秒钟到几十秒——中间的大头是嵌入调用和向量写入。这中间的差距足足有两个数量级。只要我能快速找出"哪些文件变了",增量更新在原理上就完全可行。

于是方案的核心从"怎么重建"变成了"怎么知道谁变了"。我想到三个状态:新增、修改、删除。围绕这三个状态做文章,就能避免重新处理 98% 的未变文件。要做到这一点,需要一个可靠的快照机制来记住每个文件的历史特征,还需要一个能感知变化的入口。这两个需求直接决定了后面的技术选型。

2. 架构与选型:增量管线、快照存储、监听器的三角组合

想清楚"增量"逻辑之后,我开始梳理需要哪些组件。整个系统说白了就是四块:向量数据库存语义向量,Ollama 提供嵌入和对话模型,一个轻量级的快照存储用来记住每个文件的指纹和历史状态,再加上一个文件系统监听器负责在代码变化的第一时间触发更新。

2.1 各组件选型:我只用本地能跑的、可替换的

选型上我没有追求新技术,全部选了成熟且容易替换的组件。向量库我用 ChromaDB,因为它原生支持按元数据删除和查询,对增量更新非常重要;嵌入模型选的是 Ollama 里的 nomic-embed-text,维度适中,本地 CPU 也能跑出不错的效果;快照存储直接用了 SQLite,单文件、零运维;监听器则是 watchdog,Python 生态里最顺手的文件事件库。

组件选型选它的核心原因
对话模型Ollama 上的 qwen / llama 系列本地推理,代码知识库不离开机器
嵌入模型nomic-embed-text768 维,CPU 可跑,代码语义表现够用
向量数据库ChromaDB支持 metadata 过滤删除,适合按块替换
快照存储SQLite + 文件哈希轻量可靠,记录文件指纹与处理状态
文件监听watchdog事件驱动,能精确感知新增/修改/删除

这套组合没有哪个环节是死绑的。向量库可以换、嵌入模型可以换,只要接口对齐,整个增量管线都能平移。我当时的原则就是:任何组件都别选那种和自家生态深度绑定的,否则后面想换都麻烦。

2.2 为什么是哈希快照 + 文件监听,而不是 git hook 或定时全量

很多人第一反应是:代码仓库本身有 git,用 hook 岂不是更准?理论上没错,但我在实际场景里发现三个问题。第一,不是所有代码变化都会经过 git,比如本地生成文件、临时脚本、IDE 的自动保存中间态;第二,post-commit 这类钩子只覆盖提交动作,而你在工作区里改完还没提交时,知识库就已经需要感知变化了;第三,真实环境里知识库索引的未必只有一个 git 仓库,可能是多个目录拼起来的,hook 的接入成本会成倍上升。

定时全量重扫更不用考虑,它是回到老路。真正的解法是两条腿走路:日常由 watchdog 监听目录变化做精细化增量更新,做到秒级响应;服务启动时再做一次全量快照对账,把监听器遗漏的、外部工具直接改文件的情况兜底处理。这个组合我在后面会详细展开。

2.3 代码切块策略:固定 token 窗口在代码场景并不够用

在全量时代我犯过一个典型错误:用固定 token 窗口去切代码文件,导致一个函数被切断,检索时只命中半个函数体,模型拿到的上下文残缺不全。后来我改成按语法结构切块——优先按函数、类、方法定义切分,大文件再按顶层定义边界拆。这样每个向量块基本对应一个逻辑单元,检索命中后能直接作为完整代码片段交给模型。

代码块还有一个容易被忽视的点:metadata 必须丰富。每个块我都会记录三个关键字段:source 是文件路径,chunk_index 是块在文件内的序号,file_hash 是该文件的哈希值。前两个字段负责删除旧向量,file_hash 负责判断文件是否真的变了。这三个字段是整个增量更新的地基,缺一个都会让后续流程卡壳。

3. 增量索引的实现细节:从哈希快照到向量分块替换

架构定好之后,我进入了编码阶段。整个增量管线大概分四步:扫描文件、比对哈希快照、按变更类型处理、更新快照与向量。下面我把每一步的关键实现和理由写清楚,这块内容基本是照着就能抄的水平。

3.1 SQLite 快照表和文件哈希:为什么选哈希而不是修改时间

先建一张快照表,用来记录每个文件的路径、哈希值以及它对应的向量块数量。表结构不用复杂,够用就行:

CREATE TABLE IF NOT EXISTS file_snapshot ( file_path TEXT PRIMARY KEY, file_hash TEXT NOT NULL, chunk_count INTEGER NOT NULL, last_updated TEXT NOT NULL );

每次扫描时,我遍历目标目录下所有代码文件,计算其哈希值,再和快照表里的记录比对。有人会问,直接用文件的 mtime 不更快吗?mtime 确实快,但它有个致命问题:不可靠。文件内容可能因为权限调整、touch 命令、版本管理操作而更新 mtime,但内容完全没变;反过来,复制文件时也可能把 mtime 一起保留,导致内容变了却看不出。哈希虽然要读一遍文件,但换来的是确定性。

我用的哈希很简单,就是 MD5,有人担心碰撞,但在本地代码文件这个场景里完全够用,计算快,代码也短:

import hashlib from pathlib import Path def file_hash(path: Path) -> str: hasher = hashlib.md5() with open(path, "rb") as f: for chunk in iter(lambda: f.read(65536), b""): hasher.update(chunk) return hasher.hexdigest()

读 65536 字节一块,避免把大文件整个载入内存。对一个几 MB 的源码文件来说,计算时间可以忽略不计。

3.2 新增、修改、删除三种状态的统一处理管线

有了快照比对,每个文件就能归类到三种状态之一。新增:快照里没有记录,目录里存在该文件。修改:目录里的哈希与快照不一致。删除:快照里有记录,但目录里已经没有该文件。

针对三种状态,我实现了一个统一的process_change函数。它的核心流程是:如果新增,走嵌入入库;如果修改,先把该文件旧向量清掉,再走嵌入入库;如果删除,直接清理向量并删掉快照记录。伪代码大致是这样:

def process_change(file_path: Path, state: str): rel_path = str(file_path.relative_to(REPO_ROOT)) match state: case "added": chunks = split_code_by_functions(file_path) embeddings = embed_chunks(chunks) vector_store.add(embeddings, metadatas=build_meta(rel_path, file_hash(file_path))) snapshot.upsert(rel_path, new_hash, len(chunks)) case "modified": vector_store.delete_by_metadata({"source": rel_path}) chunks = split_code_by_functions(file_path) embeddings = embed_chunks(chunks) vector_store.add(embeddings, metadatas=build_meta(rel_path, file_hash(file_path))) snapshot.upsert(rel_path, new_hash, len(chunks)) case "deleted": vector_store.delete_by_metadata({"source": rel_path}) snapshot.remove(rel_path)

新增和修改在向量处理上几乎一样,差别只在删除旧向量这一步。这也是为什么我强调 metadata 里必须有 source 路径——没有它,你就没法精准删掉属于某个文件的全部向量。如果你一开始切块时没有记录这些元数据,到这一步只能含泪重建整个向量库。

3.3 ChromaDB 的按块删除:先查 ID 再删的隐藏细节

这里有一个我踩过的坑值得单独说。ChromaDB 的delete方法可以传where条件,但搭配的 metadata 值必须和写入时完全一致,类型也不能变。上面我写的delete_by_metadata({"source": rel_path})其实是个辅助函数,真实逻辑里它需要先按条件查询出所有匹配的向量 ID,再调delete(ids=...)。

def delete_by_metadata(collection, where: dict): existing = collection.get(where=where) if existing and existing["ids"]: collection.delete(ids=existing["ids"])

原因在于,某些向量库版本不支持纯 where 直接删除,或者删除后不会自动清理内部索引,用先查后删的方式更稳妥。另外,修改文件时我用的删除维度是 source + chunk_index 的组合,而不是只按 source。因为一个文件修改后 chunk 数量可能变化,只按路径删不干净旧块,必须连同 chunk_index 一起定位。实际操作中我干脆按 source 全部删掉,然后全部重新嵌入——反正一个文件块数有限,重新嵌入的成本很小。

3.4 嵌入调用的批量处理:为什么要一次请求塞多个文本块

嵌入模型是按 token 计费的——不过我这里是指计算时间,不是指花钱,Ollama 是本地跑的。调用时最忌讳一个一个嵌入,那样网络/本地进程往返开销极大。我实测过,一条条调用嵌入接口,一次 800 块的嵌入任务耗时接近十分钟;改成批量请求后,同一批 800 块只需要几分钟,差距非常明显。

批量调用的写法不复杂,Ollama 的/api/embed接口本身就支持传一个文本数组:

import requests def embed_chunks(chunks: list[str], model: str = "nomic-embed-text", batch_size: int = 32): results = [] for i in range(0, len(chunks), batch_size): batch = chunks[i:i + batch_size] resp = requests.post("http://localhost:11434/api/embed", json={ "model": model, "input": batch }) data = resp.json() for item in data["embeddings"]: results.append({"embedding": item, "text": ""}) # 按实际需求决定是否保留原文 return results

这里的text字段我最终选择不存向量库,因为代码原文完全可以通过路径重新读出来,省一点磁盘空间。但要注意,如果你需要做关键词精确检索,那还是得存一份原文。

3.5 启动全量快照对账:把漏网之鱼一网打尽

监听器能覆盖"运行期间的变化",但它覆盖不了几种情况:服务没启动的时段、进程被系统杀掉后的时间窗、以及某些编辑器直接通过替换文件方式保存导致事件丢失。因此我在服务启动时加了一个对账动作:扫描一遍目录,和 SQLite 快照比对,把不一致项补齐。

def reconcile_at_startup(): current_files = indexable_files(REPO_ROOT) snap_map = {r.file_path: r.file_hash for r in snapshot.all()} for rel_path in current_files: h = file_hash(Path(REPO_ROOT) / rel_path) if snap_map.get(rel_path) != h: changes.append((rel_path, "modified" if rel_path in snap_map else "added")) for rel_path in snap_map: if rel_path not in current_files: changes.append((rel_path, "deleted")) for rel_path, state in changes: process_change(Path(REPO_ROOT) / rel_path, state)

这段逻辑把三种状态全部覆盖,即使你完全没跑过监听器,只要启动过一次,知识库也能自愈。我个人习惯是给这个对账过程加一个计时,如果差异超过较大阈值,会打印日志提醒自己——历史上出现过因为磁盘快照损坏导致大面积重建的情况。

4. 自动同步的双保险:文件监听与启动时快照对账

增量管线有了,接下来要解决的是"自动"两个字。所谓自动同步,我理解成两件事:日常运行中代码一变就马上感知并处理;万一感知失败了,系统也得有兜底。所以我最终采用了文件监听 + 启动对账的双保险结构。

4.1 watchdog 监听落地:事件风暴是我踩过的第一个坑

watchdog 的用法很直接,我实现了一个继承FileSystemEventHandler的类,监听目录的创建、修改、删除事件。但第一次实跑就遇到了事件风暴:用编辑器保存文件时,会触发modified、created、deleted多个事件,而且同一事件可能被触发多次。如果不加去重,一次保存就会导致重复嵌入、重复删除。

我的处理方式是:事件处理器只负责把"可疑路径"放进一个带锁的待处理集合,然后由一个后台线程每隔固定时间批量消费集合,去重后再做增量处理。这样既能合并同一文件在几百毫秒内的多次事件,也能避免在主线程里做耗时操作。

import threading import time from watchdog.events import FileSystemEventHandler class RepoChangeHandler(FileSystemEventHandler): def __init__(self, callback): self.pending = set() self.lock = threading.Lock() self.callback = callback threading.Thread(target=self._consumer, daemon=True).start() def _push(self, path): with self.lock: self.pending.add(path) def on_modified(self, event): if not event.is_directory and event.src_path.endswith((".py", ".ts", ".go", ".md", ".sql", ".rst", ".txt", ".yaml", ".yml", ".json", ".toml")): self._push(event.src_path) def on_created(self, event): if not event.is_directory and event.src_path.endswith((".py", ".ts", ".go", ".md", ".sql", ".rst", ".txt", ".yaml", ".yml", ".json", ".toml")): self._push(event.src_path) def on_deleted(self, event): if not event.is_directory and event.src_path.endswith((".py", ".ts", ".go", ".md", ".sql", ".rst", ".txt", ".yaml", ".yml", ".json", ".toml")): self._push(event.src_path) def _consumer(self): while True: time.sleep(1) with self.lock: paths = list(self.pending) self.pending.clear() for p in paths: self.callback(p)

这个去重思路很土,但非常有效。注意我列的扩展名是白名单机制,只有这些文件类型才会进入索引流程,其它如__pycache__下的文件、图片、二进制文件等直接忽略,避免无效扫描。

4.2 为什么保存事件会重复:编辑器替换写入与原子重命名

写代码的时候我对"保存一个文件为什么会有删除事件"困惑了很久。后来查了资料才明白:很多编辑器保存文件时不是直接往原文件里写,而是先写一个临时文件,再通过原子重命名替换原文件。从文件系统监听器的角度看,这就是"删除旧文件 + 新建新文件"两个事件,甚至中间还有短暂的文件不存在窗口。

如果处理逻辑写得不够健壮,这个窗口会导致两种情况:一是把文件误判为删除,清掉所有向量后又被新增事件加回来,白白浪费一轮处理;二是事件顺序倒置,先新增后删除,导致最终状态错误。我的解决办法就是前面说的批量去重:把同一路径的事件在 1 秒内合并,然后处理时直接以磁盘当前状态为准,重新计算哈希,而不是依赖事件本身携带的类型。也就是说,事件只是一个"提示",处理逻辑永远以文件系统当前事实为判断依据。

4.3 监听器与对账如何分工:谁漏了谁兜住

日常运行中,watchdog 负责秒级响应,启动对账负责兜底。两者不是重复劳动,而是互补的。监听器面对的场景是"代码正在频繁变化",比如你一边改代码一边问知识库,这时候秒级增量更新能保证答案紧跟你的编辑动作;对账面对的场景是"系统离线了一段时间",比如你关闭服务,第二天开机再跑,这个时间段内无法依赖事件触发,必须靠一次性全量比对来修复。

我实际运行了几个月,发现对账启动后真正处理的通常是以下几类:前一天关闭服务前没来得及同步的改动、外部脚本批量改文件触发的差异、以及本地生成物被我加入了忽略列表导致快照里残留的垃圾。对账的扫描加上哈希计算,对一个 3000 文件的仓库大概耗时一两分钟,远低于全量重建,可接受。

索引状态可视化也是"自动同步"能长期运行的关键。我在快照表里顺手维护了一个变更记录表,每次增量处理都会写入一条日志,包括时间、文件路径、变更类型、耗时。这样出了诡异问题,直接查表就能定位,而不是对着空白的终端发呆。

5. 压测数据与踩坑清单:那些文档里不会写的细节

系统跑通之后,我做了一轮比较系统的压测,把全量索引和增量更新的差距量化了一下。同时记录下了几类高频踩坑问题,这里一并整理出来,希望能帮你省掉几个晚上的调试时间。

5.1 一组实测数据:3000 文件的仓库,增量更新到底多快

我拿本地一个模拟业务仓库做测试,目录规模大约 3000 个文本/代码文件,总容量约 120MB,我随机修改了 20 个文件,比对全量重建和增量同步的耗时差异。

项目全量重建增量更新
文件扫描3000 个全部扫描3000 个扫描用于比对,但仅 20 个进入处理
嵌入调用所有块的嵌入,约 1.8 万次文本块仅改动文件的块,约 300 块
向量写入全量写入,约 1.8 万条先删 300 条旧向量,再写 300 条新向量
总耗时约 18 分钟约 1 分 30 秒

注意,这 1 分 30 秒里还包括了启动时对账的固定成本,如果完全由监听器驱动,跳过对账,实际耗时大概只有二三十秒。增量更新的效率优势非常夸张。当然,哈希扫描本身也需要时间,但它不需要调用嵌入模型,所以瓶颈完全不在那里。如果你的仓库更大,比如上万个文件,对账时的哈希扫描会成为主要开销,那时候可以考虑把文件清单缓存起来,只对新增和目录变化做系统调用级别的判断,但我目前这个规模还不需要。

5.2 踩坑清单:嵌入模型温启动、事务锁冲突、误删向量库

最后说说我实际踩过的几个坑,每个都花了我不少时间排查。

第一,嵌入模型的温启动问题。Ollama 加载模型需要时间,首次调用接口时等待很吓人,有时候要接近十秒。如果是增量更新里首次调用,在 Consumer 线程里会被用户感知为"保存文件后没反应"。解决方法是服务启动时先向 Ollama 发送一次预热请求,把模型固定在内存里,后面调用延迟就正常了。

第二,SQLite 的并发写锁。增量更新本身是后台线程在跑,但如果启动对账和监听器同时触发了写操作,SQLite 会报database is locked。我一开始只设置了一个后台线程,后来发现对账和监听都往同一个消费队列推任务,内部其实还是单线程,倒是避免了这个问题。但如果你自己实现了多线程直接写快照,一定要在写入时加锁或者用BEGIN IMMEDIATE事务。

第三,向量库误删。有一版代码我在处理"修改"状态时,把where条件写成了{"source": rel_path},但部分早期写入的 metadata 里 source 存储的是绝对路径,后面偶尔又写相对路径,结果同一文件出现新旧两套向量,删除也删不干净。排查了半天才意识到是路径格式不统一。这个问题最终靠强制在所有 metadata 中统一使用从仓库根目录算起的相对路径解决,并在入口处做了校验和转换。

还有一个小但隐蔽的坑:file_hash字段我没有放进向量 metadata,而是只放在 SQLite 快照里。为什么?因为向量库的 metadata 每次更新都要重新传,如果你把哈希放进去,修改文件后删除旧向量时,旧哈希会变成脏数据,反而干扰判断。哈希的唯一职责是快照比对,不要让它在向量库里承担额外角色。

5.3 这套增量管线的边界:它不解决检索质量,只解决索引保鲜

这里必须说句公道话。增量更新解决的是"索引不新鲜"的问题,它不能解决"代码语义理解不够好"的问题。如果你的嵌入模型本身对代码的理解一般,或者切块策略太粗糙,增量做得再快,回答质量也可能一般。我见过有人把检索质量不佳归结为"需要增量更新",其实是两码事。

我的经验是,先把切块策略和 metadata 设计做好,再上增量更新。切块决定了检索粒度的上限,增量更新只是保证你检索到的内容是最新的。顺序反了,后面会很被动。我自己是先实现了增量,后来才发现切块粒度太大导致检索到半个类,又回头调整切块逻辑,多花了一轮功夫。

最后再分享一个我目前还在扩展的小方向:把历史提交信息也做成增量索引。代码库真正的知识不只存在于"当前快照",很多设计决策、为什么这么写的上下文,都藏在提交说明里。但提交日志的量级比代码文件还要大,全量索引不现实,正好这套增量更新机制可以用上——监听新提交,只看 diff,只嵌入变更部分。我已经在内部跑出一个雏形,效果还有待验证,但思路和这套管线的设计是一致的:只处理变化,不碰存量。

RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进