비동기 호출과 스트리밍

.aio 아래의 비동기 API

client.models의 모든 메서드는 client.aio.models에 비동기 짝이 있어. 인자와 반환 구조는 같고 await만 더하면 돼. client.aio.chats와 client.aio.files도 같은 방식이야.

스트리밍에서는 함수 이름이 달라져

스트리밍할 때는 generate_content 대신 generate_content_stream을 호출해. 일반 호출은 응답 하나를 반환하고, 스트리밍 호출은 청크를 순회하는 동기 이터레이터나 비동기 이터레이터를 반환해.

스트리밍이 중요한 이유

Flash가 첫 토큰을 내놓는 시간(TTFT)은 보통 200–400ms야. 500단어 답변의 마지막 토큰까지는 3–8초가 걸려. 스트리밍은 첫 토큰이 오는 순간 화면에 그리기 시작하므로, 전체 시간이 같아도 훨씬 빠르게 느껴져.

각 청크는 부분 GenerateContentResponse야

청크는 전체 응답과 같은 구조지만 chunk.text에는 텍스트 조각만 들어 있어. 들어오는 대로 이어 붙이면 돼. usage_metadata는 마지막 청크에만 나타나.

Code

동기 스트리밍·python

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()

비동기 스트리밍(운영 패턴)·python

import asyncio
from google import genai

client = genai.Client()

async def stream_story():
    full = []
    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.',
    ):
        if chunk.text:
            print(chunk.text, end='', flush=True)
            full.append(chunk.text)
        if chunk.usage_metadata:  # final chunk 에만 채워짐
            final_usage = chunk.usage_metadata
    print(f'\n[total tokens: {final_usage.total_token_count}]')
    return ''.join(full)

asyncio.run(stream_story())

멀티모달 입력 스트리밍·python

from google.genai import types

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

Exercise

명령줄 인자로 프롬프트를 받아 Flash의 응답을 토큰이 오는 대로 표준 출력에 흘려보내는 작은 CLI를 만들어. 마지막에는 총 토큰 수와 실제 경과 시간을 출력해. 그다음 같은 프롬프트와 seed=0으로 두 번 실행해 봐. 텍스트는 같아야 하지만 시간은 달라질 거야. TTFT와 TTLT의 지연 차이를 기록해.