WebSocket モード | OpenAI API

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 などのトランスポート固有のフィールドは使用しません。

JavaScript

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43import 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();
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24from openai import OpenAI

client = OpenAI()

with client.responses.connect() as connection:
    connection.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=[],
    )
    for event in connection:
        if event.type == "response.completed":
            print(event.response.output_text)
            break
        if event.type in {"response.failed", "response.incomplete", "error"}:
            raise RuntimeError(event.to_json())

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31require "async"
require "openai"

def wait_for_response(connection)
  while (event = connection.receive)
    case event.type.to_s
    when "response.completed" then return event.response
    when "response.failed", "response.incomplete", "error"
      raise "Response failed: #{event.to_json}"
    end
  end
  raise "Connection closed before the response finished"
end

client = OpenAI::Client.new
Sync do |task|
  task.with_timeout(120) do
    client.responses.connect(request_options: { timeout: 10 }) do |connection|
      connection.response.create(
        stream_id: "main", model: "gpt-6-astra", store: false,
        input: [
          {
            role: "user",
            content: "Find fizz_buzz()"
          }
        ], tools: []
      )
      puts(wait_for_response(connection).output_text)
    end
  end
end

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

差分入力による継続

レスポンスの処理中にユーザーの指示を追加するには、ターン途中の指示変更を使用します。この機能は完了済みの作業を保持し、新しい指示を継続処理に含めます。通常のターン間の継続やツール結果の送信には、以下の response.create パターンを使用します。

実行を継続するには、次の内容を指定して、もう一度 response.create を送信します。

previous_response_id に前のレスポンス ID を設定
input には新しい項目（ツールの出力や次のユーザーメッセージなど）のみを指定

JavaScript

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79import 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();
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74import json

from openai import OpenAI
from openai.resources.responses.responses import ResponsesConnection
from openai.types.responses import FunctionToolParam, Response

client = OpenAI()
model = "gpt-6-astra"
tools: list[FunctionToolParam] = [
    {
        "type": "function",
        "name": "get_test_results",
        "description": "Read the demo test results.",
        "parameters": {
            "type": "object",
            "properties": {},
            "required": [],
            "additionalProperties": False,
        },
        "strict": True,
    }
]


def get_test_results():
    # Demo data. Replace this function with your test runner.
    return {
        "test": "test_fizz_buzz",
        "failure": 'Expected "FizzBuzz" for 15, got "Fizz".',
    }


def wait_for_response(connection: ResponsesConnection) -> Response:
    for event in connection:
        if event.type == "response.completed":
            return event.response
        if event.type in {"response.failed", "response.incomplete", "error"}:
            raise RuntimeError(event.to_json())
    raise RuntimeError("Connection closed before the response finished.")


with client.responses.connect() as connection:
    connection.response.create(
        stream_id="main",
        model=model,
        store=False,
        input="Find the failing test and suggest a fix.",
        tools=tools,
        tool_choice={"type": "function", "name": "get_test_results"},
        parallel_tool_calls=False,
    )
    response = wait_for_response(connection)
    call = next(item for item in response.output if item.type == "function_call")
    if call.name != "get_test_results" or json.loads(call.arguments) != {}:
        raise ValueError("Expected a get_test_results call with no arguments")

    # Continue on the same connection using the actual response and tool-call IDs.
    connection.response.create(
        stream_id="main",
        model=model,
        store=False,
        previous_response_id=response.id,
        input=[
            {
                "type": "function_call_output",
                "call_id": call.call_id,
                "output": json.dumps(get_test_results()),
            },
            {"role": "user", "content": "Now optimize it."},
        ],
        tools=tools,
        tool_choice="none",
    )
    print(wait_for_response(connection).output_text)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73require "async"
require "openai"
require "json"

def wait_for_response(connection)
  while (event = connection.receive)
    case event.type.to_s
    when "response.completed" then return event.response
    when "response.failed", "response.incomplete", "error"
      raise "Response failed: #{event.to_json}"
    end
  end
  raise "Connection closed before the response finished"
end

tools = [
  {
    type: "function",
    name: "get_test_results",
    description: "Read the demo test results.",
    parameters: {
      type: "object",
      properties: {},
      required: [],
      additionalProperties: false
    },
    strict: true
  }
]

