Update app.py

app.py

@@ -1,8 +1,6 @@
 """
 多策略 RAG 文件問答系統 v2 — ChromaDB + PDF/DOCX 版本（含 Telegram 推送）
-
-安裝依賴：
-    pip install gradio groq pypdf python-docx sentence-transformers numpy chromadb scikit-learn requests
+已修正：Telegram ReadTimeout，加入 retry + 指數退避 + 分離 connect/read timeout
 """
 
 from __future__ import annotations
@@ -17,6 +15,8 @@ import chromadb
 import gradio as gr
 import numpy as np
 import requests
+from requests.adapters import HTTPAdapter
+from urllib3.util.retry import Retry
 from docx import Document
 from docx.oxml.table import CT_Tbl
 from docx.oxml.text.paragraph import CT_P
@@ -32,11 +32,32 @@ from sklearn.feature_extraction.text import TfidfVectorizer
 #  Telegram 推送設定
 # ══════════════════════════════════════════════════════════
 DEFAULT_TELEGRAM_CHAT_ID = "8874400558"
-TELEGRAM_MAX_LEN = 4000  # Telegram 訊息長度限制 (4096)，留一些緩衝
+TELEGRAM_MAX_LEN = 4000
+
+
+def _make_session() -> requests.Session:
+    """建立帶有 retry 策略的 requests Session。"""
+    session = requests.Session()
+    retry = Retry(
+        total=4,                        # 最多重試 4 次
+        backoff_factor=1.5,             # 退避：1.5 / 3 / 4.5 / 6.75 秒
+        status_forcelist=[429, 500, 502, 503, 504],
+        allowed_methods=["POST"],
+        raise_on_status=False,
+    )
+    adapter = HTTPAdapter(max_retries=retry)
+    session.mount("https://", adapter)
+    session.mount("http://", adapter)
+    return session
 
 
 def send_telegram_message(text: str, chat_id: str, token: str) -> dict:
-    """將文字訊息送到 Telegram。若文字過長會自動分段傳送。"""
+    """
+    將文字訊息送到 Telegram。
+    - 分離 connect timeout（10 s）與 read timeout（30 s）
+    - 自動重試最多 4 次（指數退避）
+    - 長訊息自動分段
+    """
     if not token:
         return {"ok": False, "error": "尚未提供 Bot Token"}
     if not chat_id:
@@ -44,15 +65,52 @@ def send_telegram_message(text: str, chat_id: str, token: str) -> dict:
     if not text:
         return {"ok": False, "error": "empty text"}
 
+    # 先驗證 token 格式（避免明顯錯誤）
+    if ":" not in token or len(token) < 20:
+        return {"ok": False, "error": "Bot Token 格式不正確，應為 123456789:AAAxxxxxx"}
+
     url = f"https://api.telegram.org/bot{token}/sendMessage"
+    session = _make_session()
     results = []
+
     for i in range(0, len(text), TELEGRAM_MAX_LEN):
-        chunk = text[i:i + TELEGRAM_MAX_LEN]
-        try:
-            r = requests.post(url, data={"chat_id": chat_id, "text": chunk}, timeout=15)
-            results.append(r.json())
-        except Exception as exc:
-            results.append({"ok": False, "error": f"{type(exc).__name__}: {exc}"})
+        chunk = text[i : i + TELEGRAM_MAX_LEN]
+        payload = {
+            "chat_id": chat_id,
+            "text": chunk,
+            "parse_mode": "HTML",
+        }
+        last_exc = None
+        # 手動額外重試迴圈（配合退避），requests Retry 已涵蓋 5xx，這裡補 timeout
+        for attempt in range(4):
+            try:
+                resp = session.post(
+                    url,
+                    data=payload,
+                    timeout=(10, 30),   # (connect_timeout, read_timeout)
+                )
+                result = resp.json()
+                results.append(result)
+                last_exc = None
+                break  # 成功就跳出重試
+            except requests.exceptions.Timeout as exc:
+                last_exc = exc
+                wait = 1.5 ** attempt
+                time.sleep(wait)
+            except requests.exceptions.ConnectionError as exc:
+                last_exc = exc
+                wait = 2 ** attempt
+                time.sleep(wait)
+            except Exception as exc:
+                last_exc = exc
+                break  # 非網路錯誤不重試
+
+        if last_exc is not None:
+            results.append({
+                "ok": False,
+                "error": f"{type(last_exc).__name__}: {last_exc} （已重試 4 次）",
+            })
+
     return results[-1] if results else {"ok": False, "error": "no chunks sent"}
 
 
