History aware retriever Accessibility in the script

[Amitt1412]

Checked other resources

• I added a very descriptive title to this question.
• I searched the LangChain documentation with the integrated search.
• I used the GitHub search to find a similar question and didn't find it.

Commit to Help

• I commit to help with one of those options 👆

Example Code

Hi folks,

I tried implementing the history_aware_retriever here in the script as below. 

The base retriever is parent child retriever here. I am receiving the responses as based on the previous context only but I'm not sure as how the history_aware_retriever  is being used in the script. Or is there any gap in the implementation. 

Can anyone help me with a clarification?

from typing import Iterable
import uuid
from fastapi import HTTPException
from fastapi.responses import StreamingResponse
from langchain_community.chat_message_histories import (
    RedisChatMessageHistory,
    ChatMessageHistory,
)
from langchain_core.runnables.history import (
    RunnableWithMessageHistory,
    BaseChatMessageHistory,
)
from langchain.chains.retrieval import create_retrieval_chain
from services import retriever, llm, redis_url
from utilities.logger.custom_logger import get_logger
# from services.prompts.prompt_templates import qa_template, qa_template_arabic
from utilities.observability.prompt_creation import (
    qa_template,
    qa_template_arabic,
    # header_prompt_template,
)
from web_drivers.rest import system_config
from web_drivers.rest.schema import QueryRequestSchema
from services.initialization import lang_fuse

# from utilities.observability.prompt_config import *

logger = get_logger("response_generator")
chat_store = dict()

metadata = {
    "app_name": "slfh_streamlit_app_lngfus",
    "app_version": "1.0",
    "qa_template": qa_template,
    "qa_template_prompt_name": qa_template.name,
    "qa_template_prompt_description": qa_template,
    "qa_template_arabic": qa_template,
    "qa_template_arabic_name": qa_template.name,
}

from langchain.chains.history_aware_retriever import create_history_aware_retriever
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.chains.combine_documents import create_stuff_documents_chain

contextualize_q_system_prompt = """
Given the chat history and the latest user question, which might reference context in the chat history, 
reformulate the question into a standalone question that can be understood without the chat history. 
If the question is related to the most recent queries or requires context from the chat history to be understood, 
include that context in the reformulated question. Do NOT answer the question; just provide the reformulated version.
"""

contextualize_q_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", contextualize_q_system_prompt),
        MessagesPlaceholder("chat_history"),
        ("human", "{input}"),
    ]
)
history_aware_retriever = create_history_aware_retriever(
    llm, retriever, contextualize_q_prompt
)


def get_message_history(session_id: str) -> ChatMessageHistory:
    if session_id not in chat_store:
        chat_store[session_id] = ChatMessageHistory()
    return chat_store[session_id]


def get_context(query) -> Iterable:
    """
    Retrieve contexts relevant to the given query.

    Args:
        query (str): The query for which contexts are to be retrieved.

    Returns:
        Iterable: A generator yielding the contexts relevant to the query.
    """
    return history_aware_retriever.invoke(
        {"input": query}, config={"callbacks": [lang_fuse.get_embedding_callback_handler(metadata=None)]}
    )


def get_response(
    session_id,
    language: str,
    query: str,
    stream: bool = True,
    enable_chat_history: bool = True,
) -> Iterable:
    """
    Generate responses for the given query.

    Args: query (str): The query for which responses are to be generated. stream (bool, optional): If True,
    response is streamed as an iterable. If False, response is generated as a single object. Defaults to True.

    Returns:
        Iterable: A generator yielding the response chunks for the query.
    """
    # chat_history = get_message_history(session_id).messages if enable_chat_history else []
    # context = "\n".join([contexts.page_content for contexts in get_context(query)])

    context = ""

    chain = create_stuff_documents_chain(llm, qa_template if language == "english" else qa_template_arabic)
    # rag_chain = create_retrieval_chain(history_aware_retriever, chain)
    
    if enable_chat_history:
        chain = RunnableWithMessageHistory(
            chain,
            get_message_history,
            input_messages_key="query",
            history_messages_key="chat_history",
        )

    if stream and not enable_chat_history:
        response = chain.stream(
            {"query": query, "context": context},
            config={"callbacks": [lang_fuse.get_callback_handler(metadata=metadata)]},
        )
        return response

    if stream and enable_chat_history:
        response = chain.stream(
            {"query": query, "context": context},
            config={
                "configurable": {
                    "session_id": f"{session_id}",
                }, "callbacks": [lang_fuse.get_callback_handler(metadata=metadata)]
            },
        )
        return response


