Custom Frame Processors

A Pipecat FrameProcessor is a single node in the bot pipeline: it receives frames, optionally transforms or absorbs them, and forwards the rest. Writing one is the lowest-scope way to add behavior that NeMo Labs Voice Agent’s builders do not provide. Examples include an automatic speech recognition (ASR) post-corrector, a Markdown sanitizer before text-to-speech (TTS), and transcript redaction.

Prerequisites

Before you add a processor, complete the following preparation:

1. Run the Quickstart with the shipped pipeline.
2. Identify the frame type you need to transform and the pipeline stage that emits it.
3. Choose the demo server or evaluation bot entrypoint where you want to insert the processor.

Processor or Builder Swap?

Choose a processor for frame transformations and a builder change for service construction.

A processor is the right tool when the stages themselves are fine and you only need to touch the frames flowing between them. It requires no changes to nemo_voice_agent/, only a few lines in your copy of the server entrypoint.

Where the Pipeline Is Assembled

Both entrypoints — examples/generic_voice_agent/server/server.py and evaluation/bot_server.py — build the pipeline in run_bot_websocket() as a flat list, appending optional stages only when their builder returned a value:

1
pipeline_list = [ws_transport.input()]
2
if vad_processor is not None:
3
    pipeline_list.append(vad_processor)
4
pipeline_list.extend([rtvi, stt])
5
if diar is not None:
6
    pipeline_list.append(diar)
7
if turn_taking is not None:
8
    pipeline_list.append(turn_taking)
9
if user_audio_buffer is not None:
10
    pipeline_list.append(user_audio_buffer)
11
pipeline_list.extend([user_agg, llm])
12
if llm_text_processor is not None:
13
    pipeline_list.append(llm_text_processor)
14
pipeline_list.extend([tts, ws_transport.output(), assistant_agg])
15
pipeline = Pipeline(pipeline_list)

Insert your processor by adding one more element to that list. Refer to How It Works for what each stage does.

Insertion Points

Pick the position by the frame type you need to see. Frame classes come from pipecat.frames.frames unless noted.

Two things decide between the last two text positions. LLMTextProcessor converts LLMTextFrame into sentence-sized AggregatedTextFrame, as implemented by build_llm_text_processor in nemo_voice_agent/pipecat/services/nemo/builders.py. Any regex that must match across token boundaries belongs after it. Note that it is only present when tts.use_text_aggregator is true — the default. When it is false, the TTS service does its own aggregation internally and only LLMTextFrame reaches that point.

Rules

Follow these rules so unrelated frames and pipeline direction continue to work as designed:

• Subclass FrameProcessor and override async def process_frame(self, frame, direction).
• Call await super().process_frame(frame, direction) first. The base implementation handles StartFrame, CancelFrame, InterruptionFrame, and pause/resume bookkeeping. It does not forward anything.
• Always forward with await self.push_frame(frame, direction) unless you intentionally consume the frame. Dropping a control frame stalls everything downstream.
• Filter on frame type. TranscriptionFrame, InterimTranscriptionFrame, LLMTextFrame, and AggregatedTextFrame are all subclasses of TextFrame, so an isinstance(frame, TextFrame) test matches user speech as well as bot speech. Match the narrowest class you mean.
• Check direction. FrameDirection.DOWNSTREAM runs input toward output. FrameDirection.UPSTREAM carries errors and control signals back. Transforms should almost always guard on DOWNSTREAM and pass upstream frames through untouched.
• Frames are mutable dataclasses, so in-place edits such as frame.text = ... work and preserve the subclass and its metadata fields.
• System frames jump the queue. InputAudioRawFrame and other SystemFrame subclasses are dispatched ahead of queued data frames, so do not assume ordering between audio and transcripts.

Example: Strip Markdown Before TTS

LLMs emit **bold**, # heading, and - bullet markup even when the system prompt forbids it, and the TTS voice reads the punctuation aloud. A sanitizer placed immediately before tts fixes this deterministically instead of relying on model compliance.

