Requesty Model
Requesty Model class
Full source code
import json
import logging
import os
from typing import Any
import requests
from pydantic import BaseModel
from tenacity import (
before_sleep_log,
retry,
retry_if_not_exception_type,
stop_after_attempt,
wait_exponential,
)
from minisweagent.models import GLOBAL_MODEL_STATS
logger = logging.getLogger("requesty_model")
class RequestyModelConfig(BaseModel):
model_name: str
model_kwargs: dict[str, Any] = {}
class RequestyAPIError(Exception):
"""Custom exception for Requesty API errors."""
pass
class RequestyAuthenticationError(Exception):
"""Custom exception for Requesty authentication errors."""
pass
class RequestyRateLimitError(Exception):
"""Custom exception for Requesty rate limit errors."""
pass
class RequestyModel:
def __init__(self, **kwargs):
self.config = RequestyModelConfig(**kwargs)
self.cost = 0.0
self.n_calls = 0
self._api_url = "https://router.requesty.ai/v1/chat/completions"
self._api_key = os.getenv("REQUESTY_API_KEY", "")
@retry(
reraise=True,
stop=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(
(
RequestyAuthenticationError,
KeyboardInterrupt,
)
),
)
def _query(self, messages: list[dict[str, str]], **kwargs):
headers = {
"Authorization": f"Bearer {self._api_key}",
"Content-Type": "application/json",
"HTTP-Referer": "https://github.com/SWE-agent/mini-swe-agent",
"X-Title": "mini-swe-agent",
}
payload = {
"model": self.config.model_name,
"messages": messages,
**(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 REQUESTY_API_KEY YOUR_KEY`."
raise RequestyAuthenticationError(error_msg) from e
elif response.status_code == 429:
raise RequestyRateLimitError("Rate limit exceeded") from e
else:
raise RequestyAPIError(f"HTTP {response.status_code}: {response.text}") from e
except requests.exceptions.RequestException as e:
raise RequestyAPIError(f"Request failed: {e}") from e
def query(self, messages: list[dict[str, str]], **kwargs) -> dict:
response = self._query([{"role": msg["role"], "content": msg["content"]} for msg in messages], **kwargs)
# Extract cost from usage information
usage = response.get("usage", {})
cost = usage.get("cost", 0.0)
# If cost is not available, raise an error
if cost == 0.0:
raise RequestyAPIError(
f"No cost information available from Requesty API for model {self.config.model_name}. "
"Cost tracking is required but not provided by the API response."
)
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 self.config.model_dump() | {"n_model_calls": self.n_calls, "model_cost": self.cost}
module-attribute
logger = getLogger('requesty_model')
Bases: BaseModel
instance-attribute
model_name: str
class-attribute
instance-attribute
model_kwargs: dict[str, Any] = {}
Bases: Exception
Custom exception for Requesty API errors.
Bases: Exception
Custom exception for Requesty authentication errors.
Bases: Exception
Custom exception for Requesty rate limit errors.
RequestyModel(**kwargs)
Source code in src/minisweagent/models/requesty_model.py
45 46 47 48 49 50 | |
instance-attribute
cost = 0.0
instance-attribute
n_calls = 0
query(messages: list[dict[str, str]], **kwargs) -> dict
Source code in src/minisweagent/models/requesty_model.py
93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 | |
get_template_vars() -> dict[str, Any]
Source code in src/minisweagent/models/requesty_model.py
118 119 | |
