main.py

from fastapi import FastAPI, File, UploadFile, HTTPException, Response, Request
from fastapi.middleware.cors import CORSMiddleware
import uuid
from datetime import datetime, timedelta
import asyncio

from typing import List, Dict, Any
from io import BytesIO, StringIO
from docx import Document
from langchain.docstore.document import Document as langchain_Document
from PyPDF2 import PdfReader
import csv

from langchain.prompts import ChatPromptTemplate, PromptTemplate
from langchain.prompts import SystemMessagePromptTemplate, HumanMessagePromptTemplate
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.memory import ConversationBufferMemory
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain.chains import ConversationalRetrievalChain

from dotenv import load_dotenv

load_dotenv()


class Document_Processor:
    def __init__(self, file_details: List[Dict[Any, str]]):
        self.file_details = file_details

    def get_docs(self) -> List[langchain_Document]:
        docs = []
        for file_detail in self.file_details:
            if file_detail["name"].endswith(".txt"):
                docs.extend(self.get_txt_docs(file_detail=file_detail))

            elif file_detail["name"].endswith(".csv"):
                docs.extend(self.get_csv_docs(file_detail=file_detail))

            elif file_detail["name"].endswith(".docx"):
                docs.extend(self.get_docx_docs(file_detail=file_detail))

            elif file_detail["name"].endswith(".pdf"):
                docs.extend(self.get_pdf_docs(file_detail=file_detail))

        return docs

    @staticmethod
    def get_txt_docs(file_detail: Dict[str, Any]) -> List[langchain_Document]:
        text = file_detail["content"].decode("utf-8")
        source = file_detail["name"]
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000, chunk_overlap=100
        )
        text_docs = text_splitter.create_documents(
            [text], metadatas=[{"source": source}]
        )
        return text_docs

    @staticmethod
    def get_csv_docs(file_detail: Dict[str, Any]) -> List[langchain_Document]:
        csv_data = file_detail["content"]
        source = file_detail["name"]
        csv_string = csv_data.decode("utf-8")
        # Use StringIO to create a file-like object from the string
        csv_file = StringIO(csv_string)
        csv_reader = csv.DictReader(csv_file)
        csv_docs = []
        for row in csv_reader:
            # Convert each row into a dictionary of key/value pairs
            page_content = ""
            for key, value in row.items():
                page_content += f"{key}: {value}\n"
            doc = langchain_Document(
                page_content=page_content, metadata={"source": source}
            )
            csv_docs.append(doc)
        return csv_docs

    @staticmethod
    def get_pdf_docs(file_detail: Dict[str, Any]) -> List[langchain_Document]:
        pdf_content = BytesIO(file_detail["content"])
        source = file_detail["name"]

        reader = PdfReader(pdf_content)
        pdf_text = ""
        for page in reader.pages:
            pdf_text += page.extract_text() + "\n"

        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000, chunk_overlap=100
        )
        pdf_docs = text_splitter.create_documents(
            texts=[pdf_text], metadatas=[{"source": source}]
        )
        return pdf_docs

    @staticmethod
    def get_docx_docs(file_detail: Dict[str, Any]) -> List[langchain_Document]:
        docx_content = BytesIO(file_detail["content"])
        source = file_detail["name"]

        document = Document(docx_content)
        docx_text = " ".join([paragraph.text for paragraph in document.paragraphs])

        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000, chunk_overlap=100
        )
        docx_docs = text_splitter.create_documents(
            [docx_text], metadatas=[{"source": source}]
        )
        return docx_docs


