RuntimeWarning: Pipeline is stuck running in a loop. Partial outputs will be returned. Check the Pipeline graph for possible issues.

[Shuntw6096]

here is my code. it works expected currently, but I don't know why I always get the warning messages.

.venv/lib/python3.11/site-packages/haystack/core/pipeline/pipeline.py:523: RuntimeWarning: Pipeline is stuck running in a loop. Partial outputs will be returned. Check the Pipeline graph for possible issues.
  warn(RuntimeWarning(msg))

Environment is

haystack-ai           2.9.0rc0 8e3f647 
haystack-experimental 0.3.0 

from haystack import component, Document, Pipeline
from haystack.dataclasses import ChatMessage
from haystack.components.builders import PromptBuilder, ChatPromptBuilder
from typing import List, Optional, Dict, Any


@component
class FakeEmbedder:
    @component.output_types(embedding=List[float])
    def run(self, text: str):
        return {"embedding": [0.2] * 101}


@component
class FakeDocumentRetriever:
    @component.output_types(
        documents_1=List[Document],
        documents_2=List[Document],
        documents_3=List[Document],
    )
    def run(
        self,
        query_embedding: List[float],
        top_k_1: Optional[int] = None,
        top_k_2: Optional[int] = None,
        top_k_3: Optional[int] = None,
    ):

        return {
            "documents_1": [Document(content="sds", embedding=[0.2] * 101)],
            "documents_2": [Document(content="sdss", embedding=[0.22] * 101)],
            "documents_3": [Document(content="dfss", embedding=[0.255] * 101)],
        }


@component
class FakeRetriever:
    i = 0
    output_order = ["error_messages", "error_messages", "error_messages", "reply"]

    @component.output_types(reply=str, error_messages=List[ChatMessage])
    def run(self, messages: List[ChatMessage]):
        res = "error_messages" if self.i < 1 else "reply"
        print(res)
        self.i += 1
        return {
            res: [
                ChatMessage.from_assistant("Ha Ha Ha"),
                ChatMessage.from_user("You a robot"),
            ]
        }


@component
class FakePromptBuilder:
    @component.output_types(prompt=List[ChatMessage])
    def run(
        self,
        question: str,
        documents_1: List[Document],
        documents_2: List[Document],
        documents_3: List[Document],
    ):
        return {"prompt": [ChatMessage.from_user("Nice to meet you, how are you?")]}


@component
class FakeGen:
    @component.output_types(replies=List[ChatMessage])
    def run(
        self,
        messages: List[ChatMessage],
        generation_kwargs: Optional[Dict[str, Any]] = None,
    ):
        return {"replies": [ChatMessage.from_assistant("I'm good!")]}


@component
class FakeGen_Err:
    @component.output_types(replies=List[ChatMessage])
    def run(
        self,
        messages: List[ChatMessage],
        error_messages: Optional[List[ChatMessage]] = None,
    ):
        # if error_messages:
        #     messages.append(error_messages[:])
        return {"replies": [ChatMessage.from_assistant("I'm good!")]}


pipe = Pipeline(max_runs_per_component=3)
pipe.add_component(
    name="query_builder",
    instance=PromptBuilder("{{query}}", required_variables=["query"]),
)
pipe.add_component(name="embedder", instance=FakeEmbedder())
pipe.add_component(name="doc_retriever", instance=FakeDocumentRetriever())
pipe.add_component(name="fake_prompt_biulder", instance=FakePromptBuilder())
pipe.add_component(name="gen_err", instance=FakeGen_Err())
pipe.add_component(name="fake_retriever", instance=FakeRetriever())
pipe.add_component(
    name="chat_prompt_builder",
    instance=ChatPromptBuilder(
        template=[
            ChatMessage.from_user(("{{data_string}} \n" "{{query}} \n")),
        ]
    ),
)
pipe.add_component(name="gen", instance=FakeGen())
pipe.connect("query_builder", "embedder")
pipe.connect("embedder.embedding", "doc_retriever.query_embedding")
pipe.connect("query_builder.prompt", "fake_prompt_biulder.question")
pipe.connect("doc_retriever.documents_1", "fake_prompt_biulder.documents_1")
pipe.connect("doc_retriever.documents_2", "fake_prompt_biulder.documents_2")
pipe.connect("doc_retriever.documents_3", "fake_prompt_biulder.documents_3")

