8 mar 2026

Risposte in Streaming degli Agenti: Output in Tempo Reale per Workflow Multi-Step

Risposte in Streaming degli Agenti: Output in Tempo Reale per Workflow Multi-Step

Il tuo agente impiega 20 secondi per ricercare una domanda, chiamare tre strumenti e sintetizzare una risposta. L’utente clicca “Chiedi” e fissa uno spinner per 20 secondi. Non è sicuro che stia funzionando. Pensa di ricaricare la pagina. Si chiede se ricominciare da capo.

Ora immagina questo: l’utente clicca “Chiedi” e entro 500 millisecondi vede apparire le prime parole dell’agente. Lo guarda cercare informazioni—”🔍 Ricerca nel database ordini…”—e vede i risultati arrivare in tempo reale. Legge la risposta mentre viene scritta, frase per frase. Stessi 20 secondi. Esperienza completamente diversa.

Lo streaming non è un optional per gli agenti rivolti agli utenti—è un requisito di UX. Il time-to-first-token è la singola metrica di latenza più importante nelle interfacce agente. Gli utenti che vedono i progressi sono pazienti. Gli utenti che non vedono nulla abbandonano. Studio dopo studio sulle performance web dimostra che la latenza percepita conta più di quella reale, e lo streaming è lo strumento più potente che hai per colmare il divario tra le due.

In questo articolo imparerai come implementare lo streaming in tempo reale per agenti AI multi-step—dalla consegna token per token alla trasparenza delle tool call, agli aggiornamenti di stato, alla progressive disclosure, al recupero degli errori e alle scelte del livello di trasporto.

Sezione 1: Basi dell’API di Streaming di Claude

Prima di poter fare lo streaming di un agente, devi fare lo streaming di una singola risposta di Claude. Partiamo dai fondamentali.

Stream vs. Non-Stream

Una chiamata API non-streaming si blocca finché l’intera risposta non è generata, poi la restituisce tutta in una volta. Una chiamata in streaming restituisce una sequenza di eventi man mano che la risposta viene prodotta, a partire dal primo token.

La differenza nel codice è minima. La differenza nell’esperienza utente è enorme.

Tipi di Evento

L’API di streaming di Claude emette una sequenza strutturata di server-sent events:

message_start — Contiene l’oggetto Message iniziale con i metadati (model, role, usage).
content_block_start — Segnala l’inizio di un blocco di contenuto (text o tool_use).
content_block_delta — Contiene contenuto incrementale: frammenti di testo o JSON parziale dell’input dello strumento.
content_block_stop — Segnala la fine di un blocco di contenuto.
message_delta — Aggiornamenti finali al messaggio (stop reason, usage finale).
message_stop — Lo stream è completo.

Per le risposte testuali, riceverai molti eventi content_block_delta, ciascuno con un piccolo frammento di testo (spesso pochi token).

Implementazione Base dello Streaming

Ecco un esempio di streaming sincrono usando l’SDK Python di Anthropic:

import anthropic

client = anthropic.Anthropic()

def stream_basic_response(user_message: str):
    """Stream a basic Claude response token by token."""
    with client.messages.stream(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[{"role": "user", "content": user_message}],
    ) as stream:
        for text in stream.text_stream:
            print(text, end="", flush=True)
    print()  # Newline after stream completes

stream_basic_response("Explain quantum entanglement in simple terms.")

E la versione asincrona, che userai nei web server in produzione:

import anthropic
import asyncio

async_client = anthropic.AsyncAnthropic()

async def stream_basic_response_async(user_message: str):
    """Async streaming with the Anthropic SDK."""
    async with async_client.messages.stream(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        messages=[{"role": "user", "content": user_message}],
    ) as stream:
        async for text in stream.text_stream:
            print(text, end="", flush=True)
    print()

asyncio.run(stream_basic_response_async("Explain quantum entanglement in simple terms."))

Gestione dei Token Parziali e dei Confini UTF-8

L’SDK gestisce la decodifica UTF-8 per te, ma se lavori con lo stream HTTP grezzo, tieni presente che i caratteri multi-byte possono essere suddivisi tra i chunk. Esegui sempre il buffer dei byte grezzi e decodifica solo quando hai sequenze UTF-8 complete. L’iteratore text_stream dell’SDK anthropic gestisce questo automaticamente—un altro motivo per usarlo invece di analizzare direttamente lo stream SSE grezzo.

Sezione 2: Streaming delle Tool Call

Lo streaming di testo di base è il minimo indispensabile. La vera sfida—e il vero valore—emerge quando il tuo agente usa gli strumenti. Un agente multi-step che chiama tre strumenti in sequenza può sembrare bloccato durante l’esecuzione degli strumenti, a meno che tu non mostri cosa sta accadendo.

Come Appaiono le Tool Call nello Stream

Quando Claude decide di usare uno strumento, lo stream emette un evento content_block_start con type: "tool_use", seguito da eventi content_block_delta contenenti frammenti del JSON di input dello strumento. Una volta assemblata la tool call completa, esegui lo strumento, inietti il risultato e continui la conversazione.

