💡 方案亮点
本系统实现了一个低成本、高可用性的 RAG 向量化方案，特别适合中小规模知识库：
- 🎯 SQLite BLOB 向量存储：无需专门的向量数据库，使用传统关系型数据库即可存储和检索高维向量，硬件要求极低（单机 SQLite 即可支撑 10 万+向量）
- 💰 成本优化 99%：增量更新机制通过 MD5 哈希比对，避免重复向量化，API 成本从每次全量处理的数元降低到几分钱
- 🔧 零运维负担：不依赖 Redis、Milvus、Elasticsearch 等外部服务，单文件数据库即可运行，备份迁移极其简单
- ⚡ 毫秒级检索：使用 SQL 自定义函数实现余弦相似度计算，1000 条向量检索仅需 2-5ms
- 📦 增量智能识别：自动检测文件新增/修改/删除，只处理变化的内容，第二次构建索引速度提升 95%+
- 🌐 阿里云 DashScope 集成：采用国产
text-embedding-v3模型（1024 维），单次调用仅 ¥0.0007，支持中英文高质量向量- 🚀 Docker 容器化部署：本地向量化构建 + 在线单次 API 调用，适合低内存服务器（1GB RAM 即可运行）
技术栈： Python + SQLite + DashScope Embedding + LangChain + Gunicorn
传统关键词搜索（如 SQL LIKE、正则表达式）： - ❌ 只能匹配字面相同的词 - ❌ 无法理解语义相似（如"电脑"和"计算机"） - ❌ 对拼写错误敏感 - ❌ 无法处理多语言语义
向量化（Embedding）将文本转换为高维向量： - ✅ 捕获语义信息（相似的文本向量接近） - ✅ 支持语义搜索（余弦相似度匹配） - ✅ 容忍拼写差异 - ✅ 跨语言理解（多语言模型）
核心思想： 将文本映射到向量空间，语义相似的文本在空间中距离更近。
| 方法 | 原理 | 优点 | 缺点 |
|---|---|---|---|
| TF-IDF | 词频-逆文档频率 | 简单快速 | 无语义理解 |
| Word2Vec | 神经网络词向量 | 捕捉词义 | 固定维度，无上下文 |
| GloVe | 全局词向量 | 训练快 | 同样无上下文 |
| 模型 | 维度 | 特点 | 适用场景 |
|---|---|---|---|
| BERT | 768 | 双向上下文 | 通用文本 |
| text-embedding-ada-002 (OpenAI) | 1536 | 高质量 | 商业应用 |
| text-embedding-v3 (阿里百炼) | 1024 | 中文优化、性价比高 | 本项目使用 |
| m3e (Moka Massive Mixed) | 768/1024 | 开源、中文优秀 | 本地部署 |
阿里云百炼 text-embedding-v3：
- 维度：1024
- 支持中文优化
- 价格：¥0.7/百万 token
- API 调用稳定
Markdown 笔记
↓
提取 frontmatter（元数据）
↓
切分文本（Chunk）
↓
调用 Embedding API
↓
存储到 SQLite（BLOB）
↓
运行时检索（余弦相似度）
def load_markdown_notes(notes_dir):
"""加载所有 Markdown 笔记"""
notes = []
for md_file in notes_path.rglob('*.md'):
content = md_file.read_text(encoding='utf-8')
frontmatter, body = extract_frontmatter(content)
title = frontmatter.get('title', md_file.stem)
notes.append({
'file_path': str(md_file),
'title': title,
'content': body
})
return notes
from langchain_text_splitters import RecursiveCharacterTextSplitter
def split_text(content, chunk_size=1000, chunk_overlap=200):
"""使用 LangChain 切分文本"""
splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size, # 每段 1000 字符
chunk_overlap=chunk_overlap, # 重叠 200 字符
separators=["\n\n", "\n", " ", ""]
)
chunks = splitter.split_text(content)
return chunks
为什么切分？ - API 有长度限制（通常 2000-8000 token） - 小块更容易精确匹配 - 节省 token 成本
from dashscope import TextEmbedding
def get_embedding(text):
"""调用阿里云百炼 API 获取向量"""
response = TextEmbedding.call(
model='text-embedding-v3',
input=text[:2000] # 限制长度
)
if response.status_code == 200:
embedding = response.output['embeddings'][0]['embedding']
return np.array(embedding, dtype=np.float32)
return None
cursor.execute('''
INSERT INTO note_vectors (file_path, title, text, embedding, content_hash)
VALUES (?, ?, ?, ?, ?)
''', (
note['file_path'],
note['title'],
chunk,
embedding.tobytes(), # 转为 BLOB 存储
content_hash
))
工作流程：
首次运行：
1. 读取文件内容
2. 计算 MD5 hash：hash = md5(content)
3. 存储：(file_path, content_hash, embedding)
下次运行：
1. 读取文件内容
2. 计算新 hash
3. 对比数据库中的 hash：
├─ 相同 → 跳过（内容未变）
├─ 不同 → 删除旧向量，重新生成
└─ 不存在 → 新增文件
import hashlib
# 计算文件内容的 MD5 hash
content_hash = hashlib.md5(note['content'].encode()).hexdigest()
# 查询数据库中已有的 hash
cursor.execute('SELECT file_path, content_hash FROM note_vectors GROUP BY file_path')
existing_files = {row[0]: row[1] for row in cursor.fetchall()}
# 对比判断
if note['file_path'] in existing_files:
if existing_files[note['file_path']] == content_hash:
# hash 相同，跳过
print(f"⏭️ 文件未修改，跳过")
skipped += 1
continue
else:
# hash 不同，更新
print(f"🔄 文件已修改，更新向量化")
cursor.execute('DELETE FROM note_vectors WHERE file_path = ?', (note['file_path'],))
updated += 1
else:
# 新文件
print(f"✨ 新增文件")
added += 1
CREATE TABLE note_vectors (
id INTEGER PRIMARY KEY AUTOINCREMENT,
file_path TEXT NOT NULL, -- 文件路径（唯一标识）
title TEXT NOT NULL, -- 笔记标题
text TEXT NOT NULL, -- 文本片段
embedding BLOB NOT NULL, -- 向量数据（BLOB 格式）
content_hash TEXT, -- 内容 MD5 hash（增量更新）
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
| 场景 | 全量更新 | 增量更新 |
|---|---|---|
| 100 篇笔记，1 篇修改 | 100 次 API 调用 | 1 次 API 调用 |
| 耗时 | ~5 分钟 | ~3 秒 |
| 成本 | ¥0.70 | ¥0.007 |
| 节省 | - | 99% |
def cosine_similarity(vec1, vec2):
"""计算两个向量的余弦相似度"""
dot_product = np.dot(vec1, vec2)
norm1 = np.linalg.norm(vec1)
norm2 = np.linalg.norm(vec2)
if norm1 == 0 or norm2 == 0:
return 0.0
return dot_product / (norm1 * norm2)
取值范围： [-1, 1] - 1.0：完全相同 - 0.0：无关 - -1.0：完全相反
def search_similar_chunks(question_embedding, top_k=3):
"""搜索最相似的文本片段"""
# 获取所有向量
cursor.execute('SELECT id, title, text, embedding FROM note_vectors')
rows = cursor.fetchall()
# 计算相似度
similarities = []
for row_id, title, text, embedding_blob in rows:
embedding = np.frombuffer(embedding_blob, dtype=np.float32)
similarity = cosine_similarity(question_embedding, embedding)
similarities.append({
'title': title,
'text': text,
'similarity': float(similarity)
})
# 返回 Top-K
similarities.sort(key=lambda x: x['similarity'], reverse=True)
return similarities[:top_k]
relevant_threshold = 0.3 # 低于此值认为不相关
if chunk['similarity'] >= relevant_threshold:
# 相关，加入上下文
context += f"【笔记】{chunk['title']}\n{chunk['text']}\n"
else:
# 不相关，忽略
pass
| 优化策略 | 效果 |
|---|---|
| 增量更新 | 减少 99% 调用 |
| 文本切分 | 控制在 2000 字符内 |
| 缓存机制 | 相同问题不重复调用 |
| 本地向量库 | 运行时无需批量调用 |
场景： 2000 篇笔记，平均每篇 3000 字符
首次构建：
- 总字符：6,000,000
- API 调用：6000 次（每次 1000 字符）
- 成本：6000 × ¥0.0007 = ¥4.20
日常更新（每天 1 篇修改）：
- 成本：3 × ¥0.0007 = ¥0.0021/天
- 月度：¥0.06
# 推荐配置
CHUNK_SIZE = 1000 # 每段 1000 字符
CHUNK_OVERLAP = 200 # 重叠 20%
# 切分优先级（按段落、句子、单词）
separators=["\n\n", "\n", " ", ""]
| 维度 | 精度 | 存储 | 速度 | 推荐场景 |
|---|---|---|---|---|
| 256 | 低 | 小 | 快 | 移动端 |
| 768 | 中 | 中 | 中 | 通用 |
| 1024 | 高 | 中 | 中 | 本项目 |
| 1536 | 最高 | 大 | 慢 | 商业级 |
-- 创建索引加速查询
CREATE INDEX idx_vectors_filepath ON note_vectors(file_path);
CREATE INDEX idx_vectors_hash ON note_vectors(content_hash);
答： - 本地模型（如 m3e）需要 GPU 加速 - 阿里云服务器无 GPU - API 成本低且稳定
答： - 小规模（<10000 条）：SQLite 足够 - 中规模（10000-100万）：Chroma、FAISS - 大规模（>100万）：Milvus、Pinecone
答： 1. 调整文本切分策略 2. 使用更高质量的 Embedding 模型 3. 增加 Top-K 数量 4. 调整相似度阈值
✅ 向量化：将文本转为高维向量，捕获语义
✅ 增量更新：MD5 Hash 判断文件变化，节省 99% 成本
✅ 余弦相似度：衡量文本语义相似性
✅ 文本切分：控制长度，提升匹配精度
✅ 成本优化：本地存储 + 增量更新 + 缓存机制
text-embedding-v3RecursiveCharacterTextSplitter📝 创建：2026-05-10
🔄 更新：2026-05-10
📂 分类：软件开发 / AI 技术