hi

app.py

@@ -30,7 +30,7 @@ import textwrap
 # ---------- 第三方函式庫 ----------
 import numpy as np
 import pandas as pd
-from fastapi import FastAPI, Request, Response, HTTPException, status
+from fastapi import FastAPI, Request, Response, HTTPException, status, BackgroundTasks
 import uvicorn
 import jieba
 from rank_bm25 import BM25Okapi
@@ -40,7 +40,6 @@ import torch
 from openai import OpenAI
 from tenacity import retry, stop_after_attempt, wait_fixed
 import requests
-from starlette.concurrency import run_in_threadpool
 
 # ==== CONFIG (從環境變數載入，或使用預設值) ====
 # 根據提供的檔案清單，將預設路徑設定為當前目錄
@@ -157,7 +156,6 @@ class RagPipeline:
         """在啟動時載入所有必要的模型與資料"""
         log.info("開始載入資料與模型...")
 
-        # Load CSV and check for required columns
         if not os.path.exists(self.csv_path):
             raise FileNotFoundError(f"找不到 CSV 檔案於 {self.csv_path}")
 
@@ -172,15 +170,19 @@ class RagPipeline:
         )
         log.info(f"成功載入 CSV: {self.csv_path} (rows={len(self.df_csv)})")
 
-        # Load corpus and index
         self.state.index, self.state.sentences, self.state.meta = self._load_or_build_sentence_index()
         self.state.bm25 = self._ensure_bm25_index()
 
-        # Check for BM25 alignment
-        bm_n = len(self.state.bm25.corpus)
+        # Check for BM25 and meta alignment
         sent_n = len(self.state.sentences)
-        if bm_n != sent_n:
+        meta_n = len(self.state.meta)
+        bm_n = getattr(self.state.bm25, 'corpus_size', len(getattr(self.state.bm25, 'doc_len', [])))
+        
+        if sent_n != bm_n:
             raise RuntimeError(f"BM25 文件數 ({bm_n}) 與 sentences ({sent_n}) 不一致，請重新生成索引。")
+        if sent_n != meta_n:
+            raise RuntimeError(f"sentences ({sent_n}) 與 meta ({meta_n}) 長度不一致，請重新生成索引。")
+
         log.info("所有模型與資料載入完成。")
 
     def _load_or_build_sentence_index(self):
@@ -194,7 +196,6 @@ class RagPipeline:
             return index, sentences, meta
 
         log.info("索引檔案不存在，正在從 CSV 重新建立...")
-        # This function should be run by a separate script, not here.
         raise RuntimeError("FAISS 和句子 PKL 檔案未找到，請先執行索引生成腳本。")
 
     def _ensure_bm25_index(self):
@@ -203,11 +204,15 @@ class RagPipeline:
             try:
                 with open(BM25_PKL, "rb") as f:
                     data = pickle.load(f)
-                    bm25 = data.get("bm25") if isinstance(data, dict) else data
-                    if not hasattr(bm25, 'get_scores'):
-                        raise ValueError("載入的 BM25 索引無效。")
-                    log.info(f"成功載入 BM25 索引，包含 {len(bm25.corpus)} 篇文件。")
-                    return bm25
+                bm25 = data.get("bm25") if isinstance(data, dict) else data
+                if not hasattr(bm25, 'get_scores'):
+                    raise ValueError("載入的 BM25 索引無效。")
+                
+                # Use a more robust way to get corpus size
+                corpus_size = getattr(bm25, 'corpus_size', len(getattr(bm25, 'doc_len', [])))
+                log.info(f"成功載入 BM25 索引，包含 {corpus_size} 篇文件。")
+                setattr(self.state, 'bm25_corpus_len', corpus_size)
+                return bm25
             except Exception as e:
                 log.error(f"載入 BM25 索引失敗 ({e})，請檢查檔案格式。")
                 raise RuntimeError("BM25 索引檔案損壞或格式不符。")
@@ -254,7 +259,10 @@ class RagPipeline:
             all_reranked_results = []
             
             log.info("步驟 3/5: 檢索與重排序...")
-            relevant_indices = {i for i, m in enumerate(self.state.meta) if m.get("drug_id") in drug_ids}
+            # Ensure drug_id is always string for robust matching
+            drug_ids_set = {str(did) for did in drug_ids}
+            relevant_indices = {i for i, m in enumerate(self.state.meta) if str(m.get("drug_id")) in drug_ids_set}
+
             if not relevant_indices:
                 log.error("找不到與藥品相關的語料。")
                 return f"找不到 drug_id {drug_ids} 對應的任何 chunks。{DISCLAIMER}"
@@ -299,6 +307,8 @@ class RagPipeline:
                 bm_vals  = np.array([b for _, _, b in candidates_list], dtype=np.float32)
 
                 def norm(x):
