OpenAI・Claude・Geminiを切り替えるAI API設計｜Provider抽象化とフォールバック実装

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Literal

ResponseFormat = Literal["text", "json", "html"]


@dataclass(frozen=True, slots=True)
class AIRequest:
    system_instruction: str
    user_input: str
    temperature: float = 0.2
    max_output_tokens: int = 1_500
    response_format: ResponseFormat = "text"
    metadata: dict[str, Any] = field(default_factory=dict)


@dataclass(frozen=True, slots=True)
class UsageMetrics:
    input_tokens: int | None = None
    output_tokens: int | None = None
    estimated_cost: float | None = None


@dataclass(frozen=True, slots=True)
class ValidationResult:
    is_valid: bool
    issues: tuple[str, ...] = ()


@dataclass(frozen=True, slots=True)
class AIResponse:
    content: str
    provider: str
    model: str
    finish_reason: str | None
    usage: UsageMetrics
    latency_ms: int
    fallback_count: int = 0
    validation_result: ValidationResult | None = None
    request_id: str | None = None


@dataclass(frozen=True, slots=True)
class ProviderAttemptLog:
    operation_id: str
    provider: str
    model: str
    attempt_in_provider: int
    fallback_count: int
    started_at: datetime
    ended_at: datetime
    latency_ms: int
    success: bool
    error_kind: str | None = None
    provider_error_code: str | None = None
    request_id: str | None = None
    input_tokens: int | None = None
    output_tokens: int | None = None
    validation_succeeded: bool | None = None

from __future__ import annotations

from enum import StrEnum
from typing import Sequence


class ProviderErrorKind(StrEnum):
    CONNECTION = "connection"
    TIMEOUT = "timeout"
    RATE_LIMIT = "rate_limit"
    SERVER = "server"
    OVERLOADED = "overloaded"
    MODEL_UNAVAILABLE = "model_unavailable"
    AUTHENTICATION = "authentication"
    PERMISSION = "permission"
    BILLING = "billing"
    INVALID_REQUEST = "invalid_request"
    INPUT_TOO_LARGE = "input_too_large"
    POLICY_REFUSAL = "policy_refusal"
    CANCELLED = "cancelled"
    OUTPUT_VALIDATION = "output_validation"
    UNKNOWN = "unknown"


class AIProviderError(Exception):
    def __init__(
        self,
        *,
        provider: str,
        kind: ProviderErrorKind,
        message: str,
        status_code: int | None = None,
        provider_code: str | None = None,
        request_id: str | None = None,
    ) -> None:
        super().__init__(message)
        self.provider = provider
        self.kind = kind
        self.status_code = status_code
        self.provider_code = provider_code
        self.request_id = request_id


class AllProvidersFailedError(RuntimeError):
    def __init__(self, errors: Sequence[AIProviderError]) -> None:
        super().__init__("利用可能なAI Providerで処理を完了できませんでした。")
        self.errors = tuple(errors)

from typing import Protocol

from app.domain.models import AIRequest, AIResponse


class AIProvider(Protocol):
    name: str
    model: str

    async def generate(self, request: AIRequest) -> AIResponse:
        ...

from typing import Protocol

from app.domain.models import ProviderAttemptLog


class ProviderLogRepository(Protocol):
    async def save(self, log: ProviderAttemptLog) -> None:
        ...

from __future__ import annotations

import anthropic
import openai
from google.genai import errors as gemini_errors

from app.domain.errors import AIProviderError, ProviderErrorKind


def _message_contains(message: str, *keywords: str) -> bool:
    lowered = message.lower()
    return any(keyword in lowered for keyword in keywords)


def _classify_rate_limit(message: str) -> ProviderErrorKind:
    # 429でも、短期的な混雑と請求・上限超過では対応が違います。
    if _message_contains(
        message,
        "insufficient_quota",
        "billing",
        "hard limit",
        "payment required",
    ):
        return ProviderErrorKind.BILLING
    return ProviderErrorKind.RATE_LIMIT


def _classify_not_found(message: str) -> ProviderErrorKind:
    # モデル廃止・利用不可と、URLや設定ミスを同じ404として扱わないための簡易分類です。
    if _message_contains(message, "model", "not found", "does not exist"):
        return ProviderErrorKind.MODEL_UNAVAILABLE
    return ProviderErrorKind.INVALID_REQUEST


def map_openai_error(exc: openai.APIError, provider: str) -> AIProviderError:
    message = str(exc)
    status_code = getattr(exc, "status_code", None)
    request_id = getattr(exc, "request_id", None)

    if isinstance(exc, openai.APITimeoutError):
        kind = ProviderErrorKind.TIMEOUT
    elif isinstance(exc, openai.APIConnectionError):
        kind = ProviderErrorKind.CONNECTION
    elif isinstance(exc, openai.AuthenticationError):
        kind = ProviderErrorKind.AUTHENTICATION
    elif isinstance(exc, openai.PermissionDeniedError):
        kind = ProviderErrorKind.PERMISSION
    elif isinstance(exc, openai.RateLimitError):
        kind = _classify_rate_limit(message)
    elif isinstance(exc, openai.NotFoundError):
        kind = _classify_not_found(message)
    elif isinstance(exc, (openai.BadRequestError, openai.UnprocessableEntityError)):
        kind = ProviderErrorKind.INVALID_REQUEST
    elif isinstance(exc, openai.InternalServerError):
        kind = ProviderErrorKind.SERVER
    else:
        kind = ProviderErrorKind.UNKNOWN

    return AIProviderError(
        provider=provider,
        kind=kind,
        message=message,
        status_code=status_code,
        provider_code=str(status_code) if status_code is not None else None,
        request_id=request_id,
    )


