How the Pipeline Builder Constructs the Complete Voice Processing Flow in Dograh

The pipeline builder in api/services/pipecat/pipeline_builder.py assembles an ordered list of processors into a single Pipecat Pipeline instance that handles audio input, optional speech-to-text, LLM inference, text-to-speech, and output, all wrapped in a PipelineTask for asynchronous execution.

Dograh is an open-source voice AI platform that orchestrates real-time conversations through a modular pipeline architecture. The pipeline builder constructs the complete voice processing flow by chaining specialized processors into a linear data flow that transforms inbound audio into intelligent responses. This construction happens in api/services/pipecat/pipeline_builder.py, which serves as the central factory for assembling Pipecat-based pipelines.

Constructing the Complete Voice Processing Flow

The builder constructs the pipeline through three logical phases, ensuring each component executes in the correct order to maintain conversational state and audio quality.

Step 1: Create Shared Components

The construction begins with create_pipeline_components, which instantiates reusable infrastructure that persists throughout the call (lines 13-25). This function builds the AudioBufferProcessor (responsible for recording both inbound and outbound audio) and the LLMContext aggregator that maintains conversation state across turns.

audio_buffer, context = create_pipeline_components(audio_config)

Step 2: Select the Proper Pipeline Shape

The builder supports two distinct flow architectures based on the LLM type. For non-realtime processing, build_pipeline creates a transcription-based sequence (lines 41-94). For realtime (speech-to-speech) models, build_realtime_pipeline omits STT/TTS and injects the realtime LLM directly after transport input (lines 97-152).

The non-realtime processor sequence follows this strict order:

processors = [
    transport.input(),              # user audio → transport

    stt,                            # speech-to-text

    user_context_aggregator,        # aggregates user utterances

    llm,                            # main LLM

    *post_llm,                      # callbacks, optional recording router

    tts,                            # text-to-speech

    transport.output(),             # bot audio → transport

    audio_buffer,                   # record everything

    assistant_context_aggregator,   # aggregate bot responses

    pipeline_metrics_aggregator,    # metrics collection

]
pipeline = Pipeline(processors)

Step 3: Wrap in a PipelineTask

Finally, create_pipeline_task encapsulates the assembled pipeline in a PipelineTask with configurable parameters (lines 55-81). This enables tracing, metrics collection, and conversation ID tracking.

task = PipelineTask(
    pipeline,
    params=pipeline_params,
    enable_tracing=True,
    enable_rtvi=False,
    conversation_id=f"{workflow_run_id}",
)

Processor Sequence and Data Flow

The complete voice processing flow follows a deterministic linear path where each processor handles specific frame transformations:

• Transport input (transport.input()): Captures raw audio from telephony or WebRTC sources at the pipeline entry point.
• STT (non-realtime only): Converts speech to text for LLM consumption, positioned immediately after transport input.
• Voicemail detector (optional): Inserts voicemail_detector.detector() and voicemail_detector.llm_gate() between STT and the user context aggregator to classify answering machines without blocking TTS frames.
• User context aggregator: Buffers user utterances into structured LLM context before inference.
• LLM: Generates conversational responses in non-realtime mode, or produces audio directly in realtime mode.
• Post-LLM callbacks: Includes pipeline_engine_callback_processor from api/services/pipecat/pipeline_engine_callbacks_processor.py for call-duration limits and optional recording_router for pre-recorded audio playback.
• TTS (non-realtime): Synthesizes bot speech from LLM text output before transport delivery.
• Transport output (transport.output()): Streams audio back to the caller.
• Audio buffer: Persists full-duplex audio for later retrieval, positioned to capture both input and output streams.
• Assistant aggregator: Captures bot responses for context compaction and historical tracking.
• Metrics aggregator (pipeline_metrics_aggregator): Emits usage and performance data to Dograh's observability stack, defined in api/services/pipecat/pipeline_metrics_aggregator.py.

Integration with the Execution Engine

The builder is invoked from api/services/pipecat/run_pipeline.py within the _run_pipeline helper (lines 90-115). After instantiating services (STT, LLM, TTS, voicemail detector), the code selects the appropriate builder function based on the is_realtime flag (lines 164-176).

if is_realtime:
    pipeline = build_realtime_pipeline(...)
else:
    pipeline = build_pipeline(...)

task = create_pipeline_task(pipeline, workflow_run_id, audio_config)

This assembled PipelineTask is then bound to the PipecatEngine and executed via await task.run(params), initiating the asynchronous frame processing that drives the conversation.

Entry Points for Telephony and WebRTC

Consumers trigger the pipeline through provider-specific entry points that eventually converge on the builder. The telephony handler initializes transport for providers like Twilio (lines 26-42):

await run_pipeline_telephony(
    websocket,
    provider_name="twilio",
    workflow_id=42,
    workflow_run_id=12345,
    user_id=7,
    call_id="CAabcd1234",
    transport_kwargs={"stream_sid": "MS123"},
)

For browser-based WebRTC connections (lines 24-33):

await run_pipeline_smallwebrtc(
    webrtc_connection,
    workflow_id=42,
    workflow_run_id=12345,
    user_id=7,
    call_context_vars={},
)

Both paths converge on _run_pipeline, which orchestrates the builder steps and executes the final task.

Summary

• The pipeline builder in api/services/pipecat/pipeline_builder.py constructs the complete voice processing flow through three logical steps: component creation, processor assembly, and task wrapping.
• Non-realtime pipelines follow STT → LLM → TTS ordering, while realtime pipelines inject speech-to-speech LLMs directly after transport input to minimize latency.
• Shared components like AudioBufferProcessor and context aggregators ensure consistent state management and recording capabilities across both architectures.
• The builder supports optional voicemail detection via processor injection without disrupting the core data flow or blocking audio output.
• Final execution occurs through PipelineTask with integrated metrics, tracing, and conversation tracking enabled.

Frequently Asked Questions

What is the difference between build_pipeline and build_realtime_pipeline?

build_pipeline constructs a standard transcription-based flow including STT and TTS processors, suitable for traditional text-based LLM interactions. build_realtime_pipeline creates a speech-to-speech flow that omits STT/TTS entirely, placing the realtime LLM directly after transport input to minimize latency for voice-native models.

How does the pipeline handle voicemail detection without blocking responses?

The builder optionally inserts voicemail_detector.detector() and voicemail_detector.llm_gate() between the STT and user context aggregator. This allows classification of answering machines while permitting TTS frames to flow through, preventing the main LLM from processing voicemail beeps as user input.

Where does audio recording occur in the pipeline flow?

The AudioBufferProcessor (created in create_pipeline_components) is appended near the end of the processor list, after transport.output(). This positioning allows it to capture both inbound user audio and outbound bot audio in a single buffer for post-call retrieval and analysis.

Which file orchestrates the actual execution of the built pipeline?

api/services/pipecat/run_pipeline.py contains the _run_pipeline helper that instantiates all services, calls the appropriate builder function based on the realtime flag, wraps the result in a PipelineTask, and executes the flow via await task.run(params).

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}