first commit

app.py

@@ -0,0 +1,184 @@
+import os
+import sys
+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
+)
+
+sys.stdout.reconfigure(encoding="utf-8")
+
+# 嘗試從Google Drive下載 Ground Truth
+gt_url = os.environ.get("GT_URL")
+gt_path = "ragas_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)
+
+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"))
+
+        records = []
+        for idx, row in tqdm(merged_df.iterrows(), total=len(merged_df), desc="Evaluating"):
+            try:
+                if not isinstance(row["Context"], list):
+                    print(f"第 {idx + 1} 筆 Context 非 list，跳過。值：{row['Context']}")
+                    continue
+                
+                sample = SingleTurnSample(
+                    user_input=row["Question"],
+                    response=row["Answer"],
+                    retrieved_contexts=row["Context"],
+                    reference=row["GroundTruth"]
+                )
+                dataset = Dataset.from_list([sample.to_dict()])
+                result = evaluate(
+                    dataset=dataset,
+                    metrics=[
+                        LLMContextPrecisionWithReference(), LLMContextRecall(), ContextEntityRecall(),
+                        NoiseSensitivity(), Faithfulness(), ResponseRelevancy(),
+                        SemanticSimilarity(), FactualCorrectness()
+                    ],
+                    llm=llm_wrapper,
+                    embeddings=embedding_wrapper,
+                    show_progress=False
+                )
+
+                score_row = result.to_pandas().iloc[0].to_dict()
+                records.append({
+                    "Question": row["Question"],
+                    "Faithfulness": score_row.get("faithfulness"),
+                    "Answer Relevancy": score_row.get("answer_relevancy"),
+                    "Semantic Similarity": score_row.get("semantic_similarity"),
+                    # "Factual Correctness": score_row.get("factual_correctness"),
+                    "Context Precision": score_row.get("llm_context_precision_with_reference"),
+                    "Context Recall": score_row.get("context_recall"),
+                    "Context Entity Recall": score_row.get("context_entity_recall"),
+                    # "noise_sensitivity_relevant": score_row.get("noise_sensitivity_relevant")
+                })
+
+            except Exception as e:
+                print(f"第 {idx + 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
+    
+def check_csv_and_run(file, key):
+    print("開始檢查CSV檔案格式並執行評估")
+    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)")
+
+    submit_btn.click(
+        fn=check_csv_and_run,
+        inputs=[file_input, api_key_input],
+        outputs=[result_output, download_link]
+)
+
+demo.launch()