AI / Google Antigravity Gemini Fundamentals Interview Questions

# Install the Python SDK
pip install google-genai

# Quickstart
from google import genai

client = genai.Client()  # reads GEMINI_API_KEY from env

# Interactions API (recommended for new projects as of June 2026)
interaction = client.interactions.create(
    model="gemini-3.5-flash",
    input="Explain recursion in one sentence.",
)
print(interaction.output_text)

from google import genai
client = genai.Client()

# Interactions API - new recommended approach
interaction = client.interactions.create(
    model="gemini-3.5-flash",
    input="I have 2 dogs in my house.",
)
print(interaction.output_text)

# Chain turns with previous_interaction_id (server manages state)
follow_up = client.interactions.create(
    model="gemini-3.5-flash",
    input="How many paws are in my house?",
    previous_interaction_id=interaction.id,
)
print(follow_up.output_text)  # Gemini knows context from turn 1

# Legacy generateContent (still works, not deprecated)
response = client.models.generate_content(
    model="gemini-3.5-flash",
    contents="I have 2 dogs in my house.",
)

from google import genai
client = genai.Client()

# Use Gemini 3.5 Flash - recommended default for new projects
interaction = client.interactions.create(
    model="gemini-3.5-flash",
    input="Write a Python async web scraper with error handling.",
)
print(interaction.output_text)

# Or use the alias (note: aliases update automatically as new releases arrive)
interaction = client.interactions.create(
    model="gemini-flash-latest",  # auto-updates to newest Flash release
    input="Review this code for bugs...",
)
# Note: use pinned model IDs in production, not -latest aliases

# GOOD: use a pinned stable model in production
response = client.models.generate_content(
    model="gemini-2.5-flash-001",   # stable, versioned, won't change
    contents="...",
)

# ACCEPTABLE: preview with awareness of deprecation risk
response = client.models.generate_content(
    model="gemini-3.1-pro-preview",  # preview: >= 2 weeks deprecation notice
    contents="...",
)

# AVOID in production: auto-updated aliases
response = client.models.generate_content(
    model="gemini-flash-latest",  # will silently upgrade to new model releases!
    contents="...",
)

# WARNING: Gemini 3 Pro Preview was deprecated March 9, 2026
# with minimal migration window - a real-world example of preview risk.
# Always use versioned IDs for customer-facing applications.

from google import genai
client = genai.Client()

# Using the Antigravity managed agent via the Interactions API
interaction = client.interactions.create(
    model="antigravity-preview-05-2026",
    input="""Analyse the top 10 Python packages on PyPI by download count.
    Fetch the data, create a bar chart, and save it as chart.png.
    Then write a brief summary of what you found.""",
)
print(interaction.output_text)

# For long-running tasks, use background execution:
interaction = client.interactions.create(
    model="antigravity-preview-05-2026",
    input="Scrape and analyse a year of release notes from github.com/google/gemini",
    background=True,  # returns immediately; poll for completion
)
print(f"Task started: {interaction.id}, status: {interaction.status}")
# Poll:
result = client.interactions.retrieve(interaction.id)
print(result.status)  # "pending" | "running" | "completed" | "failed"

from google import genai
client = genai.Client()

# Invoke a managed agent exactly like a model call:
interaction = client.interactions.create(
    model="antigravity-preview-05-2026",  # managed agent
    input="Build and test a Python REST API with FastAPI.",
)

# The agent handles:
# 1. Planning the approach
# 2. Writing the FastAPI code
# 3. Installing dependencies (pip install fastapi uvicorn)
# 4. Running tests
# 5. Reporting results
# All inside its sandbox - zero infrastructure from you

# You can also mix agents and models in one conversation:
research = client.interactions.create(
    model="gemini-deep-research-preview",  # Deep Research agent
    input="Research best practices for API authentication",
)

# Follow up with a standard model using the research context:
followup = client.interactions.create(
    model="gemini-3.5-flash",
    input="Now implement authentication based on this research.",
    previous_interaction_id=research.id,  # carries context forward
)

from google import genai
from google.genai import types

client = genai.Client()

# Enable search grounding in generateContent
response = client.models.generate_content(
    model="gemini-3.5-flash",
    contents="What are the latest Gemini API changes announced this week?",
    config=types.GenerateContentConfig(
        tools=[types.Tool(google_search=types.GoogleSearch())]
    ),
)
print(response.text)

# Access grounding metadata (sources used):
if response.candidates[0].grounding_metadata:
    for chunk in response.candidates[0].grounding_metadata.grounding_chunks:
        print(f"Source: {chunk.web.title} - {chunk.web.uri}")

# In the Interactions API:
interaction = client.interactions.create(
    model="gemini-3.5-flash",
    input="What happened in AI news this week?",
    tools=[{"google_search": {}}],
)

