Modo WebSocket | OpenAI API

A API Responses oferece um modo WebSocket para fluxos de trabalho de longa duração com muitas chamadas de ferramentas. Além de reduzir a latência, esse modo permite usar stream_id para multiplexar conversas via WebSocket. Com uma única conexão persistente a /v1/responses, você pode executar conversas em paralelo e criar um fork de uma conversa existente em um novo fluxo. Para dar continuidade à conversa a cada turno, envie apenas os novos itens de entrada junto com previous_response_id.

O modo WebSocket é compatível tanto com zero retenção de dados (ZDR) quanto com store=false.

Por que usar o modo WebSocket

O modo WebSocket é especialmente útil quando um fluxo de trabalho exige muitas trocas entre o modelo e as ferramentas, como na programação agêntica ou em ciclos de orquestração com chamadas repetidas de ferramentas.

Como a conexão permanece aberta e cada turno envia apenas as novas entradas, o modo WebSocket reduz o custo de dar continuidade a cada turno e diminui a latência de ponta a ponta em cadeias longas. Em execuções com 20 ou mais chamadas de ferramentas, observamos ganhos de velocidade de até cerca de 40% de ponta a ponta.

Conectar e criar respostas

Instale as dependências do WebSocket com pip install "openai[realtime]>=3.8.0" para Python, npm install openai@^7.10.0 ws para JavaScript ou gem install openai async-websocket para Ruby.

Para Go, execute go get github.com/openai/openai-go/v3@v3.98.0. Para Java, adicione a dependência Maven com.openai:openai-java:4.101.0 e use o Java 17 ou posterior nestes exemplos. Essas versões dos SDKs de Go e Java oferecem suporte nativo ao WebSocket da API Responses.

No modo WebSocket, inicie cada turno enviando um evento response.create a partir do cliente. O conteúdo do evento segue o formato habitual do corpo da requisição de criação de respostas, com a diferença de que campos específicos do transporte, como stream e background, não são usados.

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

Os clientes podem preparar o estado da requisição com antecedência enviando response.create com generate: false. Isso é útil quando você já sabe quais ferramentas, instruções ou mensagens personalizadas pretende enviar em um dos próximos turnos. Com generate: false, a requisição não retorna uma saída do modelo, mas prepara o estado para que a geração no próximo turno comece mais rápido. Essa requisição preparatória retorna um ID de resposta que você pode usar em previous_response_id para continuar a cadeia, inclusive em turnos posteriores. A próxima seção explica como continuar uma sessão usando previous_response_id e entradas incrementais.

Continuar com entradas incrementais

Para adicionar instruções do usuário enquanto uma resposta está sendo gerada, use o direcionamento durante o turno. Esse recurso preserva o trabalho já concluído e incorpora as novas instruções à continuação. Use o padrão de response.create a seguir para continuar a conversa entre turnos ou enviar resultados de ferramentas.

Para continuar uma execução, envie outro evento response.create com os seguintes campos:

previous_response_id definido como o ID da resposta anterior.
input contendo apenas os novos itens (por exemplo, as saídas das ferramentas e a próxima mensagem do usuário).

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

Como funciona a continuação

O modo WebSocket segue as mesmas regras do modo HTTP para encadear respostas com previous_response_id, mas permite continuar a conversa com menor latência pelo socket ativo.

Enquanto a conexão WebSocket está ativa, o serviço mantém o estado recente das respostas anteriores em um cache em memória próprio da conexão. Quando você usa stream_id, cada fluxo mantém sua resposta mais recente no cache. Continuar a partir dessa resposta é rápido porque o serviço pode reutilizar o estado mantido na conexão. Como o serviço mantém o estado das respostas anteriores apenas na memória, sem gravá-lo em disco, você pode usar o modo WebSocket de forma compatível com store=false e com zero retenção de dados (ZDR).

Se um previous_response_id não estiver no cache em memória, o comportamento dependerá de você ter optado por armazenar as respostas:

Com store=true, o serviço pode recuperar o estado de respostas mais antigas pelos respectivos IDs quando esse estado estiver disponível no armazenamento persistente. A continuação ainda pode funcionar, mas perde a vantagem de latência do cache em memória.
Com store=false (inclusive com ZDR), não há uma cópia persistente à qual recorrer. Se o ID não estiver no cache, a requisição retornará previous_response_not_found.