def map_anthropic_error(
    exc: anthropic.APIError,
    provider: str,
) -> AIProviderError:
    message = str(exc)
    status_code = getattr(exc, "status_code", None)
    request_id = getattr(exc, "request_id", None)

    if isinstance(exc, anthropic.APITimeoutError):
        kind = ProviderErrorKind.TIMEOUT
    elif isinstance(exc, anthropic.APIConnectionError):
        kind = ProviderErrorKind.CONNECTION
    elif isinstance(exc, anthropic.AuthenticationError):
        kind = ProviderErrorKind.AUTHENTICATION
    elif isinstance(exc, anthropic.PermissionDeniedError):
        kind = ProviderErrorKind.PERMISSION
    elif isinstance(exc, anthropic.RateLimitError):
        kind = _classify_rate_limit(message)
    elif isinstance(exc, anthropic.NotFoundError):
        kind = _classify_not_found(message)
    elif isinstance(exc, (anthropic.BadRequestError, anthropic.UnprocessableEntityError)):
        kind = ProviderErrorKind.INVALID_REQUEST
    elif isinstance(exc, anthropic.InternalServerError):
        kind = ProviderErrorKind.SERVER
    else:
        kind = ProviderErrorKind.UNKNOWN

    return AIProviderError(
        provider=provider,
        kind=kind,
        message=message,
        status_code=status_code,
        provider_code=str(status_code) if status_code is not None else None,
        request_id=request_id,
    )


def map_gemini_error(
    exc: gemini_errors.APIError,
    provider: str,
) -> AIProviderError:
    status_code = getattr(exc, "code", None)
    message = getattr(exc, "message", str(exc))

    if status_code == 401:
        kind = ProviderErrorKind.AUTHENTICATION
    elif status_code == 403:
        kind = ProviderErrorKind.PERMISSION
    elif status_code == 404:
        kind = _classify_not_found(message)
    elif status_code in {408, 504}:
        kind = ProviderErrorKind.TIMEOUT
    elif status_code == 429:
        kind = _classify_rate_limit(message)
    elif status_code is not None and status_code >= 500:
        kind = ProviderErrorKind.SERVER
    elif status_code in {400, 422}:
        kind = ProviderErrorKind.INVALID_REQUEST
    else:
        kind = ProviderErrorKind.UNKNOWN

    return AIProviderError(
        provider=provider,
        kind=kind,
        message=message,
        status_code=status_code,
        provider_code=str(status_code) if status_code is not None else None,
    )

from __future__ import annotations

from time import perf_counter

import openai
from openai import AsyncOpenAI

from app.domain.errors import AIProviderError, ProviderErrorKind
from app.domain.models import AIRequest, AIResponse, UsageMetrics
from app.providers.error_mapping import map_openai_error


class OpenAIProvider:
    name = "openai"

    def __init__(
        self,
        *,
        api_key: str,
        model: str,
        timeout_seconds: float,
    ) -> None:
        self.model = model
        self._client = AsyncOpenAI(
            api_key=api_key,
            timeout=timeout_seconds,
            max_retries=0,
        )

    async def generate(self, request: AIRequest) -> AIResponse:
        started = perf_counter()
        instructions = _build_instructions(request)

        try:
            response = await self._client.responses.create(
                model=self.model,
                instructions=instructions or None,
                input=request.user_input,
                max_output_tokens=request.max_output_tokens,
            )
        except openai.APIError as exc:
            raise map_openai_error(exc, self.name) from exc

        refusal = _extract_refusal(response)
        if refusal is not None:
            raise AIProviderError(
                provider=self.name,
                kind=ProviderErrorKind.POLICY_REFUSAL,
                message=refusal,
                request_id=getattr(response, "_request_id", None),
            )

        usage = getattr(response, "usage", None)
        return AIResponse(
            content=response.output_text or "",
            provider=self.name,
            model=self.model,
            finish_reason=getattr(response, "status", None),
            usage=UsageMetrics(
                input_tokens=getattr(usage, "input_tokens", None),
                output_tokens=getattr(usage, "output_tokens", None),
            ),
            latency_ms=int((perf_counter() - started) * 1_000),
            request_id=getattr(response, "_request_id", None),
        )


def _build_instructions(request: AIRequest) -> str:
    format_instruction = {
        "text": "プレーンテキストで返してください。",
        "json": "コードフェンスを使わず、有効なJSONだけを返してください。",
        "html": "説明文やコードフェンスを付けず、HTML断片だけを返してください。",
    }[request.response_format]
    return "\n\n".join(
        part for part in (request.system_instruction.strip(), format_instruction) if part
    )


