CLIエージェントでのストリーミング出力

CLIエージェントでストリーミングが重要な理由

OpenAI SDKでストリーミングを有効にする

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()とsys.stdout.write()の比較

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

ストリーミング中に完全な応答を収集する

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

AsyncOpenAIによる非同期ストリーミング

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

finish_reasonでストリームの終了を検知する

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カラー

# 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ライブラリ

# 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

ツール呼び出しとのストリーミング

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

トークンカウンターの表示

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

ストリーミングのベストプラクティス

理解度チェック：ストリーミング出力

振り返り：CLIエージェントでのストリーミング出力