Update app.py

app.py

@@ -1,39 +1,44 @@
+# ==============================================================================
+#  Text-to-SQL 智能助手 - Hugging Face CPU 最终版 v6
+#  (融合模板引擎 + 强化 Prompt + 修复所有 Bug)
+# ==============================================================================
 import gradio as gr
 import os
 import re
 import json
 import torch
 import numpy as np
+import gc
+import tempfile
 from datetime import datetime
 from datasets import load_dataset
 from huggingface_hub import hf_hub_download
 from llama_cpp import Llama
 from typing import List, Dict, Tuple, Optional
 import faiss
-from functools import lru_cache
+import traceback
 
-# 使用 transformers 替代 sentence-transformers
 from transformers import AutoModel, AutoTokenizer
 import torch.nn.functional as F
 
-# ==================== 配置區 ====================
-DATASET_REPO_ID = "Paul720810/Text-to-SQL-Softline"
-GGUF_REPO_ID = "Paul720810/gguf-models"
-#GGUF_FILENAME = "qwen2.5-coder-1.5b-sql-finetuned.q4_k_m.gguf"
+# ==================== 配置參數 ====================
+# --- Hugging Face CPU 部署配置 ---
 GGUF_FILENAME = "qwen2.5-coder-1.5b-sql-finetuned.q8_0.gguf"
+N_GPU_LAYERS = 0 # 在 Hugging Face CPU 环境下设置为 0
 
-# 添加這一行：你的原始微調模型路徑
-FINETUNED_MODEL_PATH = "Paul720810/qwen2.5-coder-1.5b-sql-finetuned"  # ← 新增這行
-
+DATASET_REPO_ID = "Paul720810/Text-to-SQL-Softline"
+GGUF_REPO_ID = "Paul720810/gguf-models"
 FEW_SHOT_EXAMPLES_COUNT = 1
-DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
+DEVICE = "cuda" if torch.cuda.is_available() and N_GPU_LAYERS != 0 else "cpu"
 EMBED_MODEL_NAME = "sentence-transformers/all-MiniLM-L6-v2"
 
+TEMP_DIR = tempfile.gettempdir()
+os.makedirs(os.path.join(TEMP_DIR, 'text_to_sql_cache'), exist_ok=True)
+
 print("=" * 60)
-print("🤖 Text-to-SQL 系統啟動中...")
-print(f"📊 數據集: {DATASET_REPO_ID}")
-print(f"🤖 嵌入模型: {EMBED_MODEL_NAME}")
-print(f"💻 設備: {DEVICE}")
+print("🤖 Text-to-SQL 智能助手 v6.0 (Hugging Face CPU 版)...")
+print(f"🚀 模型: {GGUF_FILENAME}")
+print(f"💻 設備: {DEVICE} (GPU Layers: {N_GPU_LAYERS})")
 print("=" * 60)
 
 # ==================== 工具函數 ====================
@@ -41,51 +46,32 @@ def get_current_time():
     return datetime.now().strftime('%Y-%m-%d %H:%M:%S')
 
 def format_log(message: str, level: str = "INFO") -> str:
-    return f"[{get_current_time()}] [{level.upper()}] {message}"
+    log_entry = f"[{get_current_time()}] [{level.upper()}] {message}"
+    print(log_entry)
+    return log_entry
 
 def parse_sql_from_response(response_text: str) -> Optional[str]:
-    """從模型輸出提取 SQL，增強版"""
-    if not response_text:
-        return None
-
-    # 清理回應文本
+    if not response_text: return None
     response_text = response_text.strip()
-
-    # 1. 先找 ```sql ... ```
     match = re.search(r"```sql\s*\n(.*?)\n```", response_text, re.DOTALL | re.IGNORECASE)
-    if match:
-        return match.group(1).strip()
-
-    # 2. 找任何 ``` 包圍的內容
+    if match: return match.group(1).strip()
     match = re.search(r"```\s*\n?(.*?)\n?```", response_text, re.DOTALL)
     if match:
         sql_candidate = match.group(1).strip()
-        if sql_candidate.upper().startswith('SELECT'):
-            return sql_candidate
-
-    # 3. 找 SQL 語句（更寬鬆的匹配）
+        if sql_candidate.upper().startswith('SELECT'): return sql_candidate
     match = re.search(r"(SELECT\s+.*?;)", response_text, re.DOTALL | re.IGNORECASE)
-    if match:
-        return match.group(1).strip()
-
-    # 4. 找沒有分號的 SQL
+    if match: return match.group(1).strip()
     match = re.search(r"(SELECT\s+.*?)(?=\n\n|\n```|$|\n[^,\s])", response_text, re.DOTALL | re.IGNORECASE)
     if match:
         sql = match.group(1).strip()
-        if not sql.endswith(';'):
-            sql += ';'
+        if not sql.endswith(';'): sql += ';'
         return sql
-
-    # 5. 如果包含 SELECT，嘗試提取整行
     if 'SELECT' in response_text.upper():
-        lines = response_text.split('\n')
-        for line in lines:
+        for line in response_text.split('\n'):
             line = line.strip()
             if line.upper().startswith('SELECT'):
-                if not line.endswith(';'):
-                    line += ';'
+                if not line.endswith(';'): line += ';'
                 return line
-
     return None
 
 # ==================== Text-to-SQL 核心類 ====================
@@ -94,445 +80,179 @@ class TextToSQLSystem:
         self.log_history = []
         self._log("初始化系統...")
         self.query_cache = {}
