Spaces:
Paused
Paused
| """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) | |