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

おすすめ

はじめに

基本概念

SDK と CLI

クラウドプロバイダー

リソース

レガシー API

API ダッシュボード
ChatGPT を試す

WebSocket モード

1 本の WebSocket 接続を維持し、複数の会話の並列実行、レスポンスチェーンのフォーク、差分入力を利用して、エージェント型ワークフローのレイテンシを低減します。

Responses API は、ツール呼び出しが多く、長時間にわたるワークフロー向けに WebSocket モードを提供しています。レイテンシを低減できるほか、stream_id を使うと WebSocket 接続を多重化できます。/v1/responses への 1 本の接続を維持したまま、複数の会話を並列に実行したり、既存の会話を新しいストリームにフォークしたりできます。次のターンに進むには、新しい入力項目と previous_response_id だけを送信します。

WebSocket モードは、ゼロデータ保持（ZDR）と store=false の両方に対応しています。

WebSocket モードの利点

WebSocket モードは、モデルとツールの間で何度もやり取りするワークフローに特に適しています。たとえば、エージェント型コーディングや、ツール呼び出しを繰り返すオーケストレーションループが該当します。

WebSocket モードでは接続を維持し、各ターンで入力の差分だけを送信するため、次のターンに進む際のオーバーヘッドが減り、長いチェーン全体のレイテンシが低下します。ツールを 20 回以上呼び出す実行では、開始から完了までの処理が最大で約 40% 高速化した例があります。

接続とレスポンスの作成

WebSocket に必要なパッケージをインストールします。Python では pip install "openai[realtime]>=3.8.0"、JavaScript では npm install openai@^7.10.0 ws、Ruby では gem install openai async-websocket を実行してください。

Go では go get github.com/openai/openai-go/v3@v3.97.0 を実行します。 Java では Maven の依存関係に com.openai:openai-java:4.101.0 を追加し、以下の例を Java 17 以降で実行してください。 これらのバージョンの Go SDK と Java SDK は、Responses API の WebSocket 接続をネイティブにサポートしています。

WebSocket モードでは、クライアントから response.create イベントを送信して各ターンを開始します。ペイロードは通常のレスポンス作成リクエストの本文と同じ形式ですが、stream や background など、通信方式に固有のフィールドは使用しません。

import OpenAI from "openai";
import { ResponsesWS } from "openai/resources/responses/ws";

const client = new OpenAI();

const ws = new ResponsesWS(client);
try {
  ws.send({
    type: "response.create",
    stream_id: "main",
    model: "gpt-6-astra",
    store: false,
    input: [
      {
        type: "message",
        role: "user",
        content: [{ type: "input_text", text: "Find fizz_buzz()" }],
      },
    ],
    tools: [],
  });
  let completed = false;
  for await (const event of ws) {
    if (event.type === "error") throw event.error;
    if (event.type !== "message") continue;
    const message = event.message;
    if (message.type === "response.output_text.delta") {
      process.stdout.write(message.delta);
    } else if (message.type === "response.completed") {
      completed = true;
      break;
    } else if (
      message.type === "response.failed" ||
      message.type === "response.incomplete"
    ) {
      throw new Error(JSON.stringify(message));
    }
  }
  if (!completed)
    throw new Error("Connection closed before the response finished.");
} finally {
  ws.close();
}

クライアントは、response.create に generate: false を指定して送信することで、リクエストの状態を事前に準備できます。今後のターンで送信するツール、指示、カスタムメッセージの一部またはすべてがすでに決まっている場合に便利です。generate: false を指定すると、モデルの出力は返されず、次に生成を行うターンを早く開始できるようリクエストの状態が準備されます。このウォームアップリクエストはレスポンス ID を返します。レスポンスチェーンの後続ターンでも、この ID を previous_response_id に指定してチェーンを続けられます。次のセクションでは、previous_response_id と差分入力を使ってセッションを継続する方法を説明します。

差分入力による継続

