# ========================================== # MILESTONE 6: PRODUCTION READY PLATFORM # Multi-Format File Support (.pdf, .txt, .docx, .csv, .md) # ========================================== print("Installing required tools...") # ADDED: docx2txt to read Microsoft Word files import gradio as gr from huggingface_hub import InferenceClient from transformers import pipeline import numpy as np import librosa import traceback import torch import os import shutil import json import zipfile # --- UPGRADED RAG IMPORTS --- # We imported Docx2txtLoader and CSVLoader! from langchain_community.document_loaders import PyPDFLoader, TextLoader, Docx2txtLoader, CSVLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_huggingface import HuggingFaceEmbeddings from langchain_community.vectorstores import FAISS # ------------------------------------------ # 0. DIRECTORIES & DYNAMIC CONFIGURATION # ------------------------------------------ DOC_DIR = "school_data" CONFIG_FILE = os.path.join(DOC_DIR, "config.json") EXPORT_DIR = "deployment_export" os.makedirs(DOC_DIR, exist_ok=True) os.makedirs(EXPORT_DIR, exist_ok=True) DEFAULT_CONFIG = { "llm_model": "Qwen/Qwen2.5-7B-Instruct", "prompt_school": "You are a school assistant. Answer using strictly this info:", "prompt_general": "You are a friendly general tutor. Answer the question nicely.", "max_audio_sec": 15, "max_file_mb": 5 } def load_config(): if os.path.exists(CONFIG_FILE): with open(CONFIG_FILE, 'r') as f: return json.load(f) return DEFAULT_CONFIG def save_config(new_config): with open(CONFIG_FILE, 'w') as f: json.dump(new_config, f) if not os.path.exists(CONFIG_FILE): save_config(DEFAULT_CONFIG) # ------------------------------------------ # 1. RAG & AI INITIALIZATION (MULTI-FILE LOGIC) # ------------------------------------------ knowledge_base = None print("Downloading the Document Reading AI...") embeddings = HuggingFaceEmbeddings(model_name="all-MiniLM-L6-v2") def rebuild_knowledge_base(): global knowledge_base files = [f for f in os.listdir(DOC_DIR) if f != "config.json" and not f.startswith('.')] if len(files) == 0: knowledge_base = None return "⚠️ All files deleted. AI brain is empty." all_chunks = [] text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50) for filename in files: filepath = os.path.join(DOC_DIR, filename) try: # --- THE MULTI-FORMAT ROUTER --- if filename.lower().endswith(".pdf"): loader = PyPDFLoader(filepath) elif filename.lower().endswith(".txt") or filename.lower().endswith(".md"): loader = TextLoader(filepath, encoding='utf-8') elif filename.lower().endswith(".docx"): loader = Docx2txtLoader(filepath) elif filename.lower().endswith(".csv"): loader = CSVLoader(filepath) else: print(f"Skipping unsupported file type: {filename}") continue all_chunks.extend(text_splitter.split_documents(loader.load())) except Exception as e: print(f"Error reading {filename}: {e}") if len(all_chunks) > 0: knowledge_base = FAISS.from_documents(all_chunks, embeddings) return f"✅ Brain updated! {len(files)} files loaded." knowledge_base = None return "⚠️ Files found, but no text extracted." my_secret_key = os.getenv("Token") client = InferenceClient(token=my_secret_key) compute_device = 0 if torch.cuda.is_available() else -1 local_whisper = pipeline("automatic-speech-recognition", model="openai/whisper-small", chunk_length_s=30, device=compute_device) rebuild_knowledge_base() # ------------------------------------------ # 2. CORE BACKEND LOGIC # ------------------------------------------ def get_file_list(): return [f for f in os.listdir(DOC_DIR) if f != "config.json" and not f.startswith('.')] def upload_files(files): if files is not None: config = load_config() max_bytes = config["max_file_mb"] * 1024 * 1024 for f in files: if os.path.getsize(f.name) > max_bytes: gr.Warning(f"Blocked: {os.path.basename(f.name)} exceeds {config['max_file_mb']}MB limit.") continue filename = os.path.basename(f.name) shutil.copy(f.name, os.path.join(DOC_DIR, filename)) gr.Info(rebuild_knowledge_base()) return gr.update(choices=get_file_list()), gr.update(value=None) def delete_selected_files(selected_filenames): if selected_filenames: for filename in selected_filenames: filepath = os.path.join(DOC_DIR, filename) if os.path.exists(filepath): os.remove(filepath) gr.Info(rebuild_knowledge_base()) return gr.update(choices=get_file_list(), value=[]) def download_selected_files(selected_filenames): if not selected_filenames: gr.Warning("Please select files to download first.") return gr.update(visible=False, value=None) return gr.update(visible=True, value=[os.path.join(DOC_DIR, f) for f in selected_filenames]) def transcribe_only(audio_input, language_choice): if audio_input is None: return "" try: config = load_config() original_sr, audio_array = audio_input audio_duration = len(audio_array) / original_sr if audio_duration > config["max_audio_sec"]: gr.Warning(f"Audio blocked! Exceeds the {config['max_audio_sec']} second limit.") return "" if audio_array.ndim > 1: audio_array = np.mean(audio_array, axis=1) audio_array = audio_array.astype(np.float32) if np.max(np.abs(audio_array)) > 1.0: audio_array = audio_array / 32768.0 if original_sr != 16000: audio_array = librosa.resample(y=audio_array, orig_sr=original_sr, target_sr=16000) ai_lang = "english" if language_choice == "English" else "chinese" result = local_whisper({"sampling_rate": 16000, "raw": audio_array}, generate_kwargs={"language": ai_lang, "task": "transcribe"}) return result['text'] except Exception: return "" def agent_chat(user_text, chat_history): global knowledge_base try: if not user_text or not str(user_text).strip(): gr.Warning("⚠️ 請輸入文字！") return "", chat_history config = load_config() user_input_str = str(user_text) # 強制轉成字串，防止意外 # --- 1. 路由判斷 (確保格式正確) --- router_prompt = f"Determine if this is SCHOOL or GENERAL: {user_input_str}" route_res = client.chat_completion( model=config["llm_model"], messages=[{"role": "user", "content": router_prompt}], # 這裡必須是字典 max_tokens=5 ) # --- 2. 取得 System Prompt --- route_verdict = route_res.choices[0].message.content.strip() if "SCHOOL" in route_verdict: if knowledge_base is not None: docs = knowledge_base.similarity_search(user_input_str, k=2) context = "\n".join([doc.page_content for doc in docs]) system_prompt = f"{config['prompt_school']}\nContext: {context}" else: system_prompt = "You are a school assistant. No documents found." else: system_prompt = config['prompt_general'] # --- 3. 關鍵修正：構建發送給 AI 的完整清單 --- # 我們在這裡明確定義 role 和 content，確保完全符合 API 要求 final_messages = [ {"role": "system", "content": str(system_prompt)} ] # 如果你想加入記憶（History），請用這個迴圈： for old_user, old_ai in chat_history[-5:]: # 取最近5次對話 final_messages.append({"role": "user", "content": str(old_user)}) final_messages.append({"role": "assistant", "content": str(old_ai)}) # 加入本次的問題 final_messages.append({"role": "user", "content": user_input_str}) # --- 4. 發送請求 --- final_response = client.chat_completion( model=config["llm_model"], messages=final_messages, # 傳入剛才建好的字典清單 max_tokens=200, temperature=0.5 ) ai_reply = final_response.choices[0].message.content chat_history.append((user_input_str, ai_reply)) return "", chat_history except Exception as e: print(f"DEBUG ERROR: {e}") # 在後台印出真正的錯誤原因 chat_history.append(("Error", f"⚠️ 格式錯誤或連線中斷: {str(e)}")) return "", chat_history # ------------------------------------------ # 3. ADMIN SETTINGS HELPERS # ------------------------------------------ def admin_login(u, p): if u == "admin" and p == "admin": return gr.update(visible=False), gr.update(visible=True) return gr.update(visible=True), gr.update(visible=False) def update_ai_settings(model, p_school, p_general): conf = load_config() final_model = model.strip() if model and model.strip() != "" else "Qwen/Qwen2.5-7B-Instruct" conf.update({"llm_model": final_model, "prompt_school": p_school, "prompt_general": p_general}) save_config(conf) gr.Info(f"✅ AI Config Saved! Model set to: {final_model}") return final_model def update_security_settings(audio_sec, file_mb): conf = load_config() conf.update({"max_audio_sec": audio_sec, "max_file_mb": file_mb}) save_config(conf) gr.Info("🛡️ Security Guards updated successfully!") def export_project(): zip_path = os.path.join(EXPORT_DIR, "school_bot_deploy.zip") req_text = "gradio\ntransformers\ntorch\nlibrosa\nsoundfile\ndocx2txt\nlangchain\nlangchain-community\nlangchain-huggingface\nlangchain-text-splitters\nfaiss-cpu\npypdf\nhuggingface_hub" with zipfile.ZipFile(zip_path, 'w') as zf: zf.writestr("requirements.txt", req_text) zf.writestr("README.md", "# Deployment Guide\n1. Upload this code to Hugging Face Spaces.\n2. Add your HF_TOKEN to the Space Secrets.\n3. Run app.py!") return gr.update(visible=True, value=zip_path) # ------------------------------------------ # 4. THE UI LAYOUT # ------------------------------------------ print("Building Final UI...") current_config = load_config() with gr.Blocks(theme=gr.themes.Soft()) as demo: with gr.Tab("💬 Student Chatbot"): gr.Markdown("# 🦸‍♂️ My Ultimate Hybrid AI Agent") chatbot = gr.Chatbot(height=400) with gr.Row(): with gr.Column(scale=3): txt_input = gr.Textbox(label="Type a message") with gr.Column(scale=2): lang_toggle = gr.Radio(choices=["English", "Chinese"], value="Chinese", label="Voice Language:") aud_input = gr.Audio(sources=["microphone"], type="numpy", label="Record Voice") with gr.Row(): submit_btn = gr.Button("🚀 Send to AI", variant="primary") clear_btn = gr.ClearButton([txt_input, aud_input, chatbot], value="🗑️ Clear Chat") aud_input.change(fn=transcribe_only, inputs=[aud_input, lang_toggle], outputs=[txt_input]) submit_btn.click(fn=agent_chat, inputs=[txt_input, chatbot], outputs=[txt_input, chatbot]) with gr.Tab("⚙️ Admin Settings"): with gr.Column(visible=True) as login_screen: gr.Markdown("### 🔒 Admin Login") with gr.Row(): usr_box = gr.Textbox(label="Username") pwd_box = gr.Textbox(label="Password", type="password") login_btn = gr.Button("Login", variant="primary") with gr.Column(visible=False) as admin_panel: gr.Markdown("## 🎛️ Platform Control Center") with gr.Tabs(): with gr.Tab("📂 Data Manager"): with gr.Row(): # THE UI UPGRADE: Now accepts .docx, .csv, and .md! upload_btn = gr.File( label="1. Upload File (Click or Drag & Drop)", file_count="multiple", type="filepath", file_types=[".pdf", ".txt", ".docx", ".csv", ".md"] ) with gr.Column(): download_btn = gr.Button("2. ⬇️ Download Selected") delete_btn = gr.Button("3. 🗑️ Delete Selected", variant="stop") download_output = gr.File(label="Click to Save", visible=False) gr.Markdown("---") with gr.Row(): gr.Markdown("### 📄 Database List View") select_all_btn = gr.Button("☑️ Select All", size="sm") file_listview = gr.CheckboxGroup(choices=get_file_list(), label="") with gr.Tab("🧠 AI Config & Prompts"): gr.Markdown("Change how the AI behaves in real-time without restarting the server.") ai_model_txt = gr.Textbox(value=current_config["llm_model"], label="Cloud LLM Model (Hugging Face ID)", placeholder="Leave blank to use default (Qwen/Qwen2.5-7B-Instruct)") prompt_school_txt = gr.TextArea(value=current_config["prompt_school"], label="School RAG System Prompt") prompt_general_txt = gr.TextArea(value=current_config["prompt_general"], label="General Conversation System Prompt") save_ai_btn = gr.Button("💾 Save AI Settings", variant="primary") with gr.Tab("🛡️ Security Guards"): gr.Markdown("Protect your server from spam and massive files.") audio_limit = gr.Slider(minimum=5, maximum=60, step=1, value=current_config["max_audio_sec"], label="Max Voice Recording Length (Seconds)") file_limit = gr.Slider(minimum=1, maximum=50, step=1, value=current_config["max_file_mb"], label="Max Upload File Size (MB)") save_sec_btn = gr.Button("💾 Save Security Settings", variant="primary") with gr.Tab("☁️ Cloud Deployment"): gr.Markdown("### Ready to go live 24/7?") gr.Markdown("Google Colab goes to sleep when you close your browser. To keep this bot online forever, export your deployment kit and upload it to a free host like **Hugging Face Spaces**.") export_btn = gr.Button("📦 Generate Deployment Kit (ZIP)", variant="primary") export_output = gr.File(label="Download your ZIP", visible=False) # WIRING login_btn.click(fn=admin_login, inputs=[usr_box, pwd_box], outputs=[login_screen, admin_panel]) upload_btn.upload(fn=upload_files, inputs=[upload_btn], outputs=[file_listview, upload_btn]) delete_btn.click(fn=delete_selected_files, inputs=[file_listview], outputs=[file_listview]) download_btn.click(fn=download_selected_files, inputs=[file_listview], outputs=[download_output]) def select_all(): return gr.update(value=get_file_list()) select_all_btn.click(fn=select_all, inputs=[], outputs=[file_listview]) save_ai_btn.click(fn=update_ai_settings, inputs=[ai_model_txt, prompt_school_txt, prompt_general_txt], outputs=[ai_model_txt]) save_sec_btn.click(fn=update_security_settings, inputs=[audio_limit, file_limit]) export_btn.click(fn=export_project, outputs=[export_output]) demo.launch(share=True)