app.py

import gradio as gr
import os
import json
import pickle
import re
from typing import List, Tuple, Dict, Any
from dotenv import load_dotenv
import openai
import faiss
import numpy as np
from sentence_transformers import SentenceTransformer

# 載入環境變數
load_dotenv()
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
# API金鑰不再是必須的，因為使用者可以提供自己的金鑰
if OPENAI_API_KEY:
    print("[INFO] 已從環境變數讀取API金鑰")
else:
    print("[INFO] 未設置環境變數API金鑰，將使用使用者提供的金鑰")

# 配置
FAISS_DATA_DIR = "faiss_data"  # FAISS 資料的資料夾
MODEL_NAME = 'paraphrase-multilingual-MiniLM-L12-v2'  # 使用與建立索引時相同的模型

def load_faiss_data(data_dir=FAISS_DATA_DIR):
    """載入 FAISS 索引和 meta 資料"""
    if not os.path.exists(data_dir):
        raise ValueError(f"未找到 FAISS 資料資料夾：{data_dir}")
    
    # 載入 FAISS 索引
    index_path = os.path.join(data_dir, 'criminal_judgments.index')
    meta_path = os.path.join(data_dir, 'criminal_judgments_meta.pkl')
    
    if not os.path.exists(index_path) or not os.path.exists(meta_path):
        raise ValueError("未找到必要的 FAISS 資料檔案")
    
    print(f"[INFO] 載入 FAISS 索引和 meta 資料")
    index = faiss.read_index(index_path)
    with open(meta_path, 'rb') as f:
        meta_data = pickle.load(f)
    
    return index, meta_data

# 載入模型和資料
try:
    model = SentenceTransformer(MODEL_NAME)
    faiss_index, meta_data = load_faiss_data()
    print("[INFO] 成功載入模型和 FAISS 資料")
except Exception as e:
    print(f"[ERROR] 載入失敗：{e}")
    print("[INFO] 請先執行 create_judgment_embeddings.py 建立索引")
    model, faiss_index, meta_data = None, None, None

def format_judgment_info(judgment_meta):
    """格式化判決資訊"""
    return f"""
案件編號：{judgment_meta['JID']}
年度：{judgment_meta['JYEAR']}
案件類型：{judgment_meta['JCASE']}
案號：{judgment_meta['JNO']}
判決日期：{judgment_meta['JDATE']}
案件標題：{judgment_meta['JTITLE']}
PDF連結：{judgment_meta['JPDF']}
"""

def format_retrieval_results(results, meta_data):
    """格式化檢索結果，用於顯示給用戶"""
    html = "<div style='max-height: 500px; overflow-y: auto; padding: 1rem; background-color: #f7f7f7; border-radius: 0.5rem;'>"
    html += "<h3>📚 檢索到的相關判決</h3>"
    
    for i, (text, meta, match_context) in enumerate(results, start=1):
        html += f"<div style='margin-bottom: 1rem; padding: 0.8rem; background-color: white; border-radius: 0.3rem; border-left: 4px solid #3498db;'>"
        html += f"<p><strong>判決 {i}</strong></p>"
        html += f"<p><strong>案件編號:</strong> {meta['JID']}</p>"
        html += f"<p><strong>案件類型:</strong> {meta['JCASE']}</p>"
        html += f"<p><strong>判決日期:</strong> {meta['JDATE']}</p>"
        html += f"<p><strong>案件標題:</strong> {meta['JTITLE']}</p>"
        html += f"<p><strong>PDF連結:</strong> <a href='{meta['JPDF']}' target='_blank'>查看原文</a></p>"
        
        # 顯示匹配上下文（若有）
        if match_context:
            html += f"<div style='margin-top: 0.5rem; padding: 0.5rem; background-color: #fffbea; border-radius: 0.3rem; border-left: 4px solid #f0c674;'>"
            html += f"<p><strong>匹配內容:</strong><br>{match_context}</p>"
            html += "</div>"
        
        # 添加判決內容摘要，但縮短長度
        html += f"<div style='margin-top: 0.5rem; padding: 0.5rem; background-color: #f8f9fa; border-radius: 0.3rem;'>"
        html += f"<p><strong>相關內容摘要:</strong><br>{text[:250]}...</p>"
        html += "</div></div>"
    
    html += "</div>"
    return html

