跳到主要内容

Embeddings 编排类与索引生命周期

30 秒导读: Embeddings 是 txtai 的总控类。它本身不实现向量算法、不实现 ANN、不实现打分,而是像一个乐队指挥:手里握着一排"乐器槽位"(ANN、database、scoring、graph、subindexes),当你调用 index() 时,它负责按正确的先后顺序把一条文档流分发给各个子系统,再让每个子系统各自建好索引。本章只讲这个"指挥"怎么工作——具体乐器怎么发声,交给 02/03/04。

1. 这是什么(零基础也能懂)​

一句话定义: Embeddings 类是 txtai 的顶层入口,负责**编排(orchestrate)**一次索引构建——把你给的一批文档,同时喂进它启用的所有存储与索引子系统。

它要解决的问题很具体。一个"语义搜索引擎"从来不是单一部件,它至少要同时维护:

  • 一份稠密向量索引(ANN),用来做"意思相近"的检索;
  • 可能还有一份稀疏/关键词索引(scoring),做 BM25 或稀疏向量;
  • 一个关系数据库(content store),存原文和元数据,支持 SQL 过滤;
  • 一张图(graph),存文档间关系;
  • 甚至一组子索引(subindexes),一份数据建多套索引。

难点在于:这些子系统各有各的构建时机和依赖关系。谁先建、谁的数据依赖谁、一条文档要被抄送几份——如果每个调用方自己处理,就会乱。Embeddings 把这套编排收进一个类里,对外只暴露 index() / upsert() / delete() / search() 这几个动词。

用起来什么样: 三行就能建一个语义索引。

from txtai import Embeddings

# 建一个默认稠密索引(自动下载 all-MiniLM-L6-v2 向量模型)
embeddings = Embeddings(content=True)
embeddings.index([(0, "US tops 5 million virus cases", None),
                  (1, "Canada's last fully intact ice shelf collapses", None)])

embeddings.search("natural disaster", 1)   # → [{'id': '1', 'score': 0.28...}]

一句话直觉: 把 Embeddings 想成一个总装配线的调度台。它不亲手拧螺丝(不算向量、不建 HNSW 图),它只决定"哪个工位先开工、零件按什么顺序流过每个工位"。

本节到此不涉及底层。下面开始看它内部怎么转。

2. 顶层全景(它大概怎么转)​

2.1 组件槽位:一个类,一排"插槽"​

Embeddings.__init__ 做的第一件事,是把自己身上所有可能的子系统都初始化成 None,然后交给 configure() 按配置点亮其中一部分(embeddings/base.py:30-83,构造函数;base.py:761 configure)。

这些插槽就是整个类的骨架:

插槽属性装的是什么由哪个工厂创建
self.model稠密向量模型(文本→向量)loadvectors → VectorsFactory
self.ann稠密近邻索引(ANN)createann → ANNFactory
self.scoring稀疏/关键词打分索引createscoring → ScoringFactory
self.database内容库(SQL content store)createdatabase → DatabaseFactory
self.graph图网络creategraph → GraphFactory
self.indexes子索引集合createindexes → Indexes
self.ids无 database 时的 indexid→id 映射createids → IndexIds
self.reducer降维模型(仅 word vectors)Reducer(超出本章)
self.query / self.functions / self.archive查询翻译 / SQL 函数 / 归档各自的 loader

关键设计:每个子系统都是"可选"的。你的配置里没开 graph,self.graph 就一直是 None,后面所有 if self.graph: 分支都跳过。整个类因此像一个"按配置组装"的可插拔框架。

self.models 是一个跨 embeddings 共享的模型缓存:启用子索引时,父索引和各子索引可以共用同一份向量模型,避免重复加载(base.py:896-898、base.py:1046-1047)。

2.2 三类判定谓词:后面所有分支都靠它们​

编排逻辑里满是 if self.issparse() / if self.isdense() 这样的判断。它们是四个一句话的谓词(base.py:713-741、base.py:1077):

谓词含义判据
isdense()有没有稠密 ANNself.ann is not None
issparse()scoring 是不是"稀疏索引"(可搜)self.scoring and self.scoring.issparse()
isweighted()scoring 是不是"词权重"(仅给 word vectors 用)self.scoring and self.scoring.isweighted()
hassparse()配置里的 scoring 是否声明为稀疏索引ScoringFactory.issparse(config)