from google import genai
from pathlib import Path

client = genai.Client()

# Upload a file via the Files API (for large files)
my_video = client.files.upload(file=Path("presentation.mp4"))

# Multimodal request: video + text question
interaction = client.interactions.create(
    model="gemini-3.5-flash",
    input=[
        {"type": "text", "text": "Summarise the key points from this video."},
        {"type": "video", "uri": my_video.uri, "mime_type": "video/mp4"}
    ]
)
print(interaction.output_text)

# Image inline (base64 for small images, Files API for large):
import base64
image_data = base64.b64encode(Path("diagram.png").read_bytes()).decode()
interaction = client.interactions.create(
    model="gemini-3.5-flash",
    input=[
        {"type": "text", "text": "Explain this architecture diagram."},
        {"type": "image", "data": image_data, "mime_type": "image/png"}
    ]
)

from google import genai
from google.genai import types

client = genai.Client()

# Count tokens before sending (avoid context overflow errors):
token_response = client.models.count_tokens(
    model="gemini-3.5-flash",
    contents="Your very long document here...",
)
print(f"Token count: {token_response.total_tokens}")  # check vs 1M limit

# Long context use cases Gemini excels at:
# 1. Entire codebase analysis
response = client.models.generate_content(
    model="gemini-3.5-flash",
    contents=[
        "Summarise all TODO comments in this codebase and prioritise them.",
        entire_codebase_text,  # can be hundreds of files
    ]
)

# 2. Full book Q&A
# 3. Hour-long video analysis
# 4. Full conversation history

# Context caching: cache large stable prompts to reduce cost:
cached = client.caches.create(
    model="gemini-3.5-flash",
    contents=[large_document],  # cache this expensive prefix
    ttl="3600s",
)
# Reference the cache in subsequent calls:
response = client.models.generate_content(
    model="gemini-3.5-flash",
    contents="Summarise section 3.",
    config=types.GenerateContentConfig(cached_content=cached.name),
)

from google import genai
from google.genai import types

client = genai.Client()

# Dynamic thinking is ON by default for Gemini 3 models
# Use thinking_level to control depth:
response = client.models.generate_content(
    model="gemini-3.1-pro-preview",
    contents="Design a thread-safe cache with O(1) operations.",
    config=types.GenerateContentConfig(
        thinking_config=types.ThinkingConfig(
            thinking_level="high",  # "none" | "low" | "medium" | "high"
        )
    )
)
# Thinking steps are visible as execution steps in the Interactions API
interaction = client.interactions.create(
    model="gemini-3.1-pro-preview",
    input="Find all race conditions in this code: ...",
)
for step in interaction.steps:
    if step.type == "thought":
        print(f"Thinking: {step.signature[:50]}...")  # encrypted thought
    elif step.type == "model_output":
        print(f"Output: {step.content[0].text}")

from google import genai
from google.genai import types
from pydantic import BaseModel
from typing import Literal

client = genai.Client()

# Define schema with Pydantic
class CodeReview(BaseModel):
    summary: str
    bugs: list[str]
    severity: Literal["low", "medium", "high", "critical"]
    line_numbers: list[int]
    suggested_fix: str

# generateContent with response_schema:
response = client.models.generate_content(
    model="gemini-3.5-flash",
    contents="Review this Python code: def add(a, b): return a - b",
    config=types.GenerateContentConfig(
        response_mime_type="application/json",
        response_schema=CodeReview,
    )
)
import json
review = CodeReview(**json.loads(response.text))
print(f"Severity: {review.severity}")
print(f"Bugs: {review.bugs}")

# Interactions API structured output:
interaction = client.interactions.create(
    model="gemini-3.5-flash",
    input="Extract: Alice is 30, an engineer from London.",
    response_mime_type="application/json",
    response_schema={"type": "object",
                     "properties": {
                         "name": {"type": "string"},
                         "age": {"type": "integer"},
                         "role": {"type": "string"},
                         "city": {"type": "string"}
                     }}
)
data = json.loads(interaction.output_text)

from google import genai
from google.genai import types
import json

client = genai.Client()

# 1. Declare your functions as tools
get_weather = types.FunctionDeclaration(
    name="get_weather",
    description="Get current weather for a city",
    parameters={
        "type": "object",
        "properties": {
            "city": {"type": "string", "description": "City name"},
            "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
        },
        "required": ["city"]
    }
)

# 2. First call: model may request a function call
response = client.models.generate_content(
    model="gemini-3.5-flash",
    contents="What's the weather in Tokyo right now?",
    config=types.GenerateContentConfig(
        tools=[types.Tool(function_declarations=[get_weather])]
    )
)

