Серверы и кластеры

Для нескольких процессов или серверов укажите брокеру общую базу данных вместо локального хранилища. API вызовов при этом не меняется.

Общая база данных

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

llmbroker.Broker()  # локальный каталог и ключи из окружения
llmbroker.Broker("broker.db")  # SQLite
llmbroker.Broker("postgresql://host/db")  # PostgreSQL
llmbroker.Broker("mongodb://host/db")  # MongoDB

Для каждой базы установите соответствующие дополнительные зависимости. См. Установка. Отдельные хранилища можно задать параметрами registry=, secrets= и store=.

Собственный реестр моделей

Чтобы самостоятельно управлять составом пула, передайте объект, реализующий протокол реестра, и явно укажите режим обновления:

broker = llmbroker.Broker(registry=MyRegistry(), sync=None)  # только ваши записи
broker = llmbroker.Broker(registry=MyRegistry(), sync="freetier")  # ваши плюс наши

источник реестра	если `sync=` не указан	собственные записи
ничего или адрес базы данных	используется `"freetier"`	не изменяются при обновлении
объект реестра	ошибка: нужно выбрать режим	не изменяются при обновлении

Параметр registry= заменяет только реестр. Источники ключей и журнала настраиваются отдельно. Если их не указать, ключи читаются из переменных окружения, а журнал сохраняется в локальном каталоге store. Для серверного приложения обычно следует задать все три хранилища явно:

from llmbroker.postgres import Secrets, Store

broker = llmbroker.AsyncBroker(
    registry=MyRegistry(),
    secrets=Secrets(pool),  # ключи в базе данных
    store=Store(pool),  # журнал в той же базе
    sync=None,
)

Обновление изменяет только записи, ранее созданные самим механизмом синхронизации. Поэтому в одном реестре можно хранить поддерживаемый бесплатный набор и собственные модели.

Добавление собственной модели в пул

Изменяйте реестр через его программный интерфейс, а не прямыми запросами к таблицам: внутренняя схема может измениться между версиями llmbroker. Загрузите существующие записи, добавьте свою и передайте полный список в mirror(). Этот метод заменяет всё содержимое реестра, поэтому записи, не переданные в вызов, будут удалены:

from llmbroker import LLMConfig
from llmbroker.postgres import Registry

registry = Registry(pool)
mine = LLMConfig(
    name="my-gateway",
    base_url="https://gw.internal/v1",
    model="m",
    api_key_ref="MY_GATEWAY_KEY",
)
await registry.mirror([*await registry.load(), mine])

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

Первоначальное заполнение базы при развёртывании

Новая база данных не содержит моделей. Заполните её отдельной задачей при развёртывании, например рядом с alembic upgrade. Используйте ту же функцию создания брокера, что и приложение, чтобы строка подключения и связанные секреты настраивались в одном месте:

async with build_broker() as broker:  # функция создания брокера в приложении
    print(await broker.sync("freetier"))  # имя поддерживаемого набора

Вход в контекстный менеджер не инициализирует пул, поэтому задача работает и с пустой базой.

Запускайте код как одноразовую задачу: на этапе выпуска, в Kubernetes Job или в контейнере инициализации. После этого отдельная задача обновления не нужна: рабочие процессы проверяют поддерживаемый список примерно раз в сутки во время обычного вызова. Несколько процессов могут выполнять проверку безопасно, потому что используют один источник и одни ключи. Первый из них записывает изменения, а остальные обнаруживают, что обновлять уже нечего. Локальные копии отдельных процессов при этом не используются для перезаписи общего реестра.

sync принимает только имя поддерживаемого набора: передать путь к файлу или второй реестр нельзя. При использовании строки подключения по умолчанию выбран набор freetier. Если в registry= передан готовый объект, необходимо явно задать sync="freetier" или sync=None. Обновление в любом случае изменяет только записи, созданные предыдущими синхронизациями, и сохраняет пользовательские записи.

Развёртывание без автоматического доступа к сети

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

llmbroker.AsyncBroker("postgresql://host/db", sync_interval=None)  # в функции создания брокера

async with build_broker() as broker:
    report = await broker.sync()  # без аргумента: то, чему следует эта установка
    if report is not None:  # при обновлении только платного каталога отчёта нет
        print(llmbroker.format_report(report))

sync_interval=None отключает все автоматические сетевые запросы к каталогу: обновление бесплатного набора и платного каталога, используемого для псевдонимов из direct=. Пустой реестр также не заполняется автоматически. В этом случае возникнет EmptyRegistryError с указанием на задачу синхронизации, и брокер не станет обращаться к сети при первом пользовательском запросе.

sync() без аргумента обновляет набор, указанный в sync=. Если sync=None, обновляется только каталог платных моделей, необходимый для псевдонимов из direct=. В реестр при этом ничего не записывается, поэтому метод не возвращает отчёт.

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

При отключённом автоматическом обновлении список нужно поддерживать самостоятельно: доступность бесплатных моделей меняется без предупреждения. Запускайте задачу при каждом развёртывании и, при необходимости, по расписанию между выпусками. Стандартный интервал автоматической проверки составляет около суток и подходит как ориентир.