client = OpenAI::Client.new
Sync do |task|
  task.with_timeout(120) do
    client.responses.connect(request_options: { timeout: 10 }) do |connection|
      connection.response.create(
        stream_id: "main", model: "gpt-6-astra", store: false,
        input: "Find the failing test and suggest a fix.", tools: tools,
        tool_choice: {
          type: "function",
          name: "get_test_results"
        }, parallel_tool_calls: false
      )
      response = wait_for_response(connection)
      call = response.output.grep(OpenAI::Responses::ResponseFunctionToolCall).first
      unless call && call.name == "get_test_results" && JSON.parse(call.arguments) == {}
        raise "Expected a get_test_results call with no arguments"
      end

      # Demo data. Replace this with your test runner.
      result = {
        test: "test_fizz_buzz",
        failure: 'Expected "FizzBuzz" for 15, got "Fizz".'
      }
      connection.response.create(
        stream_id: "main", model: "gpt-6-astra", store: false,
        previous_response_id: response.id,
        input: [
          {
            type: "function_call_output",
            call_id: call.call_id,
            output: JSON.generate(result)
          },
          {
            role: "user",
            content: "Now optimize it."
          }
        ],
        tools: tools, tool_choice: "none"
      )
      puts(wait_for_response(connection).output_text)
    end
  end
end

継続の仕組み

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 に設定して、新しいチェーンを開始します。コンパクション済みの出力はそのまま渡し、返されたウィンドウの内容を削らないでください。

JavaScript

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47import { 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();
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37from typing import cast

from openai import OpenAI
from openai.types.responses import ResponseInputParam

# Compact your current window (HTTP call).
compacted = client.responses.compact(
    model="gpt-6-astra",
    input=long_input_items_array,
)
next_input = cast(
    ResponseInputParam,
    [item.to_dict() for item in compacted.output],
)
next_input.append(
    {
        "type": "message",
        "role": "user",
        "content": [{"type": "input_text", "text": "Continue from here."}],
    }
)

# Start a new response on the WebSocket using the compacted window.
with client.responses.connect() as connection:
    connection.response.create(
        stream_id="main",
        model="gpt-6-astra",
        store=False,
        input=next_input,
        tools=[],
    )
    for event in connection:
        if event.type == "response.completed":
            print(event.response.output_text)
            break
        if event.type in {"response.failed", "response.incomplete", "error"}:
            raise RuntimeError(event.to_json())

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41require "async"
require "openai"

def wait_for_response(connection)
  while (event = connection.receive)
    case event.type.to_s
    when "response.completed" then return event.response
    when "response.failed", "response.incomplete", "error"
      raise "Response failed: #{event.to_json}"
    end
  end
  raise "Connection closed before the response finished"
end

client = OpenAI::Client.new
compacted = client.responses.compact(
  model: "gpt-6-astra",
  input: [
    {
      role: :user,
      content: "Find the failing test."
    }
  ]
)
next_input = compacted.output.map(&:to_h)
next_input << {
  role: :user,
  content: "Continue from here."
}

Sync do |task|
  task.with_timeout(120) do
    client.responses.connect(request_options: { timeout: 10 }) do |connection|
      connection.response.create(
        stream_id: "main", model: "gpt-6-astra", store: false,
        input: next_input, tools: []
      )
      puts(wait_for_response(connection).output_text)
    end
  end
end

会話の並列実行

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_5

previous_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,
)

完全な例

会話を並列に実行し、そのうち 1 つをフォーク

JavaScript

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111import 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();
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93from openai import OpenAI
from openai.resources.responses.responses import ResponsesConnection

client = OpenAI()
latest_response_id_by_lane: dict[str, str] = {}


def send_create(
    connection: ResponsesConnection,
    stream_id: str,
    text: str,
    previous_response_id: str | None = None,
):
    if previous_response_id is None:
        previous_response_id = latest_response_id_by_lane.get(stream_id)
    connection.response.create(
        stream_id=stream_id,
        model="gpt-6-astra",
        store=False,
        input=[
            {
                "type": "message",
                "role": "user",
                "content": [{"type": "input_text", "text": text}],
            }
        ],
        previous_response_id=previous_response_id,
    )


def drain_until_complete(
    connection: ResponsesConnection, expected_stream_ids: set[str]
):
    completed: set[str] = set()
    for event in connection:
        stream_id = event.stream_id
        if event.type == "error" and stream_id is None:
            raise RuntimeError(f"Connection error: {event.to_json()}")
        if stream_id is None or stream_id not in expected_stream_ids:
            continue

        if event.type == "response.completed":
            latest_response_id_by_lane[stream_id] = event.response.id
            completed.add(stream_id)
            if completed == expected_stream_ids:
                return
        elif event.type in {"response.failed", "response.incomplete", "error"}:
            raise RuntimeError(f"Lane {stream_id} failed: {event.to_json()}")
    raise RuntimeError("Connection closed before all responses finished.")


