tests_results.py

from agent import Agent, process_data, process_retrievers
import pandas as pd
import torch
from huggingface_hub import InferenceClient
from tqdm import tqdm
import os

def prGreen(skk): print("\033[92m {}\033[00m".format(skk))

def organize_data(df):
    # df = df[['Question', 'Réponse', 'Lien du site', 'Année ']]
    df = df[['Question', 'Réponse', 'Lien du site']]
    df = df[df['Question'].notna()]
    # df = df.set_index('Question')
    # TODO
    return df

def get_reference_from_query(query, df):
    df = df.loc[df['Question'] == query]
    # res = res[['Réponse', 'Lien du site']]
    res = [(r,clean_links(l)) for r,l in zip(df['Réponse'].to_list(), df['Lien du site'].to_list())]
    # res = res.to_dict('index')
    # res.values = [r['Lien du site'] for r in res.values]
    return res

def clean_links(links):
    start_link = "https"
    l = links.split(' ')
    # l = [li for li in l if start_link in li]
    l = [li[li.index(start_link):] for li in l if start_link in li]
    return l

def ask_a_question(use_context, reformulation, use_HyDE, use_HyDE_cut, use_context_and_wt, ask_again, query, TOP_K):
    DATA_DIR             = "data_websites"
    PATH_SAVE_CHUNKS     = "chunks_saved.json" 
    PATH_SAVE_CONTEXT    = "chunks_with_context.json"  
    PATH_IDX             = "index_faiss_data_sh"
    PATH_IDX_CONTEXT     = "index_faiss_context_sh" 
    PATH_IDX_CONTEXT_AND_WT= "index_faiss_context_and_wt_sh"   
    # embedding_model_names = ["Geotrend/distilbert-base-en-fr-cased"]
    embedding_model_names = []
    agent_name           = "Geotrend/distilbert-base-en-fr-cased"


    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    # device = 'cpu'
    # print('Using device:', device)

    # print("QUERY:", query)


    # agent = Agent(DATA_DIR, PATH_SAVE_CHUNKS, agent_name, use_context, PATH_SAVE_CONTEXT, reformulation, use_HyDE, use_HyDE_cut, use_context_and_wt)

    # path_idx = PATH_IDX_CONTEXT_AND_WT if use_context_and_wt else PATH_IDX_CONTEXT if use_context else PATH_IDX
    # # agent.add_embedding_retriever(embedding_model_name, path_idx)
    # agent.add_BM25_retriever(TOP_K)


    list_dir, chunks = process_data(DATA_DIR, PATH_SAVE_CHUNKS, PATH_SAVE_CONTEXT, use_context_and_wt, use_context)
    embedding_models, BM25_retriever = process_retrievers(embedding_model_names, chunks, TOP_K, use_context, 
                                                        use_context_and_wt, PATH_IDX, PATH_IDX_CONTEXT, 
                                                        PATH_IDX_CONTEXT_AND_WT, device)

    agent = Agent(list_dir, chunks, embedding_models, BM25_retriever, TOP_K, reformulation, 
                use_HyDE, use_HyDE_cut, ask_again)

    # agent = Agent(DATA_DIR, PATH_SAVE_CHUNKS, PATH_IDX, PATH_IDX_CONTEXT, PATH_IDX_CONTEXT_AND_WT, agent_name, 
    #             embedding_model_names, TOP_K, use_context, PATH_SAVE_CONTEXT, reformulation, use_HyDE, use_HyDE_cut, 
    #             use_context_and_wt, ask_again, device=device)

    reply, sources = agent.get_a_reply(query)

    # agent.ask_a_question(query, ask_again)

    # # agent.retrieve_data_from_embeddings(TOP_K)
    # agent.retrieve_data_from_BM25()

    # # print(agent.ranks)

    # rank = agent.RRF()
    # prompt, chunks = agent.create_prompt_with_context(rank)

    # reply = agent.ask_agent()

    # agent.retrieve_query_from_BM25(reply, 'source')
    # sources_BM25 = [agent.chunks[r].metadata['source'] for r in agent.ranks['source']]
    # sources = list({s for s in sources_BM25 if sources_BM25.count(s) > 1})
    # if not sources:
    #     sources = [sources_BM25[0]]
    # sources = agent.get_url_from_paths(sources)
    # # print(sources)

    return (reply, sources)

