AWS

exception pipecat.services.aws_nova_sonic.aws.AWSNovaSonicUnhandledFunctionException

Bases: Exception

class pipecat.services.aws_nova_sonic.aws.ContentType(*values)

Bases: Enum

AUDIO = 'AUDIO'

TEXT = 'TEXT'

TOOL = 'TOOL'

class pipecat.services.aws_nova_sonic.aws.TextStage(*values)

Bases: Enum

FINAL = 'FINAL'

SPECULATIVE = 'SPECULATIVE'

class pipecat.services.aws_nova_sonic.aws.CurrentContent(type: pipecat.services.aws_nova_sonic.aws.ContentType, role: pipecat.services.aws_nova_sonic.context.Role, text_stage: pipecat.services.aws_nova_sonic.aws.TextStage, text_content: str)

Bases: object

Parameters:

• type (ContentType)

• role (Role)

• text_stage (TextStage)

• text_content (str)

type: ContentType

role: Role

text_stage: TextStage

text_content: str

class pipecat.services.aws_nova_sonic.aws.Params(*, input_sample_rate=16000, input_sample_size=16, input_channel_count=1, output_sample_rate=24000, output_sample_size=16, output_channel_count=1, max_tokens=1024, top_p=0.9, temperature=0.7)

Bases: BaseModel

Parameters:

• input_sample_rate (int | None)

• input_sample_size (int | None)

• input_channel_count (int | None)

• output_sample_rate (int | None)

• output_sample_size (int | None)

• output_channel_count (int | None)

• max_tokens (int | None)

• top_p (float | None)

• temperature (float | None)

input_sample_rate: int | None

input_sample_size: int | None

input_channel_count: int | None

output_sample_rate: int | None

output_sample_size: int | None

output_channel_count: int | None

max_tokens: int | None

top_p: float | None

temperature: float | None

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pipecat.services.aws_nova_sonic.aws.AWSNovaSonicLLMService(*, secret_access_key, access_key_id, region, model='amazon.nova-sonic-v1:0', voice_id='matthew', params=None, system_instruction=None, tools=None, send_transcription_frames=True, **kwargs)

Bases: LLMService

Parameters:

• secret_access_key (str)

• access_key_id (str)

• region (str)

• model (str)

• voice_id (str)

• params (Params | None)

• system_instruction (str | None)

• tools (ToolsSchema | None)

• send_transcription_frames (bool)

adapter_class

alias of AWSNovaSonicLLMAdapter

async start(frame)

Start the LLM service.

Parameters:

frame (StartFrame) – The start frame.

async stop(frame)

Stop the LLM service.

Parameters:

frame (EndFrame) – The end frame.

async cancel(frame)

Cancel the LLM service.

Parameters:

frame (CancelFrame) – The cancel frame.

async reset_conversation()

async process_frame(frame, direction)

Process a frame.

Parameters:

• frame (Frame) – The frame to process.

• direction (FrameDirection) – The direction of frame processing.

create_context_aggregator(context, *, user_params=LLMUserAggregatorParams(aggregation_timeout=0.5), assistant_params=LLMAssistantAggregatorParams(expect_stripped_words=True))

Create a context aggregator for managing LLM conversation context.

Must be implemented by subclasses.

Parameters:

• context (OpenAILLMContext) – The LLM context to create an aggregator for.

• user_params (LLMUserAggregatorParams) – Parameters for user message aggregation.

• assistant_params (LLMAssistantAggregatorParams) – Parameters for assistant message aggregation.

Returns:

A context aggregator instance.

Return type:

AWSNovaSonicContextAggregatorPair

AWAIT_TRIGGER_ASSISTANT_RESPONSE_INSTRUCTION = "Start speaking when you hear the user say 'ready', but don't consider that 'ready' to be a meaningful part of the conversation other than as a trigger for you to start speaking."

async trigger_assistant_response()