Асинхронные вызовы и потоковая выдача

AsyncBroker предоставляет асинхронный API. Broker использует тот же механизм, но предоставляет синхронные методы. При работе с AsyncBroker методы вызываются через await:

async with llmbroker.AsyncBroker() as broker:
    reply = await broker.ask("Привет")
    print(reply.text)

Для асинхронного цикла инструментов используйте await llmbroker.arun_tool_loop(...). Подробнее см. в разделе Инструменты и агенты.

Потоковая выдача

Метод stream() возвращает части текста по мере их поступления. Выбор модели и переход к другой модели при ошибке выполняются так же, как для обычного вызова:

async with llmbroker.AsyncBroker() as broker:
    stream = broker.stream("Напиши хокку про брокеров", operation="write")
    async for delta in stream:
        print(delta, end="", flush=True)

    print(stream.llm_name, stream.usage)  # имя модели и сведения об использовании
    await stream.record_quality(0.9)  # оценить полученный ответ

stream(...) возвращает объект асинхронного итератора. После завершения ответа в его полях доступны имя модели и сведения об использовании. Синхронный Broker выдаёт ответ по частям так же, через обычный цикл for; подробнее см. в разделе Потоковая выдача в синхронном коде.

The broker owns its streams. Leaving its context closes unfinished work and waits for journal writes before closing HTTP and storage. No separate stream context is required.

До получения первой части текста брокер может перейти к другой модели. Это происходит, если провайдер вернул ошибку, ограничил число запросов или завершил ответ без текста и вызовов инструментов. После получения первой части сменить модель без замены уже показанного текста нельзя. Если ответ прерывается в этот момент, возникает StreamInterruptedError, а ранее полученные части остаются у приложения. Поведение при fastest_of > 1 описано в разделе Параллельная потоковая выдача.

Параллельная потоковая выдача

При fastest_of > 1 брокер одновременно обращается к нескольким моделям. Все они продолжают работу, пока одна не завершит полный ответ. Модель, которая первой вернула часть текста, не обязательно первой завершит генерацию. Поэтому брокер не прерывает остальные запросы после появления первого текста.

Части текста, полученные до завершения одного из ответов, считаются предварительными. Если первой завершит работу другая модель, возникнет StreamReplacementError. Поле replacement этого исключения содержит полный итоговый ответ. Приложение должно заменить им весь ранее показанный текст. Части ответов разных моделей не объединяются.

stream = broker.stream("Напиши хайку", fastest_of=2, stream_selection_window=1.0)
parts = []
try:
    async for delta in stream:
        parts.append(delta)
        show(delta)
except llmbroker.StreamReplacementError as exc:
    replace_everything_with(exc.replacement.text)  # заменить предварительный текст
else:
    text = "".join(parts)

print(stream.llm_name)  # модель, чей ответ выбран итоговым
await stream.record_quality(0.9)  # оценить итоговый ответ

Оценивайте параллельный вызов через объект stream или exc.replacement. При fastest_of в журнале могут остаться несколько успешных попыток с одним trace_id, хотя пользователь получил только один ответ. Вызов record_quality(..., trace_id=...) не позволяет определить, какой именно ответ был показан. Поэтому используйте возвращённый объект или сохраните его call_id.

После StreamReplacementError итератор завершён и больше не возвращает данные, но тот же объект остаётся открытым для another(). Обрабатывайте это исключение до более общего LLMRequestError, иначе готовый ответ будет ошибочно обработан как сбой.

Параметр stream_selection_window влияет только на выбор текста, который будет показан первым. В течение указанного числа секунд, по умолчанию одной секунды, брокер ждёт данные от модели, стоящей первой в текущей очереди. Если она не начала отвечать, показываются уже полученные данные другой модели. При ошибке первой модели ожидание прекращается сразу. Значение 0 означает, что нужно показывать первый поступивший текст независимо от порядка моделей.

Истечение stream_selection_window не считается ошибкой или превышением времени ожидания и не влияет на дальнейший порядок моделей. Итоговым остаётся полный ответ модели, которая завершила работу первой.

Каждый параллельный запрос расходует квоту провайдера, а завершённые ответы хранятся в памяти до запроса или закрытия объекта. Это происходит независимо от скорости, с которой приложение читает данные.

Если fastest_of не задан или равен 1, брокер после получения первой части текста продолжает работать только с этой моделью, а stream_selection_window не влияет на поведение.

Получение ещё одного полного ответа

Call another() while the broker is open when the application rejects a complete answer. The broker uses retained attempts before trying unattempted pool models:

stream = broker.stream(prompt, fastest_of=3, wait=25)
try:
    text = "".join([delta async for delta in stream])
    rated = stream
except llmbroker.StreamReplacementError as exc:
    text, rated = exc.replacement.text, exc.replacement

while not acceptable(text):
    await rated.record_quality(0.0)
    answer = await stream.another()
    if answer is None:
        break
    text, rated = answer.text, answer

У каждого результата свои llm_name, call_id, usage и объект для оценки. Исходный поток сохраняет идентификатор первого итогового ответа. Значение None означает, что получить ещё один полный ответ в рамках этого вызова нельзя; последующие вызовы another() также вернут None.

Само хранение открытого объекта не запускает новые запросы после первого ответа: это может сделать только another(). Уже запущенные попытки могут продолжаться. Закрытие прерывает незавершённые попытки и дожидается записи журнала и освобождения занятых мест.

Ограничение времени для полного ответа

