main.py

import os
import json
import bcrypt
from typing import List
from pathlib import Path
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_huggingface import HuggingFaceEndpoint
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.schema import StrOutputParser

from operator import itemgetter
from pinecone import Pinecone
from langchain_pinecone import PineconeVectorStore
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain.memory import ConversationBufferMemory
from langchain.schema.runnable import Runnable, RunnablePassthrough, RunnableConfig, RunnableLambda
from langchain.callbacks.base import BaseCallbackHandler
from langchain.chains import (
    StuffDocumentsChain, ConversationalRetrievalChain
)
from langchain_core.tracers.context import tracing_v2_enabled

import chainlit as cl
from chainlit.input_widget import TextInput, Select, Switch, Slider

from deep_translator import GoogleTranslator

@cl.password_auth_callback
def auth_callback(username: str, password: str):
    auth = json.loads(os.environ['CHAINLIT_AUTH_LOGIN'])
    ident = next(d['ident'] for d in auth if d['ident'] == username)
    pwd = next(d['pwd'] for d in auth if d['ident'] == username)
    resultLogAdmin = bcrypt.checkpw(username.encode('utf-8'), bcrypt.hashpw(ident.encode('utf-8'), bcrypt.gensalt())) 
    resultPwdAdmin = bcrypt.checkpw(password.encode('utf-8'), bcrypt.hashpw(pwd.encode('utf-8'), bcrypt.gensalt())) 
    resultRole = next(d['role'] for d in auth if d['ident'] == username)
    if resultLogAdmin and resultPwdAdmin and resultRole == "admindatapcc":
        return cl.User(
            identifier=ident + " : 🧑‍💼 Admin Datapcc", metadata={"role": "admin", "provider": "credentials"}
        )
    elif resultLogAdmin and resultPwdAdmin and resultRole == "userdatapcc":
        return cl.User(
            identifier=ident + " : 🧑‍🎓 User Datapcc", metadata={"role": "user", "provider": "credentials"}
        )

def LLModel():
    os.environ['HUGGINGFACEHUB_API_TOKEN'] = os.environ['HUGGINGFACEHUB_API_TOKEN']
    repo_id = "mistralai/Mixtral-8x7B-Instruct-v0.1"
    model = HuggingFaceEndpoint(
        repo_id=repo_id, max_new_tokens=5500, temperature=1.0, task="text2text-generation", streaming=True
    )
    return model
    
def VectorDatabase(categorie):
    if categorie != "year":
        os.environ['PINECONE_API_KEY'] = ""
        os.environ['PINECONE_API_KEY'] = os.getenv('PINECONE_API_KEY')
        index_name = "all-venus"
    else:
        os.environ['PINECONE_API_KEY'] = ""
        os.environ['PINECONE_API_KEY'] = os.getenv('PINECONE_API_KEYJDLP')
        index_name = "all-jdlp"

    embeddings = HuggingFaceEmbeddings()
    vectorstore = PineconeVectorStore(
        index_name=index_name, embedding=embeddings
    )
    return vectorstore

def Retriever(categorie):
    vectorstore = VectorDatabase(categorie)
    if categorie != "year":
        retriever = vectorstore.as_retriever(search_type="similarity_score_threshold", search_kwargs={"score_threshold": .7, "k": 150,"filter": {'categorie': {'$eq': categorie}}})
    else:
        retriever = vectorstore.as_retriever(search_type="similarity_score_threshold", search_kwargs={"score_threshold": .7, "k": 6,"filter": {'year': {'$gte': 2019}}})
    #search = vectorstore.similarity_search(query,k=50, filter={"categorie": {"$eq": "bibliographie-OPP-DGDIN"}, 'Source': {'$eq': 'Source : Persée'}})    
    return retriever

def Search(input, categorie):
    vectorstore = VectorDatabase(categorie)
    results = []
    test = []
    sources_text = ""
    verbatim_text = ""
    count = 0
    if categorie != "year":
        search = vectorstore.similarity_search(input,k=50, filter={"categorie": {"$eq": categorie}})
        for i in range(0,len(search)):
            if search[i].metadata['Lien'] not in test:
                if count <= 15:
                    count = count + 1
                    test.append(search[i].metadata['Lien'])
                    sources_text = sources_text + str(count) + ". " + search[i].metadata['Titre'] + ', ' + search[i].metadata['Auteurs'] + ', ' + search[i].metadata['Lien'] + "\n"
                    verbatim_text = verbatim_text + "<p>" + str(count) + ". " + search[i].metadata['Phrase'] + "</p><p>&nbsp;</p>"
        
    else:
        search = vectorstore.similarity_search(input,k=50, filter={"year": {"$gte": 2019}})
        for i in range(0,len(search)):
            if count <= 15:
                count = count + 1
                sources_text = sources_text + str(count) + ". " + search[i].metadata['title'] + ' (JDLP : ' + str(search[i].metadata['year']) + '), ' + search[i].metadata['author'] + ', https://cipen.univ-gustave-eiffel.fr/fileadmin/CIPEN/OPP/' + search[i].metadata['file'] + "\n"
                verbatim_text = verbatim_text + "<p>" + str(count) + ". JDLP : " + search[i].metadata['jdlp'] + "</p><p>" + search[i].page_content + "</p>"
                    
    results = [sources_text, verbatim_text]
    return results
    
