Python 스트리밍

generate_content_stream — 같은 인자로 이터레이터 받기

스트리밍 메서드는 일반 메서드와 나란히 있고 같은 인자를 받아. 다른 점은 반환값뿐이야. GenerateContentResponse 하나 대신 부분 응답의 동기 이터레이터나 비동기 이터레이터를 받아.

기억할 패턴 세 가지

동기 스트림: for chunk in client.models.generate_content_stream(...)
비동기 스트림: async for chunk in await client.aio.models.generate_content_stream(...) — 호출 자체를 await한 뒤 async for로 순회해.
스트리밍 채팅: chat.send_message_stream(...) / chat.aio.send_message_stream(...)

진행하면서 이어 붙이고 마지막에 사용량을 잡아

각 청크의 chunk.text에는 그 구간의 텍스트 조각이 있어. 모두 이어 붙이면 전체 답변이 돼. usage_metadata는 마지막 청크에만 있으므로 순회하다 값이 나타날 때 저장해.

멀티모달도 같은 방식으로 스트리밍해

이미지, 업로드한 파일, 일반 호출에 쓸 여러 part의 콘텐츠를 그대로 전달할 수 있어. 모델이 첫 텍스트 토큰을 만드는 순간 스트림이 시작돼.

Code

동기 스트리밍·python

from google import genai

client = genai.Client()

for chunk in client.models.generate_content_stream(
    model='gemini-2.5-flash',
    contents='Tell me a 200-word story about a coffee scale.',
):
    if chunk.text:
        print(chunk.text, end='', flush=True)
print()  # final newline

사용량을 기록하는 비동기 스트리밍·python

import asyncio
from google import genai

client = genai.Client()

async def stream_with_usage():
    full_text = []
    final_usage = None

    async for chunk in await client.aio.models.generate_content_stream(
        model='gemini-2.5-flash',
        contents='Write a 300-word story about an octopus learning JavaScript.',
    ):
        if chunk.text:
            print(chunk.text, end='', flush=True)
            full_text.append(chunk.text)
        if chunk.usage_metadata:
            final_usage = chunk.usage_metadata

    print(f'\n\n[total tokens: {final_usage.total_token_count}]')
    return ''.join(full_text)

asyncio.run(stream_with_usage())

스트리밍 채팅·python

chat = client.chats.create(model='gemini-2.5-flash')

for chunk in chat.send_message_stream('Tell me a haiku.'):
    if chunk.text:
        print(chunk.text, end='', flush=True)
print()

for chunk in chat.send_message_stream('And another.'):
    if chunk.text:
        print(chunk.text, end='', flush=True)
print()

이미지 입력 스트리밍·python

from google.genai import types

for chunk in client.models.generate_content_stream(
    model='gemini-2.5-flash',
    contents=[
        "Describe what's happening in this image:",
        types.Part.from_uri(
            file_uri='gs://my-bucket/cafe.jpg',
            mime_type='image/jpeg',
        ),
    ],
):
    if chunk.text:
        print(chunk.text, end='', flush=True)

Exercise

명령줄 인자로 프롬프트를 받아 Flash 응답을 표준 출력으로 흘려보내고, 마지막에 사용한 총 토큰 수와 실제 경과 시간, 초당 토큰 수를 출력해. 그다음 --sync 플래그를 추가해 비동기 대신 동기 스트리밍 API를 쓰게 하고, 두 방식이 얼마나 부드럽게 느껴지는지 비교해.