Handle voice interruptions with Pipecat and Rasa

A caller interrupts before the assistant finishes answering. Pipecat handles the speech interruption, while Rasa supplies the dialogue. This tutorial connects them and tests how to discard replies that arrive after an interruption.

Our test interrupts the pipeline while a dialogue request is still running. Remove one check from the adapter, and the old reply enters the pipeline after the interruption.

The native control holds the dialogue reply until an actual InterruptionFrame reaches the sink. It then completes the request. Both runs use the same native pipeline machinery:

The replies are scripted and the sink records native text frames. Assess microphone input, synthesis, playback and acoustic interruption using your own speech services and recordings. This exercise isolates the late response handoff that sits before those services.

Keep the speech pipeline you already own

Juniper Mobile is fictional. Its public task asks which area the caller means, north or central, then explains a synthetic maintenance record. The supplied Rasa skill collects the area and calls a public lookup tool. Your existing Pipecat bot keeps its transport, speech recognition, speech synthesis and turn-aggregation pair.

The composition function places a Rasa dialogue processor between the user aggregator and the existing TTS service. It consumes the latest committed user text and emits native response-start, text and response-end frames. There is one speech path.

The Rasa project implements public lookup. Its accepted records carry JM-N-101 or JM-C-102. The decorated tool controls check both records, an unknown area and tool-shaped area text. Account requests need their own authorised domain tools before they can become part of this task.

A pending request outlives the current processing queue

Pipecat’s interruption guide says each processor “discards queued DataFrames and ControlFrames”. Our control adds a callback that completes after that native interruption. The adapter starts its dialogue request as a separately managed task. That request can still finish after the interruption has passed through the pipeline.

The controlled request passes through four ordered phases:

1. Request accepted. A committed context starts the controlled dialogue request. The reply waits on a future.
2. Turn interrupted. The native interruption reaches the sink. The application advances its response generation.
3. Old request completes. The future returns Synthetic old answer.. The generation comparison decides whether that text can enter the pipeline.
4. Current turn continues. A subsequent accepted request runs in order; its checked reply uses the existing speech path.

The ownership method has one comparison that the mutation removes:

def _owns_response(self, generation: int) -> bool:
    return generation == self._generation and not self._closing

The response holds the generation captured at dispatch. An interruption, a newer committed context or shutdown changes whether it can emit text.

The mutation retains the shutdown check. It removes only generation == self._generation and from the actual method source. The native control then records old text after the same interruption.

The processor keeps accepted dialogue requests in order with an async lock. Interrupting speech changes reply ownership; the accepted request can finish before the new one starts. A superseded request still waiting for that lock is skipped before dispatch. The ordering and lifecycle controls check both paths, mutable context capture, a rejected new context and shutdown.

Send committed turns to the existing conversation

The REST client owns a configured sender for its connection. Caller text stays in message; it cannot choose that sender. The actual loopback HTTP and pipeline controls inspect the received request and the resulting native text frame.

The processor skips speculative context before starting a dialogue request. It accepts only a final plain-text user message. Empty, oversized, multimodal and non-user-ending contexts are rejected. The default supplied aggregation pair disables empty-turn recovery; a retained pair can keep its own settings. A developer recovery prompt alone cannot dispatch another Rasa user message. Show repeat/retry state through your interface when no usable transcript arrives.

An unsuccessful request produces a generic native ErrorFrame with its raw exception withheld. Use the worker’s on_pipeline_error handler for interface recovery. The failure control checks the error signal and the absence of answer text together.

Run the native control before connecting speech

Use Python 3.12. From a fresh checkout, create only the selected Pipecat environment:

git clone https://github.com/RasaHQ/rasa-community-resources.git
cd rasa-community-resources
git fetch origin 0f8b765d4b66e7c407fccc58c689efa1de9a259f
git checkout 0f8b765d4b66e7c407fccc58c689efa1de9a259f
python3 scripts/workspace.py start tutorials/frameworks-with-rasa/pipecat --owner juniper-pipeline-demo
cd tutorials/frameworks-with-rasa/pipecat
make test
make demo

The recorded scripted native suite completes successfully:

Ran 17 tests in 0.217s

OK

The demo prints two JSON records. Their native interruption and request count match; their response ownership and emitted text differ.

.venv/bin/python - <<'PYCODE'
import asyncio
from rasa_dialogue.native_fixture import observe_late_response
async def check():
    guarded = await observe_late_response()
    mutated = await observe_late_response(remove_ownership_guard=True)
    assert guarded["native_interruption_reached_sink"]
    assert mutated["native_interruption_reached_sink"]
    assert guarded["response_text_after_interruption"] == []
    assert mutated["response_text_after_interruption"] == ["Synthetic old answer."]
    print(guarded)
    print(mutated)
asyncio.run(check())
PYCODE

If Mantle is new to you, start with the Rasa quickstart. Keep a suitable existing REST conversation if you already own one. Otherwise create only the supplied Rasa project’s separate locked environment, from the repo root:

python3 scripts/workspace.py start tutorials/frameworks-with-rasa/pipecat-mantle --owner juniper-dialogue-demo
cd tutorials/frameworks-with-rasa/pipecat-mantle
make test
make validate

Its five actual decorated tool controls pass. Project validation, licensed training and cold archive loading are recorded in the qualification. The native controls make zero provider model requests. Generated conversation wording and audio acceptance have their own checks.

Export your RASA_PRO_LICENSE and OPENAI_API_KEY using the blank .env.example as the variable list, then make train and make run. The local REST server uses port 5020. In another terminal, from the Pipecat project:

.venv/bin/python run_text_demo.py --message 'Is maintenance planned for north?'
.venv/bin/python run_text_demo.py --message 'And for central?'

Inspect the replies against JM-N-101 and JM-C-102, then test missing-area and retry behaviour. The command sends native text frames and prints the returned text plus operational errors. It uses your configured dialogue model.

For your voice bot, retain its component creation and transport event handlers, then replace its dialogue stage using the complete composition example. These are the adaptation files, relative to the Pipecat project:

Pass the existing transport, STT, TTS and aggregation pair to with_rasa, alongside the connection’s REST client. The native composition controls verify that those objects appear once in the pipeline. Keep actual acoustic recordings and interface recovery evidence when accepting the resulting voice configuration.

Stop your own bot/server and finish or cancel its requests. Keep needed state, models and recordings, then retire only the environments you created using the companion retirement commands. Source and locks remain available.