For the complete documentation index, see llms.txt. Markdown versions of documentation pages are available by appending .md to the page URL.
メインナビゲーション

おすすめ

Agents API

Agents SDK

ChatKit

API Dashboard
ChatGPT を試す

イベントとアイテム

リアルタイムの更新を受信し、保存された作業内容を取得します。

イベントは、エージェントの作業中に起きたことを通知します。アイテムは、後から取得できる保存済みのメッセージやツール呼び出しです。イベントを使ってアプリケーションをリアルタイムに更新し、アイテムを使って保存された履歴を表示します。

アプリケーションは入力イベントを送信して、メッセージの送信、ターンのキャンセル、ツールの結果の返却を行います。エージェントは、出力やセッションの変更を通知するイベントを送信します。入力の送信については、セッションの実行と継続を参照してください。

ストリームの受信

ターンの初期イベントをアプリケーションで受信できるように、作業を送信する前に購読を開始します。API クライアント、会話のセッション ID、イベントハンドラーを渡します。

セッションイベントのストリーミング
# Pass your saved session ID to this helper.
def stream_session(client: OpenAI, session_id: str, handle_event):
    with client.beta.agents.sessions.events.stream(session_id) as events:
        for event in events:
            handle_event(event)
            match event.type:
                case "agent.session.idle":
                    continue
                case "error":
                    raise RuntimeError(event.error.message)
                case "agent.session.failed" | "agent.session.environment.failed":
                    raise RuntimeError(f"Agent lifecycle failure: {event.type}")
                case "agent.session.turn.failed":
                    if event.turn.subagent_id is None:
                        detail = event.turn.error.message if event.turn.error else ""
                        raise RuntimeError(f"{event.type}: {detail}")
                case "agent.session.turn.cancelled":
                    if event.turn.subagent_id is None:
                        raise RuntimeError("The agent turn was cancelled")
                case "agent.session.turn.completed":
                    if event.turn.subagent_id is None:
                        return
    raise RuntimeError("Stream closed before a turn ended. Retrieve the saved state.")

ヘルパーは各イベントをハンドラーに渡した後、一般的なイベントタイプを確認します。agent.session.idle を受信しても処理を継続し、ルートターンが完了すると呼び出し元に戻ります。ルートターンが失敗またはキャンセルされた場合、セッションや環境に障害が発生した場合、または error イベントを受信した場合は、エラーを送出します。サブエージェントのターンイベントではストリームは終了しません。出力の表示方法はハンドラーが決定し、ヘルパーからのエラーは呼び出し元が処理します。ターンが終了する前にストリームが閉じた場合、ヘルパーはエラーを送出します。切断されたストリームの復旧を参照してください。

購読開始後のメッセージ送信

こちらのバージョンはメッセージを受け取り、ストリームを開いてから送信します。

メッセージの送信とストリーミング
# Pass your saved session ID and message to this helper.
def send_and_stream(client: OpenAI, session_id: str, text, handle_event):
    with client.beta.agents.sessions.events.stream(session_id) as events:
        client.beta.agents.sessions.events.create(
            session_id,
            events=[
                {
                    "type": "agent.session.input.message",
                    "input": [
                        {
                            "role": "user",
                            "content": [{"type": "input_text", "text": text}],
                        }
                    ],
                }
            ],
        )
        for event in events:
            handle_event(event)
            match event.type:
                case "agent.session.idle":
                    continue
                case "error":
                    raise RuntimeError(event.error.message)
                case "agent.session.failed" | "agent.session.environment.failed":
                    raise RuntimeError(f"Agent lifecycle failure: {event.type}")
                case "agent.session.turn.failed":
                    if event.turn.subagent_id is None:
                        detail = event.turn.error.message if event.turn.error else ""
                        raise RuntimeError(f"{event.type}: {detail}")
                case "agent.session.turn.cancelled":
                    if event.turn.subagent_id is None:
                        raise RuntimeError("The agent turn was cancelled")
                case "agent.session.turn.completed":
                    if event.turn.subagent_id is None:
                        return
    raise RuntimeError("Stream closed before a turn ended. Retrieve the saved state.")

更新の処理

イベントの type に基づいて、アプリケーションで行う処理を決定します。

  • テキストの表示： 該当するコンテンツパートに agent.session.turn.output_text.delta を追加します。agent.session.turn.output_text.done を受信したら、そのパートを完全なテキストで置き換えます。差分が届かない場合もあります。
  • 作業の追跡： セッション、ターン、アイテムのイベントは進捗を通知します。agent.session.turn.completed、agent.session.turn.failed、agent.session.turn.cancelled のいずれかを確認して、ターンの結果を判断します。
  • 必要な入力の提供： agent.session.requires_action を受信したら、セッションを取得して required_actions を確認します。コードで関数の結果を返したり、環境を接続したりする必要がある場合があります。

セッションがアイドル状態になったことやストリームが閉じたことだけでは、成功したと判断できません。また、ターンが完了しても、すべてのツールが成功したとは限りません。エージェントの出力を確認してください。

item_id、output_index、content_index を使って、テキストの更新を同じコンテンツパートに対応付けます。たとえば、以下の省略したイベント例は、1 つのパートを更新します。

{
  "type": "agent.session.turn.output_text.delta",
  "item_id": "msg_789",
  "output_index": 0,
  "content_index": 0,
  "delta": "Acme competes"
}
{
  "type": "agent.session.turn.output_text.done",
  "item_id": "msg_789",
  "output_index": 0,
  "content_index": 0,
  "text": "Acme competes on price and distribution."
}

各イベントには固有の event_id があります。共通の item_id は保存済みのアイテムを識別し、そのアイテムにはメッセージの内容、ステータス、フェーズが含まれます。保存された作業内容の取得を参照してください。

すべてのイベントタイプとフィールドについては、ストリーミングイベントのリファレンスを参照してください。これらのストリームイベントは Webhook とは異なります。サブエージェントのアクティビティや、コマンドの実行元の特定については、委任状況の確認を参照してください。

アイテムとターンの取得

アプリケーションの会話の状態に含まれるセッション ID を使って、保存された作業内容を取得します。

一覧取得エンドポイントは、1 回に 1 ページを返します。さらに結果を取得するには、SDK のページネーションヘルパーまたは after カーソルを使います。1 ページにターンのすべてのアイテムが含まれるとは限りません。アイテムを古い順に読み取るには、order: "asc" を使います。

切断されたストリームの復旧方法

ストリームは、受信できなかったイベントを再送しません。アプリケーションの表示を復元するには、次の手順を実行します。

  1. 新しいストリームを開き、受信するイベントをバッファーに保存します。
  2. ストリームの接続を維持したまま、セッションと保存済みのアイテムを取得します。
  3. アイテム ID をキーとして、それらのアイテムからローカルの状態を復元します。
  4. item_id を使って、バッファーに保存したアイテムの更新を適用します。取得した履歴ですでに最終状態に達しているアイテムについては、更新を破棄します。
  5. リアルタイムのイベント処理を再開します。

output_text.done イベントを使うと、一時的なテキストバッファーを完全なテキストで置き換えられます。保存済みのアイテムから完了した作業内容を復元できますが、受信できなかった途中のイベントをすべて復元できるわけではありません。