feat: implement Search Architecture V3 (Hybrid search with Qwen query transformation, batch embeddings, batch Qdrant searches, and optimized CPU reranking)

.gitignore

@@ -93,9 +93,10 @@ update_colab.py
 web_expert.py
 webui_tool_tipitaka*.py
 
-# Old architecture docs
+# architecture docs
 Dhamma-LM*.md
 TIPITAKA_WEB_ARCHITECTURE*.md
+Tipitaka-Web-Application-Tech-Stack*.md
 
 # Notebooks
 *.ipynb

webapp/tipitaka-api/app/config.py

@@ -36,6 +36,8 @@ class Settings(BaseSettings):
     QDRANT_PATH: str = ""
     SNAPSHOT_DIR: str = ""
     QDRANT_URL: str | None = None
+    ST_EMBED_MODEL: str = "jinaai/jina-embeddings-v5-text-small-retrieval"
+    QDRANT_COLLECTION: str = "tipitaka_chunks"
     
     def __init__(self, **values):
         super().__init__(**values)

webapp/tipitaka-api/app/database/sqlite_db.py

@@ -104,6 +104,71 @@ class SQLiteDB:
     def is_in_memory(self) -> bool:
         return self._mem_conn is not None
 
+    def ensure_query_cache_table(self) -> None:
+        """Create query_cache table on disk if not exists."""
+        with sqlite3.connect(self.db_path) as conn:
+            conn.execute("""
+                CREATE TABLE IF NOT EXISTS query_cache (
+                    cache_key   TEXT PRIMARY KEY,
+                    result_json TEXT NOT NULL,
+                    hit_count   INTEGER DEFAULT 1,
+                    created_at  DATETIME DEFAULT CURRENT_TIMESTAMP,
+                    last_used   DATETIME DEFAULT CURRENT_TIMESTAMP
+                )
+            """)
+            conn.execute("CREATE INDEX IF NOT EXISTS idx_query_cache_last_used ON query_cache(last_used)")
+            conn.commit()
+
+    def get_query_cache(self, key: str) -> dict | None:
+        """Retrieve query cache from disk and increment hit count."""
+        import json
+        try:
+            with self.get_disk_connection() as conn:
+                row = conn.execute(
+                    "SELECT result_json FROM query_cache WHERE cache_key = ?", (key,)
+                ).fetchone()
+                if row:
+                    conn.execute(
+                        """UPDATE query_cache
+                           SET hit_count = hit_count + 1, last_used = datetime('now','localtime')
+                           WHERE cache_key = ?""", (key,)
+                    )
+                    return json.loads(row["result_json"])
+        except Exception:
+            pass
+        return None
+
+    def set_query_cache(self, key: str, result: dict) -> None:
+        """Store query cache to disk and evict oldest if too large."""
+        import json
+        try:
+            self.evict_query_cache(max_entries=10000)
+            with self.get_disk_connection() as conn:
+                conn.execute(
+                    """INSERT INTO query_cache (cache_key, result_json) VALUES (?, ?)
+                       ON CONFLICT(cache_key) DO UPDATE SET
+                           result_json = excluded.result_json,
+                           last_used = datetime('now','localtime')""",
+                    (key, json.dumps(result, ensure_ascii=False))
+                )
+        except Exception:
+            pass
+
+    def evict_query_cache(self, max_entries: int = 10000) -> None:
+        """Evict oldest cache entries on disk if count exceeds max_entries."""
+        try:
+            with self.get_disk_connection() as conn:
+                count = conn.execute("SELECT COUNT(*) FROM query_cache").fetchone()[0]
+                if count > max_entries:
+                    conn.execute("""
+                        DELETE FROM query_cache WHERE cache_key IN (
+                            SELECT cache_key FROM query_cache
+                            ORDER BY last_used ASC LIMIT ?
+                        )
+                    """, (count - max_entries,))
+        except Exception:
+            pass
+
 
 # Singleton database instance
 _db: SQLiteDB | None = None
@@ -117,4 +182,5 @@ def get_db() -> SQLiteDB:
         _db = SQLiteDB(settings.DATABASE_PATH)
         _db.ensure_search_log_table()
         _db.ensure_reference_tables()
+        _db.ensure_query_cache_table()
     return _db

webapp/tipitaka-api/app/routers/search.py

@@ -2,12 +2,18 @@ from fastapi import APIRouter, Depends, Query
 from typing import List
 from app.schemas import SearchResponse
 from app.services.search_service import SearchService
+from app.services.query_transform_service import QueryTransformService
 from app.database.sqlite_db import get_db, SQLiteDB
+from app.routers.ai import get_llm_service
 
 router = APIRouter(prefix="/search", tags=["Search"])
 
