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

[WIP] AWS Nova Sonic service - log an error message if you try to use AWS Nova Sonic without the proper dependency (e.g. without having done pip install pipecat-ai[aws])

Browse Source
This commit is contained in:
Paul Kompfner
2025-05-07 10:55:43 -04:00
parent b013e375fb
commit c78f779800
1 changed files with 21 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -15,23 +15,8 @@ from enum import Enum
from importlib.resources import files from importlib.resources import files
from typing import Any, List, Optional from typing import Any, List, Optional
from aws_sdk_bedrock_runtime.client import (
BedrockRuntimeClient,
InvokeModelWithBidirectionalStreamOperationInput,
)
from aws_sdk_bedrock_runtime.config import Config, HTTPAuthSchemeResolver, SigV4AuthScheme
from aws_sdk_bedrock_runtime.models import (
BidirectionalInputPayloadPart,
InvokeModelWithBidirectionalStreamInput,
InvokeModelWithBidirectionalStreamInputChunk,
InvokeModelWithBidirectionalStreamOperationOutput,
InvokeModelWithBidirectionalStreamOutput,
)
from loguru import logger from loguru import logger
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
from smithy_aws_core.credentials_resolvers.static import StaticCredentialsResolver
from smithy_aws_core.identity import AWSCredentialsIdentity
from smithy_core.aio.eventstream import DuplexEventStream
from pipecat.adapters.schemas.tools_schema import ToolsSchema from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.adapters.services.aws_nova_sonic_adapter import AWSNovaSonicLLMAdapter from pipecat.adapters.services.aws_nova_sonic_adapter import AWSNovaSonicLLMAdapter
@@ -45,15 +30,11 @@ from pipecat.frames.frames import (
LLMFullResponseStartFrame, LLMFullResponseStartFrame,
LLMTextFrame, LLMTextFrame,
StartFrame, StartFrame,
StartInterruptionFrame,
StopInterruptionFrame,
TranscriptionFrame, TranscriptionFrame,
TTSAudioRawFrame, TTSAudioRawFrame,
TTSStartedFrame, TTSStartedFrame,
TTSStoppedFrame, TTSStoppedFrame,
TTSTextFrame, TTSTextFrame,
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
) )
from pipecat.processors.aggregators.llm_response import ( from pipecat.processors.aggregators.llm_response import (
LLMAssistantAggregatorParams, LLMAssistantAggregatorParams,
@@ -75,6 +56,27 @@ from pipecat.services.aws_nova_sonic.frames import AWSNovaSonicFunctionCallResul
from pipecat.services.llm_service import LLMService from pipecat.services.llm_service import LLMService
from pipecat.utils.time import time_now_iso8601 from pipecat.utils.time import time_now_iso8601
try:
from aws_sdk_bedrock_runtime.client import (
BedrockRuntimeClient,
InvokeModelWithBidirectionalStreamOperationInput,
)
from aws_sdk_bedrock_runtime.config import Config, HTTPAuthSchemeResolver, SigV4AuthScheme
from aws_sdk_bedrock_runtime.models import (
BidirectionalInputPayloadPart,
InvokeModelWithBidirectionalStreamInput,
InvokeModelWithBidirectionalStreamInputChunk,
InvokeModelWithBidirectionalStreamOperationOutput,
InvokeModelWithBidirectionalStreamOutput,
)
from smithy_aws_core.credentials_resolvers.static import StaticCredentialsResolver
from smithy_aws_core.identity import AWSCredentialsIdentity
from smithy_core.aio.eventstream import DuplexEventStream
except ModuleNotFoundError as e:
logger.error(f"Exception: {e}")
logger.error("In order to use AWS services, you need to `pip install pipecat-ai[aws]`.")
raise Exception(f"Missing module: {e}")
class AWSNovaSonicUnhandledFunctionException(Exception): class AWSNovaSonicUnhandledFunctionException(Exception):
pass pass