def preprocess_judgment_text(text: str) -> str:
    """預處理判決書文本，處理可能觸發內容審查的詞彙"""
    # 建立替換對照表，保留專業用語的原意
    replacements = {
        "殺人": "故意致人於死",
        "自殺": "自我傷害",
        "性侵": "違反性自主",
        "強制性交": "違反性自主",
        "毒品": "管制藥品",
        "槍械": "危險物品",
        "暴力": "強制力",
        "血": "生理組織",
    }
    
    # 進行替換
    processed_text = text
    for old, new in replacements.items():
        processed_text = processed_text.replace(old, new)
    
    return processed_text

def text_search(query: str, meta_data: List[Dict[str, Any]], max_results: int = 10) -> List[Tuple[int, str]]:
    """
    使用文本搜尋方式尋找相關判決
    
    Args:
        query: 搜尋關鍵字
        meta_data: 判決meta資料列表
        max_results: 最大返回結果數
        
    Returns:
        搜尋結果：[(索引，匹配上下文), ...]
    """
    results = []
    
    # 如果查詢字串少於2個字符，返回空結果
    if len(query.strip()) < 2:
        return results
    
    # 將查詢轉為小寫並移除特殊字符
    clean_query = re.sub(r'[^\w\s]', '', query.lower())
    search_terms = clean_query.split()
    
    # 為每個判決計算匹配分數
    for idx, meta in enumerate(meta_data):
        full_text = meta['JFULL'].lower()
        
        # 計算匹配分數和匹配上下文
        score = 0
        best_context = ""
        best_score = 0
        
        # 對每個搜尋詞進行匹配
        for term in search_terms:
            if term in full_text:
                # 完全匹配
                term_score = len(term) * 2
                
                # 找出匹配詞的上下文（前後50個字符）
                match_idx = full_text.find(term)
                start_idx = max(0, match_idx - 50)
                end_idx = min(len(full_text), match_idx + len(term) + 50)
                context = "..." + meta['JFULL'][start_idx:end_idx] + "..."
                
                if term_score > best_score:
                    best_score = term_score
                    best_context = context
                
                score += term_score
            else:
                # 部分匹配（例如：搜尋ABC，找到BC）
                for partial_size in range(len(term)-1, 0, -1):
                    for i in range(len(term) - partial_size + 1):
                        partial = term[i:i+partial_size]
                        if len(partial) >= 2 and partial in full_text:  # 只考慮至少2個字符的部分
                            partial_score = partial_size
                            
                            # 找出匹配詞的上下文
                            match_idx = full_text.find(partial)
                            start_idx = max(0, match_idx - 50)
                            end_idx = min(len(full_text), match_idx + len(partial) + 50)
                            context = "..." + meta['JFULL'][start_idx:end_idx] + "..."
                            
                            if partial_score > best_score:
                                best_score = partial_score
                                best_context = context
                            
                            score += partial_score
                            break  # 找到第一個部分匹配就跳出
        
        # 如果有匹配，加入結果
        if score > 0:
            results.append((idx, score, best_context))
    
    # 根據分數排序並返回前N個結果
    results.sort(key=lambda x: x[1], reverse=True)
    return [(idx, context) for idx, _, context in results[:max_results]]