-def get_search_service(db: SQLiteDB = Depends(get_db)) -> SearchService:
-    return SearchService(db)
+def get_search_service(
+    db: SQLiteDB = Depends(get_db),
+    llm_service = Depends(get_llm_service)
+) -> SearchService:
+    qts = QueryTransformService(db)
+    return SearchService(db, rag_service=llm_service.rag_service, query_transform_service=qts)
 
 @router.get("/suggestions", response_model=List[str])
 def suggestions(
@@ -19,10 +25,10 @@ def suggestions(
     return service.get_suggestions(q, limit)
 
 @router.get("", response_model=SearchResponse)
-def search(
+async def search(
     q: str = Query(..., min_length=1),
     limit: int = Query(50, ge=1, le=100),
     offset: int = Query(0, ge=0),
     service: SearchService = Depends(get_search_service)
 ):
-    return service.search(q, limit, offset)
+    return await service.search(q, limit, offset)

webapp/tipitaka-api/app/services/onnx_reranker.py

@@ -2,6 +2,7 @@ import onnxruntime as ort
 from transformers import AutoTokenizer
 import numpy as np
 import os
+import multiprocessing
 
 class ONNXReranker:
     def __init__(self, model_dir, device="cpu"):
@@ -9,8 +10,6 @@ class ONNXReranker:
         self.tokenizer = AutoTokenizer.from_pretrained(model_dir, fix_mistral_regex=True)
         
         # Determine the ONNX file path
-        # Jina v2 repo puts it in onnx/model.onnx or model_quantized.onnx
-        # We'll check both.
         model_path = os.path.join(model_dir, "onnx", "model_int8.onnx")
         if not os.path.exists(model_path):
              model_path = os.path.join(model_dir, "onnx", "model_quantized.onnx")
@@ -18,12 +17,23 @@ class ONNXReranker:
              model_path = os.path.join(model_dir, "onnx", "model.onnx")
         
         if not os.path.exists(model_path):
-            raise FileNotFoundError(f"ONNX model not found in {model_dir}")
+             raise FileNotFoundError(f"ONNX model not found in {model_dir}")
             
         print(f"Loading ONNX model from {model_path}...")
-        self.session = ort.InferenceSession(model_path, providers=['CPUExecutionProvider'])
         
-    def predict(self, pairs, batch_size=16, **kwargs):
+        # Optimize ONNX Runtime session for CPU execution
+        opts = ort.SessionOptions()
+        opts.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL
+        
+        # Set thread count to physical CPU cores (cap at 4 to prevent thread thrashing)
+        cores = multiprocessing.cpu_count()
+        opts.intra_op_num_threads = min(4, cores)
+        opts.inter_op_num_threads = 1
+        opts.execution_mode = ort.ExecutionMode.ORT_SEQUENTIAL
+        
+        self.session = ort.InferenceSession(model_path, sess_options=opts, providers=['CPUExecutionProvider'])
+        
+    def predict(self, pairs, batch_size=4, max_length=256, **kwargs):
         """
         Predict scores for (query, passage) pairs.
         pairs: list of [query, passage]
@@ -34,23 +44,19 @@ class ONNXReranker:
             queries = [p[0] for p in batch]
             passages = [p[1] for p in batch]
             
-            # Jina v2 usually expects the pair as single input or separated by EOS
-            # We'll use the tokenizer's __call__ which handles pairs correctly
+            # Use max_length=256 for fast CPU reranking of snippet relevance
             inputs = self.tokenizer(
                 queries,
                 passages,
                 padding=True,
                 truncation=True,
-                max_length=1024,
+                max_length=max_length,
                 return_tensors="np"
             )
             
-            # ONNX model usually expects input_ids, attention_mask
-            # Some also expect token_type_ids
             onnx_inputs = {k: v for k, v in inputs.items() if k in [n.name for n in self.session.get_inputs()]}
             
             outputs = self.session.run(None, onnx_inputs)
-            # The output is usually logits. Jina v2 has a single output dimension.
             logits = outputs[0]
             
             # Apply sigmoid to get scores in [0, 1]
@@ -58,3 +64,4 @@ class ONNXReranker:
             all_scores.extend(scores.flatten().tolist())
             
         return all_scores
+

webapp/tipitaka-api/app/services/query_transform_service.py

@@ -0,0 +1,133 @@
+import os
+import json
+import re
+import unicodedata
+from typing import Optional
+
+FALLBACK = lambda q: {"fts_queries": [q], "vector_queries": [q]}
+
+SYSTEM_PROMPT = """คุณคือผู้เชี่ยวชาญพระไตรปิฎกฉบับมหาจุฬา 45 เล่ม
+ตอบด้วย JSON เท่านั้น รูปแบบ: {"fts_queries":[...],"vector_queries":[...]}
+แต่ละช่องมีได้สูงสุด 2 รายการ ห้ามมีข้อความอื่นนอกจาก JSON"""
+
+class QueryTransformService:
+    def __init__(self, db):
+        self.db = db
+        
+        # ตั้งค่าพารามิเตอร์และ Runtime จากสภาพแวดล้อม
+        self.runtime = os.getenv("QWEN_RUNTIME", "transformers")
+        self.ollama_url = os.getenv("QWEN_OLLAMA_URL", "http://localhost:11434")
+        self.ollama_model = os.getenv("QWEN_OLLAMA_MODEL", "qwen3.5:0.8b")
+        self.model_id = os.getenv("QWEN_MODEL_ID", "Qwen/Qwen3.5-0.8B")
+        
+        # ตัวแปรสำหรับ Lazy Loading เมื่อใช้ transformers ในเครือข่าย HF Space
+        self._tokenizer = None
+        self._model = None
+
+    def _normalize_key(self, query: str) -> str:
+        # 1. ปรับรูปอักขระภาษาไทย (NFC) และแปลงเป็นตัวพิมพ์เล็ก
+        q = unicodedata.normalize("NFC", query.strip().lower())
+        # 2. ลบเครื่องหมายวรรคตอนและสัญลักษณ์พิเศษยกเว้นเว้นวรรค
+        q = re.sub(r"[^\w\s\u0e00-\u0e7f]", "", q)
+        # 3. บีบช่องว่างที่ติดกันและลบช่องว่างหัวท้าย
+        return re.sub(r"\s+", " ", q).strip()
+
+    async def transform(self, query: str) -> dict:
+        key = self._normalize_key(query)
+
+        # 1. ตรวจ cache ก่อนเสมอเพื่อความเร็วสูงสูด
+        cached = self.db.get_query_cache(key)
+        if cached:
+            return cached
+
+        # 2. เรียกใช้โมเดลตาม runtime หาก cache miss
+        try:
+            if self.runtime == "ollama":
+                result = await self._call_ollama(query)
+            else:
+                result = await self._call_transformers(query)
+            
+            # บันทึกผลลัพธ์ใหม่ลงใน cache
+            self.db.set_query_cache(key, result)
+            return result
+        except Exception:
+            return FALLBACK(query)
+
+    async def _call_ollama(self, query: str) -> dict:
+        """Local: ยิง Ollama HTTP API"""
+        import httpx
+        url = f"{self.ollama_url}/api/chat"
+        payload = {
+            "model": self.ollama_model,  # qwen3.5:0.8b
+            "messages": [
+                {"role": "system", "content": SYSTEM_PROMPT},
+                {"role": "user", "content": query}
+            ],
+            "stream": False,
+            "options": {"num_predict": 80, "temperature": 0}
+        }
+        async with httpx.AsyncClient(timeout=10.0) as client:
+            resp = await client.post(url, json=payload)
+            raw = resp.json()["message"]["content"].strip()
+        return self._parse_json(raw, query)
+
+    async def _call_transformers(self, query: str) -> dict:
+        """HF Space: Lazy Loading ในแอปเพื่อไม่ให้ Startup บล็อคและเกิด Timeout"""
+        import torch
+        from transformers import AutoTokenizer, AutoModelForCausalLM
+
+        # โหลดโมเดลในหน่วยความจำเฉพาะเมื่อถูกใช้งานจริงครั้งแรก (Lazy Loading)
+        if self._model is None or self._tokenizer is None:
+            self._tokenizer = AutoTokenizer.from_pretrained(
+                self.model_id, 
+                trust_remote_code=True
+            )
+            # โหลดด้วย float32 เพื่อความปลอดภัยและรันได้เสถียรบน CPU ของ HF Space
+            self._model = AutoModelForCausalLM.from_pretrained(
+                self.model_id,
+                torch_dtype=torch.float32, 
+                device_map="cpu",
+                trust_remote_code=True
+            )
+            self._model.eval()
+
+        messages = [
+            {"role": "system", "content": SYSTEM_PROMPT},
+            {"role": "user", "content": query}
+        ]
+        text = self._tokenizer.apply_chat_template(
+            messages, tokenize=False, add_generation_prompt=True
+        )
+        inputs = self._tokenizer(text, return_tensors="pt")
+
+        with torch.no_grad():
+            outputs = self._model.generate(
+                **inputs,
+                max_new_tokens=80,
+                do_sample=False,
+                temperature=None,
+                top_p=None,
+                pad_token_id=self._tokenizer.eos_token_id
+            )
+        raw = self._tokenizer.decode(
+            outputs[0][inputs["input_ids"].shape[1]:],
+            skip_special_tokens=True
+        ).strip()
+        return self._parse_json(raw, query)
+
+    def _parse_json(self, raw: str, original: str) -> dict:
+        """สกัดส่วน JSON จากคู่เครื่องหมาย { และ } เพื่อลดข้อผิดพลาดกรณี LLM ตอบประโยคอภิปรายอื่นปนมา"""
+        try:
+            start_idx = raw.find('{')
+            end_idx = raw.rfind('}')
+            if start_idx != -1 and end_idx != -1:
+                json_str = raw[start_idx:end_idx+1]
+                data = json.loads(json_str)
+                if "fts_queries" in data and "vector_queries" in data:
+                    # คลีนข้อมูลภายในให้ชัวร์ว่าเป็น List of string
+                    fts = [str(x).strip() for x in data["fts_queries"] if x][:2]
+                    vec = [str(x).strip() for x in data["vector_queries"] if x][:2]
+                    return {"fts_queries": fts, "vector_queries": vec}
+        except Exception:
+            pass
+        return FALLBACK(original)

webapp/tipitaka-api/app/services/rag_service.py

@@ -47,11 +47,11 @@ class RAGService:
         self.qdrant_path = settings.QDRANT_PATH
         self.snapshot_dir = Path(settings.SNAPSHOT_DIR)
         self.ollama_url = getattr(settings, "OLLAMA_URL", "http://localhost:11434")
-        self.collections = ["tipitaka_chunks", "tipitaka_scripture"]
+        self.collections = [settings.QDRANT_COLLECTION, "tipitaka_scripture"]
         
         self.model = None  # Flag: None = not verified, 'ready' = OK
         self.reranker = None
-        self.actual_chunks_col = "tipitaka_chunks" # Default name
+        self.actual_chunks_col = settings.QDRANT_COLLECTION # Default name
         self._embed_cache = OrderedDict()     # LRU cache: text → embedding
         self._embed_cache_max = 256           # max cached entries
         self.client = None
@@ -127,6 +127,7 @@ class RAGService:
             logger.error(f"Failed to list collections: {e}")
             all_cols = []
 
+        settings = get_settings()
         for col_name in self.collections:
             actual_col = None
             if col_name in all_cols:
@@ -137,7 +138,7 @@ class RAGService:
                     actual_col = matches[0]
 
             if actual_col:
-                if col_name == "tipitaka_chunks":
+                if col_name == settings.QDRANT_COLLECTION:
                     self.actual_chunks_col = actual_col
             else:
                 snap_filename = f"{col_name}.snapshot"
@@ -153,7 +154,7 @@ class RAGService:
                         abs_snap_path = "/" + abs_snap_path
                     try:
                         self.client.recover_snapshot(col_name, location=f"file://{abs_snap_path}")
-                        if col_name == "tipitaka_chunks":
+                        if col_name == settings.QDRANT_COLLECTION:
                             self.actual_chunks_col = col_name
                     except Exception as e:
                         logger.error(f"Failed to restore: {e}")
@@ -185,39 +186,69 @@ class RAGService:
                 self.reranker = "error"
 
     def _get_embedding(self, text: str) -> list:
-        """Get embedding vector — try Ollama first, fallback to sentence-transformers."""
-        # Check LRU cache first
-        if text in self._embed_cache:
-            self._embed_cache.move_to_end(text)  # Mark as recently used
-            return self._embed_cache[text]
+        """Get embedding vector — wrapper around batched implementation."""
+        res = self._get_embeddings([text])
+        return res[0] if res else []
+
+    def _get_embeddings(self, texts: list[str]) -> list[list[float]]:
+        """Get embedding vectors for a list of texts using batching."""
+        if not texts:
+            return []
+
+        results = [None] * len(texts)
+        missing_indices = []
+        missing_texts = []
+        
+        # 1. Check LRU cache first
+        for idx, text in enumerate(texts):
+            if text in self._embed_cache:
+                self._embed_cache.move_to_end(text)  # Mark as recently used
+                results[idx] = self._embed_cache[text]
+            else:
+                missing_indices.append(idx)
+                missing_texts.append(text)
+                
+        if not missing_texts:
+            return results
 
-        # ── Try 1: Ollama (local dev) ──
+        # 2. Try Ollama batch embedding
+        ollama_failed = False
         try:
             response = httpx.post(
                 f"{self.ollama_url}/api/embed",
-                json={"model": EMBED_MODEL, "input": text},
-                timeout=5
+                json={"model": EMBED_MODEL, "input": missing_texts},
+                timeout=10
             )
             response.raise_for_status()
-            embedding = response.json()["embeddings"][0]
-            self._store_cache(text, embedding)
-            return embedding
+            embeddings = response.json()["embeddings"]
+            for idx, emb in zip(missing_indices, embeddings):
+                self._store_cache(texts[idx], emb)
+                results[idx] = emb
+            return results
         except Exception as e:
-            logger.info(f"Ollama embedding unavailable ({e}), falling back to sentence-transformers")
+            logger.info(f"Ollama batch embedding unavailable ({e}), falling back to sentence-transformers")
+            ollama_failed = True
+
+        # 3. Fallback to sentence-transformers batch embedding
+        if ollama_failed:
+            try:
+                if self._st_model is None:
+                    settings = get_settings()
+                    logger.info(f"Loading sentence-transformers model: {settings.ST_EMBED_MODEL}")
+                    from sentence_transformers import SentenceTransformer
+                    self._st_model = SentenceTransformer(settings.ST_EMBED_MODEL, trust_remote_code=True)
+
+                embeddings = self._st_model.encode(missing_texts, normalize_embeddings=True, show_progress_bar=False).tolist()
+                for idx, emb in zip(missing_indices, embeddings):
+                    self._store_cache(texts[idx], emb)
+                    results[idx] = emb
+                return results
+            except Exception as e:
+                logger.error(f"sentence-transformers batch embedding failed: {e}")
+                for idx in missing_indices:
+                    results[idx] = []
+                return results
 
-        # ── Try 2: sentence-transformers (HF Space / no Ollama) ──
-        try:
-            if self._st_model is None:
-                from sentence_transformers import SentenceTransformer
-                logger.info(f"Loading sentence-transformers model: {ST_EMBED_MODEL}")
-                self._st_model = SentenceTransformer(ST_EMBED_MODEL, trust_remote_code=True)
-
-            embedding = self._st_model.encode(text, normalize_embeddings=True).tolist()
-            self._store_cache(text, embedding)
-            return embedding
-        except Exception as e:
-            logger.error(f"sentence-transformers embedding failed: {e}")
-            return []
 
     def _store_cache(self, text: str, embedding: list):
         """Store embedding in LRU cache."""

webapp/tipitaka-api/app/services/search_service.py

@@ -1,6 +1,7 @@
 import time
 import re
-from typing import List
+import asyncio
+from typing import List, Optional
 from app.database.sqlite_db import SQLiteDB
 from app.schemas import SearchResponse, SearchResultItem
 from app.services.pali_utils import (
@@ -13,11 +14,14 @@ from app.services.pali_utils import (
 
 
 class SearchService:
-    def __init__(self, db: SQLiteDB):
+    def __init__(self, db: SQLiteDB, rag_service=None, query_transform_service=None):
         self.db = db
+        self.rag_service = rag_service
+        self.qts = query_transform_service
 
     # ── Autocomplete suggestions ────────────────────────────────────
 
+
     def get_suggestions(self, prefix: str, limit: int = 8) -> List[str]:
         """
         Return autocomplete suggestions for `prefix`.
@@ -130,37 +134,184 @@ class SearchService:
             result = result[: cut if cut > 0 else max_len].strip()
         return result or text[:max_len]
 
-    def highlight(self, text: str, keyword: str) -> str:
-        """Wrap keyword occurrences in <mark> tags (Thai digits variant too)."""
-        for kw in set([keyword, to_thai_digits(keyword)]):
-            if kw and kw in text:
-                text = text.replace(kw, f"<mark>{kw}</mark>")
-        return text
+    def highlight_multiple(self, text: str, fts_queries: List[str]) -> str:
+        """Wrap keywords from fts_queries in <mark> tags without nesting tags."""
+        if not text or not fts_queries:
+            return text
+
+        words_to_highlight = set()
+        for q in fts_queries:
+            parts = re.split(r"[\s\+\-\*\"\'\(\)]+", q)
+            for p in parts:
+                p_clean = p.strip()
+                # Filter out single character words to prevent over-highlighting
+                if len(p_clean) > 1:
+                    words_to_highlight.add(p_clean)
+                    words_to_highlight.add(to_thai_digits(p_clean))
+
+        sorted_words = sorted(list(words_to_highlight), key=len, reverse=True)
+
+        # ค้นหาช่วงตำแหน่งที่ต้องไฮไลท์แบบไม่ซ้อนทับกัน
+        ranges = []
+        for word in sorted_words:
+            if not word:
+                continue
+            start = 0
+            while True:
+                idx = text.find(word, start)
+                if idx == -1:
+                    break
+                end = idx + len(word)
+
+                # ตรวจสอบการซ้อนทับกับช่วงที่มาร์คไปแล้ว
+                overlap = False
+                for rs, re_range in ranges:
+                    if idx < re_range and end > rs:
+                        overlap = True
+                        break
 
-    def search(self, query: str, limit: int = 50, offset: int = 0) -> SearchResponse:
-        start_time = time.time()
+                if not overlap:
+                    ranges.append((idx, end))
 
-        # ── 1. Auto-correct pipeline ─────────────────────────────
-        q = query.strip()
+                start = idx + 1
 
-        # Stage A: Pali-specific correction (explicit dict + auto-generated variants)
-        corrected_q = PALI_CORRECTIONS_FULL.get(q, q)
+        # เรียงลำดับตำแหน่งจากน้อยไปมาก
+        ranges.sort()
 
-        # Stage B: General Thai spelling correction via PyThaiNLP
-        if corrected_q == q:
-            pythai_q = _pythai_autocorrect(q)
-            if pythai_q != q and _similar_enough(q, pythai_q):
-                corrected_q = pythai_q
+        # สร้างข้อความใหม่พร้อมเครื่องหมาย <mark>
+        result = []
+        last_idx = 0
+        for rs, re_range in ranges:
+            result.append(text[last_idx:rs])
+            result.append(f"<mark>{text[rs:re_range]}</mark>")
+            last_idx = re_range
+        result.append(text[last_idx:])
+        return "".join(result)
 
-        thai_q = to_thai_digits(corrected_q)
-        like_pattern = f"%{thai_q}%"
+    def highlight(self, text: str, keyword: str) -> str:
+        """Wrap keyword in <mark> tags for backward compatibility."""
+        return self.highlight_multiple(text, [keyword]) if keyword else text
+
+    async def _get_vector_results_batch(self, query_texts: List[str], query_vectors: List[List[float]], limit: int = 30) -> List[List[dict]]:
+        """Fetch semantic vector results from Qdrant in batch."""
+        if not self.rag_service or not self.rag_service.client or not query_vectors:
+            return [[] for _ in query_texts]
+
+        import anyio
+        from qdrant_client.http import models as qmodels
+
+        def _blocking():
+            col = self.rag_service.actual_chunks_col
+            
+            # Map request index to original query indices
+            valid_requests = []
+            req_mapping = []
+            
+            for idx, vec in enumerate(query_vectors):
+                if vec:
+                    valid_requests.append(
+                        qmodels.SearchRequest(
+                            vector=qmodels.NamedVector(name="dense", vector=vec),
+                            limit=limit,
+                            with_payload=True,
+                            score_threshold=0.2
+                        )
+                    )
+                    req_mapping.append(idx)
+                    
+            if not valid_requests:
+                return [[] for _ in query_texts]
 
-        with self.db.get_connection() as conn:
-            cursor = conn.cursor()
+            try:
+                # Qdrant search_batch works in both local and server clients
+                batch_responses = self.rag_service.client.search_batch(
+                    collection_name=col,
+                    requests=valid_requests
+                )
+            except Exception as e:
+                import logging
+                logging.getLogger(__name__).warning(f"Qdrant batch query failed: {e}")
+                return [[] for _ in query_texts]
+
+            # Reconstruct list of list of dicts matching query_texts length
+            final_results = [[] for _ in query_texts]
+            for req_idx, hits in zip(req_mapping, batch_responses):
+                vec_results = []
+                for hit in hits:
+                    vol = hit.payload.get("volume", hit.payload.get("volume_id"))
+                    page = hit.payload.get("page", hit.payload.get("page_number"))
+                    content = hit.payload.get("content", "")
+                    vec_results.append({
+                        "volume_id": vol,
+                        "page_number": page,
+                        "content_text": content,
+                        "score": hit.score
+                    })
+                final_results[req_idx] = vec_results
+            return final_results
+
+        return await anyio.to_thread.run_sync(_blocking)
+
+    async def _get_vector_results(self, query_text: str, limit: int = 30) -> List[dict]:
+        """Fetch semantic vector results from Qdrant via RAGService."""
+        if not self.rag_service or not self.rag_service.client:
+            return []
 
-            # ── 2. FTS5 search ──────────────────────────────────────
-            fts_query = f'"{thai_q}"' if "*" not in thai_q else thai_q
+        import anyio
+
+        def _blocking():
+            query_vector = self.rag_service._get_embedding(query_text)
+            if not query_vector:
+                return []
+
+            col = self.rag_service.actual_chunks_col
             try:
+                if hasattr(self.rag_service.client, "search"):
+                    hits = self.rag_service.client.search(
+                        collection_name=col,
+                        query_vector=("dense", query_vector),
+                        limit=limit,
+                        with_payload=True,
+                        score_threshold=0.2
+                    )
+                else:
+                    response = self.rag_service.client.query_points(
+                        collection_name=col,
+                        query=query_vector,
+                        using="dense",
+                        limit=limit,
+                        with_payload=True,
+                        score_threshold=0.2
+                    )
+                    hits = response.points
+            except Exception as e:
+                import logging
+                logging.getLogger(__name__).warning(f"Qdrant query failed: {e}")
+                return []
+
+            vec_results = []
+            for hit in hits:
+                vol = hit.payload.get("volume", hit.payload.get("volume_id"))
+                page = hit.payload.get("page", hit.payload.get("page_number"))
+                content = hit.payload.get("content", "")
+                vec_results.append({
+                    "volume_id": vol,
+                    "page_number": page,
+                    "content_text": content,
+                    "score": hit.score
+                })
+            return vec_results
+
+        return await anyio.to_thread.run_sync(_blocking)
+
+    def _get_fts_results(self, query_text: str, limit: int = 50) -> List[dict]:
+        """Run SQLite FTS5 search on query_text."""
+        thai_q = to_thai_digits(query_text)
+        fts_query = f'"{thai_q}"' if "*" not in thai_q else thai_q
+
+        try:
+            with self.db.get_connection() as conn:
+                cursor = conn.cursor()
                 cursor.execute("""
                     SELECT p.id, p.volume_id, p.page_number, p.content_text, rank
                     FROM pages_fts f
@@ -168,50 +319,128 @@ class SearchService:
                     WHERE pages_fts MATCH ?
                     ORDER BY rank
                     LIMIT ?
-                """, (fts_query, limit + offset))
-                fts_rows = cursor.fetchall()
-            except Exception:
-                fts_rows = []
+                """, (fts_query, limit))
+                rows = cursor.fetchall()
+                return [dict(r) for r in rows]
+        except Exception:
+            return []
 
-            fts_keys = {(r["volume_id"], r["page_number"]) for r in fts_rows}
+    async def _rerank_candidates(self, original_query: str, candidates: list) -> list:
+        """Rerank top candidates against the original query using ONNX Reranker."""
+        if not self.rag_service or not self.rag_service.reranker or self.rag_service.reranker == "error":
+            return candidates
 
-            # ── 3. LIKE fallback ────────────────────────────────────
-            cursor.execute("""
-                SELECT p.id, p.volume_id, p.page_number, p.content_text
-                FROM pages p
-                WHERE p.content_text LIKE ?
-                ORDER BY volume_id, page_number
-                LIMIT ?
-            """, (like_pattern, limit + offset))
-            like_rows = cursor.fetchall()
-
-            # ── 4. Merge & deduplicate ──────────────────────────────
-            merged: list[SearchResultItem] = []
-
-            for r in fts_rows:
-                clean = self.clean_text(r["content_text"])
-                snip  = self.make_snippet(clean, thai_q)
-                merged.append(SearchResultItem(
-                    volume_id=r["volume_id"],
-                    page_number=r["page_number"],
-                    snippet=self.highlight(snip, thai_q),
-                    rank=r["rank"],
-                ))
-
-            for r in like_rows:
-                if (r["volume_id"], r["page_number"]) not in fts_keys:
-                    clean = self.clean_text(r["content_text"])
-                    snip  = self.make_snippet(clean, thai_q)
-                    merged.append(SearchResultItem(
-                        volume_id=r["volume_id"],
-                        page_number=r["page_number"],
-                        snippet=self.highlight(snip, thai_q),
-                        rank=0.0,
-                    ))
-
-            paginated = merged[offset: offset + limit]
-
-            # ── 5. Breakdown by volume ──────────────────────────────
+        import anyio
+        import torch
+
+        def _blocking_rerank():
+            passages = [self.clean_text(c["content_text"])[:1000] for c in candidates]
+            pairs = [[original_query, p] for p in passages]
+
+            try:
+                with torch.no_grad():
+                    scores = self.rag_service.reranker.predict(pairs, show_progress_bar=False, batch_size=4)
+
+                for i, cand in enumerate(candidates):
+                    cand["rerank_score"] = float(scores[i])
+
+                return sorted(candidates, key=lambda x: x.get("rerank_score", 0.0), reverse=True)
+            except Exception as e:
+                import logging
+                logging.getLogger(__name__).warning(f"Reranking failed in search service: {e}")
+                return candidates
+
+        return await anyio.to_thread.run_sync(_blocking_rerank)
+
+    async def search(self, query: str, limit: int = 50, offset: int = 0) -> SearchResponse:
+        start_time = time.time()
+        q = query.strip()
+
+        # ── 1. Auto-correct pipeline ─────────────────────────────
+        corrected_q = PALI_CORRECTIONS_FULL.get(q, q)
+        if corrected_q == q:
+            pythai_q = _pythai_autocorrect(q)
+            if pythai_q != q and _similar_enough(q, pythai_q):
+                corrected_q = pythai_q
+
+        thai_q = to_thai_digits(corrected_q)
+        like_pattern = f"%{thai_q}%"
+
+        # ── 2. Query Transformation (LLM) ─────────────────────────
+        fts_queries = [corrected_q]
+        vector_queries = [corrected_q]
+        
+        if self.qts:
+            transformed = await self.qts.transform(corrected_q)
+            fts_queries = transformed.get("fts_queries", fts_queries)
+            vector_queries = transformed.get("vector_queries", vector_queries)
+
+        # ── 3. Parallel Retrieval (FTS5 + Qdrant Vector) ───────────
+        # Retrieve vector embeddings in batch and perform batch Qdrant searches
+        async def retrieve_vector_results():
+            if not self.rag_service:
+                return [[] for _ in vector_queries]
+            query_vectors = await asyncio.to_thread(self.rag_service._get_embeddings, vector_queries)
+            return await self._get_vector_results_batch(vector_queries, query_vectors)
+
+        # Run FTS searches and Vector searches concurrently
+        fts_tasks = [asyncio.to_thread(self._get_fts_results, fq) for fq in fts_queries]
+
+        # Gather FTS results and Vector results concurrently
+        fts_results_list, vector_results_list = await asyncio.gather(
+            asyncio.gather(*fts_tasks),
+            retrieve_vector_results()
+        )
+
+        # ── 4. RRF Fusion ─────────────────────────────────────────
+        k = 60
+        rrf_scores = {}
+        candidate_data = {}
+
+        for results in fts_results_list:
+            for rank, item in enumerate(results):
+                key = (item["volume_id"], item["page_number"])
+                rrf_scores[key] = rrf_scores.get(key, 0.0) + (1.0 / (k + rank + 1))
+                candidate_data.setdefault(key, item)
+
+        for results in vector_results_list:
+            for rank, item in enumerate(results):
+                key = (item["volume_id"], item["page_number"])
+                rrf_scores[key] = rrf_scores.get(key, 0.0) + (1.0 / (k + rank + 1))
+                candidate_data.setdefault(key, item)
+
+        # Get top 15 candidates by RRF score (down from 30 to speed up reranking)
+        top_keys = sorted(rrf_scores, key=rrf_scores.get, reverse=True)[:15]
+        top_candidates = [candidate_data[ky] for ky in top_keys]
+
+        # ── 5. Reranking (using Jina Reranker v2 ONNX) ────────────
+        if top_candidates:
+            final_results = await self._rerank_candidates(corrected_q, top_candidates)
+        else:
+            final_results = []
+
+        # ── 6. Pagination & Formatting ────────────────────────────
+        paginated = final_results[offset: offset + limit]
+
+        
+        formatted_results = []
+        for r in paginated:
+            clean = self.clean_text(r["content_text"])
+            snip = self.make_snippet(clean, corrected_q)
+            
+            # Highlight with FTS keywords
+            highlighted_snip = self.highlight_multiple(snip, fts_queries)
+            
+            formatted_results.append(SearchResultItem(
+                volume_id=r["volume_id"],
+                page_number=r["page_number"],
+                snippet=highlighted_snip,
+                rank=r.get("rerank_score", r.get("score", 0.0))
+            ))
+
+        # ── 7. Volume Breakdown (SQLite LIKE counts) ───────────────
+        with self.db.get_connection() as conn:
+            cursor = conn.cursor()
             cursor.execute("""
                 SELECT volume_id, COUNT(*) as cnt,
                        MIN(page_number) as first_page,
@@ -223,27 +452,30 @@ class SearchService:
             """, (like_pattern,))
             breakdown_rows = cursor.fetchall()
 
-            breakdown_data = {
-                row["volume_id"]: {
-                    "count": row["cnt"],
-                    "first_page": row["first_page"],
-                    "last_page": row["last_page"],
-                }
-                for row in breakdown_rows
+        breakdown_data = {
+            row["volume_id"]: {
+                "count": row["cnt"],
+                "first_page": row["first_page"],
+                "last_page": row["last_page"],
             }
-            total_count = sum(r["cnt"] for r in breakdown_rows)
+            for row in breakdown_rows
+        }
+        total_count = sum(r["cnt"] for r in breakdown_rows)
 
-            time_ms = (time.time() - start_time) * 1000
+        # If zero results via FTS5/LIKE but we got vector results, guarantee total_count > 0
+        if total_count == 0 and final_results:
+            total_count = len(final_results)
 
-            response = SearchResponse(
-                query=corrected_q,
-                total_results=total_count,
-                results=paginated,
-                time_taken_ms=time_ms,
-                breakdown_by_volume=breakdown_data,
-            )
+        time_ms = (time.time() - start_time) * 1000
 
-        # Log only if results found (don't pollute search_log with zero-result queries)
+        # Log search only if results found
         if total_count > 0:
             self.log_search(corrected_q)
-        return response
+
+        return SearchResponse(
+            query=corrected_q,
+            total_results=total_count,
+            results=formatted_results,
+            time_taken_ms=time_ms,
+            breakdown_by_volume=breakdown_data,
+        )

webapp/tipitaka-api/requirements.txt

@@ -10,9 +10,10 @@ huggingface_hub>=0.20
 torch>=2.0
 pythainlp>=5.0
 onnxruntime>=1.17.0
-transformers>=4.40.0
+transformers>=4.51.0
 numpy>=1.24.0
 httpx>=0.27.0
 anyio>=4.0.0
 pytest>=8.0
 pytest-asyncio>=0.24
+accelerate>=0.27.0

webapp/tipitaka-api/tests/test_search_service.py

@@ -69,10 +69,11 @@ class TestAutocorrectPipeline:
         result = self.service.clean_text(text)
         assert result == "ข้อความปกติ"
 
-    def test_search_accepts_query(self):
+    @pytest.mark.asyncio
+    async def test_search_accepts_query(self):
         """Verify search method accepts a query without raising."""
         # Will not raise if mock handles context manager properly
-        self.service.search("อริยสัจ", limit=5)
+        await self.service.search("อริยสัจ", limit=5)
 
 
 class TestLogSearch: