Throughput optimization patterns

import google.generativeai as genai
import os
import time
from typing import Generator

genai.configure(api_key=os.environ['GOOGLE_API_KEY'])
model = genai.GenerativeModel('gemini-2.0-flash')

def throughput_optimization_demo():
    # Pattern 1: Synchronous batching with concurrent requests
    prompts = [
        'Classify sentiment: I love this product',
        'Classify sentiment: This broke after one day',
        'Classify sentiment: It works as described'
    ]
    
    # Naive approach: sequential (slow)
    print('=== Sequential (inefficient) ===')
    start = time.time()
    for prompt in prompts:
        response = model.generate_content(prompt)
        print(f'Response: {response.text[:50]}...')
    sequential_time = time.time() - start
    print(f'Time: {sequential_time:.2f}s\n')
    
    # Pattern 2: Streaming for lower time-to-first-token
    print('=== Streaming (optimized for latency) ===')
    start = time.time()
    response = model.generate_content(
        'List 5 optimization tips for API throughput',
        stream=True
    )
    first_chunk_time = None
    for i, chunk in enumerate(response):
        if i == 0:
            first_chunk_time = time.time() - start
        print(f'Chunk {i}: {chunk.text[:40]}...')
    total_stream_time = time.time() - start
    print(f'Time to first token: {first_chunk_time:.3f}s, Total: {total_stream_time:.2f}s\n')
    
    # Pattern 3: Context reuse via system_instruction
    print('=== Cached system instruction (optimized for cost) ===')
    system_prompt = '''You are a concise sentiment classifier. Respond with only one word: POSITIVE, NEGATIVE, or NEUTRAL.'''
    
    test_inputs = [
        'I absolutely love this',
        'Terrible experience',
        'It is what it is'
    ]
    
    start = time.time()
    for text in test_inputs:
        response = model.generate_content(
            text,
            system_instruction=system_prompt
        )
        print(f'Input: {text} → {response.text}')
    cached_time = time.time() - start
    print(f'Time with system_instruction: {cached_time:.2f}s\n')
    
    # Pattern 4: Parallel-safe batch simulation (async-friendly)
    print('=== Batch simulation with timing analysis ===')
    batch_size = 3
    total_prompts = 9
    
    start = time.time()
    for i in range(0, total_prompts, batch_size):
        batch = prompts + prompts[:batch_size]  # Repeat for demo
        responses = []
        for prompt in batch[i:i+batch_size]:
            resp = model.generate_content(prompt)
            responses.append(resp.text[:30])
        print(f'Batch {i//batch_size + 1}: {len(responses)} responses')
    batch_time = time.time() - start
    print(f'Batch processing time: {batch_time:.2f}s')
    print(f'\nSpeedup: {sequential_time/batch_time:.2f}x faster than sequential')

if __name__ == '__main__':
    throughput_optimization_demo()