Visualizzare in anteprima le risposte con i delta degli eventi

Per impostazione predefinita, il testo della risposta dell'agente raggiunge lo stream degli eventi della sessione come eventi agent.message bufferizzati. Ciascuno viene emesso solo dopo che la richiesta al modello che lo ha prodotto è terminata. Gli "event deltas" (delta degli eventi) ti consentono di visualizzare quel testo in modo incrementale, come anteprima in tempo reale, mentre il modello lo sta ancora generando.

Le anteprime sono un ausilio di visualizzazione fornito secondo il principio del "best effort" (massimo impegno), e l'agent.message bufferizzato è sempre il record autorevole. Un client che ignora le anteprime riceve comunque uno stream completo e corretto.

Attivare le anteprime

Le anteprime sono facoltative e si attivano per singola connessione di stream. Aggiungi il parametro di query event_deltas[] allo stream che stai leggendo e ripetilo una volta per ogni tipo di evento di cui vuoi l'anteprima. I valori accettati sono agent.message e agent.thinking. Qualsiasi altro valore restituisce un errore 400, così come una richiesta con più di 100 valori.

Entrambi gli endpoint di stream accettano il parametro:

• Stream a livello di sessione: GET /v1/sessions/{session_id}/events/stream
• Stream del thread di sessione: GET /v1/sessions/{session_id}/threads/{thread_id}/stream

Le anteprime di un subagente compaiono sullo stream del thread di quel subagente.

[] è un pattern glob della shell, quindi racchiudi l'URL tra virgolette ogni volta che costruisci la richiesta in una shell. Gli esempi codificano le parentesi quadre in percent-encoding come %5B%5D, che funziona ugualmente.

Eventi di anteprima

Quando inizia un evento in anteprima, lo stream emette un event_start che riporta il tipo e l'id dell'evento imminente:

{
  "type": "event_start",
  "event": {
    "type": "agent.message",
    "id": "sevt_01abc..."
  }
}

Per agent.message, lo start è seguito da eventi event_delta che trasportano testo incrementale. Ogni delta indica l'evento che estende in event_id e il blocco di contenuto che estende in delta.index:

{
  "type": "event_delta",
  "event_id": "sevt_01abc...",
  "delta": {
    "type": "content_delta",
    "index": 0,
    "content": {
      "type": "text",
      "text": "Here is the summary"
    }
  }
}

Per agent.thinking, viene emesso solo l'event_start, come segnale che un blocco di ragionamento è iniziato. Non seguono eventi event_delta. L'evento agent.thinking bufferizzato che conclude l'anteprima è un segnale di avanzamento e non contiene alcun contenuto di ragionamento.

A differenza degli eventi persistiti, event_start e event_delta non hanno un proprio id o processed_at. L'unico identificatore che riportano è l'id dell'evento di cui forniscono l'anteprima. Le loro stringhe di tipo costituiscono inoltre l'eccezione alla convenzione di denominazione {domain}.{action} degli eventi persistiti.

Accumulare e riconciliare

Ogni SDK che supporta i delta degli eventi include un helper di accumulo che gestisce per te la contabilità degli index. Il pattern manuale descritto in questa sezione funziona in ogni linguaggio quando hai bisogno di una gestione personalizzata. Applicalo ai tipi di evento generati.

Nel pattern manuale, conserva il testo di anteprima in una mappa temporanea con chiave (event_id, index) e tratta l'evento bufferizzato come il record. Riconcilia i due per ogni richiesta al modello.

Un turno si apre con un singolo evento session.status_running. In un turno che si completa normalmente, ogni richiesta al modello produce quindi questi eventi, in ordine:

1. span.model_request_start
2. event_start
3. Gli eventi event_delta
4. L'agent.message bufferizzato
5. span.model_request_end (nella scheda Eventi span)

Sulla connessione, questa è la parte in anteprima di quella sequenza, intercalata con gli altri eventi bufferizzati della connessione:

event_start     {"event": {"type": "agent.message", "id": "sevt_01abc..."}}
event_delta     {"event_id": "sevt_01abc...", "delta": {"type": "content_delta", "index": 0, "content": {"type": "text", "text": "..."}}}
...
agent.message   {"id": "sevt_01abc...", "content": [...]}

La riga event_delta si ripete una volta per ogni frammento di testo. Elabora ogni evento man mano che arriva:

1. Su event_start, annota l'id annunciato. Gli identificatori corrispondono sempre: event_start.event.id, ogni event_delta.event_id e l'id dell'agent.message bufferizzato hanno lo stesso valore.
2. Su ogni event_delta, aggiungi delta.content.text alla voce in (event_id, delta.index) e visualizza il testo accumulato. Il primo delta per un index crea quella voce.
3. Quando arriva l'agent.message bufferizzato, associalo tramite id, scarta l'anteprima accumulata e visualizza invece il contenuto del messaggio.
4. Su span.model_request_end, chiudi qualsiasi anteprima che non sia stata riconciliata dal relativo evento bufferizzato. Non arriveranno altri delta per essa. Se il turno va in errore o viene interrotto, l'evento bufferizzato potrebbe non arrivare mai, ma span.model_request_end arriva comunque.

Il pattern si basa su due garanzie:

• Concatenando i delta di un'anteprima nell'ordine di arrivo, con chiave (event_id, index), si ottiene un prefisso di content[index].text nell'evento bufferizzato. Non è necessariamente il testo completo, perché i delta potrebbero essere scartati sotto carico.
• Una connessione emette al massimo un event_start per event_id, e l'evento bufferizzato è l'ultima cosa che quella connessione consegna per quell'id.

