How do I create a shared variable for a pipeline, such as query, which needs to be used by many components in the pipeline?

[springrain]

How do I create a shared variable for a pipeline, such as query, which needs to be used by many components in the pipeline?

[davidsbatista]

can you give a concrete example of the pipeline you are building and what do you want to share and across which components?

[springrain]

@davidsbatista I use conditional routing, and if the query entered by the user is too short, nothing is done, but because the query parameter of the reranker component must be passed in the run method, the reranker component must be executed, as you can see from the pipeline diagram

rag_pipeline.connect("router.ok_query", "text_embedder.text")
rag_pipeline.connect("text_embedder.embedding", "retriever.query_embedding")
rag_pipeline.connect("retriever", "documentScore")
rag_pipeline.connect("documentScore", "reranker")
rag_pipeline.connect("reranker", "prompt_builder")
rag_pipeline.connect("prompt_builder", "llm")

query = "test?"
result = rag_pipeline.run(data={"router": {"query":query},"reranker": {"query":query},"prompt_builder": {"query":query}})
print(result)

[davidsbatista]

a quick solution would be to have a function - detached from the pipeline - that checks the user query before passing it to the pipeline and making it run.

[springrain]

How can each query parameter from user input be automatically obtained without passing through the rag_pipeline.run function? If the query parameters are passed through the rag_pipeline.run function, then the process connection from start to reranker does not exist.

[davidsbatista]

I'm not aware of the whole of your code, but this was what I had in mind:

def check_user_query(query: str):
    """
    Check if the user query is valid
    """
    if query is None:
        raise ValueError("Query is None")
    if not isinstance(query, str):
        raise ValueError("Query is not a string")
    if len(query) < MIN_QUERY_LENGTH:
        raise ValueError("Query must be larger")
    return True

if check_user_query(query):
    result = rag_pipeline.run(
        data={"router": {"query": query}, "reranker": {"query": query}, "prompt_builder": {"query": query}})
    print(result)
else:
    print(f"Your query needs to be larger than {MIN_QUERY_LENGTH}")

NOTE: I'm assuming that you only want to run the pipeline if the query is larger than some MIN_QUERY_LENGTH value.

[springrain]

Ended up like this: #8028