Building Your Own Pipeline

This is the deepest extension tier in NeMo Labs Voice Agent: you write your own run_bot_websocket() instead of reusing the shipped one. Every service, processor, and pipeline stage becomes yours. Only the transport and real-time voice inference (RTVI) control-plane contracts must remain compatible.

Use this approach only after the lower-scope tiers are exhausted. Swapping a model behind an existing stage is YAML-only (Server Configuration). A transformation between two stages requires one class (Custom Frame Processors). This tier supports a different pipeline shape or services that the builders do not cover.

Prerequisites

Before you build a replacement pipeline, complete the following preparation:

1. Run the Quickstart with the shipped pipeline.
2. Choose whether the replacement must serve the browser client, the evaluation harness, or both.
3. Identify the stage ordering or service requirement that the shipped builders cannot express.

Two Entry Points, Two Contracts

There are two run_bot_websocket() implementations, and they require different behavior from a replacement:

Both call the shared runner run_bot_websocket_server() from nemo_voice_agent/pipecat/bot_server.py, which owns the boilerplate around the task: transport connect/disconnect handlers, the RTVI on_client_ready kickoff, audio-logger finalization, and shutdown. It makes no assumptions about pipeline contents, so a custom pipeline can keep using it.

The Contract

A replacement bot must satisfy exactly two requirements.

1. Speak Pipecat’s WebSocket server protocol. Use SingleClientWebsocketServerTransport from pipecat.transports.websocket.server with a ProtobufFrameSerializer, bound to the port in WEBSOCKET_PORT. build_ws_transport() in nemo_voice_agent/pipecat/services/nemo/builders.py constructs exactly this and is the path of least resistance — it also reads the transport sample rates from your config. Since Pipecat 1.0 the transport no longer runs VAD, so a VAD processor goes in the pipeline right after transport.input().

2. Carry an RTVIProcessor with the handlers your consumer expects. Build the handlers with the factories in nemo_voice_agent/pipecat/processors/frameworks/rtvi_actions.py and install them in one shot with register_client_message_handlers(). Each factory returns a (message_type, handler) pair. The registrar installs a single on_client_message dispatcher over all of them. An unhandled type produces an error-response rather than a silent hang.

The six factories, keyed by the wire message type they answer:

Refer to RTVI Control Plane for handler semantics and RTVI Messages for payload shapes.

What You May Not Change

Preserve the following contracts if the browser client or evaluation harness must connect to the new pipeline:

• The wire protocol. Use a Pipecat WebSocket server transport. Re-implementing protobuf framing by hand is out of scope.
• The six message types and their response shapes. The bridge sends and parses these literally. Renaming get_scenario_summary, or returning something other than the actions plus db_hash pair, silently breaks scoring. The run completes with unusable metrics.
• The tool-registry namespace key. The bridge passes scenario.domain as tool_domain. Your tool-registration callback must accept the same tool_factory(name, domain=...) interface that get_schema_tool_for_eval implements. Refer to Custom Tools.

Free-Choice Points

Everything outside the connection and RTVI contracts can be adapted to the agent’s requirements.

Pass your turn-taking service to build_context_and_aggregators() if you use it. In Pipecat 1.0+, the pipeline permits exactly one component to emit user-speaking frames, and that argument determines the component. Omitting it derives the answer again from turn_taking.enabled, which is incorrect for a bot that builds the service inline.

Skeleton

Mirrors the structure of evaluation/bot_server.py, trimmed to the required parts.

1
from pipecat.frames.frames import LLMRunFrame
2
from pipecat.pipeline.pipeline import Pipeline
3
from pipecat.pipeline.worker import PipelineParams, PipelineWorker
4
from pipecat.processors.frameworks.rtvi import RTVIProcessor
5
6
from nemo_voice_agent.evaluation.tools import get_schema_tool_for_eval
7
from nemo_voice_agent.pipecat.bot_server import run_bot_websocket_server
8
from nemo_voice_agent.pipecat.processors.frameworks.rtvi import RTVIObserver
9
from nemo_voice_agent.pipecat.processors.frameworks.rtvi_actions import (
10
    SharedStateRef, TaskRef, register_client_message_handlers,
11
    create_apply_initialization_action, create_apply_sync_delta_action,
12
    create_get_context_history_action, create_get_scenario_summary_action,
13
    create_reset_context_action, create_update_system_prompt_action,
14
)
15
from nemo_voice_agent.pipecat.services.nemo.builders import build_ws_transport
16
from nemo_voice_agent.utils import ConfigManager
17
from nemo_voice_agent.utils.tool_calling import register_schema_tools_to_llm
18
19
20
async def run_custom_bot(host: str, port: int):
21
    config_manager = ConfigManager(server_base_path=..., server_config_path=...)
