Предварительный просмотр ответов с помощью дельт событий

По умолчанию текст ответа агента поступает в поток событий сессии в виде буферизованных событий agent.message. Каждое из них отправляется только после завершения запроса к модели, который его создал. «Event deltas» (дельты событий) позволяют отображать этот текст постепенно, в виде живого предварительного просмотра, пока модель ещё генерирует его.

Предварительный просмотр — это вспомогательное средство отображения, работающее по принципу «best effort» (по мере возможности), а буферизованное agent.message всегда является авторитетной записью. Клиент, который игнорирует предварительный просмотр, всё равно получает полный и корректный поток.

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

Предварительный просмотр включается отдельно для каждого подключения к потоку. Добавьте параметр запроса event_deltas[] к потоку, который вы читаете, и повторите его по одному разу для каждого типа событий, для которого вам нужен предварительный просмотр. Допустимые значения — agent.message и agent.thinking. Любое другое значение возвращает ошибку 400, как и запрос с более чем 100 значениями.

Обе конечные точки потоков принимают этот параметр:

• Поток уровня сессии: GET /v1/sessions/{session_id}/events/stream
• Поток событий потока выполнения сессии: GET /v1/sessions/{session_id}/threads/{thread_id}/stream

Предварительный просмотр субагента появляется в собственном потоке событий потока выполнения этого субагента.

[] — это шаблон подстановки оболочки (shell glob), поэтому заключайте URL в кавычки всякий раз, когда формируете запрос в оболочке. В примерах скобки закодированы в процентной кодировке как %5B%5D, что также работает.

События предварительного просмотра

Когда начинается событие с предварительным просмотром, поток отправляет event_start, содержащее тип и id предстоящего события:

{
  "type": "event_start",
  "event": {
    "type": "agent.message",
    "id": "sevt_01abc..."
  }
}

Для agent.message за началом следуют события event_delta, содержащие инкрементальный текст. Каждая дельта указывает событие, которое она дополняет, в event_id, а блок содержимого, который она дополняет, — в delta.index:

{
  "type": "event_delta",
  "event_id": "sevt_01abc...",
  "delta": {
    "type": "content_delta",
    "index": 0,
    "content": {
      "type": "text",
      "text": "Here is the summary"
    }
  }
}

Для agent.thinking отправляется только event_start как сигнал о том, что начался блок размышлений. События event_delta за ним не следуют. Буферизованное событие agent.thinking, завершающее предварительный просмотр, является сигналом прогресса и не содержит содержимого размышлений.

В отличие от сохраняемых событий, у event_start и event_delta нет собственных id или processed_at. Единственный идентификатор, который они содержат, — это id события, для которого они показывают предварительный просмотр. Их строки типов также являются исключением из соглашения об именовании {domain}.{action} для сохраняемых событий.

Накопление и согласование

Каждый SDK, поддерживающий дельты событий, включает вспомогательный аккумулятор, который выполняет учёт index за вас. Ручной шаблон из этого раздела работает на любом языке, когда вам нужен собственный учёт. Применяйте его к сгенерированным типам событий.

В ручном шаблоне храните текст предварительного просмотра во временном сопоставлении с ключом (event_id, index) и считайте буферизованное событие записью. Согласовывайте их для каждого запроса к модели.

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

1. span.model_request_start
2. event_start
3. События event_delta
4. Буферизованное agent.message
5. span.model_request_end (на вкладке «События span»)

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

event_start     {"event": {"type": "agent.message", "id": "sevt_01abc..."}}
event_delta     {"event_id": "sevt_01abc...", "delta": {"type": "content_delta", "index": 0, "content": {"type": "text", "text": "..."}}}
...
agent.message   {"id": "sevt_01abc...", "content": [...]}

Строка event_delta повторяется по одному разу для каждого фрагмента текста. Обрабатывайте каждое событие по мере поступления:

1. При event_start запомните объявленный id. Идентификаторы всегда совпадают: event_start.event.id, каждый event_delta.event_id и id буферизованного agent.message имеют одно и то же значение.
2. При каждом event_delta добавляйте delta.content.text к записи по ключу (event_id, delta.index) и отображайте текущий текст. Первая дельта для index создаёт эту запись.
3. Когда поступает буферизованное agent.message, сопоставьте его по id, отбросьте накопленный предварительный просмотр и вместо него отобразите содержимое сообщения.
4. При span.model_request_end закройте любой предварительный просмотр, который не был согласован со своим буферизованным событием. Больше дельт для него не поступит. Если в ходе возникает ошибка или он прерывается, буферизованное событие может так и не поступить, но span.model_request_end всё равно поступит.

Шаблон опирается на две гарантии:

• Конкатенация дельт предварительного просмотра в порядке поступления с ключом (event_id, index) даёт префикс content[index].text в буферизованном событии. Это не обязательно весь текст, поскольку дельты могут быть отброшены под нагрузкой.
• Подключение отправляет не более одного event_start на каждый event_id, а буферизованное событие — последнее, что это подключение доставляет для данного id.

