Add LLM client interface with LiteLLM adapter (A5/A11)
CI / checks (pull_request) Successful in 10s
CI / checks (push) Successful in 9s

PR 2 of the steps-3/4 plan.

- llm.py: LLMClient protocol — complete(events, tools, model) ->
  canonical events — present from the first real LLM call.
  LiteLLMClient renders requests via the A10 renderer, targets the
  proxy's OpenAI-compatible route, and maps provider tool-call IDs
  to locally-minted canonical IDs at the boundary
- RoleMap (A11): executor/planner/delegate/consultant resolved from
  env; pinned() returns the mapping to record at task start
- Config strictly from env: NISSE_LLM_BASE_URL, NISSE_LLM_TOKEN;
  envfile.py is a minimal .env loader where real env wins; .env now
  gitignored with a committed .env.example documenting all variables
  (Matrix ones included for the upcoming transport PR)
- Live smoke test against the proxy is integration-marked and skips
  without credentials — CI stays credential-free per operator
  decision; the test runs locally once .env exists
- httpx added as nisse-dev's HTTP client

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
dongarandClaude Fable 5 committed 2026-09-06 20:57:51 +00:00
1 parent e185bb7c88
commit 0c2021fced
8 files changed
+462 -2

No files matched your search

+144
View File
@@ -0,0 +1,144 @@
"""LLM access (spec A5, A11).
One client interface, present from the first real LLM call. All traffic goes
to the LiteLLM proxy — the VM's single LLM destination; the agent holds one
LiteLLM token and no vendor credentials. Config comes from the environment
(``NISSE_LLM_BASE_URL``, ``NISSE_LLM_TOKEN``), nothing hardcoded.
Model identity is expressed as roles (A11), resolved to concrete model strings
from env config; the resolved mapping is pinned at task start so a mid-task
config change cannot split a task across two planners.
"""
import json
import os
from collections.abc import Mapping
from dataclasses import asdict, dataclass
from typing import Any, ClassVar, Protocol
import httpx
from nisse_dev.canonical import (
AssistantText,
CanonicalEvent,
StoredEvent,
ToolCall,
mint_call_id,
)
from nisse_dev.render import OpenAIChatRenderer
class LLMError(RuntimeError):
pass
class LLMClient(Protocol):
def complete(
self,
events: list[StoredEvent],
tools: list[dict[str, Any]] | None,
model: str,
system: str | None = None,
) -> list[CanonicalEvent]: ...
@dataclass(frozen=True)
class RoleMap:
executor: str
planner: str
delegate: str
consultant: str
_ENV_VARS: ClassVar[dict[str, str]] = {
"executor": "NISSE_ROLE_EXECUTOR",
"planner": "NISSE_ROLE_PLANNER",
"delegate": "NISSE_ROLE_DELEGATE",
"consultant": "NISSE_ROLE_CONSULTANT",
}
@classmethod
def from_env(cls, env: Mapping[str, str] = os.environ) -> "RoleMap":
missing = [var for var in cls._ENV_VARS.values() if not env.get(var)]
if missing:
raise LLMError(f"role mapping incomplete; set: {', '.join(missing)}")
return cls(**{role: env[var] for role, var in cls._ENV_VARS.items()})
def resolve(self, role: str) -> str:
if role not in self._ENV_VARS:
raise LLMError(f"unknown role: {role!r}")
return str(getattr(self, role))
def pinned(self) -> dict[str, str]:
"""The full resolved mapping, to record at task start (A11)."""
return asdict(self)
class LiteLLMClient:
def __init__(
self,
base_url: str,
token: str,
*,
timeout_s: float = 120.0,
transport: httpx.BaseTransport | None = None,
) -> None:
self._http = httpx.Client(
base_url=base_url.rstrip("/"),
headers={"Authorization": f"Bearer {token}"},
timeout=timeout_s,
transport=transport,
)
self._renderer = OpenAIChatRenderer()
@classmethod
def from_env(cls, env: Mapping[str, str] = os.environ) -> "LiteLLMClient":
base_url = env.get("NISSE_LLM_BASE_URL")
token = env.get("NISSE_LLM_TOKEN")
if not base_url or not token:
raise LLMError("set NISSE_LLM_BASE_URL and NISSE_LLM_TOKEN")
return cls(base_url, token)
def complete(
self,
events: list[StoredEvent],
tools: list[dict[str, Any]] | None,
model: str,
system: str | None = None,
) -> list[CanonicalEvent]:
body = self._renderer.render(events, tools, model, system=system)
try:
response = self._http.post("/chat/completions", json=body)
except httpx.HTTPError as exc:
raise LLMError(f"LLM request failed: {exc}") from exc
if response.status_code != 200:
raise LLMError(
f"LLM request returned {response.status_code}: {response.text[:500]}"
)
return self._parse(response.json(), requested_model=model)
def _parse(
self, data: dict[str, Any], requested_model: str
) -> list[CanonicalEvent]:
try:
message = data["choices"][0]["message"]
except (KeyError, IndexError) as exc:
raise LLMError(f"malformed completion response: {exc}") from exc
produced_by = data.get("model") or requested_model
out: list[CanonicalEvent] = []
if message.get("content"):
out.append(AssistantText(text=message["content"], model=produced_by))
# Provider tool-call IDs are mapped at the boundary (A10): each call
# gets a locally-minted canonical ID; the provider's is dropped, since
# every future request is rendered wholly from the canonical log.
for call in message.get("tool_calls") or []:
out.append(
ToolCall(
call_id=mint_call_id(),
tool=call["function"]["name"],
arguments=json.loads(call["function"]["arguments"] or "{}"),
)
)
return out
def close(self) -> None:
self._http.close()