L3.2. LLM API и SDK

Анатомия запроса и ответа

import anthropic

client = anthropic.Anthropic(max_retries=3, timeout=60.0)
resp = client.messages.create(
    model="claude-opus-5-5",
    max_tokens=16000,
    system="Ты консультант магазина фильтров для воды. Отвечай кратко.",
    output_config={"effort": "medium"},
    messages=[{"role": "user", "content": "Как часто менять картридж?"}],
)

В ответе важны: content (список блоков — текст, рассуждение, вызовы инструментов), stop_reason и usage.

Особенности текущих моделей, о которых спотыкаются. У claude-opus-5-5 рассуждение не отключается и тратит тот же max_tokens, поэтому маленький лимит (например, 300) часто даёт пустой или обрезанный ответ. Параметры temperature, top_p, top_k у этой модели не принимаются (ошибка 400): глубину регулирует effort. Это детали конкретного провайдера и версии — они записаны в технологическом срезе курса и меняются; при старте потока сверяйтесь с документацией.

История — это состояние, которое храните вы

API не помнит прошлых запросов. «Память» диалога — это список messages, который вы отправляете каждый раз целиком (MEM-01·F).

history: list[dict] = []

def ask(question: str) -> str:
    history.append({"role": "user", "content": question})
    resp = client.messages.create(model="claude-opus-5-5", max_tokens=16000, messages=history)
    answer = next(b.text for b in resp.content if b.type == "text")
    history.append({"role": "assistant", "content": answer})
    return answer

Следствия: каждый следующий запрос дороже предыдущего (вся история — входные токены); длинный диалог упирается в окно контекста; история пользователя — это его данные, и вы отвечаете за их хранение (L1.5, L6.5).

Стриминг

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

with client.messages.stream(
    model="claude-opus-5-5",
    max_tokens=64000,
    messages=[{"role": "user", "content": "Сравни три типа фильтров для квартиры."}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
    final = stream.get_final_message()       # полный ответ с usage — для учёта денег
print("\nтокенов:", final.usage.input_tokens, "+", final.usage.output_tokens)

Стриминг нужен ещё и технически: при большом max_tokens обычный запрос может не уложиться в HTTP-таймаут.

Ошибки, ретраи, лимиты

SDK уже повторяет 408, 409, 429, 5xx и сетевые ошибки с экспоненциальной паузой (по умолчанию 2 повтора, настраивается max_retries). Свой цикл повторов пишут, только если нужно что-то сверх этого: например, переключиться на запасную модель.

Экспоненциальная пауза с джиттером: 1 с, 2 с, 4 с… плюс случайная добавка. Без случайности тысяча клиентов после сбоя повторяет запрос синхронно и снова роняет сервер.

Деньги: считать до, считать после

Цена = входные токены × цена входа + выходные × цена выхода. Выходные токены дороже в несколько раз, и в них входит рассуждение.

PRICES = {  # $ за 1M токенов; сверяйте с прайсом при старте потока
    "claude-opus-5-5": (4.00, 20.00),
    "claude-sonnet-5-5": (2.00, 10.00),
    "claude-haiku-4-5": (1.00, 5.00),
}

def cost_usd(model: str, usage) -> float:
    p_in, p_out = PRICES[model]
    return (usage.input_tokens * p_in + usage.output_tokens * p_out) / 1_000_000

До запроса токены можно посчитать без генерации: client.messages.count_tokens(model=..., messages=..., system=...) вернёт input_tokens. Это нужно, чтобы не отправить документ, который не влезет в окно, и чтобы оценить бюджет пакетной обработки.

Пример расчёта: 3 000 входных и 1 500 выходных токенов на Opus 5.5 — 3 000 × 4 / 10⁶ + 1 500 × 20 / 10⁶ = $0,012 + $0,030 = $0,042. На 10 000 документов — $420. На Haiku 4.5 — $0,0105 за документ, $105 на всё. Будет ли малая модель достаточно хороша, решает eval (L3.7), а не интуиция.

Практикум: надёжный клиент со счётчиком

Модуль llm.py с функцией complete(prompt, *, model, effort, system=None) -> Completion, где Completion — dataclass с полями text, input_tokens, output_tokens, cost_usd, stop_reason, latency_ms. Требования:

• stop_reason == "refusal" и "max_tokens" возвращаются явно, а не прячутся;
• накопительный счётчик расходов за процесс и печать итога при выходе;
• провайдер echo (возвращает начало промпта, 0 токенов): тесты и разработка без сети;
• 3 теста на echo и 1 тест с подменой клиента, которая бросает anthropic.APIConnectionError: функция возвращает понятную ошибку, а не трассировку.