レスポンスの生成中にユーザーの指示を追加するには、ターン途中のステアリングを使用します。ステアリングでは、完了した作業を保持したまま、新しい指示を反映して処理を続けます。通常どおり次のターンに進む場合や、ツールの実行結果を渡す場合は、以下のように response.create を使用します。

実行を継続するには、次の項目を指定して response.create を再度送信します。

  • previous_response_id に直前のレスポンス ID を設定します。
  • input には新しい項目（ツールの出力や次のユーザーメッセージなど）だけを含めます。
import OpenAI from "openai";
import { ResponsesWS } from "openai/resources/responses/ws";

const client = new OpenAI();
const model = "gpt-6-astra";

const tools = [
  {
    type: "function",
    name: "get_test_results",
    description: "Return a local demo test result.",
    parameters: { type: "object", properties: {}, additionalProperties: false },
    strict: true,
  },
];

async function waitForResponse(ws) {
  for await (const event of ws) {
    if (event.type === "error") throw event.error;
    if (event.type !== "message") continue;
    const message = event.message;
    if (message.type === "response.output_text.delta") {
      process.stdout.write(message.delta);
    } else if (message.type === "response.completed") {
      return message.response;
    } else if (
      message.type === "response.failed" ||
      message.type === "response.incomplete"
    ) {
      throw new Error(JSON.stringify(message));
    }
  }
  throw new Error("Connection closed before the response finished.");
}

const ws = new ResponsesWS(client);
try {
  ws.send({
    type: "response.create",
    stream_id: "main",
    model,
    store: false,
    input: "Find the failing test and suggest a fix.",
    tools,
    tool_choice: { type: "function", name: "get_test_results" },
    parallel_tool_calls: false,
  });
  const first = await waitForResponse(ws);
  const call = first.output.find((item) => item.type === "function_call");
  if (!call || call.name !== "get_test_results") {
    throw new Error("Expected a get_test_results function call.");
  }
  const result = {
    test: "test_fizz_buzz",
    failure: 'Expected "FizzBuzz" for 15, got "Fizz".',
  };

  // Continue on the same socket with the actual response and tool-call IDs.
  ws.send({
    type: "response.create",
    stream_id: "main",
    model,
    store: false,
    previous_response_id: first.id,
    input: [
      {
        type: "function_call_output",
        call_id: call.call_id,
        output: JSON.stringify(result),
      },
      { role: "user", content: "Now optimize it." },
    ],
    tools,
    tool_choice: "none",
  });
  await waitForResponse(ws);
} finally {
  ws.close();
}

継続の仕組み

WebSocket モードでは、previous_response_id を使ってレスポンスをつなぐ仕組みは HTTP モードと同じですが、接続中のソケットを使って、より低いレイテンシで処理を継続できます。

WebSocket 接続が有効な間、サービスは直近のレスポンスの状態を接続ごとのインメモリキャッシュに保持します。stream_id を使用すると、各レーンの最新のレスポンスがキャッシュに保持されます。そのレーンの最新のレスポンスから継続する場合、サービスは接続内の状態を再利用できるため、すばやく処理を開始できます。サービスは過去のレスポンスの状態をメモリにのみ保持し、ディスクには書き込まないため、store=false やゼロデータ保持（ZDR）と互換性のある形で WebSocket モードを利用できます。

previous_response_id に対応するレスポンスがインメモリキャッシュにない場合の動作は、レスポンスを保存するかどうかによって異なります。

  • store=true の場合、保存済みの状態があれば、サービスは古いレスポンス ID に対応する状態を復元することがあります。その場合も継続は可能ですが、インメモリキャッシュによるレイテンシ低減の効果は得られません。
  • store=false の場合（ZDR を含む）、保存済みの状態からは復元できません。指定した ID がキャッシュにない場合、リクエストは previous_response_not_found を返します。

同じレーンでの継続リクエストが 4xx または 5xx を返した場合、サービスは参照先の previous_response_id に対応する状態をその接続のキャッシュから削除します。別のレーンへのフォークでエラーが返った場合は、共有している親レスポンスが保持されるため、フォーク元のレーンは継続できます。

コンパクションと新しいレスポンスの作成