# 3. Check if model requested a function call
for part in response.candidates[0].content.parts:
    if hasattr(part, "function_call"):
        fc = part.function_call
        print(f"Model wants to call: {fc.name}({dict(fc.args)})")
        result = your_weather_api(fc.args["city"])   # execute!

        # 4. Return result and continue
        response2 = client.models.generate_content(
            model="gemini-3.5-flash",
            contents=[
                types.Content(role="user", parts=[types.Part(text="What's the weather in Tokyo?")]),
                response.candidates[0].content,  # model turn with function_call
                types.Content(
                    role="user",
                    parts=[types.Part(
                        function_response=types.FunctionResponse(
                            name=fc.name,
                            response={"result": result}
                        )
                    )]
                )
            ],
        )
        print(response2.text)

from google import genai
from pathlib import Path

client = genai.Client()

# Upload a file
uploaded = client.files.upload(
    file=Path("large-document.pdf"),
    config={"display_name": "Q3 Report"}
)
print(f"File URI: {uploaded.uri}")
print(f"MIME type: {uploaded.mime_type}")
print(f"State: {uploaded.state}")  # PROCESSING | ACTIVE | FAILED

# Wait for processing (videos may take time)
import time
while uploaded.state == "PROCESSING":
    time.sleep(2)
    uploaded = client.files.get(name=uploaded.name)

# Use in multiple requests without re-uploading:
for question in ["Summarise section 2", "List all action items", "Find risks"]:
    response = client.models.generate_content(
        model="gemini-3.5-flash",
        contents=[
            question,
            types.Part(file_data=types.FileData(file_uri=uploaded.uri))
        ]
    )
    print(response.text)

# List and delete files:
for f in client.files.list():
    print(f.name, f.display_name)
client.files.delete(name=uploaded.name)

import asyncio
from google import genai

client = genai.Client()

async def realtime_session():
    # Live API uses async context manager
    async with client.aio.live.connect(
        model="gemini-3.1-flash-live-preview",
        config={
            "response_modalities": ["AUDIO"],   # get audio back
            "speech_config": {
                "voice_config": {"prebuilt_voice_config": {"voice_name": "Aoede"}}
            },
            "system_instruction": "You are a helpful coding assistant."
        }
    ) as session:
        # Send audio input
        await session.send(input=audio_chunk, end_of_turn=True)
        # Receive streaming audio response
        async for response in session.receive():
            if response.data:  # audio bytes
                play_audio(response.data)

asyncio.run(realtime_session())

from google import genai

client = genai.Client()

# Generate an image with Nano Banana 2 (Gemini 3.1 Flash Image)
interaction = client.interactions.create(
    model="gemini-3.1-flash-image-preview",  # Nano Banana 2
    input="A photorealistic image of a Python snake debugging code on a glowing terminal.",
)
# Access the generated image:
for step in interaction.steps:
    if step.type == "model_output":
        for part in step.content:
            if part.type == "image":
                with open("output.png", "wb") as f:
                    f.write(part.data)  # image bytes

# Multi-turn editing (conversational editing):
edit = client.interactions.create(
    model="gemini-3.1-flash-image-preview",
    input="Now make the snake wearing a debugging hat and smiling.",
    previous_interaction_id=interaction.id,  # maintains image context
)

# Or via generateContent with Gen Media endpoint:
response = client.models.generate_images(
    model="gemini-3-pro-image-preview",   # Nano Banana Pro
    prompt="Studio quality portrait of an AI robot",
)

from google import genai
from google.genai import types

client = genai.Client()

# Generate a video with Veo 3.1
operation = client.models.generate_videos(
    model="veo-3.1-generate-preview",
    prompt="A drone flies over a misty mountain range at sunrise, cinematic 4K",
    config=types.GenerateVideoConfig(
        aspect_ratio="16:9",
        duration_seconds=5,
        sample_count=1,
    )
)
# Video generation is async - poll for completion:
import time
while not operation.done:
    time.sleep(10)
    operation = client.operations.get(operation.name)

# Download the generated video:
video = operation.result.generated_videos[0]
client.files.download(file=video)
print(f"Video saved: {video.video.uri}")

# Note: Veo 2.0 models were deprecated June 30, 2026
# Migrate to veo-3.1-generate-preview or veo-3.1-fast-generate-preview

import time
from google import genai
from google.api_core import exceptions

client = genai.Client()

# Handle rate limiting with exponential backoff:
def generate_with_retry(prompt, model="gemini-2.5-flash", max_retries=3):
    for attempt in range(max_retries):
        try:
            response = client.models.generate_content(
                model=model, contents=prompt
            )
            return response.text
        except exceptions.ResourceExhausted:  # 429 rate limit error
            wait = 2 ** attempt
            print(f"Rate limited. Waiting {wait}s (attempt {attempt+1})")
            time.sleep(wait)
    raise Exception("Max retries exceeded")

