数据截至 (上游 commit 059ecec2eeac)
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() | 有没有稠密 ANN | self.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" 等 | 随机 UUID | uuid():调用对应 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) # 收尾批
这里有两条容易漏掉的暗线:
- "入库的文档集" ≠ "被向量化的文档集"。 一个只有元数据、没有
text/object字段的 dict 会被存进 database,但不会产生向量。所以下游的 offset 管理必须小心(见下)。 - 一次遍历,多路分发。 上游生成器只能消费一次,
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 下一个固定子目录:
| 插槽 | 落盘子目录 |
|---|---|
ann | embeddings |
reducer | lsa |
ids | ids |
database | documents |
scoring | scoring |
indexes | indexes |
graph | graph |
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:114vsbase.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.py | Embeddings.__init__ |
| 配置驱动的组件点亮 | src/python/txtai/embeddings/base.py | Embeddings.configure |
| 索引前置:先建各子系统 | src/python/txtai/embeddings/base.py | Embeddings.initindex |
| 索引主流程 / build 顺序 | src/python/txtai/embeddings/base.py | Embeddings.index |
| 增量:删了再加 | src/python/txtai/embeddings/base.py | Embeddings.upsert |
| 多存储联动删除 | src/python/txtai/embeddings/base.py | Embeddings.delete |
| 换配置重建 | src/python/txtai/embeddings/base.py | Embeddings.reindex |
| 判定谓词 | src/python/txtai/embeddings/base.py | issparse / isdense / isweighted / hassparse |
| 元素计数 | src/python/txtai/embeddings/base.py | Embeddings.count |
| 插槽落盘 / 读回 | src/python/txtai/embeddings/base.py | Embeddings.save / Embeddings.load |
| 子系统工厂 | src/python/txtai/embeddings/base.py | createann / createdatabase / creategraph / createscoring / createindexes / createids |
| 一遍遍历:分发+向量化 | src/python/txtai/embeddings/index/transform.py | Transform.__call__ / stream / load |
| 向量化 vs 仅收 id | src/python/txtai/embeddings/index/transform.py | Transform.vectors / Transform.ids |
| 输入归一化为三元组 | src/python/txtai/embeddings/index/stream.py | Stream.__call__ |
| 动作枚举 | src/python/txtai/embeddings/index/action.py | Action |
| 自动 id(序列/UUID) | src/python/txtai/embeddings/index/autoid.py | AutoId.sequence / AutoId.uuid |
| 子索引磁盘文档队列 | src/python/txtai/embeddings/index/documents.py | Documents.add / __iter__ |
| 子索引集合转发器 | src/python/txtai/embeddings/index/indexes.py | Indexes.insert / Indexes.index |
| 无 database 时的 id 映射 | src/python/txtai/embeddings/index/indexids.py | IndexIds |
| 配置存取(JSON/pickle) | src/python/txtai/embeddings/index/configuration.py | Configuration.load / save |