Los eventos informan lo que sucede mientras un agente trabaja. 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 su historial guardado.
Tu aplicación envía 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. Consulta Ejecutar y continuar sesiones para saber cómo enviar entradas.
Suscríbete antes de enviar trabajo para que tu aplicación reciba los primeros eventos del turno. Pasa tu cliente de API, el ID de sesión de la conversación y un manejador de eventos:
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 los tipos de eventos comunes. Continúa al recibir agent.session.idle y retorna cuando se completa el turno raíz. Genera un error si el turno raíz falla o se cancela, si la sesión o el entorno fallan, o si llega un evento error. Los eventos de turnos de subagentes no finalizan el flujo. Tu manejador decide cómo mostrar la salida; el código que invoca la función auxiliar maneja los errores que esta genera. Si el flujo se cierra antes de que termine un turno, la función auxiliar genera un error. Consulta Recuperar un flujo desconectado.
Enviar un mensaje después de suscribirte
Esta versión acepta un mensaje y lo envía después de abrir el flujo:
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
Usa el type del evento para decidir qué debe hacer tu aplicación:
- Mostrar texto: agrega
agent.session.turn.output_text.delta al final de la parte de contenido correspondiente. Cuando llegue agent.session.turn.output_text.done, reemplaza esa parte por su texto completo. Es posible que no haya deltas.
- Seguir el trabajo: los eventos de sesión, turno y elemento informan sobre el progreso. Comprueba si se recibe
agent.session.turn.completed, agent.session.turn.failed o agent.session.turn.cancelled para determinar el resultado del turno.
- Proporcionar la entrada requerida: al recibir
agent.session.requires_action, recupera la sesión e inspecciona required_actions. Es posible que tu código deba devolver el resultado de una función o conectar un entorno.
Una sesión inactiva o un flujo cerrado no indican por sí solos que el trabajo haya tenido éxito. Un turno completado tampoco garantiza que todas las herramientas hayan tenido éxito. Inspecciona la salida del agente.
Usa item_id, output_index y content_index para asociar las actualizaciones de texto con la misma parte de contenido. Por ejemplo, estos eventos abreviados actualizan una 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 compartido 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 de transmisión continua para conocer todos los tipos de eventos y sus campos. Estos eventos del flujo son distintos de los webhooks. Para obtener información sobre la actividad de los subagentes y la atribución de comandos, consulta Observar la delegación.
Usa el ID de sesión del estado de la conversación de tu aplicación para recuperar el trabajo guardado:
Los puntos de acceso de listado devuelven 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.
Los flujos no vuelven a transmitir los eventos perdidos. Para restaurar la vista de tu aplicación:
- Abre un nuevo flujo y almacena los eventos entrantes en un búfer.
- Recupera la sesión y sus elementos guardados mientras el flujo permanece conectado.
- Restaura tu estado local a partir de esos elementos, usando el ID de cada elemento como clave.
- Aplica las actualizaciones de elementos almacenadas en el búfer usando
item_id. Descarta las actualizaciones de los elementos que ya alcanzaron su estado final en el historial recuperado.
- Reanuda el manejo de eventos en tiempo real.
Un evento output_text.done puede reemplazar un búfer temporal de texto por el texto completo. Los elementos guardados te permiten recuperar el trabajo completado, pero no todos los eventos intermedios que te perdiste.