コンパクションを使用する場合、継続方法は次の 2 通りです。

サーバー側のコンパクション（context_management）

サーバー側のコンパクションを有効にすると（context_management に compact_threshold を指定）、通常の /responses での生成中にコンパクションが行われます。WebSocket モードでは通常どおり継続できます。次の response.create には、最新の previous_response_id と新しい入力項目だけを指定して送信します。

単独で使用する /responses/compact

単独で使用する /responses/compact エンドポイントは、レスポンス ID の代わりに、コンパクション後の新しい入力ウィンドウを返します。コンパクション後は、このウィンドウに次のユーザー入力やツール出力の項目を追加して input に指定し、WebSocket 接続で新しいレスポンスを作成します。

新しいチェーンを開始するには、previous_response_id を省略するか、null に設定します。コンパクションの出力はそのまま渡してください。返されたウィンドウから項目を削除しないでください。

import { toResponseInputItems } from "openai/lib/responses/ResponseInputItems";

// Compact your current window with an HTTP request.
const compacted = await client.responses.compact({
  model: "gpt-6-astra",
  input: longInputItems,
});
const nextInput = toResponseInputItems(compacted.output);
nextInput.push({
  type: "message",
  role: "user",
  content: [{ type: "input_text", text: "Continue from here." }],
});

// Start a new response on the WebSocket using the compacted window.
const ws = new ResponsesWS(client);
try {
  ws.send({
    type: "response.create",
    stream_id: "main",
    model: "gpt-6-astra",
    store: false,
    input: nextInput,
    tools: [],
  });
  let completed = false;
  for await (const event of ws) {
    if (event.type === "error") throw event.error;
    if (event.type !== "message") continue;
    const message = event.message;
    if (message.type === "response.output_text.delta") {
      process.stdout.write(message.delta);
    } else if (message.type === "response.completed") {
      completed = true;
      break;
    } else if (
      message.type === "response.failed" ||
      message.type === "response.incomplete"
    ) {
      throw new Error(JSON.stringify(message));
    }
  }
  if (!completed)
    throw new Error("Connection closed before the response finished.");
} finally {
  ws.close();
}

複数の会話の並列実行

stream_id パラメータを使用すると、同じ接続上で複数の会話を並列に進められます。独立した response.create イベントにそれぞれ異なる stream_id を指定して続けて送信してください。サーバーは 1 本の接続でこれらを同時に処理できます。異なる会話のイベントが入り交じって届くことがあるため、読み取りループは 1 つにし、各イベントを stream_id に応じて振り分けます。

stream_id は、1 本の WebSocket 接続内でリクエストを順番に処理するレーンを識別します。stream_id と previous_response_id の役割を区別してください。

  • stream_id は、イベントの振り分け先と、どのリクエストを先入れ先出しで処理するかを決めます。
  • previous_response_id は、会話をどのレスポンスから引き継ぐかを決めます。

このように役割が分かれているため、会話の並列実行とフォークが可能です。

1 本の WebSocket 接続
├─ stream_id="planner"   デプロイ計画を作成
└─ stream_id="research"  デプロイのリスクを列挙

同じ stream_id を持つリクエストは先入れ先出しで処理され、実行時間は重なりません。異なる stream_id を持つリクエストは同時に実行できます。

接続ごとの上限

  • 1 本の接続で同時に処理できるレスポンスは、名前付きレーンとデフォルトレーンを合わせて最大 16 件です。上限に達しても、接続は追加の response.create イベントを受け付け、処理中のレスポンスのいずれかが完了するまでキューに入れます。
  • 1 本の接続で使用できる名前付きの stream_id は最大 32 種類です。暗黙のデフォルトレーンは、この名前付きストリーム数の上限に含まれません。上限に達したら、既存の stream_id を再利用するか、新しい接続を開いてください。

新しいストリームへの会話のフォーク

完了したレスポンスから分岐するには、その ID を previous_response_id に指定し、新しい stream_id とともに送信します。そのレスポンスが引き続き利用できれば、新しいストリームはそのコンテキストを引き継ぎ、元のストリームでも会話を続けられます。フォークの開始後は、2 つのブランチで異なるストリーム ID を使用するため、両方を同時に実行できます。

