rfc-app/backend/app/providers.py

"""Multi-provider LLM abstraction — §18 stack.

Each provider speaks a common interface — `send` and `send_streaming` —
so the chat layer in `chat.py` is provider-agnostic. Enabled providers
and their API keys are configured via env via the `ENABLED_MODELS`
contract; per §16 / §19.2, per-RFC model availability and credential
delegation are deferred until the topic is settled.
"""
from __future__ import annotations

from typing import Iterator


class BaseProvider:
    name: str = "base"
    display_name: str = "Base"

    def send(self, system: str, history: list[dict]) -> str:
        raise NotImplementedError

    def send_streaming(self, system: str, history: list[dict]) -> Iterator[str]:
        raise NotImplementedError


# ---------------------------------------------------------------------------
# Anthropic — Claude
# ---------------------------------------------------------------------------

class AnthropicProvider(BaseProvider):
    name = "claude"

    def __init__(self, api_key: str, model: str = "claude-sonnet-4-6", display_name: str = "Claude"):
        import anthropic
        self.client = anthropic.Anthropic(api_key=api_key)
        self.model = model
        self.display_name = display_name

    def send(self, system: str, history: list[dict]) -> str:
        response = self.client.messages.create(
            model=self.model,
            max_tokens=4096,
            system=system,
            messages=history,
        )
        return response.content[0].text

    def send_streaming(self, system: str, history: list[dict]) -> Iterator[str]:
        with self.client.messages.stream(
            model=self.model,
            max_tokens=4096,
            system=system,
            messages=history,
        ) as stream:
            for text in stream.text_stream:
                yield text


# ---------------------------------------------------------------------------
# Google — Gemini
# ---------------------------------------------------------------------------

class GeminiProvider(BaseProvider):
    name = "gemini"

    def __init__(self, api_key: str, model: str = "gemini-1.5-pro", display_name: str = "Gemini"):
        import google.generativeai as genai
        genai.configure(api_key=api_key)
        self._genai = genai
        self.model_name = model
        self.display_name = display_name

    def _build_model(self, system: str):
        return self._genai.GenerativeModel(model_name=self.model_name, system_instruction=system)

    def _convert_history(self, history: list[dict]) -> list[dict]:
        return [
            {"role": "user" if msg["role"] == "user" else "model", "parts": [msg["content"]]}
            for msg in history
        ]

    def send(self, system: str, history: list[dict]) -> str:
        model = self._build_model(system)
        prior = self._convert_history(history[:-1])
        chat = model.start_chat(history=prior)
        response = chat.send_message(history[-1]["content"])
        return response.text

    def send_streaming(self, system: str, history: list[dict]) -> Iterator[str]:
        model = self._build_model(system)
        prior = self._convert_history(history[:-1])
        chat = model.start_chat(history=prior)
        response = chat.send_message(history[-1]["content"], stream=True)
        for chunk in response:
            if chunk.text:
                yield chunk.text


# ---------------------------------------------------------------------------
# OpenAI-compatible — OpenAI, Copilot, or any compatible endpoint
# ---------------------------------------------------------------------------

class OpenAIProvider(BaseProvider):
    name = "openai"

    def __init__(self, api_key: str, model: str = "gpt-4o", base_url: str | None = None, display_name: str = "Copilot"):
        from openai import OpenAI
        self.client = OpenAI(api_key=api_key, base_url=base_url or "https://api.openai.com/v1")
        self.model = model
        self.display_name = display_name

    def _messages(self, system: str, history: list[dict]) -> list[dict]:
        return [{"role": "system", "content": system}] + [
            {"role": msg["role"], "content": msg["content"]} for msg in history
        ]

    def send(self, system: str, history: list[dict]) -> str:
        response = self.client.chat.completions.create(
            model=self.model, max_tokens=4096, messages=self._messages(system, history)
        )
        return response.choices[0].message.content

    def send_streaming(self, system: str, history: list[dict]) -> Iterator[str]:
        stream = self.client.chat.completions.create(
            model=self.model, max_tokens=4096, messages=self._messages(system, history), stream=True
        )
        for chunk in stream:
            delta = chunk.choices[0].delta.content
            if delta:
                yield delta


# ---------------------------------------------------------------------------
# Variants and factory.
# ---------------------------------------------------------------------------

_CLAUDE_VARIANTS: dict[str, tuple[str, str]] = {
    "claude":         ("claude-sonnet-4-6",         "Claude"),
    "claude-sonnet":  ("claude-sonnet-4-6",         "Claude Sonnet"),
    "claude-opus":    ("claude-opus-4-6",           "Claude Opus"),
    "claude-haiku":   ("claude-haiku-4-5-20251001", "Claude Haiku"),
}

_GEMINI_VARIANTS: dict[str, tuple[str, str]] = {
    "gemini":         ("gemini-1.5-pro",   "Gemini"),
    "gemini-pro":     ("gemini-1.5-pro",   "Gemini Pro"),
    "gemini-flash":   ("gemini-1.5-flash", "Gemini Flash"),
    "gemini-2-flash": ("gemini-2.0-flash", "Gemini 2 Flash"),
}


