Рецепты

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

Примеры на Python, ключ читается из окружения:

export WAIBEE_API_KEY="sk_live_..."
pip install openai

Цикл вызова инструментов

Модель просит вызов, вы выполняете функцию, возвращаете результат, повторяете. Ключевая деталь это предел шагов: без него запутавшаяся модель будет ходить по кругу за ваш счёт. См. Вызов инструментов.

tool_loop.py

import json
import os

from openai import OpenAI

client = OpenAI(base_url="https://api.waibee.com/v1", api_key=os.environ["WAIBEE_API_KEY"])

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_order",
            "description": "Состояние заказа по его номеру",
            "parameters": {
                "type": "object",
                "properties": {"order_id": {"type": "string"}},
                "required": ["order_id"],
            },
        },
    }
]

def get_order(order_id: str) -> dict:
    return {"order_id": order_id, "status": "shipped", "eta_days": 2}

HANDLERS = {"get_order": get_order}
MAX_STEPS = 5

def call_tool(name: str, raw_arguments: str) -> str:
    """Результат для сообщения role=tool. Ошибку возвращаем модели, а не бросаем."""
    handler = HANDLERS.get(name)
    if handler is None:
        return json.dumps({"error": f"инструмента {name} не существует"}, ensure_ascii=False)
    try:
        arguments = json.loads(raw_arguments)
    except json.JSONDecodeError as error:
        return json.dumps({"error": f"аргументы не разобрались как JSON: {error}"}, ensure_ascii=False)
    return json.dumps(handler(**arguments), ensure_ascii=False)

def answer(question: str) -> str:
    messages = [{"role": "user", "content": question}]
    for _ in range(MAX_STEPS):
        choice = client.chat.completions.create(
            model="waibee/auto", messages=messages, tools=TOOLS
        ).choices[0]
        if choice.finish_reason != "tool_calls":
            return choice.message.content
        messages.append(choice.message)
        messages += [
            {
                "role": "tool",
                "tool_call_id": call.id,
                "content": call_tool(call.function.name, call.function.arguments),
            }
            for call in choice.message.tool_calls
        ]
    raise RuntimeError(f"инструменты вызываются по кругу, {MAX_STEPS} шагов исчерпаны")

print(answer("Что с заказом A-1042? Ответь одним предложением."))

Заказ A-1042 отправлен и будет доставлен через 2 дня.

Ответ модели добавляется в историю как есть, а на каждый вызов уходит отдельное сообщение role: "tool" с тем же tool_call_id.

Почему разбор аргументов обёрнут в try: arguments это строка, которую сочинила модель, и она бывает невалидной. Например, часть моделей каталога присылает JSON, дополнительно завёрнутый в одинарные кавычки. Ошибку разбора лучше вернуть модели тем же сообщением role: "tool": следующим ходом она обычно исправляет вызов сама, а цикл не падает.

Повторные вопросы к одному документу

Когда большой неизменный контекст повторяется в каждом запросе, отметьте его маркером кэша: первый запрос платит за запись, остальные читают дешевле. См. Кэширование промптов.

cached_context.py

import os
import pathlib

from openai import OpenAI

client = OpenAI(base_url="https://api.waibee.com/v1", api_key=os.environ["WAIBEE_API_KEY"])

POLICY = pathlib.Path("policy.txt").read_text(encoding="utf-8")


def ask(question: str) -> tuple[str, int]:
    response = client.chat.completions.create(
        model="anthropic/claude-haiku-4.5",
        messages=[
            {
                "role": "user",
                "content": [
                    # Кэшируемый блок идёт первым, меняющийся вопрос после него.
                    {"type": "text", "text": POLICY, "cache_control": {"type": "ephemeral"}},
                    {"type": "text", "text": f"{question} Ответь одной строкой."},
                ],
            }
        ],
    )
    # prompt_tokens_details отсутствует, когда провайдер не прислал разбивку.
    details = response.usage.prompt_tokens_details
    cached = getattr(details, "cached_tokens", 0) or 0
    return response.choices[0].message.content, cached


for question in ["Какой срок возврата?", "Кто оплачивает доставку возврата?"]:
    answer, cached = ask(question)
    print(f"из кэша {cached:>6} токенов | {answer.splitlines()[0][:50]}")

из кэша      0 токенов | Возврат в течение 14 дней. Доставку возврата оплач
из кэша  24333 токенов | Покупатель оплачивает доставку возврата.

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

