#================imports============== import uuid import requests import os os.environ["USER_AGENT"] = "RAG-App/1.0" from typing import Dict, List, Any, Generator from dotenv import load_dotenv from bs4 import BeautifulSoup from langchain_core.globals import set_llm_cache from langchain_core.caches import InMemoryCache from langchain_community.document_loaders import WebBaseLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_huggingface import HuggingFaceEmbeddings from langchain_community.vectorstores import Weaviate from langchain_community.vectorstores import FAISS from langchain_groq import ChatGroq from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_classic.chains.combine_documents import create_stuff_documents_chain from langchain_classic.chains import create_retrieval_chain from langchain_core.runnables.history import RunnableWithMessageHistory from langchain_community.chat_message_histories import ChatMessageHistory from langchain_core.chat_history import BaseChatMessageHistory import gradio as gr #================== CONFIG================== load_dotenv() set_llm_cache(InMemoryCache()) api_key = os.environ["GROQ_API_KEY"] print("api chargée:" if api_key else "y'a probleme!!") #========== charger et decouper documents================= urls = [ "https://fr.wikipedia.org/wiki/%C3%89levage", "https://fr.wikipedia.org/wiki/La_P%C3%AAche" ] loader = WebBaseLoader(urls, requests_kwargs={ "headers": { "User-Agent": "RAG-App/1.0" } } ) docs = loader.load() splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200) chunks = splitter.split_documents(docs) #============embeding et indexation vers faiss_db================ embeddings = HuggingFaceEmbeddings(model_name="sentence-transformers/all-MiniLM-L6-v2") faiss_db = FAISS.from_documents( documents=chunks, embedding=embeddings ) retriever = faiss_db.as_retriever(search_type="similarity", search_kwargs={"k": 3}) #=============== LLM et Prompt================= llm = ChatGroq( model="llama-3.3-70b-versatile", temperature=0.0, max_tokens=1200, streaming=True # Activer le streaming ) prompt = ChatPromptTemplate.from_messages([ ("system", """Tu es un assistant expert en dans le domaine de l'elevage et la pêche. Réponds clairement. Si tu ne connais pas, n'invente pas. Garde un ton amical. Contexte : {context}"""), MessagesPlaceholder(variable_name="chat_history"), ("human", "{input}"), ]) #============= CHAINE DE RECUPERATION======= stuff_chain = create_stuff_documents_chain(llm, prompt) rag_chain = create_retrieval_chain(retriever, stuff_chain) # ====== GESTION DE L'HISTORIQUE ====== store = {} def get_session_history(session_id: str) -> BaseChatMessageHistory: if session_id not in store: store[session_id] = ChatMessageHistory() return store[session_id] # ====== CHAIN AVEC MÉMOIRE ====== convers_chain = RunnableWithMessageHistory( rag_chain, get_session_history, input_messages_key="input", history_messages_key="chat_history", output_messages_key="answer" ) # =============FONCTION CHAT AVEC STREAMING ================ SESSION_ID = str(uuid.uuid4()) # session globale pour l'historique def chat_fn_stream(message: str, history: list) -> Generator[str, None, None]: """ Fonction de chat avec streaming en temps réel. Yield chaque token de la réponse au fur et à mesure. """ # Récupérer l'historique de la session session_history = get_session_history(SESSION_ID) # Construire le contexte à partir de l'historique result = convers_chain.invoke( {"input": message}, config={"configurable": {"session_id": SESSION_ID}} ) # Récupérer la réponse complète full_response = result.get("answer", str(result)) # Simuler le streaming en yieldant caractère par caractère partial_response = "" for char in full_response: partial_response += char yield partial_response def get_history_list() -> list: """ Récupère l'historique de la conversation sous forme de liste pour l'affichage dans la sidebar. """ session_history = get_session_history(SESSION_ID) messages = session_history.messages history_list = [] for i in range(0, len(messages), 2): if i + 1 < len(messages): question = messages[i].content answer = messages[i + 1].content history_list.append({ "question": question[:100] + "..." if len(question) > 100 else question, "full_question": question, "full_answer": answer }) return history_list def load_conversation(question: str, history: list) -> list: """ Charge une conversation précédente et affiche la réponse. """ session_history = get_session_history(SESSION_ID) messages = session_history.messages # Trouver la question et sa réponse correspondante for i in range(0, len(messages), 2): if i + 1 < len(messages) and messages[i].content == question: history.append({"role": "user", "content": question}) history.append({"role": "assistant", "content": messages[i + 1].content}) return history return history # ================= INTERFACE GRADIO AVEC HISTORIQUE ==================== with gr.Blocks(title="🤖 RAG: Specialist en Science Animale") as demo: gr.Markdown("# 🤖 RAG: Specialist en Science Animale") gr.Markdown("Posez vos questions sur l'élévage et la pêche") with gr.Row(): # Sidebar pour l'historique with gr.Column(scale=1, min_width=300): gr.Markdown("### 📚 Historique des conversations") # Bouton pour rafraîchir l'historique refresh_btn = gr.Button("🔄 Rafraîchir l'historique") # Liste des questions précédentes history_list = gr.Dataframe( headers=["Question", "Action"], label="Questions précédentes", interactive=True, wrap=True ) def update_history_list(): """Met à jour la liste des questions dans le dataframe.""" session_history = get_session_history(SESSION_ID) messages = session_history.messages data = [] for i in range(0, len(messages), 2): if i + 1 < len(messages): question = messages[i].content data.append([question[:100] + "..." if len(question) > 100 else question, "📋 Voir"]) return data refresh_btn.click( fn=update_history_list, outputs=[history_list] ) # Zone principale de chat with gr.Column(scale=3): chatbot = gr.Chatbot( label="Assistant RAG", height=500 ) # Barre de saisie et bouton d'envoi with gr.Row(): msg = gr.Textbox( label="Votre question", placeholder="Posez votre question sur l'élevage ou la pêche...", scale=4 ) send_btn = gr.Button("Envoyer", variant="primary", scale=1) # Bouton pour effacer la conversation clear_btn = gr.Button("🗑️ Effacer la conversation") # Exemples de questions gr.Examples( examples=[ "C'est quoi la pêche ?", "Explique l'élévage", "Quelle est la différence entre l'élévage et pêche ?" ], inputs=[msg] ) # Fonction pour gérer l'envoi de message def respond(message: str, history: list) -> tuple: """Gère l'envoi du message et met à jour le chatbot.""" # Ajouter le message utilisateur à l'historique history.append({"role": "user", "content": message}) # Obtenir la réponse result = convers_chain.invoke( {"input": message}, config={"configurable": {"session_id": SESSION_ID}} ) response = result.get("answer", str(result)) # Ajouter la réponse à l'historique history.append({"role": "assistant", "content": response}) return "", history def clear_conversation(): """Efface la conversation actuelle.""" store[SESSION_ID] = ChatMessageHistory() return [], [] # Gestionnaires d'événements msg.submit( respond, inputs=[msg, chatbot], outputs=[msg, chatbot] ) send_btn.click( respond, inputs=[msg, chatbot], outputs=[msg, chatbot] ) clear_btn.click( clear_conversation, outputs=[chatbot, history_list] ) # Charger l'historique initial demo.load( fn=update_history_list, outputs=[history_list] ) # ===================LANCEMENT ================ demo.launch()