NexaraNexara

Python SDK

Официальная Python-библиотека Nexara: транскрибация, диаризация, роли и LLM-анализ в несколько строк кода.

nexara — официальная Python-библиотека для Nexara API. Она берёт на себя сборку запросов, разбор ответов в типизированные модели, обработку ошибок и повторные попытки, а асинхронные задачи превращает в один вызов job.wait(). Библиотека полностью типизирована (mypy strict) и работает на Python 3.10+.

pip install nexara

Сохраните API-ключ в переменной окружения NEXARA_API_KEY. SDK читает её автоматически при создании клиента, поэтому ключ не нужно указывать в коде — и он не попадёт в репозиторий:

export NEXARA_API_KEY=nx-...

Команда export задаёт переменную только для текущей сессии терминала. Чтобы ключ сохранялся между сессиями, добавьте эту строку в файл конфигурации вашей оболочки (например, ~/.zshrc или ~/.bashrc). На сервере переменную удобнее задавать через настройки окружения или менеджер секретов.

Быстрый старт

from nexara import Nexara

client = Nexara()  # ключ из переменной окружения NEXARA_API_KEY

text = client.transcriptions.create(file="audio.mp3").text
print(text)

Клиент берёт ключ из NEXARA_API_KEY — той самой переменной, что вы задали выше, — поэтому в коде его указывать не нужно. Передать ключ явным аргументом api_key= тоже можно, но не храните его в коде и не коммитьте в репозиторий.

Клиент

client = Nexara(
    # api_key по умолчанию берётся из переменной окружения NEXARA_API_KEY,
    # либо можете передать ключ сами: api_key="nx-..."
    timeout=600.0,      # таймаут запроса в секундах
    max_retries=2,      # повторы при 429 и сетевых сбоях
)

Клиент можно использовать как контекстный менеджер (with Nexara() as client:) — тогда HTTP-соединения закроются автоматически.

Транскрибация

Передайте ровно один из параметров: file (путь, bytes или открытый бинарный файл) или url (прямая ссылка). Файл, переданный путём, стримится с диска и не загружается в память целиком.

# Локальный файл
result = client.transcriptions.create(file="audio.mp3")
print(result.text)

# Файл по ссылке
result = client.transcriptions.create(url="https://example.com/audio.mp3")
print(result.text)

Параметры те же, что и у API: model, language, response_format, timestamp_granularities, profanity_filter и другие. Форматы text, srt и vtt возвращаются обычной строкой, json и verbose_json — типизированными объектами:

srt = client.transcriptions.create(file="video.mp4", response_format="srt")
with open("video.srt", "w", encoding="utf-8") as f:
    f.write(srt)

verbose = client.transcriptions.create(
    file="audio.mp3",
    response_format="verbose_json",
    timestamp_granularities=["word"],
)
for word in verbose.words:
    print(f"{word.start:.2f}{word.end:.2f}: {word.word}")

Диаризация и роли

Разделение по говорящим включается так же, как в API, — task="diarize". Список ролей передаётся обычным списком или словарём Python: SDK сам сериализует его в JSON.

call = client.transcriptions.create(
    file="call.mp3",
    task="diarize",
    num_speakers=2,
    roles=["Оператор", "Клиент"],  # или "auto", или {"Роль": "описание"}
)

for segment in call.segments:
    print(f"{segment.speaker}: {segment.text}")

Подробнее о режимах — на странице спикеры и роли.

Асинхронные задачи

Для длинных записей используйте асинхронный режим: create_job() ставит задачу и сразу возвращает объект Job, а job.wait() опрашивает статус и возвращает готовый результат.

job = client.transcriptions.create_job(file="long_recording.mp3")
print(job.job_id, job.status)  # in_progress

result = job.wait()  # опрашивает статус; таймаут по умолчанию — 1800 секунд
print(result.text)

Задачу можно забрать позже — в том числе из другого процесса:

job = client.transcriptions.retrieve_job(job_id)
if job.status == "complete":
    print(job.result)

Что важно знать:

  • Результат хранится 12 часов с момента создания задачи, затем удаляется — после этого retrieve_job() вернёт NotFoundError.
  • Одновременно на один ключ может выполняться до 200 задач.
  • Если задача завершилась ошибкой, wait() бросает JobFailedError. Неудавшаяся задача не тарифицируется — повторная отправка бесплатна.
  • Если задача не успела за таймаут, wait() бросает JobTimeoutError; сама задача при этом не отменяется, её можно забрать позже через retrieve_job().

«Асинхронный» здесь означает фоновую обработку на сервере, а не asyncio. Для asyncio используйте AsyncNexara — оба клиента поддерживают и create(), и create_job().

LLM-анализ

Передайте prompt, чтобы прогнать расшифровку через языковую модель, и json_schema — чтобы получить структурированный ответ. Схема передаётся обычным словарём Python, а llm_output возвращается уже разобранным объектом — никакого ручного json.dumps/json.loads.

result = client.transcriptions.create(
    file="meeting.mp3",
    prompt="Составь протокол встречи: резюме, решения и задачи.",
    json_schema={
        "type": "object",
        "properties": {
            "summary": {"type": "string"},
            "decisions": {"type": "array", "items": {"type": "string"}},
            "action_items": {"type": "array", "items": {"type": "string"}},
        },
        "required": ["summary", "decisions", "action_items"],
    },
)

print(result.llm_output["summary"])       # dict по вашей схеме
print(result.transcription.text)          # расшифровка, из которой он получен

Без json_schema поле llm_output — строка со свободным ответом модели.

asyncio

AsyncNexara — тот же интерфейс под await:

import asyncio
from nexara import AsyncNexara

async def main():
    async with AsyncNexara() as client:
        result = await client.transcriptions.create(file="audio.mp3")
        print(result.text)

asyncio.run(main())

Ошибки

Некорректные запросы (например, file и url одновременно, слишком много ролей) отклоняются на стороне клиента с NexaraValidationError — ещё до обращения к серверу. Ошибки сервера превращаются в типизированные исключения по коду статуса:

ИсключениеКодКогда возникает
BadRequestError400Некорректные параметры запроса
InsufficientBalanceError402Недостаточно средств на балансе
AuthenticationError403Ключ не передан или недействителен
NotFoundError404Задача, модель или ключ не найдены
RateLimitError429Превышен лимит частоты или число одновременных задач
InternalServerError500Внутренняя ошибка сервера
APIConnectionErrorЗапрос не дошёл до сервера (сеть, таймаут)

Все они наследуются от NexaraError, а у ошибок API есть поля status_code и detail с сообщением сервера. Полный список кодов — на странице ошибок.

from nexara import Nexara, InsufficientBalanceError, RateLimitError

client = Nexara()

try:
    result = client.transcriptions.create(file="audio.mp3")
except InsufficientBalanceError as e:  # 402
    print("Пополните баланс:", e.detail)
except RateLimitError:
    print("Слишком много запросов, попробуйте позже.")

Ссылки

On this page