itismouad/pythonic-raqa-app

1# You can find this code for Chainlit python streaming here (https://docs.chainlit.io/concepts/streaming/python)2 3# OpenAI Chat completion4import os5from openai import AsyncOpenAI  # importing openai for API usage6import chainlit as cl  # importing chainlit for our app7from chainlit.prompt import Prompt, PromptMessage  # importing prompt tools8from chainlit.playground.providers import ChatOpenAI  # importing ChatOpenAI tools9from dotenv import load_dotenv10 11import asyncio12 13from aimakerspace.text_utils import TextFileLoader, CharacterTextSplitter14from aimakerspace.vectordatabase import VectorDatabase15from aimakerspace.openai_utils.prompts import (16    UserRolePrompt,17    SystemRolePrompt,18    AssistantRolePrompt,19)20 21load_dotenv()22 23RAQA_PROMPT_TEMPLATE = """24Use the provided context to answer the user's query. 25 26You may not answer the user's query unless there is specific context in the following text.27 28If you do not know the answer, or cannot answer, please respond with "I don't know".29 30Context:31{context}32"""33 34USER_PROMPT_TEMPLATE = """35User Query:36{user_query}37"""38 39def load_vector_db_from_local_file(file_path="data/KingLear.txt"):40    """generates the vector database object base on a local file"""41    42    # load text file and split into chunk of documents43    text_loader = TextFileLoader(file_path)44    documents = text_loader.load_documents()45    text_splitter = CharacterTextSplitter()46    split_documents = text_splitter.split_texts(documents)47    48    # initialize vector db and build from list of documents49    vector_db = VectorDatabase()50    vector_db = asyncio.run(vector_db.abuild_from_list(split_documents))51    return vector_db52    53    54def get_formatted_prompts(vector_db_retriever: VectorDatabase, user_query: str):55 56    raqa_prompt = SystemRolePrompt(RAQA_PROMPT_TEMPLATE)57    user_prompt = UserRolePrompt(USER_PROMPT_TEMPLATE)58 59    context_list = vector_db_retriever.search_by_text(user_query, k=4)60    61    context_prompt = ""62    for context in context_list:63        context_prompt += context[0] + "\n"64 65    formatted_system_prompt = raqa_prompt.create_message(context=context_prompt)66 67    formatted_user_prompt = user_prompt.create_message(user_query=user_query)68    69    return formatted_system_prompt, formatted_user_prompt70 71@cl.on_chat_start  # marks a function that will be executed at the start of a user session72async def start_chat():73 74    settings = {75        "model": "gpt-3.5-turbo",76        "temperature": 0,77        "max_tokens": 500,78        "top_p": 1,79        "frequency_penalty": 0,80        "presence_penalty": 0,81    }82 83    cl.user_session.set("settings", settings)84 85 86@cl.on_message  # marks a function that should be run each time the chatbot receives a message from a user87async def main(message: cl.Message):88 89    settings = cl.user_session.get("settings")90 91    client = AsyncOpenAI()92 93    # print(f"This is the message received by the user : {message.content}")94 95    # this the loading of the vector database96    vector_db = load_vector_db_from_local_file()97 98    formatted_system_prompt, formatted_user_prompt = list(99        get_formatted_prompts(100            vector_db_retriever=vector_db,101            user_query=message.content102            )103            )104    105    # print(f"formatted_system_prompt : {formatted_system_prompt}")106    # print(f"formatted_user_prompt : {formatted_user_prompt}")107 108    formatted_messages =[formatted_system_prompt, formatted_user_prompt]109 110    msg = cl.Message(content="")111 112    # Call OpenAI113    async for stream_resp in await client.chat.completions.create(114        messages=formatted_messages, stream=True, **settings115    ):116        token = stream_resp.choices[0].delta.content117        if not token:118            token = ""119        await msg.stream_token(token)120 121    # print(f"This is the message sent by the model : {msg.content}")122 123    # Send and close the message stream124    await msg.send()125