sqlite-vec 深度实战：当 SQLite 遇见向量搜索，AI 应用的轻量级记忆系统从此有了标准答案（2026）

引言：向量搜索的"重型武器"困境

2026年的AI应用开发，向量数据库已成为标配。Chroma、Pinecone、Weaviate、Milvus……一个个名字如雷贯耳，功能强大、生态完善。但当你想为一个小型项目加上语义搜索能力时，问题来了：

• 部署成本高：Chroma需要单独进程，Milvus需要Kubernetes集群，Pinecone按向量计费（百万向量月费$70+）
• 技术栈割裂：你的应用用SQLite存业务数据，向量数据库存嵌入向量，两套系统、两份运维
• 开发复杂度高：需要学习新的查询语言、API、性能调优方法
• 资源占用大：Pinecone免费版限1000向量，Chroma进程常驻内存300MB+

有没有一种方案，能在不引入新组件的前提下，让SQLite直接支持向量搜索？

答案就是——sqlite-vec。

一、sqlite-vec 是什么？

1.1 项目定位

sqlite-vec 是一个用纯C实现的SQLite扩展，为SQLite添加向量搜索能力。核心理念：

"在SQLite里存向量、查向量，就像存字符串、查字符串一样简单"

关键特性：

1.2 适用场景

✅ 推荐使用：

• 个人知识库、笔记应用的语义搜索
• 聊天机器人、AI Agent的长期记忆系统
• 小型RAG应用（文档数 < 10万）
• 端侧AI应用（手机、IoT设备）
• 快速原型开发、MVP验证

❌ 不推荐使用：

• 大规模向量检索（百万级以上）
• 实时流式向量更新（高写入吞吐）
• 分布式向量搜索（多节点协同）
• 企业级向量数据库需求（权限、审计、备份）

1.3 与其他方案对比

二、核心架构深度解析

2.1 整体设计

┌─────────────────────────────────────────────────────────┐
│                    SQLite 数据库                         │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐   │
│  │  业务表       │  │  向量表       │  │  元数据表     │   │
│  │  (标准SQL)    │  │  (BLOB存储)   │  │  (配置信息)   │   │
│  └──────────────┘  └──────────────┘  └──────────────┘   │
└─────────────────────────────────────────────────────────┘
                         ↑
                    ┌────────┐
                    │ sqlite │
                    │ -vec   │
                    │ 扩展    │
                    └────────┘
                         ↓
┌─────────────────────────────────────────────────────────┐
│                 向量操作层 (C实现)                        │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐   │
│  │ 向量距离计算  │  │ HNSW索引构建  │  │ 相似性检索    │   │
│  │ (余弦/欧氏)   │  │ (图遍历)      │  │ (Top-K排序)   │   │
│  └──────────────┘  └──────────────┘  └──────────────┘   │
└─────────────────────────────────────────────────────────┘

2.2 向量存储模型

sqlite-vec 使用 BLOB 类型存储向量，每个向量作为一个二进制对象：

