AI Agents · Lezione

Output in streaming negli agenti CLI

Stampa dei token trasmessi in streaming, carattere per carattere, nelle interfacce da terminale

Output in streaming negli agenti CLI è una lezione AI Agents gratuita su CoddyKit. Questa è la lezione 4 di 4. Puoi leggere la lezione completa qui gratuitamente — poi esercitati direttamente nel browser con un editor di codice integrato e un tutor IA disponibile 24/7. Fa parte del percorso di apprendimento AI Agents, e i tuoi progressi si sincronizzano tra il web e l'app CoddyKit. Il corso AI Agents include 4 lezioni in totale.

Perché lo streaming è importante per gli agenti CLI

Senza streaming, l'agente CLI non stampa nulla finché la risposta completa del LLM non è pronta: possono trascorrere 5-30 secondi. Gli utenti fissano un terminale vuoto chiedendosi se il programma si sia bloccato.

Con lo streaming, i token vengono visualizzati man mano che vengono generati, fornendo un feedback immediato e un'esperienza molto migliore.

Abilitazione dello streaming nell'SDK OpenAI

Passi stream=True a chat.completions.create(). La chiamata restituisce un generatore anziché un oggetto di risposta completo. Lo iteri per elaborare i blocchi man mano che arrivano.

import openai

client = openai.OpenAI(api_key='YOUR_API_KEY')

stream = client.chat.completions.create(
    model='gpt-4o-mini',
    messages=[{'role': 'user', 'content': 'Explain Python generators in 3 sentences.'}],
    stream=True  # <-- enable streaming
)

# Each chunk arrives as it is generated
for chunk in stream:
    delta = chunk.choices[0].delta
    if delta.content:
        print(delta.content, end='', flush=True)

print()  # newline after the response is complete

print() e sys.stdout.write()

Durante lo streaming, usi print(text, end='', flush=True) oppure sys.stdout.write(text) seguito da sys.stdout.flush(). Senza flush=True, Python potrebbe memorizzare l'output nel buffer e stamparlo tutto insieme, vanificando lo scopo dello streaming.

import sys
import openai

client = openai.OpenAI(api_key='YOUR_API_KEY')

def stream_to_terminal(messages: list):
    stream = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=messages,
        stream=True
    )

    full_response = ''
    for chunk in stream:
        token = chunk.choices[0].delta.content or ''
        full_response += token

        # Option 1: print with flush
        print(token, end='', flush=True)

        # Option 2: sys.stdout.write + flush
        # sys.stdout.write(token)
        # sys.stdout.flush()

    print()  # final newline
    return full_response

Raccolta della risposta completa durante lo streaming

Spesso è necessario disporre del testo completo della risposta al termine dello streaming, per archiviarlo, elaborarlo ulteriormente o visualizzarlo. Accumuli i token in una stringa mentre li stampa.

import openai

client = openai.OpenAI(api_key='YOUR_API_KEY')

def stream_and_collect(messages: list) -> str:
    full_text = ''

    print('Agent: ', end='', flush=True)

    stream = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=messages,
        stream=True
    )

    for chunk in stream:
        token = chunk.choices[0].delta.content or ''
        full_text += token
        print(token, end='', flush=True)

    print()  # newline
    return full_text

# The return value contains the complete response for storage
# response_text = stream_and_collect(history)
# history.append({'role': 'assistant', 'content': response_text})

Streaming asincrono con AsyncOpenAI

Per le architetture di agenti asincrone, usi AsyncOpenAI e async for per iterare sui blocchi dello streaming senza bloccare il ciclo degli eventi.

import asyncio
import openai

async def async_stream_agent(query: str) -> str:
    client = openai.AsyncOpenAI(api_key='YOUR_API_KEY')

    stream = await client.chat.completions.create(
        model='gpt-4o-mini',
        messages=[{'role': 'user', 'content': query}],
        stream=True
    )

    full_text = ''
    print('Agent: ', end='', flush=True)

    async for chunk in stream:
        token = chunk.choices[0].delta.content or ''
        full_text += token
        print(token, end='', flush=True)

    print()
    return full_text

