Help needed with Pipeline

[marfago]

Hello,

I am completely new to Haystack, and I am trying to learn it by example.
I am currently focusing on combining pipelines and Gemini, in particular I am following this example:

import os
from haystack.components.retrievers.in_memory import InMemoryBM25Retriever
from haystack.components.builders import PromptBuilder
from haystack import Pipeline
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack_integrations.components.generators.google_ai import GoogleAIGeminiGenerator

os.environ["GOOGLE_API_KEY"] = "<MY_API_KEY>"

docstore = InMemoryDocumentStore()

template = """
Given the following information, answer the question.

Context: 
{% for document in documents %}
    {{ document.content }}
{% endfor %}

Question: What's the official language of {{ country }}?
"""
pipe = Pipeline()

pipe.add_component("retriever", InMemoryBM25Retriever(document_store=docstore))
pipe.add_component("prompt_builder", PromptBuilder(template=template))
pipe.add_component("gemini", GoogleAIGeminiGenerator(model="gemini-pro"))
pipe.connect("retriever", "prompt_builder.documents")
pipe.connect("prompt_builder", "gemini")

pipe.run({
    "prompt_builder": {
        "country": "France"
    }
})

However, when I run the code it returns an error:

raise ValueError(f"Missing input for component {component_name}: {socket_name}")
ValueError: Missing input for component retriever: query

I don't know what is wrong with this example and I am looking for some help/direction.
Thank you.

Environment: Microsoft Windows 11 Home

haystack-ai==2.8.0
google-ai-haystack==3.0.2

[d-kleine]

The error says that retriever (which is InMemoryBM25Retriever in your case) does not provide any input for query. But this is mandatory for running your retriever, see here in the table:

Mandatory run variables | "query": A query string

So, you need to change the prompt format in the template to

country = "France"
question = f"What is the official language of {country}?"

template = """
...

Question: {{ question }}
"""

and pass a query string (contained in the question variable for the value) to the "query" key for the retriever in your pipe.run() workflow, for example:

result = pipe.run({
        "retriever": {"query": question}, # added 
        "prompt_builder": {"question": question} # changed
    })

Both stages require question as:

• "retriever": Needs question to find relevant documents.
• "prompt_builder": Needs question to format it into a coherent prompt for the LLM.

---

If you want to iterate through each country defined in a list countries, you could do it like this:

countries = ["France", "Germany", "Spain"] 
...

for country in countries: # iterate through every country in list
    question = f"What is the official language of {country}?"
    
    result = pipe.run({
        "retriever": {"query": question},
        "prompt_builder": {"question": question}
    })

I cannot test it, but this would be the logic I would go with. I hope this helps.

[marfago]

@d-kleine thank you for explaining. I have a follow up question. Let's say that you have a pipeline with 2 components A and B connected in a loop A->B->A. Suppose that initially you provide a query to A which generates a single input for B which in turns produces 5 different texts. Is it possible to feed A with each independent piece of text (which would then generate 5 new loops)? Also, what would be the way to define the exit branch in the graph?

Note: Please note that this example above is a copy&paste from the official doc here which according to the explanation above requires a correction.

[marfago]

Can you please provide an example or a reference doc?

[d-kleine]

I'm just a fellow user but my idea/approach would be the following:

1. Defining two custom components ComponentA (A) and ComponentB (B) performing the tasks, where A generates a single query and B produces 5 different output texts (e.g. for a for-loop). A would return an output that would serve as input for B.
   https://docs.haystack.deepset.ai/docs/custom-components

2. Setting up ConditionalRouter as router to control the loop for given conditions, setting a condition for exiting the loop when the output it wrong (e.g. if the length of the output of A = 0, so there is output created), and a condition for continuing the loop when the outputs is correct (e.g. if the length of the output of A > 0).
   https://docs.haystack.deepset.ai/docs/conditionalrouter

3. Creating a pipeline by adding the components (ComponentA, ComponentB, router) and connecting them accordingly to your logic.
   https://docs.haystack.deepset.ai/reference/pipeline-api

[marfago]

I am not sure that what you suggest will work. The point is that the output B does not produce 5 different outputs but rather 1 output with 5 different texts. So, if I feed back A with it it would mean that A takes 1 single input with 5 pieces of text to process. What I want to achieve is B producing a List[str] and A consuming str. So, if B produces a list with 5 strings, A should be executed 5 times with each different str from the list.

[julian-risch]

Thank you @marfago for your question and @d-kleine for answering! I updated the code example. You're right that two components in the code example, the retriever and the prompt builder need to receive the query/country.
For the other part of the question, my suggestion is to use a custom component that processes the List[str] internally instead of splitting the list in str again and having a component that takes str as input in a loop.
@marfago Feel free to describe a bit more what you are trying to build and I'd be happy to help!

[marfago]

@julian-risch Thank you for replying. I have been thinking about processing the list in a single node, but it does not work with my use case. Imagine there is a pipeline like this:

(Some links)->A->B->C->A

with a node A which perpetually crawls documents based on a input list of links, then it calls a node B to extract new links from each document and then a node C which summarizes each document with a LLM. Node B feeds back the new links to the node A and the cycle restarts. Assuming that each doc contains a number of links to other documents, I see 2 problems:

1. Since C needs to parse multiple documents together the output of the LLM may be less accurate than when calling the
2. The number of documents propagated in the pipeline will grow exponentially and at some point a single node will not be able to process all them

The solution that I have in mind is a custom node D that can accept a list of items and return a single item. This node will work as a LIFO/FIFO queue and would keep accumulating documents in its internal queue. It will also eventually dedupe the documents.
In this case the pipeline would be

(some links)->D(dedupes and emits a single node)->A(crawls a single document)->B(extracts links)->C(summarizes a single document)->D

I think it can be easily built, just I wanted to understand if such a component is already implemented and if my approach makes sense to you.