sample_with_response_stream ¶

@contextmanager
def sample_with_response_stream(
    self,
    job_arn: "capo_sagemakerjobruntime.types.job_arn.JobArn",
    trajectory_id: "capo_sagemakerjobruntime.types.trajectory_id.TrajectoryId",
    body: "capo_sagemakerjobruntime.types.inference_request_body.InferenceRequestBody",
    *,
    config_overrides: Optional[SagemakerJobRuntimeClientConfig] = None,
) -> "Generator[capo_sagemakerjobruntime.types.sample_with_response_stream_response.SampleWithResponseStreamResponse]":
    """Sends a streaming inference request to the model during a job execution. Returns the response as a stream of payload chunks. Each turn is captured for later use.

    Args:
        job_arn: The job ARN that identifies which model session to route the inference request to.
        trajectory_id: The trajectory ID for grouping turns into a single rollout. Each turn is captured for later use.
        body: The raw inference request body in OpenAI-compatible JSON format.

    Raises:
        capo_sagemakerjobruntime.errors.access_denied_exception.AccessDeniedException: You do not have permission to perform this operation.
        capo_sagemakerjobruntime.errors.internal_service_error.InternalServiceError: An internal service error occurred. Retry the request.
        capo_sagemakerjobruntime.errors.resource_not_found_exception.ResourceNotFoundException: The specified resource was not found.
        capo_sagemakerjobruntime.errors.service_quota_exceeded_exception.ServiceQuotaExceededException: You have exceeded a service quota.
        capo_sagemakerjobruntime.errors.throttling_exception.ThrottlingException: The request was throttled. Retry the request after a brief wait.
        capo_sagemakerjobruntime.errors.validation_exception.ValidationException: The request is not valid. Check the request syntax and parameters
        capo_sagemakerjobruntime.errors.UnknownServiceError: The service returned an error code this client does not model.

    Examples:
        Invoke SampleWithResponseStream
        Sends a streaming inference request and receives the response as a stream of payload chunks.

        >>> client.sample_with_response_stream(job_arn='arn:aws:sagemaker:us-east-1:123456789012:job/AgentRFT/my-training-job', trajectory_id='trajectory-001', body='eyJtb2RlbCI6Im15LW1vZGVsIiwibWVzc2FnZXMiOlt7InJvbGUiOiJ1c2VyIiwiY29udGVudCI6IkhlbGxvIn1dfQ==')
    """

    def _handler(
        req: "OperationRequest[capo_sagemakerjobruntime.types.sample_with_response_stream_request.SampleWithResponseStreamRequest]",
    ) -> OperationResponse[
        "capo_sagemakerjobruntime.types.sample_with_response_stream_response.SampleWithResponseStreamResponse"
    ]:
        import capo_sagemakerjobruntime._operations.agentic_rft_runtime_service.sample_with_response_stream

        output, http_response = (
            capo_sagemakerjobruntime._operations.agentic_rft_runtime_service.sample_with_response_stream.sample_with_response_stream(
                req.options, req.input
            )
        )
        return OperationResponse(output=output, response=http_response)

    def _redact_input(
        input_: "capo_sagemakerjobruntime.types.sample_with_response_stream_request.SampleWithResponseStreamRequest",
    ) -> dict[str, Any]:
        import capo_sagemakerjobruntime.types.sample_with_response_stream_request

        return capo_sagemakerjobruntime.types.sample_with_response_stream_request.redact_sensitive(
            input_
        )

    def _redact_output(
        output: "capo_sagemakerjobruntime.types.sample_with_response_stream_response.SampleWithResponseStreamResponse",
    ) -> dict[str, Any]:
        import capo_sagemakerjobruntime.types.sample_with_response_stream_response

        return capo_sagemakerjobruntime.types.sample_with_response_stream_response.redact_sensitive(
            output
        )

    interceptors_, options_ = self.operation_options(
        config_overrides,
        OperationLogging(
            "SampleWithResponseStream",
            redact_input=_redact_input,
            redact_output=_redact_output,
        ),
    )
    input_: capo_sagemakerjobruntime.types.sample_with_response_stream_request.SampleWithResponseStreamRequest = {
        "job_arn": job_arn,
        "trajectory_id": trajectory_id,
        "body": body,
    }

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