Se uma continuação no mesmo fluxo retornar um erro 4xx ou 5xx, o serviço removerá do cache da conexão o previous_response_id referenciado. Se um fork para outro fluxo retornar um erro, o serviço preservará a resposta de origem compartilhada para que o fluxo original possa continuar.

Compactação e criação de novas respostas

Se você usa compactação, há duas formas de continuar:

Compactação no servidor (`context_management`)

Quando você habilita a compactação no servidor (context_management com compact_threshold), a compactação ocorre durante a geração normal de respostas em /responses. No modo WebSocket, continue como de costume. Envie o próximo evento response.create com o previous_response_id mais recente e apenas os novos itens de entrada.

Uso independente de `/responses/compact`

O endpoint /responses/compact retorna uma nova janela de entrada compactada em vez de um ID de resposta. Após a compactação, crie uma nova resposta na conexão WebSocket usando a janela compactada como input e acrescentando os próximos itens do usuário ou das ferramentas.

Para iniciar uma nova cadeia, omita previous_response_id ou defina seu valor como null. Envie a saída compactada como a recebeu, sem remover itens da janela retornada.

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

Executar conversas em paralelo

Você pode manter conversas em paralelo na mesma conexão usando o parâmetro stream_id. Envie eventos response.create independentes, um após o outro, com valores diferentes de stream_id. O servidor pode executá-los simultaneamente em uma única conexão. Como os eventos podem chegar intercalados, mantenha um único laço de leitura e encaminhe cada evento de acordo com seu stream_id.

Um stream_id identifica um fluxo ordenado dentro de uma conexão WebSocket. Distinga as funções de stream_id e previous_response_id:

stream_id determina o destino dos eventos e quais requisições são executadas em ordem de chegada.
previous_response_id determina a partir de qual resposta anterior a conversa continua.

Essa separação permite usar a mesma conexão de duas maneiras.

uma conexão WebSocket
├─ stream_id="planner"   elabore um plano de implantação
└─ stream_id="research"  liste os riscos da implantação

Requisições com o mesmo stream_id são executadas em ordem de chegada, uma por vez. Requisições com valores diferentes de stream_id podem ser executadas simultaneamente.

Limites por conexão

Uma conexão pode ter até 16 respostas em andamento no total, contando os fluxos nomeados e o fluxo padrão. A conexão aceita novos eventos response.create e os mantém na fila até que uma resposta em andamento seja concluída.
Uma conexão aceita até 32 valores distintos de stream_id para fluxos nomeados. O fluxo padrão implícito não conta para esse limite. Ao atingir o limite, reutilize um stream_id existente ou abra uma nova conexão.

Criar um fork de uma conversa em um novo fluxo

Para criar uma ramificação a partir de uma resposta concluída, envie o ID dessa resposta em previous_response_id junto com um novo stream_id. Enquanto a resposta estiver disponível, o novo fluxo herdará seu contexto e o fluxo original poderá continuar. Depois que o fork começar, as duas ramificações poderão ser executadas simultaneamente porque usam IDs de fluxo diferentes.

Com store=false (inclusive com ZDR), um fork para outro fluxo depende de a resposta de origem permanecer no cache da conexão. Se o fluxo original avançar ou falhar enquanto o fork estiver na fila, o serviço poderá remover essa resposta do cache antes que o fork comece. Nesse caso, o fork retornará previous_response_not_found. Aguarde até que o novo fluxo emita response.in_progress antes de avançar o fluxo original, ou tente novamente com previous_response_id definido como null e reenvie todo o contexto de entrada.

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

Reutilizar um stream_id sem previous_response_id inicia uma nova resposta sem dar continuidade à conversa.

Veja as principais chamadas:

# Um socket, duas conversas independentes.
send_create(connection, "planner", "Elabore um plano de implantação.")
send_create(connection, "research", "Liste os riscos da implantação.")

