duocthu/apps/ai-service/adapters/bedrock_claude.py

"""Claude on Amazon Bedrock as the answer generator.

The only module that names the `anthropic` SDK, imported lazily — the same
arrangement that confines `boto3` to `embedding.py` and `qdrant_client` to
`qdrant.py`, so `rag/` imports and the whole suite runs with no SDK and no
cloud account.

Two provider facts, taken from the Anthropic API reference rather than from
memory: Bedrock model ids carry an `anthropic.` prefix (`anthropic.claude-opus-5`),
and the Messages-API path on Bedrock is the Mantle client — not the legacy
`bedrock-runtime` InvokeModel route the embedding adapter uses.
"""
from __future__ import annotations

import json
import re
from typing import Any

from rag.ports import AnswerGenerationUnavailable

BEDROCK_CLAUDE_OPUS_5 = "anthropic.claude-opus-5"

# Generation must not outrun the evidence it is rewriting. The section route
# can return a long section, so this is sized for the rewrite, not the source.
MAX_OUTPUT_TOKENS = 4096


def _provider_error_types() -> tuple[type[BaseException], ...]:
    """SDK and transport error classes, or none when the SDK is absent."""
    collected: list[type[BaseException]] = []
    try:
        import anthropic

        collected.append(anthropic.APIError)
    except ImportError:
        pass
    try:
        from botocore.exceptions import BotoCoreError, ClientError

        collected.extend((BotoCoreError, ClientError))
    except ImportError:
        pass
    return tuple(collected)


class BedrockClaudeAnswerGenerator:
    """Rewrites evidence into prose under a schema the API enforces.

    The output shape is constrained by `output_config.format` rather than by
    asking for JSON in the prompt, so a malformed envelope is the provider's
    error rather than this code's parsing problem. What the schema cannot
    constrain is whether the *content* is faithful — that is
    `rag.grounding.verify`'s job, and it runs on every response this returns.
    """

    def __init__(
        self,
        region: str = "us-east-1",
        client: Any | None = None,
        model_id: str = BEDROCK_CLAUDE_OPUS_5,
        max_tokens: int = MAX_OUTPUT_TOKENS,
    ) -> None:
        self._region = region
        self._client = client
        self._model_id = model_id
        self._max_tokens = max_tokens

    @property
    def model_id(self) -> str:
        return self._model_id

    def _runtime(self) -> Any:
        if self._client is None:
            from anthropic import AnthropicBedrockMantle

            # max_retries: the SDK's own default (2) is lower than the retry
            # budget given the other Bedrock adapters (bedrock_converse.py,
            # embedding.py) after a real throttling burst measured live
            # 2026-08-07 — matched here for consistency, in case this
            # provider is ever selected instead of bedrock-converse.
            self._client = AnthropicBedrockMantle(aws_region=self._region, max_retries=4)
        return self._client

    def generate(self, system: str, user: str, schema: dict) -> str:
        try:
            response = self._runtime().messages.create(
                model=self._model_id,
                max_tokens=self._max_tokens,
                system=system,
                messages=[{"role": "user", "content": user}],
                output_config={"format": {"type": "json_schema", "schema": schema}},
            )
        except _provider_error_types() as error:
            raise AnswerGenerationUnavailable(
                f"{self._model_id} could not be invoked: {type(error).__name__}"
            ) from error

        # A refusal is a successful HTTP response with no usable content, not
        # an exception. Treating it as an outage routes it to the extractive
        # fallback instead of letting `content[0]` raise.
        if getattr(response, "stop_reason", None) == "refusal":
            raise AnswerGenerationUnavailable(
                f"{self._model_id} declined the request"
            )

        text = "".join(
            block.text
            for block in response.content
            if getattr(block, "type", None) == "text"
        )
        if not text.strip():
            raise AnswerGenerationUnavailable(
                f"{self._model_id} returned no text content"
            )
        return text


class StubAnswerGenerator:
    """Returns a fixed payload; lets the whole answer path run with no cloud.

    Not a fake for tests only — it is what `EMBEDDING_PROVIDER`-style local
    demos use to exercise prompt building, schema parsing, grounding
    verification and the fallback branch without spending anything.
    """

    def __init__(self, answer: str, evidence_sufficient: bool = True) -> None:
        # `answer` keeps its old free-text-with-[n]-markers shape for this
        # constructor's own callers (bootstrap.py's demo string) — wrapped
        # into a single structured claim here since 2026-08-10's schema
        # change (see rag/prompt.py's ANSWER_SCHEMA).
        citations = [int(n) for n in re.findall(r"\[(\d+)\]", answer)]
        text = re.sub(r"\s*\[\d+\]", "", answer).strip()
        self._payload = json.dumps(
            {
                "claims": [{"text": text, "citations": citations}] if text else [],
                "evidence_sufficient": evidence_sufficient,
            },
            ensure_ascii=False,
        )
        self.calls: list[tuple[str, str]] = []

    def generate(self, system: str, user: str, schema: dict) -> str:  # noqa: ARG002
        # `schema` is unused here; the stub returns an already-valid payload.
        self.calls.append((system, user))
        return self._payload