向量检索场景里,FAISS几乎是不二之选,它的索引结构和底层优化能把百万级向量的相似度查询压到毫秒级。但FAISS本身只是一个内存计算库,它不擅长数据持久化、事务管理和复杂条件过滤,而这些恰恰是SQLite的强项。把两者组合起来,用FAISS做向量召回、用SQLite做元数据管理和结果回表,是中小规模场景下非常实用的架构方案。本文会围绕一个图像以图搜图的实战项目,完整演示这套组合的搭建过程。

一、整体架构设计:为什么需要两个组件配合
先明确分工。FAISS负责的是向量相似度计算这一件事:把高维向量组织成高效的索引结构(比如IVF、HNSW、PQ量化),查询时快速返回最相似的k条向量的编号和距离。它返回的不是业务数据,而是一个内部ID。SQLite则负责存储这些ID对应的业务信息:图片路径、标题、上传时间、所属分类等字段。
这种分离带来的好处很直接。第一,业务数据可以随时用SQL做条件查询、分页、更新,不需要重建向量索引;第二,FAISS索引可以独立序列化到磁盘,服务重启后直接加载,不用重新计算所有向量;第三,两者的存储格式互不干扰,扩容和维护都很灵活。如果是纯FAISS方案,把图片路径硬塞进向量ID的映射关系里管理起来会非常痛苦,尤其是需要按条件过滤检索结果的时候。
一个典型的数据流是:用户上传图片后,先由特征提取模型生成向量,向量写入FAISS索引,同时把业务字段和向量ID写入SQLite。检索时,FAISS先返回Top-N的ID列表,再用一条SQL把这些ID对应的详细信息查出来返回给前端。
二、数据表设计与索引构建
SQLite的表结构要围绕向量ID来设计。vec_id是FAISS内部的索引位置,必须和业务表一一对应。下面是建表语句和插入逻辑:
CREATE TABLE IF NOT EXISTS images (
vec_id INTEGER PRIMARY KEY, -- 对应FAISS索引中的ID
file_path TEXT NOT NULL,
title TEXT,
category TEXT,
created_at TEXT DEFAULT (datetime('now', 'localtime'))
);
CREATE INDEX IF NOT EXISTS idx_images_category ON images(category);注意这里把vec_id设为主键,而不是自增ID。因为向量的编号是由FAISS的IndexIDMap分配的,业务表要被动跟随这个编号。使用IndexIDMap2包裹基础索引,可以支持自定义ID和后续删除操作:
import faiss
import numpy as np
import sqlite3
dim = 512 # 特征向量维度
base_index = faiss.IndexFlatIP(dim) # 内积索引,配合归一化向量等同余弦相似度
index = faiss.IndexIDMap2(base_index)
conn = sqlite3.connect('image_db.sqlite')
cur = conn.cursor()
# 批量写入向量与元数据
vectors = np.random.randn(1000, dim).astype('float32')
faiss.normalize_L2(vectors) # 归一化,让内积等价于余弦相似度
vec_ids = np.arange(1000)
index.add_with_ids(vectors, vec_ids)
for vid in vec_ids:
cur.execute(
"INSERT INTO images (vec_id, file_path, title, category) VALUES (?, ?, ?, ?)",
(int(vid), f"images/{vid}.jpg", f"图片{vid}", "demo")
)
conn.commit()
# 持久化FAISS索引
faiss.write_index(index, "image.index")这里有个细节值得展开:IndexFlatIP是暴力检索索引,向量量在50万以内可以接受,超过这个规模建议换成IndexIVFFlat或者IndexHNSWFlat,查询速度会有数量级的提升,代价是轻微的召回率损失。
三、检索与回表:完整的查询链路
检索流程分两步走。先在FAISS里做向量召回,拿到Top-K的ID和距离,再用SQLite回表取业务字段。代码如下:
def search(query_vector, top_k=10):
# 归一化查询向量
qv = query_vector.reshape(1, -1).astype('float32')
faiss.normalize_L2(qv)
distances, ids = index.search(qv, top_k)
valid_ids = [int(i) for i in ids[0] if i != -1]
if not valid_ids:
return []
placeholders = ','.join('?' * len(valid_ids))
cur.execute(
f"SELECT vec_id, file_path, title, category FROM images "
f"WHERE vec_id IN ({placeholders})",
valid_ids
)
rows = {r[0]: r for r in cur.fetchall()}
# 按FAISS返回的相似度顺序组装结果
results = []
for dist, vid in zip(distances[0], ids[0]):
if vid in rows:
results.append({
"vec_id": int(vid),
"score": float(dist),
"file_path": rows[vid][1],
"title": rows[vid][2]
})
return results有两个容易踩的坑要提醒。第一,FAISS在结果不足Top-K时会用-1填充ID数组,必须过滤掉再去查库,否则IN查询会出问题。第二,SQLite的IN查询不保证返回顺序,所以先用字典接住结果,再按FAISS给出的距离顺序重新组装,否则前端拿到的排序会乱掉。
回表之后还可以做二次过滤。比如只想看某个分类下的相似图片,可以在SQL里加上AND category = ?条件。为了弥补过滤导致的数量减少,通常会把召回的K放大到2到3倍再过滤,最后截取需要的前N条,这是ANN检索配合业务过滤的标准做法。
四、增量更新与索引维护
实际业务中向量会不断增加,IndexIDMap2支持追加操作,直接调用add_with_ids即可,同时记得同步写SQLite并重新持久化索引文件:
def add_image(vector, file_path, title, category):
new_id = index.ntotal # 用当前向量总数作为新ID
v = vector.reshape(1, -1).astype('float32')
faiss.normalize_L2(v)
index.add_with_ids(v, np.array([new_id]))
cur.execute(
"INSERT INTO images (vec_id, file_path, title, category) VALUES (?, ?, ?, ?)",
(new_id, file_path, title, category)
)
conn.commit()
faiss.write_index(index, "image.index")
return new_id删除操作要谨慎处理。FAISS的remove_ids只是逻辑删除,频繁增删后索引内部会产生空洞,占用空间且影响性能。推荐的做法是软删除:SQLite里加一个is_deleted标记字段,检索回表时过滤掉已删除的记录,等删除比例积累到一定阈值(比如20%)再全量重建索引。重建时遍历SQLite中所有有效记录,重新提取向量构建新索引,同时重排vec_id并更新数据库,保证两边编号一致。
最后说说持久化和启动恢复。服务启动时先检查索引文件是否存在,存在就直接faiss.read_index加载,再核对index.ntotal和SQLite记录数是否一致,不一致说明上次写入中断,需要触发一次一致性校验或重建。这套校验机制成本很低,但能避免绝大多数数据错位问题。SQLite本身启用WAL模式可以提升并发读写能力,配合这套向量方案足以支撑日增几万条数据的中型应用。