refact(settings): store on database and manage changes

Author: Daniele Briggi

.github/workflows/test.yaml

@@ -25,6 +25,12 @@ jobs:
       run: |
         pip install .[dev]
 
+    - name: Download the model
+      run: |
+        pip install huggingface_hub
+        chmod +x ./scripts/get_gguf_model.sh
+        ./scripts/get_gguf_model.sh "${{ vars.HF_MODEL_ID }}" "${{ vars.HF_GGUF_FILE }}"
+
     - name: Test
       run: |
         pytest ./tests

scripts/get_gguf_model.sh

@@ -0,0 +1,32 @@
+#!/bin/bash
+
+# Usage: ./get_gguf_model.sh <HF_MODEL_ID> <HF_GGUF_FILEname>
+# or via Environment variables: HF_MODEL_ID, HF_HF_GGUF_FILE
+
+# Example: ./get_gguf_model.sh Qwen/Qwen3-Embedding-0.6B-GGUF Qwen3-Embedding-0.6B-Q8_0.gguf
+
+set -e
+
+# Priority: CLI arguments > environment variables
+if [ $# -eq 2 ]; then
+    HF_MODEL_ID="$1"
+    HF_GGUF_FILE="$2"
+elif [ -z "$HF_MODEL_ID" ] || [ -z "$HF_GGUF_FILE" ]; then
+    echo "Error: Missing required parameters"
+    echo "Provide either:"
+    echo "  CLI: $0 <HF_MODEL_ID> <HF_GGUF_FILE>"
+    echo "  ENV: HF_MODEL_ID and HF_GGUF_FILE environment variables"
+    exit 1
+fi
+
+# Check if huggingface-hub is installed
+if ! command -v huggingface-cli &> /dev/null; then
+    echo "Error: huggingface-cli not found. Install with: pip install huggingface_hub"
+    exit 1
+fi
+
+# Download specific GGUF file
+echo "Downloading $HF_GGUF_FILE from $HF_MODEL_ID..."
+hf download "$HF_MODEL_ID" "$HF_GGUF_FILE" --local-dir "./models/$HF_MODEL_ID"
+
+echo "Downloaded to: ./models/$HF_MODEL_ID/$HF_GGUF_FILE"

src/sqlite_rag/cli.py

@@ -26,6 +26,12 @@ def __call__(self, *args, **kwds):
 cli = CLI(app)
 
 
+@app.command()
+def set(settings: Optional[str] = typer.Argument(None)):
+    """Set the model and database path"""
+    pass
+
+
 @app.command()
 def add(
     path: str = typer.Argument(..., help="File or directory path to add"),
@@ -44,11 +50,10 @@ def add(
         help="Optional metadata in JSON format to associate with the document",
         metavar="JSON",
         show_default=False,
-        prompt="Metadata (JSON format, e.g. {'author': 'John Doe', 'date': '2023-10-01'}'",
     ),
 ):
     """Add a file path to the database"""
-    rag = SQLiteRag()
+    rag = SQLiteRag.create()
     rag.add(
         path,
         recursive=recursive,
@@ -71,14 +76,14 @@ def add_text(
     ),
 ):
     """Add a text to the database"""
-    rag = SQLiteRag()
+    rag = SQLiteRag.create()
     rag.add_text(text, uri=uri, metadata=json.loads(metadata or "{}"))
 
 
 @app.command("list")
 def list_documents():
     """List all documents in the database"""
-    rag = SQLiteRag()
+    rag = SQLiteRag.create()
     documents = rag.list_documents()
 
     if not documents:
@@ -108,7 +113,7 @@ def remove(
     yes: bool = typer.Option(False, "-y", "--yes", help="Skip confirmation prompt"),
 ):
     """Remove document by path or UUID"""
-    rag = SQLiteRag()
+    rag = SQLiteRag.create()
 
     # Find the document first
     document = rag.find_document(identifier)
@@ -151,7 +156,7 @@ def rebuild(
     )
 ):
     """Rebuild embeddings and full-text index"""
-    rag = SQLiteRag()
+    rag = SQLiteRag.create()
 
     typer.echo("Rebuild process...")
 
@@ -169,14 +174,13 @@ def reset(
     yes: bool = typer.Option(False, "-y", "--yes", help="Skip confirmation prompt")
 ):
     """Reset/clear the entire database"""
-    rag = SQLiteRag()
+    rag = SQLiteRag.create()
 
     # Show warning and ask for confirmation unless -y flag is used
     if not yes:
         typer.echo(
             "WARNING: This will permanently delete all documents and data from the database!"
         )
-        typer.echo(f"Database file: {rag.settings.db_path}")
         typer.echo()
         confirm = typer.confirm("Are you sure you want to reset the entire database?")
         if not confirm:
@@ -203,7 +207,7 @@ def search(
     ),
 ):
     """Search for documents using hybrid vector + full-text search"""
-    rag = SQLiteRag()
+    rag = SQLiteRag.create()
     results = rag.search(query, top_k=limit)
 
     if not results:

src/sqlite_rag/database.py

