invoke_model_with_bidirectional_stream

Operation

invoke_model_with_bidirectional_stream async

invoke_model_with_bidirectional_stream(input: InvokeModelWithBidirectionalStreamOperationInput, plugins: list[Plugin] | None = None) -> DuplexEventStream[InvokeModelWithBidirectionalStreamInput, InvokeModelWithBidirectionalStreamOutput, InvokeModelWithBidirectionalStreamOperationOutput]

Invoke the specified Amazon Bedrock model to run inference using the bidirectional stream. The response is returned in a stream that remains open for 8 minutes. A single session can contain multiple prompts and responses from the model. The prompts to the model are provided as audio files and the model's responses are spoken back to the user and transcribed.

It is possible for users to interrupt the model's response with a new prompt, which will halt the response speech. The model will retain contextual awareness of the conversation while pivoting to respond to the new prompt.

Parameters:

Name	Type	Description	Default
input	InvokeModelWithBidirectionalStreamOperationInput	An instance of InvokeModelWithBidirectionalStreamOperationInput.	required
plugins	list[Plugin] | None	A list of callables that modify the configuration dynamically. Changes made by these plugins only apply for the duration of the operation execution and will not affect any other operation invocations.	None

Returns:

Type	Description
DuplexEventStream[InvokeModelWithBidirectionalStreamInput, InvokeModelWithBidirectionalStreamOutput, InvokeModelWithBidirectionalStreamOperationOutput]	A DuplexEventStream for bidirectional streaming of InvokeModelWithBidirectionalStreamInput and InvokeModelWithBidirectionalStreamOutput events with initial InvokeModelWithBidirectionalStreamOperationOutput response.

Source code in src/aws_sdk_bedrock_runtime/client.py

async def invoke_model_with_bidirectional_stream(
    self,
    input: InvokeModelWithBidirectionalStreamOperationInput,
    plugins: list[Plugin] | None = None,
) -> DuplexEventStream[
    InvokeModelWithBidirectionalStreamInput,
    InvokeModelWithBidirectionalStreamOutput,
    InvokeModelWithBidirectionalStreamOperationOutput,
]:
    """Invoke the specified Amazon Bedrock model to run inference using the
    bidirectional stream. The response is returned in a stream that remains
    open for 8 minutes. A single session can contain multiple prompts and
    responses from the model. The prompts to the model are provided as audio
    files and the model's responses are spoken back to the user and
    transcribed.

    It is possible for users to interrupt the model's response with a new
    prompt, which will halt the response speech. The model will retain
    contextual awareness of the conversation while pivoting to respond to
    the new prompt.

    Args:
        input:
            An instance of `InvokeModelWithBidirectionalStreamOperationInput`.
        plugins:
            A list of callables that modify the configuration dynamically.
            Changes made by these plugins only apply for the duration of the
            operation execution and will not affect any other operation
            invocations.

    Returns:
        A `DuplexEventStream` for bidirectional streaming of
            `InvokeModelWithBidirectionalStreamInput` and
            `InvokeModelWithBidirectionalStreamOutput` events with initial
            `InvokeModelWithBidirectionalStreamOperationOutput` response.
    """
    operation_plugins: list[Plugin] = []
    if plugins:
        operation_plugins.extend(plugins)
    config = deepcopy(self._config)
    for plugin in operation_plugins:
        plugin(config)
    if config.protocol is None or config.transport is None:
        raise ExpectationNotMetError(
            "protocol and transport MUST be set on the config to make calls."
        )
    pipeline = RequestPipeline(protocol=config.protocol, transport=config.transport)
    call = ClientCall(
        input=input,
        operation=INVOKE_MODEL_WITH_BIDIRECTIONAL_STREAM,
        context=TypedProperties({"config": config}),
        interceptor=InterceptorChain(config.interceptors),
        auth_scheme_resolver=config.auth_scheme_resolver,
        supported_auth_schemes=config.auth_schemes,
        endpoint_resolver=config.endpoint_resolver,
        retry_strategy=config.retry_strategy,
    )

    return await pipeline.duplex_stream(
        call,
        InvokeModelWithBidirectionalStreamInput,
        InvokeModelWithBidirectionalStreamOutput,
        _InvokeModelWithBidirectionalStreamOutputDeserializer().deserialize,
    )

