Skip to content

Requesty Model

Requesty Model class

Full source code
import json
import logging
import os
from dataclasses import asdict, dataclass, field
from typing import Any

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

logger = logging.getLogger("requesty_model")


@dataclass
class RequestyModelConfig:
    model_name: str
    model_kwargs: dict[str, Any] = field(default_factory=dict)


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(
        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 asdict(self.config) | {"n_model_calls": self.n_calls, "model_cost": self.cost}

minisweagent.models.requesty_model

logger module-attribute

logger = getLogger('requesty_model')

RequestyModelConfig dataclass

RequestyModelConfig(
    model_name: str, model_kwargs: dict[str, Any] = dict()
)

model_name instance-attribute

model_name: str

model_kwargs class-attribute instance-attribute

model_kwargs: dict[str, Any] = field(default_factory=dict)

RequestyAPIError

Bases: Exception

Custom exception for Requesty API errors.

RequestyAuthenticationError

Bases: Exception

Custom exception for Requesty authentication errors.

RequestyRateLimitError

Bases: Exception

Custom exception for Requesty rate limit errors.

RequestyModel

RequestyModel(**kwargs)
Source code in src/minisweagent/models/requesty_model.py
46
47
48
49
50
51
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", "")

config instance-attribute

config = RequestyModelConfig(**kwargs)

cost instance-attribute

cost = 0.0

n_calls instance-attribute

n_calls = 0

query

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
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
        },
    }

get_template_vars

get_template_vars() -> dict[str, Any]
Source code in src/minisweagent/models/requesty_model.py
118
119
def get_template_vars(self) -> dict[str, Any]:
    return asdict(self.config) | {"n_model_calls": self.n_calls, "model_cost": self.cost}