Hugging Face's logo

Spaces:
Bogdan555
/
grantforge-api
Running

App Files Community
grantforge-api
File size: 4,370 Bytes
afd56bc
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
# ruff: noqa: E402
import os
import json
import logging
from datasets import Dataset
from dotenv import load_dotenv

load_dotenv()

# Konfiguracja logowania
logging.basicConfig(level=logging.INFO, format="%(name)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

# Pobranie naszych domyślnych modułów
from core.llm_router import get_llm
from rag_pipeline.hybrid_retriever import get_hybrid_retriever
from rag_pipeline.retriever import generate_answer
from langchain_google_genai import GoogleGenerativeAIEmbeddings

try:
    from ragas import evaluate
    from ragas.metrics import (
        context_precision,
        faithfulness,
        answer_relevancy,
        context_recall,
    )
    from ragas.llms.prompt import LangchainLLMWrapper
    from ragas.embeddings.base import LangchainEmbeddingsWrapper
except ImportError:
    logger.error(
        "Brak zainstalowanych bibliotek 'ragas' / 'datasets'. Zainstaluj środowisko testowe."
    )
    exit(1)


def run_evaluation():
    dataset_path = os.path.join(
        os.path.dirname(__file__), "..", "tests", "golden_dataset.json"
    )

    if not os.path.exists(dataset_path):
        logger.error(f"Nie znaleziono pliku {dataset_path}")
        return

    with open(dataset_path, "r", encoding="utf-8") as f:
        golden_data = json.load(f)

    logger.info(f"Loaded {len(golden_data)} pytań z Golden Dataset.")

    # Inicjalizacja retrievera i łańcucha RAG do generowania odpowiedzi bieżącego systemu
    retriever = get_hybrid_retriever()
    if not retriever:
        logger.error(
            "Nie znaleziono funkcjonującego wektora w systemie. Uruchom najpierw ingest."
        )
        return

    questions = []
    answers = []
    contexts = []
    ground_truths = []

    logger.info("🛠️ Trwa generowanie odpowiedzi RAG celem poddania ewaluacji...")
    for idx, item in enumerate(golden_data):
        q = item["question"]
        gt = item["ground_truth_answer"]

        docs = retriever.invoke(q)
        retrieved_texts = [d.page_content for d in docs]

        # Generowanie używając naszego agenta
        answer = generate_answer(q, docs)

        questions.append(q)
        answers.append(answer)
        contexts.append(retrieved_texts)
        # Ragas < 0.2 używa ground_truths, Ragas >= 0.2 używa ground_truth, dodajmy format listowy, standardowy dla starszych
        # Jeśli użyjemy Ragas >= 0.2.x, ground_truth powinno być typu string. Dopasujmy do nowszego RAGAS:
        ground_truths.append(gt)

    data = {
        "question": questions,
        "answer": answers,
        "contexts": contexts,
        "ground_truth": ground_truths,
    }

    dataset = Dataset.from_dict(data)

    # Skonfigurowanie RAGAS jako evaluatora używającego Google Gemini z naszego routera (unikanie kosztów OpenAI)
    logger.info(
        "📊 Rozpoczynam ocenę metryk (RAGAS). Używam LLM od Google via LangChainWrapper..."
    )
    eval_llm = get_llm(task_type="critical")
    eval_embeddings = GoogleGenerativeAIEmbeddings(
        model="text-embedding-004", google_api_key=os.environ.get("GOOGLE_API_KEY")
    )

    try:
        ragas_eval_llm = LangchainLLMWrapper(eval_llm)
        ragas_eval_emb = LangchainEmbeddingsWrapper(eval_embeddings)

        result = evaluate(
            dataset=dataset,
            metrics=[context_precision, faithfulness, answer_relevancy, context_recall],
            llm=ragas_eval_llm,
            embeddings=ragas_eval_emb,
        )

        print("\n\n=== 🏆 WYNIKI EWALUACJI RAGAS ===")
        print(result)

        import datetime

        timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")

        # Safe format for json
        result_dict = result.to_pandas().to_dict(orient="records")
        out_json = f"rag_evaluation_results_{timestamp}.json"
        with open(out_json, "w", encoding="utf-8") as f:
            json.dump(result_dict, f, ensure_ascii=False, indent=2)

        out_csv = f"rag_evaluation_results_{timestamp}.csv"
        df = result.to_pandas()
        df.to_csv(out_csv, index=False)

        print(
            f"💡 Analiza szczegółowa została zapisana w lokalizacji: {out_json} oraz {out_csv}"
        )
    except Exception as e:
        logger.error(f"Nie udało się zakończyć ewaluacji: {e}")


if __name__ == "__main__":
    run_evaluation()