def load_providers(env: dict) -> dict[str, BaseProvider]:
    """Instantiate enabled providers from env."""
    enabled = [m.strip() for m in env.get("ENABLED_MODELS", "claude").split(",") if m.strip()]
    providers: dict[str, BaseProvider] = {}

    anthropic_key = env.get("ANTHROPIC_API_KEY") or ""
    google_key = env.get("GOOGLE_API_KEY") or ""
    openai_key = env.get("OPENAI_API_KEY") or ""

    for key in enabled:
        prefix = key.upper().replace("-", "_")
        if key in _CLAUDE_VARIANTS and anthropic_key:
            default_model, default_name = _CLAUDE_VARIANTS[key]
            providers[key] = AnthropicProvider(
                api_key=anthropic_key,
                model=env.get(f"{prefix}_MODEL", default_model),
                display_name=env.get(f"{prefix}_DISPLAY_NAME", default_name),
            )
        elif key in _GEMINI_VARIANTS and google_key:
            default_model, default_name = _GEMINI_VARIANTS[key]
            providers[key] = GeminiProvider(
                api_key=google_key,
                model=env.get(f"{prefix}_MODEL", default_model),
                display_name=env.get(f"{prefix}_DISPLAY_NAME", default_name),
            )
        elif key == "openai" and openai_key:
            providers["openai"] = OpenAIProvider(
                api_key=openai_key,
                model=env.get("OPENAI_MODEL", "gpt-4o"),
                base_url=env.get("OPENAI_BASE_URL"),
                display_name=env.get("OPENAI_DISPLAY_NAME", "Copilot"),
            )

    return providers


def construct_haiku(api_key: str) -> AnthropicProvider:
    """A dedicated Claude Haiku provider, independent of the
    `ENABLED_MODELS` chat-picker universe.

    The §9.1 tag-suggestion surface (roadmap #27) always wants the
    cheap + fast model regardless of which models the operator surfaces
    in the §8.12 picker, so it constructs Haiku directly from the
    operator's Anthropic key rather than going through `load_providers`.
    The model id is sourced from the same `_CLAUDE_VARIANTS` table the
    picker uses, so a model-string bump lands in one place.
    """
    model, name = _CLAUDE_VARIANTS["claude-haiku"]
    return AnthropicProvider(api_key=api_key, model=model, display_name=name)


def load_from_config(config) -> dict[str, BaseProvider]:
    """Convenience adapter so callers can pass our Config dataclass directly."""
    env = {
        "ENABLED_MODELS": ",".join(config.enabled_models),
        "ANTHROPIC_API_KEY": config.anthropic_api_key,
        "GOOGLE_API_KEY": config.google_api_key,
        "OPENAI_API_KEY": config.openai_api_key,
    }
    return load_providers(env)


# ---------------------------------------------------------------------------
# §6.7 helpers — per-RFC funder credential plumbing.
#
# A picker key (the operator-enabled identifier like "claude" or
# "gemini-flash") names exactly one provider family (Anthropic, Google,
# OpenAI). The funder registers a key per family; the resolver in
# funder.py picks the right family for a requested picker key, then
# constructs a fresh provider instance keyed on the funder's API key.
# ---------------------------------------------------------------------------

FUNDER_PROVIDER_FAMILIES = ("anthropic", "google", "openai")


def provider_family_for_picker_key(picker_key: str) -> str | None:
    """Map a picker key to its provider-family name per §6.7.

    Returns 'anthropic' / 'google' / 'openai' or None for keys outside
    the known variant tables.
    """
    if picker_key in _CLAUDE_VARIANTS:
        return "anthropic"
    if picker_key in _GEMINI_VARIANTS:
        return "google"
    if picker_key == "openai":
        return "openai"
    return None


def picker_keys_for_family(family: str, enabled_picker_keys: list[str]) -> list[str]:
    """Subset of `enabled_picker_keys` served by `family`. Preserves the
    operator's enabled order so the picker reads stably."""
    return [k for k in enabled_picker_keys if provider_family_for_picker_key(k) == family]


def construct_for_funder(picker_key: str, api_key: str) -> BaseProvider | None:
    """Instantiate a provider for a funder-supplied API key per §6.7.

    Mirrors the variant table `load_providers` uses, without the env
    contract. Returns None for picker keys outside the known families —
    the funder cannot serve them and the caller treats the (slug,
    picker_key) request as unavailable.
    """
    if picker_key in _CLAUDE_VARIANTS:
        default_model, default_name = _CLAUDE_VARIANTS[picker_key]
        return AnthropicProvider(api_key=api_key, model=default_model, display_name=default_name)
    if picker_key in _GEMINI_VARIANTS:
        default_model, default_name = _GEMINI_VARIANTS[picker_key]
        return GeminiProvider(api_key=api_key, model=default_model, display_name=default_name)
    if picker_key == "openai":
        return OpenAIProvider(api_key=api_key)
    return None