update big big

core/chunking.py

@@ -1,38 +1,14 @@
 import re
-from typing import List, Tuple
-import logging
-
+from typing import List
 from langchain_text_splitters import RecursiveCharacterTextSplitter
-
 from .config import CHUNK_SIZE, CHUNK_OVERLAP
 
-logger = logging.getLogger(__name__)
-
-STRUCTURE_PATTERNS = [
-    r"(?m)^\s*(Điều\s+\d+[\.:]?)",
-    r"(?m)^\s*(Chương\s+[IVXLC\d]+[\.:]?)",
-    r"(?m)^\s*(Mục\s+\d+[\.:]?)",
-    r"(?m)^\s*(Khoản\s+\d+[\.:]?)",
-    r"(?m)^\s*(Điểm\s+[a-zA-Z0-9]+[\.:]?)",
-    r"(?m)^\s*(\d+(?:\.\d+)*[\)\.])",
-    r"(?m)^\s*([a-zA-Z][\)\.])",
-]
-
-LIST_PATTERNS = [
-    (r"(?m)^\s*a\.", "<LIST_A>"),
-    (r"(?m)^\s*b\.", "<LIST_B>"),
-    (r"(?m)^\s*c\.", "<LIST_C>"),
-    (r"(?m)^\s*\d+\.", "<LIST_NUM>"),
-    (r"(?m)^\s*\d+\)", "<LIST_NUM_PAREN>"),
-    (r"(?m)^\s*-\s+", "<LIST_DASH>"),
-    (r"(?m)^\s*•\s+", "<LIST_BULLET>"),
-]
-
-
-def extract_and_protect_tables(text: str) -> Tuple[str, dict]:
-    table_pattern = re.compile(r"(?:\|.*\|[\r\n]+)+")
+def extract_and_protect_tables(text: str) -> tuple[str, dict]:
+    """Tìm và bọc các bảng Markdown để bảo vệ chúng khỏi việc bị cắt gãy."""
+    # Pattern tìm bảng Markdown (các dòng bắt đầu và chứa ký tự | liên tiếp)
+    table_pattern = re.compile(r'(?:\|.*\|[\r\n]+)+')
     tables = {}
-
+    
     def replace_table(match):
         table_id = f"<TABLE_{len(tables)}>"
         tables[table_id] = match.group(0)
@@ -41,94 +17,47 @@ def extract_and_protect_tables(text: str) -> Tuple[str, dict]:
     protected_text = re.sub(table_pattern, replace_table, text)
     return protected_text, tables
 
-
-def protect_lists(text: str) -> Tuple[str, dict]:
-    placeholders = {}
-    protected = text
-
-    for pattern, token in LIST_PATTERNS:
-        matches = list(re.finditer(pattern, protected))
-        for index, match in enumerate(matches):
-            placeholder = f"{token}_{index}"
-            placeholders[placeholder] = match.group(0)
-            protected = protected.replace(match.group(0), placeholder, 1)
-
-    return protected, placeholders
-
-
-def restore_placeholders(text: str, placeholders: dict) -> str:
-    restored = text
-    for placeholder, original in placeholders.items():
-        restored = restored.replace(placeholder, original)
-    return restored
-
-
-def split_by_structure(text: str) -> List[str]:
-    parts = [text]
-
-    for pattern in STRUCTURE_PATTERNS:
-        next_parts = []
-        for part in parts:
-            matches = list(re.finditer(pattern, part))
-            if len(matches) <= 1:
-                next_parts.append(part)
-                continue
-
-            last_pos = 0
-            for idx, match in enumerate(matches):
-                start = match.start()
-                if idx > 0 and start > last_pos:
-                    chunk = part[last_pos:start].strip()
-                    if chunk:
-                        next_parts.append(chunk)
-                last_pos = start
-
-            tail = part[last_pos:].strip()
-            if tail:
-                next_parts.append(tail)
-
-        parts = next_parts or parts
-
-    return [part for part in parts if part.strip()]
-
-
 def smart_chunking(docs: List) -> List:
-    logger.info("Chunking theo cau truc + do dai...")
-    length_splitter = RecursiveCharacterTextSplitter(
+    print("Đang áp dụng Smart Chunking (Bảo toàn Bảng & Danh sách)...")
+    legal_splitter = RecursiveCharacterTextSplitter(
         chunk_size=CHUNK_SIZE,
         chunk_overlap=CHUNK_OVERLAP,
-        separators=["\n\n", "\n", ". ", " ", ""],
+        separators=[
+            "\nĐiều ", "\nChương ", "\nMục ", "\nKhoản ",
+            "\n\n", "\n", ". ", " ", ""
+        ],
         length_function=len,
-        is_separator_regex=False,
+        is_separator_regex=False
     )
-
+    
     chunks = []
-
     for doc in docs:
-        text = doc.page_content or ""
-        protected_text, tables = extract_and_protect_tables(text)
-        protected_text, list_placeholders = protect_lists(protected_text)
-
-        structure_parts = split_by_structure(protected_text)
-
-        for part in structure_parts:
-            if len(part) <= CHUNK_SIZE:
-                sub_parts = [part]
-            else:
-                sub_parts = length_splitter.split_text(part)
-
-            for chunk_text in sub_parts:
-                restored = restore_placeholders(chunk_text, list_placeholders)
-                restored = restore_placeholders(restored, tables)
-
-                if not restored.strip():
-                    continue
-
-                new_doc = type(doc)(
-                    page_content=restored.strip(),
-                    metadata=doc.metadata.copy() if isinstance(doc.metadata, dict) else {},
-                )
-                chunks.append(new_doc)
-
-    logger.info("Da tao %s chunks.", len(chunks))
+        # 1. Bảo vệ List đang có
+        protected_text = doc.page_content.replace('\na.', '<LIST_a>') \
+                                         .replace('\nb.', '<LIST_b>') \
+                                         .replace('\nc.', '<LIST_c>')
+        
+        # 2. Bảo vệ Table
+        protected_text, tables = extract_and_protect_tables(protected_text)
+        
+        # 3. Tiến hành cắt
+        doc_chunks = legal_splitter.split_text(protected_text)
+        
+        # 4. Phục hồi dữ liệu
+        for chunk_text in doc_chunks:
+            restored = chunk_text.replace('<LIST_a>', '\na.') \
+                                 .replace('<LIST_b>', '\nb.') \
+                                 .replace('<LIST_c>', '\nc.')
+            
+            for table_id, table_content in tables.items():
+                if table_id in restored:
+                    restored = restored.replace(table_id, table_content)
+                    
+            new_doc = type(doc)(
+                page_content=restored,
+                metadata=doc.metadata.copy()
+            )
+            chunks.append(new_doc)
+            
+    print(f" Đã tạo {len(chunks)} chunks thông minh (giữ nguyên cấu trúc bảng)")
     return chunks

core/collection_router_retriever.py

@@ -3,6 +3,7 @@ import logging
 from typing import List
 
 from langchain_core.documents import Document as LangChainDocument
+from rank_bm25 import BM25Okapi
 
 from .collection_utils import collection_matches_cohort
 from .document_db import SessionLocal, list_active_collection_names
@@ -22,6 +23,7 @@ class CollectionRouterRetriever:
         self.qdrant_client = qdrant_client
         self.embeddings_model = embeddings_model
         self.top_n_collections = max(1, int(top_n_collections or 3))
+        self._bm25_cache = {}  # {collection_name -> BM25Okapi instance}
 
     @staticmethod
     def _doc_key(doc) -> str:
@@ -61,7 +63,57 @@ class CollectionRouterRetriever:
 
         return active_collections[: self.top_n_collections]
 
-    def _search_target_collections(self, query: str, collections: List[str], limit: int) -> List:
+    def _ensure_bm25_loaded(self, collection_name: str) -> BM25Okapi | None:
+        """Lazy load and cache BM25 index for a collection.
+        
+        First time: fetch all docs from Qdrant, build BM25, cache it (~0.3s)
+        Subsequent times: reuse from cache (~0.001s)
+        """
+        # Check if already cached
+        if collection_name in self._bm25_cache:
+            return self._bm25_cache[collection_name]
+        
+        try:
+            # Fetch ALL documents from collection (no query vector, get full corpus)
+            all_points = self.qdrant_client.scroll(
+                collection_name=collection_name,
+                limit=10000,  # Batch size
+            )
+            
+            points_list = all_points[0] if isinstance(all_points, tuple) else all_points
+            
+            if not points_list:
+                logger.warning("No documents found in collection=%s for BM25 indexing", collection_name)
+                return None
+            
+            # Extract documents and tokenize for BM25
+            docs_for_bm25 = []
+            for point in points_list:
+                payload = point.payload if isinstance(point.payload, dict) else {}
+                content = str(payload.get("content") or "").strip()
+                if content:
+                    docs_for_bm25.append(content)
+            
+            if not docs_for_bm25:
+                logger.warning("No valid content found in collection=%s for BM25 indexing", collection_name)
+                return None
+            
+            # Build BM25 index
+            tokenized_docs = [doc.lower().split() for doc in docs_for_bm25]
+            bm25 = BM25Okapi(tokenized_docs, k1=1.5, b=0.5)
+            
+            # Cache it
+            self._bm25_cache[collection_name] = bm25
+            logger.info("BM25 index built and cached for collection=%s (docs=%d)", collection_name, len(docs_for_bm25))
+            
+            return bm25
+            
+        except Exception:
+            logger.exception("Failed to build BM25 index for collection=%s", collection_name)
+            return None
+
+    def _search_target_collections(self, query: str, collections: List[str], limit: int, alpha: float = 0.6) -> List:
+        """Hybrid search: BM25 + Vector + RRF (Option 2 with cached BM25)"""
         if not collections:
             return []
 
@@ -71,7 +123,11 @@ class CollectionRouterRetriever:
             logger.exception("Failed to embed query for collection routing")
             return []
 
-        scored_docs = []
+        # Step 1: Vector search (từ Qdrant)
+        all_docs_dict = {}  # {doc_key -> LangChainDocument}
+        vector_ranked = {}  # {doc_key -> rank}
+        
+        vector_rank = 0
         for collection_name in collections:
             try:
                 points = self.qdrant_client.search(
@@ -101,15 +157,73 @@ class CollectionRouterRetriever:
                     "chunk_index": payload.get("chunk_index"),
                     "page_number": payload.get("page_number"),
                 }
-                scored_docs.append(
-                    (
-                        float(getattr(point, "score", 0.0) or 0.0),
-                        LangChainDocument(page_content=content, metadata=metadata),
-                    )
-                )
+                doc = LangChainDocument(page_content=content, metadata=metadata)
+                doc_key = self._doc_key(doc)
+                
+                all_docs_dict[doc_key] = doc
+                if doc_key not in vector_ranked:
+                    vector_rank += 1
+                    vector_ranked[doc_key] = vector_rank
+
+        # Step 2: BM25 search (lexical) - using CACHED index
+        bm25_ranked = {}  # {doc_key -> rank}
+        if all_docs_dict:
+            try:
+                tokenized_query = query.lower().split()
+                
+                # For each collection, use cached BM25 index
+                for collection_name in collections:
+                    # Load cached BM25 (or build if first time)
+                    bm25 = self._ensure_bm25_loaded(collection_name)
+                    if bm25 is None:
+                        continue
+                    
+                    # Get BM25 scores for vector results
+                    docs_from_collection = [
+                        doc for doc in all_docs_dict.values()
+                        if doc.metadata.get("collection_name") == collection_name
+                    ]
+                    
+                    if not docs_from_collection:
+                        continue
+                    
+                    # Get BM25 ranks
+                    bm25_results = bm25.get_top_n(tokenized_query, docs_from_collection, n=len(docs_from_collection))
+                    
+                    bm25_rank = 0
+                    for doc in bm25_results:
+                        doc_key = self._doc_key(doc)
+                        if doc_key not in bm25_ranked:
+                            bm25_rank += 1
+                            bm25_ranked[doc_key] = bm25_rank
+                            
+            except Exception:
+                logger.exception("BM25 search failed, falling back to vector-only")
 
-        scored_docs.sort(key=lambda row: row[0], reverse=True)
-        return [doc for _, doc in scored_docs]
+        # Step 3: RRF combination (Reciprocal Rank Fusion)
+        alpha = max(0.0, min(1.0, float(alpha)))
+        bm25_weight = 1.0 - alpha
+        vector_weight = alpha
+        rrf_c = 60
+        
+        rrf_scores = {}
+        for doc_key, doc in all_docs_dict.items():
+            score = 0.0
+            
+            # Vector score
+            if doc_key in vector_ranked:
+                score += vector_weight / (rrf_c + vector_ranked[doc_key])
+            
+            # BM25 score
+            if doc_key in bm25_ranked:
+                score += bm25_weight / (rrf_c + bm25_ranked[doc_key])
+            
+            if score > 0:
+                rrf_scores[doc_key] = score
+        
+        # Sort by RRF score
+        sorted_results = sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True)
+        return [all_docs_dict[doc_key] for doc_key, _ in sorted_results[:limit]]
 
     def search(self, query: str, k: int = 10, alpha: float = 0.6, cohort_key: str | None = None) -> List:
         if k <= 0:
@@ -126,6 +240,7 @@ class CollectionRouterRetriever:
             query=query,
             collections=target_collections,
             limit=candidate_k,
+            alpha=alpha,
         )
 
         if cohort_scoped:

core/config.py

@@ -39,8 +39,8 @@ GEMINI_API_KEYS = os.getenv('GEMINI_API_KEYS', '').strip()
 # Name models
 LLM_MODEL = os.getenv('LLM_MODEL', 'llama-3.1-70b-versatile')
 FAST_LLM_MODEL = os.getenv('FAST_LLM_MODEL', 'llama-3.1-8b-instant')
-EMBED_MODEL = os.getenv('EMBED_MODEL', 'BAAI/bge-m3')
-CROSS_ENCODER_MODEL = os.getenv('CROSS_ENCODER_MODEL', 'BAAI/bge-reranker-v2-m3')
+EMBED_MODEL = os.getenv('EMBED_MODEL', 'bkai-foundation-models/vietnamese-bi-encoder')
+CROSS_ENCODER_MODEL = os.getenv('CROSS_ENCODER_MODEL', 'itdainb/PhoRanker')
 
 # Chunking and retrieval settings
 CHUNK_SIZE = int(os.getenv('CHUNK_SIZE', '800'))

core/qa_pipeline.py