+                    if len(x) == 0 or (x.max() - x.min()) == 0:
+                        return np.zeros_like(x)
                     rng = x.max() - x.min()
                     return (x - x.min()) / (rng + 1e-8)
                 
@@ -441,6 +451,9 @@ class RagPipeline:
     def _semantic_search(self, index, query: str, top_k: int, embedding_model) -> Tuple[List[int], List[float]]:
         if not query:
             return [], []
+        
+        top_k = min(top_k, index.ntotal)
+        
         q_emb = embedding_model.encode([query], convert_to_numpy=True).astype("float32")
         faiss.normalize_L2(q_emb)
 
@@ -528,16 +541,16 @@ app = FastAPI()
 rag_pipeline = None
 
 class AppConfig:
-    CHANNEL_ACCESS_TOKEN = os.getenv("CHANNEL_ACCESS_TOKEN")
-    CHANNEL_SECRET = os.getenv("CHANNEL_SECRET")
+    CHANNEL_ACCESS_TOKEN = os.getenv("LINE_CHANNEL_ACCESS_TOKEN")
+    CHANNEL_SECRET = os.getenv("LINE_CHANNEL_SECRET")
 
 @app.on_event("startup")
 async def startup_event():
     """應用程式啟動時執行的任務"""
     log.info("===== Application Startup =====")
     missing = []
-    if not AppConfig.CHANNEL_ACCESS_TOKEN: missing.append("CHANNEL_ACCESS_TOKEN")
-    if not AppConfig.CHANNEL_SECRET: missing.append("CHANNEL_SECRET")
+    if not AppConfig.CHANNEL_ACCESS_TOKEN: missing.append("LINE_CHANNEL_ACCESS_TOKEN")
+    if not AppConfig.CHANNEL_SECRET: missing.append("LINE_CHANNEL_SECRET")
     if not LLM_API_CONFIG.get("api_key"): missing.append("LITELLM_API_KEY")
     if not LLM_API_CONFIG.get("base_url"): missing.append("LITELLM_BASE_URL")
     if not LLM_API_CONFIG.get("model"): missing.append("LM_MODEL")
@@ -556,8 +569,18 @@ async def health_check():
     """健康檢查端點，用於 Docker HEALTHCHECK"""
     return {"status": "ok"}
 
+def process_and_reply(reply_token: str, user_text: str):
+    """將耗時的 RAG 處理和回覆任務移至後台執行"""
+    try:
+        answer = rag_pipeline.answer_question(user_text)
+    except Exception as e:
+        log.error(f"後台處理錯誤: {e}", exc_info=True)
+        answer = "處理時發生錯誤，請稍後再試。"
+    line_reply(reply_token, answer)
+
+
 @app.post("/webhook")
-async def handle_webhook(request: Request, response: Response):
+async def handle_webhook(request: Request, background_tasks: BackgroundTasks):
     """處理 LINE Message API 的 Webhook 請求"""
     signature = request.headers.get("X-Line-Signature")
     if not signature:
@@ -580,16 +603,11 @@ async def handle_webhook(request: Request, response: Response):
         if event.get("type") == "message" and event.get("message", {}).get("type") == "text":
             reply_token = event.get("replyToken")
             user_text = event.get("message", {}).get("text", "").strip()
-
-            if not user_text: continue
             
-            # Offload heavy work to a thread pool
-            answer = await run_in_threadpool(rag_pipeline.answer_question, user_text)
-
-            if reply_token:
-                line_reply(reply_token, answer)
+            if reply_token and user_text:
+                background_tasks.add_task(process_and_reply, reply_token, user_text)
                 
-    return {"status": "ok"}
+    return {"status": "ok"} # Immediately return 200 OK
 
 def line_reply(reply_token: str, text: str):
     """透過 LINE Message API 回覆訊息，並進行分塊以避免長度限制"""
@@ -597,9 +615,9 @@ def line_reply(reply_token: str, text: str):
         "Content-Type": "application/json",
         "Authorization": f"Bearer {AppConfig.CHANNEL_ACCESS_TOKEN}"
     }
-    # LINE 文本長度上限約為 5000 字元
-    chunks = textwrap.wrap(text, 4900)
-    messages = [{"type": "text", "text": c} for c in chunks] or [{"type": "text", "text": text[:4900]}]
+    # LINE 文本長度上限約為 5000 字元，且回覆訊息數上限為 5 則
+    chunks = textwrap.wrap(text, 4000)
+    messages = [{"type": "text", "text": c} for c in chunks[:5]] or [{"type": "text", "text": text[:4000]}]
     data = {"replyToken": reply_token, "messages": messages}
     try:
         r = requests.post("https://api.line.me/v2/bot/message/reply", headers=headers, json=data, timeout=10)