async def query_request_handler(
    language: str, payload: QueryRequestSchema, session_id: str
) -> StreamingResponse:
    """
    Handles incoming requests and returns streaming response.

    Args:
        payload (QueryRequest): Incoming request containing query.
        language (str): Language of conversation.
        session_id (str): Session ID for maintaining chat history.

    Returns:
        StreamingResponse: Streaming response containing chunks of data.
    """
    lang_fuse.set_attr(session_id=session_id, user_id=str(uuid.uuid1()))
    try:
        # Generate stream of chunks based on query
        chunk_stream_gen = get_response(
            language=language,
            query=payload.query,
            session_id=session_id,
            enable_chat_history=system_config.get("ENABLE_CHAT_HISTORY"),
        )
        return StreamingResponse(
            (chunk for chunk in chunk_stream_gen), media_type="text/event-stream"
        )
    except Exception as e:
        # Raise HTTP exception if there's an error
        raise HTTPException(status_code=500, detail=str(e)) ```

[dosubot[bot]]

Hey there, @Amitt1412! 👋 I'm here to give you a hand while waiting for a human maintainer.

The history_aware_retriever is correctly implemented and used in your script. The function create_history_aware_retriever is designed to create a chain that takes conversation history and returns documents. It checks if the input is part of the prompt variables and raises a ValueError if not. The RunnableBranch is used to handle the logic of whether to use the chat history or not. If there is no chat history, the input is passed directly to the retriever. If there is chat history, the input is processed through the prompt, LLM, and StrOutputParser before being passed to the retriever ^[1].

Here is a brief overview of how the history_aware_retriever is integrated into your script:

1. Creating the History-Aware Retriever:

   from langchain.chains.history_aware_retriever import create_history_aware_retriever
   from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
   
   contextualize_q_system_prompt = """
   Given the chat history and the latest user question, which might reference context in the chat history, 
   reformulate the question into a standalone question that can be understood without the chat history. 
   If the question is related to the most recent queries or requires context from the chat history to be understood, 
   include that context in the reformulated question. Do NOT answer the question; just provide the reformulated version.
   """
   
   contextualize_q_prompt = ChatPromptTemplate.from_messages(
       [
           ("system", contextualize_q_system_prompt),
           MessagesPlaceholder("chat_history"),
           ("human", "{input}"),
       ]
   )
   history_aware_retriever = create_history_aware_retriever(
       llm, retriever, contextualize_q_prompt
   )

2. Using the History-Aware Retriever:

   def get_context(query) -> Iterable:
       return history_aware_retriever.invoke(
           {"input": query}, config={"callbacks": [lang_fuse.get_embedding_callback_handler(metadata=None)]}
       )

3. Integrating with the Response Generation:

   def get_response(
       session_id,
       language: str,
       query: str,
       stream: bool = True,
       enable_chat_history: bool = True,
   ) -> Iterable:
       context = ""
       chain = create_stuff_documents_chain(llm, qa_template if language == "english" else qa_template_arabic)
       
       if enable_chat_history:
           chain = RunnableWithMessageHistory(
               chain,
               get_message_history,
               input_messages_key="query",
               history_messages_key="chat_history",
           )
   
       if stream and not enable_chat_history:
           response = chain.stream(
               {"query": query, "context": context},
               config={"callbacks": [lang_fuse.get_callback_handler(metadata=metadata)]},
           )
           return response
   
       if stream and enable_chat_history:
           response = chain.stream(
               {"query": query, "context": context},
               config={
                   "configurable": {
                       "session_id": f"{session_id}",
                   }, "callbacks": [lang_fuse.get_callback_handler(metadata=metadata)]
               },
           )
           return response

The history_aware_retriever is used to retrieve relevant contexts based on the query and chat history, which are then used to generate responses. The RunnableWithMessageHistory class is used to manage the chat history and integrate it with the main chain ^[2].

To continue talking to Dosu, mention @dosu.

[Amitt1412]

Here is my QA flow

Q1 - What is Service Policy?
Ans - Got service Policy related Answer.

Q2 - Can you tell me more about it?
Ans - Got more detailed answer about the service policy.

Q3 - What is HR policy?
Ans - got HR policy related answer.
Q-4 Can you help me with more information on it?
Ans - Got answer about the service policy.

Here in Q-4 the answer could have been related to the HR policy, but it is answering about service policy which is due to the default mechanism if history aware retriever. How can I control its flow properly?

[dosubot[bot]]

To control the flow of the history_aware_retriever and ensure that the context is correctly maintained and the answers are relevant to the most recent query, you can follow these steps:

1. Instantiate the History-Aware Retriever: Use the create_history_aware_retriever function to create a retriever that incorporates the chat history into the retrieval process.

2. Manage Chat History: Use BaseChatMessageHistory and RunnableWithMessageHistory to manage and update the chat history.

Here is the relevant code to achieve this:

from langchain.chains import create_history_aware_retriever, create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_openai import ChatOpenAI
from langchain_community.chat_message_histories import ChatMessageHistory

# Initialize the LLM
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

# Define the contextualization prompt
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("chat_history"),
        ("human", "{input}"),
    ]
)

# Create the history-aware retriever
history_aware_retriever = create_history_aware_retriever(
    llm, retriever, contextualize_q_prompt
)

# Define the QA prompt
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, say that you "
    "don't know. Use three sentences maximum and keep the "
    "answer concise."
    "\n\n"
    "{context}"
)
qa_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", system_prompt),
        MessagesPlaceholder("chat_history"),
        ("human", "{input}"),
    ]
)
question_answer_chain = create_stuff_documents_chain(llm, qa_prompt)

# Create the retrieval chain
rag_chain = create_retrieval_chain(history_aware_retriever, question_answer_chain)

# Manage chat history
store = {}

def get_session_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]

conversational_rag_chain = RunnableWithMessageHistory(
    rag_chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="chat_history",
    output_messages_key="answer",
)

# Example usage
response = conversational_rag_chain.invoke(
    {"input": "What is Task Decomposition?"},
    config={"configurable": {"session_id": "abc123"}}
)["answer"]

print(response)

This setup ensures that the context is correctly maintained and the answers are relevant to the most recent query by incorporating the chat history into the retrieval process and managing the chat history statefully. The create_history_aware_retriever function is designed to handle cases where there is no chat history by directly passing the input to the retriever, and when there is chat history, it uses the prompt and LLM to generate a search query before passing it to the retriever ^[1][2].

To continue talking to Dosu, mention @dosu.

[Amitt1412]

Actually, upon integrating the same I'm receiving an error message. Can you help me as where I'm doing wrong? `from typing import Iterable
import uuid
from fastapi import HTTPException
from fastapi.responses import StreamingResponse
from langchain_community.chat_message_histories import (
RedisChatMessageHistory,
ChatMessageHistory,
)
from langchain_core.runnables.history import (
RunnableWithMessageHistory,
BaseChatMessageHistory,
)
from langchain.chains.retrieval import create_retrieval_chain
from services import retriever, llm, redis_url
from utilities.logger.custom_logger import get_logger