@@ -4,6 +4,8 @@ import logging
 import groq
 import google.generativeai as genai
 import json 
+from concurrent.futures import ThreadPoolExecutor
+from threading import Lock
 
 from .models import llm
 from .config import TOP_K_RESULTS, FINAL_TOP_K
@@ -239,6 +241,53 @@ def ask_ai_stream_delta(message: str, history: List, hybrid_retriever, cohort_ke
         yield "Chào bạn 👋 Mình hỗ trợ tra cứu quy chế đào tạo. Bạn cần hỏi điều gì?"
         return
 
+    # [SKIP LLM] Nếu đây là câu hỏi đầu tiên (history trống), bỏ qua LLM, chỉ trả về các tài liệu liên quan
+    if not history or len(history) == 0:
+        logger.info(f"[FIRST TURN] CÂU HỎI GỐC: {message}")
+        question = message.strip()
+        processed_data = analyze_and_expand_query(question)
+        queries = processed_data.get('expanded_queries', [question])
+        
+        # Chỉ tìm kiếm docs, không gọi LLM
+        all_docs: List = []
+        seen = set()
+        seen_lock = Lock()
+        
+        def search_query(query: str):
+            current_alpha = 0.4 if "CNTT" in query.upper() else 0.5
+            return hybrid_retriever.search(
+                query,
+                k=TOP_K_RESULTS,
+                alpha=current_alpha,
+                cohort_key=cohort_key,
+            )
+        
+        with ThreadPoolExecutor(max_workers=min(3, len(queries))) as executor:
+            futures = {executor.submit(search_query, q): q for q in queries}
+            for future in futures:
+                try:
+                    docs = future.result(timeout=30)
+                    for doc in docs:
+                        content_hash = hashlib.sha256(doc.page_content.encode("utf-8")).hexdigest()
+                        with seen_lock:
+                            if content_hash not in seen:
+                                seen.add(content_hash)
+                                all_docs.append(doc)
+                except Exception:
+                    logger.exception("Search error")
+        
+        # Format documents thành văn bản trả về
+        if all_docs:
+            result_text = "📚 **Các tài liệu liên quan:**\n\n"
+            for i, doc in enumerate(all_docs[:FINAL_TOP_K], 1):
+                source = doc.metadata.get("source") or "Không rõ"
+                content_preview = doc.page_content[:300] + ("..." if len(doc.page_content) > 300 else "")
+                result_text += f"{i}. **Nguồn:** {source}\n{content_preview}\n\n"
+            yield result_text + "\n💡 *Hãy đặt câu hỏi cụ thể hơn để được hỗ trợ tốt hơn!*"
+        else:
+            yield "❌ Không tìm thấy tài liệu liên quan. Vui lòng hãy đặt câu hỏi cụ thể hơn!"
+        return
+
     logger.info(f" CÂU HỎI GỐC: {message}")
     question = generate_standalone_query(message, history)
 
@@ -255,23 +304,33 @@ def ask_ai_stream_delta(message: str, history: List, hybrid_retriever, cohort_ke
 
     all_docs: List = []
     seen = set()
+    seen_lock = Lock()
     if cohort_key:
         logger.info(f"Sử dụng cohort_key: {cohort_key}")
     
-    for query in queries:
-        #Giữ nguyên logic alpha ngành CNTT của Minh
+    # Gửi song song các truy vấn đến Qdrant 
+    def search_query(query: str):
         current_alpha = 0.4 if "CNTT" in query.upper() else 0.5
-        docs = hybrid_retriever.search(
+        return hybrid_retriever.search(
             query,
             k=TOP_K_RESULTS,
             alpha=current_alpha,
             cohort_key=cohort_key,
         )
-        for doc in docs:
-            content_hash = hashlib.sha256(doc.page_content.encode("utf-8")).hexdigest()
-            if content_hash not in seen:
-                all_docs.append(doc)
-                seen.add(content_hash)
+    
+    with ThreadPoolExecutor(max_workers=min(3, len(queries))) as executor:
+        futures = {executor.submit(search_query, q): q for q in queries}
+        for future in futures:
+            try:
+                docs = future.result(timeout=30)
+                for doc in docs:
+                    content_hash = hashlib.sha256(doc.page_content.encode("utf-8")).hexdigest()
+                    with seen_lock:
+                        if content_hash not in seen:
+                            all_docs.append(doc)
+                            seen.add(content_hash)
+            except Exception as e:
+                logger.error(f"Lỗi khi search query '{futures[future]}': {e}")
 
     logger.info(f"Tìm thấy tổng {len(all_docs)} documents.")
     if not all_docs:
@@ -300,10 +359,12 @@ def ask_ai_stream_delta(message: str, history: List, hybrid_retriever, cohort_ke
     logger.info("Đang tạo câu trả lời cuối cùng ...")
     
     success = False
-    # Thử với Groq 
-    for _ in range(len(api_manager.groq_keys)):
+    # Ưu tiên Groq (tiết kiệm token)
+    for _ in range(len(api_manager.groq_keys) if api_manager.groq_keys else 1):
         try:
             client = api_manager.get_groq_client()
+            if not client:
+                break
             stream = client.chat.completions.create(
                 model="llama-3.3-70b-versatile",
                 messages=[{"role": "user", "content": prompt}],
@@ -316,16 +377,16 @@ def ask_ai_stream_delta(message: str, history: List, hybrid_retriever, cohort_ke
             success = True
             break
         except Exception as e:
-            if "429" in str(e): # Lỗi Rate Limit
+            if "429" in str(e):  # Rate Limit
                 api_manager.rotate_groq()
                 continue
             logger.error(f"Lỗi Groq: {e}")
             break
-            
-    # Dự phòng sang Gemini (nếu Groq lỗi hoặc hết key)
+    
+    # Fallback sang Gemini nếu Groq lỗi
     if not success:
         logger.warning("Chuyển sang Gemini ...")
-        for _ in range(max(1, len(api_manager.gemini_keys))):
+        for _ in range(len(api_manager.gemini_keys) if api_manager.gemini_keys else 1):
             try:
                 genai.configure(api_key=api_manager.get_gemini_key())
                 model = genai.GenerativeModel('gemini-2.5-flash')
@@ -338,6 +399,6 @@ def ask_ai_stream_delta(message: str, history: List, hybrid_retriever, cohort_ke
             except Exception as e:
                 api_manager.rotate_gemini()
                 logger.error(f"Lỗi Gemini: {e}")
-
+    
     if not success:
         yield "Đã xảy ra lỗi hệ thống hoặc quá tải. Vui lòng thử lại sau giây lát!"

core/rerank.py

@@ -1,15 +1,21 @@
 from typing import List
 from .models import cross_encoder
-
-MAX_RERANK_CHARS = 1200
+import logging
+logger = logging.getLogger(__name__)
+MAX_RERANK_CHARS = 800
 
 def advanced_rerank(question: str, docs: List, top_k: int = 5) -> List:
     if not docs:
         return []
-    print(f"Đang rerank {len(docs)} documents với Cross-Encoder...")
-    pairs = [(question, (doc.page_content or "")[:MAX_RERANK_CHARS]) for doc in docs]
-    scores = cross_encoder.predict(pairs)
-    ranked = sorted(zip(scores, docs), key=lambda x: x[0], reverse=True)
-    print(f" Top 3 scores: {[f'{s:.3f}' for s, _ in ranked[:3]]}")
+    MAX_DOCS_TO_RERANK = 15
+    pruned_docs = docs[:MAX_DOCS_TO_RERANK]
+    
+    logger.info("Đang rerank %s tài liệu với Cross-Encoder...", len(pruned_docs))
+    pairs = [(question, (doc.page_content or "")[:MAX_RERANK_CHARS]) for doc in pruned_docs]
+    
+    scores = cross_encoder.predict(pairs, show_progress_bar=False)
+    ranked = sorted(zip(scores, pruned_docs), key=lambda x: x[0], reverse=True)
+    
+    logger.info("Top 3 điểm: %s", [f"{s:.3f}" for s, _ in ranked[:3]])
     return [doc for score, doc in ranked[:top_k]]
 

main.py

@@ -348,6 +348,10 @@ async def chat_endpoint(payload: ChatRequest, request: Request):
     
     history = await get_history_async(db_pool, session_id)
     
+    # Nếu lịch sử < 2 messages, bỏ qua (không dùng context)
+    if len(history) < 2:
+        history = []
+    
     # Tập hợp toàn bộ response từ generator
     full_response = ""
     try:
@@ -380,6 +384,10 @@ async def chat_stream_endpoint(payload: ChatRequest, request: Request):
     
     history = await get_history_async(db_pool, session_id)
     
+    # Nếu lịch sử < 2 messages, bỏ qua (không dùng context)
+    if len(history) < 2:
+        history = []
+    
     async def event_stream_generator():
         """Generator SSE - yield mỗi delta chunk và cuối cùng done=true"""
         full_response = ""