sammo.runners

`sammo.runners`#

Module Contents#

Classes#

`MockedRunner`
`BaseRunner`	Base class for OpenAI API runners.
`OpenAIBaseRunner`
`OpenAIChat`
`AzureMixIn`	Mix-in class for Azure API runners.
`OpenAIVisionChat`
`OpenAIEmbedding`
`AzureChat`
`AzureVisionChat`
`AzureEmbedding`
`DeepInfraEmbedding`

Data#

`logger`
`prompt_logger`

API#

sammo.runners.logger = 'getLogger(...)'#

sammo.runners.prompt_logger = 'getLogger(...)'#

exception sammo.runners.RetriableError#: Bases: Exception

exception sammo.runners.NonRetriableError#: Bases: Exception

class sammo.runners.MockedRunner(return_value='')#

Initialization

async generate_text(prompt: str, *args, **kwargs)#

class sammo.runners.BaseRunner(model_id: str, api_config: dict | str | pathlib.Path, cache: None | collections.abc.MutableMapping | str | os.PathLike = None, equivalence_class: str | beartype.typing.Literal[major, exact] = 'major', rate_limit: sammo.throttler.AtMost | list[sammo.throttler.AtMost] | sammo.throttler.Throttler | int = 2, max_retries: int = 50, max_context_window: int | None = None, retry: bool = True, timeout: float | int = 60, max_timeout_retries: int = 1, use_cached_timeouts: bool = True)#

Bases: sammo.base.Runner

Base class for OpenAI API runners.

Parameters:

model_id – Model specifier as listed in the API documentation.
cache – A dict-like object to use for storing results.
api_config – The path to the API config file or a dictionary containing the API information.
rate_limit – The rate limit to use. If an integer, it specifies max calls per second.
max_retries – The maximum number of retries to attempt.
debug_mode – Enable debug mode where queries do not get issued.
retry – Enable retrying when retriable error is raised (defined in each subclass).
timeout – The timeout (in s) to use for a query.
max_context_window – The maximum number of tokens to use for the context window. Defaults to None, which

means that the maximum context window is used. :param max_timeout_retries: The maximum number of retries to attempt when a timeout occurs. :param use_cached_timeouts: Whether to use cached timeouts.

Initialization

RETRY_ERRORS = ()#

DEFAULT_CACHE = None#

class sammo.runners.OpenAIBaseRunner(model_id: str, api_config: dict | str | pathlib.Path, cache: None | collections.abc.MutableMapping | str | os.PathLike = None, equivalence_class: str | beartype.typing.Literal[major, exact] = 'major', rate_limit: sammo.throttler.AtMost | list[sammo.throttler.AtMost] | sammo.throttler.Throttler | int = 2, max_retries: int = 50, max_context_window: int | None = None, retry: bool = True, timeout: float | int = 60, max_timeout_retries: int = 1, use_cached_timeouts: bool = True)#: Bases: sammo.runners.BaseRunner

class sammo.runners.OpenAIChat(model_id: str, api_config: dict | str | pathlib.Path, cache: None | collections.abc.MutableMapping | str | os.PathLike = None, equivalence_class: str | beartype.typing.Literal[major, exact] = 'major', rate_limit: sammo.throttler.AtMost | list[sammo.throttler.AtMost] | sammo.throttler.Throttler | int = 2, max_retries: int = 50, max_context_window: int | None = None, retry: bool = True, timeout: float | int = 60, max_timeout_retries: int = 1, use_cached_timeouts: bool = True)#

Bases: sammo.runners.OpenAIBaseRunner

BASE_URL = 'https://api.openai.com/v1'#

SUFFIX = '/chat/completions'#

async generate_text(prompt: str, max_tokens: int | None = None, randomness: float | None = 0, seed: int = 0, priority: int = 0, system_prompt: str | None = None, history: list[dict] | None = None, json_mode: bool = False) → sammo.base.LLMResult#

Calls the chat endpoint of the OAI model.

Args:: prompt: The user prompt. max_tokens: The maximum number of tokens to generate. If not set, corresponds to maximum available tokens. randomness: The randomness to use when generating tokens. seed: When using randomness, use this seed for local reproducibility (achieved by caching). priority: The priority of the request (used for throttling).
Returns:: Dictionary with keys “data” (the generated text), “cost” (the number of tokens used), and “retries” (the number of retries).

class sammo.runners.AzureMixIn#

Mix-in class for Azure API runners.

Parameters:: api_config – The path to the API config file or a dictionary containing the API information.

