llm_client.py

#!/usr/bin/env python3
"""OpenAI-compatible chat client for model-to-model experiments."""

from __future__ import annotations

import json
import os
import subprocess
import urllib.error
import urllib.request


class LLMError(RuntimeError):
    pass


def _keychain_secret(service: str) -> str:
    result = subprocess.run(
        [
            "/usr/bin/security",
            "find-generic-password",
            "-a",
            os.environ.get("USER", ""),
            "-s",
            service,
            "-w",
        ],
        text=True,
        capture_output=True,
    )
    if result.returncode != 0:
        raise LLMError(f"Cannot read Keychain service {service}: {result.stderr.strip()}")
    secret = result.stdout.strip()
    if not secret:
        raise LLMError(f"Keychain service {service} is empty")
    return secret


def api_key(model_config: dict) -> str:
    env_name = model_config.get("api_key_env")
    if env_name and os.environ.get(env_name):
        return os.environ[env_name]

    keychain_service = model_config.get("keychain_service")
    if keychain_service:
        return _keychain_secret(keychain_service)

    raise LLMError(f"No api_key_env or keychain_service configured for {model_config.get('name')}")


def chat(model_config: dict, messages: list[dict], *, temperature: float = 0.3) -> str:
    base_url = (model_config.get("base_url") or "").rstrip("/")
    model = model_config.get("model")
    if not base_url or not model:
        raise LLMError(f"Missing base_url/model for {model_config.get('name')}")

    payload = {
        "model": model,
        "messages": messages,
        "temperature": temperature,
        "max_tokens": model_config.get("max_tokens", 900),
    }
    request = urllib.request.Request(
        f"{base_url}/v1/chat/completions",
        data=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
        headers={
            "Authorization": f"Bearer {api_key(model_config)}",
            "Content-Type": "application/json",
        },
        method="POST",
    )

    try:
        with urllib.request.urlopen(request, timeout=model_config.get("timeout_seconds", 60)) as response:
            data = json.loads(response.read().decode("utf-8"))
    except urllib.error.HTTPError as exc:
        body = exc.read().decode("utf-8", errors="replace")
        raise LLMError(f"HTTP {exc.code} from {model_config.get('name')}: {body[:500]}") from exc
    except urllib.error.URLError as exc:
        raise LLMError(f"Request failed for {model_config.get('name')}: {exc}") from exc

    text = data.get("choices", [{}])[0].get("message", {}).get("content") or ""
    if not text.strip():
        raise LLMError(f"Empty response from {model_config.get('name')}")
    return text.strip()