# asyncio.run(async_stream_agent('What is asyncio?'))

Rilevamento della fine dello stream con finish_reason

L'ultimo blocco di uno stream ha un valore finish_reason diverso da null. Lo controlli per sapere perché lo stream è terminato: 'stop' = completamento normale, 'length' = risposta troncata, 'tool_calls' = è necessaria una chiamata di funzione.

import openai

client = openai.OpenAI(api_key='YOUR_API_KEY')

def stream_with_finish_detection(messages: list):
    stream = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=messages,
        stream=True
    )

    finish_reason = None
    for chunk in stream:
        delta = chunk.choices[0].delta
        if delta.content:
            print(delta.content, end='', flush=True)
        if chunk.choices[0].finish_reason:
            finish_reason = chunk.choices[0].finish_reason

    print()

    if finish_reason == 'length':
        print('[WARNING: Response was truncated. Try increasing max_tokens.]')
    elif finish_reason == 'stop':
        pass  # normal completion

    return finish_reason

Colori ANSI nell'output del terminale

I codici di escape ANSI aggiungono colore all'output del terminale. Li usi per distinguere visivamente il prefisso dell'agente, il prompt dell'input dell'utente e gli avvisi. La libreria colorama offre supporto multipiattaforma, incluso Windows.

# pip install colorama
from colorama import Fore, Style, init
init(autoreset=True)  # reset color after each print

def print_colored_stream(messages: list, client):
    # Print agent prefix in cyan
    print(Fore.CYAN + 'Agent: ' + Style.RESET_ALL, end='', flush=True)

    stream = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=messages,
        stream=True
    )

    for chunk in stream:
        token = chunk.choices[0].delta.content or ''
        print(token, end='', flush=True)

    print()

# Also useful:
# print(Fore.GREEN + 'Success!') — green
# print(Fore.RED + 'Error!') — red
# print(Fore.YELLOW + 'Warning') — yellow

La libreria Rich per un output del terminale più ricco

La libreria rich offre il rendering Markdown, blocchi di codice con evidenziazione della sintassi, tabelle e spinner nel terminale. Si abbina bene all'output dell'agente trasmesso in streaming.

# pip install rich
from rich.console import Console
from rich.live import Live
from rich.markdown import Markdown

console = Console()

def stream_with_rich(messages: list, client):
    full_text = ''

    with Live(console=console, refresh_per_second=10) as live:
        stream = client.chat.completions.create(
            model='gpt-4o-mini',
            messages=messages,
            stream=True
        )

        for chunk in stream:
            token = chunk.choices[0].delta.content or ''
            full_text += token
            # Render accumulated text as Markdown in real time
            live.update(Markdown(full_text))

    return full_text

Streaming con chiamate agli strumenti

Quando lo streaming viene combinato con le chiamate di funzione, anche il campo tool_calls viene trasmesso in parti. Accumuli la stringa JSON nei vari blocchi prima di analizzarla.

import json
import openai

client = openai.OpenAI(api_key='YOUR_API_KEY')

def stream_with_tools(messages: list, tools: list) -> dict:
    stream = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=messages,
        tools=tools,
        stream=True
    )

    tool_call_chunks = {}
    for chunk in stream:
        delta = chunk.choices[0].delta
        if delta.tool_calls:
            for tc in delta.tool_calls:
                idx = tc.index
                if idx not in tool_call_chunks:
                    tool_call_chunks[idx] = {'name': '', 'args': ''}
                if tc.function.name:
                    tool_call_chunks[idx]['name'] += tc.function.name
                if tc.function.arguments:
                    tool_call_chunks[idx]['args'] += tc.function.arguments

    # Parse accumulated tool calls
    return {v['name']: json.loads(v['args']) for v in tool_call_chunks.values()}

Visualizzazione del contatore dei token