# Tips for free tier:
# - Add delays between calls (1-2 seconds minimum)
# - Use Flash-Lite (higher RPD: 1,000/day)
# - Monitor quota in AI Studio dashboard
# - Enable billing for Tier 1 ($250/month cap available)

from google import genai
from google.genai import types

client = genai.Client()

# Create a batch job with multiple requests:
batch = client.batches.create(
    model="gemini-2.5-flash",
    src=types.CreateBatchJobSource(
        requests=[
            types.EmbedContentRequest(
                content=types.Content(parts=[types.Part(text="Summarise: " + doc)])
            )
            for doc in list_of_1000_documents
        ]
    ),
)
print(f"Batch ID: {batch.name}, state: {batch.state}")

# Poll for completion (up to 24 hours):
import time
while batch.state in ("JOB_STATE_PENDING", "JOB_STATE_RUNNING"):
    time.sleep(60)
    batch = client.batches.get(name=batch.name)

# Retrieve results:
for result in client.batches.list_responses(name=batch.name):
    print(result.response.candidates[0].content.parts[0].text)

# All current Gemini 3 and 2.5 models support the Batch API

# Python SDK setup:
pip install google-genai

import os
from google import genai

# Authentication - reads GEMINI_API_KEY from environment
os.environ["GEMINI_API_KEY"] = "your-key-here"  # or set externally
client = genai.Client()

# Basic text generation:
interaction = client.interactions.create(
    model="gemini-3.5-flash",
    input="Write a Python web scraper.",
)
print(interaction.output_text)

# Async version:
import asyncio
async def async_example():
    interaction = await client.aio.interactions.create(
        model="gemini-3.5-flash",
        input="Async web scraping example",
    )
    return interaction.output_text

asyncio.run(async_example())

// JavaScript/TypeScript SDK:
npm install @google/genai

import { GoogleGenAI } from "@google/genai";

const ai = new GoogleGenAI({});  // reads GEMINI_API_KEY from env

// Basic interaction:
const interaction = await ai.interactions.create({
    model: "gemini-3.5-flash",
    input: "Explain async/await in JavaScript.",
});
console.log(interaction.outputText);

// Streaming:
const stream = await ai.interactions.create({
    model: "gemini-3.5-flash",
    input: "Write a long blog post about AI.",
    stream: true,
});
for await (const chunk of stream) {
    process.stdout.write(chunk.outputText ?? "");
}

from google import genai
from google.genai import types
from datetime import timedelta

client = genai.Client()

# Create a cache with large stable content:
large_document = open("entire-codebase.py").read()  # 500k tokens

cached = client.caches.create(
    model="gemini-3.5-flash",
    config=types.CreateCachedContentConfig(
        contents=[
            types.Content(
                parts=[types.Part(text=large_document)]
            )
        ],
        system_instruction="You are a code review expert. Always give actionable feedback.",
        ttl=timedelta(hours=2),  # cache lives for 2 hours
        display_name="codebase-cache",
    )
)
print(f"Cache name: {cached.name}")
print(f"Cached tokens: {cached.usage_metadata.total_token_count}")

# Use the cache across many requests:
for pr_diff in pull_requests:
    response = client.models.generate_content(
        model="gemini-3.5-flash",
        contents=pr_diff,  # only the diff changes per request
        config=types.GenerateContentConfig(
            cached_content=cached.name
        )
    )
    # You pay full price for pr_diff tokens
    # but discounted price for the 500k cached codebase tokens
    print(response.text)

# Delete cache when done:
client.caches.delete(name=cached.name)

# Cost estimation:
# Scenario: Process 10,000 support tickets
# Average: 500 input tokens + 200 output tokens each

docs = 10_000
per_doc_input = 500
per_doc_output = 200

# Gemini 2.5 Flash (illustrative: $0.30/1M input, $1.20/1M output)
input_cost = (docs * per_doc_input / 1_000_000) * 0.30   # $1.50
output_cost = (docs * per_doc_output / 1_000_000) * 1.20  # $2.40
total_sync = input_cost + output_cost  # $3.90

# With Batch API (~50% discount):
total_batch = total_sync * 0.5  # $1.95

print(f"Synchronous: ${total_sync:.2f}")
print(f"Batch API:   ${total_batch:.2f}")

# Always check current prices at:
# ai.google.dev/gemini-api/docs/pricing
# Prices change frequently with new model releases

# Google AI Studio (simple)
from google import genai
client = genai.Client()  # reads GEMINI_API_KEY

# Vertex AI (enterprise)
from google import genai
client = genai.Client(
    vertexai=True,
    project="my-gcp-project",
    location="us-central1",
)
# Vertex AI also accepts service account credentials
# and integrates with IAM roles

