sample_training_session_with_response_stream ¶

@contextmanager
def sample_training_session_with_response_stream(
    self,
    training_session_name: "capo_sagemakertrainingsessionruntime.types.training_session_name.TrainingSessionName",
    *,
    config_overrides: Optional[SageMakerTrainingSessionRuntimeClientConfig] = None,
    content_type: Optional[str] = None,
    accept: Optional[str] = None,
    model_package_arn: Optional[str] = None,
    body: Optional[
        "capo_sagemakertrainingsessionruntime.types.inference_body.InferenceBody"
    ] = None,
) -> "Generator[capo_sagemakertrainingsessionruntime.types.sample_training_session_with_response_stream_output.SampleTrainingSessionWithResponseStreamOutput]":
    """<p>Streams inference responses from the model associated with the specified training session. Use this operation for real-time token-by-token generation. The service passes the request body to the model without modification.</p>

    Args:
        training_session_name: <p>The name of the training session whose model to invoke.</p>
        content_type: <p>The MIME type of the input data in the request body.</p>
        accept: <p>The desired MIME type of the inference body in the response.</p>
        model_package_arn: <p>The ARN of the model package whose checkpoint to invoke.</p>
        body: <p>The raw inference request body.</p>

    Raises:
        capo_sagemakertrainingsessionruntime.errors.conflict_exception.ConflictException: <p>The training session is in a state that cannot accept new operations. The session must be in <code>Running</code> or <code>Blocked</code> state to accept requests.</p>
        capo_sagemakertrainingsessionruntime.errors.internal_server_exception.InternalServerException: <p>An internal service error occurred. Retry the request using exponential backoff.</p>
        capo_sagemakertrainingsessionruntime.errors.resource_not_found_exception.ResourceNotFoundException: <p>The specified training session or operation does not exist, or the operation does not belong to the specified training session.</p>
        capo_sagemakertrainingsessionruntime.errors.throttling_exception.ThrottlingException: <p>The request was throttled. Retry the request after a brief wait using exponential backoff.</p>
        capo_sagemakertrainingsessionruntime.errors.validation_exception.ValidationException: <p>The request is not valid. Check the request syntax and parameters. Examples include invalid sequence numbers, malformed client tokens, missing required fields, or invalid operation configuration.</p>
        capo_sagemakertrainingsessionruntime.errors.UnknownServiceError: The service returned an error code this client does not model.

    Examples:
        Stream samples from a training session
        Sends a chat completion request to the session's current weights and receives the response as a stream of payload chunks.

        >>> client.sample_training_session_with_response_stream(training_session_name='my-training-session', content_type='application/json', accept='application/json', body='eyJtZXNzYWdlcyI6W3sicm9sZSI6InVzZXIiLCJjb250ZW50IjoiV2hhdCBpcyB0aGUgY2FwaXRhbCBvZiBGcmFuY2U/In1dLCJtYXhfdG9rZW5zIjoxMjgsInN0cmVhbSI6dHJ1ZX0=')
    """

    def _handler(
        req: "OperationRequest[capo_sagemakertrainingsessionruntime.types.sample_training_session_with_response_stream_input.SampleTrainingSessionWithResponseStreamInput]",
    ) -> OperationResponse[
        "capo_sagemakertrainingsessionruntime.types.sample_training_session_with_response_stream_output.SampleTrainingSessionWithResponseStreamOutput"
    ]:
        import capo_sagemakertrainingsessionruntime._operations.lego_router_service.sample_training_session_with_response_stream

        output, http_response = (
            capo_sagemakertrainingsessionruntime._operations.lego_router_service.sample_training_session_with_response_stream.sample_training_session_with_response_stream(
                req.options, req.input
            )
        )
        return OperationResponse(output=output, response=http_response)

    def _redact_input(
        input_: "capo_sagemakertrainingsessionruntime.types.sample_training_session_with_response_stream_input.SampleTrainingSessionWithResponseStreamInput",
    ) -> dict[str, Any]:
        import capo_sagemakertrainingsessionruntime.types.sample_training_session_with_response_stream_input

        return capo_sagemakertrainingsessionruntime.types.sample_training_session_with_response_stream_input.redact_sensitive(
            input_
        )

    def _redact_output(
        output: "capo_sagemakertrainingsessionruntime.types.sample_training_session_with_response_stream_output.SampleTrainingSessionWithResponseStreamOutput",
    ) -> dict[str, Any]:
        import capo_sagemakertrainingsessionruntime.types.sample_training_session_with_response_stream_output

        return capo_sagemakertrainingsessionruntime.types.sample_training_session_with_response_stream_output.redact_sensitive(
            output
        )

    interceptors_, options_ = self.operation_options(
        config_overrides,
        OperationLogging(
            "SampleTrainingSessionWithResponseStream",
            redact_input=_redact_input,
            redact_output=_redact_output,
        ),
    )
    input_: capo_sagemakertrainingsessionruntime.types.sample_training_session_with_response_stream_input.SampleTrainingSessionWithResponseStreamInput = {
        "training_session_name": training_session_name
    }
    if content_type is not None:
        input_["content_type"] = content_type
    if accept is not None:
        input_["accept"] = accept
    if model_package_arn is not None:
        input_["model_package_arn"] = model_package_arn
    if body is not None:
        input_["body"] = body

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