Mode WebSocket | OpenAI API

L’API Responses propose un mode WebSocket pour les workflows de longue durée qui font de nombreux appels d’outils. Outre la réduction de la latence, stream_id permet le multiplexage WebSocket. Une seule connexion persistante à /v1/responses permet de mener des conversations en parallèle et de forker une conversation existante sur un nouveau flux. À chaque nouveau tour, envoyez uniquement les nouveaux éléments d’entrée et previous_response_id.

Le mode WebSocket est compatible avec la politique de non-conservation des données (ZDR) et avec store=false.

Pourquoi utiliser le mode WebSocket ?

Le mode WebSocket est particulièrement utile lorsqu’un workflow nécessite de nombreux allers-retours entre le modèle et les outils, par exemple en programmation agentique ou dans des boucles d’orchestration qui appellent plusieurs fois des outils.

Comme la connexion reste ouverte et que chaque tour ne transmet que les nouveaux éléments d’entrée, le mode WebSocket réduit le temps nécessaire pour passer d’un tour au suivant. La latence de bout en bout diminue ainsi sur les longues chaînes de réponses. Lors d’exécutions comportant au moins 20 appels d’outils, nous avons observé des gains de vitesse allant jusqu’à environ 40 % de bout en bout.

Ouvrez une connexion et créez des réponses

Installez les dépendances WebSocket avec pip install "openai[realtime]>=3.8.0" pour Python, npm install openai@^7.10.0 ws pour JavaScript ou gem install openai async-websocket pour Ruby.

Pour Go, exécutez go get github.com/openai/openai-go/v3@v3.98.0. Pour Java, ajoutez la dépendance Maven com.openai:openai-java:4.101.0 et utilisez Java 17 ou une version ultérieure pour ces exemples. Ces versions des SDK Go et Java prennent en charge nativement le mode WebSocket de l’API Responses.

En mode WebSocket, commencez chaque tour en envoyant un événement response.create depuis le client. Sa charge utile reprend le corps habituel d’une requête de création de réponse, à l’exception des champs propres au transport comme stream et background, qui ne sont pas utilisés.

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

Les clients peuvent préparer l’état de la requête à l’avance en envoyant response.create avec generate: false. Cette option est utile si vous connaissez déjà les outils, les instructions ou les messages personnalisés que vous prévoyez d’envoyer lors d’un prochain tour. Avec generate: false, le modèle ne produit aucune sortie, mais l’état de la requête est prêt pour que la génération puisse démarrer plus rapidement au prochain tour. La requête de préparation renvoie un identifiant de réponse que vous pouvez utiliser avec previous_response_id pour poursuivre la chaîne, y compris lors de tours ultérieurs. La section suivante explique comment poursuivre une session avec previous_response_id et des entrées incrémentales.

Poursuivez avec des entrées incrémentales

Pour ajouter des instructions de l’utilisateur pendant la génération d’une réponse, orientez la réponse en cours de génération. Cette méthode préserve le travail déjà effectué et intègre les nouvelles instructions à la suite de la génération. Pour passer normalement d’un tour au suivant ou transmettre des résultats d’outils, utilisez response.create comme indiqué ci-dessous.

Pour poursuivre une exécution, envoyez un nouvel événement response.create avec les champs suivants :

previous_response_id doit contenir l’identifiant de la réponse précédente.
input doit contenir uniquement les nouveaux éléments, par exemple les sorties des outils et le prochain message de l’utilisateur.

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

Fonctionnement de l’enchaînement des réponses

Le mode WebSocket applique les mêmes règles que le mode HTTP pour enchaîner les réponses avec previous_response_id, mais permet de poursuivre l’exécution avec une latence plus faible sur la connexion active.

Sur une connexion WebSocket active, le service conserve l’état des réponses précédentes les plus récentes dans un cache en mémoire propre à la connexion. Lorsque vous utilisez stream_id, chaque flux conserve sa dernière réponse en cache. Vous pouvez donc enchaîner rapidement à partir de cette réponse, car le service peut réutiliser l’état déjà disponible sur la connexion. Comme le service conserve cet état uniquement en mémoire, sans l’écrire sur disque, vous pouvez utiliser le mode WebSocket avec store=false et respecter la politique de non-conservation des données (ZDR).

Si un previous_response_id est absent du cache en mémoire, le comportement dépend de votre choix de stocker ou non les réponses :

Avec store=true, le service peut retrouver des réponses plus anciennes à partir de leur identifiant si leur état est disponible dans le stockage persistant. L’enchaînement reste alors possible, mais sans le gain de latence que procure le cache en mémoire.
Avec store=false (y compris sous ZDR), le service ne peut pas se rabattre sur le stockage persistant. Si l’identifiant est absent du cache, la requête renvoie previous_response_not_found.