# Gemma 4 models (latest as of mid-2026):
# gemma-4-26b-a4b-it (26B instruct)
# gemma-4-31b-it (31B instruct)
# Available on AI Studio and via Gemini API for inference
# Also available for local use via:

# Hugging Face:
from transformers import AutoModelForCausalLM, AutoTokenizer
model = AutoModelForCausalLM.from_pretrained("google/gemma-4-9b")

# Ollama:
# ollama run gemma3

# Google also makes Gemma available through the Gemini API:
# This lets you use the API interface without leaving Google's ecosystem
response = client.models.generate_content(
    model="gemma-4-26b-a4b-it",  # Gemma via Gemini API
    contents="Explain gradient descent.",
)

from google import genai

client = genai.Client()

# Method 1: generateContent with stream=True
for chunk in client.models.generate_content_stream(
    model="gemini-3.5-flash",
    contents="Write a comprehensive guide to Python async programming.",
):
    print(chunk.text, end="", flush=True)  # prints each chunk as it arrives
print()  # final newline

# Method 2: Interactions API with stream=True
for chunk in client.interactions.create(
    model="gemini-3.5-flash",
    input="Explain the history of computer science.",
    stream=True,
):
    if chunk.output_text:
        print(chunk.output_text, end="", flush=True)

# Async streaming (for FastAPI, async web apps):
async def stream_async():
    async for chunk in await client.aio.models.generate_content_stream(
        model="gemini-3.5-flash",
        contents="Async streaming example.",
    ):
        print(chunk.text, end="", flush=True)
        yield chunk.text  # yield to web framework (SSE, WebSocket)

# Python Easter egg
import antigravity
# -> Opens https://xkcd.com/353/ in your default web browser

# The antigravity module also contains a hidden geocode function:
import antigravity
print(dir(antigravity))
# Output: ['__builtins__', '__doc__', '__file__', '__loader__',
#          '__name__', '__package__', '__spec__', 'fly', 'geohash']

# The geohash function implements the xkcd geohashing algorithm:
antigravity.geohash(37.421542, -122.085589, b"2005-05-26-10458.68")
# Computes a Geohash based on date and Dow Jones opening price

# The module's source code is a fun read:
import inspect
import antigravity
print(inspect.getsource(antigravity))

from google import genai
from google.genai import types

client = genai.Client()

# Configure safety thresholds:
response = client.models.generate_content(
    model="gemini-3.5-flash",
    contents="Your prompt here",
    config=types.GenerateContentConfig(
        safety_settings=[
            types.SafetySetting(
                category="HARM_CATEGORY_DANGEROUS_CONTENT",
                threshold="BLOCK_ONLY_HIGH",  # options below
            ),
            types.SafetySetting(
                category="HARM_CATEGORY_HARASSMENT",
                threshold="BLOCK_MEDIUM_AND_ABOVE",
            ),
        ]
    )
)

# Check safety ratings on the response:
for rating in response.candidates[0].safety_ratings:
    print(f"{rating.category}: {rating.probability}")

# Check if response was blocked:
if response.prompt_feedback.block_reason:
    print(f"Blocked: {response.prompt_feedback.block_reason}")

from google import genai
client = genai.Client()

# Deep Research agent - model ID may vary; check current docs
# Typically accessed via AI Studio or the interactions API:
interaction = client.interactions.create(
    model="gemini-deep-research-preview",
    input="""Conduct comprehensive research on the performance
    characteristics of Python async frameworks in 2026:
    FastAPI, Starlette, Litestar, and Blacksheep.
    Compare throughput, latency, ecosystem maturity,
    and real-world adoption. Produce a structured report
    with data sources cited.""",
    background=True,  # Long research tasks run in background
)
print(f"Research started: {interaction.id}")

# Poll for completion (research may take 5-30 minutes):
import time
while True:
    result = client.interactions.retrieve(interaction.id)
    print(f"Status: {result.status}")
    if result.status == "completed":
        print(result.output_text)  # full research report
        break
    elif result.status == "failed":
        print("Research failed")
        break
    time.sleep(60)

// Firebase AI Logic - Android (Kotlin)
import com.google.firebase.ai.generativeai.GenerativeModel

val model = Firebase.ai.generativeModel("gemini-3.5-flash")
val response = model.generateContent("Explain Kotlin coroutines.")
println(response.text)

// iOS (Swift)
let model = FirebaseAI.ai().generativeModel(modelName: "gemini-3.5-flash")
let response = try await model.generateContent("Explain Swift concurrency.")
print(response.text ?? "")

// Web (JavaScript)
import { getAI, getGenerativeModel } from "firebase/ai";
const ai = getAI(firebaseApp);
const model = getGenerativeModel(ai, { model: "gemini-3.5-flash" });
const result = await model.generateContent("Explain Firebase.");
console.log(result.response.text());