这里有个容易混淆但很重要的区别:同样是 scoring,它有两副面孔。一副是可独立检索的稀疏索引(BM25 / 稀疏向量,issparse()),另一副只是给 word-vectors 模型算词权重的辅助表(isweighted())。configure() 里正是用 hassparse() 把这两者分开——只有当 scoring 不是稀疏索引时,才在配置阶段就提前建它给词权重用(base.py:776-777)。

2.3 顶层数据流​

一次 index() 的宏观流向(怎么读这张图:从上到下是时间顺序,虚线是"数据被抄送"):

                       embeddings.index(documents)
                                  │
                                  ▼
                    ┌──────────────────────────┐
                    │  initindex(): 先把各子系统 │  database / scoring /
                    │  建好(ANN 除外)          │  graph / subindexes 就位
                    └──────────────┬───────────┘
                                   ▼
             Stream(documents)  规整成 (id, data, tags) 三元组
                                   │  (缺 id 时用 AutoId 补)
                                   ▼
        ┌──────────────  Transform.__call__  ──────────────┐
        │  遍历文档流,一遍完成两件事:                       │
        │                                                   │
        │   (A) 抽出"待向量化文本" ──► 向量模型 ──► 大向量数组 │
        │   (B) 原始文档批量抄送:                            │
        │        ├┈┈► database.insert                        │
        │        ├┈┈► scoring.insert                         │
        │        ├┈┈► indexes.insert (子索引文档流)           │
        │        └┈┈► graph.insert                           │
        └───────────────────────┬───────────────────────────┘
                                 ▼
             createann() → ann.index(embeddings)   ← 稠密索引在此才建
                                 ▼
             scoring.index()     ← 必须在 graph 之前
                                 ▼
             indexes.index()     ← 各子索引各自 build
                                 ▼
             graph.index(...)    ← 最后,可复用上面的索引结果

一句话总结这张图:先建容器(initindex)→ 一条流同时喂满所有容器(Transform)→ 按依赖顺序逐个 build。 下一节把每一步拆开讲。

3. 核心原理​

3.1 生命周期第一步:initindex —— 为什么 ANN 偏偏不在这建​

index() 的第一行就是 self.initindex(reindex)(base.py:114)。这一步的职责是把除 ANN 外的子系统全部先建好(base.py:785-821):

# 示意,非源码:initindex 的核心动作
self.defaults()                       # 补默认配置(默认模型、稀疏简写展开)
self.ids = None
if not reindex:
    self.database = self.createdatabase()   # 内容库
    self.archive = None
if self.ann:
    self.ann.close()
self.ann = None                       # 关键:ANN 先清空,晚点才建
if self.hassparse():
    self.scoring = self.createscoring()     # 稀疏索引
self.indexes = self.createindexes()   # 子索引
self.graph  = self.creategraph()      # 图

为什么 database / scoring / graph 现在就建,ANN 却要等? 因为它们的构建输入不同:

  • database / scoring / graph 是流式接收原始文档的——文档在 Transform 遍历时就要往里 insert,所以它们必须先于遍历存在。
  • ANN 存的是算好的向量,而向量的维度(dimensions)只有算完第一批才知道。所以 ANN 必须等 Transform 跑完、拿到向量数组和维度后才能创建(base.py:129-136)。

这就是整个生命周期最核心的一条时序依赖:"能流式接收原始数据的先建;需要向量维度的后建"。

注意 reindex=True 时 createdatabase 被跳过——因为 reindex 是"数据库已有内容,只重建索引"的场景,原数据库要保留(见 §3.6)。

3.2 生命周期第二步:Stream —— 把杂乱输入规整成三元组​

txtai 允许你喂各种形状的文档:纯字符串、(id, data)、(id, data, tags)、或 dict。Stream 是第一道整流器,把它们统一成 (id, data, tags)(index/stream.py:37-62):

# 示意,非源码:Stream 的归一化逻辑
for document in documents:
    if isinstance(document, dict):
        document = document.get("id"), document, document.get("tags")
    elif isinstance(document, tuple):
        document = document if len(document) >= 3 else (document[0], document[1], None)
    else:
        document = None, document, None          # 裸数据 → id 留空
    if self.action and document[0] is None:      # id 为空时才补
        document = (self.autoid(document[1]), document[1], document[2])
    yield document

