0Pricing
AI Agents · Aula

Transmitindo a saída em agentes de CLI

Impressão de tokens transmitidos, caractere por caractere, em interfaces de terminal.

Transmitindo a saída em agentes de CLI é uma aula grátis de AI Agents no CoddyKit. Esta é a aula 4 de 4. Você pode ler a aula completa abaixo gratuitamente — depois pratica ao vivo no navegador com um editor de código integrado e um tutor de IA 24/7. Faz parte do caminho de aprendizado de AI Agents, e seu progresso é sincronizado entre a web e o app CoddyKit. O curso de AI Agents inclui 4 aulas no total.

Por que a transmissão é importante para agentes de CLI

Sem transmissão, seu agente de CLI não exibe nada até que a resposta completa do LLM esteja pronta — isso pode levar de 5 a 30 segundos. Os usuários ficam olhando para um terminal vazio, sem saber se o programa travou.

Com a transmissão, os tokens aparecem à medida que são gerados, proporcionando retorno imediato e uma experiência muito melhor.

Ativando a transmissão no SDK OpenAI

Passe stream=True para chat.completions.create(). A chamada retorna um gerador em vez de um objeto de resposta completo. Itere sobre ele para processar os fragmentos à medida que chegam.

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

Ao transmitir, use print(text, end='', flush=True) ou sys.stdout.write(text) seguido de sys.stdout.flush(). Sem flush=True, o Python pode armazenar a saída temporariamente e imprimi-la toda de uma vez, anulando o objetivo da transmissão.

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

Coletando a resposta completa durante a transmissão

Muitas vezes, você precisa do texto completo da resposta depois que a transmissão termina — para armazená-lo, processá-lo posteriormente ou exibi-lo. Acumule os tokens em uma cadeia de caracteres enquanto os 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})

Transmissão assíncrona com AsyncOpenAI

Para arquiteturas de agentes assíncronas, use AsyncOpenAI e async for para iterar pelos fragmentos transmitidos sem bloquear o ciclo 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?'))

Detectando o fim do fluxo com finish_reason

O último fragmento de um fluxo tem um finish_reason diferente de nulo. Verifique-o para saber por que o fluxo terminou: 'stop' = conclusão normal, 'length' = truncado, 'tool_calls' = chamada de função necessária.

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

Cores ANSI na saída do terminal

Códigos de escape ANSI adicionam cor à saída do terminal. Use-os para distinguir visualmente o prefixo do agente, a solicitação de entrada do usuário e os avisos. A biblioteca colorama oferece suporte multiplataforma, inclusive no 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 uma saída de terminal mais elaborada

A biblioteca rich oferece renderização de Markdown, blocos de código com realce de sintaxe, tabelas e indicadores giratórios no terminal. Ela combina bem com a saída transmitida do 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

Transmissão com chamadas de ferramentas

Quando a transmissão é combinada com chamadas de funções, o campo tool_calls também é transmitido em partes. Acumule a cadeia JSON entre os fragmentos antes de analisá-la.

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

Exibição do contador de tokens

Mostre um contador de tokens em tempo real durante a transmissão para ajudar os usuários a monitorar o uso e entender os custos. O fluxo da OpenAI inclui dados de uso no fragmento final quando stream_options={'include_usage': True} é definido.

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}]')

Boas práticas de transmissão

Resumo das boas práticas de saída transmitida para agentes de CLI:

  • Sempre use flush=True ou sys.stdout.flush() para evitar o armazenamento temporário em buffer
  • Acumule os tokens em uma cadeia de caracteres para armazená-los após a transmissão
  • Verifique finish_reason para detectar truncamentos
  • Use cores ANSI ou rich para melhorar a clareza visual
  • Ao transmitir chamadas de ferramentas, acumule os fragmentos de argumentos JSON

Verificação de conhecimento: saída transmitida

Verifique sua compreensão sobre a saída transmitida em agentes de CLI.

Recapitulação: Saída em fluxo em agentes CLI

Agora você pode criar agentes CLI com saída em fluxo, que parecem responsivos e modernos:

  • Passe stream=True para habilitar a transmissão do SDK OpenAI
  • Use print(token, end='', flush=True) para exibir os tokens imediatamente
  • Acumule os tokens em uma string para processamento após a transmissão
  • Verifique finish_reason no último bloco para detectar truncamento
  • Use async for com AsyncOpenAI para agentes assíncronos
  • Adicione cores ANSI ou rich para uma experiência de terminal refinada

Perguntas Frequentes

A aula “Transmitindo a saída em agentes de CLI” é grátis?

Sim — o texto completo de “Transmitindo a saída em agentes de CLI” é grátis para ler aqui na web. Para praticá-la interativamente (um editor de código integrado e um tutor de IA 24/7) e desbloquear o restante do curso de AI Agents, atualize para CoddyKit PRO. O curso de AI Agents inclui 4 aulas no total.

O que vou aprender em “Transmitindo a saída em agentes de CLI”?

Impressão de tokens transmitidos, caractere por caractere, em interfaces de terminal. Você pratica AI Agents com código prático que executa diretamente no navegador, e um tutor de IA 24/7 responde suas dúvidas enquanto trabalha na aula.

Preciso ter experiência prévia para começar AI Agents?

Nenhuma experiência prévia é necessária. AI Agents no CoddyKit é estruturado para alunos iniciantes até avançados, então você pode começar aqui ou desde o início e aprender no seu ritmo. Esta é a aula 4 de 4.

Quanto tempo leva a aula “Transmitindo a saída em agentes de CLI”?

A maioria das aulas CoddyKit leva cerca de 5–10 minutos. Cada uma é compacta e interativa, então você faz progresso constante e retoma exatamente de onde parou entre web e app.

Posso escrever e executar código nesta aula de AI Agents?

Sim. Cada aula de AI Agents inclui um editor de código integrado, então você escreve e executa código real direto no navegador e recebe feedback de IA instantaneamente — nenhuma configuração local necessária.

Todas as aulas deste curso

  1. Criando interfaces de agentes de linha de comando
  2. Agentes interativos no estilo REPL
  3. Análise de argumentos e texto de ajuda
  4. Transmitindo a saída em agentes de CLI
← Voltar para AI Agents