OpenRouter Model
OpenRouter Model class
Full source code
import json
import logging
import os
from dataclasses import asdict, dataclass, field
from typing import Any, Literal
import requests
from tenacity import (
before_sleep_log,
retry,
retry_if_not_exception_type,
stop_after_attempt,
wait_exponential,
)
from minisweagent.models import GLOBAL_MODEL_STATS
from minisweagent.models.utils.cache_control import set_cache_control
logger = logging.getLogger("openrouter_model")
@dataclass
class OpenRouterModelConfig:
model_name: str
model_kwargs: dict[str, Any] = field(default_factory=dict)
set_cache_control: Literal["default_end"] | None = None
"""Set explicit cache control markers, for example for Anthropic models"""
cost_tracking: Literal["default", "ignore_errors"] = os.getenv("MSWEA_COST_TRACKING", "default")
"""Cost tracking mode for this model. Can be "default" or "ignore_errors" (ignore errors/missing cost info)"""
class OpenRouterAPIError(Exception):
"""Custom exception for OpenRouter API errors."""
pass
class OpenRouterAuthenticationError(Exception):
"""Custom exception for OpenRouter authentication errors."""
pass
class OpenRouterRateLimitError(Exception):
"""Custom exception for OpenRouter rate limit errors."""
pass
class OpenRouterModel:
def __init__(self, **kwargs):
self.config = OpenRouterModelConfig(**kwargs)
self.cost = 0.0
self.n_calls = 0
self._api_url = "https://openrouter.ai/api/v1/chat/completions"
self._api_key = os.getenv("OPENROUTER_API_KEY", "")
@retry(
stop=stop_after_attempt(int(os.getenv("MSWEA_MODEL_RETRY_STOP_AFTER_ATTEMPT", "10"))),
wait=wait_exponential(multiplier=1, min=4, max=60),
before_sleep=before_sleep_log(logger, logging.WARNING),
retry=retry_if_not_exception_type(
(
OpenRouterAuthenticationError,
KeyboardInterrupt,
)
),
)
def _query(self, messages: list[dict[str, str]], **kwargs):
headers = {
"Authorization": f"Bearer {self._api_key}",
"Content-Type": "application/json",
}
payload = {
"model": self.config.model_name,
"messages": messages,
"usage": {"include": True},
**(self.config.model_kwargs | kwargs),
}
try:
response = requests.post(self._api_url, headers=headers, data=json.dumps(payload), timeout=60)
response.raise_for_status()
return response.json()
except requests.exceptions.HTTPError as e:
if response.status_code == 401:
error_msg = "Authentication failed. You can permanently set your API key with `mini-extra config set OPENROUTER_API_KEY YOUR_KEY`."
raise OpenRouterAuthenticationError(error_msg) from e
elif response.status_code == 429:
raise OpenRouterRateLimitError("Rate limit exceeded") from e
else:
raise OpenRouterAPIError(f"HTTP {response.status_code}: {response.text}") from e
except requests.exceptions.RequestException as e:
raise OpenRouterAPIError(f"Request failed: {e}") from e
def query(self, messages: list[dict[str, str]], **kwargs) -> dict:
if self.config.set_cache_control:
messages = set_cache_control(messages, mode=self.config.set_cache_control)
response = self._query([{"role": msg["role"], "content": msg["content"]} for msg in messages], **kwargs)
usage = response.get("usage", {})
cost = usage.get("cost", 0.0)
if cost <= 0.0 and self.config.cost_tracking != "ignore_errors":
raise RuntimeError(
f"No valid cost information available from OpenRouter API for model {self.config.model_name}: "
f"Usage {usage}, cost {cost}. Cost must be > 0.0. Set cost_tracking: 'ignore_errors' in your config file or "
"export MSWEA_COST_TRACKING='ignore_errors' to ignore cost tracking errors "
"(for example for free/local models), more information at https://klieret.short.gy/mini-local-models "
"for more details. Still stuck? Please open a github issue at https://github.com/SWE-agent/mini-swe-agent/issues/new/choose!"
)
self.n_calls += 1
self.cost += cost
GLOBAL_MODEL_STATS.add(cost)
return {
"content": response["choices"][0]["message"]["content"] or "",
"extra": {
"response": response, # already is json
},
}
def get_template_vars(self) -> dict[str, Any]:
return asdict(self.config) | {"n_model_calls": self.n_calls, "model_cost": self.cost}
Guide
Setting up OpenRouter models is covered in the quickstart guide.
module-attribute
logger = getLogger('openrouter_model')
dataclass
OpenRouterModelConfig(
model_name: str,
model_kwargs: dict[str, Any] = dict(),
set_cache_control: Literal["default_end"] | None = None,
cost_tracking: Literal[
"default", "ignore_errors"
] = getenv("MSWEA_COST_TRACKING", "default"),
)
instance-attribute
model_name: str
class-attribute
instance-attribute
model_kwargs: dict[str, Any] = field(default_factory=dict)
class-attribute
instance-attribute
set_cache_control: Literal['default_end'] | None = None
Set explicit cache control markers, for example for Anthropic models
class-attribute
instance-attribute
cost_tracking: Literal["default", "ignore_errors"] = getenv(
"MSWEA_COST_TRACKING", "default"
)
Cost tracking mode for this model. Can be "default" or "ignore_errors" (ignore errors/missing cost info)
Bases: Exception
Custom exception for OpenRouter API errors.
Bases: Exception
Custom exception for OpenRouter authentication errors.
Bases: Exception
Custom exception for OpenRouter rate limit errors.
OpenRouterModel(**kwargs)
Source code in src/minisweagent/models/openrouter_model.py
51 52 53 54 55 56 | |
instance-attribute
cost = 0.0
instance-attribute
n_calls = 0
query(messages: list[dict[str, str]], **kwargs) -> dict
Source code in src/minisweagent/models/openrouter_model.py
97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 | |
get_template_vars() -> dict[str, Any]
Source code in src/minisweagent/models/openrouter_model.py
124 125 | |
