Trouble linking ContextualCompressionRetriever to already working RAG pipeline with Context Aware Retriever. AttributeError: 'str' object has no attribute 'get'

[ignaceHelsen]

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

from typing import List

from langchain_core.documents import Document
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_huggingface import HuggingFaceEmbeddings
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from pydantic import BaseModel, Field

from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain.chains.history_aware_retriever import create_history_aware_retriever
from langchain.chains.retrieval import create_retrieval_chain
from langchain.retrievers.contextual_compression import ContextualCompressionRetriever
from langchain_community.document_compressors.rankllm_rerank import RankLLMRerank
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.messages import BaseMessage
from langchain_core.tracers import ConsoleCallbackHandler
from langchain_ollama import OllamaLLM
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder, FewShotChatMessagePromptTemplate
from langchain_core.runnables import RunnableWithMessageHistory
from langchain_community.vectorstores import VectorStore
from langchain_community.chat_message_histories import ChatMessageHistory


class InMemoryHistory(BaseChatMessageHistory, BaseModel):
    """In memory implementation of chat message history."""

    messages: List[BaseMessage] = Field(default_factory=list)

    def add_messages(self, messages: List[BaseMessage]) -> None:
        """Add a list of messages to the store"""
        self.messages.extend(messages)

    def clear(self) -> None:
        self.messages = []


class Chatbot:
    def __init__(self, llm: OllamaLLM, vector_store: VectorStore, amount_of_chunks: int = 4, score_threshold: float = 0.5):
        self.llm = llm
        self.vectorstore = vector_store
        self.amount_of_chunks = amount_of_chunks
        self.threshold = score_threshold
        self.prompts = {'rag_prompt': "Answer... \n Chat history:\n        {chat_history}\n\n        Context:\n        {context}",
                        'rewrite_question_by_history_prompt': "Given... reformulate...",
                        'few_shot_examples': [{"input": "Where...", "output": "You can..."},
            {"input": "Tell me...", "output": "You douche...."}]}

        ### RETRIEVER ###
        retriever = vector_store.as_retriever(search_kwargs={"k": amount_of_chunks, "score_threshold": score_threshold})

        ### CONTEXT AWARE RETRIEVER
        contextualize_q_system_prompt = self.prompts['rewrite_question_by_history_prompt']

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

        history_aware_retriever = create_history_aware_retriever(
            llm=llm, retriever=retriever, prompt=contextualize_q_prompt
        )

        template = self.prompts['rag_prompt']

        ### FEW-SHOT ###
        # prompt = ChatPromptTemplate.from_template(template)
        examples = self.prompts['few_shot_examples']
        example_prompt = ChatPromptTemplate.from_messages(
            [
                ("human", "{input}"),
                ("ai", "{output}"),
            ]
        )
        few_shot_prompt = FewShotChatMessagePromptTemplate(
            example_prompt=example_prompt,
            examples=examples,
        )

        prompt = ChatPromptTemplate.from_messages(
            [
                ("system", template),
                few_shot_prompt,
                MessagesPlaceholder(variable_name="chat_history"),
                ("human", "{input}"),
            ]
        )

        # Message history handling
        self.memory = {}

        ### RERANKER ###
        compressor = RankLLMRerank(top_n=3, model="zephyr", device='cpu')
        compression_retriever = ContextualCompressionRetriever(
            base_compressor=compressor, base_retriever=history_aware_retriever
        )

        ### CHAIN ###
        question_answer_chain = create_stuff_documents_chain(llm, prompt)

        rag_chain = create_retrieval_chain(retriever=compression_retriever, combine_docs_chain=question_answer_chain)

        self.chain = RunnableWithMessageHistory(
            rag_chain,
            self.get_chat_history_by_session_id,
            input_messages_key="input",
            history_messages_key="chat_history",
            output_messages_key="answer"
        )

    def get_chat_history_by_session_id(self, session_id: str) -> BaseChatMessageHistory:
        if session_id not in self.memory:
            self.memory[session_id] = ChatMessageHistory()
        return self.memory[session_id]

    # Returns the answer to the question
    def generate_answer(self, question: str, session_id ='abc-123'):
        #docs = await self.retrieve_with_scores(question)
        #print(session_id)

        #config = {"configurable": {"session_id": session_id}, 'callbacks': [ConsoleCallbackHandler()]}
        config = {"configurable": {"session_id": session_id}}
        return self.chain.invoke({"input": question}, config)



