nemo_voice_agent.pipecat.services.nemo.builders

Small composable builders for the pipecat services a voice-agent bot uses.

These are thin wrappers around the existing service constructors so bot scripts can skip the repeated boilerplate of reading ConfigManager properties. Each builder is independent: a bot imports only what it needs. Novel services that aren’t covered here can still be constructed inline.

Module Contents

Functions

Name	Description
`build_audio_logger`	Build an AudioLogger if `transport.record_audio_data` is enabled.
`build_context_and_aggregators`	Build the `LLMContext` and its user/assistant aggregators.
`build_diar`	Build the diarization service, or return `None` if `diar.enabled` is False.
`build_llm`	Build the LLM service via `get_llm_service_from_config`.
`build_llm_text_processor`	Build the processor that segments LLM text into TTS-sized chunks.
`build_stt`	Build the NeMo STT service from config.
`build_tts`	Build the TTS service via `get_tts_service_from_config`.
`build_turn_taking`	Build the turn-taking service. `use_diar` defaults to `config_manager.USE_DIAR`.
`build_vad_analyzer`	Build the Silero VAD analyzer at the transport’s input sample rate.
`build_vad_processor`	Wrap `vad_analyzer` in the pipeline processor that emits VAD frames.
`build_ws_transport`	Build the no-timeout websocket server transport used by all bots.
`overwrite_existing_log`	Whether to delete (True) or rename (False) a pre-existing log file on startup.
`resolve_log_file_path`	Read the `server.{log_file,log_level,create_new_log,overwrite_existing_log}` block.

Data

__all__

API

nemo_voice_agent.pipecat.services.nemo.builders.build_audio_logger(
    config_manager: nemo_voice_agent.utils.ConfigManager
) -> typing.Optional[nemo_voice_agent.pipecat.services.nemo.audio_logger.AudioLogger]

Build an AudioLogger if transport.record_audio_data is enabled.

nemo_voice_agent.pipecat.services.nemo.builders.build_context_and_aggregators(
    llm: pipecat.services.openai.base_llm.BaseOpenAILLMService,
    config_manager: nemo_voice_agent.utils.ConfigManager,
    turn_taking: typing.Optional[nemo_voice_agent.pipecat.services.nemo.turn_taking.NeMoTurnTakingService] = None
)

Build the LLMContext and its user/assistant aggregators.

Returns (context, user_aggregator, assistant_aggregator, original_messages). original_messages is a fresh deep-copy of the initial message list, safe to hand to the reset/update-prompt RTVI handler factories.

turn_taking is the service from build_turn_taking (or None when turn-taking is disabled). Pass it. In pipecat 1.0+ the user aggregator owns turn detection, so exactly one component may emit UserStartedSpeakingFrame / UserStoppedSpeakingFrame, and this argument is what decides which:

turn_taking given: NeMoTurnTakingService pushes those frames itself, so we select ExternalUserTurnStrategies — the supported way to tell the aggregator an upstream processor owns turn detection. It sets enable_user_speaking_frames=False on both the start and stop strategy, so the aggregator stays quiet and there is no double emission. This replaces the old transport.can_create_user_frames=False knob, which pipecat removed along with transport-side VAD.
turn_taking is None (the *_nvidia.yaml configs, which set turn_taking.enabled: false): nothing upstream emits user-turn frames, so drive the turn from VAD directly and let the aggregator emit them. These configs used to rely on the transport’s VAD for this, which is what their now-obsolete can_create_user_frames: true was for. Note we name the stop strategy explicitly rather than taking pipecat’s default, which would pull in LocalSmartTurnAnalyzerV3.

Either way the strategies read the VADUserStartedSpeakingFrame / VADUserStoppedSpeakingFrame that build_vad_processor emits right after transport.input() — neither branch needs its own analyzer.

Omitting turn_taking falls back to re-deriving the answer from turn_taking.enabled, which is correct for the stock builders but silently wrong for a bot that constructs its turn-taking service inline.

nemo_voice_agent.pipecat.services.nemo.builders.build_diar(
    config_manager: nemo_voice_agent.utils.ConfigManager,
    audio_logger: typing.Optional[nemo_voice_agent.pipecat.services.nemo.audio_logger.AudioLogger] = None
) -> typing.Optional[nemo_voice_agent.pipecat.services.nemo.diar.NemoDiarService]

Build the diarization service, or return None if diar.enabled is False.

nemo_voice_agent.pipecat.services.nemo.builders.build_llm(
    config_manager: nemo_voice_agent.utils.ConfigManager
) -> pipecat.services.llm_service.LLMService