def ask_multiple_questions(use_context, reformulation, use_HyDE, use_HyDE_cut, use_context_and_wt, ask_again, list_queries, TOP_K=10):
    print("Ask multiple questions")
    replies_sources = []
    for query in tqdm(list_queries):
        replies_sources += [ask_a_question(use_context, reformulation, use_HyDE, use_HyDE_cut, use_context_and_wt, ask_again, query, TOP_K)]
    return replies_sources

def get_score_LLM_as_judge(reference, llm_reply):
    prompt = f"""You are an expert judge. Your task is to rate how relevant the LLM response is based on the provided reference answer(s). Rate on a scale from 1 to 5, where:

    1 = Completely irrelevant  
    2 = Mostly irrelevant  
    3 = Somewhat relevant but with noticeable issues  
    4 = Mostly relevant with minor issues  
    5 = Fully correct and accurate

    reference answer(s):
    "{reference}"

    LLM response:
    "{llm_reply}"

    Please return only the numeric score (1 to 5) and no explanation.

    Score:"""

    response = InferenceClient().chat_completion(
        model="deepseek-ai/DeepSeek-V3-0324",
        messages=[
            {
                "role": "user",
                "content": prompt,
            },
        ],
    )

    response = response.choices[0].message.content
    score = extract_number_from_string(response)

    return score

def extract_number_from_string(sentence):
    number = -1
    for char in sentence:
        if char.isdigit():
            number = int(char)
    return number

def test_eval_RAG_simple(data, list_question):
    query                = "Quelles sont les missions du collège Sciences de l’Homme ?"
    reference_response = """Dans le domaine des sciences humaines et sociales, le collège Sciences de l’Homme a pour missions :

    * d’élaborer et coordonner la politique de formation initiale et tout au long de la vie, mise en œuvre par les composantes internes de formation rattachées au collège Sciences de l’Homme,
    * de construire et porter cette politique en cohérence avec la politique générale de formation de l’établissement,
    * de participer au dialogue de gestion avec la direction de l’université pour assurer les moyens financiers, humains, logistiques et patrimoniaux nécessaires à la réalisation de ses missions,
    * de participer à la coordination de l’orientation et à l’aide à l’insertion professionnelle des étudiants,
    * d’assurer le développement des formations internationales et promouvoir la mobilité des étudiants,
    * de promouvoir la vie de campus sur les différents sites du collège, en lien étroit avec la politique de l’établissement,
    * d’organiser la liaison et les interactions avec le département Sciences Humaines et Sociales, et tout autre département partenaire, notamment dans le cadre de la préparation du plan de gestion des emplois et lors des procédures d’accréditation, afin de garantir l’articulation entre la formation et la recherche.
    """
    ref_sources = ["../data_sh/le-college_nos-missions.json"]


    # use_context          = True
    use_context          = False
    # reformulation        = True
    reformulation        = False
    # use_HyDE             = True
    use_HyDE             = False
    reply, sources = ask_a_question(use_context, reformulation, use_HyDE, query)
    print("Reply:\n"+reply)
    print("\n\nSources:\n"+str(sources))

    score = get_score_LLM_as_judge(reference_response, reply)
    print("Score:",score)

    acc_sources = check_sources(ref_sources, sources)
    print("Sources acc:",acc_sources)

# def check_sources(ref_sources, llm_sources):
#     if len(ref_sources) != len(llm_sources):
#         return 0
#     for ref in ref_sources:
#         if ref not in llm_sources:
#             return 0
#     return 1

def check_sources(ref_sources, llm_sources):
    cpt = 0
    for ref in ref_sources:
        if ref in llm_sources:
            cpt += 1
    return cpt/len(ref_sources)

def write_replies(replies, list_question, filename='replies_tests_results_simple.txt'):
    with open(filename, 'a') as the_file:
        the_file.write('\n**** REPLIES ****\n')
    for q, (r,s) in tqdm(zip(list_question,replies)):
        with open(filename, 'a') as the_file:
            the_file.write('\nQuery:'+q+'\n')
            # the_file.write('Dept:'+str(s)+'\n')
            the_file.write('Reply:\n'+str(r)+'\n')
            the_file.write('Sources:'+str(s)+'\n')