Si une requête de continuation sur le même flux renvoie une erreur 4xx ou 5xx, le service retire du cache de la connexion le previous_response_id auquel la requête faisait référence. Si un fork vers un autre flux renvoie une erreur, le service conserve la réponse parente commune pour que le flux d’origine puisse se poursuivre.

Compactage et création de nouvelles réponses

Si vous utilisez le compactage, la manière de poursuivre dépend du type de compactage utilisé.

Compactage côté serveur (`context_management`)

Lorsque vous activez le compactage côté serveur (context_management avec compact_threshold), le contexte est compacté pendant la génération habituelle via /responses. En mode WebSocket, poursuivez comme d’habitude : envoyez le prochain événement response.create avec le dernier previous_response_id et uniquement les nouveaux éléments d’entrée.

Compactage autonome avec `/responses/compact`

Le point de terminaison autonome /responses/compact renvoie une nouvelle fenêtre d’entrée compactée, sans identifiant de réponse. Après le compactage, créez une réponse sur votre connexion WebSocket en utilisant la fenêtre compactée comme input et en y ajoutant les prochains éléments provenant de l’utilisateur ou des outils.

Démarrez une nouvelle chaîne en omettant previous_response_id ou en lui attribuant la valeur null. Transmettez le résultat du compactage tel quel, sans retirer d’éléments de la fenêtre renvoyée.

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

Menez des conversations en parallèle

Le paramètre stream_id permet de mener plusieurs conversations en parallèle sur une même connexion. Envoyez à la suite plusieurs événements response.create indépendants avec des valeurs de stream_id différentes. Le serveur peut les traiter simultanément sur la même connexion. Leurs événements peuvent s’entrelacer : utilisez une seule boucle de lecture et acheminez chaque événement en fonction de son stream_id.

Un stream_id identifie un flux ordonné au sein d’une connexion WebSocket. Distinguez bien stream_id de previous_response_id :

stream_id détermine la destination des événements et regroupe les requêtes à exécuter dans leur ordre d’arrivée.
previous_response_id indique à partir de quelle réponse la conversation se poursuit.

Cette distinction permet les deux usages décrits ci-dessous.

une connexion WebSocket
├─ stream_id="planner"   rédigez un plan de déploiement
└─ stream_id="research"  listez les risques du déploiement

Les requêtes ayant le même stream_id s’exécutent dans leur ordre d’arrivée, sans se chevaucher. Les requêtes ayant des valeurs de stream_id différentes peuvent s’exécuter simultanément.

Limites par connexion

Une connexion peut traiter jusqu’à 16 réponses simultanément, tous flux confondus, y compris le flux par défaut. Elle accepte les événements response.create supplémentaires et les met en file d’attente jusqu’à ce qu’une réponse en cours se termine.
Une connexion accepte jusqu’à 32 valeurs distinctes de stream_id pour les flux nommés. Le flux implicite par défaut ne compte pas dans cette limite. Une fois la limite atteinte, réutilisez un stream_id existant ou ouvrez une nouvelle connexion.

Forkez une conversation sur un nouveau flux

Pour créer une branche à partir d’une réponse terminée, envoyez son identifiant dans previous_response_id avec un nouveau stream_id. Tant que cette réponse reste disponible, le nouveau flux hérite de son contexte et le flux d’origine peut se poursuivre. Une fois le fork démarré, les deux branches peuvent s’exécuter en parallèle puisqu’elles utilisent des identifiants de flux différents.

Avec store=false (y compris sous ZDR), un fork vers un autre flux exige que la réponse parente soit toujours dans le cache de la connexion. Si le fork est en file d’attente et que le flux d’origine progresse ou échoue, la réponse parente peut être retirée du cache avant le démarrage du fork. Celui-ci renvoie alors previous_response_not_found. Attendez que le nouveau flux émette response.in_progress avant de poursuivre le flux d’origine, ou réessayez en attribuant à previous_response_id la valeur null et en renvoyant l’intégralité du contexte d’entrée.

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

Si vous réutilisez un stream_id sans previous_response_id, vous démarrez une nouvelle réponse indépendante de la conversation précédente.

Voici les principaux appels :

# Une seule connexion WebSocket pour deux conversations indépendantes.
send_create(connection, "planner", "Rédigez un plan de déploiement.")
send_create(connection, "research", "Listez les risques du déploiement.")

