import pandas as pd import hashlib import os from openai import OpenAI from llama_index.core import Document from llama_index.core.indices import VectorStoreIndex from llama_index.core.postprocessor import LLMRerank from llama_index.core import QueryBundle df=pd.read_excel("metropole_with_range.xlsx") df = df.applymap(lambda x: x.lower().strip() if isinstance(x, str) else x) # df = df.head(10) df['Direction DGA'] = df['Direction DGA'].replace('ressources', 'ressources humaines et modernisation') df['Min'] = df['Min'].apply(lambda x: '0' if x=='nan' else str(x)) df['Max'] = df['Max'].apply(lambda x: '0' if x=='nan' else str(x)) df['Min'] = df['Min'].apply(lambda x: '221000' if ',' in x else str(x)) df['Max'] = df['Max'].apply(lambda x: '100000000' if ',' in x else str(x)) df['Min'] = df['Min'].apply(lambda x: int(x)) df['Max'] = df['Max'].apply(lambda x: int(x)) df2=df[['Direction DGA','Fonction','Liste Service Text','Item Text','Signataire', 'Min', 'Max']] df2['is_director_label'] = df2['Fonction'].apply( lambda x: 'Yes' if isinstance(x, str) and ('directeur' in x.lower() or 'directrice' in x.lower()) else 'No' ) df_directeur=df2[df2['is_director_label']=='Yes'] df_directeur[['Direction DGA','Liste Service Text']] df_directeur_filtered = df_directeur[df_directeur['Direction DGA'] == 'direction générale des services'] def hashage(row): return hashlib.sha1(row.encode("utf-8")).hexdigest() unique_combinations = df_directeur[['Direction DGA', 'Fonction', 'Liste Service Text']].drop_duplicates() unique_combinations['hash'] = unique_combinations['Direction DGA'] + ' ' + unique_combinations['Fonction'] + unique_combinations['Liste Service Text'] unique_combinations['hash'] = unique_combinations['hash'].apply(lambda x:hashage(x)) # unique_combinations['description'] = f"{unique_combinations['hash']} : {}" unique_combinations_list = unique_combinations.apply(lambda x: f"{x['hash']} : {x['Direction DGA']} - {x['Fonction']} - {x['Liste Service Text']}", axis=1).tolist() def get_direction_dga(question, unique_combinations_list, unique_combinations): client = OpenAI() completion = client.chat.completions.create( model="gpt-4o", frequency_penalty=0.98, messages=[ {"role": "system", "content": f"""Tu es expert en lexique en français. Voici la liste suivante ecrite sous forme [id : description] : {unique_combinations_list}. Selon le champ lexical, réponds à la question en donnant l'ID de la réponse qui correspond le mieux. Donne uniquement la réponse."""}, { "role": "user", "content": question } ] ) id = completion.choices[0].message.content # print(id) # Filter to get the row where 'hash' matches the ID and return the Direction DGA row = unique_combinations[unique_combinations['hash'] == id] # Assuming 'Direction' is the column holding the DGA information direction_dga = row['Direction DGA'].values[0] if not row.empty else None print(direction_dga) return direction_dga # Créer les documents à partir des colonnes spécifiées dans le dataframe documents = [] for _, row in df.iterrows(): try: # Extraire les champs pertinents du dataframe item_text = row.get('Item Text', 'Texte indisponible') theme_title = row.get('Theme Title', '') sous_theme_title = row.get('SousTheme Title', '') liste_service_text = row.get('Liste Service Text', '') signataire = row.get('Signataire', 'Signataire inconnu') fonction = row.get('Fonction', '') suppleant = row.get('Suppleant', '') collectivite = row.get('Collectivite', '') date_debut = row.get('Date Debut', '') # Construire le texte avec les champs spécifiés text = f""" item: {item_text}, service: {liste_service_text}, signataire: {signataire}, fonction: {fonction}, collectivité: {collectivite}, suppléant: {suppleant} """ # Créer le document avec texte et seulement 'Direction DGA' dans les métadonnées document = Document( text=text, metadata={ "Direction DGA": row.get('Direction DGA', 'Direction DGA inconnue'), "Min":row['Min'], "Max":row['Max'], } ) documents.append(document) except Exception as e: print(f"Erreur lors du traitement de la ligne {row['Numero']}: {e}") # Si des documents sont créés, construire l'index if documents: index = VectorStoreIndex.from_documents(documents, show_progress=True) else: print("Aucun document valide n'a été généré.") from llama_index.core.vector_stores import ( MetadataFilter, MetadataFilters, FilterOperator, ) def getValue(question): client = OpenAI() completion = client.chat.completions.create( model="gpt-4o", # top_p=0.98, frequency_penalty=0.98, messages=[ { "role": "system", "content": f""" Sois une requete donnee qui peut contenir ou non le prix de quelqonce objet reponds uniquement par cette somme (sans devise et en chiffre) si elle existe sinon reponds par 0 """ }, { "role": "user", "content": question } ] ) return completion.choices[0].message.content def get_all_text(new_nodes): print(new_nodes) texts = [] for i, node in enumerate(new_nodes, 1): texts.append(f"\nDocument {i} : {node.get_text()}") return ' '.join(texts) def retrieve_filtered_documents(question, index, unique_combinations_list, unique_combinations, similarity_top_k=10, reranker_top_n=5): value = int(getValue(question)) print(value) # Get direction DGA based on the question direction_dga = get_direction_dga(question, unique_combinations_list, unique_combinations) # Create metadata filters filters = MetadataFilters( filters=[ MetadataFilter(key="Direction DGA", operator=FilterOperator.EQ, value=direction_dga), MetadataFilter(key="Min", value=value, operator=FilterOperator.LTE), MetadataFilter(key="Max", value=value, operator=FilterOperator.GTE), ] ) # Retrieve relevant documents using the filters retrieved_nodes = index.as_retriever(similarity_top_k=similarity_top_k, filters=filters).retrieve(question) # reranker = LLMRerank( # choice_batch_size=5, # top_n=reranker_top_n, # ) # retrieved_nodes = reranker.postprocess_nodes( # retrieved_nodes, QueryBundle(question) # ) # Extract and return all text from the query result return get_all_text(retrieved_nodes) from openai import OpenAI def answer(question, docs): client = OpenAI() completion = client.chat.completions.create( model="gpt-4o", # top_p=0.98, frequency_penalty=0.98, messages=[ { "role": "system", "content": f""" agit come un expert financier et un agent de la metropole expert dans la recherche des deleguation de signature . L'utilisateur posera une question et tu devras trouver la réponse dans les documents suivants.Focalise sur les service et la direction du signataire que l'utilisateur cherche. Tu ne dois pas poser de question en retour.Tu ne dois pas mentionner le numéro des documents. Tu t'exprimes dans la même langue que l'utilisateur. DOCUMENTS : {docs} instruction : -prend en consideration la fonction du signataire si elle match avec la delegation de signature -ta reponse peut se trouver sur plusieurs document -donne les signataire et les supplient et reponds de facon directe. -reponds par une liste structuree """ }, { "role": "user", "content": question } ] ) print(docs) return completion.choices[0].message.content