Visualizzare in anteprima le risposte con i delta degli eventi

Per impostazione predefinita, il testo della risposta dell'agente raggiunge il flusso di 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 un flusso completo e corretto.

Attivare le anteprime

Le anteprime sono facoltative e si attivano per singola connessione di flusso. Aggiungi il parametro di query event_deltas[] al flusso che stai leggendo e ripetilo una volta per ogni tipo di evento di cui desideri 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 flusso accettano il parametro:

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

Le anteprime di un subagente compaiono sul flusso 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, il flusso emette un event_start che riporta il tipo e l'id dell'evento in arrivo:

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

Per agent.message, l'inizio è 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 trasportano è 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à dell'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 dell'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 Span events)

Sul canale di trasmissione, 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 parziale. 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 genera un 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à dell'index. Gli helper Go, Java, Ruby e C# usano inoltre come chiave dell'anteprima in accumulo l'id dell'evento. Con gli helper Python, TypeScript e PHP, mantieni 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 integra ogni
# event_start / event_delta in uno snapshot agent.message; l'agent.message
# nel buffer lo sostituisce.
previews: dict[str, BetaManagedAgentsAgentMessageEvent] = {}

# Abilita le anteprime di 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 nel buffer è 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 nel buffer 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 flusso di eventi. Accetta lo stesso parametro event_deltas[] con gli stessi valori.

Una connessione fornisce l'anteprima solo del thread che sta leggendo. Il flusso 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 il flusso del thread di quel subagente.

Il percorso del flusso 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 sul flusso di un thread e sul flusso a livello di sessione, e il pattern accumulare e riconciliare si applica così come descritto. Esegui un'istanza di accumulatore per ogni connessione di flusso.

# Elenca i thread della sessione e scegli un figlio: i thread figli hanno parent_thread_id
# non null, 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 nel buffer è il record autorevole; renderizzane il 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 termina 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 altro delta per quell'evento. L'agent.message bufferizzato arriva comunque completo. Non considerare mai definitiva un'anteprima accumulata.
• Nessuna riproduzione alla riconnessione: I delta vengono consegnati solo alla connessione che li ha attivati, finché è aperta. Questo vale sia per il flusso a livello di sessione sia per ogni flusso di thread di sessione. Una connessione aperta dopo l'inizio di una richiesta al modello non riceve alcun delta per quell'evento in corso. Non c'è modo di richiedere nuovamente i delta persi.
• Un solo thread, solo testo: Le anteprime coprono 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 sul flusso 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