nemo_voice_agent.pipecat.services.nemo.llm

Module Contents

Classes

Name	Description
`HuggingFaceLLMLocalService`	HuggingFace LLM local service.
`HuggingFaceLLMService`	LLM service that hosts a HuggingFace model.
`LLMUtilsMixin`	Utils for local LLM services.
`VLLMService`	LLM service that hosts a vLLM server.

Functions

Name	Description
`get_llm_service_from_config`	Get an LLM service from the configuration.

Data

DEFAULT_GENERATION_KWARGS

API

class nemo_voice_agent.pipecat.services.nemo.llm.HuggingFaceLLMLocalService(
    model: str = 'meta-llama/Meta-Llama-3-8B...,
    device: str = 'cuda:0',
    dtype: str = 'bfloat16',
    reasoning_budget: int = 0,
    generation_kwargs: dict = None,
    apply_chat_template_kwargs: dict = None
)

Bases: LLMUtilsMixin

HuggingFace LLM local service.

apply_chat_template_kwargs

generation_kwargs

model

tokenizer

= AutoTokenizer.from_pretrained(model)

nemo_voice_agent.pipecat.services.nemo.llm.HuggingFaceLLMLocalService._apply_chat_template(
    messages: typing.List[openai.types.chat.ChatCompletionMessageParam]
) -> str

Apply the chat template to the messages.

nemo_voice_agent.pipecat.services.nemo.llm.HuggingFaceLLMLocalService._get_prompt_from_messages(
    messages: typing.List[openai.types.chat.ChatCompletionMessageParam]
) -> str

Get the formatted prompt from the conversation history messages. This function also tries to fix the messages if the LLM cannot handle consecutive turns of the same role, or requires a user turn after the system prompt.

nemo_voice_agent.pipecat.services.nemo.llm.HuggingFaceLLMLocalService.generate_stream(
    messages: typing.List[openai.types.chat.ChatCompletionMessageParam],
    kwargs = {}
) -> typing.AsyncGenerator[openai.types.chat.ChatCompletionChunk, None]

async

Generate a stream of chat completion chunks from the messages.

class nemo_voice_agent.pipecat.services.nemo.llm.HuggingFaceLLMService(
    model: str = 'google/gemma-7b-it',
    device: str = 'cuda',
    dtype: str = 'bfloat16',
    reasoning_budget: int = 0,
    generation_kwargs: dict = None,
    apply_chat_template_kwargs: dict = None,
    kwargs = {}
)

Bases: OpenAILLMService

LLM service that hosts a HuggingFace model.

_apply_chat_template_kwargs

_generation_kwargs

nemo_voice_agent.pipecat.services.nemo.llm.HuggingFaceLLMService._process_context(
    context: pipecat.processors.aggregators.llm_context.LLMContext
)

async

Process a context through the LLM and push text frames.

Parameters:

context

LLMContext

The context to process, containing messages and other information needed for the LLM interaction.

nemo_voice_agent.pipecat.services.nemo.llm.HuggingFaceLLMService.create_client(
    api_key = None,
    base_url = None,
    kwargs = {}
)

Create a HuggingFaceLLMLocalService client.

nemo_voice_agent.pipecat.services.nemo.llm.HuggingFaceLLMService.run_inference(
    context: pipecat.processors.aggregators.llm_context.LLMContext,
    max_tokens: typing.Optional[int] = None,
    system_instruction: typing.Optional[str] = None
) -> typing.Optional[str]

async

Run a one-shot, out-of-pipeline inference against the local HF model.

BaseOpenAILLMService.run_inference calls self._client.chat.completions.create(...) directly, but our create_client returns a HuggingFaceLLMLocalService, which only exposes generate_stream — so the inherited implementation raises AttributeError here. Nothing called run_inference before pipecat 1.0; it now backs auto context summarization and the LLM-turn-completion strategies, so implement it rather than leave a latent crash behind an opt-in flag.

class nemo_voice_agent.pipecat.services.nemo.llm.LLMUtilsMixin()

Utils for local LLM services.

