WebSocket 모드 | OpenAI API

Responses API는 도구를 자주 호출하며 장시간 실행되는 워크플로우를 위해 WebSocket 모드를 지원합니다. WebSocket 모드는 지연 시간을 줄이며, stream_id를 사용하면 하나의 WebSocket 연결에서 여러 스트림을 운용할 수 있습니다. /v1/responses 연결 하나를 유지하면서 여러 대화를 병렬로 실행하거나 기존 대화를 새 스트림으로 포크할 수 있습니다. 다음 턴을 이어갈 때는 새 입력 항목과 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.98.0을 실행하세요. Java에서는 Maven 종속성 com.openai:openai-java:4.101.0을 추가하고 Java 17 이상에서 예제를 실행하세요. 위 버전의 Go 및 Java SDK는 Responses 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
24
25
26from 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())
    else:
        raise RuntimeError("Connection closed before the response finished.")

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
27ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
defer cancel()
client := openai.NewClient()
conn, err := client.Responses.Connect(ctx, responses.ResponseConnectionOptions{})
if err != nil {
	log.Fatal(err)
}
defer conn.Close()
if err := conn.Create(ctx, responses.ResponsesClientEventResponseCreateParam{
	Model: "gpt-6-astra",
	Store: openai.Bool(false),
	Input: responses.ResponsesClientEventResponseCreateInputUnionParam{
		OfString: openai.String("Find fizz_buzz()"),
	},
	StreamID: openai.String("main"),
	Tools:    []responses.ToolUnionParam{},
}); err != nil {
	log.Fatal(err)
}
response, err := conn.FinalResponse(ctx)
if err != nil {
	log.Fatal(err)
}
if response.Status != responses.ResponseStatusCompleted {
	log.Fatalf("Response ended with status %s", response.Status)
}
fmt.Println(response.OutputText())

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
28import com.openai.client.okhttp.OpenAIOkHttpClient;
import com.openai.models.responses.*;
import java.util.*;

var client = OpenAIOkHttpClient.fromEnv();