关键细节:自动 id(AutoId)。 当文档没带 id,Stream 用 AutoId 补一个(index/autoid.py)。它支持两种策略:

配置 autoid策略实现
整数(默认)自增序列sequence():返回当前值再 +1(autoid.py:53-68)
"uuid1"/"uuid4" 等随机 UUIDuuid():调用对应 uuid.uuidN()(autoid.py:70-82)
"uuid3"/"uuid5"确定性 UUID用数据内容 + NAMESPACE_DNS 生成,同内容同 id

自增序列的"当前值"会在流结束时写回 self.config["autoid"](stream.py:64-67),这样下次 upsert 能接着上次的序号往下发,而不会从 0 重来。

3.3 生命周期第三步:Transform —— 一条流,同时干两件事​

Transform 是本章的心脏。它接收 Stream 吐出的三元组流,一次遍历同时完成分发和向量化(index/transform.py)。入口 __call__ 根据"有没有稠密模型"分岔(transform.py:60-82):

# 示意,非源码:Transform 入口的分岔
if self.model:                     # 有稠密模型 → 走向量化路径
    ids, dimensions, embeddings = self.vectors(documents, buffer)
else:                              # 纯稀疏/无向量 → 只收集 id
    ids = self.ids(documents)
return (ids, dimensions, embeddings)

真正的编排发生在 stream() 这个生成器里(transform.py:123-168)。它一边筛出要向量化的文本 yield 出去,一边攒批把原始文档抄送给各数据存储:

# 示意,非源码:Transform.stream 的双重职责
batch, offset = [], 0
for document in documents:
    if isinstance(document[1], dict):
        # dict:只把 text 或 object 字段送去向量化
        if self.text in document[1]:
            yield (document[0], document[1][self.text], document[2])
            offset += 1
        elif self.object in document[1]:
            yield (document[0], document[1][self.object], document[2])
            offset += 1
        # 没有可向量化字段的 dict:只入库,不 yield
    else:
        yield document              # 非 dict:整体向量化
        offset += 1

    batch.append(document)          # 无论是否向量化,原文都要攒批
    if len(batch) == self.batch:
        self.load(batch, offset)    # 满批 → 抄送各数据存储
        batch, offset = [], 0
if batch:
    self.load(batch, offset)        # 收尾批

这里有两条容易漏掉的暗线:

  1. "入库的文档集" ≠ "被向量化的文档集"。 一个只有元数据、没有 text/object 字段的 dict 会被存进 database,但不会产生向量。所以下游的 offset 管理必须小心(见下)。
  2. 一次遍历,多路分发。 上游生成器只能消费一次,Transform 因此在同一个 for 循环里既做筛选又做抄送,避免把大数据流缓存两遍。

load() 负责真正的抄送(transform.py:170-208):

# 示意,非源码:Transform.load —— 把一批原文分发到各存储
if self.action == Action.UPSERT:          # upsert 时先删旧
    deletes = [uid for uid,_,_ in batch if uid not in self.deletes]
    if deletes:
        self.delete(deletes)
        self.deletes.update(deletes)
if self.database and self.action != Action.REINDEX:
    self.database.insert(batch, self.offset)
if self.scoring:
    self.scoring.insert(batch, self.offset, self.checkpoint)
if self.indexes:
    self.indexes.insert(batch, self.offset, self.checkpoint)
if self.graph:
    self.graph.insert(batch, self.offset)
self.offset += offset                      # 只按"向量化了的数量"递增

offset(索引偏移量)是这里最精妙的一环。 每条向量在 ANN 里的位置是一个整数下标(indexid),它必须和 database / scoring / graph 里记录的下标对齐。注意 self.offset 只加 offset(本批被 yield 去向量化的数量),不是 len(batch)——因为没被向量化的文档不占 ANN 的下标位。这保证了"第 N 条向量"在所有子系统里指向同一份原文。

  • INDEX 动作 offset 从 0 起;UPSERT 从 config["offset"] 续起(transform.py:42)。
  • 无稠密模型时,ids() 路径遍历完后把最终 offset 写回 config["offset"](transform.py:118-119),供下次 upsert 接续。

