Spaces:
Paused
Paused
File size: 4,031 Bytes
536e671 8501103 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 | """Free models through OpenCode Zen, used as a chat model (for debugging repairs) without OpenRouter.
`ZenLLM` speaks the same `complete()` interface as the OpenRouter client, but runs `opencode run` with only the
OpenCode Zen provider configured and every tool denied: it reads the conversation and answers, nothing else.
`WithFreeModels` sends calls for `opencode/...` models to it and everything else to the paid client.
"""
from __future__ import annotations
import time
from collections.abc import Sequence
from .llm import LLM, Completion, LLMError, Message
from .opencode_runner import OpenCodeError, OpenCodeWorkspace, find_opencode
from .routing import is_free
def zen_config(settings, model_id: str) -> dict:
"""An OpenCode config with only the Zen provider and one free model, and no tools."""
provider, _, model = model_id.partition("/")
options = {"apiKey": "{env:ZEN_API_KEY}"} if settings.zen_api_key else {}
return {
"$schema": "https://opencode.ai/config.json",
"model": model_id,
"provider": {provider: {"options": options, "models": {model: {"name": model}}}},
"permission": {
"edit": "deny",
"bash": "deny",
"webfetch": "deny",
"websearch": "deny",
"external_directory": "deny",
},
"autoupdate": False,
"share": "disabled",
}
def zen_text(events: list[dict]) -> str:
texts = []
for e in events:
part = e.get("part") or {}
if (e.get("type") == "text" or part.get("type") == "text") and isinstance(part.get("text"), str):
texts.append(part["text"])
return "\n".join(texts).strip()
def _transcript(messages: Sequence[Message]) -> str:
"""OpenCode takes one message per run, so the conversation is passed as a labelled transcript."""
parts = []
for m in messages:
label = {"system": "INSTRUCTIONS", "user": "USER", "assistant": "YOUR EARLIER ANSWER"}.get(m.role, m.role)
parts.append(f"### {label}\n{m.content}")
parts.append("### Reply to the last USER message now, following the INSTRUCTIONS exactly.")
return "\n\n".join(parts)
class ZenLLM:
"""A free OpenCode Zen model as a chat model. Raises LLMError on any failure, so callers can fall back."""
def __init__(self, settings, model: str | None = None):
self.settings = settings
self.model = model or settings.debug_model
def complete(self, messages: Sequence[Message], temperature: float = 0.2, model: str | None = None) -> Completion:
model_id = model or self.model
if not find_opencode(self.settings):
raise LLMError("OpenCode is not installed, so the free model is unavailable")
provider, _, name = model_id.partition("/")
start = time.monotonic()
try:
with OpenCodeWorkspace(self.settings, config=zen_config(self.settings, model_id)) as ws:
run = ws.run(_transcript(messages), context="", model=name, provider=provider)
except OpenCodeError as e:
raise LLMError(f"free model: {e}") from e
if run.error:
raise LLMError(f"free model: {run.error}")
text = zen_text(run.events)
if not text:
raise LLMError("free model returned no text")
usage = {**run.usage(), "cost": 0.0}
return Completion(text, model_id, int((time.monotonic() - start) * 1000), usage)
class WithFreeModels:
"""Routes `opencode/...` models to the free client and everything else to the paid one."""
def __init__(self, paid: LLM, free: LLM):
self.paid, self.free = paid, free
self.model = paid.model
def complete(self, messages: Sequence[Message], temperature: float = 0.2, model: str | None = None) -> Completion:
if is_free(model):
return self.free.complete(messages, temperature, model=model)
return self.paid.complete(messages, temperature, model=model) if model else self.paid.complete(messages, temperature)
|