store=false の場合（ZDR を含む）、別のレーンにフォークするには、親レスポンスがその接続のキャッシュに残っている必要があります。フォークがキューで待機している間にフォーク元のレーンが先に進んだり失敗したりすると、フォークの開始前に親レスポンスがキャッシュから削除されることがあります。その場合、フォークは previous_response_not_found を返します。フォーク元のレーンを先に進める前に、フォーク先のレーンから response.in_progress イベントが届くのを待ってください。あるいは、previous_response_id を null に設定し、入力コンテキスト全体を再送信してリトライしてください。

main:   resp_1 ──▶ resp_2 ──▶ resp_3
                       ╲
critic:                 resp_4 ──▶ resp_5

stream_id を再利用しても、previous_response_id を指定しなければ、新しいレスポンスが開始されます。それまでの会話は引き継がれません。

主な呼び出しの流れは次のとおりです。

# 1 本のソケットで 2 つの独立した会話を開始します。
send_create(connection, "planner", "デプロイ計画を作成してください。")
send_create(connection, "research", "デプロイのリスクを列挙してください。")

# planner のレスポンスをフォークし、元のブランチでも並行して会話を続けます。
send_create(
    connection,
    "critic",
    "この計画に抜けている点があれば指摘してください。",
    previous_response_id=planner_response_id,
)
wait_for_in_progress(connection, "critic")
send_create(
    connection,
    "planner",
    "ロールバックの手順を追加してください。",
    previous_response_id=planner_response_id,
)

実装例の全体像

複数の会話を並列に実行し、そのうちの 1 つをフォーク
import OpenAI from "openai";
import { ResponsesWS } from "openai/resources/responses/ws";

const client = new OpenAI();

const latestResponseIdByLane = new Map();

function sendCreate(
  ws,
  streamId,
  text,
  previousResponseId = latestResponseIdByLane.get(streamId)
) {
  ws.send({
    type: "response.create",
    stream_id: streamId,
    model: "gpt-6-astra",
    store: false,
    input: [
      {
        type: "message",
        role: "user",
        content: [{ type: "input_text", text }],
      },
    ],
    previous_response_id: previousResponseId,
  });
}

async function readMessage(events) {
  while (true) {
    const { value: event, done } = await events.next();
    if (done)
      throw new Error("Connection closed before all responses finished.");
    if (event.type === "error") throw event.error;
    if (event.type !== "message") continue;
    const message = event.message;
    if (
      message.type === "response.failed" ||
      message.type === "response.incomplete"
    ) {
      throw new Error(
        `Lane ${message.stream_id} failed: ${JSON.stringify(message)}`
      );
    }
    return message;
  }
}

async function drainUntilComplete(events, expectedStreamIds) {
  const remaining = new Set(expectedStreamIds);
  while (remaining.size > 0) {
    const message = await readMessage(events);
    const streamId = message.stream_id;
    if (!streamId || !remaining.has(streamId)) continue;
    if (message.type === "response.completed") {
      latestResponseIdByLane.set(streamId, message.response.id);
      remaining.delete(streamId);
    }
  }
}

async function waitForInProgress(events, streamId) {
  while (true) {
    const message = await readMessage(events);
    if (
      message.type === "response.in_progress" &&
      message.stream_id === streamId
    )
      return;
  }
}

const ws = new ResponsesWS(client);
// Keep one iterator so events stay queued while moving between phases.
const events = ws.stream();
try {
  // Run two independent conversations in parallel.
  sendCreate(
    ws,
    "planner",
    "Draft a deployment plan for a stateless API service."
  );
  sendCreate(
    ws,
    "research",
    "List common deployment risks for a stateless API service."
  );
  await drainUntilComplete(events, new Set(["planner", "research"]));

  // Fork the planner conversation and continue its original branch in parallel.
  const plannerResponseId = latestResponseIdByLane.get("planner");
  sendCreate(
    ws,
    "critic",
    "Find gaps in this deployment plan.",
    plannerResponseId
  );
  // Let the fork load its parent before advancing the original lane's cache.
  await waitForInProgress(events, "critic");
  sendCreate(
    ws,
    "planner",
    "Add rollback and monitoring steps to the plan.",
    plannerResponseId
  );
  await drainUntilComplete(events, new Set(["critic", "planner"]));
} finally {
  await events.return?.();
  ws.close();
}