from google import genai
from google.genai import types

client = genai.Client()

# Approach 1: Manual history with generateContent (legacy)
history = []
while True:
    user_input = input("You: ")
    history.append(types.Content(role="user", parts=[types.Part(text=user_input)]))

    response = client.models.generate_content(
        model="gemini-3.5-flash",
        contents=history,
    )
    assistant_reply = response.candidates[0].content
    history.append(assistant_reply)   # add model turn to history
    print(f"Gemini: {assistant_reply.parts[0].text}")

# Approach 2: Interactions API with previous_interaction_id (recommended)
# First turn:
first = client.interactions.create(
    model="gemini-3.5-flash",
    input="I am building a task management app in Python.",
)

# Second turn - model remembers the first:
second = client.interactions.create(
    model="gemini-3.5-flash",
    input="What database would you recommend for my app?",
    previous_interaction_id=first.id,  # server retrieves first turn context
)

# Third turn:
third = client.interactions.create(
    model="gemini-3.5-flash",
    input="Show me the schema for that database.",
    previous_interaction_id=second.id,
)
print(third.output_text)

from google import genai
from google.genai import types
import wave

client = genai.Client()

# Single-speaker TTS:
response = client.models.generate_content(
    model="gemini-3.1-flash-tts-preview",
    contents="Welcome to the Gemini API tutorial. Today we explore text-to-speech.",
    config=types.GenerateContentConfig(
        response_modalities=["AUDIO"],
        speech_config=types.SpeechConfig(
            voice_config=types.VoiceConfig(
                prebuilt_voice_config=types.PrebuiltVoiceConfig(
                    voice_name="Aoede"   # one of 30 available voices
                )
            )
        )
    )
)
# Extract and save audio:
audio_data = response.candidates[0].content.parts[0].inline_data.data
with open("output.wav", "wb") as f:
    f.write(audio_data)

# Multi-speaker TTS (new in 2026):
response = client.models.generate_content(
    model="gemini-3.1-flash-tts-preview",
    contents="""TTS the following conversation:
    <speaker name="Narrator">The story begins on a dark night.</speaker>
    <speaker name="Alice">Who goes there?</speaker>
    <speaker name="Bob">It is I, your long-lost friend!</speaker>
    """,
    config=types.GenerateContentConfig(
        response_modalities=["AUDIO"],
        speech_config=types.SpeechConfig(
            multi_speaker_voice_config=types.MultiSpeakerVoiceConfig(
                speaker_voice_configs=[
                    types.SpeakerVoiceConfig(speaker="Alice", voice_config=types.VoiceConfig(
                        prebuilt_voice_config=types.PrebuiltVoiceConfig(voice_name="Aoede")
                    )),
                    types.SpeakerVoiceConfig(speaker="Bob", voice_config=types.VoiceConfig(
                        prebuilt_voice_config=types.PrebuiltVoiceConfig(voice_name="Charon")
                    )),
                ]
            )
        )
    )
)

from google import genai
from google.genai import types
import numpy as np

client = genai.Client()

# Generate embeddings with Gemini Embedding:
result = client.models.embed_content(
    model="gemini-embedding-exp-03-07",  # current experimental embedding model
    contents=[
        "Python is a versatile programming language.",
        "Gemini is Google's AI model family.",
        "The stock market fell today.",
    ]
)

vectors = [e.values for e in result.embeddings]
print(f"Embedding dimensions: {len(vectors[0])}")  # typically 768 or 3072

# Compute cosine similarity:
def cosine_sim(a, b):
    a, b = np.array(a), np.array(b)
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

print(f"Python vs Gemini: {cosine_sim(vectors[0], vectors[1]):.3f}")  # medium
print(f"Python vs stocks: {cosine_sim(vectors[0], vectors[2]):.3f}")  # low

# Task type specification (improves accuracy):
result = client.models.embed_content(
    model="gemini-embedding-exp-03-07",
    contents="How does async/await work in Python?",
    config=types.EmbedContentConfig(
        task_type="RETRIEVAL_QUERY"  # vs RETRIEVAL_DOCUMENT, SEMANTIC_SIMILARITY
    )
)

from google import genai
from google.genai import types

client = genai.Client()

# generateContent with system instruction:
response = client.models.generate_content(
    model="gemini-3.5-flash",
    contents="Explain recursion.",   # user input
    config=types.GenerateContentConfig(
        system_instruction="You are a Python tutor for beginners.
        Always give examples in Python.
        Keep explanations under 200 words.
        Never use jargon without explaining it first.",
    )
)
print(response.text)

# Interactions API with system instruction:
interaction = client.interactions.create(
    model="gemini-3.5-flash",
    input="Explain recursion.",
    system_instruction="You are a Python tutor for beginners...",
)

