Unknown key for a START_OBJECT in [knn]

[bansh317]

Hi. Please help!

Using Haystack 2.0, I have built a RAG pipeline that works without error when I use either InMemoryDocumentStore with InMemoryBM25Retreiver or with ElasticsearchDocumentStore with ElasticsearchBM25Retriever. The issue is, I want more robust results for slightly more nuanced retrieval so I'm trying to implement a hybrid retrieval pipeline as such:

document_store = ElasticsearchDocumentStore(hosts="http://localhost:9200", index="cls_rx_document")

initialized:
first, I initialize:

text_embedder = SentenceTransformersTextEmbedder(model="sentence-transformers/all-MiniLM-L6-v2")

bm25_retriever = ElasticsearchBM25Retriever(document_store=document_store, top_k=3)
embedding_retriever = ElasticsearchEmbeddingRetriever(document_store=document_store, top_k=3)

document_joiner = DocumentJoiner(join_mode="concatenate", top_k=5)

ranker = TransformersSimilarityRanker(model="cross-encoder/ms-marco-MiniLM-L-6-v2", top_k=5)

(I also define my template and context for the prompt generator)

Then, i build my pipelines:
Indexing:

filepath = "my_file.docx"
def read_docx(file_path):
    doc = DocxDocument(file_path)
    fullText = []
    for para in doc.paragraphs:
        fullText.append(para.text)
    return "\n".join(fullText)


docx_content = read_docx(filepath)

# have to create a Document object with the content of the .docx file
doc = Document(content=docx_content)

# the document needs to be in a list for the document splitter to do its thing
docs = [doc]

document_splitter = DocumentSplitter(split_by="word", split_length=250, split_overlap=50)
document_embedder = SentenceTransformersDocumentEmbedder(model="BAAI/bge-small-en-v1.5")
document_writer = DocumentWriter(document_store, policy=DuplicatePolicy.SKIP)

pipeline = Pipeline()
pipeline.add_component("document_splitter", document_splitter)
pipeline.add_component("document_embedder", document_embedder)
pipeline.add_component("document_writer", document_writer)
pipeline.connect("document_splitter", "document_embedder")
pipeline.connect("document_embedder", "document_writer")
pipeline.run({"document_splitter": {"documents": docs}})

prompt_builder = PromptBuilder(template=template)

hybrid_retrieval = Pipeline()
hybrid_retrieval.add_component("text_embedder", text_embedder)
hybrid_retrieval.add_component("embedding_retriever", embedding_retriever)
hybrid_retrieval.add_component("bm25_retriever", bm25_retriever)
hybrid_retrieval.add_component("prompt_builder", prompt_builder)
hybrid_retrieval.add_component("llm", generator)
hybrid_retrieval.add_component("document_joiner", document_joiner)
hybrid_retrieval.add_component("ranker", ranker)

hybrid_retrieval.connect("text_embedder", "embedding_retriever")
hybrid_retrieval.connect("bm25_retriever", "document_joiner")
hybrid_retrieval.connect("embedding_retriever", "document_joiner")
hybrid_retrieval.connect("document_joiner", "ranker")
hybrid_retrieval.connect("ranker", "prompt_builder.documents")
hybrid_retrieval.connect("prompt_builder", "llm")

Now, I manage my conversational/context history and launch the UI with gradio:

MAX_HISTORY = 1  # this is so we dont unnecessarily rack up the input token count

def ask_question(question):
    conversation_history.append({"question": question, "answer": ""})
    
    response = hybrid_retrieval.run(
        {   
            "text_embedder": {"text": question},
            "bm25_retriever": {"query": question},
            # "embedding_retriever": {"query": question},
            "document_joiner": {"documents": docs},
            "ranker" : {"query": question},
            "prompt_builder": {"question": question, "history": conversation_history}
        }
    )
    
    conversation_history[-1]["answer"] = response["llm"]["replies"][0]

    # lets keep only the latest MAX_HISTORY entries in the conversation history
    if len(conversation_history) > MAX_HISTORY:
        conversation_history[:] = conversation_history[-MAX_HISTORY:]
    
    return response["llm"]["replies"][0]