标量量化(quantize)。 若配置了整数 quantize(如 1-bit),Transform 把向量 dtype 设为 np.uint8 而非 float32,向量以量化字节写入(transform.py:46-47、transform.py:96-100)。量化的具体位打包算法属于向量化层,见 02;这里只负责把 qbits 传下去。向量数组本身走一个磁盘上的 memmap 临时文件(buffer)以支持超内存的大批量。

3.4 生命周期第四步:index() 里的 build 顺序 —— 依赖驱动的先后​

Transform 跑完后,回到 index()(base.py:120-153),开始逐个 build。这段顺序不是随意的,每一步都卡着前一步的产物:

transform() 返回 (ids, dimensions, embeddings)
   │
   ├─ 若有 embeddings:
   │     [可选] Reducer(PCA) 去主成分         base.py:124-127
   │     config["dimensions"] = dimensions    ← 记下维度
   │     self.ann = createann()               ← 现在才建 ANN(需要维度)
   │     ann.index(embeddings)                ← 灌入向量
   │
   ├─ 若无 database 且非 reindex:
   │     self.ids = createids(ids)            ← 建 indexid→id 映射
   │
   ├─ if issparse():  scoring.index()         base.py:144-145  ← 必须在 graph 前
   ├─ if indexes:     indexes.index()         base.py:148-149
   └─ if graph:       graph.index(Search(...), Ids(...), batchsimilarity)  base.py:152-153

为什么 scoring 一定在 graph 之前? 源码里专门有注释点明这条(base.py:143、base.py:191):图的构建可能要用到 scoring 的检索能力来发现文档间关系,所以 scoring 必须先 build 完、可用了,graph.index() 才能拿它去建边。同理 graph.index() 接收的三个参数——一个 Search(索引态搜索)、一个 Ids、一个 batchsimilarity 回调——都是"已经建好的检索能力",图靠它们去连边。图子系统内部见 06。

createids 的取舍: 只有当没启用 content 数据库时才需要 self.ids(base.py:1016-1028)。有 database 时,indexid↔id 的映射交给数据库表,不需要额外的内存 id 列表;没有 database 时,才用 IndexIds 在内存里维护这个映射(index/indexids.py)。

3.5 upsert:增量的"删了再加"​

upsert()(base.py:155-201)和 index() 结构几乎一样,差别在于它是追加而非重建:

upsert(documents)
   │
   ├─ 若 count()==0 → 直接退化成 index()        base.py:167-169
   │
   ├─ Transform(action=UPSERT):
   │     load() 里对每条已存在的 id 先 delete 再插   ← "更新 = 删+加"
   │     ann.append(embeddings)                     ← 追加而非重建
   │     reducer 若存在则复用(不重训)
   │
   ├─ if issparse():  scoring.upsert()
   ├─ if indexes:     indexes.upsert()
   └─ if graph:       graph.upsert(...)

三个关键区别:

  • 空索引 → 退化。 upsert 发现 count() 为 0(base.py:336-354,依次问 ann/scoring/database/ids 谁知道数量),就直接调 index(),省掉增量逻辑。
  • 更新即删除+插入。 Transform.load 在 UPSERT 动作下,对每个出现的 id 先调 self.delete 清掉旧记录,再照常插入(transform.py:181-189),并用 self.deletes 集合保证同一 id 一次动作内只删一次。
  • ANN 用 append 不用 index。 向量追加到已有索引尾部(base.py:184),reducer(若有)复用旧模型不重训。

3.6 delete:多存储的联动清理​

delete(ids)(base.py:203-258)要在所有子系统里同时把这些 id 抹掉,难点是"把用户给的 id 翻译成各索引里的整数下标"。

delete(ids)
   │
   ├─ 有 database:  database.ids(ids) 查出 (下标, id) 对
   │                database.delete(deletes)
   │
   ├─ 无 database:  在内存 self.ids 里线性查下标,置为 None
   │
   └─ 若有下标要删:
        if isdense():  ann.delete(indices)
        if issparse(): scoring.delete(indices)
        if indexes:    indexes.delete(indices)
        if graph:      graph.delete(indices)