def _extract_refusal(response: object) -> str | None:
    # Responses APIの出力を汎用的に走査し、refusalを別Providerへ自動転送しないよう検出します。
    for item in getattr(response, "output", ()) or ():
        for part in getattr(item, "content", ()) or ():
            if getattr(part, "type", None) == "refusal":
                return getattr(part, "refusal", None) or "OpenAIが応答を拒否しました。"
    return None

from __future__ import annotations

from time import perf_counter

import anthropic
from anthropic import AsyncAnthropic

from app.domain.errors import AIProviderError, ProviderErrorKind
from app.domain.models import AIRequest, AIResponse, UsageMetrics
from app.providers.error_mapping import map_anthropic_error


class AnthropicProvider:
    name = "anthropic"

    def __init__(
        self,
        *,
        api_key: str,
        model: str,
        timeout_seconds: float,
    ) -> None:
        self.model = model
        self._client = AsyncAnthropic(
            api_key=api_key,
            timeout=timeout_seconds,
            max_retries=0,
        )

    async def generate(self, request: AIRequest) -> AIResponse:
        started = perf_counter()
        system = _build_system_instruction(request)

        try:
            message = await self._client.messages.create(
                model=self.model,
                max_tokens=request.max_output_tokens,
                temperature=request.temperature,
                system=system,
                messages=[{"role": "user", "content": request.user_input}],
            )
        except anthropic.APIError as exc:
            raise map_anthropic_error(exc, self.name) from exc

        if getattr(message, "stop_reason", None) == "refusal":
            raise AIProviderError(
                provider=self.name,
                kind=ProviderErrorKind.POLICY_REFUSAL,
                message="Anthropicが応答を拒否しました。",
                request_id=getattr(message, "_request_id", None),
            )

        content = "".join(
            block.text
            for block in message.content
            if getattr(block, "type", None) == "text"
        )

        return AIResponse(
            content=content,
            provider=self.name,
            model=self.model,
            finish_reason=message.stop_reason,
            usage=UsageMetrics(
                input_tokens=message.usage.input_tokens,
                output_tokens=message.usage.output_tokens,
            ),
            latency_ms=int((perf_counter() - started) * 1_000),
            request_id=getattr(message, "_request_id", None),
        )


def _build_system_instruction(request: AIRequest) -> str:
    format_instruction = {
        "text": "プレーンテキストで返してください。",
        "json": "コードフェンスを使わず、有効なJSONだけを返してください。",
        "html": "説明文やコードフェンスを付けず、HTML断片だけを返してください。",
    }[request.response_format]
    return "\n\n".join(
        part for part in (request.system_instruction.strip(), format_instruction) if part
    )

from __future__ import annotations

from time import perf_counter

from google import genai
from google.genai import errors as gemini_errors
from google.genai import types

from app.domain.errors import AIProviderError, ProviderErrorKind
from app.domain.models import AIRequest, AIResponse, UsageMetrics
from app.providers.error_mapping import map_gemini_error


class GeminiProvider:
    name = "gemini"

    def __init__(
        self,
        *,
        api_key: str,
        model: str,
        timeout_seconds: float,
    ) -> None:
        self.model = model
        self._client = genai.Client(
            api_key=api_key,
            http_options=types.HttpOptions(
                # google-genaiのtimeoutはミリ秒です。
                timeout=int(timeout_seconds * 1_000),
                # 初回リクエストだけにし、Router側Policyと再試行を二重化しません。
                retry_options=types.HttpRetryOptions(attempts=1),
            ),
        )

    async def generate(self, request: AIRequest) -> AIResponse:
        started = perf_counter()

        try:
            response = await self._client.aio.models.generate_content(
                model=self.model,
                contents=request.user_input,
                config=types.GenerateContentConfig(
                    system_instruction=_build_system_instruction(request),
                    max_output_tokens=request.max_output_tokens,
                    temperature=request.temperature,
                ),
            )
        except gemini_errors.APIError as exc:
            raise map_gemini_error(exc, self.name) from exc

        prompt_feedback = getattr(response, "prompt_feedback", None)
        if getattr(prompt_feedback, "block_reason", None):
            raise AIProviderError(
                provider=self.name,
                kind=ProviderErrorKind.POLICY_REFUSAL,
                message=(
                    getattr(prompt_feedback, "block_reason_message", None)
                    or "Geminiが応答を拒否しました。"
                ),
                request_id=getattr(response, "response_id", None),
            )

        candidate = response.candidates[0] if response.candidates else None
        finish_reason = getattr(candidate, "finish_reason", None)
        if str(finish_reason).upper().endswith(
            ("SAFETY", "BLOCKLIST", "PROHIBITED_CONTENT")
        ):
            raise AIProviderError(
                provider=self.name,
                kind=ProviderErrorKind.POLICY_REFUSAL,
                message="Geminiの安全設定により応答が停止しました。",
                request_id=getattr(response, "response_id", None),
            )

        usage = getattr(response, "usage_metadata", None)
        return AIResponse(
            content=response.text or "",
            provider=self.name,
            model=self.model,
            finish_reason=str(finish_reason) if finish_reason is not None else None,
            usage=UsageMetrics(
                input_tokens=getattr(usage, "prompt_token_count", None),
                output_tokens=getattr(usage, "candidates_token_count", None),
            ),
            latency_ms=int((perf_counter() - started) * 1_000),
            request_id=getattr(response, "response_id", None),
        )


