Building Private Offline RAG in Pure Python: Ollama, SQLite-Vec & Function Calling Agents

[User Query / Document Input]
              ↓
  [Local Embedding Model: nomic-embed-text / bge-m3]
              ↓
  [Single-File Serverless Vector DB: SQLite + BLOB Math]
  (Cosine Similarity via NumPy or sqlite-vec)
              ↓
  [Top-K Semantic Context Assembly + Prompt Guardrails]
              ↓
  [Local Reasoning Model: Ollama / DeepSeek-R1 / Qwen 2.5]
  (Strict Deterministic JSON Extraction via Pydantic V2)
              ↓
  [Autonomous Tool Execution / Structured Response]

# app/db/vector_store.py
import sqlite3
import struct
import numpy as np
import ollama
from typing import List, Tuple

class LocalVectorDatabase:
    def __init__(self, db_path: str = "knowledge_vault.db", embed_model: str = "nomic-embed-text"):
        self.db_path = db_path
        self.embed_model = embed_model
        self.client = ollama.Client()
        self._init_database()

    def _init_database(self):
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS document_chunks (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                source_file TEXT,
                chunk_index INTEGER,
                content TEXT,
                embedding BLOB
            )
        """)
        conn.commit()
        conn.close()

    def get_embedding(self, text: str) -> List[float]:
        response = self.client.embeddings(model=self.embed_model, prompt=text)
        return response["embedding"]

    def insert_chunk(self, source_file: str, chunk_index: int, content: str):
        vec = self.get_embedding(content)
        blob = struct.pack(f"{len(vec)}f", *vec)

        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute(
            "INSERT INTO document_chunks (source_file, chunk_index, content, embedding) VALUES (?, ?, ?, ?)",
            (source_file, chunk_index, content, blob)
        )
        conn.commit()
        conn.close()

    def semantic_search(self, query: str, top_k: int = 5) -> List[Tuple[str, float]]:
        query_vec = np.array(self.get_embedding(query), dtype=np.float32)

        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute("SELECT id, content, embedding FROM document_chunks")
        rows = cursor.fetchall()
        conn.close()

        if not rows:
            return []

        results = []
        for doc_id, content, blob in rows:
            doc_vec = np.array(struct.unpack(f"{len(query_vec)}f", blob), dtype=np.float32)
            cosine_sim = float(np.dot(query_vec, doc_vec) / (np.linalg.norm(query_vec) * np.linalg.norm(doc_vec)))
            results.append((content, cosine_sim))

        results.sort(key=lambda x: x[1], reverse=True)
        return results[:top_k]

# app/rag/chunker.py
import re
from typing import List

def chunk_markdown_document(text: str, max_chunk_size: int = 1200, overlap: int = 150) -> List[str]:
    sections = re.split(r'(
#{1,3}s+[^
]+)', text)
    chunks = []
    current_chunk = ""

    for sec in sections:
        if len(current_chunk) + len(sec) < max_chunk_size:
            current_chunk += sec
        else:
            if current_chunk.strip():
                chunks.append(current_chunk.strip())
            current_chunk = current_chunk[-overlap:] + sec if len(current_chunk) > overlap else sec

    if current_chunk.strip():
        chunks.append(current_chunk.strip())
    return chunks

# app/agents/tool_caller.py
import json
import ollama
from pydantic import BaseModel, Field

class DatabaseQueryIntent(BaseModel):
    action: str = Field(description="Query action: select, insert, or update")
    table_name: str = Field(description="Target table name")
    filters: dict = Field(default_factory=dict, description="Query filters")
    limit: int = Field(default=10, description="Max rows to return")

async def parse_user_command(user_prompt: str) -> DatabaseQueryIntent:
    client = ollama.AsyncClient()
    schema = DatabaseQueryIntent.model_json_schema()

    system_prompt = (
        "You are an autonomous command parser. "
        "Convert user intent into the following JSON schema strictly:\n"
        f"{json.dumps(schema)}\n"
        "Return ONLY raw valid JSON. No conversational preamble."
    )

    response = await client.chat(
        model="qwen2.5-coder:7b",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt}
        ],
        format="json"
    )

    raw_json = response["message"]["content"]
    return DatabaseQueryIntent(**json.loads(raw_json))

# app/main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import ollama
from app.db.vector_store import LocalVectorDatabase

app = FastAPI(title="Local AI Sovereign Knowledge API", version="1.0.0")
db = LocalVectorDatabase()
client = ollama.AsyncClient()

class AskRequest(BaseModel):
    query: str
    top_k: int = 4

class AskResponse(BaseModel):
    answer: str
    sources: list[str]

@app.post("/api/v1/ask", response_model=AskResponse)
async def ask_local_ai(req: AskRequest):
    chunks = db.semantic_search(req.query, top_k=req.top_k)
    context_text = "\n---\n".join([c[0] for c in chunks])

    prompt = (
        f"REFERENCE CONTEXT:\n{context_text}\n\n"
        f"USER QUESTION: {req.query}\n\n"
        "Answer strictly based on the reference context above. If not found, say 'Information not found.'"
    )

    res = await client.chat(
        model="deepseek-r1:8b",
        messages=[{"role": "user", "content": prompt}]
    )

    return AskResponse(
        answer=res["message"]["content"],
        sources=[f"Relevance: {c[1]:.3f}" for c in chunks]
    )