Урок 5. Стриминг: ответ по мере генерации

До сих пор мы вызывали client.messages.create() и получали ответ целиком. Работает — но есть нюанс: содержимое приходит только когда модель дописала последнее слово. Пока она пишет эссе на 800 токенов, пользователь смотрит в пустой экран.

Стриминг решает ровно это: текст приходит кусочками по мере генерации. Именно так устроен claude.ai — вы видите, как ответ печатается по словам, а не ждёте его целиком.

stream=True

Включить стриминг просто — один параметр:

stream = client.messages.create(
    messages=[
        {"role": "user", "content": "Напиши предложение из трёх слов. Без вступления — только три слова."}
    ],
    model="claude-haiku-4-5",
    max_tokens=100,
    temperature=0,
    stream=True,
)

А вот дальше начинается интересное. Если напечатать сам stream, вы увидите не текст, а объект-генератор. Он отдаёт server-sent events (SSE) — отдельные события — по мере того, как они приходят от API. Данные больше не лежат в одном готовом куске, по ним нужно пройти циклом:

for event in stream:
    print(event)

На экран посыплется десяток разнотипных объектов. Разберёмся, что это.

Типы событий стрима

Каждый стрим содержит события в фиксированном порядке:

• MessageStartEvent — сообщение с пустым содержимым (старт).
• Серия блоков контента, каждый из которых состоит из:
  • ContentBlockStartEvent — блок открылся;
  • одного или нескольких ContentBlockDeltaEvent — собственно текст по кусочкам;
  • ContentBlockStopEvent — блок закрылся.
• Один или несколько MessageDeltaEvent — изменения верхнего уровня в итоговом сообщении.
• Финальный MessageStopEvent.

Весь сгенерированный текст живёт в ContentBlockDeltaEvent — у них type равен "content_block_delta", а сам текст лежит в delta.text. Отфильтруем только их:

for event in stream:
    if event.type == "content_block_delta":
        print(event.delta.text)

Текст появился, но читать его неудобно — каждый кусочек на своей строке. Двум аргументам print() тут есть что сказать:

• end="" — не добавлять перевод строки после каждого кусочка, чтобы всё печаталось в одну строку;
• flush=True — выводить сразу, не дожидаясь заполнения буфера. Без этого «эффект печатной машинки» пропадёт.

stream = client.messages.create(
    messages=[{"role": "user", "content": "Как работают большие языковые модели?"}],
    model="claude-haiku-4-5",
    max_tokens=1000,
    temperature=0,
    stream=True,
)
for event in stream:
    if event.type == "content_block_delta":
        print(event.delta.text, flush=True, end="")

На длинном ответе разница видна сразу: текст ползёт по экрану, а не появляется через несколько секунд стеной.

Токены внутри стрима

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

• MessageStartEvent — сколько токенов ушло на промпт (event.message.usage.input_tokens);
• MessageDeltaEvent — сколько токенов сгенерировала модель (event.usage.output_tokens).

for event in stream:
    if event.type == "message_start":
        input_tokens = event.message.usage.input_tokens
        print(f"Токенов на входе: {input_tokens}", flush=True)
        print("========================")
    elif event.type == "content_block_delta":
        print(event.delta.text, flush=True, end="")
    elif event.type == "message_delta":
        output_tokens = event.usage.output_tokens
        print("\n========================", flush=True)
        print(f"Токенов сгенерировано: {output_tokens}", flush=True)

Что ещё может прилететь

• Ping events — служебные пинги, их в стриме может быть сколько угодно. Просто игнорируйте.
• Error events — иногда в поток приходят ошибки. Например, в часы пиковой нагрузки — overloaded_error, который в обычном (нестриминговом) запросе был бы HTTP 529.

Событие-ошибка выглядит так:

event: error
data: {"type": "error", "error": {"type": "overloaded_error", "message": "Overloaded"}}

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

Time to first token

Главная причина использовать стриминг — time to first token (TTFT), время до первого кусочка контента. Замерим на одном и том же запросе (длинное эссе, потолок 500 токенов).

import time

def measure_streaming_ttft():
    start_time = time.time()
    stream = client.messages.create(
        max_tokens=500,
        messages=[{"role": "user", "content": "Напиши длинное эссе про историю Американской революции"}],
        temperature=0,
        model="claude-haiku-4-5",
        stream=True,
    )
    have_received_first_token = False
    for event in stream:
        if event.type == "content_block_delta":
            if not have_received_first_token:
                ttft = time.time() - start_time
                have_received_first_token = True
            print(event.delta.text, flush=True, end="")
        elif event.type == "message_delta":
            total_time = time.time() - start_time

    print(f"\nДо первого токена: {ttft:.3f} с", flush=True)
    print(f"До полного ответа: {total_time:.3f} с", flush=True)

Результаты у авторов курса получились такие:

• Без стриминга: первый токен — 4.194 с, полный ответ — 4.194 с (это одно и то же событие).
• Со стримингом: первый токен — 0.492 с, полный ответ — 4.274 с.

