app.py

import os
import sys
import math
import requests
import gradio as gr
import pandas as pd
from datasets import Dataset
from tqdm import tqdm
from ragas import evaluate, SingleTurnSample
from ragas.llms import LangchainLLMWrapper
from ragas.embeddings import LangchainEmbeddingsWrapper
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from ragas.metrics import (
    ResponseRelevancy, LLMContextPrecisionWithReference, LLMContextRecall,
    ContextEntityRecall, Faithfulness, NoiseSensitivity, SemanticSimilarity, FactualCorrectness
)

# 設定輸出編碼為 UTF-8（解決中文顯示問題）
sys.stdout.reconfigure(encoding="utf-8")

# 支援從Google Drive下載 Ground Truth
gt_url = os.environ.get("GT_URL")
gt_path = "tender_groundtruth.csv"

if gt_url and not os.path.exists(gt_path):
    print("嘗試下載 Ground Truth...")
    r = requests.get(gt_url)
    print("HTTP 狀態碼：", r.status_code)
    if r.status_code != 200:
        print("下載失敗內容預覽：", r.text[:500])
    else:
        with open(gt_path, "wb") as f:
            f.write(r.content)

# 綁定實驗室Google帳號(Python TA)Google Sheet，以記錄評估logs
def log_to_google_sheet(question, answer, contexts, scores):
    url = os.environ.get("G_SHEET_URL")
    if not url:
        print("G_SHEET_URL 未設定，略過記錄")
        return
    try:
        payload = {
            "question": question,
            "answer": answer,
            "contexts": contexts,
            "faithfulness": scores.get("Faithfulness"),
            "answer_relevancy": scores.get("Answer Relevancy"),
            "semantic_similarity": scores.get("Semantic Similarity"),
            "context_precision": scores.get("Context Precision"),
            "context_recall": scores.get("Context Recall"),
            "context_entity_recall": scores.get("Context Entity Recall")
        }
        response = requests.post(url, json=payload)
        print("成功寫入 Google Sheet:", response.status_code)
    except Exception as e:
        print("寫入 Google Sheet 失敗:", str(e))

def RAG_evaluation(uploaded_file, user_api_key):
    try:
        os.environ["OPENAI_API_KEY"] = user_api_key
        print("評估開始")

        if not os.path.exists(gt_path):
            print("找不到 Ground Truth！")
            return pd.DataFrame(), None

        gt_df = pd.read_csv(gt_path)
        df = pd.read_csv(uploaded_file.name, converters={"Context": eval})
        print(f"上傳檔案筆數：{len(df)}，GT 檔案筆數：{len(gt_df)}")

        merged_df = pd.merge(df, gt_df[["Question", "Answer"]], on="Question", suffixes=("", "_GroundTruth"))
        merged_df = merged_df.rename(columns={"Answer_GroundTruth": "GroundTruth"})
        print(f"成功合併筆數：{len(merged_df)} / {len(df)}")
        if len(merged_df) < len(df):
            missing = df[~df["Question"].isin(merged_df["Question"])]
            print("未合併題目：", missing["Question"].tolist())
        if merged_df.empty:
            return pd.DataFrame([{"錯誤訊息": "合併後無資料，請確認題目與 GT 是否對應"}]), None

        llm_wrapper = LangchainLLMWrapper(ChatOpenAI(model="gpt-4o-mini-2024-07-18"))
        embedding_wrapper = LangchainEmbeddingsWrapper(OpenAIEmbeddings(model="text-embedding-3-large"))

        batch_size = 10
        records = []
        for batch_start in tqdm(range(0, len(merged_df), batch_size), desc="RAGAS Batch Evaluating"):
            batch_df = merged_df.iloc[batch_start:batch_start + batch_size]

            samples = []
            for _, row in batch_df.iterrows():
                if not isinstance(row["Context"], list):
                    print(f"Context 非 list，跳過。值：{row['Question']}")
                    continue

                sample = SingleTurnSample(
                    user_input=row["Question"],
                    response=row["Answer"],
                    retrieved_contexts=row["Context"],
                    reference=row["GroundTruth"],
                )
                samples.append(sample)
        
            try:
                dataset = Dataset.from_list([s.to_dict() for s in samples])
                result = evaluate(
                    dataset=dataset,
                    metrics=[
                        LLMContextPrecisionWithReference(), LLMContextRecall(), ContextEntityRecall(),
                        NoiseSensitivity(), Faithfulness(), ResponseRelevancy(),
                        SemanticSimilarity(), FactualCorrectness()
                    ],
                    llm=llm_wrapper,
                    embeddings=embedding_wrapper,
                    show_progress=False
                )

                result_df = result.to_pandas()

                for i, row in enumerate(result_df.itertuples()):
                    input_row = batch_df.iloc[i]
                    record = {
                        "Question": input_row["Question"],
                        "Faithfulness": getattr(row, "faithfulness", None),
                        "Answer Relevancy": getattr(row, "answer_relevancy", None),
                        "Semantic Similarity": getattr(row, "semantic_similarity", None),
                        # "Factual Correctness": getattr(row, "factual_correctness", None),
                        "Context Precision": getattr(row, "llm_context_precision_with_reference", None),
                        "Context Recall": getattr(row, "context_recall", None),
                        "Context Entity Recall": getattr(row, "context_entity_recall", None),
                        # "Noise Sensitivity": getattr(row, "noise_sensitivity_relevant", None)
                    }
                    
                    for key in list(record.keys()):
                        val = record[key]
                        if isinstance(val, float) and not math.isfinite(val):
                            record[key] = ""
                
                    records.append(record)

                    log_to_google_sheet(
                        question=input_row["Question"],
                        answer=input_row["Answer"],
                        contexts=input_row["Context"],
                        scores=record
                    )

            except Exception as e:
                print(f"批次評估失敗（第 {batch_start+1} 筆起）：{e}")
                continue

        score_df = pd.DataFrame(records).fillna("")
        print("完成評估筆數：", len(score_df))
        
        numeric_cols = score_df.drop(columns=["Question"]).select_dtypes(include="number")
        if not numeric_cols.empty:
            avg_row = numeric_cols.mean().to_dict()
            avg_row["Question"] = "Average"
            score_df = pd.concat([score_df, pd.DataFrame([avg_row])], ignore_index=True)

        output_path = "result_output.csv"
        score_df.to_csv(output_path, index=False, encoding="utf-8-sig")
        print("評估結果已儲存為 CSV：", output_path)

        return score_df, output_path

    except Exception as e:
        print("評估函式整體錯誤：", str(e))
        return pd.DataFrame([{"錯誤訊息": f"系統錯誤：{str(e)}"}]), None