要点:下标从哪来取决于有没有 database。有 database 时下标由数据库查出;没有时在内存 self.ids 列表里线性扫描,并把删掉的槽位置 None 而非移除(base.py:236-238)——保留位置,避免打乱后续所有 indexid 的对齐。真正的物理删除下放给各子系统的 delete(indices)。

3.7 reindex:换配置重建,但保住原始内容​

reindex()(base.py:260-290)是"我想换个向量模型/换个索引后端,但不想重新喂一遍数据"的场景。它只在启用了 content 数据库时可用——因为原文得从数据库里读回来:

# 示意,非源码:reindex 的核心
config["content"] = self.config["content"]     # 强制保留 content,别把库搞没
self.configure(config)                          # 套用新配置
if function:
    self.index(function(self.database.reindex(self.config)), True)  # reindex=True
else:
    self.index(self.database.reindex(self.config), True)

它把 self.database.reindex(...) 读出的文档流,重新喂给 index(..., reindex=True)。那个 reindex=True 正是前面反复出现的开关:它让 initindex 跳过 createdatabase(base.py:800)、让 Transform.load 跳过 database.insert(transform.py:192),从而只重建索引、不动原始内容库。

4. 深入实现:两个配角类​

4.1 Documents:子索引的临时文档队列​

子索引不能和父索引共用同一个内存流(流只能消费一次),所以 Indexes.insert 把要给子索引的文档落到磁盘临时文件排队,等所有批次到齐再统一建。这就是 Documents(index/documents.py):

  • add(batch):用 pickle 把一批文档追加序列化进一个 .docs 临时文件,累加 batch/size 计数(documents.py:52-73)。
  • __iter__:关掉写句柄,重新打开文件,按批 loadstream 逐条 yield(documents.py:35-51)——即一个"可重复遍历"的磁盘文档流。
  • __len__ 返回排队总数;close() 删临时文件并清零。

4.2 Indexes:子索引集合的转发器​

Indexes(index/indexes.py)把一组子 Embeddings 实例包成一个对象,父索引对它的调用都转发给每个子索引:

  • insert():把父索引收到的文档,按各子索引自己的 offset 过滤后塞进 Documents 队列(indexes.py:102-130)。
  • index():遍历每个子索引调它自己的 index(self.documents, ...),建完清空队列(indexes.py:143-154)。注意 checkpoint 目录按子索引名分开:f"{self.checkpoint}/{name}"。
  • __getattr__/__getitem__:允许用 embeddings.indexes["name"] 或属性方式取某个子索引(indexes.py:48-75)。

子索引里的每个元素本身就是一个完整的 Embeddings(base.py:1043-1050),共享父级的 self.models 模型缓存——递归的编排。子索引的搜索融合见 04。

4.3 Configuration:配置的存取​

Configuration(index/configuration.py)只做两件小事:load/save 索引配置,默认 JSON、兼容旧的 pickle(configuration.py:16-72)。load 会往 config 里塞一个 format 字段记录来源格式,save 据此决定写 config.json 还是 config。exists() 就靠"目录里有 config 且含 offset 字段"来判断一个索引是否已建成(base.py:530)。

4.4 save / load:插槽的对称落盘​

save(base.py:605-661)和 load(base.py:532-603)是 §2.1 那张插槽表的镜像——每个非空插槽存/取到 path 下一个固定子目录:

插槽落盘子目录
annembeddings
reducerlsa
idsids
databasedocuments
scoringscoring
indexesindexes
graphgraph

load 里的创建顺序也有讲究:先建各存储再 loadvectors——因为向量模型的创建可能依赖 scoring(见 findmodel,base.py:743-759,稀疏 scoring 也能提供模型)。若 path 是 .tar.gz/.zip 等,先经 checkarchive 解归档(base.py:921-940);若配了 cloud,先从云端拉/存(base.py:942-956)。归档与云存储的内部超出本章。

