invoke_model_with_bidirectional_stream ¶

@contextmanager
def invoke_model_with_bidirectional_stream(
    self,
    model_id: "capo_bedrock_runtime.types.invoke_model_identifier.InvokeModelIdentifier",
    body: "Iterator[capo_bedrock_runtime.types.invoke_model_with_bidirectional_stream_input._InvokeModelWithBidirectionalStreamInput] | capo_bedrock_runtime.types.invoke_model_with_bidirectional_stream_input._InvokeModelWithBidirectionalStreamInput",
    *,
    config_overrides: Optional[BedrockRuntimeClientConfig] = None,
) -> "Generator[capo_bedrock_runtime.types.invoke_model_with_bidirectional_stream_response.InvokeModelWithBidirectionalStreamResponse]":
    """<p>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.</p> <p>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.</p>

    Args:
        model_id: <p>The model ID or ARN of the model ID to use. Currently, only <code>amazon.nova-sonic-v1:0</code> is supported.</p>
        body: <p>The prompt and inference parameters in the format specified in the <code>BidirectionalInputPayloadPart</code> in the header. You must provide the body in JSON format. To see the format and content of the request and response bodies for different models, refer to <a href="https://docs.aws.amazon.com/bedrock/latest/userguide/model-parameters.html">Inference parameters</a>. For more information, see <a href="https://docs.aws.amazon.com/bedrock/latest/userguide/api-methods-run.html">Run inference</a> in the Bedrock User Guide.</p>

    Raises:
        capo_bedrock_runtime.errors.access_denied_exception.AccessDeniedException: <p>The request is denied because you do not have sufficient permissions to perform the requested action. For troubleshooting this error, see <a href="https://docs.aws.amazon.com/bedrock/latest/userguide/troubleshooting-api-error-codes.html#ts-access-denied">AccessDeniedException</a> in the Amazon Bedrock User Guide</p>
        capo_bedrock_runtime.errors.internal_server_exception.InternalServerException: <p>An internal server error occurred. For troubleshooting this error, see <a href="https://docs.aws.amazon.com/bedrock/latest/userguide/troubleshooting-api-error-codes.html#ts-internal-failure">InternalFailure</a> in the Amazon Bedrock User Guide</p>
        capo_bedrock_runtime.errors.model_error_exception.ModelErrorException: <p>The request failed due to an error while processing the model.</p>
        capo_bedrock_runtime.errors.model_not_ready_exception.ModelNotReadyException: <p>The model specified in the request is not ready to serve inference requests. The AWS SDK will automatically retry the operation up to 5 times. For information about configuring automatic retries, see <a href="https://docs.aws.amazon.com/sdkref/latest/guide/feature-retry-behavior.html">Retry behavior</a> in the <i>AWS SDKs and Tools</i> reference guide.</p>
        capo_bedrock_runtime.errors.model_stream_error_exception.ModelStreamErrorException: <p>An error occurred while streaming the response. Retry your request.</p>
        capo_bedrock_runtime.errors.model_timeout_exception.ModelTimeoutException: <p>The request took too long to process. Processing time exceeded the model timeout length.</p>
        capo_bedrock_runtime.errors.resource_not_found_exception.ResourceNotFoundException: <p>The specified resource ARN was not found. For troubleshooting this error, see <a href="https://docs.aws.amazon.com/bedrock/latest/userguide/troubleshooting-api-error-codes.html#ts-resource-not-found">ResourceNotFound</a> in the Amazon Bedrock User Guide</p>
        capo_bedrock_runtime.errors.service_quota_exceeded_exception.ServiceQuotaExceededException: <p>Your request exceeds the service quota for your account. You can view your quotas at <a href="https://docs.aws.amazon.com/servicequotas/latest/userguide/gs-request-quota.html">Viewing service quotas</a>. You can resubmit your request later.</p>
        capo_bedrock_runtime.errors.service_unavailable_exception.ServiceUnavailableException: <p>The service isn't currently available. For troubleshooting this error, see <a href="https://docs.aws.amazon.com/bedrock/latest/userguide/troubleshooting-api-error-codes.html#ts-service-unavailable">ServiceUnavailable</a> in the Amazon Bedrock User Guide</p>
        capo_bedrock_runtime.errors.throttling_exception.ThrottlingException: <p>Your request was denied due to exceeding the account quotas for <i>Amazon Bedrock</i>. For troubleshooting this error, see <a href="https://docs.aws.amazon.com/bedrock/latest/userguide/troubleshooting-api-error-codes.html#ts-throttling-exception">ThrottlingException</a> in the Amazon Bedrock User Guide</p>
        capo_bedrock_runtime.errors.validation_exception.ValidationException: <p>The input fails to satisfy the constraints specified by <i>Amazon Bedrock</i>. For troubleshooting this error, see <a href="https://docs.aws.amazon.com/bedrock/latest/userguide/troubleshooting-api-error-codes.html#ts-validation-error">ValidationError</a> in the Amazon Bedrock User Guide</p>
        capo_bedrock_runtime.errors.UnknownServiceError: The service returned an error code this client does not model.
    """

    def _handler(
        req: "OperationRequest[capo_bedrock_runtime.types.invoke_model_with_bidirectional_stream_request.InvokeModelWithBidirectionalStreamRequest]",
    ) -> OperationResponse[
        "capo_bedrock_runtime.types.invoke_model_with_bidirectional_stream_response.InvokeModelWithBidirectionalStreamResponse"
    ]:
        import capo_bedrock_runtime._operations.amazon_bedrock_frontend_service.invoke_model_with_bidirectional_stream

        output, http_response = (
            capo_bedrock_runtime._operations.amazon_bedrock_frontend_service.invoke_model_with_bidirectional_stream.invoke_model_with_bidirectional_stream(
                req.options, req.input
            )
        )
        return OperationResponse(output=output, response=http_response)

    def _redact_input(
        input_: "capo_bedrock_runtime.types.invoke_model_with_bidirectional_stream_request.InvokeModelWithBidirectionalStreamRequest",
    ) -> dict[str, Any]:
        import capo_bedrock_runtime.types.invoke_model_with_bidirectional_stream_request

        return capo_bedrock_runtime.types.invoke_model_with_bidirectional_stream_request.redact_sensitive(
            input_
        )

    def _redact_output(
        output: "capo_bedrock_runtime.types.invoke_model_with_bidirectional_stream_response.InvokeModelWithBidirectionalStreamResponse",
    ) -> dict[str, Any]:
        import capo_bedrock_runtime.types.invoke_model_with_bidirectional_stream_response

        return capo_bedrock_runtime.types.invoke_model_with_bidirectional_stream_response.redact_sensitive(
            output
        )

    interceptors_, options_ = self.operation_options(
        config_overrides,
        OperationLogging(
            "InvokeModelWithBidirectionalStream",
            redact_input=_redact_input,
            redact_output=_redact_output,
        ),
    )
    input_: capo_bedrock_runtime.types.invoke_model_with_bidirectional_stream_request.InvokeModelWithBidirectionalStreamRequest = {
        "model_id": model_id,
        "body": ensure_sync_iterator(body),
    }

    response = execute_pipeline(
        OperationRequest(input=input_, options=options_),
        handler=_handler,
        interceptors=list(interceptors_),
    )
    try:
        yield response.output
    finally:
        response.response.close()