PricingLoginStart free
AI Agents · Lektion

Streaming-Ausgabe in CLI-Agenten

Gestreamte Tokens in Terminal-Oberflächen Zeichen für Zeichen ausgeben.

Lektion 4 von 413 Schritte

Streaming-Ausgabe in CLI-Agenten ist eine kostenlose AI Agents-Lektion auf CoddyKit. Dies ist Lektion 4 von 4. Du kannst die komplette Lektion unten kostenlos lesen – dann übst du sie direkt im Browser mit einem integrierten Code-Editor und einem KI-Tutor rund um die Uhr. Sie ist Teil des AI Agents-Lernpfads, und dein Fortschritt wird über Web und CoddyKit-App synchronisiert. Der AI Agents-Kurs umfasst insgesamt 4 Lektionen.

Warum Streaming für CLI-Agenten wichtig ist

Ohne Streaming gibt Ihr CLI-Agent nichts aus, bis die vollständige LLM-Antwort bereit ist – das kann 5–30 Sekunden dauern. Benutzer starren auf ein leeres Terminal und fragen sich, ob das Programm abgestürzt ist.

Beim Streaming erscheinen Tokens, sobald sie generiert werden. Das liefert sofortiges Feedback und sorgt für eine deutlich bessere Benutzererfahrung.

Streaming im OpenAI SDK aktivieren

Übergeben Sie stream=True an chat.completions.create(). Der Aufruf gibt einen Generator statt eines vollständigen Antwortobjekts zurück. Iterieren Sie darüber, um die eintreffenden Chunks zu verarbeiten.

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

Verwenden Sie beim Streaming print(text, end='', flush=True) oder sys.stdout.write(text), gefolgt von sys.stdout.flush(). Ohne flush=True puffert Python die Ausgabe möglicherweise und gibt alles auf einmal aus – damit wird der Zweck des Streamings verfehlt.

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

Die vollständige Antwort während des Streamings sammeln

Oft benötigen Sie nach Abschluss des Streamings den vollständigen Antworttext – zum Speichern, zur weiteren Verarbeitung oder zur Anzeige. Sammeln Sie die Tokens in einer Zeichenkette, während Sie sie ausgeben.

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

Asynchrones Streaming mit AsyncOpenAI

Verwenden Sie für asynchrone Agent-Architekturen AsyncOpenAI und async for, um über Streaming-Chunks zu iterieren, ohne die Ereignisschleife zu blockieren.

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

Das Ende des Streams mit finish_reason erkennen

Der letzte Chunk eines Streams enthält einen Wert für finish_reason, der nicht null ist. Prüfen Sie ihn, um den Grund für das Streamende zu ermitteln: 'stop' = normaler Abschluss, 'length' = abgeschnitten, 'tool_calls' = Funktionsaufruf erforderlich.

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

ANSI-Farben in der Terminalausgabe

ANSI-Escape-Codes fügen der Terminalausgabe Farben hinzu. Verwenden Sie sie, um das Agent-Präfix, die Eingabeaufforderung für Benutzereingaben und Warnungen visuell zu unterscheiden. Die Bibliothek colorama bietet plattformübergreifende Unterstützung einschließlich 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

Rich Library für eine ansprechendere Terminalausgabe

Die Bibliothek rich bietet Markdown-Darstellung, syntaxhervorgehobene Codeblöcke, Tabellen und Spinner im Terminal. Sie eignet sich gut für gestreamte Agent-Ausgaben.

# 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 mit Tool-Aufrufen

Wenn Streaming mit Function Calling kombiniert wird, wird auch das Feld tool_calls stückweise gestreamt. Sammeln Sie die JSON-Zeichenkette über mehrere Chunks hinweg, bevor Sie sie parsen.

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

Tokenzähler anzeigen

Zeigen Sie während des Streamings einen laufenden Tokenzähler an, damit Benutzer die Nutzung überwachen und die Kosten nachvollziehen können. Der OpenAI-Stream enthält Nutzungsdaten im letzten Chunk, wenn stream_options={'include_usage': True} gesetzt ist.

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

Bewährte Vorgehensweisen für Streaming

