Update app.py

app.py

@@ -1,53 +1,92 @@
-# ==============================================================================
-#  Text-to-SQL 智能助手 - Hugging Face CPU 最终版 v7
-#  (修复所有已知 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
-import traceback
+from functools import lru_cache
 
+# 使用 transformers 替代 sentence-transformers
 from transformers import AutoModel, AutoTokenizer
 import torch.nn.functional as F
 
-# ==================== 配置參數 ====================
-GGUF_FILENAME = "qwen2.5-coder-1.5b-sql-finetuned.q8_0.gguf"
-N_GPU_LAYERS = 0 # CPU 环境
+# ==================== 配置區 ====================
 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"
+GGUF_FILENAME = "qwen2.5-coder-1.5b-sql-finetuned.q8_0.gguf"
+
+# 添加這一行：你的原始微調模型路徑
+FINETUNED_MODEL_PATH = "Paul720810/qwen2.5-coder-1.5b-sql-finetuned"  # ← 新增這行
+
 FEW_SHOT_EXAMPLES_COUNT = 1
-DEVICE = "cpu"
+DEVICE = "cuda" if torch.cuda.is_available() 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 智能助手 v7.0 (Hugging Face CPU 版)...")
-print(f"🚀 模型: {GGUF_FILENAME}")
-print(f"💻 設備: {DEVICE} (GPU Layers: {N_GPU_LAYERS})")
+print("🤖 Text-to-SQL 系統啟動中...")
+print(f"📊 數據集: {DATASET_REPO_ID}")
+print(f"🤖 嵌入模型: {EMBED_MODEL_NAME}")
+print(f"💻 設備: {DEVICE}")
 print("=" * 60)
 
 # ==================== 工具函數 ====================
-def get_current_time(): return datetime.now().strftime('%Y-%m-%d %H:%M:%S')
+def get_current_time():
+    return datetime.now().strftime('%Y-%m-%d %H:%M:%S')
+
 def format_log(message: str, level: str = "INFO") -> str:
-    log_entry = f"[{get_current_time()}] [{level.upper()}] {message}"; print(log_entry); return log_entry
+    return f"[{get_current_time()}] [{level.upper()}] {message}"
+
 def parse_sql_from_response(response_text: str) -> Optional[str]:
-    if not response_text: return None
+    """從模型輸出提取 SQL，增強版"""
+    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()
-    return response_text if response_text.upper().startswith("SELECT") else None
+    if match:
+        return match.group(1).strip()
+
+    # 2. 找任何 ``` 包圍的內容
+    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 語句（更寬鬆的匹配）
+    match = re.search(r"(SELECT\s+.*?;)", response_text, re.DOTALL | re.IGNORECASE)
+    if match:
+        return match.group(1).strip()
+
+    # 4. 找沒有分號的 SQL
+    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 += ';'
+        return sql
+
+    # 5. 如果包含 SELECT，嘗試提取整行
+    if 'SELECT' in response_text.upper():
+        lines = response_text.split('\n')
+        for line in lines:
+            line = line.strip()
+            if line.upper().startswith('SELECT'):
+                if not line.endswith(';'):
+                    line += ';'
+                return line
+
+    return None
 
 # ==================== Text-to-SQL 核心類 ====================
 class TextToSQLSystem:
@@ -55,227 +94,759 @@ 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(f"載入嵌入模型: {embed_model_name}")
-            self.embed_tokenizer = AutoTokenizer.from_pretrained(embed_model_name)
-            self.embed_model = AutoModel.from_pretrained(embed_model_name)
-            self.schema = self._load_schema()
-            self.dataset, self.faiss_index = self._load_and_index_dataset()
-            self._load_gguf_model()
-            self._log("✅ 系統初始化完成")
+            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 模型載入成功")
+
         except Exception as e:
-            self._log(f"❌ 系統初始化過程中發生嚴重錯誤: {e}", "CRITICAL"); self._log(traceback.format_exc(), "DEBUG"); self.llm = None
+            self._log(f"❌ GGUF 載入失敗: {e}", "ERROR")
+            self._log("系統將無法生成 SQL。請檢查模型檔案或 llama-cpp-python 安裝。", "CRITICAL")
+            self.llm = None
 
-    def _log(self, message: str, level: str = "INFO"): self.log_history.append(format_log(message, level))
-    
-    def _load_gguf_model(self):
+    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 _load_transformers_model(self):
+        """使用 Transformers 載入你的微調模型"""
         try:
-            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 模型成功載入")
+            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 模型載入成功")
+
         except Exception as e:
-            self._log(f"❌ GGUF 載入失敗: {e}", "CRITICAL"); self.llm = None
+            self._log(f"❌ Transformers 載入也失敗: {e}", "ERROR")
+            self.llm = None
 
     def huggingface_api_call(self, prompt: str) -> str:
-        if self.llm is None: return ""
+        """調用 GGUF 模型，並加入詳細的原始輸出日誌"""
+        if self.llm is None:
+            self._log("模型未載入，返回 fallback SQL。", "ERROR")
+            return self._generate_fallback_sql(prompt)
+
+        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 ""
+
+        except Exception as e:
+            self._log(f"❌ 模型生成過程中發生嚴重錯誤: {e}", "CRITICAL")
+            import traceback
+            self._log(traceback.format_exc(), "DEBUG")
+            return ""
+
+    def _load_gguf_model_fallback(self, model_path):
+        """備用載入方式"""
         try:
-            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"]["text"] if output and "choices" in output and len(output["choices"]) > 0 else ""
-            self._log(f"🧠 模型原始輸出: {generated_text.strip()}", "DEBUG")
-            return generated_text.strip()
+            # 嘗試不同的參數組合
+            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}", "CRITICAL"); return ""
-    
+            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")
-            with open(schema_path, "r", encoding="utf-8") as f: schema_data = json.load(f)
-            self._log(f"📊 Schema 載入成功，包含 {len(schema_data)} 個表格。")
-            return schema_data
+            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
+
         except Exception as e:
-            self._log(f"❌ 載入 schema 失敗: {e}", "ERROR"); return {}
+            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 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()}
+
         with torch.no_grad():
             outputs = self.embed_model(**inputs)
-        return outputs.last_hidden_state.mean(dim=1).cpu()
+
+        # 使用平均池化
+        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")
-            dataset = dataset.filter(lambda ex: isinstance(ex.get("messages"), list) and len(ex["messages"]) >= 2)
+
+            # 先過濾不完整樣本，避免 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]
             self._log(f"正在編碼 {len(corpus)} 個問題...")
-            all_embeddings = torch.cat([self._encode_texts(corpus[i:i+32]) for i in range(0, len(corpus), 32)], dim=0).numpy()
+
+            # 批量編碼
+            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 索引
             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
-            
+            self._log(f"❌ 載入數據失敗: {e}", "ERROR")
+            return None, None
+
     def _identify_relevant_tables(self, question: str) -> List[str]:
+        """根據實際 Schema 識別相關表格"""
         question_lower = question.lower()
         relevant_tables = []
-        keyword_to_table = {'TSR53SampleDescription': ['客戶', '買方', '申請', '發票對象'], 'JobsInProgress': ['進行中', '買家', '申請方'], 'JobTimeline': ['時間', '完成', '創建', '實驗室'], 'TSR53Invoice': ['發票', '金額', '費用']}
+
+        # 根據實際表格的關鍵詞映射
+        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']
+        }
+
         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: return ['TSR53SampleDescription', 'JobsInProgress', 'JobTimeline']
-        return relevant_tables[:3]
+            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 內部
 
     def _format_relevant_schema(self, table_names: List[str]) -> str:
-        if not self.schema: return "No schema available.\n"
-        formatted = ""
+        """
+        生成一個簡化的、不易被模型錯誤模仿的 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']
+
+        formatted = ""
+        for table in real_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_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})")
+                    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})")
                 formatted += f"Columns: {', '.join(cols_str)}\n\n"
-        return formatted.strip()
-
-    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):
-                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'][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[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 = "\n---\n".join([f"Q: {ex['question']}\nA: ```sql\n{ex['sql']}\n```" for ex in examples]) if examples else "No examples provided."
-        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()}
+        return formatted.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, "生成成功"
+    # in class TextToSQLSystem:
 
     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
 
-        if any(kw in q_lower for kw in ['報告號碼', '報告清單', '列出報告']):
-            year_match, month_match = re.search(r'(\d{4})\s*年?', question), re.search(r'(\d{1,2})\s*月', question)
-            from_clause = "FROM JobTimeline AS jt JOIN TSR53SampleDescription AS sd ON jt.JobNo = sd.JobNo"
-            select_clause = "SELECT jt.JobNo, sd.BuyerName, jt.ReportAuthorization"
-            where, log_parts = ["jt.ReportAuthorization IS NOT NULL"], []
-            if year_match: where.append(f"strftime('%Y', jt.ReportAuthorization) = '{year_match.group(1)}'"); log_parts.append(f"{year_match.group(1)}年")
-            if month_match: where.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: where.append("sd.OverallRating = 'Fail'"); log_parts.append("Fail")
-            elif 'pass' in q_lower or '通過' in q_lower: where.append("sd.OverallRating = 'Pass'"); log_parts.append("Pass")
+        # --- 预先检测其他意图 ---
+        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)
+            from_clause = "FROM JobTimeline AS jt"
+            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 '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"
+                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"
+                where_conditions.append("sd.OverallRating = 'Pass'"); log_parts.append("Pass")
+            
             if entity_match_data:
-                op = "=" if entity_match_data["column"].endswith("ID") else "LIKE"
-                val = f"'{entity_match_data['name']}'" if op == "=" else f"'%{entity_match_data['name']}%'"
-                where.append(f"{entity_match_data['column']} {op} {val}"); log_parts.append(entity_match_data["name"])
-            final_where = "WHERE " + " AND ".join(where) if where else ""
-            log_msg = " ".join(log_parts) if log_parts else "全部"
-            self._log(f"🔄 檢測到【{log_msg} 報告列表】意圖，啟用模板。", "INFO")
-            template_sql = f"{select_clause} {from_clause} {final_where} ORDER BY jt.ReportAuthorization DESC;"
-            return self._finalize_sql(template_sql, f"模板覆寫: {log_msg}")
-
-        if '報告' in q_lower and any(kw in q_lower for kw in ['幾份', '多少', '數量', '總數']) and not 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"
+                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)
+            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:
             year_match = re.search(r'(\d{4})\s*年?', question)
-            time_cond = f"WHERE strftime('%Y', ReportAuthorization) = '{year_match.group(1)}'" if year_match else ""
-            time_log = f"{year_match.group(1)}年" if year_match else "总"
-            self._log(f"🔄 檢測到【{time_log}报告总数】意图，启用模板。", "INFO")
-            template_sql = f"SELECT COUNT(DISTINCT JobNo) AS report_count FROM JobTimeline {time_cond};"
-            return self._finalize_sql(template_sql, f"模板覆寫: {time_log}")
+            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}年"
+            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")
         
-        # Fallback to AI
-        self._log("未觸發模板，由 AI 生成...", "INFO")
         parsed_sql = parse_sql_from_response(raw_response)
-        if not parsed_sql: return None, f"無法解析SQL: {raw_response}"
-        return self._finalize_sql(parsed_sql, "AI 生成")
+        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'
+            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}'")
+        
+        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]:
-        if question in self.query_cache: self._log("⚡ 使用緩存結果"); return self.query_cache[question]
+        """處理使用者問題 (V2 / 最終版)"""
+        # 檢查緩存
+        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)
-        self._log(f"📏 Prompt 長度: {len(prompt)} 字符")
+
+        # 3. 生成 AI 回應
         self._log("🧠 開始生成 AI 回應...")
         response = self.huggingface_api_call(prompt)
+
+        # 4. **新的核心步驟**: 呼叫決策引擎來生成最終 SQL
         final_sql, status_message = self._validate_and_fix_sql(question, response)
-        result = (final_sql, status_message) if final_sql else (status_message, "生成失敗")
+
+        if final_sql:
+            result = (final_sql, status_message)
+        else:
+            result = (status_message, "生成失敗")
+
+        # 緩存結果
         self.query_cache[question] = result
         return result
 
 # ==================== Gradio 介面 ====================
 text_to_sql_system = TextToSQLSystem()
+
 def process_query(q: str):
-    if not q.strip(): return "", "等待輸入", "請輸入問題"
-    if text_to_sql_system.llm is None: return "模型未能成功載入...", "模型載入失敗", "\n".join(text_to_sql_system.log_history)
+    if not q.strip():
+        return "", "等待輸入", "請輸入問題"
+
     sql, status = text_to_sql_system.process_question(q)
-    return sql, status, "\n".join(text_to_sql_system.log_history[-15:])
+    logs = "\n".join(text_to_sql_system.log_history[-10:])  # 只顯示最後10條日誌
+
+    return sql, status, logs
+
+# 範例問題
+examples = [
+    "2024年每月完成多少份報告？",
+    "統計各種評級(Pass/Fail)的分布情況",
+    "找出總金額最高的10個工作單",
+    "哪些客戶的工作單數量最多？",
+    "A組昨天完成了多少個測試項目？"
+]
 
-examples = ["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("# ⚡ Text-to-SQL 智能助手")
+    gr.Markdown("輸入自然語言問題，自動生成SQL查詢語句")
+
     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=10, label="日誌", interactive=False)
-    gr.Examples(examples=examples, inputs=inp, label="💡 點擊試用範例問題")
+
+        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="💡 點擊試用範例問題"
+    )
+
+    # 綁定事件
     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=True)
+    demo.launch(
+        server_name="0.0.0.0",
+        server_port=7860,
+        share=False
+    )