nemo_voice_agent.pipecat.services.nemo.tts

Module Contents

Classes

Name	Description
`BaseNemoTTSService`	Text-to-Speech service using Nemo TTS models.
`KokoroTTSService`	Text-to-Speech service using Kokoro-82M model.
`MagpieTTSService`	Text-to-Speech service using Magpie TTS model.
`NeMoFastPitchHiFiGANTTSService`	Text-to-Speech service using NeMo FastPitch-Hifigan model.

Functions

Name	Description
`build_text_aggregator`	Build the sentence aggregator for TTS, or `None` if disabled by config.
`get_tts_service_from_config`	Get the TTS service from the configuration.

API

class nemo_voice_agent.pipecat.services.nemo.tts.BaseNemoTTSService(
    model,
    device: str = 'cuda',
    sample_rate: int = 22050,
    think_tokens: typing.Optional[typing.List[str]] = None,
    audio_logger: typing.Optional[nemo_voice_agent.pipecat.services.nemo.audio_logger.AudioLogger] = None,
    ignore_strings: typing.Optional[typing.List[str]] = None,
    voice: typing.Optional[str] = None,
    language: typing.Optional[str] = None,
    kwargs = {}
)

Bases: TTSService, ToolCallingMixin

Text-to-Speech service using Nemo TTS models.

This service works with any TTS model that exposes a generate(text) method that returns audio data. The TTS generation runs in a dedicated background thread to avoid blocking the main asyncio event loop, following the same pattern as NemoDiarService.

Parameters:

model

TTS model instance with a generate(text) method

sample_rate

int" default="22050

Audio sample rate in Hz (defaults to 22050)

**kwargs

Defaults to {}

Additional arguments passed to TTSService

_ignore_strings

_model

= self._setup_model()

_pending_requests

= {}

_tts_queue

= asyncio.Queue()

nemo_voice_agent.pipecat.services.nemo.tts.BaseNemoTTSService._convert_to_bytes(
    audio_data
) -> bytes

Convert various audio data formats to bytes.

nemo_voice_agent.pipecat.services.nemo.tts.BaseNemoTTSService._drop_special_tokens(
    text: str
) -> typing.Optional[str]

Drop the special tokens from the text.

nemo_voice_agent.pipecat.services.nemo.tts.BaseNemoTTSService._generate_audio(
    text: str
) -> typing.Iterator[numpy.ndarray]

nemo_voice_agent.pipecat.services.nemo.tts.BaseNemoTTSService._get_response_queue(
    request_id: str
)

async

Get the response queue for a specific request.

nemo_voice_agent.pipecat.services.nemo.tts.BaseNemoTTSService._handle_think_tokens(
    text: str
) -> typing.Optional[str]

Handle the thinking tokens for TTS. If the thinking tokens are not provided, return the text as it is. Otherwise: If both thinking tokens appear in the text, return the text after the end of thinking tokens. If the LLM is thinking, return None. If the LLM is done thinking, return the text after the end of thinking tokens. If the LLM starts thinking, return the text before the start of thinking tokens. If the LLM is not thinking, return the text as is.

nemo_voice_agent.pipecat.services.nemo.tts.BaseNemoTTSService._initialize_tool_calling()

Initialize the tool calling mixin by registering all available tools.

nemo_voice_agent.pipecat.services.nemo.tts.BaseNemoTTSService._processing_task_handler()

async

Handler for background processing task.

nemo_voice_agent.pipecat.services.nemo.tts.BaseNemoTTSService._setup_model()

nemo_voice_agent.pipecat.services.nemo.tts.BaseNemoTTSService._stop_tasks()

async

Stop background processing tasks.

nemo_voice_agent.pipecat.services.nemo.tts.BaseNemoTTSService._tts_processor()

Background processor that handles TTS generation calls.

nemo_voice_agent.pipecat.services.nemo.tts.BaseNemoTTSService.can_generate_metrics() -> bool

If the TTS service can generate metrics.

nemo_voice_agent.pipecat.services.nemo.tts.BaseNemoTTSService.cancel(
    frame: pipecat.frames.frames.CancelFrame
)

async

Handle service cancellation.

nemo_voice_agent.pipecat.services.nemo.tts.BaseNemoTTSService.reset()

Reset the TTS service.

nemo_voice_agent.pipecat.services.nemo.tts.BaseNemoTTSService.run_tts(
    text: str,
    context_id: str
) -> collections.abc.AsyncGenerator[pipecat.frames.frames.Frame, None]

async

Generate speech from text using the Nemo TTS model.

nemo_voice_agent.pipecat.services.nemo.tts.BaseNemoTTSService.setup_tool_calling()

Setup the tool calling mixin by registering all available tools.

nemo_voice_agent.pipecat.services.nemo.tts.BaseNemoTTSService.start(
    frame: pipecat.frames.frames.StartFrame
)

