索引、增量更新与删除

索引是为了快速查找而提前建立的数据结构。倒排索引把“词项”映射到含有它的文档编号；词项是切词后用于查找的单位。资料更新时既要加入新词，也要移除旧词的关联；删除文档需要同步清除所有索引引用。

用退款资料观察索引变化

我们给退款和配送资料建索引，先查“退款”，再把退款政策从“三至五日”改成“二至四日”，最后删除退款资料。观察旧期限、更新期限与删除结果。

为看清词项对应关系，固定正文用空格明确切出业务词：“退款 到账 三至五日”包含三个词项，编号P1；“配送 两天”包含两个词项，编号D1。这里的字典和集合都存放在内存中，一次运行从空索引开始。

保存完整程序并运行

先在 PowerShell 创建并进入学习目录：

mkdir "F:\学习\l5-indexing" -Force
cd "F:\学习\l5-indexing"

保存下面的文件后，在这个终端运行 dir 核对文件名，再执行本节给出的运行命令。

新建学习目录，例如 F:\学习\l5-indexing。将下面代码完整保存为 indexing_demo.py；代码包含固定资料和运行入口，保存后就能执行。

先在 PowerShell 创建并进入学习目录：

mkdir "F:\学习\l5-indexing" -Force
cd "F:\学习\l5-indexing"

保存下面的文件后，在这个终端运行 dir 核对文件名，再执行本节给出的运行命令。

"""用倒排索引观察退款资料的新增、更新与删除。"""
class Index:
    def __init__(self):
        self.documents = {}
        self.postings = {}  # 词项 -> 包含它的文档编号集合。

    def delete(self, doc_id):
        old = self.documents.pop(doc_id, None)
        if old is None:
            return False
        for term in set(old.split()):
            self.postings[term].discard(doc_id)
            if not self.postings[term]:
                del self.postings[term]
        return True

    def upsert(self, doc_id, text):
        # 输入先检查，再删除旧关联；空更新不会误删当前资料。
        if not text.strip():
            raise ValueError("正文不能为空")
        self.delete(doc_id)
        self.documents[doc_id] = text
        for term in set(text.split()):
            self.postings.setdefault(term, set()).add(doc_id)

    def search(self, term):
        return sorted(self.postings.get(term, set()))

def main():
    index = Index()
    # 用空格标出业务词项，便于直接观察倒排表。
    index.upsert("P1", "退款 到账 三至五日")
    index.upsert("D1", "配送 两天")
    print("退款关联：" + str(index.search("退款")))
    print("三至五日关联：" + str(index.search("三至五日")))
    index.upsert("P1", "退款 到账 二至四日")
    print("更新后旧期限：" + str(index.search("三至五日")))
    print("更新后新期限：" + str(index.search("二至四日")))
    index.delete("P1")
    print("删除后退款：" + str(index.search("退款")))
    print("配送仍在：" + str(index.search("配送")))

if __name__ == "__main__":
    main()

在保存文件的文件夹地址栏输入 powershell 并回车，再执行：

macOS：先在 Python 篇的学习文件夹中打开终端，执行 source .venv/bin/activate。再输入 cd ，把当前例子的文件夹拖入终端并回车，运行下面的命令。

cd "F:\学习\l5-indexing"
ls
python -X utf8 indexing_demo.py

使用其他目录时，把第一行换成自己的路径。dir 应能列出 indexing_demo.py。终端输出如下：

退款关联：['P1']
三至五日关联：['P1']
更新后旧期限：[]
更新后新期限：['P1']
删除后退款：[]
配送仍在：['D1']

首次查找通过词项直接定位P1；更新后“三至五日”没有关联，“二至四日”指向同一P1；删除后退款词项也没有关联。配送资料D1始终保留，说明更新和删除按文档范围执行。

文档与词项关联怎样一起更新

documents保存文档编号与正文，postings保存词项对应的编号集合。upsert表示“存在就更新，不存在就新增”。它先验证正文，再移除旧关联，保存新正文，建立新的词项关联。集合去掉重复词项，因此一份资料写两次“退款”不会产生两个编号。

self.postings.setdefault(term, set()).add(doc_id) 可以分两步读：这个词还没有索引时，先建立空集合；随后把当前文档编号放进集合。例如第一次加入“退款”时得到 退款 → {P1}，以后命中同一词就继续加入编号。

delete先移除文档，再遍历旧正文中的词项并删去编号。某个词项再无文档关联时，连空集合一并清理。文档存储与索引必须同时更新，否则搜索可能返回已删资料，或保留旧条款。实际服务宜采用事务、版本化索引切换或可重放的更新队列，避免更新中途被查询看到半成品。

全文检索常用倒排索引。向量索引则存放嵌入向量，即把文本转换为数字表示，再按向量距离查找邻近资料；高维向量常用近似近邻结构加快检索，会在速度与召回之间取舍。向量索引也需要稳定文档/块编号，修改正文后重算向量，删除时清理旧向量。

有些服务通过墓碑标记暂时隐藏已删项，再批量压缩物理存储。逻辑删除后应立即在查询阶段过滤，并检查缓存、复制节点和备份保留策略。这里的空格切词用于观察映射；中文全文检索需要一致的切词规则，让入库与查询使用同一套词项。

改一处输入，核对机制变化

把 main 中 index.delete("P1") 改为 index.delete("D1")，运行 python -X utf8 indexing_demo.py。最后两行退款仍返回P1、配送变为空列表；标签沿用原样，实际删除对象由参数决定。恢复P1再重跑，确认只删除目标文档的关联。

运行配套检查

把下面测试保存为 test_indexing_demo.py，与主程序放在同一目录。5项测试检查上面的实际处理结果，也覆盖更新、删除与空更新保留原文。

"""核对索引、增量更新与删除的正常结果与具体边界。"""
import unittest
import indexing_demo as demo


class DemoTests(unittest.TestCase):
    def test_update_removes_old_postings(self):
        # 更新要清除旧词项并建立新关联。
        index = demo.Index()
        index.upsert("P1", "退款 旧期限")
        index.upsert("P1", "退款 新期限")
        self.assertEqual(index.search("旧期限"), [])
        self.assertEqual(index.search("新期限"), ["P1"])

    def test_delete_keeps_other_documents(self):
        # 删除只移除目标编号，其他文档保留。
        index = demo.Index()
        index.upsert("P1", "退款")
        index.upsert("P2", "退款")
        self.assertTrue(index.delete("P1"))
        self.assertEqual(index.search("退款"), ["P2"])

    def test_repeated_upsert_has_single_posting(self):
        # 重复入库与重复词项不会重复编号。
        index = demo.Index()
        for _ in range(2):
            index.upsert("P1", "退款 退款")
        self.assertEqual(index.search("退款"), ["P1"])

    def test_invalid_update_preserves_existing_document(self):
        # 错误更新不能破坏已有资料。
        index = demo.Index()
        index.upsert("P1", "退款")
        with self.assertRaises(ValueError):
            index.upsert("P1", " ")
        self.assertEqual(index.search("退款"), ["P1"])

    def test_unknown_delete_is_safe(self):
        # 删除不存在的编号返回明确结果。
        self.assertFalse(demo.Index().delete("missing"))


if __name__ == "__main__":
    unittest.main()

python -X utf8 -m unittest -v test_indexing_demo.py

测试名称后应显示 ok，末尾是 Ran 5 tests 和 OK。

运行时怎样定位问题

索引完成后，检索会比较关键词路径与向量路径。

参考阅读：LlamaIndex 源码与检索流程。阅读时可以沿文档、节点、索引和检索入口对照这里的数据流。

下一篇：稀疏、稠密与混合检索。