-
-        # 1. 載入嵌入模型
-        self._log(f"載入嵌入模型: {embed_model_name}")
-        self.embed_tokenizer = AutoTokenizer.from_pretrained(embed_model_name)
-        self.embed_model = AutoModel.from_pretrained(embed_model_name)
-        if DEVICE == "cuda":
-            self.embed_model = self.embed_model.cuda()
-
-        # 2. 載入數據庫結構
-        self.schema = self._load_schema()
-
-        # 3. 載入數據集並建立索引
-        self.dataset, self.faiss_index = self._load_and_index_dataset()
-
-        # 4. 載入 GGUF 模型（添加錯誤處理）
-        self._load_gguf_model()
-
-        self._log("✅ 系統初始化完成")
-        # 載入數據庫結構
-        self.schema = self._load_schema()
-
-        # 暫時添加：打印 schema 信息
-        if self.schema:
-            print("=" * 50)
-            print("數據庫 Schema 信息:")
-            for table_name, columns in self.schema.items():
-                print(f"\n表格: {table_name}")
-                print(f"欄位數: {len(columns)}")
-                print("欄位列表:")
-                for col in columns[:5]:  # 只顯示前5個
-                    print(f"  - {col['name']} ({col['type']})")
-            print("=" * 50)
-
-    # in class TextToSQLSystem:
-
-    def _load_gguf_model(self):
-        """載入 GGUF 模型，使用更穩定、簡潔的參數"""
         try:
-            self._log("載入 GGUF 模型 (使用穩定性參數)...")
-            model_path = hf_hub_download(
-                repo_id=GGUF_REPO_ID,
-                filename=GGUF_FILENAME,
-                repo_type="dataset"
-            )
-
-            # 使用一組更基礎、更穩定的參數來載入模型
-            self.llm = Llama(
-                model_path=model_path,
-                n_ctx=2048,      # 將上下文增加到 2048 以確保 Prompt 不會超長
-                n_threads=4,     # 保持 4 線程
-                n_batch=512,     # 建議值
-                verbose=False,   # 設為 False 避免 llama.cpp 本身的日誌干擾
-                n_gpu_layers=0   # 確認在 CPU 上運行
-            )
-
-            # 簡單測試模型是否能回應
-            self.llm("你好", max_tokens=3)
-            self._log("✅ GGUF 模型載入成功")
-
+            self._log(f"載入嵌入模型: {embed_model_name}")
+            self.embed_tokenizer = AutoTokenizer.from_pretrained(embed_model_name)
+            self.embed_model = AutoModel.from_pretrained(embed_model_name)
+            if DEVICE == "cuda":
+                self.embed_model.to(DEVICE)
+
+            self.schema = self._load_schema()
+            self.dataset, self.faiss_index = self._load_and_index_dataset()
+            self._load_gguf_model()
+            self._log("✅ 系統初始化完成")
         except Exception as e:
-            self._log(f"❌ GGUF 載入失敗: {e}", "ERROR")
-            self._log("系統將無法生成 SQL。請檢查模型檔案或 llama-cpp-python 安裝。", "CRITICAL")
+            self._log(f"❌ 系統初始化過程中發生嚴重錯誤: {e}", "CRITICAL")
+            self._log(traceback.format_exc(), "DEBUG")
             self.llm = None
 
-    def _try_gguf_loading(self):
-        """嘗試載入 GGUF"""
-        try:
-            model_path = hf_hub_download(
-                repo_id=GGUF_REPO_ID,
-                filename=GGUF_FILENAME,
-                repo_type="dataset"
-            )
-
-            self.llm = Llama(
-                model_path=model_path,
-                n_ctx=512,
-                n_threads=4,
-                verbose=False,
-                n_gpu_layers=0
-            )
-
-            # 測試生成
-            test_result = self.llm("SELECT", max_tokens=5)
-            self._log("✅ GGUF 模型載入成功")
-            return True
-
-        except Exception as e:
-            self._log(f"GGUF 載入失敗: {e}", "WARNING")
-            return False
+    def _log(self, message: str, level: str = "INFO"):
+        self.log_history.append(format_log(message, level))
 
-    def _load_transformers_model(self):
-        """使用 Transformers 載入你的微調模型"""
+    def _load_gguf_model(self):
         try:
-            from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline
-            import torch
-
-            self._log(f"載入 Transformers 模型: {FINETUNED_MODEL_PATH}")
-
-            # 載入你的微調模型
-            self.transformers_tokenizer = AutoTokenizer.from_pretrained(FINETUNED_MODEL_PATH)
-            self.transformers_model = AutoModelForCausalLM.from_pretrained(
-                FINETUNED_MODEL_PATH,
-                torch_dtype=torch.float32,  # CPU 使用 float32
-                device_map="cpu",           # 強制使用 CPU
-                trust_remote_code=True      # Qwen 模型可能需要
-            )
-
-            # 創建生成管道
-            self.generation_pipeline = pipeline(
-                "text-generation",
-                model=self.transformers_model,
-                tokenizer=self.transformers_tokenizer,
-                device=-1,  # CPU
-                max_length=512,
-                do_sample=True,
-                temperature=0.1,
-                top_p=0.9,
-                pad_token_id=self.transformers_tokenizer.eos_token_id
-            )
-
-            self.llm = "transformers"  # 標記使用 transformers
-            self._log("✅ Transformers 模型載入成功")
-
+            model_path = hf_hub_download(repo_id=GGUF_REPO_ID, filename=GGUF_FILENAME, repo_type="dataset", cache_dir=TEMP_DIR)
+            self._log(f"模型路徑: {model_path}")
+            self._log(f"載入 GGUF 模型 (GPU Layers: {N_GPU_LAYERS})...")
+            self.llm = Llama(model_path=model_path, n_ctx=2048, n_threads=4, n_batch=512, verbose=False, n_gpu_layers=N_GPU_LAYERS)
+            self._log("✅ GGUF 模型成功載入")
         except Exception as e:
-            self._log(f"❌ Transformers 載入也失敗: {e}", "ERROR")
+            self._log(f"❌ GGUF 載入失敗: {e}", "CRITICAL")
             self.llm = None
 
     def huggingface_api_call(self, prompt: str) -> str:
-        """調用 GGUF 模型，並加入詳細的原始輸出日誌"""
-        if self.llm is None:
-            self._log("模型未載入，返回 fallback SQL。", "ERROR")
-            return self._generate_fallback_sql(prompt)
-
+        if self.llm is None: return ""
         try:
