Eventos y elementos | OpenAI API

Los eventos indican qué sucede mientras trabaja un agente. Los elementos son los mensajes y las llamadas a herramientas que se guardan y que puedes recuperar más adelante. Usa los eventos para actualizar tu aplicación en tiempo real y los elementos para mostrar el historial guardado.

Tu aplicación usa eventos de entrada para enviar mensajes, cancelar turnos o devolver resultados de herramientas. El agente envía eventos que informan sobre la salida y los cambios en la sesión. Para enviar datos de entrada, consulta Ejecutar y continuar sesiones.

Consumir un flujo de eventos

Suscríbete antes de enviar una tarea para que tu aplicación reciba los primeros eventos del turno. Pasa como argumentos tu cliente de API, el ID de sesión de la conversación y un manejador de eventos:

Recibir eventos de sesión en tiempo real

Python

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// Pass your saved session ID to this helper.
async function streamSession(client, sessionId, handleEvent) {
  const events = await client.beta.agents.sessions.events.stream(sessionId);
  try {
    for await (const event of events) {
      await handleEvent(event);
      switch (event.type) {
        case "agent.session.idle":
          continue;
        case "error":
          throw new Error(event.error.message);
        case "agent.session.failed":
        case "agent.session.environment.failed":
          throw new Error(`Agent lifecycle failure: ${event.type}`);
        case "agent.session.turn.failed":
          if (event.turn.subagent_id === null) {
            throw new Error(
              `${event.type}: ${event.turn.error?.message ?? ""}`
            );
          }
          break;
        case "agent.session.turn.cancelled":
          if (event.turn.subagent_id === null) {
            throw new Error("The agent turn was cancelled");
          }
          break;
        case "agent.session.turn.completed":
          if (event.turn.subagent_id === null) return;
          break;
      }
    }
    throw new Error(
      "Stream closed before a turn ended. Retrieve the saved state."
    );
  } finally {
    events.controller.abort();
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23# Pass your saved session ID to this helper.
def stream_session(client: OpenAI, session_id: str, handle_event):
    with client.beta.agents.sessions.events.stream(session_id) as events:
        for event in events:
            handle_event(event)
            match event.type:
                case "agent.session.idle":
                    continue
                case "error":
                    raise RuntimeError(event.error.message)
                case "agent.session.failed" | "agent.session.environment.failed":
                    raise RuntimeError(f"Agent lifecycle failure: {event.type}")
                case "agent.session.turn.failed":
                    if event.turn.subagent_id is None:
                        detail = event.turn.error.message if event.turn.error else ""
                        raise RuntimeError(f"{event.type}: {detail}")
                case "agent.session.turn.cancelled":
                    if event.turn.subagent_id is None:
                        raise RuntimeError("The agent turn was cancelled")
                case "agent.session.turn.completed":
                    if event.turn.subagent_id is None:
                        return
    raise RuntimeError("Stream closed before a turn ended. Retrieve the saved state.")

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// Pass your saved session ID to this helper.
func streamSession(ctx context.Context, client *openai.Client, sessionID string, handleEvent func(openai.AgentSessionEventUnion)) error {
	events := client.Beta.Agents.Sessions.Events.StreamStreaming(ctx, sessionID)
	defer events.Close()
	for events.Next() {
		event := events.Current()
		handleEvent(event)
		switch event.Type {
		case "agent.session.idle":
			continue
		case "error":
			return fmt.Errorf("agent error: %s", event.RawJSON())
		case "agent.session.failed", "agent.session.environment.failed":
			return fmt.Errorf("agent lifecycle failure: %s", event.RawJSON())
		case "agent.session.turn.failed", "agent.session.turn.cancelled":
			if event.Turn.SubagentID == "" {
				return fmt.Errorf("agent turn did not complete: %s", event.RawJSON())
			}
		case "agent.session.turn.completed":
			if event.Turn.SubagentID == "" {
				return nil
			}
		}
	}
	if err := events.Err(); err != nil {
		return err
	}
	return fmt.Errorf("stream closed before a turn ended; retrieve the saved state")
}

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// Pass your saved session ID to this helper.
public static void streamSession(
    OpenAIClient client, String sessionId, Consumer<AgentSessionEvent> handleEvent) {
  try (StreamResponse<AgentSessionEvent> events =
      client.beta().agents().sessions().events().streamStreaming(sessionId)) {
    var iterator = events.stream().iterator();
    while (iterator.hasNext()) {
      var event = iterator.next();
      handleEvent.accept(event);
      if (event.idle().isPresent()) {
        continue;
      }
      if (event.error().isPresent()) {
        throw new IllegalStateException("Agent error: " + event);
      }
      if (event.failed().isPresent() || event.environmentFailed().isPresent()) {
        throw new IllegalStateException("Agent lifecycle failure: " + event);
      }
      if (event.turnFailed().filter(e -> e.turn().subagentId().isEmpty()).isPresent()
          || event.turnCancelled().filter(e -> e.turn().subagentId().isEmpty()).isPresent()) {
        throw new IllegalStateException("Agent turn did not complete: " + event);
      }
      if (event.turnCompleted().filter(e -> e.turn().subagentId().isEmpty()).isPresent()) {
        return;
      }
    }
    throw new IllegalStateException(
        "Stream closed before a turn ended. Retrieve the saved state.");
  }
}

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# Pass your saved session ID to this helper.
def stream_session(client, session_id, &handle_event)
  events = client.beta.agents.sessions.events.stream_streaming(session_id)
  begin
    events.each do |event|
      handle_event.call(event)
      case event.type.to_s
      when "agent.session.idle"
        next
      when "error"
        raise event.error.message
      when "agent.session.failed", "agent.session.environment.failed"
        raise "Agent lifecycle failure: #{event.type}"
      when "agent.session.turn.failed"
        raise "#{event.type}: #{event.turn.error&.message}" if event.turn.subagent_id.nil?
      when "agent.session.turn.cancelled"
        raise "The agent turn was cancelled" if event.turn.subagent_id.nil?
      when "agent.session.turn.completed"
        return nil if event.turn.subagent_id.nil?
      end
    end
    raise "Stream closed before a turn ended. Retrieve the saved state."
  ensure
    events.close
  end
end

1
2
3
4
5curl -N \
  "https://api.openai.com/v1/agents/sessions/$session_id/events?stream=true" \
  -H "OpenAI-Beta: agents=v1" \
  -H "Authorization: Bearer $OPENAI_API_KEY" \
  -H "Accept: text/event-stream"

La función auxiliar pasa cada evento a tu manejador y luego comprueba si corresponde a uno de los tipos de evento habituales. Si recibe agent.session.idle, sigue procesando eventos. Cuando se completa el turno del agente raíz, la función devuelve el control al código que la llamó. Lanza un error si el turno del agente raíz falla o se cancela, si falla la sesión o el entorno, o si llega un evento error. Los eventos de los turnos de subagentes no ponen fin al flujo. Tu manejador decide cómo mostrar la salida. El código que llama a la función auxiliar se encarga de manejar los errores que esta lanza. Si el flujo se cierra antes de que termine un turno, la función auxiliar lanza un error. Consulta Cómo recuperar un flujo tras una desconexión.

Enviar un mensaje después de suscribirte

Esta versión acepta un mensaje y lo envía después de abrir el flujo:

Enviar un mensaje y recibir eventos en tiempo real

Python

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// Pass your saved session ID and message to this helper.
async function sendAndStream(client, sessionId, text, handleEvent) {
  const events = await client.beta.agents.sessions.events.stream(sessionId);
  try {
    await client.beta.agents.sessions.events.create(sessionId, {
      events: [
        {
          type: "agent.session.input.message",
          input: [{ role: "user", content: [{ type: "input_text", text }] }],
        },
      ],
    });
    for await (const event of events) {
      await handleEvent(event);
      switch (event.type) {
        case "agent.session.idle":
          continue;
        case "error":
          throw new Error(event.error.message);
        case "agent.session.failed":
        case "agent.session.environment.failed":
          throw new Error(`Agent lifecycle failure: ${event.type}`);
        case "agent.session.turn.failed":
          if (event.turn.subagent_id === null) {
            throw new Error(
              `${event.type}: ${event.turn.error?.message ?? ""}`
            );
          }
          break;
        case "agent.session.turn.cancelled":
          if (event.turn.subagent_id === null) {
            throw new Error("The agent turn was cancelled");
          }
          break;
        case "agent.session.turn.completed":
          if (event.turn.subagent_id === null) return;
          break;
      }
    }
    throw new Error(
      "Stream closed before a turn ended. Retrieve the saved state."
    );
  } finally {
    events.controller.abort();
  }
}

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# Pass your saved session ID and message to this helper.
def send_and_stream(client: OpenAI, session_id: str, text, handle_event):
    with client.beta.agents.sessions.events.stream(session_id) as events:
        client.beta.agents.sessions.events.create(
            session_id,
            events=[
                {
                    "type": "agent.session.input.message",
                    "input": [
                        {
                            "role": "user",
                            "content": [{"type": "input_text", "text": text}],
                        }
                    ],
                }
            ],
        )
        for event in events:
            handle_event(event)
            match event.type:
                case "agent.session.idle":
                    continue
                case "error":
                    raise RuntimeError(event.error.message)
                case "agent.session.failed" | "agent.session.environment.failed":
                    raise RuntimeError(f"Agent lifecycle failure: {event.type}")
                case "agent.session.turn.failed":
                    if event.turn.subagent_id is None:
                        detail = event.turn.error.message if event.turn.error else ""
                        raise RuntimeError(f"{event.type}: {detail}")
                case "agent.session.turn.cancelled":
                    if event.turn.subagent_id is None:
                        raise RuntimeError("The agent turn was cancelled")
                case "agent.session.turn.completed":
                    if event.turn.subagent_id is None:
                        return
    raise RuntimeError("Stream closed before a turn ended. Retrieve the saved state.")

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// Pass your saved session ID and message to this helper.
func sendAndStream(ctx context.Context, client *openai.Client, sessionID string, text string, handleEvent func(openai.AgentSessionEventUnion)) error {
	events := client.Beta.Agents.Sessions.Events.StreamStreaming(ctx, sessionID)
	defer events.Close()
	if err := events.Err(); err != nil {
		return err
	}
	err := client.Beta.Agents.Sessions.Events.New(ctx,
		sessionID,
		openai.BetaAgentSessionEventNewParams{
			Events: []openai.AgentSessionInputParamUnion{
				{
					OfParamAgentSessionInputMessage: &openai.AgentSessionInputParamAgentSessionInputMessage{
						Input: []openai.AgentSessionInputMessageParam{
							{
								Content: []openai.InputContentParamUnion{
									{
										OfParamInputText: &openai.InputContentParamInputText{
											Text: text,
										},
									},
								},
							},
						},
					},
				},
			},
		})
	if err != nil {
		return err
	}
	for events.Next() {
		event := events.Current()
		handleEvent(event)
		switch event.Type {
		case "agent.session.idle":
			continue
		case "error":
			return fmt.Errorf("agent error: %s", event.RawJSON())
		case "agent.session.failed", "agent.session.environment.failed":
			return fmt.Errorf("agent lifecycle failure: %s", event.RawJSON())
		case "agent.session.turn.failed", "agent.session.turn.cancelled":
			if event.Turn.SubagentID == "" {
				return fmt.Errorf("agent turn did not complete: %s", event.RawJSON())
			}
		case "agent.session.turn.completed":
			if event.Turn.SubagentID == "" {
				return nil
			}
		}
	}
	if err := events.Err(); err != nil {
		return err
	}
	return fmt.Errorf("stream closed before a turn ended; retrieve the saved state")
}

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// Pass your saved session ID and message to this helper.
public static void sendAndStream(
    OpenAIClient client, String sessionId, String text, Consumer<AgentSessionEvent> handleEvent) {
  try (StreamResponse<AgentSessionEvent> events =
      client.beta().agents().sessions().events().streamStreaming(sessionId)) {
    client
        .beta()
        .agents()
        .sessions()
        .events()
        .create(
            EventCreateParams.builder()
                .sessionId(sessionId)
                .addEvent(
                    AgentSessionInputParam.AgentSessionInputMessage.builder()
                        .addInput(
                            AgentSessionInputMessageParam.builder()
                                .addInputTextContent(text)
                                .build())
                        .build())
                .build());
    var iterator = events.stream().iterator();
    while (iterator.hasNext()) {
      var event = iterator.next();
      handleEvent.accept(event);
      if (event.idle().isPresent()) {
        continue;
      }
      if (event.error().isPresent()) {
        throw new IllegalStateException("Agent error: " + event);
      }
      if (event.failed().isPresent() || event.environmentFailed().isPresent()) {
        throw new IllegalStateException("Agent lifecycle failure: " + event);
      }
      if (event.turnFailed().filter(e -> e.turn().subagentId().isEmpty()).isPresent()
          || event.turnCancelled().filter(e -> e.turn().subagentId().isEmpty()).isPresent()) {
        throw new IllegalStateException("Agent turn did not complete: " + event);
      }
      if (event.turnCompleted().filter(e -> e.turn().subagentId().isEmpty()).isPresent()) {
        return;
      }
    }
    throw new IllegalStateException(
        "Stream closed before a turn ended. Retrieve the saved state.");
  }
}

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# Pass your saved session ID and message to this helper.
def send_and_stream(client, session_id, text, &handle_event)
  events = client.beta.agents.sessions.events.stream_streaming(session_id)
  begin
    client.beta.agents.sessions.events.create(
      session_id,
      events: [
        {
          type: "agent.session.input.message",
          input: [
            {
              role: "user",
              content: [
                {
                  type: "input_text",
                  text: text
                }
              ]
            }
          ]
        }
      ]
    )
    events.each do |event|
      handle_event.call(event)
      case event.type.to_s
      when "agent.session.idle"
        next
      when "error"
        raise event.error.message
      when "agent.session.failed", "agent.session.environment.failed"
        raise "Agent lifecycle failure: #{event.type}"
      when "agent.session.turn.failed"
        raise "#{event.type}: #{event.turn.error&.message}" if event.turn.subagent_id.nil?
      when "agent.session.turn.cancelled"
        raise "The agent turn was cancelled" if event.turn.subagent_id.nil?
      when "agent.session.turn.completed"
        return nil if event.turn.subagent_id.nil?
      end
    end
    raise "Stream closed before a turn ended. Retrieve the saved state."
  ensure
    events.close
  end
end

Manejar las actualizaciones

Usa el campo type del evento para decidir qué debe hacer tu aplicación:

Mostrar texto: agrega el texto de agent.session.turn.output_text.delta al final de la parte del contenido que corresponda. Cuando llegue agent.session.turn.output_text.done, reemplaza esa parte con el texto completo. Es posible que no recibas actualizaciones parciales.
Seguir el progreso: los eventos de sesión, turno y elemento indican cómo avanza el trabajo. Comprueba si recibes agent.session.turn.completed, agent.session.turn.failed o agent.session.turn.cancelled para determinar el resultado del turno.
Proporcionar los datos de entrada requeridos: cuando recibas agent.session.requires_action, recupera la sesión y revisa required_actions. Es posible que tu código deba devolver el resultado de una función, conectar un entorno o gestionar el acceso a un origen desde el navegador o el inicio de sesión.

Que la sesión esté inactiva o que el flujo se haya cerrado no basta para confirmar que el turno se completó correctamente. Que un turno se haya completado tampoco garantiza que todas las herramientas se hayan ejecutado correctamente. Revisa la salida del agente.

Usa item_id, output_index y content_index para asociar las actualizaciones de texto con la misma parte del contenido. Por ejemplo, estos eventos abreviados actualizan una misma parte:

1234567{
  "type": "agent.session.turn.output_text.delta",
  "item_id": "msg_789",
  "output_index": 0,
  "content_index": 0,
  "delta": "Acme competes"
}

1234567{
  "type": "agent.session.turn.output_text.done",
  "item_id": "msg_789",
  "output_index": 0,
  "content_index": 0,
  "text": "Acme competes on price and distribution."
}

Cada evento tiene su propio event_id. El item_id que comparten identifica el elemento guardado, que incluye el contenido, el estado y la fase del mensaje. Consulta Recuperar el trabajo guardado.

Consulta la referencia de eventos en tiempo real para ver todos los tipos de eventos y sus campos. Los eventos de este flujo son distintos de los webhooks. Para ver la actividad de los subagentes y saber cuál ejecutó cada comando, consulta Observar la delegación.

Obtener elementos y turnos

Usa el ID de sesión que figura en el estado de la conversación de tu aplicación para recuperar el trabajo guardado:

Elementos de la sesión: obtén la lista de elementos para recuperar los mensajes y las llamadas a herramientas del agente raíz a lo largo de los turnos.
Turnos: obtén la lista de turnos para explorar el trabajo de la sesión. Recupera un turno por su ID para revisar su estado, sus marcas de tiempo, sus datos de uso y cualquier error registrado.
Elementos de un turno: para un turno del agente raíz, usa el punto de acceso Listar elementos del turno. Cada subagente tiene su propio historial de elementos y un punto de acceso para consultar los elementos de cada turno.

Los puntos de acceso que devuelven listas entregan una página a la vez. Usa las funciones auxiliares de paginación del SDK o el cursor after para recuperar más resultados. Es posible que una sola página no contenga todos los elementos de un turno. Usa order: "asc" para leer los elementos del más antiguo al más reciente.

Cómo recuperar un flujo tras una desconexión

Recupera la salida que no recibiste a partir de los elementos guardados. Para restaurar la vista de tu aplicación:

Abre un flujo nuevo y almacena temporalmente los eventos entrantes.
Recupera la sesión y sus elementos guardados mientras el flujo sigue conectado.
Restaura el estado local a partir de esos elementos y usa el ID de cada elemento como clave.
Usa item_id para aplicar a cada elemento las actualizaciones que almacenaste temporalmente. Descarta las actualizaciones de los elementos que ya alcanzaron su estado final en el historial recuperado.
Reanuda el procesamiento de eventos en tiempo real.

Cuando te vuelves a conectar a una sesión que falló, el flujo informa del fallo registrado y se cierra. Si la sesión recuperada tiene status: "failed" o recibes agent.session.failed, deja de intentar conectarte y sigue las instrucciones para recuperar la sesión.

Restaura los formularios de entrada pendientes a partir de required_actions de la sesión recuperada. Los elementos del historial no indican qué solicitudes aún requieren una respuesta. Para las aprobaciones del navegador, identifica el formulario de cada solicitud mediante request_id y elimina los formularios de las solicitudes que ya no estén pendientes. Después de volver a conectarte, no necesitas reenviar la tarea ni las aprobaciones anteriores.

Un evento output_text.done puede reemplazar el contenido de un búfer de texto temporal con el texto completo. Los elementos guardados te permiten recuperar el trabajo completado, pero no todos los eventos intermedios que no recibiste.

Sugerencias

Primeros pasos

Conceptos fundamentales

SDK y CLI

Proveedores de servicios en la nube

Recursos

API heredadas

Elige un modelo

Texto y código

Redacción de prompts

Razonamiento

Imágenes

Tiempo real y audio

Modelos especializados

API de agentes

Agents SDK

ChatKit

Búsqueda y recuperación de información

Conecta herramientas y datos

Crea flujos de trabajo con herramientas

Computadora y código

Multimedia

GPT-Live

Realtime API

Desarrollar aplicaciones de voz

Conexiones

Procesamiento de audio

Pasa a producción

Rendimiento y calidad

Evaluación

Costo y rendimiento

Seguridad y gobernanza

Infraestructura y acceso

Operaciones

Conceptos básicos

Planificación

Desarrollo

Pruebas y publicación

Especificaciones de conversión

Guías

Recursos

Identidad

Uso del plan de ChatGPT

Guías

Carga de archivos

API

Medición

API para anunciantes

Referencia de la API

Primeros pasos

Primeros pasos

Fundamentos

Explorar

Disponible en

Versiones

Flujos de trabajo

Capacidades

Sites

dots

ChatGPT Espacio

Referencia

Personalización

Archivo de configuración

Configuración de agentes

Ampliar las funciones de ChatGPT y Codex

Linux

Windows

Flujos de trabajo de desarrollo

Ampliar funciones y automatizar

Entornos

Codex Cloud

Desarrollar con Codex

Integraciones de terceros

Referencia

Permisos

Codex Security

Ciberseguridad

Primeros pasos

Configuración de funciones

Identidad y acceso