Zusammenfassung bewährter Vorgehensweisen für Streaming-Ausgaben bei CLI-Agenten:

  • Verwenden Sie immer flush=True oder sys.stdout.flush(), um Pufferung zu verhindern
  • Sammeln Sie Tokens nach dem Streaming zum Speichern in einer Zeichenkette
  • Prüfen Sie finish_reason, um ein Abschneiden zu erkennen
  • Verwenden Sie ANSI-Farben oder rich für bessere visuelle Übersicht
  • Behandeln Sie das Streaming von Tool-Aufrufen, indem Sie JSON-Argument-Chunks sammeln

Wissensüberprüfung: Streaming-Ausgabe

Testen Sie Ihr Verständnis von Streaming-Ausgaben in CLI-Agenten.

Beantworte es in der interaktiven Lektion

Zusammenfassung: Streaming-Ausgabe in CLI-Agenten

Sie können jetzt Streaming-CLI-Agenten erstellen, die reaktionsschnell und modern wirken:

  • stream=True übergeben, um Streaming im OpenAI SDK zu aktivieren
  • print(token, end='', flush=True) verwenden, um Tokens sofort anzuzeigen
  • Tokens für die Verarbeitung nach dem Streaming in einer Zeichenkette sammeln
  • finish_reason im letzten Chunk prüfen, um eine Trunkierung zu erkennen
  • async for mit AsyncOpenAI für asynchrone Agenten verwenden
  • ANSI-Farben oder rich für ein ansprechendes Terminal-Erlebnis hinzufügen
Kostenlos starten

Lerne AI Agents mit einem KI-Tutor — kostenlos

Schreibe und führe echten Code in deinem Browser aus, bekomme sofortige Hilfe von einem 24/7 KI-Tutor und setze dein Lernen im Web oder in der App fort.

Kurse
60
Lektionen
239
Kostenlos lernenDen vollständigen Kurs ansehen

Häufig gestellte Fragen

Ist die Lektion „Streaming-Ausgabe in CLI-Agenten“ kostenlos?

Ja — der vollständige Text von „Streaming-Ausgabe in CLI-Agenten“ ist hier im Web kostenlos zu lesen. Um sie interaktiv zu üben (integrierter Code-Editor und 24/7 KI-Tutor) und den Rest des AI Agents-Kurses freizuschalten, upgrade auf CoddyKit PRO. Der AI Agents-Kurs umfasst insgesamt 4 Lektionen.

Was lerne ich in „Streaming-Ausgabe in CLI-Agenten“?

Gestreamte Tokens in Terminal-Oberflächen Zeichen für Zeichen ausgeben. Du übst AI Agents mit praktischem Code, den du direkt im Browser ausführst, und ein 24/7 KI-Tutor beantwortet deine Fragen während du die Lektion bearbeitest.

Brauche ich Erfahrung, um AI Agents zu starten?

Keine Vorkenntnisse erforderlich. AI Agents auf CoddyKit ist für Anfänger bis fortgeschrittene Lernende strukturiert, sodass du hier starten oder von Anfang an beginnen und in deinem eigenen Tempo voranschreiten kannst. Dies ist Lektion 4 von 4.

Wie lange dauert die Lektion „Streaming-Ausgabe in CLI-Agenten“?

Die meisten CoddyKit-Lektionen dauern etwa 5–10 Minuten. Jede ist kompakt und interaktiv, sodass du stetig Fortschritte machst und genau dort weitermachst, wo du aufgehört hast – im Web und in der App.

Kann ich in dieser AI Agents-Lektion Code schreiben und ausführen?

Ja. Jede AI Agents-Lektion enthält einen integrierten Code-Editor, sodass du echten Code direkt in deinem Browser schreibst und ausführst und sofort KI-Feedback erhältst — ohne lokale Einrichtung erforderlich.

Alle Lektionen in diesem Kurs

  1. Kommandozeilen-Oberflächen für Agenten erstellen
  2. Interaktive Agenten im REPL-Stil
  3. Argumente parsen und Hilfetexte erstellen
  4. Streaming-Ausgabe in CLI-Agenten
← Zurück zu AI Agents