L’intuizione chiave: conosci il nome dello strumento non appena arriva content_block_start, anche prima che il JSON di input sia completo. Questo significa che puoi immediatamente mostrare all’utente qualcosa come ”🔍 Ricerca nel database ordini…” senza aspettare la tool call completa.

Il Loop Completo dell’Agente in Streaming

Ecco un loop completo di agente in streaming che mostra le invocazioni degli strumenti in tempo reale:

import anthropic
import json

client = anthropic.Anthropic()

# Define tools
tools = [
    {
        "name": "search_orders",
        "description": "Search customer orders by order ID or customer email.",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Order ID or email"},
            },
            "required": ["query"],
        },
    },
    {
        "name": "get_shipping_status",
        "description": "Get real-time shipping status for an order.",
        "input_schema": {
            "type": "object",
            "properties": {
                "order_id": {"type": "string", "description": "The order ID"},
            },
            "required": ["order_id"],
        },
    },
]


def execute_tool(tool_name: str, tool_input: dict) -> str:
    """Execute a tool and return the result as a string."""
    if tool_name == "search_orders":
        # Simulated database lookup
        return json.dumps({
            "order_id": "ORD-12345",
            "customer": "jane@example.com",
            "items": ["Blue Widget x2", "Red Gadget x1"],
            "total": "$47.99",
            "status": "shipped",
        })
    elif tool_name == "get_shipping_status":
        return json.dumps({
            "order_id": tool_input["order_id"],
            "carrier": "FedEx",
            "tracking": "7891011",
            "estimated_delivery": "2026-03-10",
            "current_location": "Memphis, TN",
        })
    return json.dumps({"error": f"Unknown tool: {tool_name}"})


def stream_agent_response(user_message: str):
    """
    Complete streaming agent loop with real-time tool call display.
    """
    messages = [{"role": "user", "content": user_message}]

    while True:
        # Stream the model response
        tool_calls = []
        current_tool = None

        with client.messages.stream(
            model="claude-sonnet-4-20250514",
            max_tokens=4096,
            tools=tools,
            messages=messages,
        ) as stream:
            response = None
            for event in stream:
                # The SDK exposes raw events via iteration
                pass

            # Use the helper to collect text and tool use
            response = stream.get_final_message()

        # Process the response content blocks
        for block in response.content:
            if block.type == "text":
                print(block.text, end="", flush=True)
            elif block.type == "tool_use":
                tool_name = block.name
                tool_input = block.input
                tool_id = block.id

                # Show the user what's happening
                print(f"\n⚙️  Calling tool: {tool_name}({json.dumps(tool_input)})")

                # Execute the tool
                result = execute_tool(tool_name, tool_input)
                print(f"✅  Result received from {tool_name}")

                tool_calls.append({
                    "tool_use_id": tool_id,
                    "tool_name": tool_name,
                    "tool_input": tool_input,
                    "result": result,
                })

        # If the model stopped because it wants to use tools, continue the loop
        if response.stop_reason == "tool_use":
            # Add assistant message with all content blocks
            messages.append({"role": "assistant", "content": response.content})

            # Add tool results
            tool_results = []
            for tc in tool_calls:
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": tc["tool_use_id"],
                    "content": tc["result"],
                })
            messages.append({"role": "user", "content": tool_results})

            print("\n--- Continuing agent loop ---")
        else:
            # Model is done (stop_reason == "end_turn")
            print()
            break


# Run the agent
stream_agent_response("Where is my order #ORD-12345? When will it arrive?")

Per uno streaming token per token con rilevamento delle tool call, puoi iterare sugli eventi grezzi:

def stream_agent_with_live_tokens(user_message: str):
    """Stream text tokens live while also detecting tool calls."""
    messages = [{"role": "user", "content": user_message}]

    while True:
        collected_content = []
        current_text = ""
        current_tool_name = None
        current_tool_input_json = ""
        current_tool_id = None
        stop_reason = None

        with client.messages.stream(
            model="claude-sonnet-4-20250514",
            max_tokens=4096,
            tools=tools,
            messages=messages,
        ) as stream:
            for event in stream:
                if hasattr(event, 'type'):
                    if event.type == 'content_block_start':
                        if event.content_block.type == 'tool_use':
                            current_tool_name = event.content_block.name
                            current_tool_id = event.content_block.id
                            current_tool_input_json = ""
                            print(f"\n🔧 Agent is calling: {current_tool_name}...")

---

## Articoli Correlati

- [Pattern di Utilizzo degli Strumenti: Interfacce Agente-Strumento Affidabili](/it/blog/agent-tool-use-patterns/)
- [Pattern Multi-Agente: Orchestratori, Worker e Pipeline](/it/blog/multi-agent-patterns/)
- [Recupero Errori degli Agenti: 5 Pattern per l'Affidabilità in Produzione](/it/blog/agent-error-recovery-patterns/)
- [Agenti di Automazione Web: Controllo del Browser con Claude e Computer Use](/it/blog/web-automation-agents-browser-control-with-claude-and-computer-use/)