Ereignisse melden, was passiert, während ein Agent arbeitet. Elemente sind die gespeicherten Nachrichten und Werkzeugaufrufe, die du später abrufen kannst. Verwende Ereignisse, um deine Anwendung in Echtzeit zu aktualisieren, und Elemente, um ihren gespeicherten Verlauf anzuzeigen.
Deine Anwendung sendet Eingabeereignisse, um Nachrichten zu übermitteln, Durchläufe abzubrechen oder Werkzeugergebnisse zurückzugeben. Der Agent sendet Ereignisse, die Ausgaben und Änderungen an der Sitzung melden. Wie du Eingaben sendest, erfährst du unter Sitzungen ausführen und fortsetzen.
Abonniere den Stream, bevor du Aufgaben sendest, damit deine Anwendung auch die ersten Ereignisse des Durchlaufs empfängt. Übergib deinen API-Client, die Sitzungs-ID der Unterhaltung und einen Ereignishandler:
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"
Die Hilfsfunktion übergibt jedes Ereignis an deinen Handler und prüft anschließend auf gängige Ereignistypen. Bei agent.session.idle setzt sie die Verarbeitung fort und kehrt zurück, sobald der Root-Turn abgeschlossen ist. Sie löst einen Fehler aus, wenn der Root-Turn fehlschlägt oder abgebrochen wird, die Sitzung oder Umgebung ausfällt oder ein error-Ereignis eintrifft. Ereignisse aus Subagenten-Turns beenden den Stream nicht. Dein Handler entscheidet, wie die Ausgabe angezeigt wird; der aufrufende Code behandelt Fehler der Hilfsfunktion. Wird der Stream geschlossen, bevor ein Turn endet, löst die Hilfsfunktion einen Fehler aus. Siehe Einen unterbrochenen Stream wiederherstellen.
Nach dem Abonnieren eine Nachricht senden
Diese Version nimmt eine Nachricht entgegen und sendet sie, nachdem der Stream geöffnet wurde:
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
Entscheide anhand von type des Ereignisses, was deine Anwendung tun soll:
- Text anzeigen: Hänge
agent.session.turn.output_text.delta an den entsprechenden Inhaltsteil an. Sobald agent.session.turn.output_text.done eintrifft, ersetze diesen Teil durch seinen vollständigen Text. Deltas können fehlen.
- Arbeit verfolgen: Ereignisse zu Sitzungen, Durchläufen und Elementen melden den Fortschritt. Prüfe auf
agent.session.turn.completed, agent.session.turn.failed oder agent.session.turn.cancelled, um das Ergebnis des Durchlaufs zu ermitteln.
- Erforderliche Eingaben bereitstellen: Rufe bei
agent.session.requires_action die Sitzung ab und prüfe required_actions. Dein Code muss möglicherweise ein Funktionsergebnis zurückgeben oder eine Umgebung verbinden.
Eine inaktive Sitzung oder ein geschlossener Stream allein belegt keinen Erfolg. Auch ein abgeschlossener Durchlauf garantiert nicht, dass jedes Werkzeug erfolgreich ausgeführt wurde. Prüfe die Ausgabe des Agenten.
Verwende item_id, output_index und content_index, um Textaktualisierungen demselben Inhaltsteil zuzuordnen. Diese gekürzt dargestellten Ereignisse aktualisieren beispielsweise einen Teil:
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."
}
Jedes Ereignis hat eine eigene event_id. Die gemeinsame item_id identifiziert das gespeicherte Element, das Inhalt, Status und Phase der Nachricht enthält. Siehe Gespeicherte Arbeit abrufen.
Alle Ereignistypen und Felder findest du in der Referenz zu Streaming-Ereignissen. Diese Stream-Ereignisse unterscheiden sich von Webhooks. Informationen zur Aktivität von Subagenten und zur Zuordnung von Befehlen findest du unter Delegation beobachten.
Verwende die Sitzungs-ID aus dem Gesprächszustand deiner Anwendung, um gespeicherte Arbeit abzurufen:
Listenendpunkte geben jeweils eine Seite zurück. Verwende die Hilfsfunktionen des SDK zur Paginierung oder den Cursor after, um weitere Ergebnisse abzurufen. Eine einzelne Seite enthält möglicherweise nicht alle Elemente eines Durchlaufs. Verwende order: "asc", um die Elemente vom ältesten zum neuesten zu lesen.
Streams geben verpasste Ereignisse nicht erneut wieder. So stellst du die Ansicht deiner Anwendung wieder her:
- Öffne einen neuen Stream und puffere eingehende Ereignisse.
- Rufe die Sitzung und ihre gespeicherten Elemente ab, während der Stream verbunden bleibt.
- Stelle deinen lokalen Zustand anhand dieser Elemente wieder her und verwende dabei die Element-ID als Schlüssel.
- Wende die gepufferten Elementaktualisierungen mithilfe von
item_id an. Verwirf Aktualisierungen für Elemente, die im abgerufenen Verlauf bereits ihren endgültigen Zustand erreicht haben.
- Setze die Verarbeitung von Live-Ereignissen fort.
Ein Ereignis vom Typ output_text.done kann einen temporären Textpuffer durch den vollständigen Text ersetzen. Mit gespeicherten Elementen kannst du abgeschlossene Arbeit wiederherstellen, aber nicht jedes verpasste Zwischenereignis.
