app/council/llm.py

"""LLM client abstraction for the council.

Specialists and the moderator only depend on `LLMClient.complete(messages,
extra_body, max_tokens, temperature) -> str`. Two concrete clients:

- `OpenAILLMClient` — wraps the OpenAI SDK (which speaks the vLLM /
  HF-Endpoints OpenAI-compatible API).
- `MockLLMClient` — deterministic, no network, used by tests + `--mock`.
"""
from __future__ import annotations

import hashlib
import json
import os
import re
from dataclasses import dataclass
from typing import Any, Iterable, Optional, Protocol


class LLMClient(Protocol):
    def complete(
        self,
        messages: list[dict],
        *,
        max_tokens: int = 256,
        temperature: float = 0.2,
        extra_body: dict | None = None,
    ) -> str: ...


@dataclass
class OpenAILLMClient:
    """OpenAI-compatible client. Lazily creates the underlying SDK client."""
    base_url: str
    api_key: str
    model: str
    _sdk: Any = None

    def __post_init__(self):
        from openai import OpenAI  # imported here so test envs without `openai` still work
        self._sdk = OpenAI(base_url=self.base_url, api_key=self.api_key)

    def complete(
        self,
        messages: list[dict],
        *,
        max_tokens: int = 256,
        temperature: float = 0.2,
        extra_body: dict | None = None,
    ) -> str:
        kwargs: dict = {
            "model": self.model,
            "messages": messages,
            "max_tokens": max_tokens,
            "temperature": temperature,
        }
        if extra_body:
            kwargs["extra_body"] = extra_body
        resp = self._sdk.chat.completions.create(**kwargs)
        return (resp.choices[0].message.content or "").strip()


