Create app.py

app.py

@@ -0,0 +1,477 @@
+import gradio as gr
+import os
+import json
+import time
+from datetime import datetime
+import numpy as np
+from datasets import load_dataset
+from huggingface_hub import HfApi
+import torch
+from transformers import AutoTokenizer, AutoModel
+from openai import OpenAI
+from docx import Document
+import io
+from typing import List, Dict, Any, Optional, Tuple
+
+# ==========================================
+# 環境變數設定
+# ==========================================
+# 從 Hugging Face Secrets 讀取
+HF_TOKEN = os.environ.get("HF_TOKEN", "")
+OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY", "")
+
+# 資料集配置
+DATASET_NAME = "s880453/interview-transcripts-vectorized"
+EMBEDDING_MODEL = "intfloat/multilingual-e5-large"
+
+# 採訪者名單（需要排除）
+INTERVIEWERS = ["徐美苓", "許弘諺", "郭禹彤"]
+
+# ==========================================
+# 全域變數
+# ==========================================
+dataset = None
+embeddings = None
+tokenizer = None
+model = None
+openai_client = None
+all_speakers = []
+
+# ==========================================
+# 初始化函數
+# ==========================================
+def initialize_system():
+    """初始化系統"""
+    global dataset, embeddings, tokenizer, model, openai_client, all_speakers
+    
+    try:
+        print("🔄 正在初始化系統...")
+        
+        # 初始化 OpenAI
+        openai_client = OpenAI(api_key=OPENAI_API_KEY)
+        print("✅ OpenAI 客戶端初始化成功")
+        
+        # 載入資料集
+        print(f"📊 正在載入資料集: {DATASET_NAME}")
+        dataset = load_dataset(DATASET_NAME, split="train", token=HF_TOKEN)
+        print(f"✅ 資料集載入成功，共 {len(dataset)} 筆資料")
+        
+        # 提取所有嵌入向量
+        embeddings = np.array([item['embedding'] for item in dataset])
+        print(f"✅ 嵌入向量提取成功，維度: {embeddings.shape}")
+        
+        # 載入嵌入模型
+        print(f"🤖 正在載入模型: {EMBEDDING_MODEL}")
+        tokenizer = AutoTokenizer.from_pretrained(EMBEDDING_MODEL)
+        model = AutoModel.from_pretrained(EMBEDDING_MODEL)
+        print("✅ 嵌入模型載入成功")
+        
+        # 提取所有發言人（排除採訪者）
+        all_speakers_set = set()
+        for item in dataset:
+            speaker = item['speaker']
+            if speaker not in INTERVIEWERS:
+                all_speakers_set.add(speaker)
+        all_speakers = sorted(list(all_speakers_set))
+        print(f"✅ 發言人列表提取成功，共 {len(all_speakers)} 位受訪者")
+        
+        return True, "系統初始化成功！"
+        
+    except Exception as e:
+        error_msg = f"系統初始化失敗: {str(e)}"
+        print(f"❌ {error_msg}")
+        return False, error_msg
+
+# ==========================================
+# 向量搜尋函數
+# ==========================================
+def average_pool(last_hidden_states, attention_mask):
+    """Average pooling for embeddings"""
+    last_hidden = last_hidden_states.masked_fill(~attention_mask[..., None].bool(), 0.0)
+    return last_hidden.sum(dim=1) / attention_mask.sum(dim=1)[..., None]
+
+def generate_query_embedding(query_text):
+    """生成查詢向量"""
+    # 添加查詢前綴
+    query_with_prefix = f"query: {query_text}"
+    
+    # Tokenize
+    inputs = tokenizer(
+        [query_with_prefix],
+        max_length=512,
+        padding=True,
+        truncation=True,
+        return_tensors='pt'
+    )
+    
+    # 生成嵌入
+    with torch.no_grad():
+        outputs = model(**inputs)
+        query_embedding = average_pool(outputs.last_hidden_state, inputs['attention_mask'])
+        query_embedding = torch.nn.functional.normalize(query_embedding, p=2, dim=1)
+    
+    return query_embedding.cpu().numpy()[0]
+
+def semantic_search(query, selected_speakers, top_k=20):
+    """語義搜尋"""
+    if not dataset:
+        return []
+    
+    # 生成查詢向量
+    query_vector = generate_query_embedding(query)
+    
+    # 計算相似度
+    similarities = []
+    for i, item in enumerate(dataset):
+        # 檢查發言人過濾
+        if selected_speakers and item['speaker'] not in selected_speakers:
+            continue
+        
+        # 計算餘弦相似度
+        item_vector = np.array(item['embedding'])
+        similarity = np.dot(query_vector, item_vector)
+        
+        similarities.append({
+            'score': float(similarity),
+            'text': item['text'],
+            'speaker': item['speaker'],
+            'turn_index': item['turn_index'],
+            'file_id': item['file_id']
+        })
+    
+    # 排序並返回前 k 個結果
+    similarities.sort(key=lambda x: x['score'], reverse=True)
+    return similarities[:top_k]
+
+# ==========================================
+# GPT-4o-mini 處理函數
+# ==========================================
+def call_gpt4o_mini(prompt, temperature=0.1):
+    """調用 GPT-4o-mini"""
+    try:
+        response = openai_client.chat.completions.create(
+            model="gpt-4o-mini",
+            messages=[
+                {"role": "system", "content": "你是一個專業的訪談分析助手，擅長從訪談內容中提取關鍵信息並回答問題。"},
+                {"role": "user", "content": prompt}
+            ],
+            temperature=temperature
+        )
+        return response.choices[0].message.content
+    except Exception as e:
+        return f"GPT 調用失敗: {str(e)}"
+
+# ==========================================
+# RAG 對話函數
+# ==========================================
+def rag_chat(question, selected_speakers, history):
+    """RAG 對話處理"""
+    if not dataset:
+        return history + [[question, "系統尚未初始化，請稍後再試。"]]
+    
+    try:
+        # 執行語義搜尋
+        search_results = semantic_search(question, selected_speakers, top_k=10)
+        
+        if not search_results:
+            return history + [[question, "未找到相關內容，請嘗試其他問題。"]]
+        
+        # 構建上下文
+        context = "相關訪談內容：\n\n"
+        for i, result in enumerate(search_results, 1):
+            context += f"[片段 {i}]\n"
+            context += f"發言人：{result['speaker']}\n"
+            context += f"內容：{result['text']}\n"
+            context += f"相似度：{result['score']:.3f}\n\n"
+        
+        # 構建 GPT prompt
+        prompt = f"""基於以下訪談內容回答問題。
+
+{context}
+
+問題：{question}
+
+請提供準確、完整的回答，並在適當時引用具體的發言人和內容。"""
+        
+        # 調用 GPT
+        answer = call_gpt4o_mini(prompt)
+        
+        return history + [[question, answer]]
+        
+    except Exception as e:
+        return history + [[question, f"處理過程中發生錯誤：{str(e)}"]]
+
+# ==========================================
+# 訪綱填答函數
+# ==========================================
+def parse_word_document(file):
+    """解析 Word 文檔中的問題"""
+    try:
+        doc = Document(file)
+        questions = []
+        
+        for para in doc.paragraphs:
+            text = para.text.strip()
+            # 識別問題（以數字、問號或特定格式開頭）
+            if text and (
+                text[0].isdigit() or 
+                '?' in text or 
+                '？' in text or
+                text.startswith('Q') or
+                text.startswith('問')
+            ):
+                questions.append(text)
+        
+        return questions
+    except Exception as e:
+        return []
+
+def fill_interview_guide(file, selected_speakers):
+    """填答訪綱"""
+    if not dataset:
+        return None, "系統尚未初始化"
+    
+    try:
+        # 解析 Word 文檔
+        questions = parse_word_document(file)
+        
+        if not questions:
+            return None, "未能從文檔中提取問題，請確認格式"
+        
+        # 創建新的 Word 文檔
+        output_doc = Document()
+        output_doc.add_heading('訪談訪綱 - AI 自動填答', 0)
+        output_doc.add_paragraph(f'處理時間：{datetime.now().strftime("%Y-%m-%d %H:%M:%S")}')
+        output_doc.add_paragraph(f'選擇的受訪者：{", ".join(selected_speakers) if selected_speakers else "全部"}')
+        output_doc.add_paragraph('')
+        
+        # 處理每個問題
+        for i, question in enumerate(questions, 1):
+            # 添加問題
+            output_doc.add_heading(f'問題 {i}', level=2)
+            output_doc.add_paragraph(question)
+            
+            # 搜尋相關內容
+            search_results = semantic_search(question, selected_speakers, top_k=5)
+            
+            if search_results:
+                # 構建上下文
+                context = ""
+                for result in search_results:
+                    context += f"發言人：{result['speaker']}\n"
+                    context += f"內容：{result['text']}\n\n"
+                
+                # 使用 GPT 生成回答
+                prompt = f"""基於以下訪談內容回答問題：
+
+{context}
+
+問題：{question}
+
+請提供結構化的回答，包含：
+1. 主要觀點
+2. 不同受訪者的觀點（如果有多位）
+3. 具體引述"""
+                
+                answer = call_gpt4o_mini(prompt)
+                
+                # 添加回答
+                output_doc.add_heading('回答：', level=3)
+                for line in answer.split('\n'):
+                    if line.strip():
+                        output_doc.add_paragraph(line)
+                
+                # 添加相關引述
+                output_doc.add_heading('相關引述：', level=3)
+                for j, result in enumerate(search_results[:3], 1):
+                    p = output_doc.add_paragraph()
+                    p.add_run(f"{j}. {result['speaker']}：").bold = True
+                    p.add_run(f" {result['text'][:200]}...")
+            else:
+                output_doc.add_paragraph("未找到相關內容")
+            
+            output_doc.add_paragraph('')  # 空行分隔
+        
+        # 保存文檔
+        output_buffer = io.BytesIO()
+        output_doc.save(output_buffer)
+        output_buffer.seek(0)
+        
+        return output_buffer, "訪綱填答完成！"
+        
+    except Exception as e:
+        return None, f"處理失敗：{str(e)}"
+
+# ==========================================
+# Gradio 介面
+# ==========================================
+def create_interface():
+    """創建 Gradio 介面"""
+    
+    with gr.Blocks(title="訪談轉錄稿 RAG 系統", theme=gr.themes.Soft()) as app:
+        # 標題
+        gr.Markdown("""
+        # 🎙️ 訪談轉錄稿智慧分析系統
+        
+        基於 RAG 技術的訪談內容分析與問答系統
+        """)
+        
+        # 系統狀態
+        with gr.Row():
+            status_text = gr.Textbox(
+                label="系統狀態",
+                value="初始化中...",
+                interactive=False
+            )
+        
+        # 主要功能區
+        with gr.Tabs():
+            # Tab 1: AI 對話
+            with gr.Tab("💬 AI 對話"):
+                with gr.Row():
+                    with gr.Column(scale=1):
+                        gr.Markdown("### 選擇受訪者")
+                        speaker_selector = gr.CheckboxGroup(
+                            choices=all_speakers,
+                            label="受訪者列表",
+                            info="不選擇則搜尋全部內容"
+                        )
+                    
+                    with gr.Column(scale=3):
+                        chatbot = gr.Chatbot(
+                            height=500,
+                            label="對話記錄"
+                        )
+                        
+                        with gr.Row():
+                            msg = gr.Textbox(
+                                label="輸入問題",
+                                placeholder="請輸入您想詢問的問題...",
+                                scale=4
+                            )
+                            send_btn = gr.Button("發送", variant="primary", scale=1)
+                        
+                        clear_btn = gr.Button("清除對話")
+            
+            # Tab 2: 訪綱填答
+            with gr.Tab("📝 訪綱填答"):
+                gr.Markdown("""
+                ### 使用說明
+                1. 選擇要分析的受訪者
+                2. 上傳 Word 格式的訪綱文件
+                3. 系統將自動識別問題並填答
+                4. 下載完成的文檔
+                """)
+                
+                with gr.Row():
+                    with gr.Column():
+                        guide_speakers = gr.CheckboxGroup(
+                            choices=all_speakers,
+                            label="選擇受訪者",
+                            info="不選擇則分析全部受訪者"
+                        )
+                        
+                        file_input = gr.File(
+                            label="上傳訪綱 (Word 格式)",
+                            file_types=[".docx", ".doc"]
+                        )
+                        
+                        process_btn = gr.Button("開始處理", variant="primary")
+                    
+                    with gr.Column():
+                        process_status = gr.Textbox(
+                            label="處理狀態",
+                            interactive=False
+                        )
+                        
+                        download_file = gr.File(
+                            label="下載結果",
+                            visible=False
+                        )
+        
+        # 關於
+        with gr.Accordion("ℹ️ 關於系統", open=False):
+            gr.Markdown("""
+            ### 系統資訊
+            - **向量模型**: multilingual-e5-large
+            - **語言模型**: GPT-4o-mini
+            - **資料來源**: Hugging Face Dataset
+            - **版本**: 1.0.0
+            
+            ### 功能特色
+            - 🔍 智慧語義搜尋
+            - 💬 自然語言問答
+            - 📝 自動訪綱填答
+            - 👥 多受訪者分析
+            """)
+        
+        # 事件處理
+        def send_message(message, speakers, history):
+            if not message:
+                return "", history
+            new_history = rag_chat(message, speakers, history)
+            return "", new_history
+        
+        def clear_chat():
+            return []
+        
+        def process_guide(file, speakers):
+            if not file:
+                return "請上傳文件", None
+            
+            result_file, status = fill_interview_guide(file.name, speakers)
+            
+            if result_file:
+                # 保存到臨時文件
+                temp_path = f"filled_guide_{datetime.now().strftime('%Y%m%d_%H%M%S')}.docx"
+                with open(temp_path, 'wb') as f:
+                    f.write(result_file.getvalue())
+                return status, gr.File(value=temp_path, visible=True)
+            else:
+                return status, None
+        
+        # 綁定事件
+        send_btn.click(
+            send_message,
+            inputs=[msg, speaker_selector, chatbot],
+            outputs=[msg, chatbot]
+        )
+        
+        msg.submit(
+            send_message,
+            inputs=[msg, speaker_selector, chatbot],
+            outputs=[msg, chatbot]
+        )
+        
+        clear_btn.click(clear_chat, outputs=[chatbot])
+        
+        process_btn.click(
+            process_guide,
+            inputs=[file_input, guide_speakers],
+            outputs=[process_status, download_file]
+        )
+        
+        # 初始化系統
+        def update_status():
+            success, message = initialize_system()
+            if success:
+                # 更新發言人列表
+                speaker_selector.choices = all_speakers
+                guide_speakers.choices = all_speakers
+            return message
+        
+        app.load(update_status, outputs=[status_text])
+    
+    return app
+
+# ==========================================
+# 主程式入口
+# ==========================================
+if __name__ == "__main__":
+    # 創建並啟動應用
+    app = create_interface()
+    app.launch(
+        share=False,
+        server_name="0.0.0.0",
+        server_port=7860
+    )