def basic_test(use_context, reformulation, use_HyDE, use_HyDE_cut, use_context_and_wt, data, list_question, name, df_results, ask_again=False):
    replies = ask_multiple_questions(use_context, reformulation, use_HyDE, use_HyDE_cut, use_context_and_wt, ask_again, list_question)
    write_replies(replies, list_question)
    acc_reply, acc_sources = calcul_accuracies(list_question, replies, data)
    print(f"Accuracy replies: {acc_reply}\nAccuracy sources: {acc_sources}")
    df_results = pd.concat([pd.DataFrame([[name, acc_reply, acc_sources]], columns=df_results.columns), df_results], ignore_index=True)
    df_results.to_csv("results.csv", index=False)
    return df_results

def calcul_accuracies(list_question, replies, data):
    print("Calcul accuracies")
    score_sources = 0
    score_reply = 0
    for q, (r,s) in tqdm(zip(list_question,replies)):
        # print("Query:",q)
        # print(f"Reply:\n{r}\n\nSources:{s}")

        ref_reply_sources = get_reference_from_query(q, data)

        score_sources_q, score_reply_q = calcul_accuracies_per_query(s, r, ref_reply_sources)
        score_sources += score_sources_q
        score_reply += score_reply_q

    score_sources /= len(list_question)
    score_reply /= len(list_question)

    return score_reply, score_sources

def calcul_accuracies_per_query(llm_sources, llm_reply, ref_reply_sources):
    score_sources_q = 0
    score_reply_q = 0
    for ref_rep, ref_source in ref_reply_sources:
        score_src_q = check_sources(ref_source, llm_sources)
        score_rep_q = get_score_LLM_as_judge(ref_rep, llm_reply)
        if check_best_answer(score_sources_q, score_reply_q, score_src_q, score_rep_q):
            score_sources_q = score_src_q
            score_reply_q = score_rep_q
    return score_sources_q, score_reply_q

def check_best_answer(score_sources_q, score_reply_q, score_src_q, score_rep_q, coef_src=5):
    new_score = score_rep_q + score_src_q * coef_src
    old_score = score_reply_q + score_sources_q * coef_src
    if new_score > old_score:
        return True
    return False

def test_RAG_simple(data, list_question, df_results):
    name = "test_RAG_simple:"
    print(name)
    use_context          = False
    reformulation        = False
    use_HyDE             = False
    use_HyDE_cut         = False
    use_context_and_wt   = False
    return basic_test(use_context, reformulation, use_HyDE, use_HyDE_cut, use_context_and_wt, data, list_question, name, df_results)

def test_RAG_all(data, list_question, df_results):
    name = "test_RAG_all:"
    print(name)
    use_context          = True
    reformulation        = True
    use_HyDE             = True
    use_HyDE_cut         = True
    use_context_and_wt   = False
    return basic_test(use_context, reformulation, use_HyDE, use_HyDE_cut, use_context_and_wt, data, list_question, name, df_results)

def test_RAG_context(data, list_question, df_results):
    name = "test_RAG_context:"
    print(name)
    use_context          = True
    reformulation        = False
    use_HyDE             = False
    use_HyDE_cut         = False
    use_context_and_wt   = False
    return basic_test(use_context, reformulation, use_HyDE, use_HyDE_cut, use_context_and_wt, data, list_question, name, df_results)

def test_RAG_reformulation(data, list_question, df_results):
    name = "test_RAG_reformulation:"
    print(name)
    use_context          = False
    reformulation        = True
    use_HyDE             = False
    use_HyDE_cut         = False
    use_context_and_wt   = False
    return basic_test(use_context, reformulation, use_HyDE, use_HyDE_cut, use_context_and_wt, data, list_question, name, df_results)

def test_RAG_HyDE(data, list_question, df_results):
    name = "test_RAG_HyDE:"
    print(name)
    use_context          = False
    reformulation        = False
    use_HyDE             = True
    use_HyDE_cut         = False
    use_context_and_wt   = False
    return basic_test(use_context, reformulation, use_HyDE, use_HyDE_cut, use_context_and_wt, data, list_question, name, df_results)