class MockLLMClient:
    """Deterministic stand-in for offline runs. Routes by system prompt
    keyword so each specialist gets a plausibly-shaped response."""

    SIGNAL_BIAS = {
        "chart-pattern": +0.15,
        "seasonal":      +0.05,
        "indicator":     -0.05,
        "news":           0.00,
        "forum":          0.00,
        "peer-commodity": +0.05,
        "geopolitics":   -0.10,
        "moderator":      0.00,
    }

    def complete(
        self,
        messages: list[dict],
        *,
        max_tokens: int = 256,
        temperature: float = 0.2,
        extra_body: dict | None = None,
    ) -> str:
        # Find the role keyword from the system prompt
        sys = next((m["content"] for m in messages if m.get("role") == "system"), "")
        role = "moderator"
        for k in self.SIGNAL_BIAS:
            if k in sys.lower():
                role = k
                break

        # Hash user content for deterministic small jitter
        user_blob = json.dumps(
            [m for m in messages if m.get("role") == "user"], default=str, sort_keys=True
        )
        h = int(hashlib.md5(user_blob.encode()).hexdigest(), 16)
        jitter = (((h % 1000) / 1000.0) - 0.5) * 0.4   # in [-0.2, +0.2]
        bias = self.SIGNAL_BIAS[role]
        signal = max(-1.0, min(1.0, bias + jitter))
        confidence = 0.4 + ((h // 1000) % 600) / 1000.0  # in [0.4, 1.0]

        if role == "moderator":
            # Moderator emits a TradeAction
            avg_signal = signal  # downstream code re-computes from votes anyway
            if avg_signal > 0.2:
                side, qty = "buy", 5
            elif avg_signal < -0.2:
                side, qty = "sell", 5
            else:
                side, qty = "hold", 0
            return json.dumps({
                "side": side,
                "quantity": qty,
                "rationale": f"mock moderator: avg_signal={avg_signal:.2f}",
            })

        return json.dumps({
            "signal": round(signal, 3),
            "confidence": round(confidence, 3),
            "rationale": f"mock {role} read of the situation",
        })


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def parse_json_object(text: str) -> dict:
    """Parse the first balanced JSON object from a model response."""
    cleaned = re.sub(r"```(?:json)?\s*", "", text).strip().rstrip("`")
    start = cleaned.find("{")
    if start < 0:
        return {}
    depth = 0
    for i in range(start, len(cleaned)):
        if cleaned[i] == "{":
            depth += 1
        elif cleaned[i] == "}":
            depth -= 1
        if depth == 0:
            try:
                return json.loads(cleaned[start : i + 1])
            except json.JSONDecodeError:
                return {}
    return {}


def build_openai_client_from_env() -> OpenAILLMClient:
    api_key = (
        os.getenv("HF_TOKEN")
        or os.getenv("API_KEY")
        or os.getenv("OPENAI_API_KEY")
        or "sk-local"
    )
    base_url = os.getenv("API_BASE_URL") or "http://localhost:8000/v1"
    model = os.getenv("MODEL_NAME") or "google/gemma-4-E4B-it"
    return OpenAILLMClient(base_url=base_url, api_key=api_key, model=model)


# ---------------------------------------------------------------------------
# Transformers-direct client (for Colab / single-process inference)
# ---------------------------------------------------------------------------
@dataclass
class TransformersLLMClient:
    """Calls a loaded HF transformers model in-process. Used by Colab/local
    when there's no separate vLLM server. Multimodal-aware: image_url parts
    are decoded back to PIL.Image and passed via the AutoProcessor.

    Construct with `from_pretrained(model_id, ...)` for a one-liner setup,
    or pass an already-loaded `model` + `processor` to share weights with
    a TRL trainer (so the GPU only holds one copy of Gemma).
    """
    model: Any
    processor: Any
    moderator_lora: Optional[str] = None  # PEFT adapter name when active

    @classmethod
    def from_pretrained(
        cls,
        model_id: str = "google/gemma-4-E4B-it",
        load_in_4bit: bool = True,
        device_map: Any = None,
        max_memory: dict | None = None,
    ) -> "TransformersLLMClient":
        """Load Gemma into memory.

        Defaults that work on a single T4 (16 GB):
          - 4-bit nf4 with double-quant
          - fp16 compute (T4 has no bf16 silicon)
          - device_map={"": 0} — force everything to GPU 0; transformers'
            "auto" sometimes spills the vision tower to CPU on tight cards
            and then refuses to load with `llm_int8_enable_fp32_cpu_offload`
            unless given an explicit dict device_map.

        Pass device_map="auto" + max_memory={0: "14GiB", "cpu": "30GiB"}
        if you need offload (e.g. running on a 12GB card).
        """
        import torch
        from transformers import AutoModelForCausalLM, AutoProcessor

        # Diagnostics so the user sees what they're working with
        if torch.cuda.is_available():
            free, total = torch.cuda.mem_get_info()
            print(
                f"[load] GPU={torch.cuda.get_device_name(0)} "
                f"free={free / 1024**3:.2f} GiB / total={total / 1024**3:.2f} GiB "
                f"bf16={torch.cuda.is_bf16_supported()}"
            )
        else:
            print("[load] no CUDA device — model will load on CPU (very slow)")

        # T4 (Turing) doesn't have bf16 hardware — fall back to fp16.
        compute_dtype = (
            "bfloat16"
            if (torch.cuda.is_available() and torch.cuda.is_bf16_supported())
            else "float16"
        )

        if device_map is None:
            device_map = {"": 0} if torch.cuda.is_available() else "cpu"

        kwargs: dict = {"dtype": compute_dtype, "device_map": device_map}
        if max_memory is not None:
            kwargs["max_memory"] = max_memory

        if load_in_4bit:
            from transformers import BitsAndBytesConfig
            kwargs["quantization_config"] = BitsAndBytesConfig(
                load_in_4bit=True,
                bnb_4bit_compute_dtype=compute_dtype,
                bnb_4bit_quant_type="nf4",
                bnb_4bit_use_double_quant=True,
                # Allows non-quantizable modules (vision tower, embeddings)
                # to live on CPU if device_map decides to put them there.
                llm_int8_enable_fp32_cpu_offload=True,
            )
        processor = AutoProcessor.from_pretrained(model_id)
        model = AutoModelForCausalLM.from_pretrained(model_id, **kwargs)
        model.eval()
        if torch.cuda.is_available():
            free, total = torch.cuda.mem_get_info()
            print(
                f"[load] done. GPU free={free / 1024**3:.2f} GiB / "
                f"total={total / 1024**3:.2f} GiB"
            )
        return cls(model=model, processor=processor)

    def complete(
        self,
        messages: list[dict],
        *,
        max_tokens: int = 256,
        temperature: float = 0.2,
        extra_body: dict | None = None,
    ) -> str:
        # Decode any embedded data-URL images back into PIL.Image objects
        # the processor expects.
        prepped = self._prep_messages(messages)
        inputs = self.processor.apply_chat_template(
            prepped,
            add_generation_prompt=True,
            tokenize=True,
            return_dict=True,
            return_tensors="pt",
        ).to(self.model.device)

        gen_kwargs: dict = {
            "max_new_tokens": max_tokens,
            "do_sample": temperature > 0.0,
        }
        if temperature > 0.0:
            gen_kwargs["temperature"] = temperature

        # Activate / deactivate the moderator LoRA if requested
        adapter_name = (
            (extra_body or {}).get("lora_request", {}).get("name")
            if extra_body
            else None
        )
        active_adapter = self._set_adapter(adapter_name)

        try:
            import torch
            with torch.no_grad():
                out = self.model.generate(**inputs, **gen_kwargs)
        finally:
            self._restore_adapter(active_adapter)

        gen_only = out[:, inputs["input_ids"].shape[1] :]
        return self.processor.batch_decode(gen_only, skip_special_tokens=True)[0].strip()

    # ----------------------------------------------------------------- helpers
    @staticmethod
    def _prep_messages(messages: list[dict]) -> list[dict]:
        """Normalize OpenAI-style messages for HF AutoProcessor chat templates.

        Two transforms:
        1. Plain string `content` is wrapped as `[{"type": "text", "text": ...}]`
           because multimodal processors iterate `content` and break on strings.
        2. `{"type": "image_url", "image_url": {"url": "data:image/..."}}` parts
           are decoded back to PIL.Image objects in `{"type": "image", "image": img}`
           form expected by Gemma's processor.
        """
        import base64
        from io import BytesIO

        from PIL import Image

        out = []
        for m in messages:
            content = m.get("content")
            if isinstance(content, str):
                # Plain string -> single text part
                out.append({**m, "content": [{"type": "text", "text": content}]})
                continue
            if isinstance(content, list):
                new_parts = []
                for part in content:
                    if isinstance(part, dict) and part.get("type") == "image_url":
                        url = part["image_url"]["url"]
                        if url.startswith("data:image"):
                            b64 = url.split(",", 1)[1]
                            img = Image.open(BytesIO(base64.b64decode(b64))).convert("RGB")
                            new_parts.append({"type": "image", "image": img})
                            continue
                    new_parts.append(part)
                out.append({**m, "content": new_parts})
            else:
                out.append(m)
        return out

    def _set_adapter(self, name: str | None):
        if name is None:
            return None
        peft_model = getattr(self.model, "peft_config", None)
        if not peft_model:
            return None
        try:
            current = self.model.active_adapter
            self.model.set_adapter(name)
            return current
        except Exception:
            return None

    def _restore_adapter(self, prev):
        if prev is None:
            return
        try:
            self.model.set_adapter(prev)
        except Exception:
            pass


def encode_image_url(path: str) -> str:
    """Encode a local PNG as a data URL for OpenAI-style image messages."""
    import base64
    from pathlib import Path

    p = Path(path)
    if not p.exists():
        return ""
    return "data:image/png;base64," + base64.b64encode(p.read_bytes()).decode()