Developer's Catalog

データベース

SQLite + sqlite-vec

概要

sqlite-vecは、世界で最も広く使われている埋め込み型データベースであるSQLiteにベクトル検索機能を追加する拡張機能です。サーバーレスで軽量、ゼロ設定で動作し、モバイルアプリ、デスクトップアプリ、エッジデバイスでのベクトル検索を可能にします。

詳細

sqlite-vecはAlex Garcia氏によって開発され、SQLiteの拡張機能として提供されています。純粋なCで実装されており、追加の依存関係なしにベクトル検索機能を提供します。軽量でありながら、基本的なベクトル類似性検索とインデックス機能を備えており、ローカルアプリケーションやエッジコンピューティング環境に最適です。

sqlite-vecの主な特徴：

• SQLiteの拡張機能として動作
• ゼロ依存関係
• 軽量（数百KB）
• ブルートフォースとANNインデックス
• L2距離、コサイン類似度、内積サポート
• JSONでのベクトル保存
• SQL関数によるベクトル操作
• クロスプラットフォーム対応
• WASM対応（ブラウザでの実行）
• モバイルフレンドリー

実装の特徴

• 純粋なC実装
• SQLite仮想テーブルメカニズム
• メモリ効率的な設計
• シンプルなAPI

メリット・デメリット

メリット

• 超軽量: 最小限のフットプリントで動作
• 簡単な導入: SQLite拡張機能をロードするだけ
• ゼロ設定: 追加の設定やサーバー不要
• ポータビリティ: どこでも動作（モバイル、エッジ、ブラウザ）
• SQLite統合: 既存のSQLiteアプリに簡単に追加
• オフライン対応: ネットワーク接続不要

デメリット

• 性能制限: 大規模データセットでは遅い
• 機能制限: 基本的なベクトル検索機能のみ
• スケーラビリティ: 単一ファイルデータベースの制約
• 同時実行性: SQLiteの書き込みロック制限
• インデックスオプション: 限定的なインデックス選択肢

主要リンク

• 公式サイト
• ドキュメント
• GitHub
• デモ

書き方の例

セットアップとインストール

import sqlite3
import numpy as np
import json

# SQLite接続と拡張機能のロード
conn = sqlite3.connect(':memory:')
conn.enable_load_extension(True)

# sqlite-vec拡張機能のロード
# Linux: ./vec0.so
# macOS: ./vec0.dylib  
# Windows: ./vec0.dll
conn.load_extension('./vec0')

# ベクトルテーブルの作成
conn.execute("""
    CREATE VIRTUAL TABLE documents USING vec0(
        id INTEGER PRIMARY KEY,
        title TEXT,
        content TEXT,
        embedding FLOAT[768]
    )
""")

基本的なベクトル操作

# ドキュメントの挿入
def insert_document(conn, title, content, embedding):
    # ベクトルをJSON形式に変換
    embedding_json = json.dumps(embedding.tolist())
    
    conn.execute("""
        INSERT INTO documents(title, content, embedding)
        VALUES (?, ?, vec_f32(?))
    """, (title, content, embedding_json))
    conn.commit()

# サンプルデータの挿入
embedding = np.random.rand(768).astype(np.float32)
insert_document(
    conn,
    "SQLiteベクトル検索",
    "軽量な埋め込み型データベースでのベクトル検索",
    embedding
)

# ベクトル検索（コサイン類似度）
def vector_search_cosine(conn, query_vector, limit=10):
    query_json = json.dumps(query_vector.tolist())
    
    cursor = conn.execute("""
        SELECT 
            id,
            title,
            content,
            vec_distance_cosine(embedding, vec_f32(?)) as distance
        FROM documents
        ORDER BY distance
        LIMIT ?
    """, (query_json, limit))
    
    return cursor.fetchall()

# L2距離による検索
def vector_search_l2(conn, query_vector, limit=10):
    query_json = json.dumps(query_vector.tolist())
    
    cursor = conn.execute("""
        SELECT 
            id,
            title,
            content,
            vec_distance_l2(embedding, vec_f32(?)) as distance
        FROM documents
        ORDER BY distance
        LIMIT ?
    """, (query_json, limit))
    
    return cursor.fetchall()