def _build_system_instruction(request: AIRequest) -> str:
    format_instruction = {
        "text": "プレーンテキストで返してください。",
        "json": "コードフェンスを使わず、有効なJSONだけを返してください。",
        "html": "説明文やコードフェンスを付けず、HTML断片だけを返してください。",
    }[request.response_format]
    return "\n\n".join(
        part for part in (request.system_instruction.strip(), format_instruction) if part
    )

from dataclasses import dataclass
from enum import StrEnum

from app.domain.errors import AIProviderError, ProviderErrorKind


class FailureAction(StrEnum):
    RETRY_SAME = "retry_same"
    NEXT_PROVIDER = "next_provider"
    STOP = "stop"


@dataclass(frozen=True, slots=True)
class FallbackPolicy:
    max_attempts_per_provider: int = 2
    base_backoff_seconds: float = 0.4

    def decide(
        self,
        *,
        error: AIProviderError,
        attempt_in_provider: int,
        has_next_provider: bool,
    ) -> FailureAction:
        stop_kinds = {
            ProviderErrorKind.AUTHENTICATION,
            ProviderErrorKind.PERMISSION,
            ProviderErrorKind.BILLING,
            ProviderErrorKind.INVALID_REQUEST,
            ProviderErrorKind.INPUT_TOO_LARGE,
            ProviderErrorKind.POLICY_REFUSAL,
            ProviderErrorKind.CANCELLED,
            ProviderErrorKind.UNKNOWN,
        }
        if error.kind in stop_kinds:
            return FailureAction.STOP

        if error.kind == ProviderErrorKind.MODEL_UNAVAILABLE:
            return FailureAction.NEXT_PROVIDER if has_next_provider else FailureAction.STOP

        retryable_kinds = {
            ProviderErrorKind.CONNECTION,
            ProviderErrorKind.TIMEOUT,
            ProviderErrorKind.RATE_LIMIT,
            ProviderErrorKind.SERVER,
            ProviderErrorKind.OVERLOADED,
            ProviderErrorKind.OUTPUT_VALIDATION,
        }
        if error.kind in retryable_kinds:
            if attempt_in_provider < self.max_attempts_per_provider:
                return FailureAction.RETRY_SAME
            return FailureAction.NEXT_PROVIDER if has_next_provider else FailureAction.STOP

        return FailureAction.STOP

    def backoff_seconds(self, attempt_in_provider: int) -> float:
        return self.base_backoff_seconds * (2 ** max(0, attempt_in_provider - 1))

from __future__ import annotations

import json
import re
from dataclasses import replace
from html.parser import HTMLParser
from urllib.parse import urlparse

from app.domain.models import AIRequest, AIResponse, ValidationResult


class _HTMLInspectionParser(HTMLParser):
    def __init__(self) -> None:
        super().__init__()
        self.tags: list[str] = []
        self.urls: list[str] = []

    def handle_starttag(
        self,
        tag: str,
        attrs: list[tuple[str, str | None]],
    ) -> None:
        self.tags.append(tag.lower())
        for name, value in attrs:
            if name.lower() in {"href", "src"} and value:
                self.urls.append(value)


class OutputValidator:
    def __init__(self, *, min_length: int = 20) -> None:
        self.min_length = min_length

    def validate(
        self,
        request: AIRequest,
        response: AIResponse,
    ) -> ValidationResult:
        issues: list[str] = []
        content = response.content.strip()

        if not content:
            issues.append("出力が空です。")
            return ValidationResult(False, tuple(issues))

        if len(content) < self.min_length:
            issues.append("出力が想定より短すぎます。")

        if request.response_format == "json":
            self._validate_json(content, request, issues)
        elif request.response_format == "html":
            self._validate_html(content, request, issues)

        expected_language = request.metadata.get("expected_language")
        if expected_language == "ja" and not re.search(r"[ぁ-んァ-ヶ一-龠]", content):
            issues.append("日本語として判定できる文字が不足しています。")

        contamination_patterns = (
            "as an ai language model",
            "i cannot comply with that request",
            "here is the requested html",
        )
        lowered = content.lower()
        if any(pattern in lowered for pattern in contamination_patterns):
            issues.append("Provider固有の説明文が混入している可能性があります。")

        return ValidationResult(is_valid=not issues, issues=tuple(issues))

    def build_repair_request(
        self,
        original: AIRequest,
        invalid_response: AIResponse,
        validation: ValidationResult,
    ) -> AIRequest:
        # 前回出力を無制限に再送するとコストが増えるため、最小例では長さを制限します。
        previous = invalid_response.content[:6_000]
        issues = "\n".join(f"- {issue}" for issue in validation.issues)
        repair_input = f"""次の出力を要件に合うよう修正してください。

問題点:
{issues}

前回出力:
{previous}
"""
        return replace(original, user_input=repair_input)

    @staticmethod
    def _validate_json(
        content: str,
        request: AIRequest,
        issues: list[str],
    ) -> None:
        try:
            parsed = json.loads(content)
        except json.JSONDecodeError:
            issues.append("有効なJSONではありません。")
            return

        required_fields = request.metadata.get("required_fields", ())
        if isinstance(parsed, dict):
            for field_name in required_fields:
                if field_name not in parsed:
                    issues.append(f"必須フィールドがありません: {field_name}")

    @staticmethod
    def _validate_html(
        content: str,
        request: AIRequest,
        issues: list[str],
    ) -> None:
        parser = _HTMLInspectionParser()
        parser.feed(content)

        prohibited_tags = {"script", "object", "embed"}
        found_prohibited = sorted(prohibited_tags.intersection(parser.tags))
        if found_prohibited:
            issues.append(
                "禁止タグが含まれています: " + ", ".join(found_prohibited)
            )

        required_tags = request.metadata.get("required_html_tags", ())
        for tag in required_tags:
            if str(tag).lower() not in parser.tags:
                issues.append(f"必須HTMLタグがありません: {tag}")

        for url in parser.urls:
            parsed = urlparse(url)
            if parsed.scheme and parsed.scheme not in {"http", "https"}:
                issues.append(f"許可していないURLスキームです: {parsed.scheme}")

