app02-chatRagLcel.py

# Run with reload mode:
# gradio app02-chatRagLcel.py

import os
import gradio as gr

from operator import itemgetter

# Langchain
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnableParallel,RunnablePassthrough,RunnableLambda
from langchain_core.output_parsers import StrOutputParser
from langchain_core.messages import AIMessage, HumanMessage

# HuggingFace
from langchain_community.embeddings import HuggingFaceEmbeddings

# GeminiPro
from langchain_google_genai import ChatGoogleGenerativeAI

# Groq
from langchain_groq import ChatGroq

# Pinecone vector database
from pinecone import Pinecone, ServerlessSpec
from langchain_pinecone import PineconeVectorStore

from dotenv import load_dotenv
load_dotenv()
# print('EMBEDDINGS_MODEL', os.getenv("EMBEDDINGS_MODEL"))


setid = "global"

embeddings = HuggingFaceEmbeddings(model_name=os.getenv("EMBEDDINGS_MODEL"))

# OpenAI
# model = ChatOpenAI(temperature=0.0)

# Gemini
# model = ChatGoogleGenerativeAI(
#     model="gemini-pro", temperature=0.1, convert_system_message_to_human=True
# )

# Groq
# llama2-70b-4096 (4k), mixtral-8x7b-32768 (32k)
model = ChatGroq(model_name='mixtral-8x7b-32768')


pc = Pinecone(api_key=os.getenv("PINECONE_API_KEY"))
index = pc.Index(setid)
vectorstore = PineconeVectorStore(index, embeddings, "text")
retriever = vectorstore.as_retriever(kwargs={"k":5})            # Find 5 documents


template_no_history = """Answer the question based only on the following context:
{context}

Question: {question}
"""
PROMPT_NH = ChatPromptTemplate.from_template(template_no_history)

template_with_history = """Given the following conversation history, answer the follow up question:
Chat History:
{chat_history}

Question: {question}
"""
PROMPT_WH = ChatPromptTemplate.from_template(template_with_history)


def pipeLog(x):
    print("***", x)
    return x


setup_and_retrieval = RunnableParallel(
    {"context": retriever, "question": RunnablePassthrough()}
)

def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

rag_chain_from_docs = (
    RunnablePassthrough.assign(context=(lambda x: format_docs(x["context"])))
    | PROMPT_NH
    | model
    | StrOutputParser()
)

rag_chain_with_source = RunnableParallel(
    {"context": retriever, "question": RunnablePassthrough()}
).assign(answer=rag_chain_from_docs)


def rag_query(question: str, history: list[list[str]]):
    if len(history)==0:
        # chain = setup_and_retrieval | PROMPT_NH | model
        # response = chain.invoke(question)
        response = rag_chain_with_source.invoke(question)
        sources = [ doc.metadata['source'] for doc in response['context'] ]
        print(response, '\n', sources)
        return response['answer'] # FAILS!!!
    else:
        chat_history = ""
        for l in history:
            chat_history += " : ".join(l)
            chat_history += "\n"
        chain = (
            { "chat_history": itemgetter('chat_history'), "question": itemgetter('question') }
            | PROMPT_WH 
            | pipeLog
            | model
        )
        response = chain.invoke({ "chat_history": chat_history, "question": question })
        return response.content

# ----------------------------------------


def pipeLog(s:str, x):
    print(s, x)
    return x
pipe_a = RunnableLambda(lambda x: pipeLog("a:",x))
pipe_b = RunnableLambda(lambda x: pipeLog("b:",x))



contextualize_q_system_prompt = """Given a chat history and the latest user question \
which might reference context in the chat history, formulate a standalone question \
which can be understood without the chat history. Do NOT answer the question, \
just reformulate it if needed and otherwise return it as is."""

contextualize_q_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", contextualize_q_system_prompt),
        MessagesPlaceholder(variable_name="chat_history"),
        ("human", "{question}"),
    ]
)

contextualize_q_chain = contextualize_q_prompt | model | StrOutputParser()




qa_system_prompt = """You are an assistant for question-answering tasks.
Use the following pieces of retrieved context to answer the question.
If you don't know the answer, just say that you don't know.
Use three sentences maximum and keep the answer concise.

{context}"""
qa_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", qa_system_prompt),
        MessagesPlaceholder(variable_name="chat_history"),
        ("human", "{question}"),
    ]
)

def contextualized_question(input: dict):
    if input.get("chat_history"):
        return contextualize_q_chain
    else:
        return input["question"]


rag_chain = (
    RunnablePassthrough.assign(
        context=pipe_b | contextualized_question | retriever | format_docs
    )
    | qa_prompt
    | model
)

rag_chain_with_source = RunnableParallel(
    {"xx": pipe_a, "context": itemgetter('question')|retriever, "question": itemgetter('question'), "chat_history": itemgetter('chat_history') }
).assign(answer=rag_chain)



def rag_query_2(question: str, history: list[list[str]]):
    response = rag_chain_with_source.invoke({ 'question':question, 'chat_history':history })
    print(response)
    # sources = [ doc.metadata['source'] for doc in response['context'] ]
    # print(response, '\n', sources)
    return response['answer'].content







gr.ChatInterface(
    rag_query_2,
    title="RAG Chatbot demo",
    description="A chatbot doing Retrieval Augmented Generation, backed by a Pinecone vector database"
    ).launch()