wait ограничивает время получения полного ответа, а не только первой части. Учитывается только время, которое библиотека ждёт данные от провайдера:

stream = broker.stream("Напиши длинный ответ", wait=20.0)
try:
    async for delta in stream:
        print(delta, end="", flush=True)
except llmbroker.LLMTimeoutError as exc:
    print(f"\nвремя ожидания истекло: {exc}")

Время, которое приложение тратит на обработку уже полученной части, не учитывается. Отсчёт приостанавливается при передаче данных приложению и продолжается при запросе следующей части. Если модель быстро начала ответ, но генерирует его слишком медленно, ограничение всё равно сработает.

Одно положительное значение wait действует для первого ответа, проверки в приложении и всех вызовов another(). После первого полного ответа отсчёт идёт непрерывно. Ответ, завершённый до истечения времени, можно получить из памяти и позже, но новый запрос к провайдеру после истечения времени не начинается.

Если wait не задан, пользовательского ограничения времени нет. Если время истекло до получения первой части, возникает NoLLMAvailableError; после первой части возникает LLMTimeoutError, а ранее полученные данные остаются у приложения.

До получения первой части потоковый вызов никогда не обращается к одной и той же модели дважды: он завершается, как только опробованы все доступные ему модели, а не ждёт возвращения одной из них, как это делает ask. Если эти модели лишь временно заблокированы, возникает NoLLMAvailableError с reason="timeout", а в retry_at указано время возвращения первой из них — приложение может повторить запрос позже или обратиться к другой модели. Значение reason="excluded" по-прежнему означает, что ни одна модель не может выполнить запрос. См. Когда никто не может ответить.

Модель, которая не вернула никаких данных за время wait, временно исключается из выбора. Если модель уже начала отвечать, временная блокировка не применяется. В обоих случаях брокер учитывает, какое значение wait оказалось недостаточным, и в следующий раз предпочитает другую модель для такого же короткого запроса. Незавершённый ответ нельзя оценить: при вызове record_quality у его объекта возникнет исключение.

До получения первой части поля llm_name и call_id равны None, поскольку брокер ещё может перейти к другой модели. Поле usage заполняется после завершения ответа.

Оценку также можно записать только после завершения ответа и появления записи в журнале. При более раннем вызове record_quality возникнет ValueError.

A break does not close the stream; the broker closes it on exit. In a long-lived broker, use the stream's optional context for immediate release when this request ends, including on errors. You can rate a partial reply after closing its stream while the broker remains open:

async with broker.stream("Write a haiku about brokers") as stream:
    async for delta in stream:
        if looks_wrong(delta):
            break

await stream.record_quality(0.0)

Keep validation and another() inside this optional context. await stream.aclose() performs the same early release without a context.

Потоковая выдача доступна и для одной конкретной модели через direct(), без выбора из пула и перехода к другой модели. Подробнее см. в разделе Прямые вызовы модели.

Потоковая выдача в синхронном коде

Синхронный Broker и объект, который возвращает его for_scope(...), предоставляют тот же метод stream() с теми же параметрами. Ответ читается обычным циклом for. Всё описанное выше действует без изменений: части текста, переход к другой модели до первой части, StreamInterruptedError, StreamReplacementError, another(), wait и оценки:

with llmbroker.Broker() as broker:
    with broker.stream("Напиши хокку про брокеров") as stream:
        for delta in stream:
            print(delta, end="", flush=True)
    stream.record_quality(0.9)

Каждое исключение возникает на той же части текста, что и в асинхронном коде. Ничего не читается заранее: следующая часть запрашивается, когда её просит цикл. Через один брокер могут одновременно читать ответы многие потоки. Поток, как и объект из for_scope(...), удерживает свой брокер, пока вы с ним не закончите, даже если других ссылок на брокер нет.

Закрытие потока отменяет запрос к провайдеру. Поток закрывается при выходе из блока with, при вызове stream.close() и когда на объект больше нет ссылок. В синхронном веб-сервере именно это останавливает генерацию, когда клиент отключился. Например, Django закрывает тело StreamingHttpResponse, когда клиент уходит, и генератор, читающий поток внутри with, в этот момент его закрывает:

def answer_body(llms, prompt):
    with llms.stream(prompt) as stream:
        for delta in stream:
            yield delta

После StreamReplacementError полный ответ берётся из exc.replacement, как и в асинхронном коде: .text, .llm_name и .call_id. Его record_quality — корутина, поэтому из синхронного кода итоговый ответ оценивается через stream.record_quality(...): после замены поток указывает на заменивший ответ. another() возвращает синхронный результат со своим record_quality.

Локальная база данных для одного процесса

SQLite хранит модели, ключи и журнал в одном файле. Этого достаточно для приложения с одним процессом. Поддерживаемый список моделей записывается в базу перед первым вызовом:

async with llmbroker.AsyncBroker("broker.db") as broker:
    print((await broker.ask("Привет")).text)

Пустая база заполняется автоматически до создания пула; отдельный шаг инициализации не нужен. В дальнейшем список также обновляется автоматически. Если удалённый каталог недоступен, брокер записывает предупреждение в журнал и использует данные из базы. Если база пуста, используется копия списка из установленного пакета.

Если приложение использует тот же файл, учтите режим WAL и блокировку файла. Подробнее см. в разделе SQLite: общая база и WAL. Для нескольких процессов или серверов обновляйте базу один раз во время развёртывания приложения — см. Серверы и кластеры.