async

Handle service start.

nemo_voice_agent.pipecat.services.nemo.tts.BaseNemoTTSService.stop(
    frame: pipecat.frames.frames.EndFrame
)

async

Handle service stop.

class nemo_voice_agent.pipecat.services.nemo.tts.KokoroTTSService(
    model: str = 'hexgrad/Kokoro-82M',
    lang_code: str = 'a',
    voice: str = 'af_heart',
    device: str = 'cuda',
    sample_rate: int = 24000,
    speed: float = 1.0,
    download_all: bool = True,
    cache_models: bool = True,
    kwargs = {}
)

Bases: BaseNemoTTSService

Text-to-Speech service using Kokoro-82M model.

Kokoro is an open-weight TTS model with 82 million parameters. More info: https://huggingface.co/hexgrad/Kokoro-82M

Parameters:

lang_code

str" default="'a'

Language code for the model (default: ‘a’ for American English)

voice

str" default="'af_heart'

Voice to use (default: ‘af_heart’)

device

str" default="'cuda'

Device to run on (default: ‘cuda’)

sample_rate

int" default="24000

Audio sample rate in Hz (default: 24000 for Kokoro)

download_all

bool" default="True

Download all models for different languages (default: True)

cache_models

bool" default="True

Cache models on GPU for faster switching between languages (default: True)

**kwargs

Defaults to {}

Additional arguments passed to BaseNemoTTSService

_gender

= 'female' if voice[1] == 'f' else 'male'

_model_maps

_original_gender

= self._gender

_original_lang_code

= self._lang_code

nemo_voice_agent.pipecat.services.nemo.tts.KokoroTTSService._attach_g2p_fallback(
    pipeline
) -> None

staticmethod

Attach the Apache-2.0 OOV fallback (see _g2p_fallback.py) to an English pipeline.

Without this, misaki’s dictionary-based G2P silently drops any word outside its lexicon (phonemes=None) since the GPL-licensed espeak fallback is excluded.

nemo_voice_agent.pipecat.services.nemo.tts.KokoroTTSService._download_all_models(
    lang_code: typing.List[str] = ['a', 'b'],
    device = 'cuda',
    repo_id = 'hexgrad/Kokoro-82M',
    cache_models = True
)

Download all models for Kokoro TTS service.

nemo_voice_agent.pipecat.services.nemo.tts.KokoroTTSService._generate_audio(
    text: str
) -> typing.Iterator[numpy.ndarray]

Generate audio using the Kokoro pipeline.

Parameters:

text

str

Text to convert to speech

nemo_voice_agent.pipecat.services.nemo.tts.KokoroTTSService._setup_model(
    lang_code: typing.Optional[str] = None,
    voice: typing.Optional[str] = None
)

Initialize the Kokoro pipeline.

nemo_voice_agent.pipecat.services.nemo.tts.KokoroTTSService.reset()

Reset the voice and speed to the original ones.

nemo_voice_agent.pipecat.services.nemo.tts.KokoroTTSService.setup_tool_calling()

Setup the tool calling mixin by registering all available tools.

nemo_voice_agent.pipecat.services.nemo.tts.KokoroTTSService.tool_tts_reset_speed(
    params: pipecat.services.llm_service.FunctionCallParams
)

async

Reset the speaking speed to the original speed.

Inform user of the result of this tool call. After calling this tool, continue the previous response if it was unfinished and was interrupted by the user, otherwise start a new response and ask if the user needs help on anything else. Avoid repeating previous responses.

nemo_voice_agent.pipecat.services.nemo.tts.KokoroTTSService.tool_tts_reset_voice(
    params: pipecat.services.llm_service.FunctionCallParams
)

async

Reset the accent and voice to the original ones.

nemo_voice_agent.pipecat.services.nemo.tts.KokoroTTSService.tool_tts_set_speed(
    params: pipecat.services.llm_service.FunctionCallParams,
    speed_lambda: float
)

async

Set a specific speaking speed of the assistant’s voice. This tool should be called only when the user specifies the speed explicitly, such as “speak twice as fast” or “speak half as slow” or “speak 1.5 times as fast”.

Parameters:

speed_lambda

float

positive float, the relative change of the speaking speed to the original speed. E.g., 1.0 for original speed, 1.25 for 25% faster than original speed, 0.8 for 20% slower than original speed.

nemo_voice_agent.pipecat.services.nemo.tts.KokoroTTSService.tool_tts_set_voice(
    params: pipecat.services.llm_service.FunctionCallParams,
    accent: str,
    gender: str
)

async

Set the accent and gender of the assistant’s voice. This tool should be called only when the user specifies the accent and/or gender explicitly.

Parameters:

accent

str

Accent for the TTS model. Must be one of ‘American English’, ‘British English’ or ‘current’ for keeping the current accent.