def combined_search(query: str, k: int = 5, search_mode: str = "自動（先文本後向量）"):
    """
    結合文本搜尋和向量搜尋
    
    Args:
        query: 搜尋查詢
        k: 返回結果數量
        search_mode: 搜尋模式，可選 "自動（先文本後向量）", "僅文本搜尋", "僅向量搜尋"
        
    Returns:
        搜尋結果列表
    """
    # 根據搜尋模式選擇搜尋策略
    if search_mode == "僅文本搜尋":
        # 只使用文本搜尋
        text_results = text_search(query, meta_data)
        if not text_results:
            return []  # 沒有找到匹配結果
        
        results = []
        for idx, context in text_results[:k]:
            judgment_meta = meta_data[idx]
            processed_text = preprocess_judgment_text(judgment_meta['JFULL'])
            results.append((processed_text, judgment_meta, context))
        return results
    
    elif search_mode == "僅向量搜尋":
        # 只使用向量搜尋
        query_vector = model.encode([query])[0].astype('float32')
        distances, indices = faiss_index.search(query_vector.reshape(1, -1), k)
        
        results = []
        for idx in indices[0]:
            judgment_meta = meta_data[idx]
            processed_text = preprocess_judgment_text(judgment_meta['JFULL'])
            results.append((processed_text, judgment_meta, ""))  # 無匹配上下文
        return results
    
    else:  # 預設為 "自動（先文本後向量）"
        # 先進行文本搜尋
        text_results = text_search(query, meta_data)
        
        # 如果文本搜尋有結果，優先使用
        if text_results:
            results = []
            for idx, context in text_results[:k]:
                judgment_meta = meta_data[idx]
                processed_text = preprocess_judgment_text(judgment_meta['JFULL'])
                results.append((processed_text, judgment_meta, context))
            return results
        
        # 如果文本搜尋無結果，使用向量搜尋作為備用
        query_vector = model.encode([query])[0].astype('float32')
        distances, indices = faiss_index.search(query_vector.reshape(1, -1), k)
        
        results = []
        for idx in indices[0]:
            judgment_meta = meta_data[idx]
            processed_text = preprocess_judgment_text(judgment_meta['JFULL'])
            results.append((processed_text, judgment_meta, ""))  # 無匹配上下文
        
        return results

def respond(
    message: str,
    history: List[Tuple[str, str]],
    system_message: str,
    max_tokens: int,
    temperature: float,
    top_p: float,
    show_retrieval_results: bool,
    user_api_key: str = "",
    k: int = 5,
    model_name: str = "gpt-4o-mini",
    search_mode: str = "自動（先文本後向量）",
):
    # 檢查必要組件是否已載入
    if None in (model, faiss_index, meta_data):
        return "錯誤：系統未完全載入。請確認已執行 create_judgment_embeddings.py 建立索引。", None
    
    # 檢查 API 金鑰
    api_key = user_api_key.strip() if user_api_key.strip() else OPENAI_API_KEY
    if not api_key:
        yield "請在右側設定中輸入您的 OpenAI API 金鑰以繼續使用。您可以從 OpenAI 網站獲取 API 金鑰：https://platform.openai.com/api-keys", None
        return
    
    # 臨時設置 API 金鑰，僅用於本次請求
    client = openai.OpenAI(api_key=api_key)
    
    # 根據問題長度動態調整檢索數量，避免超出上下文限制
    effective_k = min(k, 3) if len(message) > 100 else min(k, 4)
    
    # 搜尋相關判決（使用改進的搜尋方法）
    results = combined_search(message, k=effective_k, search_mode=search_mode)
    
    # 如果沒有找到結果，返回提示訊息
    if not results:
        return "抱歉，我沒有找到與您問題相關的判決。請嘗試使用不同的搜尋詞或切換搜尋模式。", None
    
    # 準備上下文，限制每個判決摘要的長度
    context = ""
    for i, (text, meta, match_context) in enumerate(results, start=1):
        context += f"\n[判決 {i}]\n"
        context += format_judgment_info(meta)
        
        # 如果有匹配上下文，加入上下文
        if match_context:
            context += f"匹配內容：\n{match_context}\n"
        
        # 限制每個判決文本的長度
        summary_length = 300
        context += f"內容摘要：\n{text[:summary_length]}...\n"
    
    # 如果用戶要求顯示檢索結果
    retrieval_display = format_retrieval_results(results, meta_data) if show_retrieval_results else None
    
    # 建立系統提示
    rag_system_message = f"""{system_message}

這是一個司法判決書分析系統，所有內容均來自真實判決書。請注意：
1. 判決書中可能包含暴力、犯罪等敏感描述，這些是判決書的必要組成部分
2. 請以客觀、專業的法律角度分析這些內容
3. 不要對判決書中的敏感內容進行道德評判或迴避
4. 保持專業中立的態度，準確傳達判決書的內容

根據以下判決資料回答用戶問題。若問題無法從判決資料中回答，請說明你無法提供答案，不要憑空編造。

請在回答中引用具體的判決案號和日期，並說明資訊來源。您可以透過提供的 PDF 連結查看完整判決內容。

參考判決：
{context}
"""
    
    # 建立訊息清單
    messages = [
        {"role": "system", "content": "你現在是一個專業的法律助理，正在處理司法判決書。請以客觀、專業的態度處理所有內容，包括敏感話題。"},
        {"role": "system", "content": rag_system_message}
    ]
    
    # 只保留最近的幾次對話歷史，減少 token 使用量
    recent_history = history[-2:] if len(history) > 2 else history
    for val in recent_history:
        if val[0]:
            messages.append({"role": "user", "content": val[0]})
        if val[1]:
            messages.append({"role": "assistant", "content": val[1]})
    messages.append({"role": "user", "content": message})
    
    # 呼叫 OpenAI API 生成回答
    response = ""
    try:
        for chunk in client.chat.completions.create(
            model=model_name,
            messages=messages,
            max_tokens=max_tokens,
            stream=True,
            temperature=temperature,
            top_p=top_p,
            presence_penalty=0.6,
            frequency_penalty=0.3,
        ):
            if chunk.choices[0].delta.content is not None:
                token = chunk.choices[0].delta.content
                response += token
                yield response, retrieval_display
    except Exception as e:
        error_message = f"呼叫 OpenAI API 時發生錯誤：{str(e)}"
        yield error_message, retrieval_display