def test_RAG_HyDE_cut(data, list_question, df_results):
    name = "test_RAG_HyDE_cut:"
    print(name)
    use_context          = False
    reformulation        = False
    use_HyDE             = False
    use_HyDE_cut         = True
    use_context_and_wt   = False
    return basic_test(use_context, reformulation, use_HyDE, use_HyDE_cut, use_context_and_wt, data, list_question, name, df_results)

def test_RAG_context_reformulation(data, list_question, df_results):
    name = "test_RAG_context_reformulation:"
    print(name)
    use_context          = True
    reformulation        = True
    use_HyDE             = False
    use_HyDE_cut         = False
    use_context_and_wt   = False
    return basic_test(use_context, reformulation, use_HyDE, use_HyDE_cut, use_context_and_wt, data, list_question, name, df_results)

def test_RAG_context_HyDE(data, list_question, df_results):
    name = "test_RAG_context_HyDE:"
    print(name)
    use_context          = True
    reformulation        = False
    use_HyDE             = True
    use_HyDE_cut         = False
    use_context_and_wt   = False
    return basic_test(use_context, reformulation, use_HyDE, use_HyDE_cut, use_context_and_wt, data, list_question, name, df_results)

def test_RAG_context_HyDE_cut(data, list_question, df_results):
    name = "test_RAG_context_HyDE_cut:"
    print(name)
    use_context          = True
    reformulation        = False
    use_HyDE             = False
    use_HyDE_cut         = True
    use_context_and_wt   = False
    return basic_test(use_context, reformulation, use_HyDE, use_HyDE_cut, use_context_and_wt, data, list_question, name, df_results)

def test_RAG_reformulation_HyDE(data, list_question, df_results):
    name = "test_RAG_reformulation_HyDE:"
    print(name)
    use_context          = False
    reformulation        = True
    use_HyDE             = True
    use_HyDE_cut         = False
    use_context_and_wt   = False
    return basic_test(use_context, reformulation, use_HyDE, use_HyDE_cut, use_context_and_wt, data, list_question, name, df_results)

def test_RAG_reformulation_HyDE_cut(data, list_question, df_results):
    name = "test_RAG_reformulation_HyDE_cut:"
    print(name)
    use_context          = False
    reformulation        = True
    use_HyDE             = False
    use_HyDE_cut         = True
    use_context_and_wt   = False
    return basic_test(use_context, reformulation, use_HyDE, use_HyDE_cut, use_context_and_wt, data, list_question, name, df_results)

def test_RAG_HyDE_HyDE_cut(data, list_question, df_results):
    name = "test_RAG_HyDE_HyDE_cut:"
    print(name)
    use_context          = False
    reformulation        = False
    use_HyDE             = True
    use_HyDE_cut         = True
    use_context_and_wt   = False
    return basic_test(use_context, reformulation, use_HyDE, use_HyDE_cut, use_context_and_wt, data, list_question, name, df_results)

def test_RAG_context_reformulation_HyDE(data, list_question, df_results):
    name = "test_RAG_context_reformulation_HyDE:"
    print(name)
    use_context          = True
    reformulation        = True
    use_HyDE             = True
    use_HyDE_cut         = False
    use_context_and_wt   = False
    return basic_test(use_context, reformulation, use_HyDE, use_HyDE_cut, use_context_and_wt, data, list_question, name, df_results)

def test_RAG_context_HyDE_HyDE_cut(data, list_question, df_results):
    name = "test_RAG_context_HyDE_HyDE_cut:"
    print(name)
    use_context          = True
    reformulation        = False
    use_HyDE             = True
    use_HyDE_cut         = True
    use_context_and_wt   = False
    return basic_test(use_context, reformulation, use_HyDE, use_HyDE_cut, use_context_and_wt, data, list_question, name, df_results)

def test_RAG_context_reformulation_HyDE_cut(data, list_question, df_results):
    name = "test_RAG_context_reformulation_HyDE_cut:"
    print(name)
    use_context          = True
    reformulation        = True
    use_HyDE             = False
    use_HyDE_cut         = True
    use_context_and_wt   = False
    return basic_test(use_context, reformulation, use_HyDE, use_HyDE_cut, use_context_and_wt, data, list_question, name, df_results)