Helper di accumulo degli SDK

L'helper di ciascun SDK gestisce la contabilità degli index. Gli helper Go, Java, Ruby e C# indicizzano inoltre l'anteprima in accumulo tramite l'id dell'evento. Con gli helper Python, TypeScript e PHP, gestisci tu stesso quella mappa e integra ogni delta nella voce corrispondente al suo id.

Gli esempi seguenti attivano le anteprime di agent.message e le riconciliano con l'evento bufferizzato:

# Snapshot di anteprima, indicizzati per id evento. accumulate_managed_agents_event accumula ogni
# event_start / event_delta in uno snapshot agent.message; l'evento
# agent.message bufferizzato lo sostituisce.
previews: dict[str, BetaManagedAgentsAgentMessageEvent] = {}

# Attiva le anteprime agent.message su questa connessione
with client.beta.sessions.events.stream(
    session.id, event_deltas=["agent.message"]
) as stream:
    client.beta.sessions.events.send(
        session.id,
        events=[
            {
                "type": "user.message",
                "content": [{"type": "text", "text": "Describe the repo in one sentence."}],
            },
        ],
    )

    for event in stream:
        match event.type:
            case "event_start":
                snapshot = accumulate_managed_agents_event(None, event)
                if snapshot is not None:
                    previews[event.event.id] = snapshot
                print(f"event_start             {event.event.type} {event.event.id}")
            case "event_delta":
                preview = accumulate_managed_agents_event(previews.get(event.event_id), event)
                if preview is not None:
                    previews[event.event_id] = preview
                    text = "".join(block.text for block in preview.content)
                    print(f"event_delta             preview: {text!r}")
            case "agent.message":
                # L'evento bufferizzato è il record: sostituisce e chiude l'anteprima
                preview = accumulate_managed_agents_event(previews.pop(event.id, None), event)
                text = "".join(block.text for block in preview.content)
                print(f"agent.message           {event.id} {text!r}")
            case "span.model_request_end":
                # Non arriveranno altri delta. Chiudi ogni anteprima il cui
                # evento bufferizzato non è mai arrivato.
                for event_id in previews:
                    print(f"span.model_request_end  closing preview for {event_id}")
                previews.clear()
            case "session.status_idle":
                break

Visualizzare in anteprima gli eventi dei thread di sessione

In una sessione multiagente, ogni thread di sessione ha il proprio stream di eventi. Accetta lo stesso parametro event_deltas[] con gli stessi valori.

Una connessione fornisce l'anteprima solo del thread che sta leggendo. Lo stream a livello di sessione fornisce l'anteprima del thread principale, e le anteprime di un thread figlio non vengono mai ripubblicate su di esso. Per osservare il testo di un subagente mentre il modello lo genera, apri lo stream del thread di quel subagente.

Il percorso dello stream di un thread termina con /threads/{thread_id}/stream. /events/stream esiste solo a livello di sessione, quindi non esiste alcun endpoint /threads/{thread_id}/events/stream.

event_start e event_delta hanno la stessa forma su uno stream di thread e sullo stream a livello di sessione, e il pattern accumulare e riconciliare si applica così come descritto. Esegui un'istanza di accumulatore per ogni connessione di stream.

# Elenca i thread della sessione e scegli un figlio: i thread figli hanno un
# parent_thread_id non nullo, mentre il parent_thread_id del thread primario è null.
child_thread = next(
    thread
    for thread in client.beta.sessions.threads.list(session.id)
    if thread.parent_thread_id is not None
)

# Lo stream del thread figlio accetta lo stesso parametro event_deltas dello
# stream della sessione.
with client.beta.sessions.threads.events.stream(
    child_thread.id,
    session_id=session.id,
    event_deltas=["agent.message"],
) as stream:
    for event in stream:
        match event.type:
            case "event_delta":
                print(event.delta.content.text, end="")
            case "agent.message":
                # L'evento bufferizzato è il record autorevole; renderizza il suo contenuto
                print()
                for block in event.content:
                    if block.type == "text":
                        print(block.text, end="")
                print()
            case "session.thread_status_idle":
                break

Il ciclo di lettura termina su session.thread_status_idle, l'evento emesso quando il turno del thread di sessione finisce e il thread diventa inattivo.

Limitazioni

• Best effort: Sotto carico, il server potrebbe scartare i delta di un evento. Quando ciò accade, ricevi un prefisso contiguo del testo e poi nessun ulteriore delta per quell'evento. L'agent.message bufferizzato arriva comunque completo. Non considerare mai definitiva un'anteprima accumulata.
• Nessun replay alla riconnessione: I delta vengono consegnati solo alla connessione che li ha attivati, finché è aperta. Questo vale sia per lo stream a livello di sessione sia per ogni stream di thread di sessione. Una connessione aperta dopo l'inizio di una richiesta al modello non riceve delta per quell'evento in corso. Non c'è modo di richiedere nuovamente i delta persi.
• Un solo thread, solo testo: Le anteprime riguardano il testo dell'assistente sul thread che la connessione sta leggendo. L'uso degli strumenti, i risultati degli strumenti e i risultati MCP non vengono mai visualizzati in anteprima.
• Mai persistiti: event_start e event_delta esistono solo sullo stream in tempo reale. Non compaiono nella cronologia degli eventi della sessione (GET /v1/sessions/{session_id}/events) né nella cronologia degli eventi di alcun thread di sessione.

Risoluzione dei problemi delle anteprime

Passaggi successivi