from services.prompts.prompt_templates import qa_template, qa_template_arabic

from utilities.observability.prompt_creation import (
qa_template,
qa_template_arabic,
# header_prompt_template,
)
from web_drivers.rest import system_config
from web_drivers.rest.schema import QueryRequestSchema
from services.initialization import lang_fuse

from utilities.observability.prompt_config import *

logger = get_logger("response_generator")
chat_store = dict()

metadata = {
"app_name": "slfh_streamlit_app_lngfus",
"app_version": "1.0",
"qa_template": qa_template,
"qa_template_prompt_name": qa_template.name,
"qa_template_prompt_description": qa_template,
"qa_template_arabic": qa_template,
"qa_template_arabic_name": qa_template.name,
}

from langchain.chains.history_aware_retriever import create_history_aware_retriever
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.chains.combine_documents import create_stuff_documents_chain

contextualize_q_system_prompt = """
Given the chat history and the latest user question, which might reference context in the chat history,
reformulate the question into a standalone question that can be understood without the chat history.
If the question is related to the most recent queries or requires context from the chat history to be understood,
include that context in the reformulated question. Do NOT answer the question; just provide the reformulated version.
"""

contextualize_q_prompt = ChatPromptTemplate.from_messages(
[
("system", contextualize_q_system_prompt),
MessagesPlaceholder("chat_history"),
("human", "{input}"),
]
)
history_aware_retriever = create_history_aware_retriever(
llm, retriever, contextualize_q_prompt
)

