做语义搜索或者文本聚类时，最核心的中间产物就是向量嵌入。BERT这类预训练模型可以把一段文本压缩成一个固定长度的稠密向量，语义相近的文本在向量空间中距离也会更近。拿到了向量之后，下一个问题就是：这些向量存在哪里？如果数据量不大、部署环境简单，直接引入一套独立的向量数据库未免太重了。SQLite作为一个单文件数据库，配合向量检索扩展，完全可以承担起BERT向量嵌入的存储和检索任务，本文就围绕这个思路做一个完整的实战演示。
一、BERT向量嵌入是怎么来的
在使用BERT做向量化之前，需要先理解它输出的张量结构。BERT接收token id序列作为输入，输出两层结果：一层是每个token对应的隐藏状态，另一层是[CLS]位置的池化输出。做句子级别的嵌入时，通常有三种池化策略可选：直接取[CLS]向量、对所有token的隐藏状态求平均（mean pooling），或者使用sentence-transformers这类库自带的优化池化方式。实践证明mean pooling在原始BERT上的效果通常优于直接使用CLS向量，因为CLS位置的训练目标并不是直接对齐语义。
维度是存储设计的关键参数。base规模的BERT模型输出768维向量，large模型是1024维，distilbert也是768维。假如我们有一万条文本，每条文本生成一个768维的float32向量，那么原始向量数据量大约是10000乘以768再乘以4字节，约30MB。这个量级放在内存里毫无压力，放到SQLite单文件里也非常轻松。
生成向量的代码可以借助sentence-transformers库，它封装了池化逻辑和归一化操作。安装命令为pip install sentence-transformers，首次运行会自动下载模型权重。下面是一段生成向量并归一化的示例代码：
from sentence_transformers import SentenceTransformer
import numpy as np
# 加载模型，768维输出
model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')
texts = [
"SQLite是一个轻量级的嵌入式数据库",
"BERT是谷歌提出的预训练语言模型",
"向量嵌入用于衡量文本语义相似度"
]
# encode返回numpy数组，normalize_embeddings=True会做L2归一化
embeddings = model.encode(texts, normalize_embeddings=True)
print(embeddings.shape) # 输出 (3, 384)
# 归一化后内积等价于余弦相似度
sim = np.dot(embeddings[0], embeddings[1])
print(f"前两条文本的相似度: {sim:.4f}")
这里特意选了一个多语言小模型，维度是384，方便演示。如果你用中文场景较多，也可以换成shibing624/text2vec-base-chinese这类中文模型，维度为768。L2归一化是一个很重要的细节：归一化之后，向量的内积就等于余弦相似度，后续计算可以省去除以模长的步骤。
二、在SQLite中设计向量存储表
向量本质上是二进制大对象，在SQLite中对应BLOB类型。最直接的做法是把numpy数组通过tobytes()方法序列化成字节串存进BLOB字段，读取时再用np.frombuffer()还原。这种方案不依赖任何扩展，兼容性最好，缺点是SQL层面无法对向量做运算，所有相似度计算都要在应用层完成。
另一种方案是使用sqlite-vec扩展，它提供了原生的向量数据类型和虚拟表支持，可以在SQL语句中直接调用vec_distance_cosine这类函数。不过扩展需要额外编译加载，部署多一步操作。下面的建表方案采用第一种通用做法，把两种需求的字段都设计进去：
CREATE TABLE IF NOT EXISTS documents (
id INTEGER PRIMARY KEY AUTOINCREMENT,
content TEXT NOT NULL, -- 原始文本
model_name TEXT NOT NULL, -- 记录生成向量的模型，避免混用
dim INTEGER NOT NULL, -- 向量维度
embedding BLOB NOT NULL, -- 序列化后的向量
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX IF NOT EXISTS idx_docs_model ON documents(model_name, dim);
写入数据时要注意字节序和精度。numpy默认使用系统本地字节序，跨平台读取时可能出现高低位颠倒的问题，稳妥的做法是统一转为<f4（小端float32）。另外不要用float64存储，精度提升对检索效果几乎没有影响，但存储空间直接翻倍。写入和读取的完整流程如下：
import sqlite3
import numpy as np
def save_embedding(conn, content, vec, model_name):
# 统一转成小端float32再序列化
blob = np.asarray(vec, dtype='<f4').tobytes()
conn.execute(
"INSERT INTO documents (content, model_name, dim, embedding) VALUES (?, ?, ?, ?)",
(content, model_name, len(vec), blob)
)
def load_embeddings(conn, model_name=None):
if model_name:
rows = conn.execute(
"SELECT id, content, embedding FROM documents WHERE model_name = ?",
(model_name,)
).fetchall()
else:
rows = conn.execute(
"SELECT id, content, embedding FROM documents"
).fetchall()
ids, texts, vecs = [], [], []
for rid, text, blob in rows:
vec = np.frombuffer(blob, dtype='<f4')
ids.append(rid)
texts.append(text)
vecs.append(vec)
return ids, texts, np.array(vecs)
还有一个容易踩的坑：np.frombuffer返回的数组是只读的，如果后续需要修改向量（比如再做一次归一化），必须先调用copy()，否则会抛出缓冲区不可写的异常。此外，同一个表里不要混存不同模型、不同维度的向量，查询时一旦混在一起，矩阵运算会直接报形状错误，这也是建表时保留model_name和dim字段的原因。
三、实现语义检索：从暴力计算到索引优化
拿到全部向量之后，语义检索的本质就是找 Top-K 近邻。数据量在几万条以内时，最简单有效的方案是把所有向量堆成一个矩阵，一次性做矩阵乘法，然后取相似度最高的K条。numpy底层调用BLAS库，计算速度非常快，一万条384维向量的一次全量检索在普通笔记本上耗时通常只有几毫秒。
def search(query_vec, texts_matrix, texts, top_k=5):
# 向量已归一化，内积即余弦相似度
scores = texts_matrix @ query_vec
top_idx = np.argsort(-scores)[:top_k]
return [(texts[i], float(scores[i])) for i in top_idx]
# 完整检索流程
query = "轻量级数据库有哪些优势"
qvec = model.encode([query], normalize_embeddings=True)[0]
ids, texts, matrix = load_embeddings(conn, model_name='paraphrase-multilingual-MiniLM-L12-v2')
results = search(qvec, matrix, texts, top_k=3)
for text, score in results:
print(f"{score:.4f} {text}")
暴力检索的优点是结果精确、实现简单，缺点是复杂度随数据量线性增长。当数据规模涨到几十万、上百万条时，全量矩阵乘法的延迟就会明显上升，内存占用也成问题。这时可以考虑两条优化路径。
第一条路径是引入sqlite-vec扩展，使用它的虚拟表和 ANN 索引能力。加载方式是在连接数据库后执行conn.enable_load_extension(True)，再调用sqlite_vec.load(conn)。它支持把向量存入vec0虚拟表，并用MATCH语法做KNN查询，同时提供chunk_size和partition key来组织数据分片，检索时会先做粗筛再精算，速度比暴力扫描快一个数量级。
第二条路径是纯应用层的近似检索，比如用hnswlib构建HNSW图索引，把SQLite当作向量数据的持久化存储，内存中维护索引结构。启动服务时从SQLite读出全部向量重建索引，写入时双写同步。这种架构兼顾了SQLite的可靠持久化和HNSW的高速检索，是一个很常见的工程折中方案。两种路径的简单对比如下：
| 方案 | 数据量级 | 精确度 | 部署复杂度 |
|---|---|---|---|
| numpy暴力检索 | 十万级以内 | 精确结果 | 最低，无额外依赖 |
| sqlite-vec扩展 | 百万级 | 精确或近似 | 中等，需加载扩展 |
| SQLite加HNSW索引 | 千万级 | 近似结果 | 较高，需维护双写 |
四、端到端实战：构建一个本地语义搜索引擎
最后把前面的模块串起来，做一个可运行的小型语义搜索引擎。它的功能是：批量导入一批文本，向量化后写入SQLite；接收用户查询，返回语义最相近的若干条结果。为了让工程结构清晰，把数据库操作、向量化、检索三部分封装成类：
import sqlite3
import numpy as np
from sentence_transformers import SentenceTransformer
class SemanticSearchEngine:
def __init__(self, db_path, model_name='paraphrase-multilingual-MiniLM-L12-v2'):
self.conn = sqlite3.connect(db_path)
self.model = SentenceTransformer(model_name)
self.model_name = model_name
self._init_db()
self._cache = None # 向量缓存，避免每次检索都查库
def _init_db(self):
self.conn.execute("""
CREATE TABLE IF NOT EXISTS documents (
id INTEGER PRIMARY KEY AUTOINCREMENT,
content TEXT NOT NULL,
model_name TEXT NOT NULL,
dim INTEGER NOT NULL,
embedding BLOB NOT NULL
)""")
def add_documents(self, texts):
vecs = self.model.encode(texts, normalize_embeddings=True)
for text, vec in zip(texts, vecs):
blob = np.asarray(vec, dtype='<f4').tobytes()
self.conn.execute(
"INSERT INTO documents (content, model_name, dim, embedding) VALUES (?, ?, ?, ?)",
(text, self.model_name, len(vec), blob))
self.conn.commit()
self._cache = None # 数据变了，缓存失效
def _load_cache(self):
if self._cache is None:
rows = self.conn.execute(
"SELECT content, embedding FROM documents WHERE model_name = ?",
(self.model_name,)).fetchall()
texts = [r[0] for r in rows]
matrix = np.array([np.frombuffer(r[1], dtype='<f4') for r in rows])
self._cache = (texts, matrix)
return self._cache
def search(self, query, top_k=5):
texts, matrix = self._load_cache()
if not texts:
return []
qvec = self.model.encode([query], normalize_embeddings=True)[0]
scores = matrix @ qvec
top_idx = np.argsort(-scores)[:top_k]
return [(texts[i], float(scores[i])) for i in top_idx]
if __name__ == '__main__':
engine = SemanticSearchEngine('semantic.db')
engine.add_documents([
"SQLite无需独立服务进程，整个数据库就是一个文件",
"BERT通过双向Transformer编码器理解上下文语义",
"余弦相似度衡量两个向量方向的接近程度",
"今天是周六，天气不错适合出去走走"
])
for text, score in engine.search("向量相似度怎么计算"):
print(f"{score:.4f} | {text}")
运行后可以看到，查询“向量相似度怎么计算”时，得分最高的正是“余弦相似度衡量两个向量方向的接近程度”这条，而与天气相关的文本得分很低，说明语义匹配确实生效了。这套代码可以直接扩展到文档问答、FAQ自动应答、个人笔记检索等场景。
从工程角度看，这个方案的最大价值在于零部署成本：一个db文件加一个模型权重，拷贝到任何机器上就能跑，特别适合桌面工具、边缘设备或者原型验证阶段。当数据规模或者并发需求增长到SQLite的边界时，再平滑迁移到专业向量数据库也不迟，因为向量生成的部分完全不用改动，只需要替换存储和检索层。这种渐进式的架构演进思路，比一开始就搭建重型基础设施要务实得多。