22
23
    # 1. Your services: stock builders, your own classes, third-party plugins.
24
    vad_analyzer, stt, llm, tts = ..., ..., ..., ...
25
26
    # 2. Transport (required) — applies the protobuf serializer and config sample rates.
27
    ws_transport = build_ws_transport(config_manager, vad_analyzer, host, port)
28
29
    # 3. Context + aggregators. Substitute your own if you need a different shape.
30
    context, user_agg, assistant_agg, original_messages = ...
31
32
    # 4. RTVI processor + handlers (required).
33
    rtvi = RTVIProcessor()
34
    task_ref = TaskRef()
35
    shared_state_ref = SharedStateRef()
36
    resettable = [stt, tts]
37
    prompt_args = (task_ref, user_agg, assistant_agg, original_messages, resettable)
38
39
    register_client_message_handlers(
40
        rtvi,
41
        [
42
            create_reset_context_action(*prompt_args),
43
            create_update_system_prompt_action(
44
                *prompt_args,
45
                system_role=config_manager.SYSTEM_ROLE,
46
                system_prompt_suffix=config_manager.SYSTEM_PROMPT_SUFFIX,
47
                enable_tool_calling=True,
48
                llm=llm, context=context, rtvi=rtvi,
49
                tool_factory=get_schema_tool_for_eval,
50
                register_schema_tools=register_schema_tools_to_llm,
51
                shared_state_ref=shared_state_ref,
52
            ),
53
            create_get_context_history_action(task_ref, assistant_agg),
54
            create_get_scenario_summary_action(task_ref, shared_state_ref),
55
            create_apply_initialization_action(shared_state_ref),
56
            create_apply_sync_delta_action(shared_state_ref),
57
        ],
58
    )
59
60
    # 5. Pipeline. The shape is yours; only these positions are fixed.
61
    pipeline = Pipeline(
62
        [ws_transport.input(), rtvi, stt, user_agg, llm, tts, ws_transport.output(), assistant_agg]
63
    )
64
65
    task = PipelineWorker(
66
        pipeline,
67
        params=PipelineParams(enable_metrics=True, enable_usage_metrics=True, idle_timeout=None),
68
        observers=[RTVIObserver(rtvi)],
69
        idle_timeout_secs=None,
70
        cancel_on_idle_timeout=False,
71
    )
72
73
    # 6. Shared runner: transport handlers, client-ready kickoff, shutdown.
74
    await run_bot_websocket_server(
75
        task=task, ws_transport=ws_transport, rtvi=rtvi, task_ref=task_ref,
76
        talk_first=True, initial_frame_factory=LLMRunFrame,
77
        on_disconnect_reset_services=resettable,
78
    )

task_ref.task and task_ref.running are populated by run_bot_websocket_server(), so construct the TaskRef before the handlers and hand it over unpopulated. To keep the FastAPI /connect endpoint the browser client uses, wrap the coroutine with create_fastapi_app() and run_bot_with_fastapi() from the same module.

Verify Against the Eval Harness

The eval bridge is the strictest consumer of the contract, so run one scenario through it. All three commands start from evaluation/ — SERVER_CONFIG_PATH resolves against the current working directory, not the script directory.

$
cd evaluation
$
$
# Terminal 1 — your custom agent bot
$
WEBSOCKET_PORT=8765 SERVER_CONFIG_PATH=server_configs/agent.yaml python path/to/your_bot.py
$
# Terminal 2 — stock user-simulator bot
$
WEBSOCKET_PORT=8766 SERVER_CONFIG_PATH=server_configs/user.yaml python bot_server.py
$
# Terminal 3 — one short scenario
$
python run_evaluation.py --scenarios restaurant__pizza_pepperoni

Then read eval_results/eval_<timestamp>/restaurant__pizza_pepperoni/:

For the full run procedure and artifact reference, refer to Evaluation Quickstart and Reading Results.

Non-Pipecat Agents

Evaluating an agent that is not built on Pipecat means re-implementing the transport protocol and the RTVI control plane yourself. Treat pipecat.transports.websocket.server and pipecat.serializers.protobuf as the wire specification. Refer to External Agents for the harness-side view.

Next Steps

Continue with the implementation guide for the extension surface you need next:

• The Builder API explains how to reuse or replace component construction.
• Create a Custom Frame Processor shows how to add a focused transformation without replacing the full pipeline.
• RTVI Control Plane documents the handlers a compatible evaluation bot must register.