5. 巧妙之处(可借鉴的技术)​

  • "能流式接收的先建,需算出维度的后建"——用一条清晰的时序规则解决多子系统的构建依赖,而不是加一堆状态标志(base.py:114 vs base.py:133)。
  • 一次遍历、多路分发——Transform.stream 在同一个 for 循环里既筛向量化文本又抄送原文,避免把大流缓存两遍(transform.py:123-168)。
  • offset 只按"向量化数量"递增——一句 self.offset += offset 而非 += len(batch),精确保证跨 ANN/DB/scoring/graph 的下标对齐,即便部分文档不产生向量(transform.py:208)。
  • "更新 = 删 + 加",且一动作内每 id 只删一次——用一个 self.deletes 集合去重,把 upsert 语义压进 insert 流水线(transform.py:181-189)。
  • 删除时置 None 而非移除——保住内存 id 列表的位置对齐(base.py:237)。
  • reindex=True 一个开关贯穿多处——同一个布尔在 initindex、Transform.load、index 里协同跳过"重建数据库"的动作,复用整条索引流水线做"只换索引"(base.py:800、transform.py:192)。
  • 子索引递归复用 Embeddings + 共享模型缓存——子索引就是嵌套的 Embeddings,靠 self.models 共享模型省内存(base.py:1043-1050)。

6. 边界与局限​

  • 写操作不是线程安全的。 类注释明确:读线程安全,写必须外部同步(base.py:32)。
  • reindex 依赖 content 数据库。 没开 content 时 reindex 直接什么都不做(base.py:270 的 if self.database:),因为没地方读回原文。
  • 无 database 时 delete 是 O(n) 线性扫描。 每个待删 id 都要遍历整个 self.ids(base.py:232-233),大索引下较慢;有 database 时才走索引化查找。
  • 只有 text/object 字段的文档才被向量化。 纯元数据 dict 会入库但无向量;这是设计,不是 bug(transform.py:145-158)。
  • 本章不覆盖: 向量模型内部(→02)、ANN/scoring 后端实现(→03)、搜索路由与分数融合(→04)、SQL/content store(→05)、图子系统(→06)。

7. 横向对比​

在 ai-agent-reference 这一 shelf 里,txtai 的取舍是**"薄编排 + 可插拔子系统"**:Embeddings 自己几乎不含算法,靠一排工厂(ANNFactory/ScoringFactory/…)把具体后端解耦。这和"把检索逻辑硬编码进一个大类"的做法相反——代价是编排时序的复杂度,收益是稠密/稀疏/图/子索引可以任意组合。同类关切见本组 index.md 的全景图。

8. 代码地图(导航索引)​

主题文件路径符号名
编排总控类 / 组件槽位src/python/txtai/embeddings/base.pyEmbeddings.__init__
配置驱动的组件点亮src/python/txtai/embeddings/base.pyEmbeddings.configure
索引前置:先建各子系统src/python/txtai/embeddings/base.pyEmbeddings.initindex
索引主流程 / build 顺序src/python/txtai/embeddings/base.pyEmbeddings.index
增量:删了再加src/python/txtai/embeddings/base.pyEmbeddings.upsert
多存储联动删除src/python/txtai/embeddings/base.pyEmbeddings.delete
换配置重建src/python/txtai/embeddings/base.pyEmbeddings.reindex
判定谓词src/python/txtai/embeddings/base.pyissparse / isdense / isweighted / hassparse
元素计数src/python/txtai/embeddings/base.pyEmbeddings.count
插槽落盘 / 读回src/python/txtai/embeddings/base.pyEmbeddings.save / Embeddings.load
子系统工厂src/python/txtai/embeddings/base.pycreateann / createdatabase / creategraph / createscoring / createindexes / createids
一遍遍历:分发+向量化src/python/txtai/embeddings/index/transform.pyTransform.__call__ / stream / load
向量化 vs 仅收 idsrc/python/txtai/embeddings/index/transform.pyTransform.vectors / Transform.ids
输入归一化为三元组src/python/txtai/embeddings/index/stream.pyStream.__call__
动作枚举src/python/txtai/embeddings/index/action.pyAction
自动 id(序列/UUID)src/python/txtai/embeddings/index/autoid.pyAutoId.sequence / AutoId.uuid
子索引磁盘文档队列src/python/txtai/embeddings/index/documents.pyDocuments.add / __iter__
子索引集合转发器src/python/txtai/embeddings/index/indexes.pyIndexes.insert / Indexes.index
无 database 时的 id 映射src/python/txtai/embeddings/index/indexids.pyIndexIds
配置存取(JSON/pickle)src/python/txtai/embeddings/index/configuration.pyConfiguration.load / save