rfc-app/backend/app/providers.py

"""Multi-provider LLM abstraction — §18 carryover from the prototype.

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 per the prototype's
`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 — preserved from the prototype to keep the contract.
# ---------------------------------------------------------------------------

_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 — same contract as the prototype."""
    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 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)