-            output = self.llm(
-                prompt,
-                max_tokens=150,
-                temperature=0.1,
-                top_p=0.9,
-                echo=False,
-                # --- 將 stop 參數加回來 ---
-                stop=["```", ";", "\n\n", "</s>"],
-            )
-
-            self._log(f"🧠 模型原始輸出 (Raw Output): {output}", "DEBUG")
-
-            if output and "choices" in output and len(output["choices"]) > 0:
-                generated_text = output["choices"][0]["text"]
-                self._log(f"📝 提取出的生成文本: {generated_text.strip()}", "DEBUG")
-                return generated_text.strip()
-            else:
-                self._log("❌ 模型的原始輸出格式不正確或為空。", "ERROR")
-                return ""
-
+            output = self.llm(prompt, max_tokens=150, temperature=0.1, top_p=0.9, echo=False, stop=["```", ";", "\n\n", "</s>", "###", "Q:"], repeat_penalty=1.1)
+            generated_text = output["choices"][0]["text"] if output and "choices" in output and len(output["choices"]) > 0 else ""
+            self._log(f"🧠 模型原始輸出: {generated_text.strip()}", "DEBUG")
+            return generated_text.strip()
         except Exception as e:
-            self._log(f"❌ 模型生成過程中發生嚴重錯誤: {e}", "CRITICAL")
-            import traceback
-            self._log(traceback.format_exc(), "DEBUG")
+            self._log(f"❌ 模型生成錯誤: {e}", "CRITICAL")
             return ""
-
-    def _load_gguf_model_fallback(self, model_path):
-        """備用載入方式"""
-        try:
-            # 嘗試不同的參數組合
-            self.llm = Llama(
-                model_path=model_path,
-                n_ctx=512,  # 更小的上下文
-                n_threads=4,
-                n_batch=128,
-                vocab_only=False,
-                use_mmap=True,
-                use_mlock=False,
-                verbose=True
-            )
-            self._log("✅ 備用方式載入成功")
-        except Exception as e:
-            self._log(f"❌ 備用方式也失敗: {e}", "ERROR")
-            self.llm = None
-
-    def _log(self, message: str, level: str = "INFO"):
-        self.log_history.append(format_log(message, level))
-        print(format_log(message, level))
-
+    
     def _load_schema(self) -> Dict:
-        """載入數據庫結構"""
         try:
-            schema_path = hf_hub_download(
-                repo_id=DATASET_REPO_ID,
-                filename="sqlite_schema_FULL.json",
-                repo_type="dataset"
-            )
+            schema_path = hf_hub_download(repo_id=DATASET_REPO_ID, filename="sqlite_schema_FULL.json", repo_type="dataset")
             with open(schema_path, "r", encoding="utf-8") as f:
                 schema_data = json.load(f)
-
-                # 添加調試信息
-                self._log(f"📊 Schema 載入成功，包含 {len(schema_data)} 個表格:")
-                for table_name, columns in schema_data.items():
-                    self._log(f"  - {table_name}: {len(columns)} 個欄位")
-                    # 顯示前3個欄位作為範例
-                    sample_cols = [col['name'] for col in columns[:3]]
-                    self._log(f"    範例欄位: {', '.join(sample_cols)}")
-
-                self._log("✅ 數據庫結構載入完成")
-                return schema_data
-
+            self._log(f"📊 Schema 載入成功，包含 {len(schema_data)} 個表格。")
+            return schema_data
         except Exception as e:
             self._log(f"❌ 載入 schema 失敗: {e}", "ERROR")
             return {}
 
-    # 也可以添加一個方法來檢查生成的 SQL 是否使用了正確的表格和欄位
-    def _analyze_sql_correctness(self, sql: str) -> Dict:
-        """分析 SQL 的正確性"""
-        analysis = {
-            'valid_tables': [],
-            'invalid_tables': [],
-            'valid_columns': [],
-            'invalid_columns': [],
-            'suggestions': []
-        }
-
-        if not self.schema:
-            return analysis
-
-        # 提取 SQL 中的表格名稱
-        table_pattern = r'FROM\s+(\w+)|JOIN\s+(\w+)'
-        table_matches = re.findall(table_pattern, sql, re.IGNORECASE)
-        used_tables = [match[0] or match[1] for match in table_matches]
-
-        # 檢查表格是否存在
-        valid_tables = list(self.schema.keys())
-        for table in used_tables:
-            if table in valid_tables:
-                analysis['valid_tables'].append(table)
-            else:
-                analysis['invalid_tables'].append(table)
-                # 尋找相似的表格名稱
-                for valid_table in valid_tables:
-                    if table.lower() in valid_table.lower() or valid_table.lower() in table.lower():
-                        analysis['suggestions'].append(f"{table} -> {valid_table}")
-
-        # 提取欄位名稱（簡單版本）
-        column_pattern = r'SELECT\s+(.*?)\s+FROM|WHERE\s+(\w+)\s*[=<>]|GROUP BY\s+(\w+)|ORDER BY\s+(\w+)'
-        column_matches = re.findall(column_pattern, sql, re.IGNORECASE)
-
-        return analysis
-
     def _encode_texts(self, texts):
-        """編碼文本為嵌入向量"""
-        if isinstance(texts, str):
-            texts = [texts]
-
-        inputs = self.embed_tokenizer(texts, padding=True, truncation=True,
-                                    return_tensors="pt", max_length=512)
-        if DEVICE == "cuda":
-            inputs = {k: v.cuda() for k, v in inputs.items()}
-
+        if isinstance(texts, str): texts = [texts]
+        inputs = self.embed_tokenizer(texts, padding=True, truncation=True, return_tensors="pt", max_length=512).to(DEVICE)
         with torch.no_grad():
             outputs = self.embed_model(**inputs)
-
-        # 使用平均池化
         embeddings = outputs.last_hidden_state.mean(dim=1)
         return embeddings.cpu()
 
     def _load_and_index_dataset(self):
-        """載入數據集並建立 FAISS 索引"""
         try:
             dataset = load_dataset(DATASET_REPO_ID, data_files="training_data.jsonl", split="train")
