イベントデルタでレスポンスをプレビューする

デフォルトでは、エージェントのレスポンステキストは、バッファリングされた agent.message イベントとしてセッションイベントストリームに届きます。各イベントは、それを生成したモデルリクエストが完了した後にのみ発行されます。「event delta」（イベントデルタ）を使用すると、モデルがまだテキストを生成している間に、そのテキストをライブプレビューとして段階的にレンダリングできます。

プレビューはベストエフォートの表示補助であり、バッファリングされた agent.message が常に正式な記録です。プレビューを無視するクライアントでも、完全で正しいストリームを受信します。

プレビューをオプトインする

プレビューはストリーム接続ごとのオプトインです。読み取っているストリームに event_deltas[] クエリパラメータを追加し、プレビューしたいイベントタイプごとに1回ずつ繰り返します。受け付けられる値は agent.message と agent.thinking です。それ以外の値を指定すると400エラーが返されます。100個を超える値を含むリクエストも同様です。

両方のストリームエンドポイントがこのパラメータを受け付けます。

• セッションレベルのストリーム： GET /v1/sessions/{session_id}/events/stream
• セッションスレッドのストリーム： GET /v1/sessions/{session_id}/threads/{thread_id}/stream

サブエージェントのプレビューは、そのサブエージェント自身のスレッドストリームに表示されます。

[] はシェルのグロブパターンであるため、シェルでリクエストを組み立てる場合は常にURLを引用符で囲んでください。例では角括弧を %5B%5D としてパーセントエンコードしていますが、これも機能します。

プレビューイベント

プレビュー対象のイベントが始まると、ストリームはこれから生成されるイベントのタイプと id を持つ event_start を発行します。

{
  "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) をキーとする一時的なマップに保持し、バッファリングされたイベントを記録として扱います。この2つをモデルリクエストごとに照合します。

ターンは単一の session.status_running イベントで始まります。正常に完了するターンでは、各モデルリクエストが次のイベントを順番に生成します。

1. span.model_request_start
2. event_start
3. event_delta イベント
4. バッファリングされた agent.message
5. span.model_request_end（スパンイベントタブ内）

ワイヤー上では、これはそのシーケンスのうちプレビューされる部分であり、接続上の他のバッファリングされたイベントと混在して届きます。

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回繰り返されます。各イベントを到着順に処理します。

1. event_start で、通知された id を記録します。識別子は常に一致します。event_start.event.id、すべての event_delta.event_id、およびバッファリングされた agent.message の id は同じ値です。
2. 各 event_delta で、delta.content.text を (event_id, delta.index) のエントリに追加し、累積テキストをレンダリングします。ある index に対する最初のデルタがそのエントリを作成します。
3. バッファリングされた agent.message が到着したら、id で照合し、蓄積したプレビューを破棄して、代わりにメッセージのコンテンツをレンダリングします。
4. span.model_request_end で、バッファリングされたイベントによって照合されていないプレビューをすべて閉じます。それらに対するデルタはこれ以上届きません。ターンがエラーになったり中断されたりした場合、バッファリングされたイベントは届かない可能性がありますが、span.model_request_end は届きます。

このパターンは2つの保証に依存しています。

• (event_id, index) をキーとして、プレビューのデルタを到着順に連結すると、バッファリングされたイベントの content[index].text のプレフィックスが得られます。デルタは負荷時に破棄される可能性があるため、必ずしもテキスト全体になるとは限りません。
• 1つの接続は event_id ごとに最大1つの event_start を発行し、バッファリングされたイベントは、その接続がその 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 は、スレッドストリームでもセッションレベルのストリームと同じ形式であり、蓄積と照合のパターンは記載どおりに適用されます。ストリーム接続ごとに1つのアキュムレーターインスタンスを実行してください。

# セッションのスレッドを一覧して子を選びます。子スレッドの 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 は引き続き完全な形で届きます。蓄積したプレビューを最終的なものとして扱わないでください。
• 再接続時のリプレイなし： デルタは、オプトインした接続が開いている間にのみ、その接続に配信されます。これはセッションレベルのストリームと各セッションスレッドのストリームの両方に当てはまります。モデルリクエストの開始後に開かれた接続は、その処理中のイベントに対するデルタを受信しません。見逃したデルタを再リクエストする方法はありません。
• 1つのスレッド、テキストのみ： プレビューは、接続が読み取っているスレッド上のアシスタントテキストを対象とします。ツール使用、ツール結果、MCP の結果がプレビューされることはありません。
• 永続化されない： event_start と event_delta はライブストリーム上にのみ存在します。セッションのイベント履歴（GET /v1/sessions/{session_id}/events）や、どのセッションスレッドのイベント履歴にも表示されません。

プレビューのトラブルシューティング

次のステップ