try (var conn = client.responses().connect()) {
  conn.send(
      ResponsesClientEvent.ofResponseCreate(
          ResponsesClientEvent.ResponseCreate.builder()
              .model("gpt-6-astra")
              .store(false)
              .input("Find fizz_buzz()")
              .streamId("main")
              .tools(List.of())
              .build()));
  var response = conn.finalResponse();
  if (response.status().filter(ResponseStatus.COMPLETED::equals).isEmpty())
    throw new IllegalStateException(
        "Response ended with status " + response.status().orElse(null));
  response.output().stream()
      .flatMap(item -> item.message().stream())
      .flatMap(message -> message.content().stream())
      .flatMap(content -> content.outputText().stream())
      .forEach(text -> System.out.println(text.text()));
} finally {
  client.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
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

클라이언트는 필요에 따라 response.create 이벤트에 generate: false를 지정해 요청 상태를 미리 준비할 수 있습니다. 이후 턴에 보낼 도구, 지침, 사용자 지정 메시지 중 일부 또는 전부를 이미 알고 있을 때 유용합니다. 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
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
defer cancel()
client := openai.NewClient()
tools := []responses.ToolUnionParam{
	{
		OfFunction: &responses.FunctionToolParam{
			Name:        "get_test_results",
			Description: openai.String("Return a local demo test result."),
			Parameters: map[string]any{
				"type":                 "object",
				"properties":           map[string]any{},
				"additionalProperties": false,
			},
			Strict: openai.Bool(true),
		},
	},
}
conn, err := client.Responses.Connect(ctx, responses.ResponseConnectionOptions{})
if err != nil {
	log.Fatal(err)
}
defer conn.Close()
if err := conn.Create(ctx, responses.ResponsesClientEventResponseCreateParam{
	Model: "gpt-6-astra",
	Store: openai.Bool(false),
	Input: responses.ResponsesClientEventResponseCreateInputUnionParam{
		OfString: openai.String("Find the failing test and suggest a fix."),
	},
	StreamID:          openai.String("main"),
	Tools:             tools,
	ParallelToolCalls: openai.Bool(false),
	ToolChoice: responses.ResponsesClientEventResponseCreateToolChoiceUnionParam{
		OfFunctionTool: &responses.ToolChoiceFunctionParam{
			Name: "get_test_results",
		},
	},
}); err != nil {
	log.Fatal(err)
}
first, err := conn.FinalResponse(ctx)
if err != nil {
	log.Fatal(err)
}
if first.Status != responses.ResponseStatusCompleted {
	log.Fatalf("Response ended with status %s", first.Status)
}
var callID string
for _, item := range first.Output {
	if item.Type == "function_call" && item.Name == "get_test_results" {
		callID = item.CallID
		break
	}
}
if callID == "" {
	log.Fatal("Expected a get_test_results function call")
}
// The result is a local demo fixture; carry the real response and call IDs.
result := `{"test":"test_fizz_buzz","failure":"Expected FizzBuzz for 15, got Fizz."}`
if err := conn.Create(ctx, responses.ResponsesClientEventResponseCreateParam{
	Model:              "gpt-6-astra",
	StreamID:           openai.String("main"),
	Store:              openai.Bool(false),
	PreviousResponseID: openai.String(first.ID),
	Tools:              tools,
	ToolChoice: responses.ResponsesClientEventResponseCreateToolChoiceUnionParam{
		OfToolChoiceMode: openai.Opt(responses.ToolChoiceOptionsNone),
	},
	Input: responses.ResponsesClientEventResponseCreateInputUnionParam{
		OfResponse: &responses.ResponseInputParam{
			{
				OfFunctionCallOutput: &responses.ResponseInputItemFunctionCallOutputParam{
					CallID: openai.String(callID),
					Output: responses.ResponseInputItemFunctionCallOutputOutputUnionParam{
						OfString: openai.String(result),
					},
				},
			},
			responses.ResponseInputItemParamOfMessage("Now optimize it.", responses.EasyInputMessageRoleUser),
		},
	},
}); err != nil {
	log.Fatal(err)
}
response, err := conn.FinalResponse(ctx)
if err != nil {
	log.Fatal(err)
}
if response.Status != responses.ResponseStatusCompleted {
	log.Fatalf("Response ended with status %s", response.Status)
}
fmt.Println(response.OutputText())

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
78import com.openai.client.okhttp.OpenAIOkHttpClient;
import com.openai.core.JsonValue;
import com.openai.models.responses.*;
import java.util.*;

var client = OpenAIOkHttpClient.fromEnv();

var tool =
    FunctionTool.builder()
        .name("get_test_results")
        .description("Return a local demo test result.")
        .strict(true)
        .parameters(
            FunctionTool.Parameters.builder()
                .putAdditionalProperty("type", JsonValue.from("object"))
                .putAdditionalProperty("properties", JsonValue.from(Map.of()))
                .putAdditionalProperty("additionalProperties", JsonValue.from(false))
                .build())
        .build();
try (var conn = client.responses().connect()) {
  conn.send(
      ResponsesClientEvent.ofResponseCreate(
          ResponsesClientEvent.ResponseCreate.builder()
              .model("gpt-6-astra")
              .store(false)
              .input("Find the failing test and suggest a fix.")
              .streamId("main")
              .addTool(tool)
              .parallelToolCalls(false)
              .toolChoice(ToolChoiceFunction.builder().name("get_test_results").build())
              .build()));
  var first = conn.finalResponse();
  if (first.status().filter(ResponseStatus.COMPLETED::equals).isEmpty())
    throw new IllegalStateException(
        "Response ended with status " + first.status().orElse(null));
  var call =
      first.output().stream()
          .flatMap(item -> item.functionCall().stream())
          .filter(item -> item.name().equals("get_test_results"))
          .findFirst()
          .orElseThrow(() -> new IllegalStateException("Expected get_test_results call"));
  // Use the real response and call IDs with a local demo result.
  var result =
      "{\"test\":\"test_fizz_buzz\",\"failure\":\"Expected FizzBuzz for 15, got Fizz.\"}";
  conn.send(
      ResponsesClientEvent.ofResponseCreate(
          ResponsesClientEvent.ResponseCreate.builder()
              .model("gpt-6-astra")
              .streamId("main")
              .store(false)
              .previousResponseId(first.id())
              .addTool(tool)
              .toolChoice(ToolChoiceOptions.NONE)
              .inputOfResponse(
                  List.of(
                      ResponseInputItem.ofFunctionCallOutput(
                          ResponseInputItem.FunctionCallOutput.builder()
                              .callId(call.callId())
                              .output(result)
                              .build()),
                      ResponseInputItem.ofEasyInputMessage(
                          EasyInputMessage.builder()
                              .role(EasyInputMessage.Role.USER)
                              .content("Now optimize it.")
                              .build())))
              .build()));
  var response = conn.finalResponse();
  if (response.status().filter(ResponseStatus.COMPLETED::equals).isEmpty())
    throw new IllegalStateException(
        "Response ended with status " + response.status().orElse(null));
  response.output().stream()
      .flatMap(item -> item.message().stream())
      .flatMap(message -> message.content().stream())
      .flatMap(content -> content.outputText().stream())
      .forEach(text -> System.out.println(text.text()));
} finally {
  client.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
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 연결에서는 지연 시간이 더 짧은 경로로 응답을 이어갈 수 있습니다.

서비스는 활성 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에 해당하는 항목을 연결별 캐시에서 제거합니다. 다른 레인으로 포크하는 요청이 오류를 반환하면 공유하는 부모 응답을 보존하므로 원래 레인에서 실행을 이어갈 수 있습니다.

컨텍스트 압축과 새 응답 생성

컨텍스트 압축을 사용할 때는 압축 방식에 따라 두 가지 방법으로 실행을 이어갈 수 있습니다.

서버 측 컨텍스트 압축(`context_management`)

서버 측 컨텍스트 압축을 활성화하면(context_management에 compact_threshold 지정) 일반적인 /responses 응답 생성 과정에서 컨텍스트를 압축합니다. WebSocket 모드에서는 평소와 같은 방식으로 이어가세요. 다음 response.create 이벤트에 최신 previous_response_id와 새 입력 항목만 담아 보내면 됩니다.

독립형 `/responses/compact`

독립형 /responses/compact 엔드포인트는 응답 ID 대신 새로 압축한 입력 윈도우를 반환합니다. 압축 후에는 WebSocket 연결에서 압축된 윈도우를 input으로 사용하고 다음 사용자 또는 도구 항목을 덧붙여 새 응답을 생성하세요.

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
37
38
39from 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())
    else:
        raise RuntimeError("Connection closed before the response finished.")

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 매개변수를 사용하면 같은 연결에서 여러 대화를 병렬로 유지할 수 있습니다. 독립적인 response.create 이벤트에 서로 다른 stream_id 값을 지정해 연달아 보내세요. 서버는 하나의 연결에서 이 요청들을 동시에 실행할 수 있습니다. 서로 다른 스트림의 이벤트가 섞여 도착할 수 있으므로 읽기 루프는 하나만 두고 stream_id에 따라 각 이벤트를 분배하세요.

stream_id는 하나의 WebSocket 연결에서 요청을 순서대로 처리하는 레인의 이름입니다. stream_id와 previous_response_id는 역할이 다릅니다.

stream_id는 이벤트를 전달할 레인과 선입선출 순서로 실행할 요청을 결정합니다.
previous_response_id는 대화가 어느 응답에서 이어지는지 결정합니다.

두 역할이 분리되어 있어 다음 두 가지 패턴을 사용할 수 있습니다.

WebSocket 연결 하나
├─ stream_id="planner"   배포 계획 초안 작성
└─ stream_id="research"  배포 위험 요인 나열

stream_id가 같은 요청은 들어온 순서대로 하나씩 실행됩니다. stream_id 값이 다른 요청은 동시에 실행될 수 있습니다.

연결별 한도

하나의 연결에서 이름이 지정된 레인과 기본 레인을 모두 합쳐 최대 16개의 응답을 동시에 생성할 수 있습니다. 연결은 추가 response.create 이벤트도 받아들이며, 진행 중인 응답이 완료될 때까지 해당 요청을 대기열에 둡니다.
하나의 연결에서 이름이 지정된 레인에 사용할 수 있는 고유한 stream_id 값은 최대 32개입니다. 암시적으로 사용되는 기본 레인은 이 한도에 포함되지 않습니다. 한도에 도달하면 기존 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

stream_id를 재사용하더라도 previous_response_id를 지정하지 않으면 대화를 이어가지 않고 새 응답을 시작합니다.

주요 호출은 다음과 같습니다.

# 소켓 하나에서 독립적인 대화 두 개를 시작합니다.
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,
)

전체 예제

대화를 병렬로 실행한 뒤 하나를 포크하기

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

음성 기능 개발

연결

오디오 처리

서비스 출시하기

성능 및 품질

평가

비용 및 처리량

안전 및 거버넌스

인프라 및 접근 권한

운영

핵심 개념

기획

개발

테스트 및 게시

전환 명세

가이드

자료

사용자 식별

ChatGPT 요금제 사용량

가이드

파일 업로드

API

성과 측정

광고주 API

API 레퍼런스

시작하기

시작하기

기초

둘러보기

지원 플랫폼

릴리스

워크플로우

주요 기능

Sites

dot

ChatGPT 스페이스

레퍼런스

맞춤 설정

설정 파일

에이전트 설정

ChatGPT와 Codex 확장하기

Linux

Windows

개발 워크플로우

확장 및 자동화

환경

Codex Cloud

Codex로 개발하기

서드 파티 통합

레퍼런스

권한

Codex Security

사이버 안전

시작하기

기능 설정

ID 및 액세스

서버 측 컨텍스트 압축(`context_management`)

독립형 `/responses/compact`