def wait_for_in_progress(connection: ResponsesConnection, expected_stream_id: str):
    for event in connection:
        if event.type == "error" and event.stream_id is None:
            raise RuntimeError(f"Connection error: {event.to_json()}")
        if event.stream_id != expected_stream_id:
            continue
        if event.type == "response.in_progress":
            return
        if event.type in {"response.failed", "response.incomplete", "error"}:
            raise RuntimeError(f"Lane {expected_stream_id} failed: {event.to_json()}")
    raise RuntimeError("Connection closed before the fork started.")


with client.responses.connect() as connection:
    # 1. Run two independent conversations in parallel.
    send_create(
        connection, "planner", "Draft a deployment plan for a stateless API service."
    )
    send_create(
        connection,
        "research",
        "List common deployment risks for a stateless API service.",
    )
    drain_until_complete(connection, {"planner", "research"})

    # 2. Fork the planner conversation and continue the original branch in parallel.
    planner_response_id = latest_response_id_by_lane["planner"]
    send_create(
        connection,
        "critic",
        "Find gaps in this deployment plan.",
        previous_response_id=planner_response_id,
    )
    # Let the fork bind its parent before advancing the original lane.
    wait_for_in_progress(connection, "critic")
    send_create(
        connection,
        "planner",
        "Add rollback and monitoring steps to the plan.",
        previous_response_id=planner_response_id,
    )
    drain_until_complete(connection, {"critic", "planner"})

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64require "async"
require "openai"
require "json"

def send_create(connection, stream_id, text, previous_response_id = nil)
  payload = {
    stream_id: stream_id,
    model: "gpt-6-astra",
    store: false,
    input: [
      {
        role: "user",
        content: text
      }
    ]
  }
  payload[:previous_response_id] = previous_response_id if previous_response_id
  connection.response.create(**payload)
end

def read_event(connection)
  event = connection.receive or raise "Connection closed before all responses finished"
  if ["response.failed", "response.incomplete", "error"].include?(event.type.to_s)
    raise "Response failed: #{event.to_json}"
  end

  event
end

def drain_responses(connection, lanes, latest_ids)
  remaining = lanes.dup
  until remaining.empty?
    event = read_event(connection)
    next unless event.type.to_s == "response.completed"

    lane = event.stream_id
    next unless remaining.include?(lane)

    latest_ids[lane] = event.response.id
    remaining.delete(lane)
  end
end

client = OpenAI::Client.new
Sync do |task|
  task.with_timeout(120) do
    client.responses.connect(request_options: { timeout: 10 }) do |connection|
      latest_ids = {}
      send_create(connection, "planner", "Draft a deployment plan for a stateless API service.")
      send_create(connection, "research", "List common deployment risks for a stateless API service.")
      drain_responses(connection, ["planner", "research"], latest_ids)
      parent_id = latest_ids.fetch("planner")
      send_create(connection, "critic", "Find gaps in this deployment plan.", parent_id)
      # Let the fork load its parent before advancing the original lane's cache.
      loop do
        event = read_event(connection)
        break if event.type.to_s == "response.in_progress" && event.stream_id == "critic"
      end
      send_create(connection, "planner", "Add rollback and monitoring steps.", parent_id)
      drain_responses(connection, ["critic", "planner"], latest_ids)
      puts(JSON.generate(latest_ids))
    end
  end
end

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

1234567891011{
  "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

12345678910{
  "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

1234567891011{
  "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

123456789{
  "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
}

おすすめ

はじめに

基本コンセプト

SDK と CLI

リソース

レガシー API

モデルの選び方

テキストとコード

プロンプト

推論

画像と動画

リアルタイムと音声

特化型モデル

Agents API

Agents SDK

ChatKit

検索と取得

ツールとデータの接続

ツールを使ったワークフローの構築

コンピューターとコード

メディア

GPT-Live

Realtime API

音声を活用した開発

接続

音声処理

本番運用の開始

性能と品質

コストとスループット

安全性とガバナンス

インフラとアクセス

オペレーション

基本コンセプト

Plan

Build

Test and publish

Conversion specs

ガイド

リソース

はじめに

ガイド

File Upload

API

Measurement

Advertiser API

API リファレンス

はじめに

基礎

詳しく見る

利用可能な環境

リリース

ワークフロー

できること

リファレンス

カスタマイズ

設定ファイル

エージェントの構成

ChatGPT と Codex の機能拡張

Linux

Windows

開発ワークフロー

拡張と自動化

環境

Codex を使った開発

サードパーティ連携

リファレンス

権限

Codex Security

サイバーセーフティ

はじめに

ChatGPT Work

ID と認証

ワークスペースのアクセス、ポリシー、モデル

プラグインとコネクタの制御

利用状況、ガバナンス、コンプライアンス

展開とモデルプロバイダー

コミュニティ

ブログ

コミュニティ

ブログ

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

単独での `/responses/compact` の使用