AI/ChatGPT

openai api streaming — python 예제와 usage 받기

2026. 10. 26.

OpenAI API 스트리밍은 요청에 stream=True만 넣으면 켜집니다. 답을 다 만든 뒤 한 번에 받는 대신, 만들어지는 대로 조각(이벤트)을 받아 바로 화면에 찍습니다. 파이썬에서는 Responses API면 response.output_text.delta 이벤트의 delta를, Chat Completions면 chunk.choices[0].delta.content를 이어 붙이면 됩니다. 토큰 사용량(usage)도 받을 수 있습니다. Responses API는 마지막 response.completed 이벤트에 들어 있고, Chat Completions는 stream_options={"include_usage": True}를 줘야 마지막 청크에 옵니다. 아래는 2026년 10월 23일 공식 문서와 openai 파이썬 SDK 3.37.0 기준입니다.

1. openai api streaming이란 — 언제 쓰나

스트리밍을 안 켜면 API는 모델이 출력을 전부 만든 뒤 HTTP 응답 하나로 돌려줍니다. 답이 길면 그동안 화면은 빈 채로 기다려야 합니다. 스트리밍을 켜면 서버가 SSE(server-sent events)로 출력 앞부분부터 흘려 보내므로, 받는 쪽은 첫 글자부터 바로 보여 줄 수 있습니다. ChatGPT 화면에서 글자가 타이핑되듯 나오는 그 방식입니다.

쓰면 좋은 곳: 사용자가 화면 앞에서 기다리는 채팅, 긴 글 생성, 터미널 도구처럼 체감 속도가 중요한 곳.
굳이 안 써도 되는 곳: 결과를 DB에 저장만 하는 백엔드 작업, 응답 전체를 JSON으로 파싱해야 하는 곳. 이런 대량 작업은 오히려 openai batch api로 묶어 보내는 편이 싸고 편합니다.

스트리밍은 두 API에서 모두 됩니다. 새로 짜는 코드라면 Responses API(client.responses)가 공식 가이드의 기본 예제이고, 기존 코드가 Chat Completions(client.chat.completions)라면 그대로 stream=True만 붙여도 됩니다. 이 글의 예제 모델 gpt-6-luna는 모델 페이지 기준으로 두 API와 스트리밍을 모두 지원합니다.

구분	Responses API	Chat Completions
켜는 법	`stream=True`	`stream=True`
받는 단위	타입이 정해진 이벤트 (`event.type`)	청크 (`chunk.choices[0].delta`)
글자 조각	`response.output_text.delta`의 `delta`	`delta.content`
토큰 사용량	`response.completed`의 `response.usage` (기본)	`include_usage`를 켜야 마지막 청크에
끝을 아는 법	`completed`·`incomplete`·`failed`·`error` 이벤트	`finish_reason` · `data: [DONE]`

2. openai api streaming python 예제 — Responses API

가장 기본 형태입니다. stream=True를 주면 create()가 응답 객체 대신 이벤트를 하나씩 내주는 스트림을 돌려주고, for문으로 돌면서 필요한 이벤트만 골라 씁니다. API 키는 OPENAI_API_KEY 환경 변수에서 읽습니다(키 만드는 법은 chatgpt api key 발급 글).

from openai import OpenAI

client = OpenAI()  # OPENAI_API_KEY 환경 변수를 읽는다

stream = client.responses.create(
    model="gpt-6-luna",
    input="스트리밍이 뭔지 한 문장으로 설명해 줘.",
    stream=True,
)

for event in stream:
    if event.type == "response.output_text.delta":
        print(event.delta, end="", flush=True)
    elif event.type == "response.completed":
        usage = event.response.usage
        print(f"\n[완료] 입력 {usage.input_tokens} / 출력 {usage.output_tokens} 토큰")
    elif event.type == "response.incomplete":
        reason = event.response.incomplete_details.reason
        print(f"\n[중간에 끊김] {reason}")
    elif event.type in ("response.failed", "error"):
        print(f"\n[실패] {event}")

API 키 없이 확인하려고 SDK에 가짜 서버를 붙여 위 코드를 그대로 돌린 출력입니다(글자 조각과 토큰 수는 가짜 서버가 보낸 값입니다).

스트리밍은 답을 조각으로 받는 방식입니다.
[완료] 입력 21 / 출력 14 토큰