gender

str

gender of the assistant’s voice. Must be one of ‘male’, ‘female’, or ‘current’ for keeping the current gender.

nemo_voice_agent.pipecat.services.nemo.tts.KokoroTTSService.tool_tts_speak_faster(
    params: pipecat.services.llm_service.FunctionCallParams
)

async

Speak faster by increasing the speaking speed 15% faster each time this function is called.

nemo_voice_agent.pipecat.services.nemo.tts.KokoroTTSService.tool_tts_speak_slower(
    params: pipecat.services.llm_service.FunctionCallParams
)

async

Speak slower by decreasing the speaking speed 15% slower each time this function is called.

class nemo_voice_agent.pipecat.services.nemo.tts.MagpieTTSService(
    model: str = 'nvidia/magpie_tts_multilin...,
    language: str = 'en',
    speaker: str = 'Sofia',
    apply_TN: bool = False,
    device: str = 'cuda',
    kwargs = {}
)

Bases: BaseNemoTTSService

Text-to-Speech service using Magpie TTS model.

Magpie is a multilingual TTS model with 357 million parameters. More info: https://huggingface.co/nvidia/magpie_tts_multilingual_357m

Parameters:

model

str" default="'nvidia/magpie_tts_multilingual_357m'

Model name or path to the Magpie TTS model.

language

str" default="'en'

Language code for the model (default: ‘en’ for English)

speaker

str" default="'Sofia'

Speaker to use for the model (default: ‘Sofia’)

apply_TN

bool" default="False

Whether to apply text normalization (default: False)

device

str" default="'cuda'

Device to run on (default: ‘cuda’)

**kwargs

Defaults to {}

Additional arguments passed to BaseNemoTTSService

SPEAKER_MAP

nemo_voice_agent.pipecat.services.nemo.tts.MagpieTTSService._generate_audio(
    text: str
) -> typing.Iterator[numpy.ndarray]

nemo_voice_agent.pipecat.services.nemo.tts.MagpieTTSService._setup_model()

nemo_voice_agent.pipecat.services.nemo.tts.MagpieTTSService.setup_tool_calling()

No tools for now for Magpie TTS service.

class nemo_voice_agent.pipecat.services.nemo.tts.NeMoFastPitchHiFiGANTTSService(
    fastpitch_model: str = 'nvidia/tts_en_fastpitch',
    hifigan_model: str = 'nvidia/tts_hifigan',
    device: str = 'cuda',
    kwargs = {}
)

Bases: BaseNemoTTSService

Text-to-Speech service using NeMo FastPitch-Hifigan model.

More info: https://huggingface.co/nvidia/tts_en_fastpitch

Parameters:

fastpitch_model

str" default="'nvidia/tts_en_fastpitch'

FastPitch model name

hifigan_model

str" default="'nvidia/tts_hifigan'

Hifigan model name

device

str" default="'cuda'

Device to run on (default: ‘cuda’)

**kwargs

Defaults to {}

Additional arguments passed to BaseNemoTTSService

nemo_voice_agent.pipecat.services.nemo.tts.NeMoFastPitchHiFiGANTTSService._generate_audio(
    text: str
) -> typing.Iterator[numpy.ndarray]

nemo_voice_agent.pipecat.services.nemo.tts.NeMoFastPitchHiFiGANTTSService._setup_fastpitch_model(
    model_name: str
)

nemo_voice_agent.pipecat.services.nemo.tts.NeMoFastPitchHiFiGANTTSService._setup_hifigan_model(
    model_name: str
)

nemo_voice_agent.pipecat.services.nemo.tts.NeMoFastPitchHiFiGANTTSService._setup_model()

nemo_voice_agent.pipecat.services.nemo.tts.build_text_aggregator(
    config: omegaconf.DictConfig
) -> typing.Optional[nemo_voice_agent.pipecat.utils.text.simple_text_aggregator.SimpleSegmentedTextAggregator]

Build the sentence aggregator for TTS, or None if disabled by config.

Since pipecat 1.0, TTSService no longer accepts a text_aggregator argument — text aggregation belongs to an LLMTextProcessor placed upstream of the TTS service. Note that pipecat silently ignores unknown constructor kwargs, so passing it to the service would drop our segmentation with no error at all.

nemo_voice_agent.pipecat.services.nemo.tts.get_tts_service_from_config(
    config: omegaconf.DictConfig,
    audio_logger: typing.Optional[nemo_voice_agent.pipecat.services.nemo.audio_logger.AudioLogger] = None
) -> nemo_voice_agent.pipecat.services.nemo.tts.BaseNemoTTSService

Get the TTS service from the configuration.

Returns: The TTS service.

Parameters:

config

DictConfig

The DictConfig object containing the TTS configuration.

audio_logger

Optional[AudioLogger]" default="None

The audio logger to use for audio logging.