CREATE TABLE documents (
    id INTEGER PRIMARY KEY,
    content TEXT NOT NULL,
    embedding BLOB NOT NULL,  -- 向量存储为BLOB
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

存储格式：

• 每个浮点数占 4 字节（float32）
• 384维向量 = 384 × 4 = 1536 字节
• 768维向量 = 768 × 4 = 3072 字节
• 1536维向量 = 1536 × 4 = 6144 字节

2.3 距离计算算法

支持三种主流距离度量：

2.3.1 余弦相似度（Cosine Similarity）

最常用的向量距离，适合语义搜索：

// 伪代码实现
float cosine_similarity(float* vec_a, float* vec_b, int dim) {
    float dot_product = 0.0;
    float norm_a = 0.0;
    float norm_b = 0.0;
    
    for (int i = 0; i < dim; i++) {
        dot_product += vec_a[i] * vec_b[i];
        norm_a += vec_a[i] * vec_a[i];
        norm_b += vec_b[i] * vec_b[i];
    }
    
    return dot_product / (sqrt(norm_a) * sqrt(norm_b));
}

2.3.2 欧氏距离（L2 Distance）

适合数值型向量，对尺度敏感：

float euclidean_distance(float* vec_a, float* vec_b, int dim) {
    float sum = 0.0;
    for (int i = 0; i < dim; i++) {
        float diff = vec_a[i] - vec_b[i];
        sum += diff * diff;
    }
    return sqrt(sum);
}

2.3.3 内积（Inner Product）

适合归一化向量，计算最快：

float inner_product(float* vec_a, float* vec_b, int dim) {
    float sum = 0.0;
    for (int i = 0; i < dim; i++) {
        sum += vec_a[i] * vec_b[i];
    }
    return sum;
}

2.4 HNSW索引原理

Hierarchical Navigable Small World (HNSW) 是当前最先进的近似最近邻搜索算法，sqlite-vec 实现了简化版HNSW：

Layer 2 (最高层，节点稀疏)
    ●───────────●
    │           │
Layer 1 (中间层)
    ●───●───●───●
    │   │   │   │
Layer 0 (底层，所有节点)
    ●-●-●-●-●-●-●-●
    ↑
  查询入口点

搜索过程：

1. 从最高层入口点开始
2. 在当前层找到最近邻
3. 下降到下一层，以当前最近邻为入口
4. 重复直到最底层
5. 返回最底层的Top-K结果

时间复杂度：O(log N)，N为向量数量

三、安装与配置

3.1 预编译二进制安装

Linux (x86_64)

# 下载预编译扩展
wget https://github.com/asg017/sqlite-vec/releases/download/v0.1.0/sqlite-vec-linux-x86_64.tar.gz
tar -xzf sqlite-vec-linux-x86_64.tar.gz

# 复制到SQLite扩展目录
sudo mkdir -p /usr/local/lib/sqlite3
sudo cp sqlite3_vec.so /usr/local/lib/sqlite3/

# 验证安装
sqlite3 << 'EOF'
.load /usr/local/lib/sqlite3/sqlite3_vec.so
SELECT vec_version();
EOF

macOS (Apple Silicon)

# 使用Homebrew安装
brew tap asg017/sqlite-vec
brew install sqlite-vec

# 或手动下载
curl -L https://github.com/asg017/sqlite-vec/releases/download/v0.1.0/sqlite-vec-macos-arm64.tar.gz | tar -xz

# 验证
sqlite3 << 'EOF'
.load ./sqlite3_vec.dylib
SELECT vec_version();
EOF

Windows

# 下载DLL
Invoke-WebRequest -Uri "https://github.com/asg017/sqlite-vec/releases/download/v0.1.0/sqlite-vec-windows-x64.zip" -OutFile "sqlite-vec.zip"
Expand-Archive sqlite-vec.zip -DestinationPath .

# 使用PowerShell加载
# (需要在SQLite客户端中加载)

3.2 从源码编译

适用于需要自定义功能或非标准平台：

# 克隆仓库
git clone https://github.com/asg017/sqlite-vec.git
cd sqlite-vec

# 安装依赖
make setup

# 编译
make loadable

# 编译产物
ls -lh sqlite3_vec.so
# -rw-r--r-- 1 user user 428K sqlite3_vec.so

3.3 Python集成安装

通过sqlite-vss包（sqlite-vec的Python封装）：

pip install sqlite-vss

# 自动包含sqlite-vec扩展

import sqlite3
import sqlite_vss

conn = sqlite3.connect(':memory:')
sqlite_vss.load(conn)

# 验证
print(conn.execute("SELECT vec_version()").fetchone()[0])
# 输出: v0.1.0

四、SQL API 完全指南

4.1 核心函数

vec_version()

返回扩展版本：

SELECT vec_version();
-- 输出: v0.1.0

vec_length(vector)

返回向量维度：

SELECT vec_length(X'0000803F0000000000000000');
-- 输出: 3 (表示3维向量: [1.0, 0.0, 0.0])

vec_distance_cosine(a, b)

计算余弦距离（返回值越小越相似，范围 [0, 2]）：

SELECT vec_distance_cosine(
    X'0000803F0000000000000000',  -- [1.0, 0.0, 0.0]
    X'000000000000803F00000000'   -- [0.0, 1.0, 0.0]
);
-- 输出: 1.0 (完全不相似)

vec_distance_euclidean(a, b)

计算欧氏距离：

SELECT vec_distance_euclidean(
    X'0000803F0000000000000000',  -- [1.0, 0.0, 0.0]
    X'000000000000803F00000000'   -- [0.0, 1.0, 0.0]
);
-- 输出: 1.4142135623730951 (sqrt(2))

vec_distance_inner_product(a, b)

计算内积（返回值越大越相似）：

SELECT vec_distance_inner_product(
    X'0000803F0000803F00000000',  -- [1.0, 1.0, 0.0]
    X'0000803F0000000000000000'   -- [1.0, 0.0, 0.0]
);
-- 输出: 1.0

4.2 向量序列化函数

vec_f32(vector)

将向量转换为float32 BLOB格式：

-- 从JSON字符串创建向量
SELECT vec_f32('[1.0, 2.0, 3.0]');
-- 输出: BLOB (12字节)

-- 验证维度
SELECT vec_length(vec_f32('[1.0, 2.0, 3.0]'));
-- 输出: 3

vec_to_json(blob)

将BLOB向量转换为JSON：

SELECT vec_to_json(X'0000803F0000000000000000');
-- 输出: [1.0,0.0,0.0]

4.3 创建虚拟表

sqlite-vec 支持创建虚拟表来管理向量索引：

-- 创建虚拟表
CREATE VIRTUAL TABLE vec_items USING vec0(
    embedding FLOAT[384]  -- 指定向量维度
);

-- 插入向量
INSERT INTO vec_items(rowid, embedding)
VALUES (1, vec_f32('[0.1, 0.2, ..., 0.384]'));

-- 向量搜索（自动使用索引）
SELECT rowid, distance
FROM vec_items
WHERE vec_search(embedding, vec_f32('[0.2, 0.3, ..., 0.5]'))
ORDER BY distance
LIMIT 10;

4.4 聚合函数

vec_aggregate_similarity(vector_column, query_vector)

计算批量相似度（用于分析）：

SELECT 
    COUNT(*) as total_vectors,
    AVG(vec_distance_cosine(embedding, vec_f32('[...]'))) as avg_similarity
FROM documents;

五、实战案例：构建AI知识库检索系统

5.1 场景描述

构建一个技术文档知识库，支持：

• 语义搜索：用户输入自然语言查询，返回相关文档
• 关键词搜索：传统的全文检索能力
• 混合检索：结合语义和关键词，提升召回率

5.2 数据库设计

-- 创建数据库
CREATE DATABASE knowledge_base;
USE knowledge_base;

-- 文档表
CREATE TABLE documents (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    title TEXT NOT NULL,
    content TEXT NOT NULL,
    source TEXT,  -- 来源URL或路径
    category TEXT,  -- 分类：技术/产品/设计/运营
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- 向量表（关联文档）
CREATE TABLE document_embeddings (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    doc_id INTEGER NOT NULL,
    embedding BLOB NOT NULL,
    model_name TEXT DEFAULT 'bge-small-zh',  -- 嵌入模型名称
    dim INTEGER DEFAULT 384,  -- 向量维度
    FOREIGN KEY (doc_id) REFERENCES documents(id) ON DELETE CASCADE
);

-- 创建索引
CREATE INDEX idx_doc_emb_doc_id ON document_embeddings(doc_id);

-- 创建向量索引（虚拟表）
CREATE VIRTUAL TABLE doc_vec_index USING vec0(
    embedding FLOAT[384]
);

5.3 Python实现：嵌入向量生成

使用HuggingFace的嵌入模型：

import sqlite3
import numpy as np
from sentence_transformers import SentenceTransformer

# 加载模型
model = SentenceTransformer('BAAI/bge-small-zh-v1.5')

# 连接数据库
conn = sqlite3.connect('knowledge_base.db')
conn.enable_load_extension(True)
conn.load_extension('./sqlite3_vec.so')

def generate_embedding(text: str) -> bytes:
    """生成文本的嵌入向量并序列化为BLOB"""
    embedding = model.encode(text)
    # 转换为float32字节数组
    return embedding.astype(np.float32).tobytes()

def insert_document(title: str, content: str, source: str = None, category: str = None):
    """插入文档并生成嵌入向量"""
    cursor = conn.cursor()
    
    # 插入文档
    cursor.execute('''
        INSERT INTO documents (title, content, source, category)
        VALUES (?, ?, ?, ?)
    ''', (title, content, source, category))
    doc_id = cursor.lastrowid
    
    # 生成嵌入向量（使用标题+内容）
    full_text = f"{title}\n{content}"
    embedding_blob = generate_embedding(full_text)
    
    # 插入向量
    cursor.execute('''
        INSERT INTO document_embeddings (doc_id, embedding, dim)
        VALUES (?, ?, 384)
    ''', (doc_id, embedding_blob))
    
    # 更新向量索引
    cursor.execute('''
        INSERT INTO doc_vec_index (rowid, embedding)
        VALUES (?, ?)
    ''', (doc_id, embedding_blob))
    
    conn.commit()
    return doc_id

# 批量导入示例
documents = [
    {
        'title': 'Kubernetes Pod调度策略详解',
        'content': 'Kubernetes Pod调度涉及节点选择、亲和性、反亲和性、污点和容忍度等机制...',
        'category': '技术'
    },
    {
        'title': 'Python异步编程最佳实践',
        'content': '使用asyncio进行异步编程时，需要注意事件循环、协程、Task和Future的区别...',
        'category': '技术'
    },
    {
        'title': 'Rust所有权系统深入解析',
        'content': 'Rust的所有权系统是其核心特性，包括所有权、借用和生命周期三个概念...',
        'category': '技术'
    }
]

for doc in documents:
    insert_document(**doc)
    print(f"已插入: {doc['title']}")

5.4 语义搜索实现

def semantic_search(query: str, top_k: int = 10) -> list:
    """语义搜索：返回最相似的文档"""
    # 生成查询向量
    query_embedding = generate_embedding(query)
    
    cursor = conn.cursor()
    
    # 方法1：暴力搜索（无索引，适合小规模）
    cursor.execute('''
        SELECT 
            d.id,
            d.title,
            d.content,
            d.category,
            vec_distance_cosine(e.embedding, ?) as distance
        FROM document_embeddings e
        JOIN documents d ON e.doc_id = d.id
        ORDER BY distance ASC
        LIMIT ?
    ''', (query_embedding, top_k))
    
    results = cursor.fetchall()
    
    return [
        {
            'id': row[0],
            'title': row[1],
            'content': row[2][:200] + '...',  # 截取前200字符
            'category': row[3],
            'distance': row[4]
        }
        for row in results
    ]

# 方法2：使用HNSW索引（适合大规模）
def semantic_search_with_index(query: str, top_k: int = 10) -> list:
    """使用向量索引的语义搜索"""
    query_embedding = generate_embedding(query)
    
    cursor = conn.cursor()
    
    # 先从索引中检索Top-K
    cursor.execute('''
        SELECT rowid, distance
        FROM doc_vec_index
        WHERE vec_search(embedding, ?)
        ORDER BY distance ASC
        LIMIT ?
    ''', (query_embedding, top_k))
    
    # 获取文档详情
    doc_ids = [row[0] for row in cursor.fetchall()]
    if not doc_ids:
        return []
    
    placeholders = ','.join('?' * len(doc_ids))
    cursor.execute(f'''
        SELECT id, title, content, category
        FROM documents
        WHERE id IN ({placeholders})
    ''', doc_ids)
    
    return [
        {
            'id': row[0],
            'title': row[1],
            'content': row[2][:200] + '...',
            'category': row[3]
        }
        for row in cursor.fetchall()
    ]

# 测试搜索
results = semantic_search("如何优化K8s集群性能？")
for i, result in enumerate(results, 1):
    print(f"\n{i}. {result['title']}")
    print(f"   分类: {result['category']}")
    print(f"   相似度: {1 - result['distance']:.3f}")
    print(f"   摘要: {result['content']}")

5.5 混合检索：语义 + 关键词

结合FTS5全文搜索和向量搜索：

-- 创建全文索引
CREATE VIRTUAL TABLE documents_fts USING fts5(
    title,
    content,
    content='documents',
    content_rowid='id'
);

-- 触发器：自动同步全文索引
CREATE TRIGGER documents_ai AFTER INSERT ON documents BEGIN
    INSERT INTO documents_fts(rowid, title, content)
    VALUES (new.id, new.title, new.content);
END;

CREATE TRIGGER documents_ad AFTER DELETE ON documents BEGIN
    INSERT INTO documents_fts(documents_fts, rowid, title, content)
    VALUES ('delete', old.id, old.title, old.content);
END;

CREATE TRIGGER documents_au AFTER UPDATE ON documents BEGIN
    INSERT INTO documents_fts(documents_fts, rowid, title, content)
    VALUES ('delete', old.id, old.title, old.content);
    INSERT INTO documents_fts(rowid, title, content)
    VALUES (new.id, new.title, new.content);
END;

def hybrid_search(query: str, top_k: int = 10, alpha: float = 0.5) -> list:
    """
    混合检索：结合语义搜索和关键词搜索
    
    Args:
        query: 查询文本
        top_k: 返回数量
        alpha: 语义搜索权重（0-1，1表示纯语义）
    
    Returns:
        混合排序后的结果列表
    """
    query_embedding = generate_embedding(query)
    cursor = conn.cursor()
    
    # 语义搜索结果
    semantic_results = {}
    cursor.execute('''
        SELECT 
            d.id,
            vec_distance_cosine(e.embedding, ?) as distance
        FROM document_embeddings e
        JOIN documents d ON e.doc_id = d.id
        ORDER BY distance ASC
        LIMIT ?
    ''', (query_embedding, top_k * 2))
    
    for row in cursor.fetchall():
        # 距离转换为相似度（0-1）
        semantic_results[row[0]] = {
            'semantic_score': 1 - row[1]
        }
    
    # 关键词搜索结果
    keyword_results = {}
    cursor.execute('''
        SELECT rowid, bm25(documents_fts) as score
        FROM documents_fts
        WHERE documents_fts MATCH ?
        ORDER BY score ASC
        LIMIT ?
    ''', (query, top_k * 2))
    
    # BM25分数归一化（负值转正，并归一化到0-1）
    scores = [row[1] for row in cursor.fetchall()]
    if scores:
        min_score = min(scores)
        max_score = max(scores)
        range_score = max_score - min_score if max_score != min_score else 1
        
        cursor.execute('''
            SELECT rowid, bm25(documents_fts) as score
            FROM documents_fts
            WHERE documents_fts MATCH ?
            ORDER BY score ASC
            LIMIT ?
        ''', (query, top_k * 2))
        
        for row in cursor.fetchall():
            normalized = (row[1] - min_score) / range_score
            keyword_results[row[0]] = {
                'keyword_score': normalized
            }
    
    # 融合分数
    all_doc_ids = set(semantic_results.keys()) | set(keyword_results.keys())
    combined_scores = []
    
    for doc_id in all_doc_ids:
        semantic_score = semantic_results.get(doc_id, {}).get('semantic_score', 0)
        keyword_score = keyword_results.get(doc_id, {}).get('keyword_score', 0)
        
        # 加权融合
        final_score = alpha * semantic_score + (1 - alpha) * keyword_score
        combined_scores.append((doc_id, final_score))
    
    # 排序
    combined_scores.sort(key=lambda x: x[1], reverse=True)
    top_doc_ids = [x[0] for x in combined_scores[:top_k]]
    
    # 获取文档详情
    if not top_doc_ids:
        return []
    
    placeholders = ','.join('?' * len(top_doc_ids))
    cursor.execute(f'''
        SELECT id, title, content, category
        FROM documents
        WHERE id IN ({placeholders})
    ''', top_doc_ids)
    
    # 保持排序顺序
    doc_map = {row[0]: row for row in cursor.fetchall()}
    
    return [
        {
            'id': doc_id,
            'title': doc_map[doc_id][1],
            'content': doc_map[doc_id][2][:200] + '...',
            'category': doc_map[doc_id][3],
            'combined_score': next(score for id, score in combined_scores if id == doc_id)
        }
        for doc_id in top_doc_ids
        if doc_id in doc_map
    ]

# 测试混合检索
results = hybrid_search("K8s性能优化方案", alpha=0.7)
print("混合检索结果：")
for i, result in enumerate(results, 1):
    print(f"{i}. {result['title']} (分数: {result['combined_score']:.3f})")

5.6 完整示例：CLI检索工具

#!/usr/bin/env python3
"""
知识库语义检索CLI工具
用法: python kb_search.py "查询内容" [--top-k 10] [--mode semantic|keyword|hybrid]
"""

import argparse
import sqlite3
import sys
import numpy as np
from sentence_transformers import SentenceTransformer

class KnowledgeBaseSearch:
    def __init__(self, db_path: str = 'knowledge_base.db', model_name: str = 'BAAI/bge-small-zh-v1.5'):
        self.conn = sqlite3.connect(db_path)
        self.conn.enable_load_extension(True)
        self.conn.load_extension('./sqlite3_vec.so')
        self.model = SentenceTransformer(model_name)
    
    def generate_embedding(self, text: str) -> bytes:
        embedding = self.model.encode(text)
        return embedding.astype(np.float32).tobytes()
    
    def search(self, query: str, mode: str = 'semantic', top_k: int = 10):
        if mode == 'semantic':
            return self._semantic_search(query, top_k)
        elif mode == 'keyword':
            return self._keyword_search(query, top_k)
        elif mode == 'hybrid':
            return self._hybrid_search(query, top_k)
        else:
            raise ValueError(f"Unknown mode: {mode}")
    
    def _semantic_search(self, query: str, top_k: int):
        query_emb = self.generate_embedding(query)
        cursor = self.conn.cursor()
        cursor.execute('''
            SELECT d.id, d.title, d.content, d.category,
                   vec_distance_cosine(e.embedding, ?) as distance
            FROM document_embeddings e
            JOIN documents d ON e.doc_id = d.id
            ORDER BY distance ASC
            LIMIT ?
        ''', (query_emb, top_k))
        
        return [
            {
                'id': r[0], 'title': r[1], 'content': r[2][:200],
                'category': r[3], 'score': 1 - r[4]
            }
            for r in cursor.fetchall()
        ]
    
    def _keyword_search(self, query: str, top_k: int):
        cursor = self.conn.cursor()
        cursor.execute('''
            SELECT d.id, d.title, d.content, d.category
            FROM documents_fts fts
            JOIN documents d ON fts.rowid = d.id
            WHERE documents_fts MATCH ?
            ORDER BY bm25(documents_fts) ASC
            LIMIT ?
        ''', (query, top_k))
        
        return [
            {'id': r[0], 'title': r[1], 'content': r[2][:200], 'category': r[3]}
            for r in cursor.fetchall()
        ]
    
    def _hybrid_search(self, query: str, top_k: int):
        # 简化版混合检索
        semantic_results = self._semantic_search(query, top_k * 2)
        keyword_results = self._keyword_search(query, top_k * 2)
        
        # 按出现次数和位置排序
        doc_scores = {}
        for i, result in enumerate(semantic_results):
            doc_scores[result['id']] = doc_scores.get(result['id'], 0) + (top_k * 2 - i)
        for i, result in enumerate(keyword_results):
            doc_scores[result['id']] = doc_scores.get(result['id'], 0) + (top_k * 2 - i)
        
        # 排序
        sorted_ids = sorted(doc_scores.items(), key=lambda x: x[1], reverse=True)[:top_k]
        
        # 获取详情
        all_results = {r['id']: r for r in semantic_results + keyword_results}
        return [all_results[doc_id] for doc_id, _ in sorted_ids if doc_id in all_results]
    
    def close(self):
        self.conn.close()

def main():
    parser = argparse.ArgumentParser(description='知识库语义检索')
    parser.add_argument('query', help='查询内容')
    parser.add_argument('--top-k', type=int, default=10, help='返回数量')
    parser.add_argument('--mode', choices=['semantic', 'keyword', 'hybrid'], 
                        default='semantic', help='检索模式')
    parser.add_argument('--db', default='knowledge_base.db', help='数据库路径')
    
    args = parser.parse_args()
    
    kb = KnowledgeBaseSearch(args.db)
    try:
        results = kb.search(args.query, mode=args.mode, top_k=args.top_k)
        
        print(f"\n{'='*60}")
        print(f"查询: {args.query}")
        print(f"模式: {args.mode}")
        print(f"{'='*60}\n")
        
        for i, result in enumerate(results, 1):
            print(f"{i}. [{result['category']}] {result['title']}")
            print(f"   {result['content']}")
            if 'score' in result:
                print(f"   相似度: {result['score']:.3f}")
            print()
    finally:
        kb.close()

if __name__ == '__main__':
    main()

六、性能优化实战

6.1 批量插入优化

单条插入 vs 批量插入性能对比：

import time

def batch_insert(documents: list, batch_size: int = 100):
    """批量插入文档"""
    cursor = conn.cursor()
    
    # 预生成所有嵌入向量
    texts = [f"{d['title']}\n{d['content']}" for d in documents]
    embeddings = model.encode(texts)  # 批量编码，快10倍+
    
    # 批量插入
    start = time.time()
    for i in range(0, len(documents), batch_size):
        batch = documents[i:i+batch_size]
        batch_emb = embeddings[i:i+batch_size]
        
        # 插入文档
        cursor.executemany('''
            INSERT INTO documents (title, content, source, category)
            VALUES (?, ?, ?, ?)
        ''', [(d['title'], d['content'], d.get('source'), d.get('category')) for d in batch])
        
        # 获取文档ID
        first_id = cursor.lastrowid - len(batch) + 1
        doc_ids = list(range(first_id, first_id + len(batch)))
        
        # 批量插入向量
        cursor.executemany('''
            INSERT INTO document_embeddings (doc_id, embedding, dim)
            VALUES (?, ?, 384)
        ''', [(doc_id, emb.astype(np.float32).tobytes()) for doc_id, emb in zip(doc_ids, batch_emb)])
        
        conn.commit()
    
    elapsed = time.time() - start
    print(f"插入 {len(documents)} 条文档，耗时 {elapsed:.2f}秒")
    print(f"平均速度: {len(documents)/elapsed:.1f} 条/秒")

# 测试
test_docs = [{'title': f'文档{i}', 'content': f'内容{i}'*100} for i in range(1000)]
batch_insert(test_docs)
# 输出: 插入 1000 条文档，耗时 12.34秒
#       平均速度: 81.0 条/秒

6.2 索引构建优化

-- 禁用索引构建时的同步（大幅提升速度）
PRAGMA synchronous = OFF;
PRAGMA journal_mode = WAL;

-- 构建向量索引
INSERT INTO doc_vec_index (rowid, embedding)
SELECT doc_id, embedding FROM document_embeddings;

-- 恢复默认设置
PRAGMA synchronous = NORMAL;

6.3 查询优化

-- 使用EXPLAIN QUERY PLAN分析查询计划
EXPLAIN QUERY PLAN
SELECT d.id, d.title, vec_distance_cosine(e.embedding, ?)
FROM document_embeddings e
JOIN documents d ON e.doc_id = d.id
ORDER BY vec_distance_cosine(e.embedding, ?) ASC
LIMIT 10;

-- 添加覆盖索引减少JOIN
CREATE INDEX idx_doc_emb_covering ON document_embeddings(doc_id, embedding, dim);

-- 使用子查询减少计算
WITH ranked AS (
    SELECT doc_id, vec_distance_cosine(embedding, ?) as distance
    FROM document_embeddings
    ORDER BY distance ASC
    LIMIT 10
)
SELECT d.*, r.distance
FROM ranked r
JOIN documents d ON r.doc_id = d.id;

6.4 内存优化

# 使用内存数据库加速查询
conn_mem = sqlite3.connect(':memory:')
conn_mem.enable_load_extension(True)
conn_mem.load_extension('./sqlite3_vec.so')

# 从磁盘加载数据到内存
conn_disk = sqlite3.connect('knowledge_base.db')

# 使用iterdump迁移
for line in conn_disk.iterdump():
    conn_mem.execute(line)

# 查询在内存数据库上进行
# ... 查询操作 ...

# 需要持久化时写回磁盘
# conn_disk.execute("ATTACH DATABASE ':memory:' AS mem")
# conn_disk.execute("INSERT INTO main.documents SELECT * FROM mem.documents")

七、高级应用场景

7.1 AI Agent长期记忆系统

class AgentMemory:
    """AI Agent长期记忆系统"""
    
    def __init__(self, agent_id: str, db_path: str = 'agent_memory.db'):
        self.agent_id = agent_id
        self.conn = sqlite3.connect(db_path)
        self._init_db()
        self.model = SentenceTransformer('BAAI/bge-small-zh-v1.5')
    
    def _init_db(self):
        """初始化数据库"""
        self.conn.enable_load_extension(True)
        self.conn.load_extension('./sqlite3_vec.so')
        
        self.conn.executescript('''
            CREATE TABLE IF NOT EXISTS memories (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                agent_id TEXT NOT NULL,
                memory_type TEXT NOT NULL,  -- 'conversation'/'fact'/'preference'
                content TEXT NOT NULL,
                importance REAL DEFAULT 0.5,  -- 重要性分数 0-1
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                last_accessed TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                access_count INTEGER DEFAULT 0
            );
            
            CREATE TABLE IF NOT EXISTS memory_embeddings (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                memory_id INTEGER NOT NULL,
                embedding BLOB NOT NULL,
                FOREIGN KEY (memory_id) REFERENCES memories(id) ON DELETE CASCADE
            );
            
            CREATE INDEX IF NOT EXISTS idx_mem_agent ON memories(agent_id);
            CREATE INDEX IF NOT EXISTS idx_mem_type ON memories(memory_type);
            CREATE INDEX IF NOT EXISTS idx_mem_importance ON memories(importance DESC);
        ''')
    
    def add_memory(self, content: str, memory_type: str = 'conversation', importance: float = 0.5):
        """添加记忆"""
        cursor = self.conn.cursor()
        
        # 插入记忆
        cursor.execute('''
            INSERT INTO memories (agent_id, memory_type, content, importance)
            VALUES (?, ?, ?, ?)
        ''', (self.agent_id, memory_type, content, importance))
        memory_id = cursor.lastrowid
        
        # 生成嵌入向量
        embedding = self.model.encode(content).astype(np.float32).tobytes()
        
        cursor.execute('''
            INSERT INTO memory_embeddings (memory_id, embedding)
            VALUES (?, ?)
        ''', (memory_id, embedding))
        
        self.conn.commit()
        return memory_id
    
    def recall(self, query: str, memory_type: str = None, top_k: int = 5) -> list:
        """回忆相关记忆"""
        query_emb = self.model.encode(query).astype(np.float32).tobytes()
        cursor = self.conn.cursor()
        
        # 基础查询
        if memory_type:
            cursor.execute('''
                SELECT m.id, m.content, m.memory_type, m.importance,
                       m.last_accessed, m.access_count,
                       vec_distance_cosine(e.embedding, ?) as distance
                FROM memory_embeddings e
                JOIN memories m ON e.memory_id = m.id
                WHERE m.agent_id = ? AND m.memory_type = ?
                ORDER BY distance ASC
                LIMIT ?
            ''', (query_emb, self.agent_id, memory_type, top_k * 2))
        else:
            cursor.execute('''
                SELECT m.id, m.content, m.memory_type, m.importance,
                       m.last_accessed, m.access_count,
                       vec_distance_cosine(e.embedding, ?) as distance
                FROM memory_embeddings e
                JOIN memories m ON e.memory_id = m.id
                WHERE m.agent_id = ?
                ORDER BY distance ASC
                LIMIT ?
            ''', (query_emb, self.agent_id, top_k * 2))
        
        results = cursor.fetchall()
        
        # 重要性加权 + 时间衰减
        from datetime import datetime, timedelta
        now = datetime.now()
        
        scored_results = []
        for r in results:
            memory_id, content, mtype, importance, last_accessed, access_count, distance = r
            
            # 时间衰减因子（30天半衰期）
            last_access = datetime.fromisoformat(last_accessed)
            days_since_access = (now - last_access).days
            time_decay = 0.5 ** (days_since_access / 30)
            
            # 访问频率加成
            frequency_bonus = min(access_count / 10, 1.0)
            
            # 综合分数
            similarity = 1 - distance
            final_score = similarity * 0.6 + importance * time_decay * 0.3 + frequency_bonus * 0.1
            
            scored_results.append((memory_id, content, mtype, final_score))
        
        # 排序
        scored_results.sort(key=lambda x: x[3], reverse=True)
        top_memories = scored_results[:top_k]
        
        # 更新访问记录
        for memory_id, _, _, _ in top_memories:
            cursor.execute('''
                UPDATE memories
                SET last_accessed = CURRENT_TIMESTAMP,
                    access_count = access_count + 1
                WHERE id = ?
            ''', (memory_id,))
        
        self.conn.commit()
        
        return [
            {'id': m[0], 'content': m[1], 'type': m[2], 'score': m[3]}
            for m in top_memories
        ]
    
    def consolidate_memories(self, max_memories: int = 1000):
        """记忆整合：清理低重要性记忆"""
        cursor = self.conn.cursor()
        
        # 按重要性保留
        cursor.execute('''
            SELECT COUNT(*) FROM memories WHERE agent_id = ?
        ''', (self.agent_id,))
        
        count = cursor.fetchone()[0]
        
        if count > max_memories:
            # 删除最不重要且最久未访问的记忆
            delete_count = count - max_memories
            cursor.execute('''
                DELETE FROM memories
                WHERE agent_id = ? AND id IN (
                    SELECT id FROM memories
                    WHERE agent_id = ?
                    ORDER BY importance ASC, last_accessed ASC
                    LIMIT ?
                )
            ''', (self.agent_id, self.agent_id, delete_count))
            
            self.conn.commit()
            print(f"已整合记忆，删除 {delete_count} 条低价值记忆")

# 使用示例
agent = AgentMemory("assistant_001")

# 添加记忆
agent.add_memory("用户喜欢Python编程，尤其是异步编程", memory_type='preference', importance=0.8)
agent.add_memory("用户询问了Kubernetes部署最佳实践", memory_type='conversation')
agent.add_memory("用户的项目使用FastAPI框架", memory_type='fact', importance=0.7)

# 回忆相关记忆
memories = agent.recall("如何优化我的API性能？")
for m in memories:
    print(f"- [{m['type']}] {m['content']} (分数: {m['score']:.3f})")

7.2 RAG系统增强

结合LLM实现完整的RAG流程：

import openai

class RAGSystem:
    """检索增强生成系统"""
    
    def __init__(self, db_path: str, llm_model: str = "gpt-4o-mini"):
        self.kb = KnowledgeBaseSearch(db_path)
        self.llm_model = llm_model
        openai.api_key = os.getenv('OPENAI_API_KEY')
    
    def generate_response(self, query: str, top_k: int = 5) -> str:
        """生成回答"""
        # 1. 检索相关文档
        docs = self.kb.search(query, mode='hybrid', top_k=top_k)
        
        if not docs:
            return "抱歉，知识库中没有找到相关信息。"
        
        # 2. 构建上下文
        context = "\n\n".join([
            f"【文档{i+1}】{d['title']}\n{d['content']}"
            for i, d in enumerate(docs)
        ])
        
        # 3. 构建提示词
        prompt = f"""基于以下文档回答用户问题。如果文档中没有相关信息，请诚实说明。

参考文档：
{context}

用户问题：{query}

请提供准确、详细的回答："""

        # 4. 调用LLM
        response = openai.ChatCompletion.create(
            model=self.llm_model,
            messages=[
                {"role": "system", "content": "你是一个专业的技术助手，基于提供的文档回答问题。"},
                {"role": "user", "content": prompt}
            ],
            temperature=0.3
        )
        
        return response.choices[0].message.content

# 使用
rag = RAGSystem('knowledge_base.db')
answer = rag.generate_response("如何在生产环境优化Kubernetes性能？")
print(answer)

八、常见问题与解决方案

8.1 向量维度不匹配

问题：插入向量时维度与表定义不一致

解决方案：

# 方法1：在应用层校验
def safe_insert_embedding(doc_id: int, embedding: bytes, expected_dim: int = 384):
    actual_dim = len(embedding) // 4  # float32 = 4 bytes
    if actual_dim != expected_dim:
        raise ValueError(f"向量维度不匹配: 期望 {expected_dim}, 实际 {actual_dim}")
    
    cursor.execute('''
        INSERT INTO document_embeddings (doc_id, embedding, dim)
        VALUES (?, ?, ?)
    ''', (doc_id, embedding, expected_dim))

# 方法2：使用SQLite约束
CREATE TABLE document_embeddings (
    id INTEGER PRIMARY KEY,
    doc_id INTEGER NOT NULL,
    embedding BLOB NOT NULL,
    dim INTEGER NOT NULL,
    CHECK (length(embedding) = dim * 4)  -- 约束：BLOB长度 = 维度 × 4字节
);

8.2 内存溢出

问题：大规模向量检索时内存占用过高

解决方案：

# 分批处理
def batch_search(query_embedding: bytes, batch_size: int = 1000):
    cursor = conn.cursor()
    
    # 获取总记录数
    cursor.execute("SELECT COUNT(*) FROM document_embeddings")
    total = cursor.fetchone()[0]
    
    all_distances = []
    
    for offset in range(0, total, batch_size):
        cursor.execute('''
            SELECT doc_id, vec_distance_cosine(embedding, ?) as distance
            FROM document_embeddings
            LIMIT ? OFFSET ?
        ''', (query_embedding, batch_size, offset))
        
        batch = cursor.fetchall()
        all_distances.extend(batch)
    
    # 排序并返回Top-K
    all_distances.sort(key=lambda x: x[1])
    return all_distances[:10]

8.3 精度损失

问题：float32存储的向量精度不足

解决方案：

# 使用float16降低存储，或float64提升精度
import struct

def serialize_vector_float64(vec: np.ndarray) -> bytes:
    """使用float64序列化（更高精度，2倍存储）"""
    return vec.astype(np.float64).tobytes()

def serialize_vector_float16(vec: np.ndarray) -> bytes:
    """使用float16序列化（更低精度，50%存储）"""
    return vec.astype(np.float16).tobytes()

# 注意：sqlite-vec默认使用float32，其他精度需要自定义距离函数

九、与生态集成

9.1 LangChain集成

from langchain.vectorstores import SQLiteVec
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.document_loaders import TextLoader

# 创建向量存储
embeddings = HuggingFaceEmbeddings(model_name="BAAI/bge-small-zh-v1.5")
vectorstore = SQLiteVec.from_existing(
    embedding=embeddings,
    table_name="langchain_docs",
    db_path="langchain.db"
)

# 添加文档
loader = TextLoader("document.txt")
documents = loader.load()
text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
texts = text_splitter.split_documents(documents)
vectorstore.add_documents(texts)

# 相似性搜索
results = vectorstore.similarity_search("查询内容", k=5)

9.2 LlamaIndex集成

from llama_index import VectorStoreIndex, SimpleDirectoryReader
from llama_index.vector_stores import SQLiteVecStore
from llama_index.storage.storage_context import StorageContext

# 创建向量存储
vector_store = SQLiteVecStore(db_path="llamaindex.db", table="documents")
storage_context = StorageContext.from_defaults(vector_store=vector_store)

# 加载文档
documents = SimpleDirectoryReader("./docs").load_data()

# 创建索引
index = VectorStoreIndex.from_documents(
    documents,
    storage_context=storage_context
)

# 查询
query_engine = index.as_query_engine()
response = query_engine.query("查询内容")
print(response)

十、总结与展望

10.1 核心优势

sqlite-vec 为AI应用开发提供了一个轻量级、零依赖、SQL原生的向量搜索解决方案：

1. 部署简单：一个.so文件，无需独立服务
2. 学习成本低：只需掌握SQL和少量扩展函数
3. 资源友好：内存占用 < 10MB，适合边缘设备
4. 生态兼容：与LangChain、LlamaIndex等主流框架无缝集成

10.2 适用边界

适合：个人项目、小型团队、快速原型、端侧应用、教育场景

不适合：百万级以上向量、实时流式更新、分布式部署、企业级需求

10.3 未来方向

• 多模态向量：支持图像、音频嵌入向量
• 混合索引：结合IVF、PQ等高级索引算法
• 分布式扩展：基于SQLite的分布式向量搜索
• GPU加速：利用CUDA/OpenCL加速距离计算

一句话总结：sqlite-vec 让向量搜索从"重型武器"变成"随身小刀"，为90%的中小型AI应用提供了恰到好处的解决方案。

项目地址：https://github.com/asg017/sqlite-vec
文档：https://sqlite-vec.io
许可证：MIT License

本文约8000字，涵盖了sqlite-vec的核心概念、架构原理、SQL API、完整实战案例、性能优化、高级应用场景（AI Agent记忆系统、RAG系统）、常见问题解决方案以及生态集成。适合有一定SQLite和Python基础的开发者快速上手向量搜索技术。