Перенос данных между хранилищами

Отдельной команды миграции нет. Данные можно загрузить из одного реестра и передать в другой в задаче развёртывания, где уже доступны обе строки подключения:

from llmbroker.mongodb import Registry as MongoRegistry
from llmbroker.postgres import Registry as PostgresRegistry

old = PostgresRegistry(old_pool)
new = MongoRegistry(new_db)
await new.mirror(await old.load())

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

Платная модель для прямого вызова задаётся при создании брокера: AsyncBroker(dsn, direct=["opus"]). В реестр она не записывается. Одна настройка в общей функции создания брокера применяется ко всем процессам. Каждый процесс периодически проверяет соответствие псевдонима текущей версии модели, поэтому долгоживущее приложение также получает обновления. См. Прямые вызовы модели.

sync приводит созданные им записи в соответствие с выбранным поддерживаемым набором: обновляет существующие, удаляет отсутствующие и добавляет новые. Пользовательские записи реестра не изменяются. Модель удаляется из поддерживаемого набора только после того, как её больше нельзя вызвать.

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

Отображение состояния пула

snapshot() одним вызовом возвращает состояние каждой модели и общие показатели пула. Изменения доступности llmbroker также записывает в журнал, поэтому системе мониторинга не обязательно регулярно опрашивать приложение. См. Наблюдение и журнал.

SQLite: общая база и WAL

llmbroker и приложение могут использовать один файл SQLite. Брокер создаёт только таблицы llmbroker_* и не изменяет остальные. Интеграция с Alembic исключает эти таблицы из автоматического создания миграций.

Значение PRAGMA user_version, которое используют многие средства миграции, llmbroker не изменяет. Версия его собственной схемы хранится в таблице llmbroker_schema_version. Удаление всех таблиц llmbroker_* удаляет и все данные llmbroker из файла.

llmbroker не изменяет journal_mode SQLite. Режим WAL сохраняется в самом файле и должен настраиваться приложением, которое им управляет. Включите WAL, если нескольким соединениям нужно одновременно читать и записывать данные.

Если приложение часто записывает данные, для брокера можно выделить отдельный файл .db, чтобы уменьшить конкуренцию за блокировку SQLite. Для отдельного файла режим WAL достаточно включить один раз:

sqlite3 broker.db 'PRAGMA journal_mode=WAL'

Это ограничение относится только к SQLite. PostgreSQL и MongoDB не используют общую файловую блокировку такого типа, поэтому база может быть общей с приложением. Отдельная схема или база данных остаётся вопросом организации.

Обновление llmbroker

llmbroker никогда не мигрирует существующую базу данных. Если новая версия меняет внутреннюю схему, брокер не запускается и возбуждает SchemaVersionError вместо чтения старых таблиц. Обновление в этом случае — сброс: таблицы llmbroker_* удаляются, и новая версия создаёт их заново.

Перед удалением нужно сохранить то, что не восстанавливается само. Реестр восстановит следующий sync(), журнал — это история, а вот ключи, которые хранит сам брокер, больше нигде не существуют, поэтому их выгрузка — обязательный шаг обновления, а не необязательный. Это касается только ключей в базе данных: ключи из переменных окружения, AWS Secrets Manager или Vault сброс не затрагивает.

Выгрузите ключи до установки новой версии. Новая версия отказывается читать старые таблицы, поэтому её API до них уже не доберётся — выгрузка выполняется, пока установлена прежняя версия:

import json
from pathlib import Path

from llmbroker.sqlite import Secrets  # или llmbroker.postgres / llmbroker.mongodb

secrets = Secrets("broker.db")
dump = {ref: await secrets.resolve(ref) for ref in sorted(await secrets.refs())}
Path("keys.json").write_text(json.dumps(dump))

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

sqlite3 broker.db 'SELECT ref, value FROM llmbroker_secrets'

Затем удалите таблицы, запустите новую версию и верните данные:

for ref, value in json.loads(Path("keys.json").read_text()).items():
    await secrets.set(ref, value)
await broker.sync("freetier")

После сброса пул стартует «холодным»: прошлые вызовы, история качества и действующие паузы моделей потеряны, и модели ранжируются заново — с первых вызовов нового развёртывания.

Ошибки запуска

До первого запроса могут возникнуть три ошибки, требующие разной обработки:

SyncRefusedError — sync() не применил обновление, поскольку после него реестр остался бы пустым. Реестр не изменён; поле report содержит запланированные изменения.
EmptyRegistryError — реестр ещё не заполнен. Нужно выполнить первоначальную настройку или синхронизацию. Ошибку вызывает первый вызов, которому нужен пул; чтобы узнать о ней при запуске, вызовите ensure_pool().
SchemaVersionError — версия схемы хранилища несовместима с текущей версией llmbroker. Поля found и expected содержат обнаруженную и ожидаемую версии. См. Обновление llmbroker.

Все три класса доступны непосредственно из модуля llmbroker, например llmbroker.SchemaVersionError, и наследуют LLMBrokerError, который, в свою очередь, наследует RuntimeError. Можно перехватывать конкретный класс или общий базовый класс:

try:
    models = broker.snapshot()
except llmbroker.EmptyRegistryError:
    models = {}  # показать пустое состояние вместо ошибки HTTP 500

Не скрывайте SchemaVersionError: сообщение содержит инструкции для администратора. Иначе несовместимая схема может выглядеть как отсутствие настроенных провайдеров. LLMBrokerError перехватывает все три перечисленные ошибки, а RuntimeError — также другие ошибки времени выполнения.

Ошибки выполнения пользовательского запроса наследуют другой класс — LLMRequestError. См. Когда ответить некому.

Закрытие брокера

Закрывайте брокер явно, если процесс создаёт несколько брокеров последовательно или использует внешнюю базу данных:

with llmbroker.Broker("broker.db") as broker:
    reply = broker.ask("...")

Для AsyncBroker используйте async with или await broker.aclose().

Закрытие брокера — блоком with, как выше, или вызовом close() — дожидается, пока его открытые потоки будут отменены и записаны в журнал вызовов. Для обычного скрипта закрытие необязательно: синхронный брокер, на который просто больше нет ссылок или который остался открытым при завершении процесса, освобождается без ожидания, и ответ, который в этот момент ещё передавался, может не попасть в журнал.

Журнал вызовов

Каждая попытка обращения к модели записывает имя модели, результат, сведения об использовании, trace_id и последующую оценку. Методы broker.calls(...) и broker.stats(...) читают журнал без инициализации пула, поэтому работают даже до заполнения реестра. См. Наблюдение и журнал.

Записи старше срока retention удаляются автоматически; значение по умолчанию — 90 дней. Этот параметр относится к хранилищу журнала, поэтому его нельзя задать в строке подключения брокера. Создайте объекты хранилищ явно и передайте retention в Store:

from datetime import timedelta

from llmbroker.postgres import Registry, Secrets, Store

broker = llmbroker.AsyncBroker(
    registry=Registry(pool),
    secrets=Secrets(pool),
    store=Store(pool, retention=timedelta(days=365)),
    sync="freetier",  # явно выбрать набор для обновления реестра
)

Один брокер для нескольких пользователей

Создавайте один брокер на весь процесс. Он хранит пул моделей, ключи, накопленные оценки и HTTP-клиент. Для отдельного пользователя можно создать лёгкий объект AsyncLLMs через for_scope(...). Он использует общий пул, но связывает записи журнала и пользовательские ключи с переданной областью данных. Создание такого объекта не обращается к хранилищам, поэтому его можно выполнять для каждого входящего запроса.

Ниже приведены четыре распространённых варианта.

Обычный скрипт. Вызывайте методы самого брокера; отдельная область данных не нужна:

broker = llmbroker.Broker()
print(broker.ask("hi").text)

Долгоживущий процесс с базой данных. Создайте брокер один раз при запуске приложения и закройте при остановке:

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.broker = llmbroker.AsyncBroker("postgresql://host/db")
    try:
        yield
    finally:
        await app.state.broker.aclose()

Несколько процессов с общими ключами. Каждый обработчик получает общий объект AsyncLLMs. Все запросы используют один пул и один набор ключей, а в каждом процессе создаётся один набор соединений:

def llms(request: Request) -> llmbroker.AsyncLLMs:
    return request.app.state.broker.llms


@app.post("/ask")
async def ask(prompt: str, llms: llmbroker.AsyncLLMs = Depends(llms)):
    return (await llms.ask(prompt)).text

Отдельный ключ для каждого пользователя. Изменяется только функция зависимости:

def llms(request: Request) -> llmbroker.AsyncLLMs:
    return request.app.state.broker.for_scope(request.headers["x-user-id"])

Пользовательский ключ хранится под именем <область>/<имя ключа>. Например, объект для области u-42 сначала запрашивает u-42/GROQ_API_KEY, а при его отсутствии использует общий GROQ_API_KEY. Общий ключ читается один раз и используется всеми областями. Пользовательский ключ можно сохранить в том же источнике, что и общие ключи: в переменной окружения, базе данных, AWS Secrets Manager или Vault.

Область — это произвольная строка, переданная в for_scope(...). llmbroker не придаёт ей отдельного смысла и не требует, чтобы она обязательно обозначала пользователя. Об особенностях символа / в именах Vault см. в разделе API-ключи.

Каждая запись журнала сохраняет область объекта, через который выполнен вызов. Историю одного пользователя можно получить через broker.for_scope(user).calls(...). Отдельного параметра scope= у calls() нет: область берётся из объекта. Методы broker.calls() и broker.stats() возвращают данные по всем областям.

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

Общие данные нескольких процессов

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

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

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

Alembic

Чтобы автогенерация миграций игнорировала таблицы llmbroker_*:

# alembic/env.py
import llmbroker.integrations.alembic

context.configure(
    connection=connection,
    target_metadata=target_metadata,
    include_object=llmbroker.integrations.alembic.include_object,
)

Если в приложении уже есть собственная функция include_object, объедините её с этой проверкой с помощью and.