# Crie um fork da resposta do fluxo planner e continue a ramificação original em paralelo.
send_create(
    connection,
    "critic",
    "Identifique lacunas neste plano.",
    previous_response_id=planner_response_id,
)
wait_for_in_progress(connection, "critic")
send_create(
    connection,
    "planner",
    "Adicione as etapas de reversão da implantação.",
    previous_response_id=planner_response_id,
)

Exemplo completo

Execute conversas em paralelo e crie um fork de uma delas

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

Um stream_id deve ter de 1 a 256 caracteres e pode conter apenas letras, números, sublinhados (_), hífens (-) e pontos (.). Use esse campo apenas em eventos response.create via WebSocket. Não o inclua em requisições HTTP POST /v1/responses.

Nos fluxos nomeados, os eventos do servidor incluem o stream_id correspondente, inclusive os eventos finais e os erros específicos de uma requisição.

Se você omitir stream_id, a requisição usará um fluxo padrão implícito e seus eventos não incluirão stream_id. Nos demais aspectos, o fluxo padrão segue as mesmas regras de ordenação e concorrência dos fluxos nomeados. Uma string vazia não é um stream_id válido. Para selecionar o fluxo padrão, omita o campo.

Comportamento e limites da conexão

Os eventos de cada resposta seguem o modelo de eventos de streaming da API Responses. Eventos de fluxos diferentes podem chegar intercalados.
Requisições com o mesmo stream_id são executadas em ordem de chegada, uma por vez. Requisições em fluxos diferentes podem ser executadas simultaneamente.
As conexões duram até 60 minutos. Reconecte-se ao atingir esse limite.

Reconectar e recuperar os fluxos

Quando uma conexão é encerrada ou atinge o limite de 60 minutos, todos os fluxos perdem o cache daquela conexão. Abra uma nova conexão WebSocket e recupere cada fluxo de uma das seguintes formas:

Se você armazenou uma resposta anterior (store=true) e tem um ID de resposta válido, continue o fluxo com previous_response_id e os novos itens de entrada.
Se não for possível continuar um fluxo (por exemplo, ao usar store=false ou ZDR, ou ao receber previous_response_not_found), inicie uma nova resposta definindo previous_response_id como null (ou omitindo o campo) e envie todo o contexto de entrada para o próximo turno desse fluxo.
Se você compactou o contexto com /responses/compact, use a janela compactada retornada como base do input da nova resposta e acrescente os itens mais recentes do usuário ou das ferramentas.

Erros a tratar

Quando o servidor consegue associar um erro a um fluxo nomeado, o evento de erro inclui stream_id. Os demais fluxos podem continuar após um erro específico de uma requisição.

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
}

Sugestões

Primeiros passos

Conceitos fundamentais

SDKs e CLI

Provedores de nuvem

Recursos

APIs legadas

Escolha um modelo

Texto e código

Como escrever prompts

Raciocínio

Imagens

Tempo real e áudio

Modelos especializados

API de agentes

SDK de agentes

ChatKit

Busca e recuperação de informações

Conecte ferramentas e dados

Crie fluxos de trabalho com ferramentas

Computador e código

Mídia

GPT-Live

Realtime API

Desenvolver aplicações de voz

Conexões

Processamento de áudio

Coloque em produção

Desempenho e qualidade

Avaliação

Custo e taxa de processamento

Segurança e governança

Infraestrutura e acesso

Operações

Conceitos fundamentais

Planejar

Desenvolver

Testar e publicar

Especificações de conversão

Guias

Recursos

Identidade

Uso do plano do ChatGPT

Guias

Envio de arquivos

API

Mensuração

API para anunciantes

Referência da API

Primeiros passos

Primeiros passos

Fundamentos

Explorar

Disponível em

Versões

Fluxos de trabalho

Capacidades

Sites

dots

ChatGPT Space

Referência

Personalização

Arquivo de configuração

Configuração de agentes

Ampliar os recursos do ChatGPT e do Codex

Linux

Windows

Fluxos de trabalho de desenvolvimento

Ampliar recursos e automatizar tarefas

Ambientes

Codex Cloud

Desenvolver com o Codex

Integrações de terceiros

Referência

Permissões

Codex Security

Segurança cibernética

Primeiros passos

Configuração de recursos

Identidade e acesso

Compactação no servidor (`context_management`)

Uso independente de `/responses/compact`