Jev на Python — установка SDK и примеры

Установка и ключ

pip install typesafe-sdk   # или: uv add typesafe-sdk
export TYPESAFE_API_KEY="ваш_ключ"

Нужен Python 3.10 или новее. Пакет называется typesafe-sdk, а импортируется как typesafe_sdk. Ключ SDK берёт из переменной TYPESAFE_API_KEY, если не передать его в клиент явно. Пробелы и переводы строк по краям ключа отбрасываются. Пустой ключ или ключ с недопустимыми символами клиент отклонит ещё при создании: конструктор выбросит TypeSafeError до первого запроса.

Проверьте версию пакета. В 0.6.0 уровни Score стали передаваться упорядоченным списком вместо словаря с числовыми ключами, а в 0.7.0 SDK перешёл с msgspec на pydantic. Код для версий до 0.6.0 со словарём уровней придётся переписать.

Переменная	Что задаёт	По умолчанию
TYPESAFE_API_KEY	Ключ API	—
TYPESAFE_BASE_URL	Адрес API	https://api.typesafe.ai
TYPESAFE_DEFAULT_MODEL	Модель	jev-latest
TYPESAFE_LOG_LEVEL	Уровень логов typesafe_sdk	не задан

Явные параметры клиента важнее переменных окружения.

Первый запрос

from typesafe_sdk import Choice, Noul, Score, TypeSafeClient

with TypeSafeClient() as client:
    response = client.system_one(
        state={"document": "С карты списали оплату дважды. Верните деньги срочно."},
        questions={
            "billing": Noul(instructions="Обращение касается оплаты?"),
            "tone": Choice(
                instructions="Какой тон у клиента?",
                criteria={"calm": None, "frustrated": None, "angry": None},
            ),
            "urgency": Score(
                instructions="Насколько срочно нужно ответить?",
                criteria=["Может подождать неделю", "Ответить за день", "Ответить за час"],
            ),
        },
    )

print(response.nouls["billing"].noul)    # 0.97
print(response.choices["tone"].choice)   # frustrated
print(response.scores["urgency"].score)  # 1.62
print(response.model, response.usage.input_tokens)

Числа в комментариях условные. Разберём, что здесь происходит.

• with закрывает соединения при выходе из блока. Без него вызовите client.close() сами.
• state принимает строку, словарь или список. Вопросы передаются словарём «ключ → вопрос». Объекты Noul, Choice и Score проверяют поля до отправки, но SDK примет и обычные словари в формате API вроде {"type": "noul", "instructions": "..."}, их можно смешивать.
• Ответы лежат в response.answers по ключам вопросов. Свойства response.nouls, response.choices и response.scores дают те же ответы, сгруппированные по типам, так что IDE подсказывает поля.
• Уровни Score в SDK индексируются целыми числами: probabilities[2] и legend[2], а не строковыми ключами, как в JSON.
• Метаданные: response.model — версия модели, response.usage — токены, response.request_id — ID запроса из заголовка x-typesafe-request-id. Сохраняйте его в логах, он понадобится при разборе проблем.

Модель по умолчанию — jev-latest. Зафиксировать версию можно в клиенте, TypeSafeClient(model="jev-1.13.0"), или в отдельном вызове: client.system_one(state, questions, model="jev-1.13.0"). Как выбирать типы вопросов и писать уровни, разобрано в статье о Choice, Score и Noul, а все поля ответа описаны в справочнике по API.

Асинхронный клиент: пачка документов

import asyncio
from typesafe_sdk import AsyncTypeSafeClient, Noul

TICKETS = ["Не могу войти в аккаунт", "Спасибо, всё работает!", "Верните деньги!!!"]

async def classify(client, limit, text):
    async with limit:
        r = await client.system_one(
            state={"ticket": text},
            questions={"complaint": Noul(instructions="Клиент жалуется на проблему?")},
        )
    return text, r.nouls["complaint"].noul

async def main():
    limit = asyncio.Semaphore(5)  # 5 одновременно ≈ 1000 запросов в минуту при ответе за 0,3 с
    async with AsyncTypeSafeClient() as client:
        results = await asyncio.gather(*(classify(client, limit, t) for t in TICKETS))
    for text, p in results:
        print(f"{p:.2f}  {text}")

asyncio.run(main())

У AsyncTypeSafeClient те же параметры, что у синхронного клиента, только system_one вызывается через await. Создайте один клиент на всю пачку и ограничьте число одновременных запросов семафором. Учтите, что семафор ограничивает одновременность, а не частоту: при ответе за 0,3 секунды каждый поток даёт около 200 запросов в минуту, поэтому 5 потоков — это примерно 1000 в минуту, с запасом под лимит. Без ограничения тысяча документов уйдёт разом, упрётся в лимиты 1200 запросов в минуту и 250 000 токенов в секунду и получит ошибки 429. SDK их повторит, но пачка обработается медленнее, чем при ровном потоке. Если к каждому документу несколько вопросов, задавайте их одним запросом, а не несколькими. Как выдерживать темп, настраивать повторы и разбирать упавшие документы, рассказано в статье об асинхронном клиенте.

Типизированный ответ

from typesafe_sdk import Noul, NoulAnswer, SystemOneResponse, TypeSafeClient

class BillingResponse(SystemOneResponse):
    billing: NoulAnswer

with TypeSafeClient() as client:
    result = client.system_one(
        "С карты дважды списали оплату.",
        {"billing": Noul(instructions="Обращение касается оплаты?")},
        response_model=BillingResponse,
    )
    print(result.billing.noul)

Параметр response_model появился в версии 0.7.0. Вы описываете ожидаемые ответы моделью pydantic и обращаетесь к ним как к атрибутам с подсказками IDE. Если ответ не совпал с описанием, SDK выбросит TypeSafeAPIResponseValidationError с путём к проблемному полю.