@@ -80,7 +80,7 @@ def _create_schema(conn: sqlite3.Connection, settings: Settings):
 
         cursor.execute(
             f"""
-            SELECT vector_init('chunks', 'embedding', 'type={settings.vector_type},dimension={settings.embedding_dim}');
+            SELECT vector_init('chunks', 'embedding', 'type={settings.vector_type},dimension={settings.embedding_dim},{settings.other_vector_config}');
         """
         )
 

src/sqlite_rag/engine.py

@@ -80,7 +80,8 @@ def search(self, query: str, limit: int = 10) -> list[DocumentResult]:
         query_embedding = self.generate_embedding([Chunk(content=query)])[0].embedding
 
         # Clean up and split into words
-        query = " ".join(re.findall(r"\b\w+\b", query.lower()))
+        # '*' is used to match while typing
+        query = " ".join(re.findall(r"\b\w+\b", query.lower())) + "*"
 
         cursor.execute(
             # TODO: use vector_convert_XXX to convert the query to the correct type
@@ -137,8 +138,7 @@ def search(self, query: str, limit: int = 10) -> list[DocumentResult]:
             ;
             """,
             {
-                # '*' is used to match while typing
-                "query": query + "*",
+                "query": query,
                 "query_embedding": query_embedding,
                 "k": limit,
                 # TODO: move to settings or costants
@@ -165,3 +165,12 @@ def search(self, query: str, limit: int = 10) -> list[DocumentResult]:
             )
             for row in rows
         ]
+
+    def close(self):
+        """Close the database connection."""
+        try:
+            self._conn.execute("SELECT llm_model_free();")
+        except sqlite3.ProgrammingError:
+            # When connection is already closed the model
+            # is already freed.
+            pass

src/sqlite_rag/settings.py

@@ -1,14 +1,86 @@
+import json
+import sqlite3
+from dataclasses import asdict, dataclass, fields
+
+
+@dataclass
 class Settings:
-    def __init__(self, model_path_or_name: str, db_path: str = "sqliterag.db"):
-        self.model_path_or_name = model_path_or_name
-        self.db_path = db_path
+    model_path_or_name: str = "./models/Qwen3-Embedding-0.6B-Q8_0.gguf"
+    model_config: str = "n_ctx=384"
+
+    embedding_dim: int = 384
+    vector_type: str = "FLOAT32"
+    other_vector_config: str = "distance=cosine"  # e.g. distance=metric,other=value,...
+
+    chunk_size: int = 12000
+    # Token overlap between chunks
+    chunk_overlap: int = 1200
+
+    # Whether to quantize the vector for faster search
+    quantize_scan: bool = True
+
+
+class SettingsManager:
+    def __init__(self, connection: sqlite3.Connection):
+        self.connection = connection
+        self._ensure_table_exists()
+
+    def _ensure_table_exists(self):
+        cursor = self.connection.cursor()
+        cursor.execute(
+            """
+            CREATE TABLE IF NOT EXISTS settings (
+                id TEXT PRIMARY KEY,
+                settings JSON NOT NULL
+            );
+        """
+        )
+        self.connection.commit()
+
+    def load_settings(self) -> Settings | None:
+        cursor = self.connection.cursor()
+
+        cursor.execute("SELECT settings FROM settings LIMIT 1")
+        row = cursor.fetchone()
+
+        if not row:
+            return None
+
+        current_settings = json.loads(row[0])
+
+        # Start from defaults, update with values from db (ignore extra keys)
+        defaults = Settings()
+        valid_keys = {f.name for f in fields(Settings)}
+        filtered = {k: v for k, v in current_settings.items() if k in valid_keys}
+
+        # Use defaults as base, update with valid properties
+        settings_dict = {**asdict(defaults), **filtered}
+        return Settings(**settings_dict)
+
+    def store(self, settings: Settings):
+        cursor = self.connection.cursor()
 
-        self.embedding_dim = 384
-        self.vector_type = "FLOAT32"
+        settings_json = json.dumps(asdict(settings))
 
-        self.model_config = "n_ctx=384"  # See: https://github.com/sqliteai/sqlite-ai/blob/e172b9c9b9d76435be635d1e02c1e88b3681cc6e/src/sqlite-ai.c#L51-L57
+        # Upsert the settings
+        cursor.execute(
+            """
+            INSERT INTO settings (id, settings)
+            VALUES ('1', ?)
+            ON CONFLICT(id) DO UPDATE SET settings = excluded.settings;
+        """,
+            (settings_json,),
+        )
 
-        self.chunk_size = 256  # Maximum tokens per chunk
-        self.chunk_overlap = 32  # Token overlap between chunks
+        self.connection.commit()
+        return settings
 
-        self.quantize_scan = True  # Whether to quantize the vector for faster search
+    def has_critical_changes(
+        self, new_settings: Settings, current_settings: Settings
+    ) -> bool:
+        """Check if the new settings have critical changes compared to the current settings."""
+        return (
+            new_settings.model_path_or_name != current_settings.model_path_or_name
+            or new_settings.embedding_dim != current_settings.embedding_dim
+            or new_settings.vector_type != current_settings.vector_type
+        )