0Pricing
AI Agents · Lección

Salida en streaming en agentes CLI

Imprima los tokens transmitidos carácter a carácter en interfaces de terminal.

Salida en streaming en agentes CLI es una lección gratuita de AI Agents en CoddyKit. Esta es la lección 4 de 4. Puedes leer la lección completa abajo gratuitamente — luego la practicas en el navegador con un editor de código integrado y un tutor de IA 24/7. Forma parte de la ruta de aprendizaje de AI Agents, y tu progreso se sincroniza en la web y la app de CoddyKit. El curso de AI Agents incluye 4 lecciones en total.

Por qué importa el streaming para los agentes de CLI

Sin streaming, su agente de CLI no muestra nada hasta que la respuesta completa del LLM está lista; esto puede tardar entre 5 y 30 segundos. Los usuarios se quedan mirando un terminal vacío, preguntándose si el programa se ha bloqueado.

Con streaming, los tokens aparecen a medida que se generan, lo que proporciona información inmediata y una experiencia mucho mejor.

Cómo habilitar el streaming en el SDK de OpenAI

Pase stream=True a chat.completions.create(). La llamada devuelve un generador en lugar de un objeto de respuesta completo. Itere sobre él para procesar los fragmentos a medida que llegan.

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() frente a sys.stdout.write()

Al transmitir datos, use print(text, end='', flush=True) o sys.stdout.write(text), seguido de sys.stdout.flush(). Sin flush=True, Python puede almacenar la salida en un búfer e imprimirla toda de una vez, lo que anula el propósito del 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

Cómo recopilar la respuesta completa durante el streaming

A menudo necesitará el texto completo de la respuesta una vez finalizado el streaming, ya sea para almacenarlo, procesarlo posteriormente o mostrarlo. Acumule los tokens en una cadena a medida que los imprime.

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 asíncrono con AsyncOpenAI

Para arquitecturas de agentes asíncronas, use AsyncOpenAI y async for para iterar sobre los fragmentos transmitidos sin bloquear el bucle de eventos.

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?'))

Detección del final del stream con finish_reason

El último fragmento de un stream tiene un valor finish_reason distinto de null. Compruébelo para saber por qué terminó el stream: 'stop' = finalización normal, 'length' = truncado, 'tool_calls' = se necesita una llamada a una función.

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

Colores ANSI en la salida de la terminal

Los códigos de escape ANSI añaden color a la salida de la terminal. Úselos para distinguir visualmente el prefijo del agente, el indicador de entrada del usuario y las advertencias. La biblioteca colorama ofrece compatibilidad multiplataforma, incluido 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

Biblioteca rich para una salida de terminal más completa

La biblioteca rich proporciona renderizado de Markdown, bloques de código con resaltado de sintaxis, tablas e indicadores giratorios en la terminal. Se combina bien con la salida transmitida del agente.

# 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 llamadas a herramientas

Cuando el streaming se combina con llamadas a funciones, el campo tool_calls también se transmite por partes. Acumule la cadena JSON a lo largo de los fragmentos antes de analizarla.

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()}

Visualización del contador de tokens

Muestre un contador de tokens en tiempo real durante el streaming para ayudar a los usuarios a supervisar el uso y comprender los costes. El stream de OpenAI incluye datos de uso en el fragmento final cuando se establece 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}]')

Buenas prácticas de streaming

Resumen de las buenas prácticas para la salida mediante streaming en agentes de CLI:

  • Use siempre flush=True o sys.stdout.flush() para evitar el almacenamiento en búfer
  • Acumule los tokens en una cadena para almacenarlos después del streaming
  • Compruebe finish_reason para detectar truncamientos
  • Use colores ANSI o rich para mejorar la claridad visual
  • Gestione el streaming de llamadas a herramientas acumulando los fragmentos de argumentos JSON

Comprobación de conocimientos: salida mediante streaming

Compruebe sus conocimientos sobre la salida mediante streaming en agentes de CLI.

Resumen: Salida en streaming en agentes CLI

Ahora puede crear agentes CLI con salida en streaming que ofrecen una experiencia ágil y moderna:

  • Pase stream=True para habilitar el streaming desde el SDK de OpenAI
  • Use print(token, end='', flush=True) para mostrar los tokens de inmediato
  • Acumule los tokens en una cadena para procesarlos después del streaming
  • Compruebe finish_reason en el último fragmento para detectar truncamientos
  • Use async for con AsyncOpenAI para agentes asíncronos
  • Añada colores ANSI o rich para mejorar la experiencia en el terminal

Preguntas frecuentes

¿La lección «Salida en streaming en agentes CLI» es gratis?

Sí — el texto completo de «Salida en streaming en agentes CLI» es gratis para leer aquí en la web. Para practicarla de forma interactiva (editor de código integrado y tutor de IA 24/7) y desbloquear el resto del curso de AI Agents, actualiza a CoddyKit PRO. El curso de AI Agents incluye 4 lecciones en total.

¿Qué aprenderé en «Salida en streaming en agentes CLI»?

Imprima los tokens transmitidos carácter a carácter en interfaces de terminal. Practicas AI Agents con código real que ejecutas directamente en el navegador, y un tutor de IA 24/7 responde tus preguntas mientras trabajas en la lección.

¿Necesito experiencia previa para empezar AI Agents?

No se requiere experiencia previa. AI Agents en CoddyKit está estructurado para principiantes hasta estudiantes avanzados, así que puedes empezar aquí o desde el inicio y avanzar a tu ritmo. Esta es la lección 4 de 4.

¿Cuánto tiempo toma la lección «Salida en streaming en agentes CLI»?

La mayoría de las lecciones de CoddyKit toman alrededor de 5–10 minutos. Cada una es compacta e interactiva, así que avanzas constantemente y retomas exactamente por donde dejaste en la web y la app.

¿Puedo escribir y ejecutar código en esta lección de AI Agents?

Sí. Cada lección de AI Agents incluye un editor de código integrado, así que escribes y ejecutas código real directamente en tu navegador y obtienes retroalimentación instantánea de IA — sin configuración local necesaria.

Todas las lecciones de este curso

  1. Creación de interfaces de agentes de línea de comandos
  2. Agentes interactivos al estilo REPL
  3. Análisis de argumentos y texto de ayuda
  4. Salida en streaming en agentes CLI
← Volver a AI Agents