Should be of the form: {‘api_key’: ??, ‘endpoint’: ‘https://??.openai.azure.com/’, ‘deployment_id’: ??}.

class sammo.runners.OpenAIVisionChat(model_id: str, api_config: dict | str | pathlib.Path, cache: None | collections.abc.MutableMapping | str | os.PathLike = None, equivalence_class: str | beartype.typing.Literal[major, exact] = 'major', rate_limit: sammo.throttler.AtMost | list[sammo.throttler.AtMost] | sammo.throttler.Throttler | int = 2, max_retries: int = 50, max_context_window: int | None = None, retry: bool = True, timeout: float | int = 60, max_timeout_retries: int = 1, use_cached_timeouts: bool = True)#

Bases: sammo.runners.OpenAIChat

classmethod find_image_segments(text)#

classmethod load_image(img_src)#

class sammo.runners.OpenAIEmbedding(model_id: str, api_config: dict | str | pathlib.Path, cache: None | collections.abc.MutableMapping | str | os.PathLike = None, equivalence_class: str | beartype.typing.Literal[major, exact] = 'major', rate_limit: sammo.throttler.AtMost | list[sammo.throttler.AtMost] | sammo.throttler.Throttler | int = 2, max_retries: int = 50, max_context_window: int | None = None, retry: bool = True, timeout: float | int = 60, max_timeout_retries: int = 1, use_cached_timeouts: bool = True)#

Bases: sammo.runners.OpenAIChat

SUFFIX = '/embeddings'#

INPUT_FIELD = 'input'#

DEFAULT_CACHE = None#

async generate_embedding(text: str | list[str], priority: int = 0) → sammo.base.LLMResult#

class sammo.runners.AzureChat(model_id: str, api_config: dict | str | pathlib.Path, cache: None | collections.abc.MutableMapping | str | os.PathLike = None, equivalence_class: str | beartype.typing.Literal[major, exact] = 'major', rate_limit: sammo.throttler.AtMost | list[sammo.throttler.AtMost] | sammo.throttler.Throttler | int = 2, max_retries: int = 50, max_context_window: int | None = None, retry: bool = True, timeout: float | int = 60, max_timeout_retries: int = 1, use_cached_timeouts: bool = True)#

Bases: sammo.runners.AzureMixIn, sammo.runners.OpenAIChat

SCENARIO = 'chat/completions'#

class sammo.runners.AzureVisionChat(model_id: str, api_config: dict | str | pathlib.Path, cache: None | collections.abc.MutableMapping | str | os.PathLike = None, equivalence_class: str | beartype.typing.Literal[major, exact] = 'major', rate_limit: sammo.throttler.AtMost | list[sammo.throttler.AtMost] | sammo.throttler.Throttler | int = 2, max_retries: int = 50, max_context_window: int | None = None, retry: bool = True, timeout: float | int = 60, max_timeout_retries: int = 1, use_cached_timeouts: bool = True)#

Bases: sammo.runners.AzureMixIn, sammo.runners.OpenAIVisionChat

SCENARIO = 'chat/completions'#

class sammo.runners.AzureEmbedding(model_id: str, api_config: dict | str | pathlib.Path, cache: None | collections.abc.MutableMapping | str | os.PathLike = None, equivalence_class: str | beartype.typing.Literal[major, exact] = 'major', rate_limit: sammo.throttler.AtMost | list[sammo.throttler.AtMost] | sammo.throttler.Throttler | int = 2, max_retries: int = 50, max_context_window: int | None = None, retry: bool = True, timeout: float | int = 60, max_timeout_retries: int = 1, use_cached_timeouts: bool = True)#

Bases: sammo.runners.AzureMixIn, sammo.runners.OpenAIEmbedding

SCENARIO = 'embeddings'#

class sammo.runners.DeepInfraEmbedding(model_id: str, api_config: dict | str | pathlib.Path, cache: None | collections.abc.MutableMapping | str | os.PathLike = None, equivalence_class: str | beartype.typing.Literal[major, exact] = 'major', rate_limit: sammo.throttler.AtMost | list[sammo.throttler.AtMost] | sammo.throttler.Throttler | int = 2, max_retries: int = 50, max_context_window: int | None = None, retry: bool = True, timeout: float | int = 60, max_timeout_retries: int = 1, use_cached_timeouts: bool = True)#

Bases: sammo.runners.BaseRunner

BASE_URL = 'https://api.deepinfra.com/v1/inference/'#

INPUT_FIELD = 'inputs'#

DEFAULT_CACHE = None#