C.W.K.
Stream
Lesson 02 of 04 · published

Python Streaming

~12 min · python, streaming, async-iterator

Level 0Spark
0 XP0/35 lessons0/10 achievements
0/140 XP to next level140 XP to go0% complete

generate_content_stream — 같은 인자, iterator 반환

Streaming method 가 non-streaming 과 나란히 살고 같은 인자 받음. 차이는 반환값뿐 — single GenerateContentResponse 대신 partial response 의 iterator (sync) 또는 async iterator (async) 받음.

기억할 패턴 세 개

  • Sync stream: for chunk in client.models.generate_content_stream(...)
  • Async stream: async for chunk in await client.aio.models.generate_content_stream(...) (호출 자체에 awaitasync for)
  • Streaming chat: chat.send_message_stream(...) / chat.aio.send_message_stream(...)

진행하면서 concat, 마지막에 usage 잡아

각 chunk 의 chunk.text 가 그 slice 의 partial text. Full reply 로 concat. usage_metadata 는 final chunk 에만 — iterate 하면서 lazy 하게 capture.

Multimodal 도 같은 방식으로 stream

Image, 업로드된 파일, non-streaming 에서 쓸 multi-part contents 어느 거나 pass 가능. 모델이 첫 text 토큰 생산하자마자 stream 시작.

Code

Sync streaming·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
Async streaming with usage capture·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())
Streaming chat·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()
Image input 으로 streaming·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)

External links

Exercise

argv 로 prompt 받아 Flash 응답을 stdout 에 stream 하고, 마지막에 출력: 사용 총 토큰, wall-clock 경과 시간, 토큰/초. 그 다음 --sync flag 추가해서 async 대신 sync streaming surface 사용. 두 방식의 perceived smoothness 비교.

Progress

Progress is local-only — sign in to sync across devices.
이 페이지에서 버그를 발견하셨거나 피드백이 있으세요?문제 신고

댓글 0

🔔 답글 알림 (로그인 필요)
로그인댓글을 남기려면 로그인해 주세요.

아직 댓글이 없어요. 첫 댓글을 남겨보세요.