if __name__ == '__main__':
    llm = ChatOpenAI(
        model="gpt-4o",
        temperature=0,
        max_tokens=None,
        timeout=None
    )

    embeddings = HuggingFaceEmbeddings(model_name="Snowflake/snowflake-arctic-embed-l-v2.0")

    vector_store = InMemoryVectorStore(embeddings)

    document_1 = Document(id="1", page_content="foo", metadata={"baz": "bar"})
    document_2 = Document(id="2", page_content="thud", metadata={"bar": "baz"})
    document_3 = Document(id="3", page_content="i will be deleted :(")

    documents = [document_1, document_2, document_3]
    vector_store.add_documents(documents=documents)
    chatbot = Chatbot(llm, vector_store, 10, 0.8)
    chatbot.generate_answer('What is langchain?')

Description

Greetings,

I have setup a conversational RAG that uses a context aware retriever, together with few-shot prompting and chat history. I am now trying to incorporate a Reranker into the already working chain but I'm failing to set it up correctly. I'm experiencing troubles setting up the link between the context aware retriever and the new reranker retriever.
Setting up the chain is successfull but when running invoke on the chain, I'm getting following traceback.

Thank you for any help!

Traceback (most recent call last):
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/gradio/queueing.py", line 625, in process_events
response = await route_utils.call_process_api(
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/gradio/route_utils.py", line 322, in call_process_api
output = await app.get_blocks().process_api(
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/gradio/blocks.py", line 2044, in process_api
result = await self.call_function(
^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/gradio/blocks.py", line 1603, in call_function
prediction = await utils.async_iteration(iterator)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/gradio/utils.py", line 728, in async_iteration
return await anext(iterator)
^^^^^^^^^^^^^^^^^^^^^
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/gradio/utils.py", line 833, in asyncgen_wrapper
response = await iterator.anext()
^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/gradio/chat_interface.py", line 898, in _stream_fn
first_response = await utils.async_iteration(generator)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/gradio/utils.py", line 728, in async_iteration
return await anext(iterator)
^^^^^^^^^^^^^^^^^^^^^
File "/home/ignace/projects/llm-finetunen/src/main.py", line 35, in chat_fn
async for chunk in rag_system.ask_question(question=message, session_id='abc-123'):
File "/home/ignace/projects/llm-finetunen/src/rag/rag_system.py", line 45, in ask_question
async for chunk in self.chatbot.generate_answer(question=question, session_id=session_id):
File "/home/ignace/projects/llm-finetunen/src/rag/chatbot.py", line 134, in generate_answer
if 'answer' in chunk:
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/base.py", line 5540, in astream
async for item in self.bound.astream(
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/base.py", line 5540, in astream
async for item in self.bound.astream(
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/base.py", line 3431, in astream
async for chunk in self.atransform(input_aiter(), config, **kwargs):
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/base.py", line 3414, in atransform
async for chunk in self._atransform_stream_with_config(
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/base.py", line 2295, in _atransform_stream_with_config
chunk: Output = await asyncio.create_task( # type: ignore[call-arg]
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/base.py", line 3384, in _atransform
async for output in final_pipeline:
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/base.py", line 5576, in atransform
async for item in self.bound.atransform(
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/base.py", line 4946, in atransform
async for output in self._atransform_stream_with_config(
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/base.py", line 2295, in _atransform_stream_with_config
chunk: Output = await asyncio.create_task( # type: ignore[call-arg]
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/base.py", line 4927, in _atransform
async for chunk in output.astream(
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/base.py", line 5540, in astream
async for item in self.bound.astream(
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/base.py", line 5540, in astream
async for item in self.bound.astream(
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/base.py", line 3431, in astream
async for chunk in self.atransform(input_aiter(), config, **kwargs):
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/base.py", line 3414, in atransform
async for chunk in self._atransform_stream_with_config(
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/base.py", line 2295, in _atransform_stream_with_config
chunk: Output = await asyncio.create_task( # type: ignore[call-arg]
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/base.py", line 3384, in _atransform
async for output in final_pipeline:
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/passthrough.py", line 627, in atransform
async for chunk in self._atransform_stream_with_config(
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/base.py", line 2295, in _atransform_stream_with_config
chunk: Output = await asyncio.create_task( # type: ignore[call-arg]
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/passthrough.py", line 605, in _atransform
async for chunk in for_passthrough:
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/utils/aiter.py", line 123, in tee_peer
item = await iterator.anext()
^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/utils/aiter.py", line 123, in tee_peer
item = await iterator.anext()
^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/passthrough.py", line 627, in atransform
async for chunk in self._atransform_stream_with_config(
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/base.py", line 2295, in _atransform_stream_with_config
chunk: Output = await asyncio.create_task( # type: ignore[call-arg]
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/passthrough.py", line 617, in _atransform
yield await first_map_chunk_task
^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/utils/aiter.py", line 74, in anext_impl
return await anext(iterator)
^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/base.py", line 3915, in atransform
async for chunk in self._atransform_stream_with_config(
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/base.py", line 2295, in _atransform_stream_with_config
chunk: Output = await asyncio.create_task( # type: ignore[call-arg]
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/base.py", line 3902, in _atransform
chunk = AddableDict({step_name: task.result()})
^^^^^^^^^^^^^
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/base.py", line 3885, in get_next_chunk
return await py_anext(generator)
^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/base.py", line 5576, in atransform
async for item in self.bound.atransform(
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/base.py", line 3414, in atransform
async for chunk in self._atransform_stream_with_config(
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/base.py", line 2295, in _atransform_stream_with_config
chunk: Output = await asyncio.create_task( # type: ignore[call-arg]
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/base.py", line 3384, in _atransform
async for output in final_pipeline:
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/base.py", line 1465, in atransform
async for output in self.astream(final, config, **kwargs):
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/base.py", line 1012, in astream
yield await self.ainvoke(input, config, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/retrievers.py", line 322, in ainvoke
result = await self._aget_relevant_documents(
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain/retrievers/contextual_compression.py", line 70, in _aget_relevant_documents
docs = await self.base_retriever.ainvoke(
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/base.py", line 5369, in ainvoke
return await self.bound.ainvoke(
^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/branch.py", line 271, in ainvoke
expression_value = await condition.ainvoke(
^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/base.py", line 4748, in ainvoke
return await self._acall_with_config(
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/base.py", line 1966, in _acall_with_config
output: Output = await asyncio.create_task(coro, context=context) # type: ignore
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/base.py", line 4672, in _ainvoke
output = await acall_func_with_variable_args(
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/base.py", line 4642, in f
return await run_in_executor(config, func, *args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/config.py", line 588, in run_in_executor
return await asyncio.get_running_loop().run_in_executor(
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/concurrent/futures/thread.py", line 58, in run
result = self.fn(*self.args, **self.kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/config.py", line 579, in wrapper
return func(*args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/base.py", line 4636, in func
return call_func_with_variable_args(
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain_core/runnables/config.py", line 396, in call_func_with_variable_args
return func(input, **kwargs) # type: ignore[call-arg]
^^^^^^^^^^^^^^^^^^^^^
File "/home/ignace/anaconda3/envs/llm/lib/python3.11/site-packages/langchain/chains/history_aware_retriever.py", line 60, in
lambda x: not x.get("chat_history", False),
^^^^^
AttributeError: 'str' object has no attribute 'get'

System Info

langchain==0.3.17
langchain-cli==0.0.35
langchain-community==0.3.16
langchain-core==0.3.33
langchain-huggingface==0.1.2
langchain-ollama==0.2.2
langchain-openai==0.3.2
langchain-qdrant==0.2.0
langchain-text-splitters==0.3.5
rank_llm==0.20.3

[ignaceHelsen]

@DosuBot

[startde]

hello, did you find solution?

[ignaceHelsen]

@startde
Kind of, I got it working by doing the following:

...

history_aware_retriever = create_history_aware_retriever(
            llm=llm, retriever=retriever, prompt=contextualize_q_prompt
        )

...

ranker = RankLLMRerank(model=reranker_model, top_n=amount_of_chunks_reranker, gpt_model=self.reranker_gpt_model)
reranker = lambda input: ranker.compress_documents(query=input['input']['input'], documents=[Document(page_content=x.page_content if x.page_content else 'empty', metadata=x.metadata) for x in input['context']])
rag_chain = (
                    RunnableParallel({"context": history_aware_retriever, "input": RunnablePassthrough(),
                                      "chat_history": lambda input: input.get("chat_history", [])})
                    | {"context": reranker, "input": lambda input: input['input'],
                       "chat_history": lambda input: input['input'].get("chat_history", [])}
                    | prompt
                    | llm
                    | StrOutputParser()
            )
chain = RunnableWithMessageHistory(
                rag_chain,
                self.get_chat_history_by_session_id,
                input_messages_key="input",
                history_messages_key="chat_history"
                # output_messages_key="answer"
            )

[startde]

@ignaceHelsen
thanks a lot