from __future__ import annotations

import asyncio
from dataclasses import replace
from datetime import datetime, timezone
from time import perf_counter
from uuid import uuid4

from app.domain.errors import (
    AIProviderError,
    AllProvidersFailedError,
    ProviderErrorKind,
)
from app.domain.models import (
    AIRequest,
    AIResponse,
    ProviderAttemptLog,
    UsageMetrics,
    ValidationResult,
)
from app.ports.ai_provider import AIProvider
from app.ports.provider_log_repository import ProviderLogRepository
from app.services.fallback_policy import FailureAction, FallbackPolicy
from app.services.output_validator import OutputValidator


class AIProviderRouter:
    def __init__(
        self,
        *,
        providers: list[AIProvider],
        fallback_policy: FallbackPolicy,
        output_validator: OutputValidator,
        log_repository: ProviderLogRepository,
        provider_timeout_seconds: float,
        total_timeout_seconds: float,
    ) -> None:
        if not providers:
            raise ValueError("少なくとも1つのProviderが必要です。")
        self._providers = providers
        self._policy = fallback_policy
        self._validator = output_validator
        self._logs = log_repository
        self._provider_timeout_seconds = provider_timeout_seconds
        self._total_timeout_seconds = total_timeout_seconds

    async def generate(self, request: AIRequest) -> AIResponse:
        errors: list[AIProviderError] = []
        operation_id = uuid4().hex

        try:
            async with asyncio.timeout(self._total_timeout_seconds):
                return await self._run_providers(request, operation_id, errors)
        except TimeoutError as exc:
            total_timeout_error = AIProviderError(
                provider="router",
                kind=ProviderErrorKind.TIMEOUT,
                message="AI処理全体の制限時間を超えました。",
            )
            errors.append(total_timeout_error)
            raise AllProvidersFailedError(errors) from exc

    async def _run_providers(
        self,
        original_request: AIRequest,
        operation_id: str,
        errors: list[AIProviderError],
    ) -> AIResponse:
        fallback_count = 0

        for provider_index, provider in enumerate(self._providers):
            has_next_provider = provider_index < len(self._providers) - 1
            working_request = original_request
            attempt_in_provider = 0

            while attempt_in_provider < self._policy.max_attempts_per_provider:
                attempt_in_provider += 1
                started_at = datetime.now(timezone.utc)
                started_clock = perf_counter()

                try:
                    async with asyncio.timeout(self._provider_timeout_seconds):
                        raw_response = await provider.generate(working_request)

                    validation = self._validator.validate(
                        original_request,
                        raw_response,
                    )
                    response = replace(
                        raw_response,
                        fallback_count=fallback_count,
                        validation_result=validation,
                    )

                    if validation.is_valid:
                        await self._save_success_log(
                            operation_id=operation_id,
                            provider=provider,
                            response=response,
                            attempt_in_provider=attempt_in_provider,
                            fallback_count=fallback_count,
                            started_at=started_at,
                            started_clock=started_clock,
                        )
                        return response

                    error = AIProviderError(
                        provider=provider.name,
                        kind=ProviderErrorKind.OUTPUT_VALIDATION,
                        message=" / ".join(validation.issues),
                        request_id=response.request_id,
                    )
                    errors.append(error)
                    await self._save_failure_log(
                        operation_id=operation_id,
                        provider=provider,
                        error=error,
                        attempt_in_provider=attempt_in_provider,
                        fallback_count=fallback_count,
                        started_at=started_at,
                        started_clock=started_clock,
                        validation=validation,
                    )
                    action = self._policy.decide(
                        error=error,
                        attempt_in_provider=attempt_in_provider,
                        has_next_provider=has_next_provider,
                    )
                    if action == FailureAction.RETRY_SAME:
                        working_request = self._validator.build_repair_request(
                            original_request,
                            response,
                            validation,
                        )
                        await asyncio.sleep(
                            self._policy.backoff_seconds(attempt_in_provider)
                        )
                        continue
                    if action == FailureAction.NEXT_PROVIDER:
                        break
                    raise AllProvidersFailedError(errors)

                except TimeoutError:
                    error = AIProviderError(
                        provider=provider.name,
                        kind=ProviderErrorKind.TIMEOUT,
                        message="Provider呼び出しがタイムアウトしました。",
                    )
                except AIProviderError as exc:
                    error = exc
                except Exception:
                    # アプリ側の未知のバグは別Providerで隠さず、そのまま上位へ返します。
                    raise

                errors.append(error)
                await self._save_failure_log(
                    operation_id=operation_id,
                    provider=provider,
                    error=error,
                    attempt_in_provider=attempt_in_provider,
                    fallback_count=fallback_count,
                    started_at=started_at,
                    started_clock=started_clock,
                    validation=None,
                )
                action = self._policy.decide(
                    error=error,
                    attempt_in_provider=attempt_in_provider,
                    has_next_provider=has_next_provider,
                )

                if action == FailureAction.RETRY_SAME:
                    await asyncio.sleep(
                        self._policy.backoff_seconds(attempt_in_provider)
                    )
                    continue
                if action == FailureAction.NEXT_PROVIDER:
                    break
                raise AllProvidersFailedError(errors)

            if has_next_provider:
                fallback_count += 1

        raise AllProvidersFailedError(errors)

    async def _save_success_log(
        self,
        *,
        operation_id: str,
        provider: AIProvider,
        response: AIResponse,
        attempt_in_provider: int,
        fallback_count: int,
        started_at: datetime,
        started_clock: float,
    ) -> None:
        await self._logs.save(
            ProviderAttemptLog(
                operation_id=operation_id,
                provider=provider.name,
                model=provider.model,
                attempt_in_provider=attempt_in_provider,
                fallback_count=fallback_count,
                started_at=started_at,
                ended_at=datetime.now(timezone.utc),
                latency_ms=int((perf_counter() - started_clock) * 1_000),
                success=True,
                request_id=response.request_id,
                input_tokens=response.usage.input_tokens,
                output_tokens=response.usage.output_tokens,
                validation_succeeded=True,
            )
        )

    async def _save_failure_log(
        self,
        *,
        operation_id: str,
        provider: AIProvider,
        error: AIProviderError,
        attempt_in_provider: int,
        fallback_count: int,
        started_at: datetime,
        started_clock: float,
        validation: ValidationResult | None,
    ) -> None:
        await self._logs.save(
            ProviderAttemptLog(
                operation_id=operation_id,
                provider=provider.name,
                model=provider.model,
                attempt_in_provider=attempt_in_provider,
                fallback_count=fallback_count,
                started_at=started_at,
                ended_at=datetime.now(timezone.utc),
                latency_ms=int((perf_counter() - started_clock) * 1_000),
                success=False,
                error_kind=error.kind.value,
                provider_error_code=error.provider_code,
                request_id=error.request_id,
                validation_succeeded=(
                    validation.is_valid if validation is not None else None
                ),
            )
        )