print(..., end="", flush=True): 줄바꿈 없이 이어 찍고, 버퍼에 쌓지 말고 바로 내보냅니다. flush=True가 없으면 터미널에 따라 글자가 몰아서 나와 스트리밍 효과가 사라집니다.
이벤트는 이 밖에도 많습니다(response.created, response.output_item.added, response.content_part.added 등). 공식 가이드가 텍스트 스트리밍에서 보라고 꼽는 것은 response.created, response.output_text.delta, response.completed, error 네 개입니다. 모르는 이벤트는 그냥 지나치면 됩니다.

3. 스트리밍이 끝나는 이벤트 — completed만 기다리면 안 된다

Responses API 레퍼런스를 보면 응답이 끝나는 이벤트가 하나가 아닙니다. 위 예제가 네 가지를 모두 받는 이유입니다.

이벤트	뜻
`response.completed`	정상 완료. `event.response`에 전체 응답과 `usage`가 들어 있음
`response.incomplete`	덜 끝난 채 종료. 이유는 `response.incomplete_details.reason` (예: `max_output_tokens`)
`response.failed`	응답 실패
`error`	에러 발생. `code`·`message`·`param`을 가짐

가짜 서버로 max_output_tokens에 걸려 끊기는 경우와 도중에 error 이벤트가 오는 경우를 흘려 보내 보면 이렇게 나옵니다(server_error는 가짜 서버에 넣은 예시 값입니다).

스트리밍은 답을 조각으로
[중간에 끊김] max_output_tokens

스트리밍은
[실패] ResponseErrorEvent(code='server_error', message='The server had an error while processing your request.', param=None, sequence_number=5, type='error')

눈여겨볼 점은 Responses API의 error 이벤트는 예외가 아니라 그냥 이벤트로 넘어온다는 것입니다. 실행해 보면 프로그램은 종료 코드 0으로 끝납니다. completed만 기다리게 짜면 에러가 나도 조용히 반복문을 빠져나와 "답이 왜 잘렸지?"만 남습니다. 끝 이벤트는 위 네 개를 모두 처리합니다.

4. responses.stream() 헬퍼 — get_final_response 함정

SDK에는 with문으로 쓰는 client.responses.stream() 헬퍼도 있습니다. 이벤트를 돌리면서 SDK가 응답을 알아서 모아 두고, 끝나면 get_final_response()로 완성된 응답을 한 번에 꺼낼 수 있어 편합니다. stream=True는 따로 안 넣어도 됩니다.

from openai import OpenAI

client = OpenAI()

with client.responses.stream(
    model="gpt-6-luna",
    input="스트리밍이 뭔지 한 문장으로 설명해 줘.",
) as stream:
    for event in stream:
        if event.type == "response.output_text.delta":
            print(event.delta, end="", flush=True)

    final = stream.get_final_response()

print()
print("전체 텍스트:", final.output_text)
print("상태:", final.status, "/ 총 토큰:", final.usage.total_tokens)

스트리밍은 답을 조각으로 받는 방식입니다.
전체 텍스트: 스트리밍은 답을 조각으로 받는 방식입니다.
상태: completed / 총 토큰: 35

그런데 같은 코드에 incomplete나 error로 끝나는 스트림을 흘려 보내면 get_final_response()에서 죽습니다. SDK 3.37.0은 response.completed 이벤트를 못 받으면 예외를 던지기 때문입니다.

    final = stream.get_final_response()
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^
RuntimeError: Didn't receive a `response.completed` event.

답이 길어 max_output_tokens에 걸리는 일은 흔하므로, 헬퍼를 쓸 때는 반복문 안에서 response.incomplete·error 이벤트를 먼저 확인하거나 get_final_response()를 try/except RuntimeError로 감쌉니다. 끊긴 지점까지의 글이 필요하면 반복문에서 delta를 직접 모아 두는 쪽이 확실합니다.

5. openai api stream usage — Chat Completions stream_options

"스트리밍하면 토큰 사용량을 못 받는다"는 설명이 아직 많이 보이는데, 지금은 받을 수 있습니다. Chat Completions에서는 stream_options={"include_usage": True}를 넣으면 data: [DONE] 바로 앞에 청크가 하나 더 오고, 거기에 요청 전체의 usage가 들어 있습니다.

from openai import OpenAI

client = OpenAI()