Mostri un contatore dei token aggiornato in tempo reale durante lo streaming, per aiutare gli utenti a monitorare l'utilizzo e comprendere i costi. Lo stream OpenAI include i dati sull'utilizzo nell'ultimo blocco quando è impostato stream_options={'include_usage': True}.

import openai

client = openai.OpenAI(api_key='YOUR_API_KEY')

def stream_with_token_count(messages: list):
    stream = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=messages,
        stream=True,
        stream_options={'include_usage': True}
    )

    for chunk in stream:
        delta = chunk.choices[0].delta
        if delta.content:
            print(delta.content, end='', flush=True)

        # Last chunk includes usage
        if chunk.usage:
            print(f'\n[Tokens: prompt={chunk.usage.prompt_tokens}, '
                  f'completion={chunk.usage.completion_tokens}, '
                  f'total={chunk.usage.total_tokens}]')

Buone pratiche per lo streaming

Riepilogo delle buone pratiche per l'output in streaming negli agenti CLI:

Usi sempre flush=True o sys.stdout.flush() per evitare il buffering
Accumuli i token in una stringa per archiviarli dopo lo streaming
Controlli finish_reason per rilevare il troncamento
Usi i colori ANSI o rich per una maggiore chiarezza visiva
Gestisca lo streaming delle chiamate agli strumenti accumulando i blocchi degli argomenti JSON

Verifica delle conoscenze: output in streaming

Verifichi la Sua comprensione dell'output in streaming negli agenti CLI.

Riepilogo: output in streaming negli agenti CLI

Ora è in grado di creare agenti CLI con output in streaming, reattivi e moderni:

Passi stream=True per abilitare lo streaming nell'SDK OpenAI
Utilizzi print(token, end='', flush=True) per visualizzare immediatamente i token
Accumuli i token in una stringa per elaborarli dopo lo streaming
Controlli finish_reason nell'ultimo chunk per rilevare un troncamento
Utilizzi async for con AsyncOpenAI per gli agenti asincroni
Aggiunga colori ANSI o rich per un'esperienza del terminale più curata

Domande Frequenti

La lezione «Output in streaming negli agenti CLI» è gratuita?

Sì — il testo completo di «Output in streaming negli agenti CLI» è gratuito qui sul web. Per esercitarvi in modo interattivo (un editor di codice integrato e un tutor IA 24/7) e sbloccare il resto del corso AI Agents, passa a CoddyKit PRO. Il corso AI Agents include 4 lezioni in totale.

Cosa imparerò in «Output in streaming negli agenti CLI»?

Stampa dei token trasmessi in streaming, carattere per carattere, nelle interfacce da terminale Eserciti AI Agents con codice pratico che esegui direttamente nel browser, e un tutor IA 24/7 risponde alle tue domande mentre lavori sulla lezione.

Ho bisogno di esperienza per iniziare AI Agents?

Non è richiesta alcuna esperienza precedente. AI Agents su CoddyKit è strutturato per principianti e studenti avanzati, quindi puoi iniziare da qui o dall'inizio e procedere al tuo ritmo. Questa è la lezione 4 di 4.

Quanto tempo richiede la lezione «Output in streaming negli agenti CLI»?

La maggior parte delle lezioni CoddyKit richiede circa 5–10 minuti. Ogni lezione è breve e interattiva, quindi fai progressi costanti e riprendi esattamente da dove hai lasciato su web e app.

Posso scrivere ed eseguire codice in questa lezione AI Agents?

Sì. Ogni lezione AI Agents include un editor di codice integrato, quindi scrivi ed esegui codice reale direttamente nel tuo browser e ricevi feedback istantaneo dall'IA — nessuna configurazione locale necessaria.

Tutte le lezioni di questo corso

Creazione di interfacce a riga di comando per agenti
Agenti interattivi in stile REPL
Parsing degli argomenti e testo di aiuto
Output in streaming negli agenti CLI

← Torna a AI Agents