wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[WIP] AWS Nova Sonic service

Browse Source
This commit is contained in:
Paul Kompfner
2025-04-24 15:23:01 -04:00
parent 8cbad070ad
commit b1d413b9be
1 changed files with 19 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
src/pipecat/services/aws_nova_sonic/aws.py
View File

@@ -286,14 +286,18 @@ class AWSNovaSonicService(LLMService):
# LLM communication: output events (LLM -> pipecat) # LLM communication: output events (LLM -> pipecat)
# #
# Receive LLM responses ("completions"). # Receive the ongoing LLM "completion".
# Each response contains up to four pieces of content, delivered sequentially: # There is generally a single completion per session.
# - User transcription # In a completion, a few different kinds of content can be delivered:
# - Tool use (optional) # - Transcription of user audio
# - Text response # - Tool use
# - Audio response # - Text preview of planned response speech before audio delivered
# Each piece of content is wrapped by "contentStart" and "contentEnd" events. # - User interruption notification
# Each overall response is wrapped by "completionStart" and "completionEnd" events. # - Text of response speech that whose audio was actually delivered
# - Audio of response speech
# Each piece of content is wrapped by "contentStart" and "contentEnd" events. The content is
# delivered sequentially: one piece of content will end before another starts.
# The overall completion is wrapped by "completionStart" and "completionEnd" events.
async def _receive_task_handler(self): async def _receive_task_handler(self):
try: try:
while self._client: while self._client:
@@ -335,7 +339,13 @@ class AWSNovaSonicService(LLMService):
content_start = event_json["contentStart"] content_start = event_json["contentStart"]
type = content_start["type"] type = content_start["type"]
role = content_start["role"] role = content_start["role"]
print(f"[pk] content start. type: {type}, role: {role}") generation_stage = None
if "additionalModelFields" in content_start:
additional_model_fields = json.loads(content_start["additionalModelFields"])
generation_stage = additional_model_fields.get("generationStage")
print(
f"[pk] content start. type: {type}, role: {role}, generation_stage: {generation_stage}"
)
async def _handle_text_output_event(self, event_json): async def _handle_text_output_event(self, event_json):
text_content = event_json["textOutput"]["content"] text_content = event_json["textOutput"]["content"]