构建一个既支持精确条件过滤又能理解自然语言语义的检索服务,通常不需要在单个数据库里硬塞所有能力。将SQLite作为本地结构化数据存储,同时把文本向量交给Pinecone这样的云向量服务管理,可以形成一种轻量且清晰的混合检索架构。SQLite负责保存文档标题、来源、更新时间、分片正文以及状态字段,Pinecone则只保存向量和指向SQLite记录的ID。查询时先通过语义相似度找到候选ID,再回到SQLite获取完整信息,或者反向先按SQL条件筛选出候选集合,再进入向量空间排序。下面这个实战项目会覆盖从建表、写入、向量化、upsert到最终混合查询的完整链路。

为什么把SQLite和Pinecone组合起来用
SQLite的强项是事务、索引、轻量部署和复杂SQL过滤。对于一个知识库类应用,文档元信息天然是结构化的,比如分类、标签、创建时间、状态等,这些数据放在SQLite中可以用一条SELECT语句快速筛选。但是如果用户输入的是“最近有什么关于数据库性能优化的内容”,SQLite的LIKE匹配很难理解“数据库性能优化”与“MySQL慢查询调优”是同义表达。这正是Pinecone擅长的部分:把文本转成768维或1024维向量后,根据余弦相似度返回语义相近的若干条记录。
单独使用Pinecone也能实现语义搜索,但Pinecone不是用来做事务型数据管理的,它不擅长频繁更新字段、维护复杂关系或者执行多表连接。如果要把标题、作者、审核状态和正文都存在向量服务的metadata里,当这些字段频繁变化时,更新成本高,metadata体积也会膨胀。让SQLite管理结构化事实,让Pinecone只维护向量索引和最小化的过滤条件,职责边界明确,后续扩展也会更舒服。例如增加一个新的SQL字段只需要ALTER TABLE,而不用重建向量索引。
这种组合特别适合单机工具、边缘设备应用、原型验证以及数据量在百万级以内但语义检索需求明确的项目。SQLite文件可以直接复制备份,Pinecone作为云服务管理向量副本,即便本地SQLite文件丢失,也可以通过重新嵌入原文恢复向量索引;反过来,如果Pinecone索引损坏,结构化数据仍然完整,只需重新写入向量即可。
SQLite数据层设计与写入实现
先设计两张核心表。documents表保存文档级别的信息,chunks表保存切分后的文本片段。之所以把分片单独建表,是因为一个长文档可能被切成多个chunk,每个chunk对应一条Pinecone向量记录。documents表和chunks表通过doc_id关联,chunks表还需要一个全局唯一的chunk_id,这个ID会同时写入Pinecone的向量记录中,作为回查SQLite的桥梁。建表SQL如下:
CREATE TABLE documents (
doc_id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT NOT NULL,
category TEXT NOT NULL DEFAULT 'general',
source TEXT,
created_at TEXT DEFAULT (datetime('now', 'localtime'))
);
CREATE TABLE chunks (
chunk_id INTEGER PRIMARY KEY AUTOINCREMENT,
doc_id INTEGER NOT NULL,
chunk_index INTEGER NOT NULL,
content TEXT NOT NULL,
token_count INTEGER DEFAULT 0,
FOREIGN KEY (doc_id) REFERENCES documents(doc_id) ON DELETE CASCADE
);
CREATE INDEX idx_chunks_doc_id ON chunks(doc_id);
CREATE INDEX idx_documents_category ON documents(category);
这里给doc_id和category建索引,是因为混合查询经常先做SQL条件过滤。chunk_id使用SQLite自增整数主键,在Pinecone中可以直接把向量ID设置为字符串形式,例如chunk_123。为了让回查更直观,写入向量时会使用统一的ID前缀,避免纯数字ID与其他来源数据冲突。表结构中还保留了chunk_index,表示当前片段在原文档中的顺序,后续展示上下文时可以方便地取前一帧和后一帧。
Python侧写入SQLite时建议使用事务批量提交。逐条INSERT虽然简单,但写入几万条数据会非常慢。把每个文档的所有chunk放入一个事务,并复用同一个预编译语句,性能可以提升一个数量级。下面是一个写入chunks的示例函数:
import sqlite3
from typing import List, Dict
def insert_chunks(chunks: List[Dict], db_path: str = "knowledge.db") -> None:
conn = sqlite3.connect(db_path)
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA synchronous=NORMAL")
try:
cur = conn.cursor()
cur.execute(
"INSERT INTO documents (title, category, source) VALUES (?, ?, ?)",
("向量数据库选型指南", "database", "internal_wiki")
)
doc_id = cur.lastrowid
cur.executemany(
"INSERT INTO chunks (doc_id, chunk_index, content, token_count) "
"VALUES (?, ?, ?, ?)",
[
(doc_id, item["index"], item["content"], item.get("token_count", 0))
for item in chunks
]
)
conn.commit()
finally:
conn.close()
上面代码先写入documents表,拿到自增doc_id后再批量写入chunks。WAL模式可以让读写并发更好,适合这种一边写入SQLite一边异步同步Pinecone的场景。synchronous设为NORMAL能在保证基本安全的前提下减少磁盘同步次数。需要注意,如果写入过程中抛出异常,事务