@cl.on_chat_start
async def on_chat_start():
    await cl.Message(f"> REVIEWSTREAM").send()
    #await cl.Message(f"Nous avons le plaisir de vous accueillir dans l'application de recherche et d'analyse des publications.").send()
    res = await cl.AskActionMessage(
        content="<div style='width:100%;text-align:center'>Sélectionnez une source documentaire</div>",
        actions=[
            cl.Action(name="bibliographie-OPP-DGDIN", value="Pédagogie durable", label="🔥 Pédagogie durable : exemple : «quels sont les modèles d'apprentissage dans les universités?»"),
            cl.Action(name="bibliographie-OPP-DGDIN", value="Lieux d'apprentissage", label="🔥 Lieux d'apprentissage : exemple : «donne des exemples de lieu d'apprentissage dans les universités?»"),
            cl.Action(name="year", value="Journée de La Pédagogie", label="🔥 Journée de La Pédagogie : exemple : «Quelles sont les bonnes pratiques des plateformes de e-learning?»"),
        ],
        timeout="3600"
    ).send()
    listPrompts_name = f"Liste des revues de recherche"
    contentPrompts = """<p><img src='/public/hal-logo-header.png' width='32' align='absmiddle' /> <strong> Hal Archives Ouvertes</strong> : Une archive ouverte est un réservoir numérique contenant des documents issus de la recherche scientifique, généralement déposés par leurs auteurs, et permettant au grand public d'y accéder gratuitement et sans contraintes.
    </p>
    <p><img src='/public/logo-persee.png' width='32' align='absmiddle' /> <strong>Persée</strong> : offre un accès libre et gratuit à des collections complètes de publications scientifiques (revues, livres, actes de colloques, publications en série, sources primaires, etc.) associé à une gamme d'outils de recherche et d'exploitation.</p>
    """
    prompt_elements = []
    prompt_elements.append(
        cl.Text(content=contentPrompts, name=listPrompts_name, display="side")
    )
    await cl.Message(content="📚 " + listPrompts_name, elements=prompt_elements).send()
    settings = await cl.ChatSettings(
        [
            Select(
                id="Model",
                label="Publications de recherche",
                values=["---", "HAL", "Persée"],
                initial_index=0,
            ),
        ]
    ).send()
    if res:
        await cl.Message(f"Vous pouvez requêter sur la thématique : {res.get('value')}").send()
        cl.user_session.set("selectRequest", res.get("name"))
        
    ########## Chain with streaming ##########
    message_history = ChatMessageHistory()
    memory = ConversationBufferMemory(memory_key="chat_history",output_key="answer",chat_memory=message_history,return_messages=True)
    
    qa = ConversationalRetrievalChain.from_llm(
        LLModel(),
        memory=memory,
        chain_type="stuff",
        return_source_documents=True,
        verbose=False,
        retriever=Retriever(res.get("name"))
    )
    cl.user_session.set("runnable", qa)
    cl.user_session.set("memory", memory)


@cl.on_message
async def on_message(message: cl.Message):
    memory = cl.user_session.get("memory")
    runnable = cl.user_session.get("runnable")  # type: Runnable
    msg = cl.Message(content="")
    
    class PostMessageHandler(BaseCallbackHandler):
        """
        Callback handler for handling the retriever and LLM processes.
        Used to post the sources of the retrieved documents as a Chainlit element.
        """

        def __init__(self, msg: cl.Message):
            BaseCallbackHandler.__init__(self)
            self.msg = msg
            self.sources = set()  # To store unique pairs

        def on_retriever_end(self, documents, *, run_id, parent_run_id, **kwargs):
            for d in documents:
                source_page_pair = (d.metadata['source'], d.metadata['page'])
                self.sources.add(source_page_pair)  # Add unique pairs to the set

        def on_llm_end(self, response, *, run_id, parent_run_id, **kwargs):
                sources_text = "\n".join([f"{source}#page={page}" for source, page in self.sources])
                self.msg.elements.append(
                    cl.Text(name="Sources", content=sources_text, display="inline")
                )
        
    async with cl.Step(type="run", name="Réponse de Mistral"):
        #async for chunk in runnable.astream(
        #    {"question": message.content},
        #    config=RunnableConfig(callbacks=[
        #        cl.AsyncLangchainCallbackHandler(stream_final_answer=True)
        #    ]),
        #):
        #    await msg.stream_token(chunk)
        cb = cl.AsyncLangchainCallbackHandler()
        with tracing_v2_enabled():
            results = await runnable.acall("Contexte : Vous êtes un chercheur de l'enseignement supérieur et vous êtes doué pour faire des analyses d'articles de recherche sur les thématiques liées à la pédagogie, en fonction des critères définis ci-avant. En fonction des informations suivantes et du contexte suivant seulement et strictement, répondez en langue française strictement à la question ci-dessous à partir du contexte ci-dessous. En plus, tu créeras 3 questions supplémentaires en relation avec le contexte initial. Tu écriras les 3 questions supplémentaires en relation avec le contexte initial, avec un titrage de niveau 1 qui a pour titre \"Questions en relation avec le contexte : \". Si vous ne pouvez pas répondre à la question sur la base des informations, dites que vous ne trouvez pas de réponse ou que vous ne parvenez pas à trouver de réponse. Essayez donc de comprendre en profondeur le contexte et répondez uniquement en vous basant sur les informations fournies. Ne générez pas de réponses non pertinentes. Question : " + message.content, callbacks=[cb])
            answer = results["answer"]

    await cl.Message(content=GoogleTranslator(source='auto', target='fr').translate(answer)).send()
    
    #search = vectorstore.similarity_search(message.content,k=50, filter={"categorie": {"$eq": "bibliographie-OPP-DGDIN"}})
    search = Search(message.content, cl.user_session.get("selectRequest"))
    
    sources = [
        cl.Text(name="Sources", content=search[0], display="inline")
    ]

    await cl.Message(
        content="Sources : ",
        elements=sources,
    ).send()
    verbatim = [
        cl.Text(name="Verbatim", content=search[1], display="side")
    ]

    await cl.Message(
        content="📚 Liste des Verbatim ",
        elements=verbatim,
    ).send()