Create app.py

app.py

@@ -0,0 +1,171 @@
+import gradio as gr
+import requests
+import json
+import os
+from datasets import load_dataset, Dataset
+from sentence_transformers import SentenceTransformer, util
+import torch
+from huggingface_hub import hf_hub_download
+
+# --- 配置區 ---
+# 從 Hugging Face Secrets 獲取 Token，這是最安全的方式
+HF_TOKEN = os.environ.get("HF_TOKEN") 
+# 您的 Dataset 倉庫 ID
+DATASET_REPO_ID = "Paul720810/Text-to-SQL-Softline"
+# 雲端 LLM 模型的 API URL (推薦使用 CodeLlama-34b，它更強大)
+LLM_API_URL = "https://api-inference.huggingface.co/models/codellama/CodeLlama-34b-Instruct-hf"
+
+# 相似度閾值，高於此值則直接返回答案
+SIMILARITY_THRESHOLD = 0.90 
+
+print("--- [1/5] 開始初始化應用 ---")
+
+# --- 1. 載入知識庫 ---
+try:
+    print(f"--- [2/5] 正在從 '{DATASET_REPO_ID}' 載入知識庫... ---")
+    # 載入問答範例
+    dataset = load_dataset(DATASET_REPO_ID, token=HF_TOKEN, trust_remote_code=True)
+    qa_dataset = dataset['train']
+    
+    # 載入並解析 Schema JSON
+    schema_file_path = "sqlite_schema_FULL.json"
+    hf_hub_download(repo_id=DATASET_REPO_ID, filename=schema_file_path, repo_type='dataset', local_dir='.', token=HF_TOKEN)
+    
+    with open(schema_file_path, 'r', encoding='utf-8') as f:
+        schema_data = json.load(f)
+        
+    print(f"--- > 成功載入 {len(qa_dataset)} 條問答範例和 Schema。 ---")
+except Exception as e:
+    print(f"!!! 致命錯誤: 無法載入 Dataset '{DATASET_REPO_ID}'. 請檢查：")
+    print("1. Dataset 倉庫是否設為 Public，或 HF_TOKEN 是否有讀取 Private 倉庫的權限。")
+    print("2. 倉庫中是否包含 training_data.jsonl 和 sqlite_schema_FULL.json。")
+    print(f"詳細錯誤: {e}")
+    # 如果載入失敗，則使用備用數據避免應用崩潰
+    qa_dataset = Dataset.from_dict({"question": ["示例問題"], "sql": ["SELECT 'Dataset failed to load'"]})
+    schema_data = {}
+
+# --- 2. 構建 DDL 和初始化檢索模型 ---
+def load_schema_as_ddl(schema_dict: dict) -> str:
+    ddl_string = ""
+    for table_name, columns in schema_dict.items():
+        if not isinstance(columns, list): continue
+        ddl_string += f"CREATE TABLE `{table_name}` (\n"
+        ddl_cols = [f"  `{col.get('name', '')}` {col.get('type', '')} -- {col.get('description', '')}" for col in columns]
+        ddl_string += ",\n".join(ddl_cols) + "\n);\n\n"
+    return ddl_string
+
+SCHEMA_DDL = load_schema_as_ddl(schema_data)
+
+print("--- [3/5] 正在載入句向量模型 (all-MiniLM-L6-v2)... ---")
+# 輕量級句向量模型，在 CPU 上運行極快
+embedder = SentenceTransformer('all-MiniLM-L6-v2', device='cpu')
+
+questions = [item['question'] for item in qa_dataset]
+print(f"--- [4/5] 正在為 {len(questions)} 個問題計算向量 (這可能需要幾分鐘)... ---")
+# 預先計算所有問題的向量，這是實現快速檢索的關鍵
+question_embeddings = embedder.encode(questions, convert_to_tensor=True, show_progress_bar=True)
+sql_answers = [item['sql'] for item in qa_dataset]
+print("--- > 向量計算完成！ ---")
+
+# --- 3. 混合系統核心邏輯 ---
+def get_sql_query(user_question: str):
+    if not user_question:
+        return "請輸入您的問題。", "日誌：用戶未輸入問題。"
+        
+    # 1. 向量檢索
+    question_embedding = embedder.encode(user_question, convert_to_tensor=True)
+    hits = util.semantic_search(question_embedding, question_embeddings, top_k=5)
+    hits = hits[0] # Get the hits for the first query
+    
+    most_similar_hit = hits[0]
+    similarity_score = most_similar_hit['score']
+    
+    log_message = f"檢索到最相似問題: '{questions[most_similar_hit['corpus_id']]}' (相似度: {similarity_score:.4f})"
+    
+    # 2. 如果相似度足夠高，直接返回預定義的 SQL
+    if similarity_score > SIMILARITY_THRESHOLD:
+        sql_result = sql_answers[most_similar_hit['corpus_id']]
+        log_message += f"\n相似度 > {SIMILARITY_THRESHOLD}，[模式: 直接返回]。"
+        return sql_result, log_message
+        
+    # 3. 否則，檢索幾個相關例子，用 LLM 生成新 SQL
+    log_message += f"\n相似度 < {SIMILARITY_THRESHOLD}，[模式: LLM生成]。正在構建 Prompt..."
+    
+    # 構建 Prompt
+    examples_context = ""
+    for hit in hits[:3]: # 取最相關的3個例子
+         examples_context += f"### A user asks: {questions[hit['corpus_id']]}\n{sql_answers[hit['corpus_id']]}\n\n"
+    
+    prompt = f"""### Task
+Generate a SQLite SQL query that answers the following user question.
+Your response must contain ONLY the SQL query. Do not add any explanation.
+
+### Database Schema
+{SCHEMA_DDL}
+### Examples
+{examples_context}
+### Question
+{user_question}
+
+### SQL Query
+"""
+    
+    # 調用 Hugging Face Inference API
+    log_message += "\n正在請求雲端 LLM..."
+    headers = {"Authorization": f"Bearer {HF_TOKEN}"}
+    payload = {"inputs": prompt, "parameters": {"max_new_tokens": 512, "temperature": 0.1, "return_full_text": False}}
+    response_text = ""
+    
+    try:
+        response = requests.post(LLM_API_URL, headers=headers, json=payload)
+        response_text = response.text # 先保存原始響應文本
+        response.raise_for_status() 
+        
+        generated_text = response.json()[0]['generated_text'].strip()
+        
+        # 清理常見的返回格式問題
+        if "```sql" in generated_text:
+            generated_text = generated_text.split("```sql")[1].split("```").strip()
+        if "```" in generated_text:
+            generated_text = generated_text.replace("```", "").strip()
+
+        
+        log_message += f"\nLLM 生成成功！"
+        return generated_text, log_message
+    except Exception as e:
+        error_msg = f"LLM API 調用失敗: {e}\nAPI 原始回應: {response_text}"
+        log_message += f"\n{error_msg}"
+        return "抱歉，調用雲端 AI 時發生錯誤。", log_message
+
+# --- 4. 創建 Gradio Web 界面 ---
+print("--- [5/5] 正在創建 Gradio Web 界面... ---")
+with gr.Blocks(theme=gr.themes.Soft()) as demo:
+    gr.Markdown("# 智能 Text-to-SQL 系統 (混合模式)")
+    gr.Markdown("輸入您的自然語言問題，系統將首先嘗試從知識庫中快速檢索答案。如果問題較新穎，則會調用雲端大語言模型生成SQL。")
+    
+    with gr.Row():
+        question_input = gr.Textbox(label="輸入您的問題", placeholder="例如：去年Nike的總業績是多少？", scale=4)
+        submit_button = gr.Button("生成SQL", variant="primary", scale=1)
+        
+    sql_output = gr.Code(label="生成的 SQL 查詢", language="sql")
+    log_output = gr.Textbox(label="系統日誌 (執行過程)", lines=4, interactive=False)
+    
+    submit_button.click(
+        fn=get_sql_query,
+        inputs=question_input,
+        outputs=[sql_output, log_output]
+    )
+    
+    gr.Examples(
+        examples=[
+            "2024 最好的5個客人以及業績",
+            "比較2023年跟2024年的業績",
+            "上禮拜C組 完成幾份報告",
+            "有沒有快到期的單子？",
+            "哪個客戶的付款最不及時？"
+        ],
+        inputs=question_input
+    )
+
+print("--- 應用準備啟動 ---")
+demo.launch()