stream = client.chat.completions.create(
    model="gpt-6-luna",
    messages=[{"role": "user", "content": "스트리밍이 뭔지 한 문장으로 설명해 줘."}],
    stream=True,
    stream_options={"include_usage": True},
)

for chunk in stream:
    if chunk.choices and chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)
    if chunk.usage:
        print(f"\n[완료] 입력 {chunk.usage.prompt_tokens} / 출력 {chunk.usage.completion_tokens} 토큰")

스트리밍은 답을 조각으로 받는 방식입니다.
[완료] 입력 21 / 출력 14 토큰

여기서 if chunk.choices and가 빠지면 마지막에 죽습니다. 공식 레퍼런스대로 usage가 담긴 마지막 청크는 choices가 빈 배열입니다. 흔히 보는 예제처럼 chunk.choices[0].delta.content를 바로 꺼내면, 글자는 다 찍힌 뒤 마지막 청크에서 이렇게 납니다.

    content = chunk.choices[0].delta.content
              ~~~~~~~~~~~~~^^^
IndexError: list index out of range

include_usage를 켜면 나머지 청크에도 usage 필드가 붙지만 값은 null이고, 마지막 청크에만 값이 있습니다. 그래서 위 예제처럼 if chunk.usage:로 거르면 됩니다. 토큰 수로 비용을 계산하는 법은 chatgpt api 요금 글에 정리해 두었습니다.

Responses API는 이 옵션이 필요 없습니다. 2번 예제처럼 response.completed 이벤트의 event.response.usage에 input_tokens·output_tokens·total_tokens가 이미 들어 있습니다.

6. 스트리밍 도중 에러 처리 — 두 API가 다르게 동작한다

요청 자체가 거절되는 에러(잘못된 키, 429 등)는 스트리밍이든 아니든 create()를 부르는 순간 예외로 납니다. 까다로운 건 글자를 받다가 중간에 나는 에러입니다. 가짜 서버로 두 API에 같은 상황을 만들어 보니 SDK 동작이 달랐습니다.

Responses API: type이 error인 이벤트로 넘어옵니다(3번). 예외가 아니라서 try/except로는 안 잡히고, 반복문 안에서 event.type을 봐야 합니다.
Chat Completions: 청크에 error가 실려 오면 SDK가 openai.APIError 예외를 던집니다. 반복문을 try로 감싸 잡습니다.

import openai
from openai import OpenAI

client = OpenAI()

stream = client.chat.completions.create(
    model="gpt-6-luna",
    messages=[{"role": "user", "content": "스트리밍이 뭔지 한 문장으로 설명해 줘."}],
    stream=True,
)

received = []
try:
    for chunk in stream:
        if chunk.choices and chunk.choices[0].delta.content:
            received.append(chunk.choices[0].delta.content)
except openai.APIError as e:
    print(f"스트림 중간 에러: {type(e).__name__}: {e.message}")
    print(f"그때까지 받은 글: {''.join(received)!r}")

스트림 중간 에러: APIError: The server had an error while processing your request.
그때까지 받은 글: '스트리밍은 '

받은 조각을 리스트에 모아 두면 에러가 나도 그때까지의 글은 남습니다. 화면에 "응답이 중간에 끊겼습니다" 같은 안내를 띄우고 다시 요청할지 정하면 됩니다. 요청 단계의 429 에러는 원인이 크레딧인지 한도인지부터 가려야 하는데, 그 방법은 openai api 429 에러 글에 있습니다.

7. openai api streaming async — AsyncOpenAI로 받기

FastAPI 같은 비동기 서버에서는 AsyncOpenAI를 씁니다. create()를 await하고, 반복문을 async for로 바꾸는 것 말고는 같습니다.

import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI()


async def main():
    stream = await client.responses.create(
        model="gpt-6-luna",
        input="스트리밍이 뭔지 한 문장으로 설명해 줘.",
        stream=True,
    )
    async for event in stream:
        if event.type == "response.output_text.delta":
            print(event.delta, end="", flush=True)
    print()


asyncio.run(main())

출력은 2번 예제와 같은 글자가 같은 순서로 나왔습니다. 비동기 서버에서 동기 OpenAI() 클라이언트로 스트림을 돌리면 그동안 이벤트 루프가 막혀 다른 요청이 줄줄이 늦어지니, 서버 코드에서는 AsyncOpenAI를 씁니다.

자주 묻는 질문 (FAQ)