stream_id の長さは 1～256 文字で、使用できるのは英字、数字、アンダースコア（_）、ハイフン（-）、ピリオド（.）のみです。WebSocket の response.create イベントでのみ使用し、HTTP の POST /v1/responses には含めないでください。

名前付きストリームでは、終了イベントや個々のリクエストに対するエラーも含め、サーバーイベントに対応する stream_id が含まれます。

stream_id を省略すると、リクエストは暗黙のデフォルトレーンを使用し、そのイベントには stream_id が含まれません。それ以外の点では、デフォルトレーンには名前付きストリームと同じ実行順序と同時実行のルールが適用されます。空文字列は有効な stream_id ではありません。デフォルトレーンを選択するには、このフィールドを省略してください。

接続の動作と制限

  • 各レスポンス内のイベントは、既存の Responses API のストリーミングイベントモデルに従います。異なるレーンのイベントは入り交じって届くことがあります。
  • 同じ stream_id を持つリクエストは先入れ先出しで処理され、実行時間は重なりません。異なるレーンのリクエストは同時に実行できます。
  • 接続を維持できるのは最大 60 分です。上限に達したら再接続してください。

再接続と復旧

接続が閉じると（60 分の上限に達した場合を含む）、その接続のすべてのレーンのキャッシュが失われます。新しい WebSocket 接続を開き、次のいずれかの方法で各レーンを復旧してください。

  1. 以前のレスポンスを保存しており（store=true）、有効なレスポンス ID がある場合は、previous_response_id と新しい入力項目を指定して、そのレーンを継続します。
  2. レーンを継続できない場合（store=false または ZDR を使用している場合や、previous_response_not_found が返った場合など）は、previous_response_id を null に設定するか省略して新しいレスポンスを開始し、そのレーンの次のターンに必要な入力コンテキスト全体を送信します。
  3. /responses/compact でコンテキストのコンパクションを行った場合は、返されたウィンドウを新しいレスポンスの input のベースにし、最新のユーザー入力やツール出力の項目を追加します。

対処が必要なエラー

サーバーがエラーを名前付きレーンに関連付けられる場合、エラーイベントには stream_id が含まれます。個々のリクエストでエラーが発生しても、他のレーンは処理を継続できます。

previous_response_not_found

{
  "type": "error",
  "status": 400,
  "stream_id": "main",
  "error": {
    "type": "invalid_request_error",
    "code": "previous_response_not_found",
    "message": "Previous response with id 'resp_abc' not found.",
    "param": "previous_response_id"
  }
}

invalid_stream_id

{
  "type": "error",
  "status": 400,
  "error": {
    "type": "invalid_request_error",
    "code": "invalid_stream_id",
    "message": "The 'stream_id' field must be a non-empty string with at most 256 characters and may only contain letters, numbers, underscores, hyphens, and periods.",
    "param": "stream_id"
  }
}

websocket_stream_limit_reached

{
  "type": "error",
  "status": 400,
  "stream_id": "agent_33",
  "error": {
    "type": "invalid_request_error",
    "code": "websocket_stream_limit_reached",
    "message": "This WebSocket connection has reached its maximum number of distinct stream IDs (32). Reuse an existing stream_id or open a new WebSocket connection.",
    "param": "stream_id"
  }
}

websocket_connection_limit_reached

{
  "type": "error",
  "error": {
    "type": "invalid_request_error",
    "code": "websocket_connection_limit_reached",
    "message": "Responses websocket connection limit reached (60 minutes). Create a new websocket connection to continue."
  },
  "status": 400
}

関連ガイド