from __future__ import annotations

import asyncio

from app.domain.models import ProviderAttemptLog


class InMemoryProviderLogRepository:
    def __init__(self) -> None:
        self._items: list[ProviderAttemptLog] = []
        self._lock = asyncio.Lock()

    async def save(self, log: ProviderAttemptLog) -> None:
        async with self._lock:
            self._items.append(log)

    async def list_all(self) -> tuple[ProviderAttemptLog, ...]:
        async with self._lock:
            return tuple(self._items)

from app.domain.models import AIRequest, AIResponse
from app.services.provider_router import AIProviderRouter


class GenerateContentUseCase:
    def __init__(self, router: AIProviderRouter) -> None:
        self._router = router

    async def execute(self, request: AIRequest) -> AIResponse:
        return await self._router.generate(request)

from functools import lru_cache

from pydantic import SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    openai_api_key: SecretStr | None = None
    anthropic_api_key: SecretStr | None = None
    gemini_api_key: SecretStr | None = None

    openai_model: str | None = None
    anthropic_model: str | None = None
    gemini_model: str | None = None

    provider_order: str = "openai,anthropic,gemini"
    provider_timeout_seconds: float = 20.0
    total_timeout_seconds: float = 45.0
    max_attempts_per_provider: int = 2

    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        extra="ignore",
    )


@lru_cache
def get_settings() -> Settings:
    return Settings()

from functools import lru_cache

from app.infrastructure.logging_repository import InMemoryProviderLogRepository
from app.ports.ai_provider import AIProvider
from app.providers.anthropic_provider import AnthropicProvider
from app.providers.gemini_provider import GeminiProvider
from app.providers.openai_provider import OpenAIProvider
from app.services.fallback_policy import FallbackPolicy
from app.services.output_validator import OutputValidator
from app.services.provider_router import AIProviderRouter
from app.settings import get_settings
from app.usecases.generate_content import GenerateContentUseCase