-
-            # 先過濾不完整樣本，避免 messages 長度不足導致索引或檢索報錯
-            try:
-                original_count = len(dataset)
-            except Exception:
-                original_count = None
-
-            dataset = dataset.filter(
-                lambda ex: isinstance(ex.get("messages"), list)
-                and len(ex["messages"]) >= 2
-                and all(
-                    isinstance(m.get("content"), str) and m.get("content") and m["content"].strip()
-                    for m in ex["messages"][:2]
-                )
-            )
-
-            if original_count is not None:
-                self._log(
-                    f"資料集清理: 原始 {original_count} 筆, 過濾後 {len(dataset)} 筆, 移除 {original_count - len(dataset)} 筆"
-                )
-
-            if len(dataset) == 0:
-                self._log("清理後資料集為空，無法建立索引。", "ERROR")
-                return None, None
-
-            corpus = [item['messages'][0]['content'] for item in dataset]
+            dataset = dataset.filter(lambda ex: isinstance(ex.get("messages"), list) and len(ex["messages"]) >= 2)
+            corpus = [item['messages']['content'] for item in dataset]
             self._log(f"正在編碼 {len(corpus)} 個問題...")
-
-            # 批量編碼
-            embeddings_list = []
-            batch_size = 32
-
-            for i in range(0, len(corpus), batch_size):
-                batch_texts = corpus[i:i+batch_size]
-                batch_embeddings = self._encode_texts(batch_texts)
-                embeddings_list.append(batch_embeddings)
-                self._log(f"已編碼 {min(i+batch_size, len(corpus))}/{len(corpus)}")
-
-            all_embeddings = torch.cat(embeddings_list, dim=0).numpy()
-
-            # 建立 FAISS 索引
+            all_embeddings = torch.cat([self._encode_texts(corpus[i:i+32]) for i in range(0, len(corpus), 32)], dim=0).numpy()
             index = faiss.IndexFlatIP(all_embeddings.shape[1])
             index.add(all_embeddings.astype('float32'))
-
             self._log("✅ 向量索引建立完成")
             return dataset, index
-
         except Exception as e:
             self._log(f"❌ 載入數據失敗: {e}", "ERROR")
+            self._log(traceback.format_exc(), "DEBUG")
             return None, None
-
+            
     def _identify_relevant_tables(self, question: str) -> List[str]:
-        """根據實際 Schema 識別相關表格"""
         question_lower = question.lower()
         relevant_tables = []
-
-        # 根據實際表格的關鍵詞映射
-        keyword_to_table = {
-            'TSR53SampleDescription': ['客戶', '買方', '申請', '發票對象', 'customer', 'invoice', 'sample'],
-            'JobsInProgress': ['進行中', '買家', '申請方', 'buyer', 'applicant', 'progress', '工作狀態'],
-            'JobTimeline': ['時間', '完成', '創建', '實驗室', 'timeline', 'creation', 'lab'],
-            'TSR53Invoice': ['發票', '金額', '費用', 'invoice', 'credit', 'amount'],
-            'JobEventsLog': ['事件', '操作', '用戶', 'event', 'log', 'user'],
-            'calendar_days': ['工作日', '假期', 'workday', 'holiday', 'calendar']
-        }
-
+        keyword_to_table = {'TSR53SampleDescription': ['客戶', '買方', '申請', '發票對象'], 'JobsInProgress': ['進行中', '買家', '申請方'], 'JobTimeline': ['時間', '完成', '創建', '實驗室'], 'TSR53Invoice': ['發票', '金額', '費用']}
         for table, keywords in keyword_to_table.items():
-            if any(keyword in question_lower for keyword in keywords):
-                relevant_tables.append(table)
-
-        # 預設重要表格
-        if not relevant_tables:
-            if any(word in question_lower for word in ['客戶', '買家', '申請', '工作單', '數量']):
-                return ['TSR53SampleDescription', 'JobsInProgress']
-            else:
-                return ['JobTimeline', 'TSR53SampleDescription']
-
-        return relevant_tables[:3]  # 最多返回3個相關表格
-
-    # 請將這整個函數複製到您的 TextToSQLSystem class 內部
+            if any(keyword in question_lower for keyword in keywords): relevant_tables.append(table)
+        if not relevant_tables: return ['TSR53SampleDescription', 'JobsInProgress', 'JobTimeline']
+        return relevant_tables[:3]
 
     def _format_relevant_schema(self, table_names: List[str]) -> str:
-        """
-        生成一個簡化的、不易被模型錯誤模仿的 Schema 字符串。
-        """
-        if not self.schema:
-            return "No schema available.\n"
-
-        actual_table_names_map = {name.lower(): name for name in self.schema.keys()}
-        real_table_names = []
-        for table in table_names:
-            actual_name = actual_table_names_map.get(table.lower())
-            if actual_name:
-                real_table_names.append(actual_name)
-            elif table in self.schema:
-                real_table_names.append(table)
-
-        if not real_table_names:
-            self._log("未識別到相關表格，使用預設核心表格。", "WARNING")
-            real_table_names = ['TSR53SampleDescription', 'JobTimeline', 'JobsInProgress']
-
+        if not self.schema: return "No schema available.\n"
         formatted = ""
-        for table in real_table_names:
+        for table in table_names:
             if table in self.schema:
-                # 使用簡單的 "Table: ..." 和 "Columns: ..." 格式
                 formatted += f"Table: {table}\n"
                 cols_str = []
-                # 只顯示前 10 個關鍵欄位
                 for col in self.schema[table][:10]:
-                    col_name = col['name']
-                    col_type = col['type']
-                    col_desc = col.get('description', '').replace('\n', ' ')
-                    # 將描述信息放在括號裡
-                    if col_desc:
-                        cols_str.append(f"{col_name} ({col_type}, {col_desc})")
-                    else:
-                        cols_str.append(f"{col_name} ({col_type})")
+                    col_name, col_type, col_desc = col['name'], col['type'], col.get('description', '').replace('\n', ' ')
+                    if col_desc: cols_str.append(f"{col_name} ({col_type}, {col_desc})")
+                    else: cols_str.append(f"{col_name} ({col_type})")
                 formatted += f"Columns: {', '.join(cols_str)}\n\n"
-
         return formatted.strip()
 