# Effective system instruction principles:
EFFECTIVE_SYSTEM_INSTRUCTION = """
Role: You are a senior Python code reviewer at a fintech company.

Constraints:
- Always check for security vulnerabilities first
- Flag any use of eval() or exec() as critical risk
- Use Decimal for financial calculations, never float
- Output in this format: Risk Level, Issues, Recommendations

Tone: Professional, direct, constructive
"""

# System instruction vs user prompt:
# System instruction: stable per session; shapes all responses; cached efficiently
# User prompt:        changes per turn; the actual task or question

from google import genai
client = genai.Client()

interaction = client.interactions.create(
    model="gemini-3.5-flash",
    input="What is the square root of the number of stars in the Milky Way?",
    tools=[{"google_search": {}}],
)

# Inspect every step the model took:
for i, step in enumerate(interaction.steps):
    print(f"Step {i}: type={step.type}")

    if step.type == "thought":
        # Encrypted thinking (content not shown but signature confirms it happened)
        print(f"  Thinking... (signature: {step.signature[:20]}...)")

    elif step.type == "tool_call":
        print(f"  Tool call: {step.tool_call.name}")
        print(f"  Arguments: {step.tool_call.arguments}")

    elif step.type == "tool_result":
        print(f"  Tool result received")

    elif step.type == "code_execution":
        print(f"  Code: {step.code}")
        print(f"  Output: {step.output}")

    elif step.type == "model_output":
        print(f"  Final answer: {step.content[0].text}")

# Use steps to build a UI showing real-time reasoning:
for step in interaction.steps:
    if step.type == "tool_call":
        show_ui("Searching the web for: " + step.tool_call.arguments.get("query",""))
    elif step.type == "thought":
        show_ui("Thinking...")
    elif step.type == "model_output":
        show_ui("Answer: " + step.content[0].text)

import time
from google import genai
from google.api_core import exceptions as google_exceptions

client = genai.Client()

def gemini_call_with_retry(prompt: str, model: str = "gemini-3.5-flash",
                           max_retries: int = 5) -> str:
    for attempt in range(max_retries):
        try:
            response = client.models.generate_content(
                model=model,
                contents=prompt,
            )
            # Check if blocked by safety filters:
            if not response.candidates:
                reason = response.prompt_feedback.block_reason
                raise ValueError(f"Response blocked: {reason}")

            return response.text

        except google_exceptions.ResourceExhausted:  # 429 rate limit
            wait = (2 ** attempt) + 0.1
            print(f"Rate limited. Waiting {wait:.1f}s...")
            time.sleep(wait)

        except google_exceptions.InvalidArgument as e:  # 400 bad request
            if "context window" in str(e).lower():
                raise ValueError("Prompt too long - reduce size.") from e
            raise  # Re-raise other bad request errors

        except google_exceptions.PermissionDenied:  # 403
            raise PermissionError("Invalid API key or insufficient permissions.")

        except google_exceptions.ServiceUnavailable:  # 503
            wait = 5 * (attempt + 1)
            print(f"Service unavailable. Waiting {wait}s...")
            time.sleep(wait)

        except google_exceptions.NotFound:  # 404
            raise ValueError(f"Model {model!r} not found or deprecated.")

    raise Exception(f"Max retries ({max_retries}) exceeded.")

from google import genai
from google.genai import types
from pathlib import Path

client = genai.Client()

# Method 1: Inline audio (small files < 20MB)
audio_bytes = Path("interview.mp3").read_bytes()
response = client.models.generate_content(
    model="gemini-3.5-flash",
    contents=[
        types.Part(
            inline_data=types.Blob(
                data=audio_bytes,
                mime_type="audio/mp3"
            )
        ),
        types.Part(text="Transcribe this interview and summarise the key points.")
    ]
)
print(response.text)

# Method 2: Upload via Files API (recommended for files > 20MB)
uploaded = client.files.upload(
    file=Path("podcast.mp3"),
    config={"display_name": "Tech Podcast Episode 42"}
)

# Wait for processing:
import time
while uploaded.state == "PROCESSING":
    time.sleep(2)
    uploaded = client.files.get(name=uploaded.name)

# Analyse the audio:
response = client.models.generate_content(
    model="gemini-3.5-flash",
    contents=[
        types.Part(file_data=types.FileData(file_uri=uploaded.uri, mime_type="audio/mp3")),
        types.Part(text="How many different speakers are in this podcast? Identify each speaker's role.")
    ]
)
print(response.text)

from google import genai
from google.genai import types

client = genai.Client()

response = client.models.generate_content(
    model="gemini-3.5-flash",
    contents="What are the most recent Gemini API updates?",
    config=types.GenerateContentConfig(
        tools=[types.Tool(google_search=types.GoogleSearch())],
    )
)