1
import re
2
3
from pipecat.frames.frames import AggregatedTextFrame
4
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
5
6
7
class MarkdownStripper(FrameProcessor):
8
    """Remove Markdown formatting from sentences on their way to TTS.
9
10
    Identifier-like tokens ('#W2378156' order IDs, '*123' extensions) survive
11
    because only formatting patterns are matched, never bare characters.
12
    """
13
14
    _BOLD = re.compile(r"\*\*([^*]+?)\*\*")
15
    _EMPH = re.compile(r"(?<!\w)\*([^*\s][^*]*?)\*(?!\w)")
16
    _UNDERLINE = re.compile(r"(?<!\w)_([^_\s][^_]*?)_(?!\w)")
17
    _HEADING = re.compile(r"^#+\s+", flags=re.MULTILINE)
18
    _LIST = re.compile(r"^[-*]\s+", flags=re.MULTILINE)
19
    _NUMBERED = re.compile(r"^\d+\.\s+", flags=re.MULTILINE)
20
21
    @classmethod
22
    def sanitize(cls, text: str) -> str:
23
        text = cls._BOLD.sub(r"\1", text)
24
        text = cls._EMPH.sub(r"\1", text)
25
        text = cls._UNDERLINE.sub(r"\1", text)
26
        text = cls._HEADING.sub("", text)
27
        text = cls._LIST.sub("", text)
28
        text = cls._NUMBERED.sub("", text)
29
        return text
30
31
    async def process_frame(self, frame, direction: FrameDirection):
32
        await super().process_frame(frame, direction)
33
        if isinstance(frame, AggregatedTextFrame) and direction == FrameDirection.DOWNSTREAM:
34
            frame.text = self.sanitize(frame.text)
35
        await self.push_frame(frame, direction)

Wire it into your copy of the entrypoint:

1
from your_package.processors import MarkdownStripper
2
3
markdown_stripper = MarkdownStripper()
4
5
pipeline_list.extend([user_agg, llm])
6
if llm_text_processor is not None:
7
    pipeline_list.append(llm_text_processor)
8
pipeline_list.append(markdown_stripper)  # after aggregation, before TTS
9
pipeline_list.extend([tts, ws_transport.output(), assistant_agg])

Side Effects to Watch

A processor can affect downstream context, timing, and logging even when it changes only one frame type:

• The assistant context sees your edits. assistant_agg sits at the end of the pipeline and builds the assistant turn from text frames whose append_to_context is true. These are the same objects your processor already mutated. For Markdown stripping that is desirable. For a change you want spoken but not remembered, emit a modified copy instead of editing in place.
• Stateful processors need a reset hook. The resettable list passed to the RTVI actions is iterated by _reset_services in nemo_voice_agent/pipecat/processors/frameworks/rtvi_actions.py. The function calls a plain synchronous reset() on any entry that has one and skips None. Add your processor to that list and give it a reset() method if it carries per-conversation state. Refer to RTVI Control Plane.
• Configuration belongs in YAML. Accept options in __init__ and read them from a section of the server config, the way the existing builders do, rather than hardcoding them.
• Reasoning spans are handled elsewhere. TTS already skips text between tts.think_tokens. Do not reimplement that in a processor. Refer to Reasoning Mode.

Test It

Processors are unit-testable without a pipeline: keep the transform in a pure classmethod, then drive process_frame directly with a stubbed push_frame to check forwarding and frame typing. tests/unit/test_runtime_state_machines.py uses exactly this pattern for UserAudioBuffer.

$
uv run pytest tests/unit -m "not gpu"

Then run the bot end to end and listen. Refer to Quickstart. If your processor does not process frames, confirm it is in pipeline_list because optional surrounding stages are appended conditionally. Also confirm that the selected frame class reaches that position.

Next Steps

Continue with the guide for the extension boundary you need:

• The Builder API — swap how a stage is constructed instead of what flows between stages.
• Building Your Own Pipeline — replace run_bot_websocket() entirely.
• Extension Overview — how the extension points fit together.