-    # in class TextToSQLSystem:
+    def find_most_similar(self, question: str, top_k: int) -> List[Dict]:
+        if self.faiss_index is None: return []
+        try:
+            q_embedding = self._encode_texts([question]).numpy().astype('float32')
+            distances, indices = self.faiss_index.search(q_embedding, min(top_k + 2, len(self.dataset)))
+            results, seen_questions = [], set()
+            for i, idx in enumerate(indices[0]):
+                if len(results) >= top_k: break
+                idx = int(idx)
+                if idx >= len(self.dataset): continue
+                item = self.dataset[idx]
+                if not (isinstance(item.get('messages'), list) and len(item['messages']) >= 2): continue
+                q_content = (item['messages']['content'] or '').strip()
+                a_content = (item['messages'].get('content') or '').strip()
+                if not q_content or not a_content: continue
+                clean_q = re.sub(r"以下是一個SQL查詢任務：\s*指令:\s*", "", q_content).strip()
+                if clean_q in seen_questions: continue
+                seen_questions.add(clean_q)
+                sql = parse_sql_from_response(a_content) or "無法解析範例SQL"
+                results.append({"similarity": float(distances[0][i]), "question": clean_q, "sql": sql})
+            return results
+        except Exception as e:
+            self._log(f"❌ 檢索失敗: {e}", "ERROR")
+            return []
+
+    def _build_prompt(self, user_q: str, examples: List[Dict]) -> str:
+        schema_str = self._format_relevant_schema(self._identify_relevant_tables(user_q))
+        example_str = ""
+        if examples:
+            example_prompts = [f"Q: {ex['question']}\nA: ```sql\n{ex['sql']}\n```" for ex in examples]
+            example_str = "\n---\n".join(example_prompts)
+        prompt = f"""You are an expert SQLite programmer. Your task is to generate a SQL query based on the database schema and a user's question.
+
+## Database Schema
+{schema_str.strip()}
+
+## Examples
+{example_str.strip()}
+
+## Task
+Based on the schema and examples, generate the SQL query for the following question.
+Q: {user_q}
+A: ```sql
+"""
+        return prompt
+        
+    def _finalize_sql(self, sql: str, log_message: str) -> Tuple[str, str]:
+        final_sql = re.sub(r'\s+', ' ', sql.strip())
+        if not final_sql.endswith(';'): final_sql += ';'
+        self._log(f"✅ SQL 已生成 ({log_message})", "INFO")
+        self._log(f"  - 最終 SQL: {final_sql}", "DEBUG")
+        return final_sql, "生成成功"
 
     def _validate_and_fix_sql(self, question: str, raw_response: str) -> Tuple[Optional[str], str]:
-        """
-        (V23 / 统一实体识别版)
-        一個全面、多層次的 SQL 驗證與生成引擎。
-        引入了全新的、统一的实体识别引擎，能够准确解析 "买家 Gap", "c0761n", 
-        "买家ID c0761n" 等多种复杂的实体提问模式。
-        """
         q_lower = question.lower()
-        
-        # ==============================================================================
-        #  第一層：高價值意圖識別與模板覆寫 (Intent Recognition & Templating)
-        # ==============================================================================
-        
-        # --- **全新的统一实体识别引擎** ---
         entity_match_data = None
-        
-        # 定义多种识别模式，【优先级从高到低】
         entity_patterns = [
-            # 模式1: 匹配 "类型 + ID" (e.g., "买家ID C0761N") - 最高优先级
             {'pattern': r"(买家|buyer)\s*(?:id|代號|代码)\s*'\"?\b([A-Z]\d{4}[A-Z])\b'\"?", 'column': 'sd.BuyerID', 'type': '买家ID'},
             {'pattern': r"(申请方|申请厂商|applicant)\s*(?:id|代號|代码)\s*'\"?\b([A-Z]\d{4}[A-Z])\b'\"?", 'column': 'sd.ApplicantID', 'type': '申请方ID'},
             {'pattern': r"(付款方|付款厂商|invoiceto)\s*(?:id|代號|代码)\s*'\"?\b([A-Z]\d{4}[A-Z])\b'\"?", 'column': 'sd.InvoiceToID', 'type': '付款方ID'},
             {'pattern': r"(代理商|agent)\s*(?:id|代號|代码)\s*'\"?\b([A-Z]\d{4}[A-Z])\b'\"?", 'column': 'sd.AgentID', 'type': '代理商ID'},
-            
-            # 模式2: 匹配 "类型 + 名称" (e.g., "买家 Gap")
             {'pattern': r"(买家|buyer|客戶)\s*'\"?([a-zA-Z0-9&.\s-]+?)(?:\s*的|\s+|$|有)", 'column': 'sd.BuyerName', 'type': '买家'},
             {'pattern': r"(申请方|申请厂商|applicant)\s*'\"?([a-zA-Z0-9&.\s-]+?)(?:\s*的|\s+|$|有)", 'column': 'sd.ApplicantName', 'type': '申请方'},
             {'pattern': r"(付款方|付款厂商|invoiceto)\s*'\"?([a-zA-Z0-9&.\s-]+?)(?:\s*的|\s+|$|有)", 'column': 'sd.InvoiceToName', 'type': '付款方'},
             {'pattern': r"(代理商|agent)\s*'\"?([a-zA-Z0-9&.\s-]+?)(?:\s*的|\s+|$|有)", 'column': 'sd.AgentName', 'type': '代理商'},
-            
-            # 模式3: 单独匹配一个 ID (e.g., "c0761n") - 较低优先级
             {'pattern': r"\b([A-Z]\d{4}[A-Z])\b", 'column': 'sd.ApplicantID', 'type': 'ID'}
         ]
-        
         for p in entity_patterns:
             match = re.search(p['pattern'], question, re.IGNORECASE)
             if match:
                 entity_value = match.group(2) if len(match.groups()) > 1 else match.group(1)
-                entity_match_data = {
-                    "type": p['type'], 
-                    "name": entity_value.strip().upper(),
-                    "column": p['column']
-                }
+                entity_match_data = {"type": p['type'], "name": entity_value.strip().upper(), "column": p['column']}
                 break
 
-        # --- 预先检测其他意图 ---
-        job_no_match = re.search(r"(?:工單|jobno)\s*'\"?([A-Z]{2,3}\d+)'\"?", question, re.IGNORECASE)
-        
-        # --- 判断逻辑: 依优先级进入对应的模板 ---
         if any(kw in q_lower for kw in ['報告號碼', '報告清單', '列出報告', 'report number', 'list of reports']):
             year_match = re.search(r'(\d{4})\s*年?', question)
             month_match = re.search(r'(\d{1,2})\s*月', question)