pipe.connect("fake_prompt_biulder.prompt", "gen_err.messages")
pipe.connect("gen_err.replies", "fake_retriever.messages")
pipe.connect("fake_retriever.error_messages", "gen_err.error_messages")
pipe.connect("fake_retriever.reply", "chat_prompt_builder.data_string")
pipe.connect("query_builder.prompt", "chat_prompt_builder.query")

pipe.connect("chat_prompt_builder.prompt", "gen.messages")
print(pipe.run({"query_builder": {"query": "boom"}}))
print(
    pipe.graph.nodes["fake_retriever"]["visits"], pipe.graph.nodes["gen_err"]["visits"]
)

# error_messages
# reply
# {'gen': {'replies': [ChatMessage(_role=<ChatRole.ASSISTANT: 'assistant'>, _content=[TextContent(text="I'm good!")], _name=None, _meta={})]}}
# 2 2

I can't draw the pipeline picture, it seems https://mermaid.ink/ not works.
the point is that there is a loop between FakeRetriever.error_messages and FakeGen_Err.error_messages,
when FakeRetriever send an error_messages to FakeGen_Err, the warning will happen.

This is minimal reproduction code.

from haystack import component, Pipeline
from haystack.dataclasses import ChatMessage
from typing import List, Optional


@component
class FakeRetriever:
    i = 0
    output_order = ["error_messages", "error_messages", "error_messages", "reply"]

    @component.output_types(reply=List[ChatMessage], error_messages=Optional[List[ChatMessage]])
    def run(self, messages: List[ChatMessage]):
        res = "error_messages" if self.i < 1 else "reply"
        print(res)
        self.i += 1
        return {
            res: [
                ChatMessage.from_assistant("Ha Ha Ha"),
                ChatMessage.from_user("You a robot"),
            ]
        }


@component
class FakeGen_Err:
    @component.output_types(replies=List[ChatMessage])
    def run(
        self,
        messages: List[ChatMessage],
        error_messages: Optional[List[ChatMessage]] = None,
    ):
        # if error_messages:
        #     messages.append(error_messages[:])
        return {"replies": [ChatMessage.from_assistant("I'm good!")]}


MX_RUN = 3
pipe = Pipeline(max_runs_per_component=MX_RUN)
pipe.add_component(name="gen_err", instance=FakeGen_Err())
pipe.add_component(name="fake_retriever", instance=FakeRetriever())
pipe.connect("gen_err.replies", "fake_retriever.messages")
pipe.connect("fake_retriever.error_messages", "gen_err.error_messages")

print(pipe.run({"gen_err": {"messages": [ChatMessage.from_user("boom!")]}}))
print(
    pipe.graph.nodes["fake_retriever"]["visits"], pipe.graph.nodes["gen_err"]["visits"]
)

# .venv/lib/python3.11/site-packages/haystack/core/pipeline/pipeline.py:523: RuntimeWarning: Pipeline is stuck running in a loop. # Partial outputs will be returned. Check the Pipeline graph for possible issues.
# {'fake_retriever': {'reply': [ChatMessage(_role=<ChatRole.ASSISTANT: 'assistant'>, _content=[TextContent(text='Ha Ha Ha')], _name=None, _meta={}), ChatMessage(_role=<ChatRole.USER: 'user'>, _content=[TextContent(text='You a robot')], _name=None, _meta={})]}}
# 2 2

Thanks for reading this.

[julian-risch]

Hello @Shuntw6096 we investigated some issues with the current pipeline run logic. One of the aspects of that logic is that it tries to determine if it is stuck in a loop and if so, breaks out of it, which is when you would get that warning.
We re-wrote the pipeline run logic and would like to invite you to try out the new logic. All you need to do is to install haystack in a fresh environment and change the line of code where you import Pipeline. For trying out the new logic, you would simply need to do: from haystack_experimental import Pipeline
Here are more details about the new logic, we are currently testing: https://github.com/deepset-ai/haystack-experimental/releases/tag/v0.5.0