hi

app.py

@@ -43,7 +43,24 @@ from openai import OpenAI
 from tenacity import retry, stop_after_attempt, wait_fixed
 import requests
 
+# [MODIFIED] 限制 PyTorch 執行緒數量，避免在 CPU 環境下過度佔用資源
+torch.set_num_threads(int(os.getenv("TORCH_NUM_THREADS", "1")))
+
 # ==== CONFIG (從環境變數載入，或使用預設值) ====
+# [MODIFIED] 新增環境變數健檢函式
+def _require_env(var: str) -> str:
+    v = os.getenv(var)
+    if not v:
+        raise RuntimeError(f"FATAL: Missing required environment variable: {var}")
+    return v
+
+# [MODIFIED] 檢查 LLM 相關環境變數
+def _require_llm_config():
+    for k in ("LITELLM_BASE_URL", "LITELLM_API_KEY", "LM_MODEL"):
+        _require_env(k)
+
+_require_llm_config()
+
 CSV_PATH = os.getenv("CSV_PATH", "cleaned_combined.csv")
 FAISS_INDEX = os.getenv("FAISS_INDEX", "drug_sentences.index")
 SENTENCES_PKL = os.getenv("SENTENCES_PKL", "drug_sentences.pkl")
@@ -51,7 +68,7 @@ BM25_PKL = os.getenv("BM25_PKL", "bm25.pkl")
 
 TOP_K_SENTENCES = int(os.getenv("TOP_K_SENTENCES", 15))
 PRE_RERANK_K = int(os.getenv("PRE_RERANK_K", 30))
-MAX_RERANK_CANDIDATES = int(os.getenv("MAX_RERANK_CANDIDATES", 30)) # 根據使用者要求調整
+MAX_RERANK_CANDIDATES = int(os.getenv("MAX_RERANK_CANDIDATES", 30))
 
 EMBEDDING_MODEL = os.getenv("EMBEDDING_MODEL", "DMetaSoul/Dmeta-embedding-zh")
 RERANKER_MODEL = os.getenv("RERANKER_MODEL", "BAAI/bge-reranker-v2-m3")
@@ -81,7 +98,6 @@ DRUG_NAME_MAPPING = {
 }
 DISCLAIMER = "本資訊僅供參考，若您對藥物使用有任何疑問，請務務必諮詢您的醫師或藥師。"
 
