tropass-sdk 1.1.0

Tropass SDK helps you develop and manage your models for Tropass platform

tropass-sdk — это инструмент для разработки и управления ML-моделями на платформе Тропасс.

---

📦 Установка

# Для pip
pip install tropass-sdk[server]

# Для uv
uv add tropass-sdk[server]

# Для poetry
poetry add tropass-sdk[server]

---

🛠 Подготовка приложения

Для инициализации сервера достаточно передать функцию предсказания в класс ModelServer.

Ключевые требования:

• Функция предсказания обязана принимать три аргумента model_input, common_resources и request_metadata.
• Функция предсказания обязана возвращать схему ответа модели MLModelResponseSchema.
• Инстанс ModelServer обязан находиться в файле main.py в корне проекта.

from tropass_sdk.server import ModelServer
from tropass_sdk.schemas.model_contract_schema import MLModelRequestMetadataSchema, MLModelRequestSchema, MLModelResponseSchema

def predict_handler(
    model_input: dict[str, list[typing.Any]],
    common_resources: dict[str, typing.Any],
    request_metadata: MLModelRequestMetadataSchema | None = None,
) -> MLModelResponseSchema:
    # Логика инференса модели
    return MLModelResponseSchema(panel_items=[])

server = ModelServer(
    model_func=predict_handler,
    model_name="my_model",
    model_description="Production model description",
    model_version="1.0.0",
    debug=False
)

Метаданные запроса

ModelServer автоматически извлекает метаданные из HTTP-заголовков запроса к /prediction и передает их в аргумент request_metadata.

Поддерживаемые заголовки:

• X-User-ID — идентификатор пользователя. Доступен как request_metadata.user_id.
• X-User-Locale — локаль пользователя. Доступна как request_metadata.locale, значение по умолчанию — ru.
• X-User-Api-Token — API-токен пользователя. Доступен как request_metadata.user_api_token.

Пример использования:

def predict_handler(
    model_input: dict[str, list[typing.Any]],
    common_resources: dict[str, typing.Any],
    request_metadata: MLModelRequestMetadataSchema | None = None,
) -> MLModelResponseSchema:
    user_id = request_metadata.user_id if request_metadata is not None else None
    locale = request_metadata.locale if request_metadata is not None else "ru"

    # Логика инференса модели с учетом пользователя и локали
    return MLModelResponseSchema(panel_items=[])

⚡ Варианты запуска

Для запуска создайте экземпляр приложения c помощью метода build_application. Важно! Имя приложения обязательно должно быть application:

application = server.build_application()

Запуск: uvicorn main:application --host 0.0.0.0 --port 8000 --workers 4

🌉 Клиент для вызова моделей через Gateway

GatewayClient — асинхронный клиент для вызова моделей через Gateway. Клиент сам создает HTTP-соединение, добавляет приватный токен в заголовок X-API-TOKEN, выполняет retry через stamina и защищает вызов circuit breaker через circuitbreaker.

import typing
import uuid

from tropass_sdk.client import GatewayClient


async def call_gateway_model() -> dict[str, typing.Any]:
    async with GatewayClient(
        gateway_url="https://api.tropass.me",
        gateway_api_token="private-token",
    ) as gateway_client:
        return await gateway_client.call_model(
            model_id=uuid.UUID("00000000-0000-0000-0000-000000000123"),
            model_request_data={
                "input_name": ["input-value"]
            },
        )

call_model отправляет запрос на api/rpc/call-model в формате:

{
    "model_id": "00000000-0000-0000-0000-000000000123",
    "model_request_data": {
        "input_name": ["input-value"]
    },
}

Для настройки клиента можно передать GatewayClientConfig в аргумент client_config.

Параметры конфигурации:

• timeout_seconds — HTTP timeout одного запроса к Gateway.
• retry_attempts — максимальное количество попыток выполнить запрос.
• retry_timeout_seconds — общий лимит времени на retry-цикл.
• retry_wait_exp_base — база экспоненциального роста паузы между retry.
• circuit_failure_threshold — количество неуспешных вызовов до открытия circuit breaker.
• circuit_recovery_seconds — время до попытки восстановить circuit breaker.
• circuit_success_threshold — зарезервирован для совместимости конфигурации.

Метод возвращает dict с ответом Gateway или выбрасывает GatewayCallError.

Оригинальная причина ошибки доступна через __cause__: например открытый circuit breaker, некорректный JSON, HTTP-ошибка или ошибка транспорта после всех retry.

🔍 Мониторинг и наблюдаемость

Благодаря интеграции с microbootstrap, сервис из коробки поддерживает:

• Метрики: Доступны по эндпоинту /metrics в формате Prometheus.
• Health Checks: Проверка состояния сервиса доступна по пути /health.
• Логирование: Структурированные логи готовы к сбору в ELK/Loki.

---