nemo_voice_agent.pipecat.services.nemo.llm.LLMUtilsMixin._maybe_add_user_message(
    messages: typing.List[openai.types.chat.ChatCompletionMessageParam]
) -> typing.List[openai.types.chat.ChatCompletionMessageParam]

Some LLMs like “nvidia/Llama-3.1-Nemotron-Nano-8B-v1” requires a user turn after the system prompt, this function is used to add a dummy user turn if the system prompt is followed by an assistant turn.

nemo_voice_agent.pipecat.services.nemo.llm.LLMUtilsMixin._maybe_merge_consecutive_user_turns(
    messages: typing.List[openai.types.chat.ChatCompletionMessageParam]
) -> typing.List[openai.types.chat.ChatCompletionMessageParam]

Merge consecutive user turns into a single turn, since some LLMs like “nvidia/Llama-3.1-Nemotron-Nano-8B-v1” do not support consecutive user turns.

class nemo_voice_agent.pipecat.services.nemo.llm.VLLMService(
    model: str,
    device: str = 'cuda',
    api_key = 'None',
    base_url = 'http://localhost:8000/v1',
    organization = 'None',
    project = 'None',
    default_headers: typing.Optional[typing.Mapping[str, str]] = None,
    settings: typing.Optional[pipecat.services.openai.llm.OpenAILLMService.Settings] = None,
    start_vllm_on_init: bool = False,
    vllm_server_params: typing.Optional[str] = None,
    vllm_server_max_wait_time: int = 3600,
    vllm_server_check_interval: int = 5,
    kwargs = {}
)

Bases: OpenAILLMService, LLMUtilsMixin

LLM service that hosts a vLLM server.

nemo_voice_agent.pipecat.services.nemo.llm.VLLMService._get_response_from_client(
    messages: typing.List[openai.types.chat.ChatCompletionMessageParam],
    params: dict
) -> openai.AsyncStream[openai.types.chat.ChatCompletionChunk]

async

Get a response from the client.

nemo_voice_agent.pipecat.services.nemo.llm.VLLMService._get_response_from_client_with_reasoning(
    messages: typing.List[openai.types.chat.ChatCompletionMessageParam],
    params: dict
) -> openai.AsyncStream[openai.types.chat.ChatCompletionChunk]

async

Get a response from the client with reasoning.

nemo_voice_agent.pipecat.services.nemo.llm.VLLMService._start_vllm_server(
    model: str,
    vllm_server_params: typing.Optional[str] = None,
    base_url: typing.Optional[str] = None
) -> str

Start a vllm server and return the base url.

nemo_voice_agent.pipecat.services.nemo.llm.VLLMService._stop_vllm_server()

Stop the vLLM server process if it’s running.

nemo_voice_agent.pipecat.services.nemo.llm.VLLMService.cancel(
    frame: pipecat.frames.frames.CancelFrame
)

async

Cancel the LLM service.

Parameters:

frame

CancelFrame

The cancel frame.

nemo_voice_agent.pipecat.services.nemo.llm.VLLMService.get_chat_completions(
    context: pipecat.processors.aggregators.llm_context.LLMContext
) -> openai.AsyncStream[openai.types.chat.ChatCompletionChunk]

async

Get streaming chat completions from the vLLM OpenAI-compatible server.

Parameters:

context

LLMContext

The LLM context containing messages, tools and tool choice.

Returns: AsyncStream[ChatCompletionChunk]

Async stream of chat completion chunks.

nemo_voice_agent.pipecat.services.nemo.llm.VLLMService.stop(
    frame: pipecat.frames.frames.EndFrame
)

async

Stop the LLM service.

Parameters:

frame

EndFrame

The end frame.

nemo_voice_agent.pipecat.services.nemo.llm.get_llm_service_from_config(
    config: omegaconf.DictConfig
) -> pipecat.services.openai.llm.OpenAILLMService

Get an LLM service from the configuration.

nemo_voice_agent.pipecat.services.nemo.llm.DEFAULT_GENERATION_KWARGS = {'max_new_tokens': 256, 'temperature': 0.7, 'top_p': 0.95, 'do_sample': True}