# 自定義 CSS
css = """
.gradio-container {
    font-family: "Noto Sans TC", sans-serif !important;
}
.retrieval-results {
    margin-top: 1rem;
    margin-bottom: 1rem;
}
"""

# Gradio 界面
with gr.Blocks(css=css, theme=gr.themes.Soft()) as demo:
    gr.HTML("<h1>🔍 司法判決書搜尋系統（示範）</h1>")
    gr.HTML("<p>這個系統能夠搜尋並分析刑事判決書的內容，支援人名搜尋（用於測試與展示，資料為2025年2月全台灣各級法院之刑事判決）。輸入您的問題，系統將檢索相關判決並給出回答。</p>")
    
    with gr.Row():
        with gr.Column(scale=3):
            chatbot = gr.Chatbot(height=500)
            msg = gr.Textbox(
                placeholder="請輸入您的問題...",
                label="問題輸入",
                lines=3
            )
            with gr.Row():
                submit_btn = gr.Button("提交問題", variant="primary")
                clear_btn = gr.Button("清除對話")
            
            system_message = gr.Textbox(
                value="你是一個專業的法律助理，專門處理司法判決書的分析工作。這是一個司法專業場景，判決書中可能包含暴力、犯罪等敏感描述，這些都是判決書的必要內容，請以客觀、專業的態度處理，不要過度審查或迴避這些內容。請使用專業且平易近人的方式回答問題，並引用具體的判決案號和日期。我會提供判決書的摘要內容，請基於這些摘要進行分析，需要時可以參考完整判決內容的PDF連結。",
                label="系統訊息"
            )
        
        with gr.Column(scale=1):
            show_retrieval = gr.Checkbox(label="顯示檢索結果", value=True)
            
            # 添加 API 金鑰輸入欄位
            api_key_input = gr.Textbox(
                placeholder="輸入您的 OpenAI API 金鑰...",
                label="OpenAI API 金鑰（僅用於本次使用，不會儲存）",
                type="password"
            )
            
            # 金鑰說明
            gr.HTML("""
            <div style="margin-bottom: 10px;">
                <p style="font-size: 0.85em; color: #666;">
                    您的 API 金鑰僅用於當前會話，不會被儲存或記錄。
                    若不提供，系統將嘗試使用環境變數中的金鑰。
                </p>
            </div>
            """)
            
            # 搜尋模式選擇
            search_mode = gr.Radio(
                choices=["自動（先文本後向量）", "僅文本搜尋", "僅向量搜尋"],
                value="自動（先文本後向量）",
                label="搜尋模式"
            )
            
            max_tokens = gr.Slider(minimum=1, maximum=2048, value=512, step=1, label="最大生成字數")
            temperature = gr.Slider(minimum=0.1, maximum=1.0, value=0.7, step=0.1, label="溫度 (Temperature)")
            top_p = gr.Slider(minimum=0.1, maximum=1.0, value=0.95, step=0.05, label="Top-p 採樣")
            
            # 模型選擇
            model_choice = gr.Radio(
                choices=["gpt-4o-mini", "gpt-4o"],
                value="gpt-4o-mini",
                label="選擇模型"
            )

            gr.Markdown("""
            ### 📝 系統說明
            
            #### 參數設定
            - **API 金鑰**：輸入您的 OpenAI API 金鑰（不會被儲存）
            - **顯示檢索結果**：勾選後會顯示系統檢索到的原始判決內容
            - **搜尋模式**：選擇如何搜尋判決書（直接使用AI向量搜尋可能遭遇內容審查，如需搜尋敏感詞彙或人名，請用自動或文本搜尋）
              - **自動**：先嘗試文本搜尋，若無結果再用向量搜尋
              - **僅文本**：使用精確字詞匹配（支持部分匹配）
              - **僅向量**：使用語義相似度搜尋
            - **最大生成字數**：控制 AI 回答的最大長度（1-2048）
            - **溫度**：控制回答的創造性，越高越有創意（0.1-1.0）
            - **Top-p 採樣**：控制用詞的多樣性（0.1-1.0）
            - **模型選擇**：選擇不同的 OpenAI 模型
            
            #### 系統特點
            - **資料來源**：2025年2月份各級法院刑事判決書（[司法院資料開放平臺](https://opendata.judicial.gov.tw/)）
            - **向量檢索**：FAISS 向量資料庫
            - **文本搜尋**：支持模糊匹配，例如搜尋 ABC 也可以找到 BC
            #### 使用提示
            1. 搜尋人名時，建議使用「文本搜尋」模式
            2. 搜尋概念或法律問題時，建議使用「向量搜尋」模式
            3. 「自動」模式適合大多數情況
            """)
    
    retrieval_results = gr.HTML(label="檢索結果", elem_classes=["retrieval-results"])
    
    def user(user_message, history):
        if history is None:
            history = []
        return "", history + [[user_message, None]]
    
    def bot(history, system_msg, max_tok, temp, top_p_val, show_ret, api_key, model_name, search_mode_val):
        if history is None:
            return "錯誤：歷史記錄為空。請重新開始對話。", None
        user_message = history[-1][0]
        history[-1][1] = ""
        for response, retrieval_html in respond(
            user_message, 
            history[:-1], 
            system_msg, 
            max_tok, 
            temp, 
            top_p_val, 
            show_ret,
            api_key,
            model_name=model_name,
            search_mode=search_mode_val
        ):
            history[-1][1] = response
            if retrieval_html:
                yield history, retrieval_html
            else:
                yield history, None
    
    def clear_history():
        return [], None
    
    msg.submit(user, [msg, chatbot], [msg, chatbot], queue=False).then(
        bot, [chatbot, system_message, max_tokens, temperature, top_p, show_retrieval, api_key_input, model_choice, search_mode], [chatbot, retrieval_results]
    )
    
    submit_btn.click(user, [msg, chatbot], [msg, chatbot], queue=False).then(
        bot, [chatbot, system_message, max_tokens, temperature, top_p, show_retrieval, api_key_input, model_choice, search_mode], [chatbot, retrieval_results]
    )
    
    clear_btn.click(clear_history, None, [chatbot, retrieval_results])

if __name__ == "__main__":
    demo.launch()