Build the LLM service via get_llm_service_from_config.

nemo_voice_agent.pipecat.services.nemo.builders.build_llm_text_processor(
    config_manager: nemo_voice_agent.utils.ConfigManager
) -> typing.Optional[pipecat.processors.aggregators.llm_text_processor.LLMTextProcessor]

Build the processor that segments LLM text into TTS-sized chunks.

Pipecat 1.0 removed TTSService(text_aggregator=...); aggregation now belongs to an LLMTextProcessor sitting immediately upstream of the TTS service. Insert the result there — pipecat silently ignores unknown constructor kwargs, so passing the aggregator to the service instead would quietly fall back to plain sentence splitting with no error to notice.

Returns None when tts.use_text_aggregator is False.

nemo_voice_agent.pipecat.services.nemo.builders.build_stt(
    config_manager: nemo_voice_agent.utils.ConfigManager,
    audio_logger: typing.Optional[nemo_voice_agent.pipecat.services.nemo.audio_logger.AudioLogger] = None
) -> pipecat.services.stt_service.STTService

Build the NeMo STT service from config.

nemo_voice_agent.pipecat.services.nemo.builders.build_tts(
    config_manager: nemo_voice_agent.utils.ConfigManager,
    audio_logger: typing.Optional[nemo_voice_agent.pipecat.services.nemo.audio_logger.AudioLogger] = None
) -> pipecat.services.tts_service.TTSService

Build the TTS service via get_tts_service_from_config.

nemo_voice_agent.pipecat.services.nemo.builders.build_turn_taking(
    config_manager: nemo_voice_agent.utils.ConfigManager,
    audio_logger: typing.Optional[nemo_voice_agent.pipecat.services.nemo.audio_logger.AudioLogger] = None,
    use_diar: typing.Optional[bool] = None,
    use_vad: bool = True
) -> nemo_voice_agent.pipecat.services.nemo.turn_taking.NeMoTurnTakingService

Build the turn-taking service. use_diar defaults to config_manager.USE_DIAR.

nemo_voice_agent.pipecat.services.nemo.builders.build_vad_analyzer(
    config_manager: nemo_voice_agent.utils.ConfigManager
) -> pipecat.audio.vad.silero.SileroVADAnalyzer

Build the Silero VAD analyzer at the transport’s input sample rate.

nemo_voice_agent.pipecat.services.nemo.builders.build_vad_processor(
    vad_analyzer: pipecat.audio.vad.silero.SileroVADAnalyzer | None
) -> typing.Optional[pipecat.processors.audio.vad_processor.VADProcessor]

Wrap vad_analyzer in the pipeline processor that emits VAD frames.

Pipecat 1.0 removed vad_analyzer from TransportParams — VAD is no longer run by the input transport. Placing a VADProcessor immediately after transport.input() restores the old frame ordering, so NeMoTurnTakingService keeps receiving VADUserStartedSpeakingFrame / VADUserStoppedSpeakingFrame at exactly the point it used to.

Returns None when there is no analyzer, so callers can drop it from the pipeline list with the same if x is not None pattern used elsewhere.

nemo_voice_agent.pipecat.services.nemo.builders.build_ws_transport(
    config_manager: nemo_voice_agent.utils.ConfigManager,
    vad_analyzer: pipecat.audio.vad.silero.SileroVADAnalyzer | None,
    host: str,
    port: int
) -> pipecat.transports.websocket.server.SingleClientWebsocketServerTransport

Build the no-timeout websocket server transport used by all bots.

vad_analyzer is accepted but unused: since pipecat 1.0 the transport no longer runs VAD. Pass the analyzer to build_vad_processor and insert the result right after transport.input() instead.

nemo_voice_agent.pipecat.services.nemo.builders.overwrite_existing_log(
    config_manager: nemo_voice_agent.utils.ConfigManager
) -> bool

Whether to delete (True) or rename (False) a pre-existing log file on startup.

nemo_voice_agent.pipecat.services.nemo.builders.resolve_log_file_path(
    config_manager: nemo_voice_agent.utils.ConfigManager,
    default_name: str = 'bot_server.log'
) -> tuple[str, str, bool]

Read the server.{log_file,log_level,create_new_log,overwrite_existing_log} block.

Returns (log_file, log_level, create_new_log). Callers pair this with setup_rotating_log from nemo_voice_agent.utils.misc to handle the rename-existing-log dance.

nemo_voice_agent.pipecat.services.nemo.builders.__all__ = ['build_audio_logger', 'build_vad_analyzer', 'build_vad_processor', 'build_ws_tr...