Feat/phase3 rag pipeline

.github/workflows/ci.yml

@@ -2,9 +2,9 @@ name: CI
 
 on:
   push:
-    branches: [master]
+    branches: [master, main, feat/*]
   pull_request:
-    branches: [master]
+    branches: [master, main, feat/*]
 
 permissions:
   contents: read
@@ -17,23 +17,34 @@ jobs:
     timeout-minutes: 10
 
     steps:
+      # ── Checkout backend ──
       - uses: actions/checkout@v4
+        with:
+          path: chatbot-plugin
+
+      # ── Checkout SDK (sibling directory so ../chatbot-plugin-sdk resolves) ──
+      - uses: actions/checkout@v4
+        with:
+          repository: Teng91/chatbot-plugin-sdk
+          path: chatbot-plugin-sdk
 
       - uses: astral-sh/setup-uv@v5
         with:
           python-version: "3.11"
 
       - name: Install dependencies
+        working-directory: ./chatbot-plugin
         run: uv sync --all-groups
 
       - name: Run tests
+        working-directory: ./chatbot-plugin
         run: uv run pytest src/tests/ -v --tb=short --cov=chatbot_plugin --cov-report=xml:coverage.xml --junitxml=pytest.xml
 
       - name: Post coverage comment
         if: github.event_name == 'pull_request'
         uses: MishaKav/pytest-coverage-comment@v1.6.0
         with:
-          pytest-xml-coverage-path: ./coverage.xml
-          junitxml-path: ./pytest.xml
+          pytest-xml-coverage-path: ./chatbot-plugin/coverage.xml
+          junitxml-path: ./chatbot-plugin/pytest.xml
           title: Test Coverage
           github-token: ${{ secrets.GITHUB_TOKEN }}

.gitignore

@@ -14,3 +14,5 @@ htmlcov/
 src/DB_COMPARISON.md
 .pytest_cache/
 providers.toml
+coverage.xml
+pytest.xml

alembic/versions/002_toolbox_schema.py

@@ -41,6 +41,9 @@ def upgrade() -> None:
     op.create_index("idx_articles_url", "articles", ["url"])
     op.create_unique_constraint("articles_url_key", "articles", ["url"])
 
+    # pgvector: enable extension
+    op.execute("CREATE EXTENSION IF NOT EXISTS vector")
+
     # Create article_chunks table
     op.create_table(
         "article_chunks",
@@ -49,24 +52,32 @@ def upgrade() -> None:
         sa.Column("chunk_index", sa.Integer(), nullable=False),
         sa.Column("content", sa.Text(), nullable=False),
         sa.Column("dense_vector", sa.Text(), nullable=True),
-        sa.Column("sparse_vector", JSONB, nullable=True),
+        sa.Column("sparse_vector", sa.Text(), nullable=True),
         sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
     )
 
-    # pgvector: enable extension + cast dense_vector to proper vector type
-    op.execute("CREATE EXTENSION IF NOT EXISTS vector")
+    # Cast dense_vector to proper vector type
     op.execute("ALTER TABLE article_chunks ALTER COLUMN dense_vector TYPE vector(1024) USING dense_vector::vector")
+    # Cast sparse_vector to proper sparsevec type
+    op.execute("ALTER TABLE article_chunks ALTER COLUMN sparse_vector TYPE sparsevec(250002) USING sparse_vector::sparsevec")
 
     op.create_unique_constraint("uq_article_chunk_idx", "article_chunks", ["article_id", "chunk_index"])
     op.create_index("idx_chunks_article_id", "article_chunks", ["article_id"])
 
-    # HNSW index for dense similarity search (required by spec)
+    # HNSW index for dense similarity search
     op.execute(
         "CREATE INDEX hnsw_chunks_dense ON article_chunks USING hnsw (dense_vector vector_cosine_ops)"
     )
+    # Simple index for sparse vector (sparsevec HNSW not available in pgvector 0.8.0)
+    op.execute(
+        "CREATE INDEX idx_chunks_sparse ON article_chunks (sparse_vector)"
+    )
 
 
 def downgrade() -> None:
+    op.execute("DROP INDEX IF EXISTS idx_chunks_sparse")
+    op.execute("DROP INDEX IF EXISTS hnsw_chunks_dense")
+
     op.drop_table("article_chunks")
     op.drop_table("articles")
 

pyproject.toml

@@ -1,26 +1,24 @@
 [project]
 name = "chatbot-plugin"
-version = "0.1.0"
-description = "Vector storage toolbox for scrape-and-analyze — chunk storage and retrieval"
+version = "0.3.0"
+description = "OpenAI-compatible RAG chat backend"
 requires-python = ">=3.11"
 dependencies = [
     "fastapi",
     "uvicorn[standard]",
     "pydantic>=2.0",
-    "pydantic-settings>=2.0",
-    "structlog",
-    "sqlalchemy[asyncio]>=2.0",
-    "asyncpg",
-    "alembic",
-    "pgvector",
+    "chatbot-plugin-sdk",
 ]
 
+[tool.uv.sources]
+chatbot-plugin-sdk = { path = "../chatbot-plugin-sdk", editable = true }
+
 [dependency-groups]
 dev = [
+    "httpx",
     "pytest",
     "pytest-cov",
     "pytest-asyncio",
-    "httpx",
 ]
 
 [build-system]

scripts/fix_sparsevec.py

@@ -0,0 +1,86 @@
+#!/usr/bin/env python3
+"""Fix existing databases: sparse_vector column type jsonb -> sparsevec.
+
+Migration 002 originally created sparse_vector as JSONB (a bug in the migration).
+This script safely converts existing data to the correct sparsevec(250002) type.
+"""
+
+import asyncio
+
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import create_async_engine
+from pgvector import SparseVector
+from chatbot_plugin.config import settings
+
+
+async def fix_sparse_vector() -> None:
+    engine = create_async_engine(settings.database_url)
+
+    async with engine.begin() as conn:
+        # Check current type
+        result = await conn.execute(text(
+            "SELECT data_type "
+            "FROM information_schema.columns "
+            "WHERE table_name = 'article_chunks' AND column_name = 'sparse_vector'"
+        ))
+        row = result.fetchone()
+        if row is None:
+            print("Column sparse_vector not found — nothing to fix.")
+            return
+
+        current_type = row[0]
+
+        if current_type == "sparsevec":
+            print(f"sparse_vector is already {current_type} — nothing to fix.")
+            return
+
+        print(f"Current type: {current_type}. Converting to sparsevec(250002)...")
+
+        # Ensure extension is available
+        await conn.execute(text("CREATE EXTENSION IF NOT EXISTS vector"))
+
+        # Drop any existing index on sparse_vector first
+        await conn.execute(text("DROP INDEX IF EXISTS idx_chunks_sparse"))
+
+        # Add a temporary sparsevec column
+        await conn.execute(text(
+            "ALTER TABLE article_chunks "
+            "ADD COLUMN IF NOT EXISTS sparse_vector_new sparsevec(250002)"
+        ))
+
+        # Transfer data: read jsonb dicts and write as SparseVector
+        result = await conn.execute(text(
+            "SELECT id, sparse_vector FROM article_chunks WHERE sparse_vector IS NOT NULL"
+        ))
+        rows = result.fetchall()
+        print(f"Converting {len(rows)} rows...")
+
+        for chunk_id, sparse_data in rows:
+            if sparse_data is None:
+                continue
+            # PostgreSQL sparsevec uses 1-based indices in text format:
+            # '{1:val,2:val}/dims'.
+            items = ",".join(f"{int(k) + 1}:{float(v)}" for k, v in sorted(sparse_data.items()))
+            sparse_text = f"{{{items}}}/{settings.sparse_dimension}"
+            await conn.execute(
+                text("UPDATE article_chunks SET sparse_vector_new = text(:sv)::sparsevec(250002) WHERE id = :id"),
+                {"sv": sparse_text, "id": chunk_id},
+            )
+
+        # Drop old column and rename
+        await conn.execute(text("ALTER TABLE article_chunks DROP COLUMN sparse_vector"))
+        await conn.execute(text("ALTER TABLE article_chunks RENAME COLUMN sparse_vector_new TO sparse_vector"))
+
+        # Note: pgvector 0.8.0 does not provide sparsevec HNSW indexing.
+        # A simple index on the column is sufficient for now.
+        await conn.execute(text(
+            "CREATE INDEX idx_chunks_sparse ON article_chunks (sparse_vector)"
+        ))
+
+        print("Done. sparse_vector is now sparsevec(250002).")
+
+    await engine.dispose()
+
+
+if __name__ == "__main__":
+    asyncio.run(fix_sparse_vector())

specs/integration.md

@@ -21,7 +21,7 @@ uvicorn chatbot_plugin.main:app --reload
 CHATBOT_DATABASE_URL=postgresql+asyncpg://... uvicorn chatbot_plugin.main:app --host 0.0.0.0 --port 8000
 ```
 
-## How External Services Send Data
+## How External Services Send Data, Search, and Chat
 
 Scrape-and-analyze (or any service) POSTs to the toolbox:
 
@@ -42,8 +42,32 @@ async def send_chunks():
         assert resp.status_code == 201
 ```
 
+```python
+async def search():
+    async with httpx.AsyncClient() as client:
+        resp = await client.post(
+            "http://toolbox:8000/tools/search",
+            json={"query": "What is RAG?", "top_k": 10},
+        )
+        assert resp.status_code == 200
+        print(resp.json()["chunks"])
+```
+
+```python
+async def chat():
+    async with httpx.AsyncClient() as client:
+        resp = await client.post(
+            "http://toolbox:8000/tools/chat",
+            json={"message": "Explain RAG in simple terms"},
+        )
+        assert resp.status_code == 200
+        print(resp.json()["reply"])
+```
+
 This exposes:
 - `POST /tools/chunks` — store article + chunks
+- `POST /tools/search` — hybrid dense + sparse search
+- `POST /tools/chat` — chat with RAG context
 
 ## Environment Variables
 
@@ -56,6 +80,14 @@ CHATBOT_DATABASE_URL=postgresql+asyncpg://postgres:postgres@localhost:5432/chatb
 # Embedding model config (must match what scrape-and-analyze uses)
 CHATBOT_EMBEDDING_MODEL=BAAI/bge-m3
 CHATBOT_EMBEDDING_DIMENSION=1024
+CHATBOT_SPARSE_DIMENSION=250002
+CHATBOT_RRF_K=60
+CHATBOT_SEARCH_CANDIDATES=50
+CHATBOT_MAX_CONTEXT_CHUNKS=10
+
+# LLM config (optional — only needed for /tools/chat)
+CHATBOT_LLM_API_KEY=sk-ant-api-...
+CHATBOT_LLM_MODEL=claude-sonnet-4-6-20250514
 ```
 
 ## Database
@@ -84,7 +116,7 @@ CREATE TABLE article_chunks (
     chunk_index   INT NOT NULL,
     content       TEXT NOT NULL,
     dense_vector  vector(1024),
-    sparse_vector JSONB,
+    sparse_vector sparsevec(250002),
     created_at    TIMESTAMPTZ DEFAULT NOW(),
     UNIQUE(article_id, chunk_index)
 );

specs/rag-pipeline.md

@@ -93,8 +93,108 @@ CREATE INDEX hnsw_chunks_dense ON article_chunks
 2. **Vector dimension validation**: Reject chunks where `len(dense_vector)` != `CHATBOT_EMBEDDING_DIMENSION` (default 1024).
 3. **No embedding generation here**: Scrape-and-analyze (or a separate embedding service) generates vectors before calling this API.
 
+## Phase 3: Query Embedding + Hybrid Retrieval + Chat
+
+### Architecture
+
+```
+User Query
+    │
+    ▼
+Embed Query (BGE-M3 in-process CPU)
+    │ → query_dense_vec (1024-dim float)
+    │ → query_sparse_vec (250002-dim sparsevec)
+    │
+    ▼
+Parallel Hybrid Search
+    │
+    ├─ Dense:  cosine similarity via pgvector HNSW
+    │          ORDER BY dense_vector <=> query_vec LIMIT candidates
+    │
+    └─ Sparse: max inner product via pgvector sparsevec
+               ORDER BY sparse_vector <#> query_sparse LIMIT candidates
+    │
+    ▼
+RRF Fusion (k=60)
+    score_d(i) = 1 / (k + rank_d(i))
+    score_s(i) = 1 / (k + rank_s(i))
+    final(i)   = score_d(i) + score_s(i)
+    │
+    ▼
+Top-K Chunks (default 10)
+    │ Deduplicate, assemble into context
+    │
+    ▼
+Chat Response (Claude API via HTTP)
+    System prompt + context chunks + user message
+    Each context segment annotated [source: article_title]
+```
+
+### Embedding Model
+
+- **Model**: `BAAI/bge-m3` (loaded via `FlagEmbedding` library)
+- **Loading**: Singleton, loaded once at startup in FastAPI lifespan
+- **CPU inference**: ~300-500ms per query, ~2GB RAM footprint
+- **Output**: dense (1024-dim) + sparse (250002-dim token weights)
+- Sparse dimension: `250002` (BGE-M3 tokenizer vocab size)
+
+### Hybrid Search Algorithm
+
+1. Embed query text via `embed_query(text)` → `(dense_vec, sparse_vec)`
+2. **Dense candidates**: `SELECT chunk_id, article_id, ... FROM article_chunks ORDER BY dense_vector <=> :dense LIMIT :candidates`
+3. **Sparse candidates**: `SELECT chunk_id, article_id, ... FROM article_chunks ORDER BY sparse_vector <#> :sparse LIMIT :candidates`
+4. **RRF fusion**: Combine the two candidate lists
+   - Rank each chunk in each list (1-indexed)
+   - `score = 1/(rrf_k + rank_dense) + 1/(rrf_k + rank_sparse)`
+   - If a chunk only appears in one list, its missing rank contributes 0
+5. Sort by final score descending
+6. Return top `top_k` chunks
+
+### Chat Service
+
+1. Call search service with `message` as query → top chunks
+2. Deduplicate chunks by `article_id`
+3. Build context string:
+   ```
+   [source: Article Title]
+   Chunk content...
+
+   [source: Another Title]
+   Another chunk...
+   ```
+4. Send to LLM (fallback chain):
+   1. **Anthropic Claude** — if `CHATBOT_LLM_API_KEY` is set. On any error, fall through.
+   2. **Google Gemini** — if `CHATBOT_GEMINI_API_KEY` is set. On any error (rate limit, unreachable, etc.), fall through.
+   3. **Raw context fallback** — if all LLM calls fail or no key is configured, return the assembled context + question directly so the caller can still inspect retrieved sources.
+   - System: "You are a helpful research assistant..."
+   - User: context + "\n\nQuestion: {message}"
+5. Return: `{reply, articles_used, chunks}`
+
+### Sparse Vector Storage
+
+- PostgreSQL type: `sparsevec(250002)` via pgvector
+- SQLAlchemy type: `pgvector.sqlalchemy.SPARSEVEC(250002)`
+- Index: no HNSW for sparsevec (pgvector 0.7+ uses btree or ivfflat)
+  - For now create a standard btree index on `(sparse_vector)`
+- Input translation: API receives `dict[str, float]` → `SparseVector(dict, dim=250002)` → DB
+- Output translation: DB → `SparseVector.from_text()` → consumed by search
+
 ## Configuration
 
 | Variable | Default | Description |
 |----------|---------|-------------|
 | `CHATBOT_EMBEDDING_DIMENSION` | `1024` | Dense vector dimension (matches embedding model) |
+| `CHATBOT_SPARSE_DIMENSION` | `250002` | Sparse vector dimension (BGE-M3 vocab size) |
+| `CHATBOT_RRF_K` | `60` | RRF constant k |
+| `CHATBOT_SEARCH_CANDIDATES` | `50` | Number of candidates from each sub-search before RRF |
+| `CHATBOT_MAX_CONTEXT_CHUNKS` | `10` | Max chunks to include in chat context |
+| `CHATBOT_LLM_API_KEY` | `""` | Anthropic API key (optional) |
+| `CHATBOT_LLM_MODEL` | `"claude-sonnet-4-6-20250514"` | Claude model name |
+| `CHATBOT_GEMINI_API_KEY` | `""` | Google Gemini API key (fallback) |
+| `CHATBOT_GEMINI_MODEL` | `"gemini-2.0-flash"` | Gemini model name |
+
+## New Dependencies
+
+- `FlagEmbedding` — BGE-M3 embedding (dense + sparse)
+- `torch` — PyTorch backend (CPU mode)
+- `anthropic` — Claude API client