Извлечение данных по схеме

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

extract.py

import json
import os

from openai import OpenAI

client = OpenAI(base_url="https://api.waibee.com/v1", api_key=os.environ["WAIBEE_API_KEY"])

INVOICE_SCHEMA = {
    "type": "object",
    "properties": {
        "company": {"type": "string"},
        "amount": {"type": "number"},
        "currency": {"type": "string", "enum": ["RUB", "USD", "EUR"]},
        # Явный null: иначе модель придумает дату, которой в тексте нет.
        "due_date": {"type": ["string", "null"], "description": "ISO 8601 или null"},
    },
    "required": ["company", "amount", "currency", "due_date"],
    "additionalProperties": False,
}


def extract_invoice(text: str) -> dict:
    choice = client.chat.completions.create(
        model="openai/gpt-5.4-mini",
        messages=[{"role": "user", "content": text}],
        response_format={
            "type": "json_schema",
            "json_schema": {"name": "invoice", "strict": True, "schema": INVOICE_SCHEMA},
        },
    ).choices[0]
    if choice.finish_reason == "length":
        raise ValueError("ответ обрезан лимитом токенов, JSON неполный")
    invoice = json.loads(choice.message.content)
    if invoice["amount"] <= 0:  # схема проверяет форму, а не правдоподобность
        raise ValueError(f"неправдоподобная сумма: {invoice['amount']}")
    return invoice


print(extract_invoice("Счёт от ООО Ромашка на 154 200 рублей, оплатить до 15 августа 2026."))
print(extract_invoice("Invoice from Acme Inc, 1200 EUR, no due date specified."))

{'company': 'ООО Ромашка', 'amount': 154200, 'currency': 'RUB', 'due_date': '2026-08-15'}
{'company': 'Acme Inc', 'amount': 1200, 'currency': 'EUR', 'due_date': None}

Картинка в структурированные данные

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

receipt.py

import base64
import json
import os
import pathlib

from openai import OpenAI

client = OpenAI(base_url="https://api.waibee.com/v1", api_key=os.environ["WAIBEE_API_KEY"])

RECEIPT_SCHEMA = {
    "type": "object",
    "properties": {
        "merchant": {"type": "string"},
        "total": {"type": "number"},
        "currency": {"type": "string"},
        "items": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {"name": {"type": "string"}, "price": {"type": "number"}},
                "required": ["name", "price"],
                "additionalProperties": False,
            },
        },
    },
    "required": ["merchant", "total", "currency", "items"],
    "additionalProperties": False,
}


def read_receipt(path: pathlib.Path) -> dict:
    image = base64.b64encode(path.read_bytes()).decode()
    choice = client.chat.completions.create(
        model="google/gemini-3.6-flash",
        messages=[
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": "Извлеки данные чека по схеме."},
                    {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{image}"}},
                ],
            }
        ],
        response_format={
            "type": "json_schema",
            "json_schema": {"name": "receipt", "strict": True, "schema": RECEIPT_SCHEMA},
        },
    ).choices[0]
    receipt = json.loads(choice.message.content)
    # Модель читает картинку, а не считает: сумма позиций может не сойтись с итогом.
    if abs(sum(item["price"] for item in receipt["items"]) - receipt["total"]) > 0.01:
        raise ValueError(f"сумма позиций не сходится с итогом {receipt['total']}")
    return receipt


print(read_receipt(pathlib.Path("receipt.png")))

{'merchant': 'COFFEE HOUSE', 'total': 10.5, 'currency': 'EUR', 'items': [{'name': 'Espresso', 'price': 2.8}, {'name': 'Croissant', 'price': 3.2}, {'name': 'Orange juice', 'price': 4.5}]}

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

Извлечение данных из PDF

Документ вкладывается прямо в сообщение, отдельной загрузки файлов нет. Обрабатывать лучше по одному документу за запрос: так дешевле и точнее, а сбой на одном файле не роняет остальные. См. Изображения и PDF.

invoices.py

import base64
import os
import pathlib

from openai import OpenAI, OpenAIError

client = OpenAI(base_url="https://api.waibee.com/v1", api_key=os.environ["WAIBEE_API_KEY"])