@@ -62,14 +120,14 @@ def send_telegram_message(text: str, chat_id: str, token: str) -> dict:
 class MultiStrategyRAG:
 
     STRATEGY_MAP = {
-        "semantic":      "1  ChromaDB 語意搜尋",
-        "tfidf":         "2  TF-IDF 關鍵詞",
-        "hybrid":        "3  混合搜尋",
-        "rerank":        "4  重新排序",
-        "multi_query":   "5  多查詢擴展",
-        "compress":      "6  上下文壓縮",
-        "parent_child":  "7  父子文檔",
-        "hyde":          "8  假設性答案 HyDE",
+        "semantic":     "1  ChromaDB 語意搜尋",
+        "tfidf":        "2  TF-IDF 關鍵詞",
+        "hybrid":       "3  混合搜尋",
+        "rerank":       "4  重新排序",
+        "multi_query":  "5  多查詢擴展",
+        "compress":     "6  上下文壓縮",
+        "parent_child": "7  父子文檔",
+        "hyde":         "8  假設性答案 HyDE",
     }
 
     def __init__(
@@ -102,12 +160,10 @@ class MultiStrategyRAG:
         self.tfidf_vectorizer: TfidfVectorizer | None = None
         self.tfidf_matrix = None
 
-    # ── API Key 管理 ─────────────────────────────────────
     def set_api_key(self, api_key: str) -> None:
         key = (api_key or "").strip()
         self.client = Groq(api_key=key) if key else None
 
-    # ── 文件載入 ─────────────────────────────────────────
     def load_document(self, file_path: str) -> str:
         try:
             path = Path(file_path)
@@ -148,7 +204,6 @@ class MultiStrategyRAG:
         except Exception as exc:
             return f"✗ 載入失敗：{type(exc).__name__}: {exc}"
 
-    # ── 文字擷取 ─────────────────────────────────────────
     def _extract_pdf(self, path: Path) -> tuple[str, str]:
         reader = PdfReader(str(path))
         parts = []
@@ -190,10 +245,9 @@ class MultiStrategyRAG:
         step = max(1, chunk_size - overlap)
         return [
             c for start in range(0, len(clean), step)
-            if (c := clean[start: start + chunk_size].strip())
+            if (c := clean[start : start + chunk_size].strip())
         ]
 
-    # ── Index 建立 ───────────────────────────────────────
     def _encode(self, texts: list[str]) -> list[list[float]]:
         return (
             self.embedding_model
@@ -232,7 +286,6 @@ class MultiStrategyRAG:
             self.child_collection.add(ids=child_ids, documents=child_docs,
                                        metadatas=child_metas, embeddings=self._encode(child_docs))
 
-    # ── 工具函式 ─────────────────────────────────────────
     def _where(self) -> dict[str, str]:
         return {"session_id": self.session_id or ""}
 
@@ -278,7 +331,6 @@ class MultiStrategyRAG:
         except Exception:
             return None
 
-    # ── 8 種策略 ──────────────────────────────────────────
     def s_semantic(self, query: str, k: int = 3) -> list[str]:
         return [r["text"] for r in self._chroma_search(query, k)]
 
@@ -290,9 +342,7 @@ class MultiStrategyRAG:
         return [self.chunks[i] for i in scores.argsort()[-k:][::-1]]
 
     def s_hybrid(self, query: str, k: int = 3) -> list[str]:
-        return self._dedupe(
-            self.s_semantic(query, k * 2) + self.s_tfidf(query, k * 2), k
-        )
+        return self._dedupe(self.s_semantic(query, k * 2) + self.s_tfidf(query, k * 2), k)
 
     def s_rerank(self, query: str, k: int = 3) -> list[str]:
         candidates = self.s_semantic(query, k * 2)
@@ -348,7 +398,6 @@ class MultiStrategyRAG:
         hypo = self._llm(prompt, max_tokens=250, temperature=0.7) or query
         return self.s_semantic(hypo, k)
 
-    # ── 策略路由 ──────────────────────────────────────────
     _FN = {
         "semantic":     s_semantic,
         "tfidf":        s_tfidf,
@@ -424,7 +473,6 @@ STRATEGY_INFO = [
     ("hyde",         "💡 HyDE",       "先生成假設答案再語意搜尋"),
 ]
 
-# 顯示用 label -> 內部 key 對照表
 STRATEGY_LABEL_TO_KEY = {label: key for key, label, _ in STRATEGY_INFO}
 STRATEGY_CHOICES = [label for _, label, _ in STRATEGY_INFO]
 STRATEGY_DESC_HTML = "<br>".join(f"<b>{label}</b> — {desc}" for _, label, desc in STRATEGY_INFO)
@@ -447,7 +495,6 @@ body, .gradio-container { background:#f5f4f1 !important; }
         font-size:11px; background:#e8f4f0; color:#2d6a4f; border:1px solid rgba(45,106,79,.2); }
 .pill-amber { background:#fdf4e3; color:#b87a1a; border-color:rgba(184,122,26,.25); }
 
-/* API Key 區塊 */
 #apikey-box {
     background: #fffbf2;
     border: 1.5px solid #f0c96a;
@@ -455,8 +502,6 @@ body, .gradio-container { background:#f5f4f1 !important; }
     padding: 12px 14px;
     margin-bottom: 8px;
 }
-
-/* Telegram 區塊 */
 #telegram-box {
     background: #eef6ff;
     border: 1.5px solid #8ec4f0;
@@ -464,8 +509,6 @@ body, .gradio-container { background:#f5f4f1 !important; }
     padding: 12px 14px;
     margin-bottom: 8px;
 }
-
-/* 策略選擇區塊 */
 #strategy-box {
     background:#fff;
     border:1.5px solid #e5e0d8;
@@ -488,9 +531,10 @@ body, .gradio-container { background:#f5f4f1 !important; }
              color:#7a6e67; font-weight:700; margin:16px 0 8px; }
 .card-box { background:#fff !important; border:1px solid #e5e0d8 !important;
             border-radius:12px !important; padding:16px !important; }
-#ask-btn { background:#2d6a4f !important; color:#fff !important; border:0 !important; border-radius:8px !important; }
+#ask-btn     { background:#2d6a4f !important; color:#fff !important; border:0 !important; border-radius:8px !important; }
 #apply-key-btn { background:#b87a1a !important; color:#fff !important; border:0 !important; border-radius:8px !important; }
 #send-tg-btn { background:#0088cc !important; color:#fff !important; border:0 !important; border-radius:8px !important; }
+#verify-tg-btn { background:#0099aa !important; color:#fff !important; border:0 !important; border-radius:8px !important; }
 """
 
 HEADER_HTML = """
@@ -525,7 +569,6 @@ def create_interface():
         rag.set_api_key(env_key)
 
     current_strategy = {"key": "semantic"}
-    # 暫存最近一次回答 / 來源片段，供 Telegram 推送使用
     last_result = {"answer": "", "source": ""}
 
     def apply_api_key(api_key: str):
@@ -552,6 +595,23 @@ def create_interface():
         last_result["source"] = source or ""
         return answer, source
 
+    def verify_bot(token_input: str, chat_id_input: str):
+        """驗證 Bot Token 與 Chat ID 是否正確，發送一則測試訊息。"""
+        token = (token_input or "").strip()
+        chat_id = (chat_id_input or "").strip() or DEFAULT_TELEGRAM_CHAT_ID
+        if not token:
+            return "⚠ 請先輸入 Bot Token"
+        result = send_telegram_message("✅ Telegram 推送測試成功！RAG 系統連線正常。", chat_id=chat_id, token=token)
+        if result.get("ok"):
+            return "✓ 測試訊息已送達，Bot Token 與 Chat ID 正確！"
+        desc = result.get("description") or result.get("error") or str(result)
+        # 給出更友善的錯誤提示
+        if "Unauthorized" in desc:
+            return "✗ Bot Token 錯誤（Unauthorized），請確認 Token 正確"
+        if "chat not found" in desc:
+            return f"✗ Chat ID「{chat_id}」找不到，請確認：\n1. 個人 ID 需先對 Bot 傳訊\n2. 群組 ID 需含負號，如 -1001234567890"
+        return f"✗ 驗證失敗：{desc}"
+
     def push_to_telegram(include_source: bool, chat_id_input: str, token_input: str):
         if not last_result["answer"]:
             return "⚠ 尚無可推送的回答，請先提問。"
@@ -562,17 +622,28 @@ def create_interface():
 
         chat_id = (chat_id_input or "").strip() or DEFAULT_TELEGRAM_CHAT_ID
 
-        text = f"📄 文件：{rag.source_name or '未知'}\n"
-        text += f"❓ 策略：{rag.STRATEGY_MAP.get(current_strategy['key'], current_strategy['key'])}\n\n"
-        text += f"💬 回答：\n{last_result['answer']}"
+        # 避免 HTML parse_mode 與純文字衝突，移除尖括號
+        answer_clean = re.sub(r"[<>]", "", last_result["answer"])
+
+        text = (
+            f"📄 文件：{rag.source_name or '未知'}\n"
+            f"❓ 策略：{rag.STRATEGY_MAP.get(current_strategy['key'], current_strategy['key'])}\n\n"
+            f"💬 回答：\n{answer_clean}"
+        )
 
         if include_source:
-            text += f"\n\n— 檢索片段 —\n{last_result['source']}"
+            source_clean = re.sub(r"[<>]", "", last_result["source"])
+            text += f"\n\n— 檢索片段 —\n{source_clean}"
 
         result = send_telegram_message(text, chat_id=chat_id, token=token)
         if result.get("ok"):
             return "✓ 已成功推送至 Telegram"
-        return f"✗ 推送失敗：{result.get('description') or result.get('error') or result}"
+        desc = result.get("description") or result.get("error") or str(result)
+        if "Unauthorized" in desc:
+            return "✗ 推送失敗：Bot Token 錯誤（Unauthorized）"
+        if "chat not found" in desc:
+            return f"✗ 推送失敗：找不到 Chat ID「{chat_id}」，請先對 Bot 傳訊或確認群組 ID"
+        return f"✗ 推送失敗：{desc}"
 
     with gr.Blocks(
         title="多策略 RAG 文件問答 v2",
@@ -588,12 +659,11 @@ def create_interface():
             # ── 左欄 ──────────────────────────────────
             with gr.Column(scale=1, min_width=320, elem_classes="card-box"):
 
-                # Step 00 · Telegram 推送（移至最上方，避免找不到）
                 gr.HTML("<div class='sec-label'>Step 00 · Telegram 推送</div>")
                 with gr.Group(elem_id="telegram-box"):
                     tg_token = gr.Textbox(
                         label="Bot Token",
-                        placeholder="例如：123456789:AAAxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
+                        placeholder="123456789:AAAxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
                         type="password",
                         lines=1,
                     )
@@ -602,17 +672,12 @@ def create_interface():
                         placeholder=f"預設：{DEFAULT_TELEGRAM_CHAT_ID}（留空使用此預設值）",
                         lines=1,
                     )
-                    tg_include_source = gr.Checkbox(
-                        label="同時推送檢索片段內容", value=False
-                    )
-                    send_tg_btn = gr.Button(
-                        "📨 推送回答到 Telegram", elem_id="send-tg-btn"
-                    )
-                    tg_status = gr.Textbox(
-                        label="推送狀態", interactive=False, lines=1
-                    )
+                    with gr.Row():
+                        verify_tg_btn = gr.Button("🔍 測試連線", elem_id="verify-tg-btn", size="sm")
+                        send_tg_btn = gr.Button("📨 推送回答", elem_id="send-tg-btn", size="sm")
+                    tg_include_source = gr.Checkbox(label="同時推送檢索片段內容", value=False)
+                    tg_status = gr.Textbox(label="推送狀態", interactive=False, lines=2)
 
-                # Step 01：Groq API Key
                 gr.HTML("<div class='sec-label'>Step 01 · Groq API Key</div>")
                 with gr.Group(elem_id="apikey-box"):
                     api_key_input = gr.Textbox(
@@ -623,9 +688,7 @@ def create_interface():
                         lines=1,
                         show_label=False,
                     )
-                    apply_key_btn = gr.Button(
-                        "套用 API Key", size="sm", elem_id="apply-key-btn"
-                    )
+                    apply_key_btn = gr.Button("套用 API Key", size="sm", elem_id="apply-key-btn")
                     api_key_status = gr.Textbox(
                         value="✓ API Key 已從環境變數載入" if env_key else "⚠ 尚未設定 API Key",
                         interactive=False,
@@ -634,13 +697,11 @@ def create_interface():
                         show_label=False,
                     )
 
-                # Step 02：上傳文件
                 gr.HTML("<div class='sec-label'>Step 02 · 上傳文件</div>")
                 file_input = gr.File(label="PDF / DOCX", file_types=[".pdf", ".docx"])
                 load_btn = gr.Button("↑ 載入文件")
                 status = gr.Textbox(label="狀態", interactive=False, lines=3)
 
-                # Step 03：RAG 策略（改用 Radio，可正常點選）
                 gr.HTML("<div class='sec-label'>Step 03 · 選擇 RAG 策略</div>")
                 with gr.Group(elem_id="strategy-box"):
                     strategy_radio = gr.Radio(
@@ -657,7 +718,6 @@ def create_interface():
                     label="目前策略",
                 )
 
-                # Step 04：參數
                 gr.HTML("<div class='sec-label'>Step 04 · 搜尋參數</div>")
                 topk = gr.Slider(minimum=1, maximum=10, value=3, step=1, label="Top-K 片段數量")
 
@@ -686,6 +746,7 @@ def create_interface():
         strategy_radio.change(fn=set_strategy, inputs=[strategy_radio], outputs=[strategy_status])
         ask_btn.click(fn=ask, inputs=[qin, topk], outputs=[ans, src])
         qin.submit(fn=ask, inputs=[qin, topk], outputs=[ans, src])
+        verify_tg_btn.click(fn=verify_bot, inputs=[tg_token, tg_chat_id], outputs=[tg_status])
         send_tg_btn.click(
             fn=push_to_telegram,
             inputs=[tg_include_source, tg_chat_id, tg_token],