Вспомогательные аккумуляторы SDK

Вспомогательный аккумулятор каждого SDK выполняет учёт index. Вспомогательные средства для Go, Java, Ruby и C# также используют id события в качестве ключа накапливаемого предварительного просмотра. При использовании вспомогательных средств для Python, TypeScript и PHP ведите это сопоставление самостоятельно и добавляйте каждую дельту в запись для её id.

Следующие примеры включают предварительный просмотр agent.message и согласовывают его с буферизованным событием:

# Снимки предпросмотра, с ключом по id события. accumulate_managed_agents_event сворачивает каждое
# событие event_start / event_delta в снимок agent.message; буферизованное
# agent.message заменяет его.
previews: dict[str, BetaManagedAgentsAgentMessageEvent] = {}

# Включаем предпросмотр agent.message для этого соединения
with client.beta.sessions.events.stream(
    session.id, event_deltas=["agent.message"]
) as stream:
    client.beta.sessions.events.send(
        session.id,
        events=[
            {
                "type": "user.message",
                "content": [{"type": "text", "text": "Describe the repo in one sentence."}],
            },
        ],
    )

    for event in stream:
        match event.type:
            case "event_start":
                snapshot = accumulate_managed_agents_event(None, event)
                if snapshot is not None:
                    previews[event.event.id] = snapshot
                print(f"event_start             {event.event.type} {event.event.id}")
            case "event_delta":
                preview = accumulate_managed_agents_event(previews.get(event.event_id), event)
                if preview is not None:
                    previews[event.event_id] = preview
                    text = "".join(block.text for block in preview.content)
                    print(f"event_delta             preview: {text!r}")
            case "agent.message":
                # Буферизованное событие — это итоговая запись: оно заменяет и закрывает предпросмотр
                preview = accumulate_managed_agents_event(previews.pop(event.id, None), event)
                text = "".join(block.text for block in preview.content)
                print(f"agent.message           {event.id} {text!r}")
            case "span.model_request_end":
                # Дельт больше не будет. Закрываем все предпросмотры, чьё
                # буферизованное событие так и не пришло.
                for event_id in previews:
                    print(f"span.model_request_end  closing preview for {event_id}")
                previews.clear()
            case "session.status_idle":
                break

Предварительный просмотр событий потоков выполнения сессии

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

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

Путь к потоку событий потока выполнения заканчивается на /threads/{thread_id}/stream. /events/stream существует только на уровне сессии, поэтому конечной точки /threads/{thread_id}/events/stream нет.

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

# Получаем список тредов сессии и выбираем дочерний: у дочерних тредов parent_thread_id
# не равен null, а у основного треда parent_thread_id равен null.
child_thread = next(
    thread
    for thread in client.beta.sessions.threads.list(session.id)
    if thread.parent_thread_id is not None
)

# Поток дочернего треда принимает тот же параметр event_deltas, что и
# поток сессии.
with client.beta.sessions.threads.events.stream(
    child_thread.id,
    session_id=session.id,
    event_deltas=["agent.message"],
) as stream:
    for event in stream:
        match event.type:
            case "event_delta":
                print(event.delta.content.text, end="")
            case "agent.message":
                # Буферизованное событие — авторитетная запись; отображаем его содержимое
                print()
                for block in event.content:
                    if block.type == "text":
                        print(block.text, end="")
                print()
            case "session.thread_status_idle":
                break

Цикл чтения завершается при session.thread_status_idle — событии, отправляемом, когда ход потока выполнения сессии завершается и поток выполнения переходит в режим ожидания.

Ограничения

• Без гарантий доставки: под нагрузкой сервер может отбрасывать дельты для события. В этом случае вы получаете непрерывный префикс текста, после чего дельты для этого события больше не поступают. Буферизованное agent.message всё равно поступает полностью. Никогда не считайте накопленный предварительный просмотр окончательным.
• Нет повторной передачи при переподключении: дельты доставляются только тому подключению, которое их включило, пока оно открыто. Это относится как к потоку уровня сессии, так и к каждому потоку выполнения сессии. Подключение, открытое после начала запроса к модели, не получает дельт для этого выполняющегося события. Запросить пропущенные дельты повторно невозможно.
• Один поток выполнения, только текст: предварительный просмотр охватывает текст ассистента в потоке выполнения, который читает подключение. Использование инструментов, результаты инструментов и результаты MCP никогда не показываются в предварительном просмотре.
• Никогда не сохраняются: event_start и event_delta существуют только в живом потоке. Они не появляются в истории событий сессии (GET /v1/sessions/{session_id}/events) или в истории событий какого-либо потока выполнения сессии.

Устранение неполадок с предварительным просмотром

Следующие шаги