feat: complete BAMS syllabus database ingestion & local LLM config

Dockerfile

@@ -6,8 +6,17 @@ RUN apt-get update && apt-get install -y \
     curl \
     git \
     build-essential \
+    cmake \
     && rm -rf /var/lib/apt/lists/*
 
+# Compile native llama-server binary
+RUN git clone --depth 1 https://github.com/ggml-org/llama.cpp /opt/llama.cpp \
+    && cmake -B /opt/llama.cpp/build -S /opt/llama.cpp \
+        -DCMAKE_BUILD_TYPE=Release -DGGML_NATIVE=OFF \
+    && cmake --build /opt/llama.cpp/build --config Release -j "$(nproc)" --target llama-server \
+    && install -m 0755 /opt/llama.cpp/build/bin/llama-server /usr/local/bin/llama-server \
+    && rm -rf /opt/llama.cpp
+
 WORKDIR /code
 
 # Copy requirements and install python packages

app/main.py

@@ -11,6 +11,7 @@ if not hasattr(torch.utils._pytree, "register_constant"):
     torch.utils._pytree.register_constant = lambda cls: cls
 
 import json
+import asyncio
 import logging
 from contextlib import asynccontextmanager
 from typing import AsyncGenerator
@@ -185,40 +186,69 @@ async def _stream_and_persist(
     """
     Internal generator that streams engine events to SSE AND persists the
     final assistant message to the database once the stream completes.
+    Sends keep-alive heartbeats if the stream is idle (e.g. waiting for llama-server).
     """
     from app.schemas import Citation
+    
+    queue = asyncio.Queue()
+    done_sentinel = object()
+
+    async def producer():
+        try:
+            async for event in engine.stream_answer(domain, query, history):
+                await queue.put(event)
+        except Exception as e:
+            logger.exception("Error in stream_answer producer:")
+            await queue.put({"type": "error", "error": str(e)})
+        finally:
+            await queue.put(done_sentinel)
+
+    producer_task = asyncio.create_task(producer())
+
     full_content: list[str] = []
     citations_data: list[dict] = []
     is_grounded = False
 
-    async for event in engine.stream_answer(domain, query, history):
-        event_type = event.get("type")
-
-        if event_type == "status":
-            yield {"data": json.dumps(event)}
-
-        elif event_type == "citations":
-            citations_data = event.get("citations", [])
-            is_grounded = event.get("is_grounded", False)
-            yield {"data": json.dumps(event)}
-
-        elif event_type == "delta":
-            full_content.append(event.get("content", ""))
-            yield {"data": json.dumps(event)}
-
-        elif event_type == "done":
-            # Persist the complete assistant message before emitting done.
-            # Apply citation expansion so MongoDB stores human-readable text
-            # (the client already received the expanded tokens via the delta events).
-            content_str = "".join(full_content)
-            domain_config = engine._load_domain_config(domain)
-            content_str = engine.make_citations_readable(content_str, domain_config)
-            citations = [Citation(**c) for c in citations_data]
-            session_svc.add_assistant_message(db, session_id, content_str, citations, is_grounded)
-            yield {"data": json.dumps(event)}
-
-        elif event_type == "error":
-            yield {"data": json.dumps(event)}
+    try:
+        while True:
+            try:
+                # 5.0 second timeout to send heartbeats during queuing or prompt evaluation
+                event = await asyncio.wait_for(queue.get(), timeout=5.0)
+                if event is done_sentinel:
+                    break
+
+                event_type = event.get("type")
+                if event_type == "status":
+                    yield {"data": json.dumps(event)}
+                elif event_type == "citations":
+                    citations_data = event.get("citations", [])
+                    is_grounded = event.get("is_grounded", False)
+                    yield {"data": json.dumps(event)}
+                elif event_type == "delta":
+                    full_content.append(event.get("content", ""))
+                    yield {"data": json.dumps(event)}
+                elif event_type == "done":
+                    # Persist the complete assistant message before emitting done.
+                    content_str = "".join(full_content)
+                    if content_str:
+                        domain_config = engine._load_domain_config(domain)
+                        content_str = engine.make_citations_readable(content_str, domain_config)
+                        citations = [Citation(**c) for c in citations_data]
+                        session_svc.add_assistant_message(db, session_id, content_str, citations, is_grounded)
+                    yield {"data": json.dumps(event)}
+                elif event_type == "error":
+                    yield {"data": json.dumps(event)}
+
+            except asyncio.TimeoutError:
+                # Keep proxy connections alive
+                yield {"data": json.dumps({"type": "heartbeat"})}
+    finally:
+        if not producer_task.done():
+            producer_task.cancel()
+            try:
+                await producer_task
+            except asyncio.CancelledError:
+                pass
 
 
 # ── Helpers ────────────────────────────────────────────────────────────────────

app/services/engine.py

@@ -287,6 +287,23 @@ def make_citations_readable(text: str, domain_config: dict) -> str:
     return text
 
 
+def estimate_tokens(text: str) -> int:
+    # Heuristic for Qwen/llama token counts (~4 chars/token)
+    return max(1, len(text) // 4)
+
+
+def condense_context(chunks: list[rag.RetrievedChunk], max_context_tokens: int = 700) -> tuple[list[rag.RetrievedChunk], int]:
+    selected = []
+    running = 0
+    for chunk in chunks:
+        cost = estimate_tokens(chunk.content)
+        if running + cost > max_context_tokens:
+            break
+        selected.append(chunk)
+        running += cost
+    return selected, running
+
+
 def _build_prompt(
     domain_config: dict,
     query: str,
@@ -487,6 +504,25 @@ async def stream_answer(
             yield {"type": "done"}
             return
 
+        # Compute query embeddings for cache and search
+        embedding_fn = rag._get_embedding_fn()
+        query_embeddings = await asyncio.to_thread(embedding_fn, [query])
+
+        # ── Step 1.5: Semantic Cache Check ──────────────────────────────────────────
+        cached_val = await check_semantic_cache(query_embeddings, f"answer_cache_{domain}")
+        if cached_val:
+            logger.info("Semantic cache HIT for query: %s", query[:80])
+            engine_mode = "hybrid" if is_online else "fallback"
+            yield {"type": "status", "engine_mode": engine_mode}
+            yield {
+                "type": "citations",
+                "citations": cached_val["citations"],
+                "is_grounded": cached_val["is_grounded"]
+            }
+            yield {"type": "delta", "content": cached_val["answer"]}
+            yield {"type": "done"}
+            return
+
         # Query condensation DISABLED on CPU deployment (saves a full LLM round trip).
         # ── Step 2: Tier 1 — Pre-emptive grounding gate ──────────────────────────────
         chunks = await asyncio.to_thread(
@@ -654,7 +690,8 @@ async def stream_answer(
 
         # Build prompt (history is trimmed + deduplicated inside _build_prompt)
         trimmed_history = history[-(history_turns * 2):]
-        messages = _build_prompt(config, query, chunks, trimmed_history)
+        condensed_chunks, _ = condense_context(chunks, max_context_tokens=700)
+        messages = _build_prompt(config, query, condensed_chunks, trimmed_history)
 
         # ── Step 4: Tier 2 — Stream buffer for OUT_OF_SYLLABUS detection ────────────
         # We do NOT emit citations before we know the model is grounded.
@@ -673,7 +710,7 @@ async def stream_answer(
             if grounded:
                 yield_data = {
                     "type": "citations",
-                    "citations": [c.to_citation().model_dump() for c in chunks],
+                    "citations": [c.to_citation().model_dump() for c in condensed_chunks],
                     "is_grounded": True,
                 }
             else:
@@ -695,6 +732,7 @@ async def stream_answer(
                 max_tokens=llm_cfg.get("max_tokens"),
             )
 
+            full_response_text = []
             tokens_yielded = 0
             async for token in raw_stream:
                 tokens_yielded += 1
@@ -728,16 +766,18 @@ async def stream_answer(
                             citation_emitted = True
                             yield {
                                 "type": "citations",
-                                "citations": [c.to_citation().model_dump() for c in chunks],
+                                "citations": [c.to_citation().model_dump() for c in condensed_chunks],
                                 "is_grounded": True,
                             }
                             # Flush buffer
                             expanded = make_citations_readable(combined, config)
+                            full_response_text.append(expanded)
                             yield {"type": "delta", "content": expanded}
                     # Still accumulating buffer — don't yield yet
                 else:
                     # Post-buffer: emit tokens with citation expansion applied
                     expanded = make_citations_readable(token, config)
+                    full_response_text.append(expanded)
                     yield {"type": "delta", "content": expanded}
 
             # Edge case: stream ended while we were still buffering (very short response)
@@ -758,10 +798,11 @@ async def stream_answer(
                         citation_emitted = True
                         yield {
                             "type": "citations",
-                            "citations": [c.to_citation().model_dump() for c in chunks],
+                            "citations": [c.to_citation().model_dump() for c in condensed_chunks],
                             "is_grounded": True,
                         }
                     expanded = make_citations_readable(combined, config)
+                    full_response_text.append(expanded)
                     yield {"type": "delta", "content": expanded}
 
             # Empty stream fallback: if we got absolutely no tokens from the LLM,
@@ -773,13 +814,13 @@ async def stream_answer(
                     citation_emitted = True
                     yield {
                         "type": "citations",
-                        "citations": [c.to_citation().model_dump() for c in chunks],
+                        "citations": [c.to_citation().model_dump() for c in condensed_chunks],
                         "is_grounded": True,
                     }
                 
-                # Format fallback content from chunks
+                # Format fallback content from condensed_chunks
                 context_parts = []
-                for chunk in chunks:
+                for chunk in condensed_chunks:
                     context_parts.append(chunk.content)
                 context_text = "\n\n".join(context_parts).strip()
                 
@@ -793,6 +834,28 @@ async def stream_answer(
                         yield {"type": "delta", "content": word + " "}
                         await asyncio.sleep(0.01)
                     yield {"type": "delta", "content": "\n\n"}
+
+                # Cache the fallback response
+                await write_semantic_cache(
+                    query,
+                    query_embeddings,
+                    intro + context_text,
+                    [c.to_citation().model_dump() for c in condensed_chunks],
+                    True,
+                    f"answer_cache_{domain}"
+                )
+            elif is_grounded_confirmed:
+                # Cache the generated LLM response
+                final_text = "".join(full_response_text)
+                await write_semantic_cache(
+                    query,
+                    query_embeddings,
+                    final_text,
+                    [c.to_citation().model_dump() for c in condensed_chunks],
+                    True,
+                    f"answer_cache_{domain}"
+                )
+
         async for event in _stream_with_buffer():
             yield event
 
@@ -802,4 +865,75 @@ async def stream_answer(
         yield {"type": "error", "error": str(e)}
 
 
+ANSWER_CACHE_COLLECTION = "answer_cache"
+SIMILARITY_THRESHOLD = 0.92
+
+
+async def check_semantic_cache(query_embeddings: list[list[float]], collection_name: str = ANSWER_CACHE_COLLECTION) -> dict | None:
+    try:
+        from app.services.rag import get_or_create_collection
+        collection = get_or_create_collection(collection_name)
+        
+        # Query the cache collection using the query embeddings
+        results = await asyncio.to_thread(
+            collection.query,
+            query_embeddings=query_embeddings,
+            n_results=1,
+            include=["documents", "metadatas", "distances"]
+        )
+        if not results["distances"] or not results["distances"][0]:
+            return None
+        
+        distance = results["distances"][0][0]
+        similarity = 1.0 - distance
+        
+        if similarity >= SIMILARITY_THRESHOLD:
+            meta = results["metadatas"][0][0]
+            citations_raw = meta.get("citations", "[]")
+            try:
+                citations = json.loads(citations_raw)
+            except Exception:
+                citations = []
+            return {
+                "answer": meta.get("answer", ""),
+                "citations": citations,
+                "is_grounded": meta.get("is_grounded", True)
+            }
+    except Exception as e:
+        logger.warning("Semantic cache lookup failed: %s", e)
+    return None
+
+
+async def write_semantic_cache(
+    query: str,
+    query_embeddings: list[list[float]],
+    answer: str,
+    citations: list[dict],
+    is_grounded: bool = True,
+    collection_name: str = ANSWER_CACHE_COLLECTION
+) -> None:
+    try:
+        from app.services.rag import get_or_create_collection
+        collection = get_or_create_collection(collection_name)
+        
+        doc_id = hashlib.sha256(query.encode("utf-8")).hexdigest()
+        
+        # Write/Update the cache record
+        await asyncio.to_thread(
+            collection.upsert,
+            ids=[doc_id],
+            embeddings=query_embeddings,
+            documents=[query],
+            metadatas=[{
+                "answer": answer,
+                "citations": json.dumps(citations),
+                "is_grounded": is_grounded,
+                "query": query
+            }]
+        )
+        logger.info("Saved response to semantic cache for query: %s", query[:50])
+    except Exception as e:
+        logger.warning("Failed to write to semantic cache: %s", e)
+
+
 

app/services/llm.py

@@ -68,6 +68,7 @@ async def stream_completion(
             stop=["<|im_end|>", "<|im_start|>", "--- STUDENT QUESTION ---", "--- SYLLABUS CONTEXT ---"],
             # Frequency penalty prevents the model from getting stuck in repetition loops (e.g. repeating emojis)
             frequency_penalty=0.1,
+            extra_body={"cache_prompt": True},
         )
         async for chunk in stream:
             delta = chunk.choices[0].delta.content

parallel_10_results.json

@@ -0,0 +1,132 @@
+[
+  {
+    "user_id": 1,
+    "question": "What is Rasa Dhatu?",
+    "duration_s": 88.5884416103363,
+    "first_token_time_s": 84.02139806747437,
+    "words": 305,
+    "grounded": true,
+    "citations": 3,
+    "heartbeats": 0,
+    "cache_hit": false,
+    "status": "Success",
+    "error": ""
+  },
+  {
+    "user_id": 7,
+    "question": "What is Shukra Dhatu?",
+    "duration_s": 117.67143845558167,
+    "first_token_time_s": null,
+    "words": 0,
+    "grounded": false,
+    "citations": 0,
+    "heartbeats": 0,
+    "cache_hit": false,
+    "status": "Failed (Too Short)",
+    "error": ""
+  },
+  {
+    "user_id": 6,
+    "question": "What is Rasa Dhatu?",
+    "duration_s": 118.67596745491028,
+    "first_token_time_s": null,
+    "words": 0,
+    "grounded": false,
+    "citations": 0,
+    "heartbeats": 0,
+    "cache_hit": false,
+    "status": "Failed (Too Short)",
+    "error": ""
+  },
+  {
+    "user_id": 4,
+    "question": "Explain Kedara-Kulya Nyaya.",
+    "duration_s": 120.67894673347473,
+    "first_token_time_s": null,
+    "words": 0,
+    "grounded": false,
+    "citations": 0,
+    "heartbeats": 0,
+    "cache_hit": false,
+    "status": "Failed (Too Short)",
+    "error": ""
+  },
+  {
+    "user_id": 2,
+    "question": "What is Shukra Dhatu?",
+    "duration_s": 122.68142056465149,
+    "first_token_time_s": null,
+    "words": 0,
+    "grounded": false,
+    "citations": 0,
+    "heartbeats": 0,
+    "cache_hit": false,
+    "status": "Failed (Too Short)",
+    "error": ""
+  },
+  {
+    "user_id": 8,
+    "question": "Explain Ksheera-Dadhi Nyaya.",
+    "duration_s": 116.67163467407227,
+    "first_token_time_s": null,
+    "words": 0,
+    "grounded": false,
+    "citations": 0,
+    "heartbeats": 0,
+    "cache_hit": false,
+    "status": "Failed (Too Short)",
+    "error": ""
+  },
+  {
+    "user_id": 9,
+    "question": "What is the function of Rasa Dhatu?",
+    "duration_s": 115.66793322563171,
+    "first_token_time_s": null,
+    "words": 0,
+    "grounded": false,
+    "citations": 0,
+    "heartbeats": 0,
+    "cache_hit": false,
+    "status": "Failed (Too Short)",
+    "error": ""
+  },
+  {
+    "user_id": 5,
+    "question": "Explain Khale-Kapotanya Nyaya.",
+    "duration_s": 119.67767238616943,
+    "first_token_time_s": null,
+    "words": 0,
+    "grounded": false,
+    "citations": 0,
+    "heartbeats": 0,
+    "cache_hit": false,
+    "status": "Failed (Too Short)",
+    "error": ""
+  },
+  {
+    "user_id": 10,
+    "question": "What are the theories of tissue nutrition in Ayurveda?",
+    "duration_s": 114.66833090782166,
+    "first_token_time_s": null,
+    "words": 0,
+    "grounded": false,
+    "citations": 0,
+    "heartbeats": 0,
+    "cache_hit": false,
+    "status": "Failed (Too Short)",
+    "error": ""
+  },
+  {
+    "user_id": 3,
+    "question": "Explain Ksheera-Dadhi Nyaya.",
+    "duration_s": 121.68023824691772,
+    "first_token_time_s": null,
+    "words": 0,
+    "grounded": false,
+    "citations": 0,
+    "heartbeats": 0,
+    "cache_hit": false,
+    "status": "Failed (Too Short)",
+    "error": ""
+  }
+]

start.sh

@@ -18,17 +18,18 @@ else
     echo "Model already exists at $MODEL_PATH"
 fi
 
-# 2. Start llama-cpp-python server in the background (runs on CPU, port 8001)
-echo "Starting local llama-cpp-python server on port 8001..."
-python -m llama_cpp.server \
+# 2. Start native llama-server in the background (runs on CPU, port 8001)
+echo "Starting native llama-server on port 8001..."
+llama-server \
     --model "$MODEL_PATH" \
     --port 8001 \
     --host 127.0.0.1 \
-    --n_ctx 4096 \
-    --n_threads 2 \
-    --n_threads_batch 2 \
-    --n_batch 256 \
-    --chat_format chatml &
+    --ctx-size 6144 \
+    --parallel 3 \
+    --cont-batching \
+    --threads 2 \
+    --threads-batch 2 &
+
 
 # 3. Wait for the local llama.cpp server to be ready
 echo "Waiting for llama.cpp server to initialize..."
@@ -45,14 +46,6 @@ export MODEL_NAME="qwen2.5-3b-instruct-q4_k_m.gguf"
 export LLM_MODEL_NAME="qwen2.5-3b-instruct-q4_k_m.gguf"
 export LLM_MAX_TOKENS=2048
 
-
-# 4.5. Extract pre-built ChromaDB index from zip if present
-if [ -f "chroma_store.zip" ]; then
-    echo "Extracting pre-built ChromaDB index from zip..."
-    python -c "import zipfile; zipfile.ZipFile('chroma_store.zip').extractall('chroma_store')"
-    echo "Extraction complete."
-fi
-
 # 5. Run ingest only if the collection is empty or new files exist
 echo "Checking knowledge base..."
 CHROMA_COUNT=$(python -c "