gr_interface = gr.Interface(
    fn=ask_question,
    inputs=gr.components.Textbox(lines=3, placeholder="Please enter your question here..."),
    outputs="text",
    title="retreiver"
)

gr_interface.launch(share=True)

I receive the following error with this set up:

Traceback (most recent call last):
  File "/home/ubuntu/dev/Optic_AI/.venv/lib/python3.10/site-packages/gradio/queueing.py", line 501, in call_prediction
    output = await route_utils.call_process_api(
  File "/home/ubuntu/dev/Optic_AI/.venv/lib/python3.10/site-packages/gradio/route_utils.py", line 253, in call_process_api
    output = await app.get_blocks().process_api(
  File "/home/ubuntu/dev/Optic_AI/.venv/lib/python3.10/site-packages/gradio/blocks.py", line 1695, in process_api
    result = await self.call_function(
  File "/home/ubuntu/dev/Optic_AI/.venv/lib/python3.10/site-packages/gradio/blocks.py", line 1235, in call_function
    prediction = await anyio.to_thread.run_sync(
  File "/home/ubuntu/dev/Optic_AI/.venv/lib/python3.10/site-packages/anyio/to_thread.py", line 56, in run_sync
    return await get_async_backend().run_sync_in_worker_thread(
  File "/home/ubuntu/dev/Optic_AI/.venv/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 2144, in run_sync_in_worker_thread
    return await future
  File "/home/ubuntu/dev/Optic_AI/.venv/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 851, in run
    result = context.run(func, *args)
  File "/home/ubuntu/dev/Optic_AI/.venv/lib/python3.10/site-packages/gradio/utils.py", line 692, in wrapper
    response = f(*args, **kwargs)
  File "/home/ubuntu/dev/Optic_AI/Copilot/RxRAG_v1.py", line 342, in ask_question
    response = hybrid_retrieval.run(
  File "/home/ubuntu/dev/Optic_AI/.venv/lib/python3.10/site-packages/haystack/core/pipeline/pipeline.py", line 823, in run
    res = comp.run(**last_inputs[name])
  File "/home/ubuntu/dev/Optic_AI/.venv/lib/python3.10/site-packages/haystack_integrations/components/retrievers/elasticsearch/embedding_retriever.py", line 75, in run
    docs = self._document_store._embedding_retrieval(
  File "/home/ubuntu/dev/Optic_AI/.venv/lib/python3.10/site-packages/haystack_integrations/document_stores/elasticsearch/document_store.py", line 369, in _embedding_retrieval
    docs = self._search_documents(**body)
  File "/home/ubuntu/dev/Optic_AI/.venv/lib/python3.10/site-packages/haystack_integrations/document_stores/elasticsearch/document_store.py", line 152, in _search_documents
    res = self._client.search(
  File "/home/ubuntu/dev/Optic_AI/.venv/lib/python3.10/site-packages/elasticsearch/_sync/client/utils.py", line 446, in wrapped
    return api(*args, **kwargs)
  File "/home/ubuntu/dev/Optic_AI/.venv/lib/python3.10/site-packages/elasticsearch/_sync/client/__init__.py", line 3836, in search
    return self.perform_request(  # type: ignore[return-value]
  File "/home/ubuntu/dev/Optic_AI/.venv/lib/python3.10/site-packages/elasticsearch/_sync/client/_base.py", line 320, in perform_request
    raise HTTP_EXCEPTIONS.get(meta.status, ApiError)(
elasticsearch.BadRequestError: BadRequestError(400, 'parsing_exception', 'Unknown key for a START_OBJECT in [knn].')

Is something misconfigured in my pipeline? I am dynamically mapping the ES index, and like I said it worked with sparse retrieval, but the integration of hybrid retrieval shows 'Unknown key for a START_OBJECT in [knn].

Thank you!

[vblagoje]

Sounds like something is off either an incorrect API endpoint or a misconfiguration in your elasticsearch setup. Let me ask a colleague who is more familiar with elasticsearch for an advice here cc @silvanocerza