Bedrock

@staticmethod
def create_payload(
    user_message: str | list[ContentItem],
    max_tokens: int | None = None,
    **kwargs: Any,
) -> dict:
    """
    Create a payload for the Bedrock Converse API request with optional multi-modal content.

    ⚠️ SECURITY WARNING: Format detection is for testing/development convenience ONLY.
    This method does NOT validate file safety, integrity, or protect against malicious
    content. DO NOT use with untrusted files (user uploads, external sources) without
    proper validation, sanitization, and security measures.

    Args:
        user_message: A single text string, or an ordered list mixing strings
            and :class:`~llmeter.prompt_utils.MediaContent` objects
            (``ImageContent``, ``AudioContent``, ``VideoContent``,
            ``DocumentContent``).  The order of items in the list controls
            the order of content blocks in the API request.
        max_tokens: Maximum number of tokens to generate. Defaults to 256.
        **kwargs: Additional keyword arguments to include in the payload.

    Returns:
        dict: The formatted payload for the Bedrock API request.

    Raises:
        TypeError: If parameters have invalid types.
        ValueError: If parameters have invalid values.

    Examples:
        Text only::

            create_payload("Hello")

        Image with text (order preserved)::

            create_payload([
                ImageContent.from_path("photo.jpg"),
                "What's in this image?",
            ])

        Mixed content::

            create_payload([
                "Compare the chart with the report:",
                ImageContent.from_path("chart.png"),
                DocumentContent.from_path("report.pdf"),
            ])
    """
    if max_tokens is None:
        max_tokens = 256
    if not isinstance(max_tokens, int) or max_tokens <= 0:
        raise ValueError("max_tokens must be a positive integer")

    # Normalise to a list of ContentItem
    if isinstance(user_message, str):
        items: list[ContentItem] = [user_message]
    elif isinstance(user_message, list):
        items = user_message
    else:
        raise TypeError(
            "user_message must be a str or list of str/MediaContent, "
            f"got {type(user_message).__name__}"
        )

    if not items:
        raise ValueError("user_message must not be empty")

    content_blocks = _build_content_blocks(items)

    payload: dict = {
        "messages": [{"role": "user", "content": content_blocks}],
    }
    payload.update(kwargs)
    if payload.get("inferenceConfig") is None:
        payload["inferenceConfig"] = {}
    payload["inferenceConfig"] = {
        **payload["inferenceConfig"],
        "maxTokens": max_tokens,
    }
    return payload

def prepare_payload(self, payload):
    """Enforce required properties on the input to Bedrock Converse* APIs

    In particular, ensure `modelId` is set in line with this Endpoint's configured model ID
    and apply `self._inference_config` if no config was explicitly set on the payload.
    """
    payload = {**payload}
    if payload.get("inferenceConfig") is None:
        payload["inferenceConfig"] = self._inference_config or {}
    payload["modelId"] = self.model_id
    return payload

@BedrockBase.llmeter_invoke
def invoke(self, payload: dict) -> ConverseResponseTypeDef:
    """Invoke the Bedrock converse API with the given payload."""
    client_response = self._bedrock_client.converse(**payload)  # type: ignore
    return client_response

def process_raw_response(
    self, raw_response, start_t: float, response: InvocationResponse
) -> None:
    """Parse the response from a Bedrock converse API call.

    Args:
        response: Raw response from the Bedrock API.
        start_t: The timestamp when the request was initiated.

    Returns:
        InvocationResponse with the generated text and metadata.
    """
    resp_meta = raw_response.get("ResponseMetadata", {})
    response.id = resp_meta.get("RequestId")
    response.retries = resp_meta.get("RetryAttempts")

    text_parts = [
        part["text"]
        for part in raw_response["output"]["message"]["content"]
        if "text" in part
    ]
    response.response_text = "".join(text_parts)

    usage = raw_response.get("usage", {})
    response.num_tokens_input = usage.get("inputTokens")
    response.num_tokens_input_cached = usage.get("cacheReadInputTokens")
    response.num_tokens_output = usage.get("outputTokens")

def process_raw_response(
    self, raw_response, start_t: float, response: InvocationResponse
) -> None:
    """Parse the streaming response from a Bedrock ConverseStream API call.

    Only `text` deltas contribute to `response_text`. `reasoningContent` deltas are used solely
    for TTFT measurement when `ttft_visible_tokens_only` is `False`.

    Args:
        raw_response: The raw response from the Bedrock API.
        start_t: The timestamp when the request was initiated.
        response: The output response object to be updated in-place.
    """
    response.id = raw_response["ResponseMetadata"].get("RequestId")
    response.retries = raw_response["ResponseMetadata"]["RetryAttempts"]

    for chunk in raw_response["stream"]:
        now = time.perf_counter()

        if "contentBlockDelta" in chunk:
            delta = chunk["contentBlockDelta"]["delta"]

            if "reasoningContent" in delta:
                # Reasoning delta -- only counts for TTFT when
                # ttft_visible_tokens_only is False.
                if (
                    not self.ttft_visible_tokens_only
                    and response.time_to_first_token is None
                ):
                    response.time_to_first_token = now - start_t

            elif "text" in delta:
                delta_text = delta["text"]
                if not isinstance(delta_text, str):
                    raise TypeError("Expected string for delta text")
                if delta_text:
                    if response.time_to_first_token is None:
                        response.time_to_first_token = now - start_t
                    if response.response_text is None:
                        response.response_text = delta_text
                    else:
                        response.response_text += delta_text
                    response.time_to_last_token = now - start_t

        if "contentBlockStop" in chunk:
            response.time_to_last_token = now - start_t

        if "metadata" in chunk:
            usage = chunk["metadata"].get("usage", {})
            input_tokens = usage.get("inputTokens")
            if input_tokens is not None:
                if response.num_tokens_input is None:
                    response.num_tokens_input = input_tokens
                else:
                    response.num_tokens_input += input_tokens
            output_tokens = usage.get("outputTokens")
            if output_tokens is not None:
                if response.num_tokens_output is None:
                    response.num_tokens_output = output_tokens
                else:
                    response.num_tokens_output += output_tokens
            cache_read_input_tokens = usage.get("cacheReadInputTokens")
            if cache_read_input_tokens is not None:
                if response.num_tokens_input_cached is None:
                    response.num_tokens_input_cached = cache_read_input_tokens
                else:
                    response.num_tokens_input_cached += cache_read_input_tokens

        # Detect Bedrock stream error events
        for error_type in BEDROCK_STREAM_ERROR_TYPES:
            if error_type in chunk:
                response.time_to_last_token = now - start_t
                raise RuntimeError(
                    f"Bedrock {error_type}: {chunk[error_type]['message']}"
                )