Input

InvokeModelWithBidirectionalStreamOperationInput dataclass

Dataclass for InvokeModelWithBidirectionalStreamOperationInput structure.

Source code in src/aws_sdk_bedrock_runtime/models.py

@dataclass(kw_only=True)
class InvokeModelWithBidirectionalStreamOperationInput:
    """Dataclass for InvokeModelWithBidirectionalStreamOperationInput
    structure.
    """

    model_id: str | None = None
    """The model ID or ARN of the model ID to use. Currently, only
    `amazon.nova-sonic-v1:0` is supported.
    """

    def serialize(self, serializer: ShapeSerializer):
        serializer.write_struct(
            _SCHEMA_INVOKE_MODEL_WITH_BIDIRECTIONAL_STREAM_OPERATION_INPUT, self
        )

    def serialize_members(self, serializer: ShapeSerializer):
        if self.model_id is not None:
            serializer.write_string(
                _SCHEMA_INVOKE_MODEL_WITH_BIDIRECTIONAL_STREAM_OPERATION_INPUT.members[
                    "modelId"
                ],
                self.model_id,
            )

    @classmethod
    def deserialize(cls, deserializer: ShapeDeserializer) -> Self:
        return cls(**cls.deserialize_kwargs(deserializer))

    @classmethod
    def deserialize_kwargs(cls, deserializer: ShapeDeserializer) -> dict[str, Any]:
        kwargs: dict[str, Any] = {}

        def _consumer(schema: Schema, de: ShapeDeserializer) -> None:
            match schema.expect_member_index():
                case 0:
                    kwargs["model_id"] = de.read_string(
                        _SCHEMA_INVOKE_MODEL_WITH_BIDIRECTIONAL_STREAM_OPERATION_INPUT.members[
                            "modelId"
                        ]
                    )

                case _:
                    logger.debug("Unexpected member schema: %s", schema)

        deserializer.read_struct(
            _SCHEMA_INVOKE_MODEL_WITH_BIDIRECTIONAL_STREAM_OPERATION_INPUT,
            consumer=_consumer,
        )
        return kwargs

Attributes

model_id class-attribute instance-attribute

model_id: str | None = None

The model ID or ARN of the model ID to use. Currently, only amazon.nova-sonic-v1:0 is supported.

Output

This operation returns a DuplexEventStream for bidirectional streaming.

Event Stream Structure

Input Event Type

InvokeModelWithBidirectionalStreamInput

Output Event Type

InvokeModelWithBidirectionalStreamOutput

Initial Response Structure

InvokeModelWithBidirectionalStreamOperationOutput dataclass

Dataclass for InvokeModelWithBidirectionalStreamOperationOutput structure.

Source code in src/aws_sdk_bedrock_runtime/models.py

@dataclass(kw_only=True)
class InvokeModelWithBidirectionalStreamOperationOutput:
    """Dataclass for InvokeModelWithBidirectionalStreamOperationOutput
    structure.
    """

    def serialize(self, serializer: ShapeSerializer):
        serializer.write_struct(
            _SCHEMA_INVOKE_MODEL_WITH_BIDIRECTIONAL_STREAM_OPERATION_OUTPUT, self
        )

    def serialize_members(self, serializer: ShapeSerializer):
        pass

    @classmethod
    def deserialize(cls, deserializer: ShapeDeserializer) -> Self:
        return cls(**cls.deserialize_kwargs(deserializer))

    @classmethod
    def deserialize_kwargs(cls, deserializer: ShapeDeserializer) -> dict[str, Any]:
        kwargs: dict[str, Any] = {}

        def _consumer(schema: Schema, de: ShapeDeserializer) -> None:
            match schema.expect_member_index():
                case _:
                    logger.debug("Unexpected member schema: %s", schema)

        deserializer.read_struct(
            _SCHEMA_INVOKE_MODEL_WITH_BIDIRECTIONAL_STREAM_OPERATION_OUTPUT,
            consumer=_consumer,
        )
        return kwargs

Errors

• AccessDeniedException

• InternalServerException

• ModelErrorException

• ModelNotReadyException

• ModelStreamErrorException

• ModelTimeoutException

• ResourceNotFoundException

• ServiceQuotaExceededException

• ServiceUnavailableException

• ThrottlingException

• ValidationException