@@ -540,259 +260,71 @@ class TextToSQLSystem:
             select_clause = "SELECT jt.JobNo, jt.ReportAuthorization"
             where_conditions = ["jt.ReportAuthorization IS NOT NULL"]
             log_parts = []
-
-            if year_match: year = year_match.group(1); where_conditions.append(f"strftime('%Y', jt.ReportAuthorization) = '{year}'"); log_parts.append(f"{year}年")
-            if month_match: month = month_match.group(1).zfill(2); where_conditions.append(f"strftime('%m', jt.ReportAuthorization) = '{month}'"); log_parts.append(f"{month}月")
-            
+            if year_match: where_conditions.append(f"strftime('%Y', jt.ReportAuthorization) = '{year_match.group(1)}'"); log_parts.append(f"{year_match.group(1)}年")
+            if month_match: where_conditions.append(f"strftime('%m', jt.ReportAuthorization) = '{month_match.group(1).zfill(2)}'"); log_parts.append(f"{month_match.group(1)}月")
             if 'fail' in q_lower or '失敗' in q_lower:
-                if "JOIN TSR53SampleDescription" not in from_clause: from_clause = "FROM JobTimeline AS jt JOIN TSR53SampleDescription AS sd ON jt.JobNo = sd.JobNo"
+                if "JOIN TSR53SampleDescription" not in from_clause: from_clause += " JOIN TSR53SampleDescription AS sd ON jt.JobNo = sd.JobNo"
                 where_conditions.append("sd.OverallRating = 'Fail'"); log_parts.append("Fail")
             elif 'pass' in q_lower or '通過' in q_lower:
-                if "JOIN TSR53SampleDescription" not in from_clause: from_clause = "FROM JobTimeline AS jt JOIN TSR53SampleDescription AS sd ON jt.JobNo = sd.JobNo"
+                if "JOIN TSR53SampleDescription" not in from_clause: from_clause += " JOIN TSR53SampleDescription AS sd ON jt.JobNo = sd.JobNo"
                 where_conditions.append("sd.OverallRating = 'Pass'"); log_parts.append("Pass")
-            
             if entity_match_data:
                 entity_name, column_name = entity_match_data["name"], entity_match_data["column"]
-                if "JOIN TSR53SampleDescription" not in from_clause: from_clause = "FROM JobTimeline AS jt JOIN TSR53SampleDescription AS sd ON jt.JobNo = sd.JobNo"
+                if "JOIN TSR53SampleDescription" not in from_clause: from_clause += " JOIN TSR53SampleDescription AS sd ON jt.JobNo = sd.JobNo"
                 match_operator = "=" if column_name.endswith("ID") else "LIKE"
                 entity_value = f"'{entity_name}'" if match_operator == "=" else f"'%{entity_name}%'"
                 where_conditions.append(f"{column_name} {match_operator} {entity_value}")
                 log_parts.append(entity_name)
                 select_clause = "SELECT jt.JobNo, sd.BuyerName, jt.ReportAuthorization"
-
-            final_where_clause = "WHERE " + " AND ".join(where_conditions)
+            final_where_clause = "WHERE " + " AND ".join(where_conditions) if where_conditions else ""
             time_log = " ".join(log_parts) if log_parts else "全部"
             self._log(f"🔄 檢測到查詢【{time_log} 報告列表】意圖，啟用智能模板。", "INFO")
             template_sql = f"{select_clause} {from_clause} {final_where_clause} ORDER BY jt.ReportAuthorization DESC;"
             return self._finalize_sql(template_sql, f"模板覆寫: {time_log} 報告列表查詢")
 
-        # ... (此处可以继续添加 V17 版本中的其他所有 if/elif 模板)
-        elif '報告' in q_lower and any(kw in q_lower for kw in ['幾份', '多少', '數量', '總數']) and not entity_match_data:
+        if '報告' in q_lower and any(kw in q_lower for kw in ['幾份', '多少', '數量', '總數']) and not entity_match_data:
             year_match = re.search(r'(\d{4})\s*年?', question)
             time_condition, time_log = "", "總"
             if year_match:
-                year = year_match.group(1)
-                time_condition = f"WHERE ReportAuthorization IS NOT NULL AND strftime('%Y', ReportAuthorization) = '{year}'"
-                time_log = f"{year}年"
+                time_condition = f"WHERE ReportAuthorization IS NOT NULL AND strftime('%Y', ReportAuthorization) = '{year_match.group(1)}'"
+                time_log = f"{year_match.group(1)}年"
             else:
                 time_condition = "WHERE ReportAuthorization IS NOT NULL"
             self._log(f"🔄 檢測到查詢【{time_log}全局報告總數】意圖，啟用模板。", "INFO")
             template_sql = f"SELECT COUNT(DISTINCT JobNo) AS report_count FROM JobTimeline {time_condition};"
             return self._finalize_sql(template_sql, f"模板覆寫: {time_log}全局報告總數查詢")
 
-        # ==============================================================================
-        #  第二层：常规修正流程 (Fallback Corrections)
-        # ==============================================================================
         self._log("未觸發任何模板，嘗試解析並修正 AI 輸出...", "INFO")
-        
         parsed_sql = parse_sql_from_response(raw_response)
         if not parsed_sql:
-            self._log(f"❌ 未能從模型回應中解析出任何 SQL。原始回應: {raw_response}", "ERROR")
             return None, f"無法解析SQL。原始回應:\n{raw_response}"
-
-        self._log(f"📊 解析出的原始 SQL: {parsed_sql}", "DEBUG")
-        
         fixed_sql = " " + parsed_sql.strip() + " "
         fixes_applied_fallback = []
-        
         dialect_corrections = {r'YEAR\s*\(([^)]+)\)': r"strftime('%Y', \1)"}