def read_invoice(path: pathlib.Path) -> str:
    document = base64.b64encode(path.read_bytes()).decode()
    response = client.chat.completions.create(
        model="anthropic/claude-haiku-4.5",
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "file",
                        "file": {
                            "filename": path.name,
                            "file_data": f"data:application/pdf;base64,{document}",
                        },
                    },
                    {"type": "text", "text": "Верни сумму и валюту документа, без пояснений."},
                ],
            }
        ],
    )
    return response.choices[0].message.content.strip()


for path in sorted(pathlib.Path("invoices").glob("*.pdf")):
    try:
        print(f"{path.name}: {read_invoice(path)}")
    except OpenAIError as error:  # один сбойный документ не должен останавливать обработку
        print(f"{path.name}: пропущен, {error}")

a.pdf: 154200 RUB
b.pdf: 1200 EUR

Многостраничные документы стоит резать и спрашивать по разделу: дешевле и точнее, чем сотня страниц одним запросом.

Задача, где нужен максимум рассуждений

Уровень усилий max доступен только на POST /v1/messages, поэтому это единственный рецепт на странице, где нужен SDK Anthropic (pip install anthropic). Полезен там, где ошибка дороже ожидания: разбор запутанной логики, вывод формулы, ревью сложного инварианта. См. Reasoning.

hard.py

import os

import anthropic

# Base URL без /v1: путь SDK добавляет сам.
client = anthropic.Anthropic(
    base_url="https://api.waibee.com",
    api_key=os.environ["WAIBEE_API_KEY"],
    timeout=600.0,
)

TASK = (
    "В корзине 12 шаров: 5 красных, 4 синих, 3 белых. Тянем три подряд без возврата. "
    "Какова вероятность, что все три разного цвета? Ответ одной строкой: несократимая дробь "
    "и десятичное значение."
)

response = client.messages.create(
    model="anthropic/claude-sonnet-5",
    max_tokens=8192,
    # Уровень max доступен только на /v1/messages; SDK передаёт поле как есть.
    extra_body={"output_config": {"effort": "max"}},
    messages=[{"role": "user", "content": TASK}],
)

answer = "".join(block.text for block in response.content if block.type == "text")
details = getattr(response.usage, "output_tokens_details", None)
print(answer.strip())
print(f"[рассуждений {getattr(details, 'thinking_tokens', 0)} из {response.usage.output_tokens} токенов вывода]")

**3/11 ≈ 0,2727 (0,(27))**
[рассуждений 107 из 402 токенов вывода]

Длинный таймаут здесь обязателен: на max модель молчит, пока думает, и короткий read-таймаут оборвёт живой запрос. Блоки thinking идут в content перед текстом, поэтому ответ собирается только из блоков с типом text.

Стриминг ответа пользователю

Текст печатается по мере генерации, а счётчики токенов приходят в последнем кадре, где содержимого уже нет. См. Стриминг.

stream.py

import os

from openai import OpenAI

client = OpenAI(base_url="https://api.waibee.com/v1", api_key=os.environ["WAIBEE_API_KEY"])

stream = client.chat.completions.create(
    model="anthropic/claude-haiku-4.5",
    messages=[{"role": "user", "content": "Три способа кэшировать HTTP-ответы, по строке на способ."}],
    stream=True,
)

usage = None
for chunk in stream:
    if chunk.usage:  # последний кадр: только счётчики, без текста
        usage = chunk.usage
    for choice in chunk.choices:
        if choice.delta.content:
            print(choice.delta.content, end="", flush=True)
print()

if usage:
    print(f"[токены: ввод {usage.prompt_tokens}, вывод {usage.completion_tokens}]")

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

Много запросов параллельно

Предел одновременных запросов задаёте вы: без него клиент упирается в лимиты и получает 429 или 503 вместо результата. См. Лимиты.

batch.py

import asyncio
import os

from openai import AsyncOpenAI

client = AsyncOpenAI(base_url="https://api.waibee.com/v1", api_key=os.environ["WAIBEE_API_KEY"])

MAX_CONCURRENCY = 4

async def summarize(text: str, limiter: asyncio.Semaphore) -> str:
    async with limiter:
        response = await client.chat.completions.create(
            model="anthropic/claude-haiku-4.5",
            messages=[{"role": "user", "content": f"Одним предложением, о чём это: {text}"}],
        )
        return response.choices[0].message.content

async def summarize_all(texts: list[str]) -> list[str | BaseException]:
    limiter = asyncio.Semaphore(MAX_CONCURRENCY)
    # return_exceptions: один упавший запрос не должен обнулить всю пачку результатов.
    return await asyncio.gather(
        *(summarize(text, limiter) for text in texts), return_exceptions=True
    )

