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 を使用します。
WebSocket モードでは、クライアントから response.create イベントを送信して各ターンを開始します。ペイロードは通常の Responses 作成リクエストの本文と同じですが、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();
}クライアントは必要に応じて、generate: false を指定した response.create を送信し、リクエストの状態を事前に準備できます。これは、今後のターンで送信するツール、指示、カスタムメッセージのいずれか、または複数がすでに決まっている場合に便利です。generate: false はモデルの出力を返さず、次の生成ターンをより速く開始できるようにリクエストの状態を準備します。この準備用リクエストはレスポンス ID を返します。previous_response_id にその 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 モードでは、通常と同じ方法で継続します。最新の previous_response_id と新しい入力項目だけを指定して、次の response.create を送信します。
単独での /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 パラメーターを使用すると、同じ接続で複数の会話を並列に進められます。異なる stream_id の値を指定して、独立した response.create イベントを続けて送信します。サーバーはこれらを 1 つの接続で同時に実行できます。それぞれのイベントは混在して届くことがあるため、読み取りループは 1 つにまとめ、stream_id に基づいて各イベントを振り分けます。
stream_id は、1 つの WebSocket 接続内で処理順序が保証されるレーンを識別します。stream_id と previous_response_id の役割を区別してください。
stream_idは、イベントの送信先と、どのリクエストを先入れ先出しの順序で実行するかを制御します。previous_response_idは、会話の継承関係を制御します。
このように役割を分けることで、2 つの便利なパターンを利用できます。
one WebSocket connection
├─ stream_id="planner" draft a deployment plan
└─ stream_id="research" list deployment risks同じ stream_id を持つリクエストは先入れ先出しの順序を維持し、実行が重なることはありません。stream_id の値が異なるリクエストは同時に実行できます。
接続ごとの制限
- 1 つの接続では、名前付きレーンとデフォルトレーンを合わせて、最大 16 件のレスポンスを同時に処理できます。それを超える
response.createイベントも受け付けますが、処理中のレスポンスが完了するまでキューに入ります。 - 1 つの接続では、最大 32 種類の名前付き
stream_idの値を使用できます。暗黙のデフォルトレーンは、この名前付きストリームの上限には含まれません。上限に達したら、既存のstream_idを再利用するか、新しい接続を開いてください。
新しいストリームへの会話のフォーク
完了済みのレスポンスから分岐するには、その ID を previous_response_id に指定し、新しい stream_id とともに送信します。そのレスポンスが利用可能な間は、新しいストリームがコンテキストを引き継ぎ、元のストリームも継続できます。フォークの開始後は、異なるストリーム 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_5previous_response_id を指定せずに stream_id を再利用すると、新しいレスポンスが開始されます。会話が継続されるわけではありません。
主な呼び出しは次のようになります。
# One socket, two independent conversations.
send_create(connection, "planner", "Draft a deployment plan.")
send_create(connection, "research", "List deployment risks.")
# Fork the planner response, then continue the original branch in parallel.
send_create(
connection,
"critic",
"Find gaps in this plan.",
previous_response_id=planner_response_id,
)
wait_for_in_progress(connection, "critic")
send_create(
connection,
"planner",
"Add rollback steps.",
previous_response_id=planner_response_id,
)完全な例
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 ストリーミングイベントモデルに従います。異なるレーンからのイベントは混在して届くことがあります。
- 同じ
stream_idを持つリクエストは先入れ先出しの順序で実行され、実行が重なることはありません。異なるレーンのリクエストは同時に実行できます。 - 接続は最大 60 分間維持されます。上限に達したら再接続してください。
再接続と復旧
接続が閉じると（または 60 分の上限に達すると）、すべてのレーンで、その接続内に保持されていたキャッシュが失われます。新しい WebSocket 接続を開き、次のいずれかの方法で各レーンを復旧します。
- 過去のレスポンスを保存済みで（
store=true）、有効なレスポンス ID がある場合は、previous_response_idと新しい入力項目を使ってそのレーンを継続します。 - レーンを継続できない場合（たとえば、
store=false/ZDR を使用している場合や、previous_response_not_foundが発生した場合）は、previous_response_idをnullに設定するか省略して新しいレスポンスを開始し、そのレーンの次のターンに必要な入力コンテキスト全体を送信します。 /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
}