Streaming — token 단위 delivery · GPT Wire Quest

Streaming 은 마법이 아니라 사용자가 기다리는 방식을 바꾸는 UX야. 전체 응답이 3 초 뒤에 한꺼번에 나타나면 멈춘 것처럼 느껴질 수 있지만, 총 4 초가 걸려도 첫 token 이 300ms 만에 보이면 바로 반응한다고 느껴. 사용자에게 보여주는 긴 텍스트는 streaming 을 우선 검토해.

sync 와 async 순회

sync 에서는 for chunk in stream:, async 에서는 async for chunk in stream: 을 써. 새 텍스트는 chunk.choices[0].delta.content 에 들어가지만 None 일 수도 있어. 첫 chunk 는 role 만, 마지막 chunk 는 finish_reason 만 담을 수 있으니 매번 확인해.

stream 은 반드시 닫아

stream 객체는 HTTP connection 을 잡고 있어. with stream: 또는 async with stream: 을 사용하거나 try/finally 에서 stream.close() 를 호출해. 닫지 않은 connection 은 시간이 지나면 connection 한도 문제로 돌아와.

사람이 보지 않는 작업은 streaming 이 필요 없을 수 있어

backend pipeline 이 최종 문자열만 필요하다면 chunk 를 하나씩 파싱하고 누적하는 비용만 늘 수 있어. 화면에 중간 결과를 보여줄 이유가 없다면 non-streaming 호출이 더 단순해.

Code

Sync 'for chunk in stream' streaming·python

stream = client.chat.completions.create(
    model="gpt-5.4",
    messages=[{"role": "user", "content": "Write a haiku about autumn."}],
    stream=True,
)

for chunk in stream:
    delta = chunk.choices[0].delta
    if delta.content:
        print(delta.content, end="", flush=True)
print()  # newline at end

Async 'async for chunk in stream' streaming·python

import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI()

async def stream_response():
    stream = await client.responses.create(
        model="gpt-5.4",
        input="Describe the solar system.",
        stream=True,
    )
    async for event in stream:
        if event.type == "response.output_text.delta":
            print(event.delta, end="", flush=True)

asyncio.run(stream_response())

Streaming — token 단위 delivery

sync 와 async 순회

stream 은 반드시 닫아

사람이 보지 않는 작업은 streaming 이 필요 없을 수 있어

Code

External links

Exercise

Progress

댓글 0