# handle exception並執行RAG評估
def check_csv_and_run(file, key):
    if file is None:
        return pd.DataFrame([{"錯誤訊息": "請上傳檔案！"}]), None
    
    if not key or key.strip() == "":
        return pd.DataFrame([{"錯誤訊息": "請輸入 OpenAI API Key"}]), None

    try:
        df = pd.read_csv(file.name, encoding="utf-8-sig")
        df.columns = [col.strip() for col in df.columns]

        required_columns = {"Question", "Context", "Answer"}
        actual_columns = set(df.columns)

        if actual_columns != required_columns:
            return pd.DataFrame([{"錯誤訊息": f"欄位錯誤：應包含欄位 {required_columns}，實際為 {actual_columns}"}]), None

        if df.shape[0] == 0:
            return pd.DataFrame([{"錯誤訊息": "檔案中沒有資料列!"}]), None

        invalid_rows = df[df["Question"].notnull() & (df["Answer"].isnull() | df["Context"].isnull())]
        if len(invalid_rows) > 0:
            missing_questions = "\n".join(f"- {q}" for q in invalid_rows["Question"].tolist())
            return pd.DataFrame([{"錯誤訊息": f"發現 {len(invalid_rows)} 筆資料中 Answer 或 Context 為空：\n{missing_questions}"}]), None
        
        try:
            for i, val in df["Context"].dropna().items():
                if not isinstance(eval(val), list):
                    return pd.DataFrame([{"錯誤訊息": f"第 {i + 1} 筆 Context 欄格式錯誤，請確認其內容應為 list"}]), None
        except Exception as e:
            return pd.DataFrame([{"錯誤訊息": f"Context 欄格式解析錯誤，請確認其為有效的 list 格式，例如 ['A', 'B']：{str(e)}"}]), None

    except Exception as e:
        return pd.DataFrame([{"錯誤訊息": f"發生錯誤：{str(e)}"}]), None

    # 若上傳之待評估檔案無錯誤，執行評估
    try:
        return RAG_evaluation(file, key)
    except Exception as e:
        return pd.DataFrame([{"錯誤訊息": f"RAG 評估失敗：{str(e)}"}]), None

# Gradio 介面
with gr.Blocks() as demo:
    gr.Markdown("## 📐 RAG系統評估工具")
    gr.Markdown("""
    ### 📄 使用說明
    請上傳您RAG系統產出的結果檔案（包含 Question, Context, Answer 欄位），並填入您的OpenAI API Key，以評估您的RAG系統。  
    #### ⏳ 完整評估需要數小時，無即時回應並不是當機，請耐心等候。
    """)

    file_input = gr.File(label="上傳 Evaluation_Dataset.csv")
    api_key_input = gr.Textbox(label="OpenAI API Key", type="password")
    submit_btn = gr.Button("開始評估")

    result_output = gr.Dataframe(label="評估結果")
    download_link = gr.File(label="下載評估結果(CSV)")

    def wrapped_fn(file, key):
        return RAG_evaluation(file, key)

    submit_btn.click(
        fn=check_csv_and_run,
        inputs=[file_input, api_key_input],
        outputs=[result_output, download_link]
)

demo.launch()