@lru_cache
def get_log_repository() -> InMemoryProviderLogRepository:
    return InMemoryProviderLogRepository()


@lru_cache
def get_generate_content_use_case() -> GenerateContentUseCase:
    settings = get_settings()
    providers_by_name: dict[str, AIProvider] = {}

    if settings.openai_api_key and settings.openai_model:
        providers_by_name["openai"] = OpenAIProvider(
            api_key=settings.openai_api_key.get_secret_value(),
            model=settings.openai_model,
            timeout_seconds=settings.provider_timeout_seconds,
        )

    if settings.anthropic_api_key and settings.anthropic_model:
        providers_by_name["anthropic"] = AnthropicProvider(
            api_key=settings.anthropic_api_key.get_secret_value(),
            model=settings.anthropic_model,
            timeout_seconds=settings.provider_timeout_seconds,
        )

    if settings.gemini_api_key and settings.gemini_model:
        providers_by_name["gemini"] = GeminiProvider(
            api_key=settings.gemini_api_key.get_secret_value(),
            model=settings.gemini_model,
            timeout_seconds=settings.provider_timeout_seconds,
        )

    ordered_names = [
        name.strip()
        for name in settings.provider_order.split(",")
        if name.strip()
    ]
    providers = [
        providers_by_name[name]
        for name in ordered_names
        if name in providers_by_name
    ]
    if not providers:
        raise RuntimeError(
            "APIキーとモデル名が設定されたProviderがありません。"
        )

    router = AIProviderRouter(
        providers=providers,
        fallback_policy=FallbackPolicy(
            max_attempts_per_provider=settings.max_attempts_per_provider
        ),
        output_validator=OutputValidator(),
        log_repository=get_log_repository(),
        provider_timeout_seconds=settings.provider_timeout_seconds,
        total_timeout_seconds=settings.total_timeout_seconds,
    )
    return GenerateContentUseCase(router)

from typing import Annotated, Any, Literal

from fastapi import APIRouter, Depends, HTTPException, status
from pydantic import BaseModel, Field

from app.dependencies import get_generate_content_use_case
from app.domain.errors import AllProvidersFailedError
from app.domain.models import AIRequest
from app.usecases.generate_content import GenerateContentUseCase

router = APIRouter(prefix="/api/ai", tags=["ai"])


class GenerateRequestBody(BaseModel):
    system_instruction: str = ""
    user_input: str = Field(min_length=1, max_length=100_000)
    temperature: float = Field(default=0.2, ge=0.0, le=2.0)
    max_output_tokens: int = Field(default=1_500, ge=1, le=32_000)
    response_format: Literal["text", "json", "html"] = "text"
    metadata: dict[str, Any] = Field(default_factory=dict)


class GenerateResponseBody(BaseModel):
    content: str
    provider: str
    model: str
    finish_reason: str | None
    fallback_count: int
    latency_ms: int
    request_id: str | None
    validation_succeeded: bool | None


@router.post("/generate", response_model=GenerateResponseBody)
async def generate_content(
    body: GenerateRequestBody,
    use_case: Annotated[
        GenerateContentUseCase,
        Depends(get_generate_content_use_case),
    ],
) -> GenerateResponseBody:
    try:
        result = await use_case.execute(
            AIRequest(
                system_instruction=body.system_instruction,
                user_input=body.user_input,
                temperature=body.temperature,
                max_output_tokens=body.max_output_tokens,
                response_format=body.response_format,
                metadata=body.metadata,
            )
        )
    except AllProvidersFailedError as exc:
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail="現在AI生成を利用できません。時間を置くか手動処理へ切り替えてください。",
        ) from exc

    return GenerateResponseBody(
        content=result.content,
        provider=result.provider,
        model=result.model,
        finish_reason=result.finish_reason,
        fallback_count=result.fallback_count,
        latency_ms=result.latency_ms,
        request_id=result.request_id,
        validation_succeeded=(
            result.validation_result.is_valid
            if result.validation_result is not None
            else None
        ),
    )

from fastapi import FastAPI

from app.api.routes import router as ai_router

app = FastAPI(title="AI Provider Fallback Example")
app.include_router(ai_router)

from __future__ import annotations

from collections import deque

from app.domain.models import AIRequest, AIResponse, UsageMetrics


class FakeProvider:
    def __init__(
        self,
        *,
        name: str,
        model: str,
        outcomes: list[AIResponse | Exception],
    ) -> None:
        self.name = name
        self.model = model
        self._outcomes = deque(outcomes)
        self.call_count = 0

    async def generate(self, request: AIRequest) -> AIResponse:
        self.call_count += 1
        if not self._outcomes:
            raise AssertionError("FakeProviderの結果が不足しています。")
        outcome = self._outcomes.popleft()
        if isinstance(outcome, Exception):
            raise outcome
        return outcome