# Forkez la réponse du flux planner, puis poursuivez la branche d’origine en parallèle.
send_create(
    connection,
    "critic",
    "Repérez les lacunes de ce plan.",
    previous_response_id=planner_response_id,
)
wait_for_in_progress(connection, "critic")
send_create(
    connection,
    "planner",
    "Ajoutez les étapes de retour à la version précédente.",
    previous_response_id=planner_response_id,
)

Exemple complet

Menez des conversations en parallèle, puis forkez l’une d’elles

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

Un stream_id doit comporter de 1 à 256 caractères. Seuls les lettres, les chiffres, les traits de soulignement (_), les traits d’union (-) et les points (.) sont autorisés. Utilisez ce champ uniquement dans les événements WebSocket response.create. Omettez-le dans les requêtes HTTP POST /v1/responses.

Pour les flux nommés, les événements du serveur contiennent le stream_id correspondant, y compris les événements de fin et les erreurs propres à une requête.

Si vous omettez stream_id, la requête utilise un flux implicite par défaut et ses événements ne contiennent pas de stream_id. Pour le reste, le flux par défaut suit les mêmes règles d’ordre et d’exécution simultanée que les flux nommés. Une chaîne vide n’est pas une valeur valide pour stream_id. Pour sélectionner le flux par défaut, omettez le champ.

Fonctionnement et limites des connexions

Au sein d’une réponse, les événements suivent le modèle existant de diffusion en continu de l’API Responses. Les événements de différents flux peuvent s’entrelacer.
Les requêtes ayant le même stream_id s’exécutent dans leur ordre d’arrivée, sans se chevaucher. Les requêtes de différents flux peuvent s’exécuter simultanément.
Une connexion peut rester ouverte jusqu’à 60 minutes. Reconnectez-vous une fois cette limite atteinte.

Reconnectez-vous et reprenez les conversations

Lorsqu’une connexion se ferme ou atteint la limite de 60 minutes, son cache disparaît pour tous les flux. Ouvrez une nouvelle connexion WebSocket et reprenez chaque flux selon le cas :

Si vous avez stocké une réponse précédente (store=true) et disposez d’un identifiant de réponse valide, poursuivez le flux avec previous_response_id et les nouveaux éléments d’entrée.
Si vous ne pouvez pas poursuivre un flux (par exemple, avec store=false ou ZDR, ou en cas d’erreur previous_response_not_found), démarrez une nouvelle réponse en attribuant à previous_response_id la valeur null ou en omettant ce champ. Envoyez l’intégralité du contexte d’entrée pour le prochain tour de ce flux.
Si vous avez compacté le contexte avec /responses/compact, utilisez la fenêtre compactée renvoyée comme base du champ input de la nouvelle réponse, puis ajoutez les derniers éléments provenant de l’utilisateur ou des outils.

Erreurs à gérer

Lorsque le serveur peut associer une erreur à un flux nommé, l’événement d’erreur contient stream_id. Les autres flux peuvent se poursuivre après une erreur propre à une requête.

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
}

Suggestions

Premiers pas

Notions fondamentales

SDK et CLI

Fournisseurs de services cloud

Ressources

Anciennes API

Choisissez un modèle

Texte et code

Rédaction de prompts

Raisonnement

Images

Temps réel et audio

Modèles spécialisés

API Agents

Agents SDK

ChatKit

Recherche et récupération d’informations

Connectez des outils et des sources de données

Créez des workflows qui font appel à des outils

Ordinateur et code

Médias

GPT-Live

Realtime API

Créer des applications vocales

Connexions

Traitement audio

Mise en production

Performances et qualité

Évaluation

Coût et débit

Sécurité et gouvernance

Infrastructure et accès

Opérations

Concepts clés

Planification

Développement

Tests et publication

Spécifications de conversion

Guides

Ressources

Identité

Utilisation du forfait ChatGPT

Guides

Envoi de fichiers

API

Mesure des performances

API pour les annonceurs

Référence de l’API

Premiers pas

Premiers pas

Notions de base

Explorer

Disponible sur

Versions

Workflows

Capacités

Sites

dots

Espace ChatGPT

Référence

Personnalisation

Fichier de configuration

Configuration des agents

Étendre les fonctionnalités de ChatGPT et de Codex

Linux

Windows

Workflows de développement

Extensions et automatisation

Environnements

Codex Cloud

Développer avec Codex

Intégrations tierces

Référence

Autorisations

Codex Security

Cybersécurité

Premiers pas

Configuration des fonctionnalités

Identité et accès

Compactage côté serveur (`context_management`)

Compactage autonome avec `/responses/compact`