chat_store = {}
def get_message_history(session_id: str) -> BaseChatMessageHistory:
if session_id not in chat_store:
chat_store[session_id] = ChatMessageHistory()
return chat_store[session_id]

def get_context(query) -> Iterable:
"""
Retrieve contexts relevant to the given query.

Args:
    query (str): The query for which contexts are to be retrieved.

Returns:
    Iterable: A generator yielding the contexts relevant to the query.
"""
return history_aware_retriever.invoke(
    {"input": query},
    config={"callbacks": [lang_fuse.get_embedding_callback_handler(metadata=None)]},
)

def get_response(
session_id,
language: str,
query: str,
stream: bool = True,
enable_chat_history: bool = True,
) -> Iterable:
"""
Generate responses for the given query.

Args: query (str): The query for which responses are to be generated. stream (bool, optional): If True,
response is streamed as an iterable. If False, response is generated as a single object. Defaults to True.

Returns:
    Iterable: A generator yielding the response chunks for the query.
"""
# chat_history = get_message_history(session_id).messages if enable_chat_history else []
# context = "\n".join([contexts.page_content for contexts in get_context(query)])

context = ""

chain = create_stuff_documents_chain(
    llm, qa_template if language == "english" else qa_template_arabic
)
rag_chain = create_retrieval_chain(history_aware_retriever, chain)

if enable_chat_history:
    chain = RunnableWithMessageHistory(
        rag_chain,
        get_message_history,
        input_messages_key="query",
        history_messages_key="chat_history",
    )

if stream and not enable_chat_history:
    response = rag_chain.stream(
        {"query": query, "context": context},
        config={"callbacks": [lang_fuse.get_callback_handler(metadata=metadata)]},
    )
    return response

if stream and enable_chat_history:
    response = rag_chain.stream(
        {"query": query, "context": context},
        config={
            "configurable": {
                "session_id": f"{session_id}",
            },
            "callbacks": [lang_fuse.get_callback_handler(metadata=metadata)],
        },
    )
    return response

async def query_request_handler(
language: str, payload: QueryRequestSchema, session_id: str
) -> StreamingResponse:
"""
Handles incoming requests and returns streaming response.

Args:
    payload (QueryRequest): Incoming request containing query.
    language (str): Language of conversation.
    session_id (str): Session ID for maintaining chat history.

Returns:
    StreamingResponse: Streaming response containing chunks of data.
"""
lang_fuse.set_attr(session_id=session_id, user_id=str(uuid.uuid1()))
try:
    # Generate stream of chunks based on query
    chunk_stream_gen = get_response(
        language=language,
        query=payload.query,
        session_id=session_id,
        enable_chat_history=system_config.get("ENABLE_CHAT_HISTORY"),
    )
    return StreamingResponse(
        (chunk for chunk in chunk_stream_gen), media_type="text/event-stream"
    )
except Exception as e:
    # Raise HTTP exception if there's an error
    raise HTTPException(status_code=500, detail=str(e))`

File "C:\Users\AmitTiwari\snapwork_projects\virtual_envs\rndenv\lib\site-packages\starlette_utils.py", line 93, in collapse_excgroups
raise exc
File "C:\Users\AmitTiwari\snapwork_projects\virtual_envs\rndenv\lib\site-packages\starlette\responses.py", line 261, in wrap
await func()
File "C:\Users\AmitTiwari\snapwork_projects\virtual_envs\rndenv\lib\site-packages\starlette\responses.py", line 252, in stream_response
chunk = chunk.encode(self.charset)
AttributeError: 'AddableDict' object has no attribute 'encode'