print(response.text)

# Access grounding metadata:
candidate = response.candidates[0]
if candidate.grounding_metadata:
    gm = candidate.grounding_metadata

    # Sources used:
    print("\nSources:")
    for chunk in gm.grounding_chunks:
        if chunk.web:
            print(f"  - {chunk.web.title}: {chunk.web.uri}")

    # Which parts of the response are grounded:
    for support in gm.grounding_supports:
        text_segment = support.segment.text
        sources = [gm.grounding_chunks[i].web.uri
                   for i in support.grounding_chunk_indices]
        print(f"\nClaim: {text_segment[:80]}...")
        print(f"Supported by: {sources}")

    # Search entry point (rendered search suggestion UI element):
    if gm.search_entry_point:
        print(f"\nSearch UI: {gm.search_entry_point.rendered_content[:100]}")

# How to detect if your code is using deprecated models:
# Run this to audit model IDs in your codebase:
import re, glob

deprecated = [
    "gemini-1.0", "gemini-1.5", "gemini-2.0-flash",
    "gemini-2.0-flash-lite", "gemini-3-pro-preview",
    "gemini-3.0", "imagen", "veo-2"
]

for file_path in glob.glob("**/*.py", recursive=True):
    with open(file_path) as f:
        content = f.read()
    for dep in deprecated:
        if dep in content:
            print(f"DEPRECATED model ID found in {file_path}: {dep}")

# Best practice: centralise model IDs in config
# config.py:
GEMINI_MODEL = "gemini-3.5-flash"   # update here only when migrating
GEMINI_EMBED_MODEL = "gemini-embedding-exp-03-07"

from google import genai
from google.genai import types
import numpy as np

client = genai.Client()

# APPROACH 1: Built-in file_search (simplest - Google manages everything)
# 1. Upload documents:
uploaded = client.files.upload(file="knowledge-base.pdf")

# 2. Query against them (no vector DB needed):
response = client.models.generate_content(
    model="gemini-3.5-flash",
    contents="What is our refund policy?",
    config=types.GenerateContentConfig(
        tools=[types.Tool(retrieval=types.Retrieval(
            vertex_ai_search=types.VertexAISearch(datastore="projects/my-project/...")
        ))]
    )
)
print(response.text)

# APPROACH 2: Custom RAG with Gemini embeddings
def embed(texts: list[str]) -> list[list[float]]:
    result = client.models.embed_content(
        model="gemini-embedding-exp-03-07",
        contents=texts,
        config=types.EmbedContentConfig(task_type="RETRIEVAL_DOCUMENT"),
    )
    return [e.values for e in result.embeddings]

# Index your documents:
docs = ["Our policy is...", "For returns, contact..."]
doc_embeddings = embed(docs)  # store in Pinecone/pgvector/Weaviate

# At query time:
query_embedding = client.models.embed_content(
    model="gemini-embedding-exp-03-07",
    contents="Can I return a product?",
    config=types.EmbedContentConfig(task_type="RETRIEVAL_QUERY"),
).embeddings[0].values

# Find most similar docs (cosine similarity):
similarities = [np.dot(query_embedding, d) /
                (np.linalg.norm(query_embedding) * np.linalg.norm(d))
                for d in doc_embeddings]
top_idx = int(np.argmax(similarities))

# Generate with context:
response = client.models.generate_content(
    model="gemini-3.5-flash",
    contents=f"Context: {docs[top_idx]}\n\nQuestion: Can I return a product?",
)

# Example production-hardened Gemini call:
from google import genai
from google.genai import types
from google.api_core import exceptions
import time, logging

logger = logging.getLogger("gemini-app")
client = genai.Client()

GEMINI_MODEL = "gemini-2.5-flash-001"  # PINNED - never use -latest

def production_generate(prompt: str, user_id: str) -> str:
    for attempt in range(4):
        try:
            response = client.models.generate_content(
                model=GEMINI_MODEL,
                contents=prompt,
                config=types.GenerateContentConfig(
                    max_output_tokens=2048,   # cap output
                    temperature=0.2,          # low for predictability
                    safety_settings=[
                        types.SafetySetting(
                            category="HARM_CATEGORY_DANGEROUS_CONTENT",
                            threshold="BLOCK_MEDIUM_AND_ABOVE"
                        )
                    ]
                )
            )
            if not response.candidates:
                raise ValueError(f"Blocked: {response.prompt_feedback.block_reason}")
            logger.info({"user": user_id, "tokens": response.usage_metadata.total_token_count})
            return response.text
        except exceptions.ResourceExhausted:
            time.sleep(2 ** attempt)
    raise RuntimeError("Max retries exceeded")