dmslanes/lanes

Lanes - team task boards with roles, history and an MCP endpoint for Claude and ChatGPT

• dodil-challenge
• dodil-challenge-boards

A remix template · 0 remixes · Remix dmslanes/lanes

app/infrastructure/models.py

"""Dodil Models: OpenAI-compatible chat and embeddings, signed with the app's service account."""
from __future__ import annotations

import os

import httpx

from app.domain.model import DomainError
from app.infrastructure.k3 import access_token

BASE_URL = os.environ.get("DODIL_MODELS_URL", "https://api.dodil.io/v1").rstrip("/")
CHAT_MODEL = os.environ.get("LANES_CHAT_MODEL", "kimi-k2.6")
# arctic ranks short card titles by topic; jina-embeddings-v4 scored "office chairs" as close to "invoice" as "unpaid bills"
EMBED_MODEL = os.environ.get("LANES_EMBED_MODEL", "arctic-embed-m-v2")


class DodilModels:
    def _post(self, path: str, body: dict, timeout: float) -> dict:
        try:
            r = httpx.post(f"{BASE_URL}{path}", json=body, timeout=timeout,
                           headers={"Authorization": f"Bearer {access_token()}"})
            r.raise_for_status()
            return r.json()
        except (httpx.HTTPError, KeyError) as e:
            raise DomainError(f"the AI did not answer ({type(e).__name__}); try again in a moment") from None

    def complete(self, system: str, prompt: str) -> str:
        d = self._post("/chat/completions", {"model": CHAT_MODEL, "max_tokens": 6000,  # kimi reasons first (~1.4k tokens); it rejects `temperature`
                                             "messages": [{"role": "system", "content": system},
                                                          {"role": "user", "content": prompt}]}, 90)
        return (d["choices"][0]["message"].get("content") or "").strip()

    def embed(self, texts: list[str]) -> list[list[float]]:
        d = self._post("/embeddings", {"model": EMBED_MODEL, "input": texts}, 120)
        return [row["embedding"] for row in d["data"]]