class Conversational_Chain:
    def __init__(self, file_details: List[Dict[Any, str]]):
        self.llm_model = ChatOpenAI()
        self.embeddings = OpenAIEmbeddings()
        self.file_details = file_details

    def create_conversational_chain(self):
        docs = Document_Processor(self.file_details).get_docs()
        memory = ConversationBufferMemory(
            memory_key="chat_history", return_messages=True
        )
        vectordb = Chroma.from_documents(
            docs,
            self.embeddings,
        )
        retriever = vectordb.as_retriever()
        conversation_chain = ConversationalRetrievalChain.from_llm(
            llm=self.llm_model,
            retriever=retriever,
            condense_question_prompt=self.get_question_generator_prompt(),
            combine_docs_chain_kwargs={
                "document_prompt": self.get_document_prompt(),
                "prompt": self.get_final_prompt(),
            },
            memory=memory,
        )

        return conversation_chain

    @staticmethod
    def get_document_prompt() -> PromptTemplate:
        document_template = """Document Content:{page_content}
    Document Path: {source}"""
        return PromptTemplate(
            input_variables=["page_content", "source"],
            template=document_template,
        )

    @staticmethod
    def get_question_generator_prompt() -> PromptTemplate:
        question_generator_template = """Combine the chat history and follow up question into
    a standalone question.\n Chat History: {chat_history}\n
    Follow up question: {question}
    """
        return PromptTemplate.from_template(question_generator_template)

    @staticmethod
    def get_final_prompt() -> ChatPromptTemplate:
        final_prompt_template = """Answer question based on the context and chat_history.
    If you cannot find answers, ask more related questions from the user.
    Use only the basename of the file path as name of the documents.
    Mention document name of the documents you used in your answer.

    context:
    {context}

    chat_history:
    {chat_history}

    question:
    {question}

    Answer:
    """

        messages = [
            SystemMessagePromptTemplate.from_template(final_prompt_template),
            HumanMessagePromptTemplate.from_template("{question}"),
        ]

        return ChatPromptTemplate.from_messages(messages)


class UserSessionManager:
    def __init__(self):
        self.sessions = {}
        self.last_request_time = {}

    def get_session(self, user_id: str):
        if user_id not in self.sessions:
            self.sessions[user_id] = None
            self.last_request_time = datetime.now()
        return self.sessions[user_id]

    def set_session(self, user_id: str, conversational_chain):
        self.sessions[user_id] = conversational_chain
        self.last_request_time[user_id] = datetime.now()

    def delete_inactive_sessions(self, inactive_period: timedelta):
        current_time = datetime.now()
        for user_id, last_request_time in list(self.last_request_time.items()):
            if current_time - last_request_time > inactive_period:
                del self.sessions[user_id]
                del self.last_request_time[user_id]


app = FastAPI()

origins = ["https://viboognesh-react-chat.static.hf.space"]
# origins = ["http://localhost:3000"]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["GET", "POST"],
    allow_headers=["*"],
)

user_session_manager = UserSessionManager()


@app.middleware("http")
async def update_last_request_time(request: Request, call_next):
    user_id = request.cookies.get("user_id")
    if user_id:
        user_session_manager.last_request_time[user_id] = datetime.now()
    response = await call_next(request)
    return response


async def check_inactivity():
    inactive_period = timedelta(hours=2)
    while True:
        await asyncio.sleep(600)
        user_session_manager.delete_inactive_sessions(inactive_period)


@app.on_event("startup")
async def startup_event():
    asyncio.create_task(check_inactivity())

@app.post("/upload_files/")
async def upload_files(response: Response, files: List[UploadFile] = File(...)):
    file_details = []
    try:
        for file in files:
            content = await file.read()
            name = f"{file.filename}"
            details = {"content": content, "name": name}
            file_details.append(details)
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

    user_id = response.cookies.get("user_id")
    if not user_id:
        user_id = str(uuid.uuid4())
        response.set_cookie(key="user_id", value=user_id)

    try:
        conversational_chain = Conversational_Chain(
            file_details
        ).create_conversational_chain()
        user_session_manager.set_session(
            user_id=user_id, conversational_chain=conversational_chain
        )
        print("conversational_chain_manager created")
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

    return {"message": "ConversationalRetrievalChain is created. Please ask questions."}


@app.get("/predict/")
async def predict(query: str):
    user_id = response.cookies.get("user_id")
    if not user_id:
        user_id = str(uuid.uuid4())
        response.set_cookie(key="user_id", value=user_id)

    try:
        conversational_chain = user_session_manager.get_session(user_id=user_id)
        if conversational_chain is None:
            system_prompt = "Answer the question and also ask the user to upload files to ask questions from the files.\n"
            llm_model = ChatOpenAI()
            response = llm_model.invoke(system_prompt + query)
            answer = response.content
        else:
            response = conversational_chain.invoke(query)
            answer = response["answer"]
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

    print("predict called")
    return {"answer": answer}