-        for pattern, replacement in dialect_corrections.items():
-            if re.search(pattern, fixed_sql, re.IGNORECASE):
-                fixed_sql = re.sub(pattern, replacement, fixed_sql, flags=re.IGNORECASE)
-                fixes_applied_fallback.append(f"修正方言: {pattern}")
-        
-        schema_corrections = {'TSR53Report':'TSR53SampleDescription', 'TSR53InvoiceReportNo':'JobNo', 'TSR53ReportNo':'JobNo', 'TSR53InvoiceNo':'JobNo', 'TSR53InvoiceCreditNoteNo':'InvoiceCreditNoteNo', 'TSR53InvoiceLocalAmount':'LocalAmount', 'Status':'OverallRating', 'ReportStatus':'OverallRating'}
-        for wrong, correct in schema_corrections.items():
-            pattern = r'\b' + re.escape(wrong) + r'\b'
+        for p, r in dialect_corrections.items():
+            if re.search(p, fixed_sql, re.IGNORECASE):
+                fixed_sql = re.sub(p, r, fixed_sql, flags=re.IGNORECASE); fixes_applied_fallback.append(f"修正方言: {p}")
+        schema_corrections = {'TSR53Report':'TSR53SampleDescription', 'TSR53InvoiceReportNo':'JobNo', 'Status':'OverallRating'}
+        for w, c in schema_corrections.items():
+            pattern = r'\b' + re.escape(w) + r'\b'
             if re.search(pattern, fixed_sql, re.IGNORECASE):
-                fixed_sql = re.sub(pattern, correct, fixed_sql, flags=re.IGNORECASE)
-                fixes_applied_fallback.append(f"映射 Schema: '{wrong}' -> '{correct}'")
-        
+                fixed_sql = re.sub(pattern, c, fixed_sql, flags=re.IGNORECASE); fixes_applied_fallback.append(f"映射 Schema: '{w}' -> '{c}'")
         log_msg = "AI 生成並成功修正" if fixes_applied_fallback else "AI 生成且無需修正"
         return self._finalize_sql(fixed_sql, log_msg)
 
-    def _finalize_sql(self, sql: str, log_message: str) -> Tuple[str, str]:
-        """一個輔助函數，用於清理最終的SQL並記錄成功日誌。"""
-        final_sql = sql.strip()
-        if not final_sql.endswith(';'):
-            final_sql += ';'
-        final_sql = re.sub(r'\s+', ' ', final_sql).strip()
-        self._log(f"✅ SQL 已生成 ({log_message})", "INFO")
-        self._log(f"  - 最終 SQL: {final_sql}", "DEBUG")
-        return final_sql, "生成成功"
-
-    def find_most_similar(self, question: str, top_k: int) -> List[Dict]:
-        """使用 FAISS 快速檢索相似問題"""
-        if self.faiss_index is None or self.dataset is None:
-            return []
-
-        try:
-            # 編碼問題
-            q_embedding = self._encode_texts([question]).numpy().astype('float32')
-
-            # FAISS 搜索
-            distances, indices = self.faiss_index.search(q_embedding, min(top_k + 2, len(self.dataset)))
-
-            results = []
-            seen_questions = set()
-
-            for i, idx in enumerate(indices[0]):
-                if len(results) >= top_k:
-                    break
-
-                # 修復：將 numpy.int64 轉換為 Python int
-                idx = int(idx)  # ← 添加這行轉換
-
-                if idx >= len(self.dataset):  # 確保索引有效
-                    continue
-
-                item = self.dataset[idx]
-                # 防呆：若樣本不完整則跳過
-                if not isinstance(item.get('messages'), list) or len(item['messages']) < 2:
-                    continue
-                q_content = (item['messages'][0].get('content') or '').strip()
-                a_content = (item['messages'][1].get('content') or '').strip()
-                if not q_content or not a_content:
-                    continue
-
-                # 提取純淨問題
-                clean_q = re.sub(r"以下是一個SQL查詢任務：\s*指令:\s*", "", q_content).strip()
-                if clean_q in seen_questions:
-                    continue
-
-                seen_questions.add(clean_q)
-                sql = parse_sql_from_response(a_content) or "無法解析範例SQL"
-
-                results.append({
-                    "similarity": float(distances[0][i]),
-                    "question": clean_q,
-                    "sql": sql
-                })
-
-            return results
-
-        except Exception as e:
-            self._log(f"❌ 檢索失敗: {e}", "ERROR")
-            return []
-
-    # in class TextToSQLSystem:
-
-    def _build_prompt(self, user_q: str, examples: List[Dict]) -> str:
-        """
-        建立一個高度結構化、以任務為導向的提示詞，使用清晰的標題分隔符。
-        """
-        relevant_tables = self._identify_relevant_tables(user_q)
-
-        # 使用我們新的、更簡單的 schema 格式化函數
-        schema_str = self._format_relevant_schema(relevant_tables)
-
-        example_str = "No example available."
-        if examples:
-            best_example = examples[0]
-            example_str = f"Question: {best_example['question']}\nSQL:\n```sql\n{best_example['sql']}\n```"
-
-        # 使用強分隔符和清晰的標題來構建 prompt
-        prompt = f"""### INSTRUCTIONS ###
-You are a SQLite expert. Your only job is to generate a single, valid SQLite query based on the provided schema and question.
-- ONLY use the tables and columns from the schema below.
-- ALWAYS use SQLite syntax (e.g., `strftime('%Y', date_column)` for years).
-- The report completion date is the `ReportAuthorization` column in the `JobTimeline` table.
-- Your output MUST be ONLY the SQL query inside a ```sql code block.
-
-### SCHEMA ###
-{schema_str}
-
-### EXAMPLE ###
-{example_str}
-
-### TASK ###
-Generate a SQLite query for the following question.
-Question: {user_q}
-SQL:
-```sql
-"""
-        self._log(f"📏 Prompt 長度: {len(prompt)} 字符")
-        # 不再需要複雜的長度截斷邏輯，因為 schema 已經被簡化
-        return prompt
-
-
-    def _generate_fallback_sql(self, prompt: str) -> str:
-        """當模型不可用時的備用 SQL 生成"""
-        prompt_lower = prompt.lower()
-
-        # 簡單的關鍵詞匹配生成基本 SQL
-        if "統計" in prompt or "數量" in prompt or "多少" in prompt:
-            if "月" in prompt:
-                return "SELECT strftime('%Y-%m', completed_time) as month, COUNT(*) as count FROM jobtimeline GROUP BY month ORDER BY month;"
-            elif "客戶" in prompt:
-                return "SELECT applicant, COUNT(*) as count FROM tsr53sampledescription GROUP BY applicant ORDER BY count DESC;"
-            else:
-                return "SELECT COUNT(*) as total_count FROM jobtimeline WHERE completed_time IS NOT NULL;"
-
-        elif "金額" in prompt or "總額" in prompt:
-            return "SELECT SUM(amount) as total_amount FROM tsr53invoice;"
-
-        elif "評級" in prompt or "pass" in prompt_lower or "fail" in prompt_lower:
-            return "SELECT rating, COUNT(*) as count FROM tsr53sampledescription GROUP BY rating;"
-
-        else:
-            return "SELECT * FROM jobtimeline LIMIT 10;"
-
-    def _validate_model_file(self, model_path):
-        """驗證模型檔案完整性"""
-        try:
-            if not os.path.exists(model_path):
-                return False
-
-            # 檢查檔案大小（至少應該有幾MB）
-            file_size = os.path.getsize(model_path)
-            if file_size < 10 * 1024 * 1024:  # 小於 10MB 可能有問題
-                return False
-
-            # 檢查 GGUF 檔案頭部
-            with open(model_path, 'rb') as f:
-                header = f.read(8)
-                if not header.startswith(b'GGUF'):
-                    return False
-
-            return True
-        except Exception:
-            return False
-
-# in class TextToSQLSystem:
-
     def process_question(self, question: str) -> Tuple[str, str]:
-        """處理使用者問題 (V2 / 最終版)"""
-        # 檢查緩存
-        if question in self.query_cache:
-            self._log("⚡ 使用緩存結果")
-            return self.query_cache[question]
-
+        if question in self.query_cache: self._log("⚡ 使用緩存結果"); return self.query_cache[question]
         self.log_history = []
         self._log(f"⏰ 處理問題: {question}")
-
-        # 1. 檢索相似範例
-        self._log("🔍 尋找相似範例...")
         examples = self.find_most_similar(question, FEW_SHOT_EXAMPLES_COUNT)
         if examples: self._log(f"✅ 找到 {len(examples)} 個相似範例")
-
-        # 2. 建立提示詞
-        self._log("📝 建立 Prompt...")
         prompt = self._build_prompt(question, examples)
-
-        # 3. 生成 AI 回應
+        self._log(f"📏 Prompt 長度: {len(prompt)} 字符")
         self._log("🧠 開始生成 AI 回應...")
         response = self.huggingface_api_call(prompt)
-
-        # 4. **新的核心步驟**: 呼叫決策引擎來生成最終 SQL
         final_sql, status_message = self._validate_and_fix_sql(question, response)
-
-        if final_sql:
-            result = (final_sql, status_message)
-        else:
-            result = (status_message, "生成失敗")
-
-        # 緩存結果
+        if not final_sql: result = (status_message, "生成失敗")
+        else: result = (final_sql, status_message)
         self.query_cache[question] = result
         return result
 
@@ -800,53 +332,36 @@ SQL:
 text_to_sql_system = TextToSQLSystem()
 
 def process_query(q: str):
-    if not q.strip():
-        return "", "等待輸入", "請輸入問題"
-
+    if not q.strip(): return "", "等待輸入", "請輸入問題"
+    if text_to_sql_system.llm is None:
+        return "模型未能成功載入，請檢查終端日誌。", "模型載入失敗", "\n".join(text_to_sql_system.log_history)
     sql, status = text_to_sql_system.process_question(q)
-    logs = "\n".join(text_to_sql_system.log_history[-10:])  # 只顯示最後10條日誌
-
+    logs = "\n".join(text_to_sql_system.log_history[-15:])
     return sql, status, logs
 
-# 範例問題
 examples = [
-    "2024年每月完成多少份報告？",
-    "統計各種評級(Pass/Fail)的分布情況",
-    "找出總金額最高的10個工作單",
-    "哪些客戶的工作單數量最多？",
-    "A組昨天完成了多少個測試項目？"
+    "2024年7月買家 Gap 的 Fail 報告號碼",
+    "列出2023年所有失败的报告",
+    "找出总金额最高的10个工作单",
+    "哪些客户的工作单数量最多？",
+    "A組2024年完成了多少個測試項目？",
+    "2024年每月完成多少份報告？"
 ]
-
 with gr.Blocks(theme=gr.themes.Soft(), title="Text-to-SQL 智能助手") as demo:
-    gr.Markdown("# ⚡ Text-to-SQL 智能助手")
-    gr.Markdown("輸入自然語言問題，自動生成SQL查詢語句")
-
+    gr.Markdown("# ⚡ Text-to-SQL 智能助手 (终极版)")
+    gr.Markdown("融合了模板引擎和 GGUF 模型的强大版本")
     with gr.Row():
         with gr.Column(scale=2):
             inp = gr.Textbox(lines=3, label="💬 您的問題", placeholder="例如：2024年每月完成多少份報告？")
             btn = gr.Button("🚀 生成 SQL", variant="primary")
             status = gr.Textbox(label="狀態", interactive=False)
-
         with gr.Column(scale=3):
             sql_out = gr.Code(label="🤖 生成的 SQL", language="sql", lines=8)
-
     with gr.Accordion("📋 處理日誌", open=False):
-        logs = gr.Textbox(lines=8, label="日誌", interactive=False)
-
-    # 範例區
-    gr.Examples(
-        examples=examples,
-        inputs=inp,
-        label="💡 點擊試用範例問題"
-    )
-
-    # 綁定事件
+        logs = gr.Textbox(lines=10, label="日誌", interactive=False)
+    gr.Examples(examples=examples, inputs=inp, label="💡 點擊試用範例問題")
     btn.click(process_query, inputs=[inp], outputs=[sql_out, status, logs])
     inp.submit(process_query, inputs=[inp], outputs=[sql_out, status, logs])
 
 if __name__ == "__main__":
-    demo.launch(
-        server_name="0.0.0.0",
-        server_port=7860,
-        share=False
-    )
+    demo.launch(server_name="0.0.0.0", server_port=7860, share=True)