TICKETS = [
    "Клиент не может войти: пароль сброшен, письмо не приходит.",
    "Оплата прошла дважды, нужен возврат одной транзакции.",
    "Приложение падает при открытии профиля на Android 13.",
]

for ticket, result in zip(TICKETS, asyncio.run(summarize_all(TICKETS))):
    if isinstance(result, BaseException):
        print(f"! {ticket[:30]}: {result}")
    else:
        print(f"- {result}")

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

Резервная модель при сбое

Полей маршрутизации в запросе нет, поэтому цепочка резерва живёт на стороне клиента. Различайте случаи: недоступность модели лечится переходом на следующую, а ошибка в самом запросе повторится на любой. См. Ошибки.

fallback.py

import os

from openai import APIStatusError, OpenAI

client = OpenAI(base_url="https://api.waibee.com/v1", api_key=os.environ["WAIBEE_API_KEY"])

# Порядок важен: сначала предпочтительная модель, дальше запасные.
MODELS = ["anthropic/claude-sonnet-4.6", "openai/gpt-5.4-mini", "google/gemini-3.6-flash"]
FALLBACK_STATUSES = frozenset({404, 429, 502, 503})


def complete(messages: list[dict], models: list[str] = MODELS):
    for model in models:
        try:
            return client.chat.completions.create(model=model, messages=messages)
        except APIStatusError as error:
            if error.status_code not in FALLBACK_STATUSES:
                raise  # ошибка в самом запросе: другая модель не спасёт
    raise RuntimeError(f"ни одна модель не ответила: {', '.join(models)}")


response = complete([{"role": "user", "content": "Ответь одним словом: резерв"}])
print(f"{response.model}: {response.choices[0].message.content}")

Альтернатива без своей цепочки это waibee/auto: роутер сам выбирает модель под запрос, см. Autorouting.

Таймауты и повторы в сервисе

Что отличает продакшн от примера: длинный read-таймаут (reasoning-модели молчат минутами), собственные повторы с решением по коду ошибки и чтение счётчиков. См. Лимиты и Токены и usage.

client.py

import os
import random
import time

import httpx
from openai import APIConnectionError, APIStatusError, APITimeoutError, OpenAI
from openai.types.chat import ChatCompletion

client = OpenAI(
    base_url="https://api.waibee.com/v1",
    api_key=os.environ["WAIBEE_API_KEY"],
    # Reasoning-модели молчат минутами: короткий read-таймаут рвёт живой запрос.
    timeout=httpx.Timeout(connect=10.0, read=600.0, write=30.0, pool=10.0),
    max_retries=0,  # повторы ведём сами, чтобы решать по коду ошибки
)

MAX_ATTEMPTS = 3
RETRIABLE_STATUSES = frozenset({429, 502, 503})

def is_retriable(error: APIStatusError) -> bool:
    # Лимит расходов ключа снимется только со сбросом счётчика: повтор не поможет.
    return error.status_code in RETRIABLE_STATUSES and error.code != "api_key_limit_exceeded"

def complete(**kwargs) -> ChatCompletion:
    for attempt in range(MAX_ATTEMPTS):
        last_attempt = attempt == MAX_ATTEMPTS - 1
        try:
            return client.chat.completions.create(**kwargs)
        except APIStatusError as error:
            if last_attempt or not is_retriable(error):
                raise
        except (APITimeoutError, APIConnectionError):
            if last_attempt:
                raise
        # Случайная добавка разносит повторы клиентов, чтобы они не сошлись в одну волну.
        time.sleep(2**attempt + random.random())
    raise AssertionError("недостижимо: цикл либо вернул ответ, либо бросил исключение")

response = complete(
    model="waibee/auto",
    messages=[{"role": "user", "content": "Ответь одним словом: готово"}],
)
usage = response.usage
cached = getattr(usage.prompt_tokens_details, "cached_tokens", 0) or 0
print(
    f"модель {response.model}: ввод {usage.prompt_tokens}, вывод {usage.completion_tokens}, "
    f"из кэша {cached}"
)

модель anthropic/claude-haiku-4.5: ввод 18, вывод 6, из кэша 0

Случайная добавка к задержке нужна, чтобы повторы клиентов не сошлись в одну волну. И логируйте response.model вместе с usage: на waibee/auto это единственный способ потом понять, какая модель отвечала и сколько это стоило.