토큰 스트리밍 이해

사용자 경험에서 스트리밍이 중요한 이유

LLM이 토큰을 생성하는 방식

# Conceptual model of autoregressive generation
prompt = 'The capital of France is'

# Step 1: process full prompt, predict next token
# token_1 = sample(logits) → ' Paris'

# Step 2: append token_1 to context, predict next
# token_2 = sample(logits) → '.'

# Step 3: append token_2 to context, predict next
# token_3 = sample(logits) → '<|end|>'

# Total time: time_to_process_prompt + n_tokens * time_per_token
# With streaming: first token arrives after time_to_process_prompt (TTFT)
# Without streaming: everything arrives after TTFT + n_tokens * time_per_token

TTFT와 TPOT: 두 가지 지연 시간 지표

import time
from openai import OpenAI

client = OpenAI()

def measure_streaming_latency(prompt: str):
    t_start = time.perf_counter()
    t_first_token = None
    token_times = []

    stream = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=[{'role': 'user', 'content': prompt}],
        stream=True,
    )
    for chunk in stream:
        if chunk.choices[0].delta.content:
            t_now = time.perf_counter()
            if t_first_token is None:
                t_first_token = t_now
                print(f'TTFT: {(t_first_token - t_start) * 1000:.0f}ms')
            else:
                token_times.append(t_now - token_times[-1] if token_times else t_now - t_first_token)
            token_times.append(t_now)
    print(f'TPOT avg: {1000 * (token_times[-1] - t_first_token) / max(len(token_times)-1, 1):.1f}ms')

stream=True 매개변수

from openai import OpenAI

client = OpenAI()

# Non-streaming: wait for complete response
response = client.chat.completions.create(
    model='gpt-4o-mini',
    messages=[{'role': 'user', 'content': 'Explain RAG in one paragraph.'}],
)
full_text = response.choices[0].message.content

# Streaming: receive tokens incrementally
stream = client.chat.completions.create(
    model='gpt-4o-mini',
    messages=[{'role': 'user', 'content': 'Explain RAG in one paragraph.'}],
    stream=True,
)
for chunk in stream:
    delta = chunk.choices[0].delta.content
    if delta:  # delta can be None for non-content chunks
        print(delta, end='', flush=True)
print()  # newline at end

전체 응답 누적

def stream_and_accumulate(prompt: str) -> str:
    stream = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=[{'role': 'user', 'content': prompt}],
        stream=True,
    )

    full_text = ''
    finish_reason = None

    for chunk in stream:
        choice = chunk.choices[0]
        delta = choice.delta.content
        if delta:
            print(delta, end='', flush=True)  # real-time display
            full_text += delta               # accumulate
        if choice.finish_reason:
            finish_reason = choice.finish_reason

    print()  # newline
    print(f'Finished: {finish_reason}, total chars: {len(full_text)}')
    return full_text

사용 통계와 함께 스트리밍

stream = client.chat.completions.create(
    model='gpt-4o-mini',
    messages=[{'role': 'user', 'content': 'What is a vector database?'}],
    stream=True,
    stream_options={'include_usage': True},  # include token counts
)

full_text = ''
usage = None

for chunk in stream:
    if chunk.choices and chunk.choices[0].delta.content:
        full_text += chunk.choices[0].delta.content
    if chunk.usage:  # arrives in the final chunk
        usage = chunk.usage

if usage:
    print(f'Prompt tokens: {usage.prompt_tokens}')
    print(f'Completion tokens: {usage.completion_tokens}')
    print(f'Total tokens: {usage.total_tokens}')

스트리밍을 사용하지 않아야 하는 경우

Anthropic 및 Gemini API를 사용한 스트리밍

import anthropic

ant_client = anthropic.Anthropic(api_key='YOUR_KEY')

# Anthropic streaming
with ant_client.messages.stream(
    model='claude-sonnet-4-5',
    max_tokens=1024,
    messages=[{'role': 'user', 'content': 'Explain hybrid search briefly.'}],
) as stream:
    for text in stream.text_stream:
        print(text, end='', flush=True)

# Final message with usage stats
final_msg = stream.get_final_message()
print(f'\nInput tokens: {final_msg.usage.input_tokens}')
print(f'Output tokens: {final_msg.usage.output_tokens}')

생성기 기반 스트리밍 인터페이스

from typing import Generator

def stream_completion(
    messages: list[dict],
    model: str = 'gpt-4o-mini',
    **kwargs,
) -> Generator[str, None, None]:
    stream = client.chat.completions.create(
        model=model,
        messages=messages,
        stream=True,
        **kwargs,
    )
    for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta:
            yield delta

# Usage: pipe to stdout
for token in stream_completion([{'role': 'user', 'content': 'Hello!'}]):
    print(token, end='', flush=True)

# Usage: accumulate
full = ''.join(stream_completion([{'role': 'user', 'content': 'Hello!'}]))

터미널 및 CLI 애플리케이션에서의 스트리밍

import sys

def stream_to_terminal(messages: list[dict]):
    stream = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=messages,
        stream=True,
    )
    token_count = 0
    for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta:
            sys.stdout.write(delta)  # no newline added
            sys.stdout.flush()       # MUST flush or output buffers
            token_count += 1
    print()  # final newline
    print(f'({token_count} tokens generated)')

스트리밍과 오류 복구

import signal

def stream_with_timeout(messages, timeout_seconds=30):
    def timeout_handler(signum, frame):
        raise TimeoutError('LLM stream timed out')

    signal.signal(signal.SIGALRM, timeout_handler)
    signal.alarm(timeout_seconds)

    try:
        stream = client.chat.completions.create(
            model='gpt-4o-mini',
            messages=messages,
            stream=True,
        )
        for chunk in stream:
            delta = chunk.choices[0].delta.content
            if delta:
                yield delta
    except TimeoutError:
        yield '\n[Response timed out]'
    except Exception as e:
        yield f'\n[Error: {str(e)}]'
    finally:
        signal.alarm(0)  # cancel timeout

빠른 확인

레슨 요약