De nombreuses entreprises d’IA introduisent désormais des systèmes décisionnels agentiques capables d’analyser instantanément vos données et de fournir des informations complètes sans avoir besoin de requêtes SQL complexes ou d’analyses manuelles de feuilles de calcul. Par exemple, un système d’analyse commerciale basé sur RAG peut répondre à des questions telles que « Quel a été notre produit le plus vendu au dernier trimestre ? » en quelques secondes plutôt qu’en heures.
Un système d’analyse commerciale basé sur RAG résout ces problèmes en utilisant des agents IA capables de raisonner de manière autonome, de prendre des décisions et de coordonner des outils externes pour permettre des requêtes en langage naturel sur l’ensemble de votre écosystème de données. Au lieu d’apprendre les schémas de base de données ou les formules de feuilles de calcul, l’agent décompose votre requête, détermine la meilleure source de données, effectue une analyse ciblée et fournit des réponses complètes et fondées sur les données avec des citations appropriées.
Par exemple, vous pouvez utiliser un LLM différent, un format de document personnalisé ou ajuster la façon dont les agents analysent et se comportent pour produire des résultats métier spécifiques. Dans ce tutoriel, vous apprendrez à construire votre propre système d’analyse commerciale basé sur RAG en utilisant l’API LLM unifiée de Novita AI, le framework d’agents LangChain et des capacités avancées de traitement de documents. Vous créerez un système qui achemine automatiquement les requêtes vers les sources de données optimales et fournit des informations exploitables pour l’entreprise.
Qu’est-ce qu’un système d’analyse commerciale basé sur RAG ?
Un système d’analyse commerciale basé sur RAG utilise la génération augmentée par récupération (Retrieval-Augmented Generation) pour fournir des réponses précises et fondées aux questions sur les ventes au détail, en combinant la puissance des grands modèles de langage avec une couche de récupération capable d’accéder à vos données métier réelles.
Ce système implique généralement :
- L’agent IA analyse la requête et détermine si elle nécessite une analyse de données structurées (SQL), une analyse de documents, ou les deux.
- L’agent achemine intelligemment les requêtes vers la source de données la plus appropriée — les bases de données SQL pour l’analyse quantitative ou les stockages de documents pour les informations qualitatives.
- L’agent utilise divers outils pour collecter des informations, notamment des agents Pandas pour analyser les fichiers CSV, des agents SQL pour interroger les bases de données et des stockages vectoriels pour récupérer des documents. L’agent ne peut pas effectuer de tâches essentielles dans le flux de travail sans ces outils.
- L’agent traite les informations collectées à l’aide de techniques spécialisées, d’analyses statistiques pour les données numériques et de recherche sémantique pour les documents texte.
- Après avoir analysé les données, l’agent crée une réponse structurée qui comprend des résumés, des calculs et des citations appropriées.
Outils dont vous aurez besoin
Avant d’entrer dans la partie construction de cet article, configurons les outils nécessaires.
Novita AI
Pour construire notre système d’analyse basé sur RAG, nous aurons besoin d’accéder à de puissants grands modèles de langage (LLM) et modèles d’embedding. Novita AI propose des API performantes et abordables qui donnent accès aux derniers grands modèles de langage (LLM), modèles d’embedding, et plus encore, via une interface unique et unifiée.
LangChain
LangChain est un framework open source conçu pour construire des applications avec les LLM. Avec LangChain, vous pouvez créer un flux de travail agentique qui raisonne étape par étape, utilise des outils et interagit avec des API. Pour notre système d’analyse commerciale, nous utiliserons LangChain pour structurer le processus d’analyse, employer des outils tels que les agents SQL et les processeurs de documents, et synthétiser toutes les données en informations structurées.
Streamlit
Nous utiliserons Streamlit pour construire l’interface web interactive. C’est parfait pour le prototypage rapide et crée des interfaces utilisateur professionnelles avec un minimum de code, rendant ce tutoriel facile à suivre.
FAISS
Pour la récupération de documents basée sur les vecteurs, FAISS offre des capacités de recherche de similarité rapides, nous permettant de trouver rapidement des fragments de documents pertinents en fonction des requêtes des utilisateurs.
SQLAlchemy & PyMySQL
Ils gèrent nos opérations de base de données SQL et la connectivité MySQL, permettant à notre système d’interroger directement les données métier structurées.
pandas
Essentiel pour la manipulation et l’analyse de données CSV. Notre système utilise des agents pandas pour effectuer des calculs complexes sur des données tabulaires.
Aperçu de l’architecture du système
Notre système RAG traite intelligemment les requêtes des utilisateurs via un flux de travail en plusieurs étapes qui détermine automatiquement la meilleure source de données et la méthode de traitement pour chaque question. Plutôt que d’obliger les utilisateurs à savoir où se trouvent leurs données, le système détermine l’approche optimale en coulisses.
Traitement des requêtes
La première étape de notre processus de travail avec notre Agent est le traitement des requêtes. Dans ce cas, l’Agent vérifiera la requête soumise par l’utilisateur et élaborera un plan de traitement personnalisé adapté au contexte de la requête. Il déterminera ensuite le type de données sur lequel la question doit être analysée : données structurées, analyse de documents, ou les deux.
Sélection de la source de données
Lorsqu’un utilisateur soumet une question, le système adopte une approche simple : il tente d’abord de résoudre le problème en consultant la base de données SQL. La sélection de la source de données a du sens car les données structurées fournissent souvent les réponses les plus précises aux questions quantitatives. L’agent SQL peut rapidement exécuter des requêtes et renvoyer des nombres exacts avec les calculs appropriés.
Analyse de documents
Lorsque l’agent SQL renvoie des résultats vides ou ne trouve pas d’informations pertinentes, le système bascule automatiquement vers l’analyse de documents. Le chemin de traitement des documents varie selon le type de fichier. Les fichiers CSV et Excel sont chargés dans des DataFrames Pandas pour une analyse complexe des données, tandis que les PDF, fichiers Word et documents texte sont découpés en morceaux et indexés à l’aide de FAISS pour la recherche sémantique.
Gestion des sources
Il est essentiel de citer correctement toutes les sources pour garantir que les informations fournies peuvent être retracées jusqu’à une autorité crédible. Cette étape dans le flux de travail contribue à établir la confiance avec le public, en lui permettant de vérifier toute affirmation ou donnée présentée dans le rapport final.
Synthèse des réponses
Pour relier tous les points entre les informations recueillies, l’Agent rassemble les informations de toutes les sources de données analysées pour former une réponse cohérente. Sans cette phase du flux de travail, la sortie pourrait être un rapport disjoint.
Implémentation du flux de travail d’analyse basé sur RAG
Jusqu’à présent, vous avez compris ce qu’est le système d’analyse commerciale basé sur RAG. Il est maintenant temps de l’implémenter. Pour commencer, nous allons effectuer quelques étapes clés :
Installation et configuration
Avant de poursuivre, établissons la structure de fichiers complète pour notre système d’analyse RAG. Vous devrez créer les fichiers suivants dans votre répertoire de projet :
| rag-analytics-system/ ├── .env # Variables d’environnement ├── requirements.txt # Dépendances Python ├── query_processor.py # Classe principale QueryProcessor ├── main.py # Interface Streamlit |
Configuration des dépendances
Tout d’abord, mettez à jour le fichier requirements.txt avec toutes les dépendances nécessaires :
| langchain==0.3.26 langchain-openai==0.3.25 python-dotenv==1.1.1 SQLAlchemy==2.0.41 pandas==2.3.0 PyPDF2==3.0.1 faiss-cpu==1.11.0 PyMySQL==1.1.1 cryptography==45.0.4 langchain-experimental==0.3.4 streamlit==1.46.0 openpyxl==3.1.5 python-docx==1.2.0 |
Connectez-vous ensuite à Novita AI. Une fois connecté, accédez à la page Manage API Keys, cliquez sur le bouton Add New Key et entrez votre Key Name.
Novita AI vous fournira des crédits gratuits pour essayer différents modèles après votre inscription, vous n’avez donc pas à vous soucier d’acheter des crédits avant de commencer à construire ou à expérimenter. Ajoutez vos informations d’identification au fichier .env, comme la clé API Novita que vous avez créée précédemment :
| NOVITA_API_KEY=“votre_clé_api_novita_ici” |
Maintenant, installez toutes les dépendances requises pour le projet :
| # Créer un environnement virtuel python -m venv rag_analytics_env source rag_analytics_env/bin/activate # Sur Windows : rag_analytics_env\Scripts\activate # Installer les dépendances pip install -r requirements.txt |
Construction du QueryProcessor
L’ensemble du flux de travail d’analyse RAG est encapsulé dans une classe modulaire appelée QueryProcessor. Cette classe gérera le processus d’acheminement des requêtes, d’analyse SQL, de traitement des documents, de synthèse des résultats et d’exécution de la boucle d’analyse complète.
Le QueryProcessor sert d’orchestrateur central pour notre système d’analyse. Plutôt que de construire des outils séparés pour différents types de données, cette approche unifiée nous permet d’acheminer intelligemment les requêtes vers la méthode d’analyse la plus appropriée, que ce soit des bases de données SQL, des fichiers CSV ou des documents non structurés.
Tout d’abord, nous importerons les bibliothèques nécessaires pour le projet dans le fichier query_processor.py. Ces importations nous fournissent tout ce dont nous avons besoin pour l’intégration du modèle de langage, la connectivité de base de données, le traitement de documents et le stockage vectoriel :
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain.agents import AgentExecutor, create_react_agent
from langchain_community.agent_toolkits.sql.base import create_sql_agent
from langchain_experimental.agents.agent_toolkits import create_csv_agent, create_pandas_dataframe_agent
from langchain_community.agent_toolkits.sql.toolkit import SQLDatabaseToolkit
from langchain.agents import AgentType
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import FAISS
from langchain_community.utilities import SQLDatabase
from langchain.prompts import PromptTemplate
from langchain_community.tools import ReadFileTool
from langchain.schema import Document
import os
from dotenv import load_dotenv
from sqlalchemy.engine import Engine
import glob
import pandas as pd
load_dotenv()
Ensuite, nous initialisons l’agent d’analyse avec un accès à un LLM via l’API Novita et divers outils de traitement de données. La méthode d’initialisation configure tous les composants principaux dont nous aurons besoin : le modèle de langage pour comprendre les requêtes, les embeddings pour la recherche de similarité de documents, et le diviseur de texte pour traiter efficacement les grands documents.
class QueryProcessor:
def __init__(self, documents_folder: str, sql_engine: Engine):
self.documents_folder = documents_folder
self.sql_engine = sql_engine
self.llm = ChatOpenAI(
model="google/gemma-3-27b-it",
temperature=0,
openai_api_key=os.getenv("NOVITA_API_KEY"),
openai_api_base="<https://api.novita.ai/v3/openai>",
default_headers={
"X-Model-Provider": "google"
}
)
self.embeddings = OpenAIEmbeddings(
model="baai/bge-m3",
openai_api_key=os.getenv("NOVITA_API_KEY"),
openai_api_base="<https://api.novita.ai/v3/openai>",
default_headers={
"X-Model-Provider": "baai"
}
)
self.text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200,
length_function=len,
)
self.sql_agent = None
self._prepare_data_sources()
Le constructeur initialise plusieurs composants clés. Nous utilisons une température de 0 pour le LLM afin de garantir des réponses cohérentes et factuelles plutôt que des variations créatives. Le diviseur de texte est configuré avec des chevauchements de fragments pour maintenir la continuité du contexte lors du traitement de grands documents. La méthode _prepare_data_sources() est appelée à la fin pour configurer nos agents spécialisés.
Création de l’intégration multi-sources
L’étape suivante consiste à créer des méthodes qui initialisent différents agents de traitement de données. Chaque agent se spécialise dans le traitement de types de données et de tâches d’analyse spécifiques, un peu comme si l’on avait différents experts pour différents types de questions.
def _prepare_data_sources(self):
"""Préparer les sources de données SQL et documents"""
# Préparer l'agent SQL
self._prepare_sql_agent()
# Préparer l'agent document
self._prepare_document_agent()
def _prepare_sql_agent(self):
"""Initialiser l'agent SQL"""
# Convertir SQLAlchemy Engine en SQLDatabase LangChain
db = SQLDatabase(self.sql_engine)
toolkit = SQLDatabaseToolkit(db=db, llm=self.llm)
self.sql_agent = create_sql_agent(
llm=self.llm,
toolkit=toolkit,
agent_type=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
verbose=True
)
def _prepare_document_agent(self):
"""Initialiser l'agent document en utilisant ReadFileTool et le stockage vectoriel"""
# Obtenir tous les fichiers pris en charge
supported_files = []
for ext in ['*.txt', '*.pdf', '*.docx', '*.xlsx', '*.xls', '*.csv']:
supported_files.extend(glob.glob(os.path.join(self.documents_folder, ext)))
print(f"\
{len(supported_files)} fichiers pris en charge trouvés dans {self.documents_folder}")
if supported_files:
# Créer ReadFileTool
read_file_tool = ReadFileTool()
# Créer la liste d'outils
tools = [read_file_tool]
# Créer le modèle de prompt pour l'agent react
prompt = PromptTemplate.from_template("""
Vous êtes un assistant utile capable de répondre à des questions sur des documents métier.
Vous avez accès aux outils suivants :
{tools}
Utilisez le format suivant :
Question : la question d'entrée à laquelle vous devez répondre
Réflexion : vous devez toujours réfléchir à ce qu'il faut faire
Action : l'action à entreprendre, doit être l'une de [{tool_names}]
Entrée de l'action : l'entrée de l'action
Observation : le résultat de l'action
... (cette séquence Réflexion/Action/Entrée de l'action/Observation peut se répéter N fois)
Réflexion : je connais maintenant la réponse finale
Réponse finale : la réponse finale à la question d'entrée originale
Commencez !
Question : {input}
Réflexion : Je devrais lire les documents pertinents pour trouver la réponse
{agent_scratchpad}
""")
agent = create_react_agent(self.llm, tools, prompt)
self.document_agent = AgentExecutor.from_agent_and_tools(
agent=agent,
tools=tools,
verbose=True
)
print("\
Agent document initialisé avec succès")
else:
print("\
Aucun document trouvé dans le dossier spécifié.")
La méthode prepare_sql_agent() crée un agent spécialisé pour les requêtes de base de données. Nous utilisons le type d’agent ZERO_SHOT_REACT_DESCRIPTION, ce qui signifie que l’agent peut raisonner sur les requêtes SQL sans avoir besoin d’exemples spécifiques, le rendant adaptable à différents schémas de base de données. La méthode _prepare_document_agent() recherche les types de fichiers pris en charge et crée un agent flexible capable de lire divers formats de documents. Le modèle de prompt ReAct (Reasoning and Acting) guide l’agent à travers un processus de réflexion structuré, similaire à la façon dont un analyste humain aborderait l’analyse de documents.
Traitement des documents et stockage vectoriel
Ensuite, nous devons gérer la manière dont le flux de travail traite les différents types de documents. Cette section crée des méthodes qui gèrent les fichiers CSV/Excel à l’aide d’agents pandas et les documents non structurés à l’aide d’un stockage vectoriel. L’idée clé ici est que différents types de données nécessitent des stratégies de traitement différentes.
CSV_PROMPT_PREFIX = """
IMPORTANT : Vous travaillez avec un DataFrame pandas appelé 'df' qui a été chargé avec les données réelles.
NE créez PAS d'exemples de données ou de données fictives. Utilisez UNIQUEMENT le DataFrame 'df' réel qui vous est disponible.
Commencez par explorer le DataFrame :
1. Définissez les options d'affichage pandas pour afficher toutes les colonnes : pd.set_option('display.max_columns', None)
2. Vérifiez la forme du DataFrame : print(df.shape)
3. Obtenez les noms des colonnes : print(df.columns.tolist())
4. Vérifiez les types de données : print(df.dtypes)
5. Regardez les premières lignes : print(df.head())
6. Répondez ensuite à la question en utilisant les données réelles du DataFrame.
"""
CSV_PROMPT_SUFFIX = """
- **CRITIQUE** : Utilisez UNIQUEMENT les données réelles du DataFrame. Ne créez PAS d'exemples de données ou de données fictives.
- **TOUJOURS** avant de donner la réponse finale, essayez une autre méthode pour vérifier vos résultats.
- Réfléchissez ensuite aux réponses des deux méthodes que vous avez utilisées et demandez-vous si elles répondent correctement à la question originale.
- Si vous n'êtes pas sûr, essayez une autre méthode.
- FORMATTEZ 4 CHIFFRES OU PLUS AVEC DES VIRGULES.
- Si les méthodes essayées ne donnent pas le même résultat, réfléchissez et réessayez jusqu'à ce que vous ayez deux méthodes qui donnent le même résultat.
- Si vous ne parvenez toujours pas à obtenir un résultat cohérent, dites que vous n'êtes pas sûr de la réponse.
- Si vous êtes sûr de la bonne réponse, créez une réponse belle et complète en utilisant Markdown.
- **N'INVENTEZ PAS DE RÉPONSE ET N'UTILISEZ PAS DE CONNAISSANCES ANTÉRIEURES, UTILISEZ UNIQUEMENT LES RÉSULTATS DES CALCULS QUE VOUS AVEZ EFFECTUÉS**.
- **TOUJOURS**, dans le cadre de votre "Réponse finale", expliquez comment vous êtes arrivé à la réponse dans une section qui commence par : "\\
\\
Explication :\\
".
- Dans l'explication, mentionnez les noms de colonnes que vous avez utilisés pour arriver à la réponse finale.
- Montrez votre travail en affichant les opérations pertinentes du DataFrame et leurs résultats.
"""
def _process_document_query(self, query: str) -> str:
"""Traiter une requête document en utilisant des agents CSV/Excel pour les données tabulaires, et LLM pour les données non structurées."""
try:
print(f"\\
Traitement de la requête : {query}")
# Obtenir tous les fichiers pris en charge
supported_files = []
for ext in ['*.txt', '*.pdf', '*.docx', '*.xlsx', '*.xls', '*.csv']:
supported_files.extend(glob.glob(os.path.join(self.documents_folder, ext)))
print(f"{len(supported_files)} fichiers pris en charge trouvés dans {self.documents_folder}")
if not supported_files:
return "Aucun document trouvé pour effectuer une recherche."
# Vérifier d'abord les fichiers CSV/Excel
csv_files = [f for f in supported_files if f.endswith('.csv')]
excel_files = [f for f in supported_files if f.endswith(('.xlsx', '.xls'))]
if csv_files:
csv_file = csv_files[0]
print(f"\
Traitement du fichier CSV : {csv_file}")
try:
# Essayer d'abord avec l'agent DataFrame pandas
df = pd.read_csv(csv_file)
print(f"CSV chargé avec succès : {len(df)} lignes et colonnes : {df.columns.tolist()}")
# Créer l'agent avec une configuration améliorée
print("Création de l'agent DataFrame pandas...")
agent = create_pandas_dataframe_agent(
self.llm,
df,
verbose=True,
include_df_in_prompt=False, # Éviter les limites de tokens avec les grands DataFrames
allow_dangerous_code=True,
max_iterations=10,
handle_parsing_errors=True
)
# Traiter la requête avec notre prompt personnalisé
print(f"Traitement de la requête avec l'agent : {query}")
# Prompt amélioré pour garantir que l'agent utilise le DataFrame réel
prompt = f"""
Vous avez accès à un DataFrame pandas appelé 'df' avec {len(df)} lignes et les colonnes suivantes : {df.columns.tolist()}.
Voici les premières lignes des données :
{df.head().to_string()}
Types de données :
{df.dtypes.to_string()}
{self.CSV_PROMPT_PREFIX}
Question : {query}
{self.CSV_PROMPT_SUFFIX}
"""
response = agent.invoke({"input": prompt})
print(f"Réponse de l'agent : {response}")
return response['output']
except Exception as e:
print(f"Erreur avec l'agent DataFrame pandas : {str(e)}")
print("Tentative avec l'agent CSV alternatif...")
# Repli sur l'agent CSV
try:
agent = create_csv_agent(
self.llm,
csv_file,
verbose=True,
allow_dangerous_code=True
)
enhanced_query = f"""
{self.CSV_PROMPT_PREFIX}
Question : {query}
{self.CSV_PROMPT_SUFFIX}
"""
response = agent.invoke({"input": enhanced_query})
return response['output']
except Exception as e2:
print(f"Erreur lors du traitement du fichier CSV : {str(e2)}")
import traceback
print(f"Pile d'appels complète : {traceback.format_exc()}")
return f"Erreur lors du traitement du fichier CSV : {str(e2)}"
elif excel_files:
excel_file = excel_files[0]
print(f"\
Traitement du fichier Excel : {excel_file}")
try:
df = pd.read_excel(excel_file)
print(f"Excel chargé avec succès : {len(df)} lignes et colonnes : {df.columns.tolist()}")
# Créer l'agent avec une configuration améliorée
print("Création de l'agent DataFrame pandas...")
agent = create_pandas_dataframe_agent(
self.llm,
df,
verbose=True,
include_df_in_prompt=False, # Éviter les limites de tokens avec les grands DataFrames
allow_dangerous_code=True,
max_iterations=10,
handle_parsing_errors=True
)
# Traiter la requête avec notre prompt personnalisé
print(f"Traitement de la requête avec l'agent : {query}")
# Prompt amélioré pour garantir que l'agent utilise le DataFrame réel
prompt = f"""
Vous avez accès à un DataFrame pandas appelé 'df' avec {len(df)} lignes et les colonnes suivantes : {df.columns.tolist()}.
Voici les premières lignes des données :
{df.head().to_string()}
Types de données :
{df.dtypes.to_string()}
{self.CSV_PROMPT_PREFIX}
Question : {query}
{self.CSV_PROMPT_SUFFIX}
"""
response = agent.invoke({"input": prompt})
print(f"Réponse de l'agent : {response}")
return response['output']
except Exception as e:
print(f"Erreur lors du traitement du fichier Excel : {str(e)}")
import traceback
print(f"Pile d'appels complète : {traceback.format_exc()}")
return f"Erreur lors du traitement du fichier Excel : {str(e)}"
# Pour les fichiers texte non structurés
print("\
Traitement des fichiers texte non structurés...")
all_content = []
for file_path in supported_files:
try:
if file_path.endswith('.txt'):
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
all_content.append(Document(page_content=content, metadata={"source": file_path}))
except Exception as e:
print(f"Erreur lors de la lecture du fichier {file_path} : {str(e)}")
continue
if not all_content:
return "Impossible de lire les documents."
# Traiter le texte non structuré à l'aide du stockage vectoriel
print("Traitement du texte avec le stockage vectoriel...")
chunks = self.text_splitter.split_documents(all_content)
vector_store = FAISS.from_documents(chunks, self.embeddings)
relevant_docs = vector_store.similarity_search(query, k=3)
context = "\\
\\
".join([doc.page_content for doc in relevant_docs])
print(f"Longueur du contexte généré : {len(context)}")
response = self.llm.invoke(
f"""En vous basant sur le contexte suivant, répondez à la question : {query}
\\
Contexte :\\
{context}\\
\\
Réponse :"""
)
return response.content
except Exception as e:
print(f"Erreur lors du traitement de la requête document : {str(e)}")
import traceback
print(f"Pile d'appels complète : {traceback.format_exc()}")
return f"Erreur lors du traitement de la requête document : {str(e)}"
La section de traitement CSV inclut une ingénierie de prompt détaillée pour garantir une analyse précise. Les prompts de préfixe et de suffixe sont cruciaux car ils empêchent l’agent d’halluciner des données ou de fournir des résultats incorrects. L’étape de vérification, où l’agent essaie plusieurs méthodes, contribue à garantir l’exactitude, un peu comme un analyste minutieux qui vérifierait ses calculs. Nous priorisons les agents DataFrame pandas par rapport aux agents CSV car ils offrent une gestion des données plus robuste et de meilleures performances avec de grands ensembles de données. Le mécanisme de repli garantit que si une approche échoue, nous avons des méthodes alternatives pour traiter les données. Complétons maintenant la méthode _process_document_query pour gérer les fichiers Excel et les documents non structurés :
elif excel_files:
excel_file = excel_files[0]
print(f"\
Traitement du fichier Excel : {excel_file}")
try:
df = pd.read_excel(excel_file)
print(f"Excel chargé avec succès : {len(df)} lignes et colonnes : {df.columns.tolist()}")
# Créer l'agent avec une configuration améliorée
print("Création de l'agent DataFrame pandas...")
agent = create_pandas_dataframe_agent(
self.llm,
df,
verbose=True,
include_df_in_prompt=False, # Éviter les limites de tokens avec les grands DataFrames
allow_dangerous_code=True,
max_iterations=10,
handle_parsing_errors=True
)
# Traiter la requête avec notre prompt personnalisé
print(f"Traitement de la requête avec l'agent : {query}")
# Prompt amélioré pour garantir que l'agent utilise le DataFrame réel
prompt = f"""
Vous avez accès à un DataFrame pandas appelé 'df' avec {len(df)} lignes et les colonnes suivantes : {df.columns.tolist()}.
Voici les premières lignes des données :
{df.head().to_string()}
Types de données :
{df.dtypes.to_string()}
{self.CSV_PROMPT_PREFIX}
Question : {query}
{self.CSV_PROMPT_SUFFIX}
"""
response = agent.invoke({"input": prompt})
print(f"Réponse de l'agent : {response}")
return response['output']
except Exception as e:
print(f"Erreur lors du traitement du fichier Excel : {str(e)}")
import traceback
print(f"Pile d'appels complète : {traceback.format_exc()}")
return f"Erreur lors du traitement du fichier Excel : {str(e)}"
# Pour les fichiers texte non structurés
print("\\
Traitement des fichiers texte non structurés...")
all_content = []
for file_path in supported_files:
try:
if file_path.endswith('.txt'):
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
all_content.append(Document(page_content=content, metadata={"source": file_path}))
except Exception as e:
print(f"Erreur lors de la lecture du fichier {file_path} : {str(e)}")
continue
if not all_content:
return "Impossible de lire les documents."
# Traiter le texte non structuré à l'aide du stockage vectoriel
print("Traitement du texte avec le stockage vectoriel...")
chunks = self.text_splitter.split_documents(all_content)
vector_store = FAISS.from_documents(chunks, self.embeddings)
relevant_docs = vector_store.similarity_search(query, k=3)
context = "\\
\\
".join([doc.page_content for doc in relevant_docs])
print(f"Longueur du contexte généré : {len(context)}")
response = self.llm.invoke(
f"""En vous basant sur le contexte suivant, répondez à la question : {query}
\\
Contexte :\\
{context}\\
\\
Réponse :"""
)
return response.content
except Exception as e:
print(f"Erreur lors du traitement de la requête document : {str(e)}")
import traceback
print(f"Pile d'appels complète : {traceback.format_exc()}")
return f"Erreur lors du traitement de la requête document : {str(e)}"
Intégration de la base de données SQL
Maintenant que nous avons traité les documents, nous devons créer le système d’acheminement intelligent des requêtes qui décide entre l’analyse SQL et l’analyse de documents. C’est le cœur de l’intelligence de notre système : déterminer quelle source de données est la plus susceptible de contenir la réponse à une question donnée.
def process_query(self, query: str) -> str:
"""
Traiter la requête en utilisant des agents pour décider intelligemment entre SQL et documents
"""
# Essayer d'abord l'agent SQL
try:
print("\
Tentative avec l'agent SQL...")
sql_result = self.sql_agent.run(query)
no_answer_phrases = [
"aucun résultat", "je ne sais pas", "inconnu", "pas sûr", "impossible de répondre", "n'ai pas", "aucune donnée", "n/a"
]
if sql_result and not any(phrase in sql_result.lower() for phrase in no_answer_phrases) and sql_result.strip():
return f"Depuis la base de données SQL : {sql_result}"
else:
print("L'agent SQL n'a pas pu répondre, tentative avec les documents...")
except Exception as e:
print(f"Erreur de l'agent SQL : {str(e)}")
print("Repli sur les documents...")
# Si l'agent SQL échoue ou ne renvoie aucun résultat, essayer le traitement de documents
try:
print("\
Traitement des documents...")
doc_result = self._process_document_query(query)
if doc_result:
return f"Depuis les documents : {doc_result}"
else:
print("Le traitement des documents n'a renvoyé aucun résultat")
except Exception as e:
print(f"Erreur du traitement des documents : {str(e)}")
return "Impossible de trouver des informations pertinentes dans la base de données SQL ou les documents."
La logique d’acheminement des requêtes suit un système de priorité : les bases de données SQL sont essayées en premier car elles contiennent généralement des données structurées et quantitatives qui peuvent répondre rapidement et précisément aux questions commerciales. Si l’agent SQL renvoie des réponses peu claires ou négatives (détectées via notre liste no_answer_phrases), le système bascule automatiquement vers le traitement de documents. Cette approche imite le travail d’un analyste humain : vérifier d’abord les sources de données structurées, puis se tourner vers les documents et rapports lorsque la base de données ne contient pas les informations nécessaires.
Création de l’interface Streamlit
Construisons maintenant l’interface utilisateur qui rendra notre système d’analyse accessible aux utilisateurs métier. L’interface Streamlit offre une expérience intuitive basée sur le chat, qui dissimule la complexité technique tout en fournissant des capacités d’analyse puissantes. Ajoutez les extraits de code suivants dans le fichier main.py :
import streamlit as st
from query_processor import QueryProcessor
import os
from dotenv import load_dotenv
from sqlalchemy import create_engine
# Charger les variables d'environnement
load_dotenv()
# Initialiser l'état de session
if 'processor' not in st.session_state:
st.session_state.processor = None
if 'messages' not in st.session_state:
st.session_state.messages = []
# Configuration de la page
st.set_page_config(
page_title="Chatbot d'analyse de documents",
page_icon="🤖",
layout="wide"
)
# CSS personnalisé pour le style des boutons et du titre
st.markdown("""
<style>
.stButton > button {
background-color: #23D57C;
color: white;
border: none;
border-radius: 8px;
padding: 0.5rem 1rem;
font-weight: 600;
transition: all 0.3s ease;
}
.stButton > button:hover {
background-color: #1fb36b;
box-shadow: 0 4px 8px rgba(35, 213, 124, 0.3);
transform: translateY(-2px);
}
.stButton > button:active {
background-color: #1a9960;
transform: translateY(0px);
}
h1 {
color: #23D57C !important;
font-weight: 700;
}
</style>
""", unsafe_allow_html=True)
# Titre et description
st.title("Chatbot d'analyse de documents utilisant Novita")
Cette configuration initiale crée une interface d’aspect professionnel avec un style personnalisé. La gestion de l’état de session garantit que les utilisateurs ne perdent pas l’historique de leurs conversations et n’ont pas besoin de réinitialiser leurs sources de données lorsqu’ils interagissent avec l’interface. Le CSS personnalisé fournit un retour visuel et maintient une expérience de marque cohérente.
Intégration finale du système
Terminons l’interface Streamlit avec l’initialisation des sources de données et la fonctionnalité de chat. Cette dernière section consolide tous nos composants en une application conviviale que les analystes métier peuvent utiliser sans nécessiter d’expertise technique.
# Vérifier si la source de données est initialisée
if st.session_state.processor is None:
# Centrer la configuration de la source de données
st.markdown("<br><br>", unsafe_allow_html=True)
# Créer des colonnes centrées
col1, col2, col3 = st.columns([1, 2, 1])
with col2:
st.subheader("🚀 Commencer")
st.write("Initialisez votre source de données pour commencer à discuter avec vos documents")
# Configuration de la base de données SQL (masquée)
db_user = "root"
db_password = "1234cisco"
db_host = "localhost"
db_name = "retail_sales_db"
# Configuration du dossier de documents
st.write("**Chemin du dossier de documents :**")
documents_folder = st.text_input(
"Chemin du dossier de documents",
placeholder="Entrez le chemin vers votre dossier de documents (ex. : docs)",
label_visibility="collapsed"
)
st.markdown("<br>", unsafe_allow_html=True)
# Centrer le bouton
button_col1, button_col2, button_col3 = st.columns([1, 1, 1])
with button_col2:
if st.button("Initialiser la source de données", use_container_width=True):
try:
# Valider et résoudre le chemin du dossier de documents
if not documents_folder:
st.error("Veuillez fournir un chemin de dossier de documents")
else:
# Convertir en chemin absolu
abs_documents_folder = os.path.abspath(documents_folder)
if not os.path.exists(abs_documents_folder):
st.error(f"Dossier de documents introuvable : {abs_documents_folder}")
elif not os.path.isdir(abs_documents_folder):
st.error(f"Le chemin n'est pas un dossier : {abs_documents_folder}")
else:
# Initialiser le QueryProcessor avec un moteur SQL factice
# Créer le moteur SQL
connection_string = f"mysql+pymysql://{db_user}:{db_password}@{db_host}/{db_name}"
sql_engine = create_engine(connection_string)
st.session_state.processor = QueryProcessor(abs_documents_folder, sql_engine)
st.success("Source de données initialisée avec succès !")
st.rerun()
except Exception as e:
st.error(f"Erreur lors de l'initialisation de la source de données : {str(e)}")
else:
# Afficher l'état de la source de données dans la barre latérale
with st.sidebar:
st.header("📊 Source de données")
st.success("✅ Source de données initialisée")
if st.button("Réinitialiser la source de données"):
st.session_state.processor = None
st.session_state.messages = []
st.rerun()
# Interface de chat principale
st.header("Interface de chat")
# Afficher les messages du chat
for message in st.session_state.messages:
with st.chat_message(message["role"]):
st.write(message["content"])
# Champ de saisie du chat
if prompt := st.chat_input("Posez une question sur vos documents"):
# Ajouter le message de l'utilisateur à l'historique
st.session_state.messages.append({"role": "user", "content": prompt})
with st.chat_message("user"):
st.write(prompt)
# Traiter la requête
try:
response = st.session_state.processor.process_query(prompt)
# Ajouter la réponse de l'assistant à l'historique
st.session_state.messages.append({"role": "assistant", "content": response})
with st.chat_message("assistant"):
st.write(response)
except Exception as e:
st.error(f"Erreur lors du traitement de la requête : {str(e)}")
# Ajouter un bouton pour effacer le chat seulement s'il y a des messages
if st.session_state.messages:
if st.button("Effacer le chat"):
st.session_state.messages = []
st.rerun()
Le processus d’initialisation inclut une gestion robuste des erreurs et une validation des chemins pour éviter les erreurs courantes des utilisateurs. La barre latérale fournit des informations claires sur l’état, et l’interface de chat suit des schémas familiers que les utilisateurs attendent des assistants IA modernes. Le système conserve l’historique des conversations, permettant aux utilisateurs de s’appuyer sur des questions précédentes et de créer un flux de travail analytique naturel. La fonctionnalité d’effacement du chat permet aux utilisateurs de repartir à zéro si nécessaire.
Exécution du système d’analyse
Maintenant que nous avons construit le système d’analyse commerciale basé sur RAG, testons-le pour voir comment il fonctionne sur des requêtes métier réelles.
Tout d’abord, créez un fichier data_generator.py et copiez le code du script Python ici pour générer des exemples de données que nous donnerons au système, puis exécutez le script avec la commande :
| python data_generator.py |
Cela créera un dossier sample_documents avec :
- large_sales_dataset.csv - 10 000 enregistrements de ventes
- business_strategy_2024.txt - Document stratégique d’entreprise
- sales_meeting_notes.txt - Notes de réunion et actions
Ensuite, démarrez l’application Streamlit :
| streamlit run main.py |
L’application s’ouvrira dans votre navigateur à l’adresse http://localhost:8501.
Une fois initialisée, vous pouvez tester le système en entrant le chemin de vos documents comme sample_documents et poser diverses questions commerciales :
| # Exemples de requêtes pour tester le système topic_1 = “Quel a été le montant total des ventes pour l’électronique en 2024 ?” topic_2 = “Quelles sont nos principales initiatives stratégiques pour l’expérience client ?” topic_3 = “Quel représentant commercial a eu le total de ventes le plus élevé, et quel était ce montant ?” |
Conclusion
Tout au long de ce tutoriel, nous avons construit un système d’analyse basé sur RAG qui démontre comment l’IA moderne peut transformer les flux de travail de business intelligence. En combinant plusieurs approches de traitement de données sous une seule interface intelligente, nous avons créé un outil capable de gérer tout l’éventail des questions commerciales auxquelles les analystes sont confrontés quotidiennement. Les principaux atouts du système incluent :
- Acheminement intelligent des requêtes qui détermine automatiquement la meilleure source de données pour chaque question.
- Prise en charge multi-formats pour traiter les bases de données SQL, fichiers CSV, feuilles de calcul Excel et documents texte.
- Gestion robuste des erreurs avec des mécanismes de repli et des messages d’erreur clairs.
- Interaction basée sur le chat ne nécessitant aucune expertise technique.
- Vérification et validation intégrées pour garantir des résultats fiables.
Vous pouvez étendre cette application pour incorporer des sources supplémentaires, des techniques d’analyse plus avancées et une logique métier personnalisée. Maintenant que vous avez ces connaissances, lancez-vous et essayez de construire vos propres agents avec Novita AI ou intégrez-les dans vos projets existants.
Novita AI est la plateforme cloud tout-en-un qui donne vie à vos ambitions en IA. API intégrées, sans serveur, Instance GPU — les outils rentables dont vous avez besoin. Éliminez l’infrastructure, commencez gratuitement et réalisez votre vision de l’IA.
