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)