インデックスの作成と管理

# ANNインデックスの作成
conn.execute("""
    CREATE INDEX idx_embedding ON documents(embedding) 
    USING vec_ann(metric='cosine', trees=10)
""")

# メタデータ付きベクトル検索
def search_with_metadata(conn, query_vector, category=None):
    query_json = json.dumps(query_vector.tolist())
    
    if category:
        cursor = conn.execute("""
            SELECT 
                d.id,
                d.title,
                d.content,
                vec_distance_cosine(d.embedding, vec_f32(?)) as distance,
                m.category,
                m.tags
            FROM documents d
            JOIN metadata m ON d.id = m.doc_id
            WHERE m.category = ?
            ORDER BY distance
            LIMIT 10
        """, (query_json, category))
    else:
        cursor = conn.execute("""
            SELECT 
                d.id,
                d.title,
                d.content,
                vec_distance_cosine(d.embedding, vec_f32(?)) as distance
            FROM documents d
            ORDER BY distance
            LIMIT 10
        """, (query_json,))
    
    return cursor.fetchall()

バッチ処理と最適化

# バッチ挿入
def batch_insert_documents(conn, documents):
    conn.execute("BEGIN TRANSACTION")
    
    for doc in documents:
        embedding_json = json.dumps(doc['embedding'].tolist())
        conn.execute("""
            INSERT INTO documents(title, content, embedding)
            VALUES (?, ?, vec_f32(?))
        """, (doc['title'], doc['content'], embedding_json))
    
    conn.execute("COMMIT")

# メモリ最適化設定
def optimize_sqlite_settings(conn):
    # ページキャッシュサイズの設定
    conn.execute("PRAGMA cache_size = 10000")
    
    # WALモードの有効化
    conn.execute("PRAGMA journal_mode = WAL")
    
    # 同期モードの調整
    conn.execute("PRAGMA synchronous = NORMAL")
    
    # メモリマップI/Oの有効化
    conn.execute("PRAGMA mmap_size = 268435456")

# ベクトル統計情報の取得
def get_vector_stats(conn):
    cursor = conn.execute("""
        SELECT 
            COUNT(*) as total_vectors,
            AVG(vec_length(embedding)) as avg_vector_length
        FROM documents
    """)
    
    return cursor.fetchone()

モバイル・エッジ向け実装

# 軽量なベクトル検索クラス
class LiteVectorSearch:
    def __init__(self, db_path):
        self.conn = sqlite3.connect(db_path)
        self.conn.enable_load_extension(True)
        self.conn.load_extension('./vec0')
        self._setup_tables()
    
    def _setup_tables(self):
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS vectors (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                embedding BLOB,
                metadata TEXT
            )
        """)
        
        self.conn.execute("""
            CREATE VIRTUAL TABLE IF NOT EXISTS vec_index USING vec0(
                embedding FLOAT[384]  -- 軽量モデル用の次元数
            )
        """)
    
    def add_vector(self, embedding, metadata=None):
        embedding_json = json.dumps(embedding.tolist())
        metadata_json = json.dumps(metadata) if metadata else None
        
        self.conn.execute("""
            INSERT INTO vectors(embedding, metadata)
            VALUES (?, ?)
        """, (embedding_json, metadata_json))
        
        self.conn.execute("""
            INSERT INTO vec_index(embedding)
            VALUES (vec_f32(?))
        """, (embedding_json,))
        
        self.conn.commit()
    
    def search(self, query_vector, k=5):
        query_json = json.dumps(query_vector.tolist())
        
        cursor = self.conn.execute("""
            SELECT 
                v.id,
                v.metadata,
                vec_distance_cosine(vi.embedding, vec_f32(?)) as distance
            FROM vec_index vi
            JOIN vectors v ON vi.rowid = v.id
            ORDER BY distance
            LIMIT ?
        """, (query_json, k))
        
        results = []
        for row in cursor:
            results.append({
                'id': row[0],
                'metadata': json.loads(row[1]) if row[1] else None,
                'distance': row[2]
            })
        
        return results