def make_response(
    *,
    provider: str,
    model: str,
    content: str = "十分な長さのテスト応答です。",
) -> AIResponse:
    return AIResponse(
        content=content,
        provider=provider,
        model=model,
        finish_reason="stop",
        usage=UsageMetrics(input_tokens=10, output_tokens=20),
        latency_ms=10,
        request_id="test-request-id",
    )

from __future__ import annotations

import asyncio

import pytest

from app.domain.errors import (
    AIProviderError,
    AllProvidersFailedError,
    ProviderErrorKind,
)
from app.domain.models import AIRequest
from app.infrastructure.logging_repository import InMemoryProviderLogRepository
from app.services.fallback_policy import FallbackPolicy
from app.services.output_validator import OutputValidator
from app.services.provider_router import AIProviderRouter
from app.tests.fakes import FakeProvider, make_response


def make_request() -> AIRequest:
    return AIRequest(
        system_instruction="日本語で回答してください。",
        user_input="Provider抽象化を説明してください。",
        metadata={"expected_language": "ja"},
    )


def make_router(
    providers: list[FakeProvider],
    *,
    provider_timeout_seconds: float = 0.2,
    total_timeout_seconds: float = 1.0,
) -> AIProviderRouter:
    return AIProviderRouter(
        providers=providers,
        fallback_policy=FallbackPolicy(
            max_attempts_per_provider=1,
            base_backoff_seconds=0.0,
        ),
        output_validator=OutputValidator(min_length=10),
        log_repository=InMemoryProviderLogRepository(),
        provider_timeout_seconds=provider_timeout_seconds,
        total_timeout_seconds=total_timeout_seconds,
    )


@pytest.mark.asyncio
async def test_primary_provider_succeeds() -> None:
    primary = FakeProvider(
        name="primary",
        model="model-a",
        outcomes=[make_response(provider="primary", model="model-a")],
    )
    secondary = FakeProvider(
        name="secondary",
        model="model-b",
        outcomes=[make_response(provider="secondary", model="model-b")],
    )

    result = await make_router([primary, secondary]).generate(make_request())

    assert result.provider == "primary"
    assert result.fallback_count == 0
    assert secondary.call_count == 0


@pytest.mark.asyncio
async def test_timeout_falls_back_to_secondary() -> None:
    primary = FakeProvider(
        name="primary",
        model="model-a",
        outcomes=[
            AIProviderError(
                provider="primary",
                kind=ProviderErrorKind.TIMEOUT,
                message="timeout",
            )
        ],
    )
    secondary = FakeProvider(
        name="secondary",
        model="model-b",
        outcomes=[make_response(provider="secondary", model="model-b")],
    )

    result = await make_router([primary, secondary]).generate(make_request())

    assert result.provider == "secondary"
    assert result.fallback_count == 1


@pytest.mark.asyncio
async def test_authentication_error_does_not_fallback() -> None:
    primary = FakeProvider(
        name="primary",
        model="model-a",
        outcomes=[
            AIProviderError(
                provider="primary",
                kind=ProviderErrorKind.AUTHENTICATION,
                message="invalid api key",
            )
        ],
    )
    secondary = FakeProvider(
        name="secondary",
        model="model-b",
        outcomes=[make_response(provider="secondary", model="model-b")],
    )

    with pytest.raises(AllProvidersFailedError):
        await make_router([primary, secondary]).generate(make_request())

    assert secondary.call_count == 0


@pytest.mark.asyncio
async def test_all_providers_fail() -> None:
    providers = [
        FakeProvider(
            name="primary",
            model="model-a",
            outcomes=[
                AIProviderError(
                    provider="primary",
                    kind=ProviderErrorKind.SERVER,
                    message="server error",
                )
            ],
        ),
        FakeProvider(
            name="secondary",
            model="model-b",
            outcomes=[
                AIProviderError(
                    provider="secondary",
                    kind=ProviderErrorKind.SERVER,
                    message="server error",
                )
            ],
        ),
    ]

    with pytest.raises(AllProvidersFailedError):
        await make_router(providers).generate(make_request())


@pytest.mark.asyncio
async def test_validation_failure_falls_back() -> None:
    primary = FakeProvider(
        name="primary",
        model="model-a",
        outcomes=[make_response(provider="primary", model="model-a", content="短い")],
    )
    secondary = FakeProvider(
        name="secondary",
        model="model-b",
        outcomes=[make_response(provider="secondary", model="model-b")],
    )

    result = await make_router([primary, secondary]).generate(make_request())

    assert result.provider == "secondary"
    assert result.validation_result is not None
    assert result.validation_result.is_valid is True


@pytest.mark.asyncio
async def test_total_timeout_is_enforced() -> None:
    class SlowProvider(FakeProvider):
        async def generate(self, request: AIRequest):  # type: ignore[override]
            await asyncio.sleep(0.2)
            return await super().generate(request)

    slow = SlowProvider(
        name="slow",
        model="model-slow",
        outcomes=[make_response(provider="slow", model="model-slow")],
    )

    with pytest.raises(AllProvidersFailedError):
        await make_router(
            [slow],
            provider_timeout_seconds=1.0,
            total_timeout_seconds=0.05,
        ).generate(make_request())