И это Haiku, самая быстрая модель, на 500 токенах. На тысяче токенов через Opus разрыв становится неприличным: 47 секунд до первого токена без стриминга против 1.8 секунды со стримингом.

Хелперы messages.stream

Ручная проверка event.type — рабочий, но нудный способ. В Python SDK есть удобная альтернатива: вместо client.messages.create(..., stream=True) использовать client.messages.stream().

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

Два главных подарка:

• stream.text_stream — итератор только по текстовым дельтам. Никаких проверок типа события;
• get_final_message() — итоговое накопленное сообщение, когда стрим дочитан до конца. Удобно, когда нужно и печатать по кусочкам, и иметь полный текст в переменной. Собрать его вручную тоже можно — но зачем.

from anthropic import AsyncAnthropic

client = AsyncAnthropic()

async def streaming_with_helpers():
    async with client.messages.stream(
        max_tokens=1024,
        messages=[{"role": "user", "content": "Напиши сонет про орхидеи"}],
        model="claude-sonnet-5",
    ) as stream:
        async for text in stream.text_stream:
            print(text, end="", flush=True)

    final_message = await stream.get_final_message()
    print("\n\nСТРИМ ЗАКОНЧЕН. ИТОГОВОЕ СООБЩЕНИЕ:")
    print(final_message.to_json())

await streaming_with_helpers()

Обработчики событий

С client.messages.stream() можно ещё и повесить свои обработчики — на любое событие или только на текст. Определяем класс-наследник AsyncMessageStream и передаём его в аргумент event_handler:

• on_text — срабатывает, когда накапливается текстовый ContentBlock. Первый аргумент — дельта текста, второй — уже накопленный текст;
• on_stream_event — срабатывает на любое событие от API.

from anthropic import AsyncAnthropic, AsyncMessageStream

client = AsyncAnthropic()

class MyStream(AsyncMessageStream):
    async def on_text(self, text, snapshot):
        # только на текстовых дельтах
        print(text, flush=True, end="")

    async def on_stream_event(self, event):
        # на любом событии стрима
        print("событие:", event.type)

async def streaming_events_demo():
    async with client.messages.stream(
        max_tokens=1024,
        messages=[{"role": "user", "content": "Напиши стихотворение из пяти слов"}],
        model="claude-sonnet-5",
        event_handler=MyStream,
    ) as stream:
        message = await stream.get_final_message()
        print("итоговое сообщение:", message.to_json())

await streaming_events_demo()

Другие доступные обработчики:

• on_message(message) — накоплен полный объект Message (соответствует SSE message_stop);
• on_content_block(content_block) — накоплен полный ContentBlock (соответствует SSE content_block_stop);
• on_exception(exception) — во время чтения стрима возникло исключение;
• on_timeout() — запрос отвалился по таймауту;
• on_end() — последнее событие стрима.

Упражнения

Упражнение 5.1 — Чат-бот со стримингом

Напишите простого консольного чат-бота, который держит историю диалога и печатает ответы Claude по мере генерации. Выход по слову quit. Запускать лучше отдельным скриптом, а не в ноутбуке — так эффект печатной машинки виден честно.

from anthropic import Anthropic

client = Anthropic()

def chat_with_claude():
    print("Чат с Claude. Напишите 'quit', чтобы выйти.")
    conversation = []

    while True:
        user_input = input("Вы: ")
        if user_input.lower() == "quit":
            print("Пока!")
            break

        conversation.append({"role": "user", "content": user_input})
        print("Claude: ", end="", flush=True)

        stream = client.messages.create(
            model="claude-haiku-4-5",
            max_tokens=1000,
            messages=conversation,
            stream=True,
        )

        assistant_response = ""
        for chunk in stream:
            if chunk.type == "content_block_delta":
                content = chunk.delta.text
                print(content, end="", flush=True)
                assistant_response += content

        print()  # перевод строки после ответа
        conversation.append({"role": "assistant", "content": assistant_response})

if __name__ == "__main__":
    chat_with_claude()

Упражнение 5.2 — Тот же бот на хелперах

Перепишите цикл чтения стрима из упражнения 5.1 через client.messages.stream(), чтобы не проверять chunk.type вручную, а полный ответ брать готовым.

with client.messages.stream(
    model="claude-haiku-4-5",
    max_tokens=1000,
    messages=conversation,
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

final_message = stream.get_final_message()
conversation.append({
    "role": "assistant",
    "content": final_message.content[0].text,
})

Что запомнить

• Без стриминга ответ приходит целиком и только когда дописан. Со стримингом — кусочками, по мере генерации.
• Включается параметром stream=True; в ответ вы получаете генератор server-sent events.
• Порядок событий: MessageStartEvent → (ContentBlockStartEvent → ContentBlockDeltaEvent… → ContentBlockStopEvent) → MessageDeltaEvent → MessageStopEvent.
• Текст живёт в content_block_delta, в поле delta.text. Печатайте с end="" и flush=True.
• Токены: входные — в message_start, выходные — в message_delta.
• Стриминг режет time to first token, но не общее время генерации.
• client.messages.stream() даёт text_stream, get_final_message() и обработчики событий — руками разбирать типы событий больше не нужно.