-# [NEW] 集中管理 Prompt 模板
 PROMPT_TEMPLATES = {
     "analyze_query": """
 請分析以下使用者問題，並完成以下兩個任務：
@@ -126,7 +142,6 @@ PROMPT_TEMPLATES = {
 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
 log = logging.getLogger(__name__)
 
-# [NEW] 使用 Dataclasses 提升程式碼可讀性
 @dataclass
 class FusedCandidate:
     idx: int
@@ -145,7 +160,9 @@ class RerankResult:
 class RagPipeline:
     def __init__(self, config):
         self.config = config
-        self.state = type('state', (), {})() # Simple state object
+        self.state = type('state', (), {})()
+        if not LLM_API_CONFIG["api_key"] or not LLM_API_CONFIG["base_url"]:
+             raise ValueError("LLM API Key or Base URL is not configured.")
         self.llm_client = OpenAI(api_key=LLM_API_CONFIG["api_key"], base_url=LLM_API_CONFIG["base_url"])
         self.embedding_model = self._load_model(SentenceTransformer, EMBEDDING_MODEL, "embedding")
         self.reranker = self._load_model(CrossEncoder, RERANKER_MODEL, "reranker")
@@ -164,24 +181,33 @@ class RagPipeline:
 
     def load_data(self):
         log.info("開始載入資料與模型...")
-        self.df_csv = pd.read_csv(CSV_PATH, dtype=str).fillna('')
-        self.df_csv['drug_name_norm_normalized'] = (
-            self.df_csv['drug_name_norm'].str.lower().str.replace(r'[^\w\s]', '', regex=True).str.strip()
-        )
-        self.drug_name_to_ids = self.df_csv.groupby('drug_name_norm_normalized')['drug_id'].unique().apply(list).to_dict()
-        self._load_drug_name_vocabulary()
+        try:
+            self.df_csv = pd.read_csv(CSV_PATH, dtype=str).fillna('')
+            # [MODIFIED] 增加必要欄位檢查
+            for col in ("drug_name_norm", "drug_id"):
+                if col not in self.df_csv.columns:
+                    raise KeyError(f"CSV 檔案 '{CSV_PATH}' 中缺少必要欄位: {col}")
+
+            self.df_csv['drug_name_norm_normalized'] = (
+                self.df_csv['drug_name_norm'].str.lower().str.replace(r'[^\w\s]', '', regex=True).str.strip()
+            )
+            self.drug_name_to_ids = self.df_csv.groupby('drug_name_norm_normalized')['drug_id'].unique().apply(list).to_dict()
+            self._load_drug_name_vocabulary()
+            
+            log.info("載入 FAISS 索引與句子資料...")
+            self.state.index = faiss.read_index(FAISS_INDEX)
+            with open(SENTENCES_PKL, "rb") as f:
+                data = pickle.load(f)
+                self.state.sentences = data["sentences"]
+                self.state.meta = data["meta"]
+
+            log.info("載入 BM25 索引...")
+            with open(BM25_PKL, "rb") as f:
+                self.state.bm25 = pickle.load(f)
+        except (FileNotFoundError, KeyError) as e:
+            log.exception(f"資料或索引檔案載入失敗: {e}")
+            raise RuntimeError(f"資料初始化失敗，請檢查檔案路徑與內容: {e}")
         
-        log.info("載入 FAISS 索引與句子資料...")
-        self.state.index = faiss.read_index(FAISS_INDEX)
-        with open(SENTENCES_PKL, "rb") as f:
-            data = pickle.load(f)
-            self.state.sentences = data["sentences"]
-            self.state.meta = data["meta"]
-
-        log.info("載入 BM25 索引...")
-        with open(BM25_PKL, "rb") as f:
-            self.state.bm25 = pickle.load(f)
-
         log.info("所有模型與資料載入完���。")
 
     def _load_drug_name_vocabulary(self):
@@ -248,15 +274,24 @@ class RagPipeline:
     @lru_cache(maxsize=128)
     def _find_drug_ids_from_name(self, query: str) -> List[str]:
         candidates = extract_drug_candidates_from_query(query.lower(), self.drug_vocab)
-        expanded = {c.lower().replace(" ", "") for c in candidates} | set(candidates)
         
         drug_ids = set()
-        for alias in expanded:
+        for alias in candidates:
+            # [MODIFIED] 英文藥名比對使用詞邊界，避免子字串誤判
+            is_english = not re.search(r'[\u4e00-\u9fff]', alias)
             for drug_name_norm, ids in self.drug_name_to_ids.items():
-                if alias in drug_name_norm:
+                match = False
+                if is_english:
+                    if re.search(rf"\b{re.escape(alias)}\b", drug_name_norm):
+                        match = True
+                elif alias in drug_name_norm:
+                    match = True
+                
+                if match:
                     drug_ids.update(ids)
         return list(drug_ids)
 
+
     def _analyze_query(self, query: str) -> Dict[str, Any]:
         prompt = PROMPT_TEMPLATES["analyze_query"].format(
             options="\n".join(f"- {c}" for c in INTENT_CATEGORIES),
@@ -280,48 +315,52 @@ class RagPipeline:
             distances, sim_indices = self.state.index.search(q_emb, PRE_RERANK_K)
 
             tokenized_query = list(jieba.cut(expanded_q))
-            # [MODIFIED] 使用 get_top_n 提升 BM25 效率
-            bm25_results = self.state.bm25.get_top_n(tokenized_query, self.state.sentences, n=PRE_RERANK_K)
             
-            doc_to_bm25_score = {self.state.sentences.index(doc): score for doc, score in zip(bm25_results, np.arange(PRE_RERANK_K, 0, -1))}
-
+            # [MODIFIED] 改為獲取真實 BM25 分數，而非使用排名
+            bm25_scores = self.state.bm25.get_scores(tokenized_query)
+            top_bm25_indices = np.argsort(bm25_scores)[::-1][:PRE_RERANK_K]
+            doc_to_bm25_score = {int(i): float(bm25_scores[i]) for i in top_bm25_indices}
+            
             candidate_scores: Dict[int, Dict[str, float]] = {}
+            
+            # [MODIFIED] FAISS L2 距離轉為相似度 (分數越高越好)
+            def dist_to_sim(d: float) -> float:
+                return 1.0 / (1.0 + d)
+
             for i, dist in zip(sim_indices[0], distances[0]):
                 if i in relevant_indices:
-                    candidate_scores[i] = {"sem": float(dist), "bm": 0.0}
+                    similarity = dist_to_sim(dist)
+                    candidate_scores[int(i)] = {"sem": float(similarity), "bm": 0.0}
 
             for i, score in doc_to_bm25_score.items():
                 if i in relevant_indices:
-                    if i in candidate_scores:
-                        candidate_scores[i]["bm"] = score
-                    else:
-                        candidate_scores[i] = {"sem": 0.0, "bm": score}
+                    candidate_scores.setdefault(i, {"sem": 0.0, "bm": 0.0})["bm"] = score
             
             if not candidate_scores: continue
             
-            sem_scores = np.array([s['sem'] for s in candidate_scores.values()])
-            bm_scores = np.array([s['bm'] for s in candidate_scores.values()])
-            
-            def norm(x): return (x - x.min()) / (x.max() - x.min() + 1e-8) if x.max() > x.min() else np.zeros_like(x)
+            # [MODIFIED] 使用固定的鍵順序來確保分數對齊
+            keys = list(candidate_scores.keys())
+            sem_scores = np.array([candidate_scores[k]['sem'] for k in keys])
+            bm_scores = np.array([candidate_scores[k]['bm'] for k in keys])
             
+            def norm(x): 
+                rng = x.max() - x.min()
+                return (x - x.min()) / (rng + 1e-8) if rng > 0 else np.zeros_like(x)
+
             sem_n, bm_n = norm(sem_scores), norm(bm_scores)
             
-            for idx, (i, scores) in enumerate(candidate_scores.items()):
+            for idx, k in enumerate(keys):
                 fused_score = sem_n[idx] * 0.6 + bm_n[idx] * 0.4
                 
-                if i not in all_fused_candidates or fused_score > all_fused_candidates[i].fused_score:
-                    all_fused_candidates[i] = FusedCandidate(
-                        idx=i, fused_score=fused_score, sem_score=scores['sem'], bm_score=scores['bm']
+                if k not in all_fused_candidates or fused_score > all_fused_candidates[k].fused_score:
+                    all_fused_candidates[k] = FusedCandidate(
+                        idx=k, fused_score=fused_score, sem_score=sem_scores[idx], bm_score=bm_scores[idx]
                     )
 
         return sorted(all_fused_candidates.values(), key=lambda x: x.fused_score, reverse=True)
 
-    @lru_cache(maxsize=128)
+    # [MODIFIED] 移除 lru_cache，因對多變的長查詢效果不佳
     def _expand_query_with_llm(self, query: str, intents: tuple) -> str:
-        """
-        Expands a query using the LLM. 
-        [CORRECTED] Now safely handles failed or empty LLM responses.
-        """
         if not intents:
             return query
             
@@ -329,14 +368,13 @@ class RagPipeline:
         
         try:
             expanded_query = self._llm_call([{"role": "user", "content": prompt}])
-            # Ensure the result is a non-empty string before returning
             if expanded_query and expanded_query.strip():
                 return expanded_query
             else:
                 log.warning(f"Query expansion for '{query}' returned an empty result. Using original query.")
                 return query
         except Exception as e:
-            log.error(f"Query expansion for '{query}' failed with an exception: {e}. Using original query.")
+            log.error(f"Query expansion for '{query}' failed: {e}. Using original query.")
             return query
 
     def _rerank_with_crossencoder(self, query: str, candidates: List[FusedCandidate]) -> List[RerankResult]:
@@ -369,7 +407,13 @@ class RagPipeline:
             additional_instruction=add_instr, context=context, query=query
         )
     
+    # [MODIFIED] 增強 JSON 解析的穩健性，從字串中提取 JSON 物件
     def _safe_json_parse(self, json_str: str, default: Any = None) -> Any:
+        # Find the JSON object within the string
+        match = re.search(r'\{.*\}', json_str, re.DOTALL)
+        if match:
+            json_str = match.group(0)
+        
         try:
             return json.loads(json_str)
         except json.JSONDecodeError:
@@ -380,9 +424,10 @@ class RagPipeline:
 app = FastAPI()
 rag_pipeline: Optional[RagPipeline] = None
 
+# [MODIFIED] 將 LINE 配置集中管理並進行啟動時檢查
 class AppConfig:
-    CHANNEL_ACCESS_TOKEN = os.getenv("CHANNEL_ACCESS_TOKEN")
-    CHANNEL_SECRET = os.getenv("CHANNEL_SECRET")
+    CHANNEL_ACCESS_TOKEN = _require_env("CHANNEL_ACCESS_TOKEN")
+    CHANNEL_SECRET = _require_env("CHANNEL_SECRET")
 
 @app.on_event("startup")
 async def startup_event():
@@ -393,40 +438,53 @@ async def startup_event():
 
 @app.post("/webhook")
 async def handle_webhook(request: Request, background_tasks: BackgroundTasks):
+    # [MODIFIED] 增強簽章驗證與環境變數檢查
     signature = request.headers.get("X-Line-Signature")
+    if not signature:
+        raise HTTPException(status_code=400, detail="Missing X-Line-Signature")
+    if not AppConfig.CHANNEL_SECRET:
+        log.error("CHANNEL_SECRET is not configured.")
+        raise HTTPException(status_code=500, detail="Server configuration error")
+
     body = await request.body()
-    
-    if not hmac.compare_digest(
-        base64.b64encode(hmac.new(AppConfig.CHANNEL_SECRET.encode("utf-8"), body, hashlib.sha256).digest()),
-        signature.encode("utf-8")
-    ):
+    try:
+        hash = hmac.new(AppConfig.CHANNEL_SECRET.encode('utf-8'), body, hashlib.sha256)
+        expected_signature = base64.b64encode(hash.digest()).decode('utf-8')
+    except Exception as e:
+        log.error(f"Failed to generate signature: {e}")
+        raise HTTPException(status_code=500, detail="Signature generation error")
+
+    if not hmac.compare_digest(expected_signature, signature):
         raise HTTPException(status_code=403, detail="Invalid signature")
 
     data = json.loads(body.decode('utf-8'))
     for event in data.get("events", []):
         if event.get("type") == "message" and event.get("message", {}).get("type") == "text":
-            reply_token = event["replyToken"]
-            user_text = event["message"]["text"].strip()
-            user_id = event["source"]["userId"]
-            
-            if reply_token and user_text:
-                background_tasks.add_task(process_user_query, user_id, reply_token, user_text)
+            reply_token = event.get("replyToken")
+            user_text = event.get("message", {}).get("text", "").strip()
+            # [MODIFIED] 安全地獲取 userId，應對群組/聊天室中可能不存在的情況
+            source = event.get("source", {})
+            user_id = source.get("userId")
+
+            if reply_token and user_id and user_text:
+                # [MODIFIED] 更改回覆策略：立即回覆處理中訊息，避免 replyToken 逾時
+                line_reply(reply_token, "收到您的問題，正在查詢資料庫，請稍候...")
+                # 將耗時的任務交給背景處理，使用 push message 回覆最終答案
+                background_tasks.add_task(process_user_query, user_id, user_text)
                 
     return Response(status_code=status.HTTP_200_OK)
 
-def process_user_query(user_id: str, reply_token: str, user_text: str):
-    # [NEW] 發送 "處理中" 訊息以改善 UX
-    line_push(user_id, "收到您的問題，正在查詢資料庫，請稍候...")
-    
+# [MODIFIED] 調整函式簽名，只接收 user_id 和 text，並使用 push message
+def process_user_query(user_id: str, user_text: str):
     try:
         if rag_pipeline:
             answer = rag_pipeline.answer_question(user_text)
-            line_reply(reply_token, answer)
         else:
-            line_reply(reply_token, "系統正在啟動中，請稍後再試。")
+            answer = "系統正在啟動中，請稍後再試。"
+        line_push(user_id, answer)
     except Exception as e:
         log.error(f"背景處理 user_id={user_id} 發生錯誤: {e}", exc_info=True)
-        line_reply(reply_token, f"抱歉，處理時發生未預期的錯誤。{DISCLAIMER}")
+        line_push(user_id, f"抱歉，處理時發生未預期的錯誤。{DISCLAIMER}")
 
 def line_api_call(endpoint: str, data: Dict):
     headers = {
@@ -447,15 +505,17 @@ def line_push(user_id: str, text: str):
     messages = [{"type": "text", "text": chunk} for chunk in textwrap.wrap(text, 4800, replace_whitespace=False)[:5]]
     line_api_call("push", {"to": user_id, "messages": messages})
 
-# ---- 額外工具函式 ----
+# [MODIFIED] 改善藥名提取的正則表達式
 def extract_drug_candidates_from_query(query: str, drug_vocab: dict) -> list:
     candidates = set()
-    words = re.findall(r'[a-z0-9]+', query)
+    q_lower = query.lower()
+    # 允許藥名中包含 -, /, . 等符號
+    words = re.findall(r"[a-z0-9][a-z0-9+\-/\.]*", q_lower)
     for word in words:
         if word in drug_vocab["en"]:
             candidates.add(word)
     
-    for token in jieba.cut(query):
+    for token in jieba.cut(q_lower):
         if token in drug_vocab["zh"]:
             candidates.add(token)