Problem with using Flows in a bot with a parallel pipeline and multiple LLMs

[chadbailey59]

I'm trying to use a prerelease version of our new parallel voicemail detection architecture, but also use Pipecat Flows in the human side of the conversation. My pipeline looks like this:

pipeline = Pipeline(
        [
            transport.input(),
            rtvi,
            ParallelPipeline(
                # Voicemail detection branch
                [
                    detection_context_aggregator.user(),
                    detection_llm,
                    # fl0,
                    voicemail_tts,
                    FunctionFilter(voicemail_filter),
                ],
                # Human conversation branch
                [
                    context_aggregator.user(),
                    # bot_mute_processor,
                    llm,
                    human_gate,
                    tts,
                    FunctionFilter(human_filter),
                    context_aggregator.assistant(),
                ],
            ),
            transport.output(),
        ]
    )

If I run flow_manager.initialize() early in the call lifecycle, the FlowManager will update the context of both llm (the human LLM) and detection_llm (the voicemail detection LLM). Here's Claude's explanation of why, and a fix that worked for me:

When the FlowManager initializes a new node with flow_manager.initialize(create_initial_node()), it calls _set_node which then calls _update_llm_context. This method sends frames to the task queue:

await self.task.queue_frames(
    [frame_type(messages=messages), LLMSetToolsFrame(tools=functions)]
)

The key issue is that these frames (LLMMessagesUpdateFrame/LLMMessagesAppendFrame and LLMSetToolsFrame) are being sent to the shared task queue, but they're not specifically targeted to the context manager that was provided to the FlowManager.

In your bot.py, both context managers (the main one and the detection one) are connected to the same task pipeline. When the FlowManager sends these frames to the task, they're processed by all context managers in the pipeline, not just the one that was provided to the FlowManager.

Here's how to fix the issue: Add a filter processor that only allows specific frames to reach the detection context manager.
Here's a suggested fix for your bot.py:

# Create a filter function that blocks LLMMessagesUpdateFrame, LLMMessagesAppendFrame, and LLMSetToolsFrame
# from reaching the detection context manager
async def flow_manager_frame_filter(frame):
    from pipecat.frames.frames import LLMMessagesUpdateFrame, LLMMessagesAppendFrame, LLMSetToolsFrame
    # Return False (block) for frames sent by FlowManager
    return not isinstance(frame, (LLMMessagesUpdateFrame, LLMMessagesAppendFrame, LLMSetToolsFrame))

# Then in your pipeline setup, add this filter before the detection_context_aggregator
pipeline = Pipeline(
    [
        transport.input(),
        rtvi,
        # stt,
        ParallelPipeline(
            # Voicemail detection branch
            [
                # audio_collector,
                FunctionFilter(flow_manager_frame_filter),  # Add this filter
                detection_context_aggregator.user(),
                detection_llm,
                # fl0,
                voicemail_tts,
                FunctionFilter(voicemail_filter),
            ],
            # Human conversation branch
            [
                context_aggregator.user(),
                # bot_mute_processor,
                llm,
                human_gate,
                tts,
                FunctionFilter(human_filter),
                context_aggregator.assistant(),
            ],
        ),
        transport.output(),
    ]
)

This filter will prevent the frames sent by the FlowManager from affecting the detection context manager, while still allowing them to reach the main context manager.

The issue occurs because the FlowManager is designed to work with a single context manager and doesn't have a way to target specific context managers when sending frames. By adding a filter, you can ensure that the frames only affect the intended context manager.

[kompfner]

I see. In this case, you have a "side" pipeline that you don't really want Pipecat Flows managing at all.

Filtering out LLMMessagesUpdateFrame, LLMMessagesAppendFrame and LLMSetToolsFrame seems like it'd mostly work—as far as I can tell none of these frame types are really used by Pipecat "internally" and just used as an API for programmatically sending messages into the pipeline—but is a bit of a blunt instrument, and probably doesn't generalize too well.

Some downsides I can think of:

• You might want to use those frames someday to programmatically control your "side" pipeline
• Pipecat Flows also sends other frames than just those in the list, if you use things like node actions

We need to give this more thought, but one random thing that popped into my head is a way to tell Pipecat Flows what its "entrypoint" to the pipeline should be, so it could push_frames from there instead of from the top of the pipeline...? 🤔 So you could set it to the processor at the top of the pipeline "branch" that you want to affect. (I'm sure we'd hit something weird with this approach, but just wanted to jot it down).

[markbackman]

@chadbailey59 is this an issue or can I close it out?