Q. 함수 호출(function calling)도 스트리밍되나요?
됩니다. Responses API에서는 인자가 response.function_call_arguments.delta로 조각조각 오고 response.function_call_arguments.done에서 완성됩니다. 인자가 다 오기 전에는 JSON이 깨진 상태이므로, 함수를 실행하는 건 .done을 받은 뒤입니다. 함수 호출의 기본 흐름은 openai function calling 예제 글에 있습니다.

Q. 구조화 출력(JSON 스키마)도 스트리밍할 수 있나요?
공식 가이드에 구조화 출력 스트리밍 절이 따로 있고, SDK 3.37.0의 client.responses.stream()도 text_format 인자를 받습니다. 다만 받는 도중의 JSON은 미완성이라 바로 파싱할 수 없습니다. 스키마 쓰는 법은 chatgpt api json 출력 글을 참고하세요.

Q. 스트리밍을 중간에 끊으면 usage는 어떻게 되나요?
Chat Completions 레퍼런스에 따르면 스트림이 끊기거나 취소되면 usage가 담긴 마지막 청크를 못 받을 수 있습니다. 사용량을 집계해야 하는 서비스라면 끝까지 받지 못한 요청은 따로 표시해 두고, 대시보드 사용량과 맞춰 봅니다.

Q. Chat Completions 청크에 obfuscation이라는 필드가 있어요.
레퍼런스 설명으로는 스트리밍 청크 크기를 고르게 맞추려고 넣는 난수 문자열입니다(일부 사이드 채널 공격 대비). 기본으로 들어가며, 네트워크 구간을 믿을 수 있으면 stream_options={"include_obfuscation": False}로 빼서 전송량을 조금 줄일 수 있습니다. 코드에서는 무시해도 됩니다.

Q. 스트리밍하면 출력 검열(moderation)이 어렵나요?
공식 가이드도 그렇게 적어 둡니다. 부분 출력은 평가하기 어렵고, 생성 요청과 함께 moderation 점수를 받더라도 점수는 전체 출력이 끝난 뒤에 오며 중간 조각에는 붙지 않습니다. 사용자에게 보여 주기 전에 걸러야 하는 서비스라면 이 점을 설계에 넣어야 합니다.

마무리

정리하면 openai api streaming은 stream=True 한 줄로 켜고, 받은 조각을 이어 찍는 방식입니다. 코드에서 놓치기 쉬운 곳은 세 가지입니다. Responses API는 completed뿐 아니라 incomplete·failed·error까지 끝 이벤트로 처리하고(헬퍼의 get_final_response()는 completed가 없으면 예외), Chat Completions에서 usage를 받으려면 include_usage를 켜고 빈 choices를 걸러야 하며, 중간 에러는 Responses는 이벤트로, Chat Completions는 예외로 온다는 점입니다.

📚 참고 출처 (2026년 10월 23일 확인 · openai 파이썬 SDK 3.37.0)
· Streaming API responses — OpenAI API
· Responses streaming events — OpenAI API Reference
· Chat Completions streaming events — OpenAI API Reference
· GPT-6 Luna — OpenAI API

저작자표시 비영리 (새창열림)

Seemingly
Online

openai api streaming — python 예제와 usage 받기

1. openai api streaming이란 — 언제 쓰나

2. openai api streaming python 예제 — Responses API

3. 스트리밍이 끝나는 이벤트 — completed만 기다리면 안 된다

4. responses.stream() 헬퍼 — get_final_response 함정

5. openai api stream usage — Chat Completions stream_options

6. 스트리밍 도중 에러 처리 — 두 API가 다르게 동작한다

7. openai api streaming async — AsyncOpenAI로 받기

자주 묻는 질문 (FAQ)

마무리

COMMENTS

티스토리툴바

SeeminglyOnline

1. openai api streaming이란 — 언제 쓰나

2. openai api streaming python 예제 — Responses API

3. 스트리밍이 끝나는 이벤트 — completed만 기다리면 안 된다

4. responses.stream() 헬퍼 — get_final_response 함정

5. openai api stream usage — Chat Completions stream_options

6. 스트리밍 도중 에러 처리 — 두 API가 다르게 동작한다

7. openai api streaming async — AsyncOpenAI로 받기

자주 묻는 질문 (FAQ)

마무리

COMMENTS

티스토리툴바

Seemingly
Online