Маршрутизация по уверенности

p = response.nouls["billing"].noul
if p >= 0.85:
    route_to("billing")
elif p <= 0.15:
    route_to("general")
else:
    route_to("human_review")  # серая зона

Пороги 0,85 и 0,15 здесь условные. У Choice и Score в ответе есть ещё confidence, и по нему строят такие же три ветки: действовать, перепроверить, отдать человеку. Как подобрать пороги на размеченной выборке, рассказано в статье об уверенности Jev.

Повторы, таймауты и ошибки

По умолчанию SDK повторяет запрос до двух раз на ответы 408, 429 и любые 5xx, включая 529, а также при обрыве соединения и таймауте. Пауза начинается с 0,5 секунды и удваивается до 5 секунд со случайным разбросом, заголовок Retry-After учитывается. На одну HTTP-операцию отводится 10 секунд, на весь вызов вместе с повторами — 30.

from typesafe_sdk import RetryPolicy, TypeSafeClient

client = TypeSafeClient(
    timeout=3.0,  # секунд на одну HTTP-операцию
    retry=RetryPolicy(max_retries=4, backoff_max=2.0, timeout=15.0),
)

RetryPolicy(max_retries=0) отключает повторы. Политику и таймаут можно передать и в отдельный вызов: client.system_one(state, questions, retry=..., timeout=...).

Если запрос так и не прошёл, SDK выбросит исключение:

Исключение	Когда
TypeSafeAuthenticationError	401, неверный ключ
TypeSafeUnprocessableEntityError	422, запрос не прошёл валидацию
TypeSafeRateLimitError	429 после всех повторов, в retry_after_ms подсказка о паузе
TypeSafeInternalServerError	Ошибки 5xx, в том числе 529
TypeSafeAPIConnectionError, TypeSafeAPITimeoutError	Нет соединения или истёк таймаут

HTTP-ошибки наследуют TypeSafeAPIError с полями status, body и request_id, а все исключения SDK — TypeSafeError.

import logging
from typesafe_sdk import TypeSafeAPIError, TypeSafeRateLimitError

log = logging.getLogger(__name__)

try:
    response = client.system_one(state, questions)
except TypeSafeRateLimitError as e:
    log.warning("лимит запросов, пауза %s мс", e.retry_after_ms)
except TypeSafeAPIError as e:
    log.error("ошибка %s, request_id=%s, тело: %s", e.status, e.request_id, e.body)

Что означает каждый код и как строить повторы без SDK, описано в статье об ошибках API Jev.

Подключение через OpenRouter, Polza.AI и Vercel

Адрес API задаёт параметр base_url или переменная TYPESAFE_BASE_URL. Указывайте корень без /v1: путь /v1/systemone SDK допишет сам.

import os
from typesafe_sdk import TypeSafeClient

openrouter = TypeSafeClient(
    api_key=os.environ["OPENROUTER_API_KEY"],
    base_url="https://openrouter.ai/api",
    model="typesafe/jev-1.13",
)

polza = TypeSafeClient(
    api_key=os.environ["POLZA_AI_API_KEY"],
    base_url="https://polza.ai/api",
    model="typesafe/jev",
)

vercel = TypeSafeClient(
    api_key=os.environ["AI_GATEWAY_API_KEY"],
    base_url="https://ai-gateway.vercel.sh/typesafe",
    model="typesafe-ai/jev",
)

У каждой площадки свои нюансы:

• OpenRouter понимает и имена TypeSafe без префикса: jev-latest он направит на ~typesafe/jev-latest, jev-1.13 — на typesafe/jev-1.13. Список моделей через client.models.list() с OpenRouter не работает, потому что OpenRouter отвечает в своём формате и SDK его отклоняет. Подробности — в статье о Jev через OpenRouter.
• Polza.AI возвращает стоимость запроса в рублях в поле usage.cost_rub, но Python SDK, по данным Polza.AI, это поле отбрасывает. Достать его можно из исходного HTTP-ответа: response.raw_http_response.json()["usage"]. Как подключиться и оплатить из России, читайте в статье о Jev через Polza.AI.
• Vercel AI Gateway работает с ключом шлюза и оплачивается кредитами Vercel. Настройка описана в статье о Jev в Vercel AI Gateway.

Частые ошибки

• Новый клиент на каждый запрос. Клиент держит HTTP-соединения, поэтому создавайте его один раз при старте приложения и переиспользуйте.
• Синхронный клиент внутри async-кода. TypeSafeClient блокирует цикл событий на время запроса. В FastAPI, aiohttp и других асинхронных фреймворках берите AsyncTypeSafeClient.
• /v1 в base_url. SDK сам добавляет /v1/systemone, поэтому адрес вида https://polza.ai/api/v1 превратится в путь с двойным /v1.
• Пустые вопросы и пустая шкала. Если словарь вопросов пуст или у Score нет уровней, SDK выбросит TypeSafeError ещё до отправки запроса.
• Поле не того типа. У ответа Noul нет confidence и choice, только noul. Читайте ответы через response.nouls, response.choices и response.scores: у каждого из них свой тип ответа, и IDE подскажет, какие поля доступны.

Логирование

import logging

logging.getLogger("typesafe_sdk").setLevel(logging.INFO)

На уровне info SDK пишет одну строку на запрос, на debug добавляет заголовки и тела запросов и ответов. Ключи и другие секретные заголовки в логах маскируются, а тела — нет. Если в state бывают персональные данные, не включайте debug в продакшене. Без правки кода уровень задаёт переменная TYPESAFE_LOG_LEVEL со значением debug, info, warning, error или off, её нужно выставить до импорта SDK.