def test_RAG_reformulation_HyDE_HyDE_cut(data, list_question, df_results):
    name = "test_RAG_reformulation_HyDE_HyDE_cut:"
    print(name)
    use_context          = False
    reformulation        = True
    use_HyDE             = True
    use_HyDE_cut         = True
    use_context_and_wt   = False
    return basic_test(use_context, reformulation, use_HyDE, use_HyDE_cut, use_context_and_wt, data, list_question, name, df_results)

def test_RAG_context_and_wt(data, list_question, df_results):
    name = "test_RAG_context_and_wt:"
    print(name)
    use_context          = False
    reformulation        = False
    use_HyDE             = False
    use_HyDE_cut         = False
    use_context_and_wt   = True
    return basic_test(use_context, reformulation, use_HyDE, use_HyDE_cut, use_context_and_wt, data, list_question, name, df_results)

def test_RAG_context_and_wt_ask_again(data, list_question, df_results):
    name = "test_RAG_context_and_wt_ask_again:"
    print(name)
    use_context          = False
    reformulation        = False
    use_HyDE             = False
    use_HyDE_cut         = False
    use_context_and_wt   = True
    ask_again            = True
    return basic_test(use_context, reformulation, use_HyDE, use_HyDE_cut, use_context_and_wt, data, list_question, name, df_results, ask_again)


if __name__ == "__main__":
    # filename = 'questions.csv'
    # filename = 'QuestionsPagesCollègeSH.csv'
    filename = 'QA_generated_SH.csv'
    # filename = 'QA_generated.csv'
    data_test = pd.read_csv(filename)
    data_test = organize_data(data_test)
    print(data_test)
    list_question = data_test['Question'].drop_duplicates().to_list()

    # list_question = list_question[:3]

    if os.path.exists("results.csv"):
        df_results = pd.read_csv("results.csv")
    else:
        df_results = pd.DataFrame(columns=['Name', 'Accuracy replies', 'Accuracy sources'])

    with open('replies_tests_results_simple.txt', 'a') as the_file:
        the_file.write(str(data_test)+'\n')

    # data = get_reference_from_query(list_question[0], data_test)
    # print(data)

    # print("*********************")
    # data = get_reference_from_query(list_question[1], data_test)
    # print(data)

    # print("*********************")
    # data = get_reference_from_query(list_question[14], data_test)
    # print(data)

    ################################################################
    # test_RAG_simple(data_test,list_question)
    # test_RAG_all(data_test, list_question)
    # test_RAG_context(data_test, list_question)
    # test_RAG_reformulation(data_test, list_question)
    # test_RAG_HyDE(data_test, list_question)
    # test_RAG_context_reformulation(data_test, list_question)
    # test_RAG_context_HyDE(data_test, list_question)
    # test_RAG_reformulation_HyDE(data_test, list_question)

    ###############################################################

    # df_results = test_RAG_simple(data_test, list_question, df_results)
    # df_results = test_RAG_all(data_test, list_question, df_results)
    # df_results = test_RAG_context(data_test, list_question, df_results)
    # df_results = test_RAG_reformulation(data_test, list_question, df_results)
    # df_results = test_RAG_HyDE(data_test, list_question, df_results)
    # df_results = test_RAG_HyDE_cut(data_test, list_question, df_results)
    # df_results = test_RAG_context_reformulation(data_test, list_question, df_results)
    # df_results = test_RAG_context_HyDE(data_test, list_question, df_results)
    # df_results = test_RAG_context_HyDE_cut(data_test, list_question, df_results)
    # df_results = test_RAG_reformulation_HyDE(data_test, list_question, df_results)
    # df_results = test_RAG_reformulation_HyDE_cut(data_test, list_question, df_results)
    # df_results = test_RAG_HyDE_HyDE_cut(data_test, list_question, df_results)
    # df_results = test_RAG_context_reformulation_HyDE(data_test, list_question, df_results)
    # df_results = test_RAG_context_HyDE_HyDE_cut(data_test, list_question, df_results)
    # df_results = test_RAG_context_reformulation_HyDE_cut(data_test, list_question, df_results)
    # df_results = test_RAG_reformulation_HyDE_HyDE_cut(data_test, list_question, df_results)
    df_results = test_RAG_context_and_wt(data_test, list_question, df_results)
    # df_results = test_RAG_context_and_wt_ask_again(data_test, list_question, df_results)

    ##############################################################

    # test_eval_RAG_simple(data_test, list_question)