How to Use Google AI Mode for Complex Search Queries

Table of Contents

• How to Use Google AI Mode for Complex Search Queries
• Understanding the AI Mode Architecture and API Design
• Key Architectural Considerations
• Production Architecture Pattern
• Prerequisites and Environment Setup
• Create project directory
• Create virtual environment
• Core dependencies
• Development dependencies
• .env

📺 Watch: Neural Networks Explained

Video by 3Blue1Brown

---

In March 2025, Google introduced an experimental "AI Mode" within its search platform, fundamentally changing how users interact with search engines. According to Wikipedia, this feature enables users to input complex, multi-part queries and receive thorough, AI-generated responses using Google's Gemini [7] model, which enhances the system's reasoning capabilities and supports multimodal inputs including text, images, and voice.

This shift represents a significant change in user behavior: instead of typing frag [2]mented keywords and manually synthesizing results from multiple searches, users can now ask nuanced, multi-faceted questions in natural language and receive synthesized answers. For developers, data scientists, and power users, understanding how to leverage this capability programmatically is essential for building next-generation search applications.

In this tutorial, we'll build a production-ready Python application that interacts with Google's AI Mode, handles complex multi-part queries, processes multimodal inputs, and manages the response pipeline efficiently. We'll cover architecture decisions, edge cases, and real-world deployment considerations.

Understanding the AI Mode Architecture and API Design

Before diving into code, it's critical to understand the architectural implications of AI Mode. Traditional search APIs return ranked lists of URLs with snippets. AI Mode, by contrast, returns synthesized responses that may include text, structured data, and references to source materials.

Key Architectural Considerations

1. Query Complexity: AI Mode excels at multi-part queries like "Compare the energy efficiency of Tesla Powerwall vs LG Chem RESU, considering installation costs in California and available tax incentives for 2025." Traditional search would require 3-4 separate queries.

2. Multimodal Support: The Gemini model underlying AI Mode supports text, images, and voice inputs. This means your application can accept image uploads alongside text queries.

3. Response Streaming: AI-generated responses can be lengthy. Production applications should implement streaming to provide real-time feedback to users.

4. Rate Limiting and Cost Management: As of May 2026, Google's AI Mode is still experimental. Production deployments must handle rate limits, quota management, and cost tracking.

Production Architecture Pattern

We'll implement a microservice architecture with the following components:

• Query Orchestrator: Handles complex query decomposition and routing
• Multimodal Processor: Converts various input types to API-compatible formats
• Response Streamer: Manages streaming responses with backpressure
• Cache Layer: Reduces API calls for similar queries
• Monitoring Stack: Tracks latency, error rates, and token usage

Prerequisites and Environment Setup

Let's set up a robust development environment. We'll use Python 3.11+ with modern async patterns.

# Create project directory
mkdir ai-mode-search-engine
cd ai-mode-search-engine

# Create virtual environment
python3.11 -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate

# Core dependencies
pip install httpx==0.27.0 # Async HTTP client
pip install pydantic==2.7.0 # Data validation
pip install python-multipart==0.0.9 # File upload handling
pip install redis==5.0.0 # Caching
pip install prometheus-client==0.20.0 # Metrics
pip install structlog==24.1.0 # Structured logging
pip install tenacity==8.2.3 # Retry logic
pip install pillow==10.3.0 # Image processing
pip install python-dotenv==1.0.1 # Environment management

# Development dependencies
pip install pytest==8.1.0
pip install pytest-asyncio==0.23.0
pip install black==24.3.0
pip install mypy==1.9.0

Create a .env file for configuration:

# .env
GOOGLE_API_KEY=your_api_key_here
AI_MODE_ENDPOINT=https://experimental.google.com/ai-mode/v1
REDIS_URL=redis://localhost:6379/0
MAX_TOKENS=4096
TEMPERATURE=0.7
RATE_LIMIT_RPM=60
CACHE_TTL_SECONDS=3600

Building the Core AI Mode Client

Now we'll implement the core client that interacts with Google's AI Mode API. This client handles authentication, request formatting, streaming, and error recovery.

# ai_mode_client.py
import asyncio
import base64
import hashlib
import json
import time
from dataclasses import dataclass, field
from enum import Enum
from io import BytesIO
from pathlib import Path
from typing import AsyncGenerator, Optional, Union
from urllib.parse import urljoin

import httpx
from pydantic import BaseModel, Field, field_validator
from tenacity import (
 retry,
 stop_after_attempt,
 wait_exponential,
 retry_if_exception_type,
)
from PIL import Image

class QueryComplexity(Enum):
 """Enum for query complexity classification."""
 SIMPLE = "simple" # Single intent, few keywords
 MODERATE = "moderate" # Multiple intents, requires synthesis
 COMPLEX = "complex" # Multi-part with conditions, comparisons

class MultimodalInput(BaseModel):
 """Represents multimodal input to AI Mode."""
 text: str = Field(.., min_length=1, max_length=10000)
 images: list[str] = Field(default_factory=list, max_length=5)
 voice_data: Optional[str] = None # Base64 encoded audio

 @field_validator('images')
 @classmethod
 def validate_images(cls, v: list[str]) -> list[str]:
 """Validate image data is proper base64."""
 for img in v:
 try:
 base64.b64decode(img, validate=True)
 except Exception:
 raise ValueError(f"Invalid base64 image data")
 return v

class AIResponse(BaseModel):
 """Structured response from AI Mode."""
 query_id: str
 content: str
 sources: list[dict] = Field(default_factory=list)
 processing_time_ms: int
 token_count: int
 complexity: QueryComplexity

@dataclass
class AIClientConfig:
 """Configuration for the AI Mode client."""
 api_key: str
 endpoint: str = "https://experimental.google.com/ai-mode/v1"
 max_retries: int = 3
 timeout_seconds: int = 60
 max_tokens: int = 4096
 temperature: float = 0.7
 enable_streaming: bool = True
 cache_enabled: bool = True
 cache_ttl: int = 3600

class AIModeClient:
 """
 Production-grade client for Google's AI Mode.

 Handles authentication, request formatting, streaming,
 caching, rate limiting, and error recovery.
 """

 def __init__(self, config: AIClientConfig):
 self.config = config
 self._client = httpx.AsyncClient(
 base_url=config.endpoint,
 timeout=config.timeout_seconds,
 headers={
 "Authorization": f"Bearer {config.api_key}",
 "Content-Type": "application/json",
 "User-Agent": "AIModeSearchEngine/1.0",
 },
 )
 self._rate_limiter = asyncio.Semaphore(10) # Max 10 concurrent requests
 self._cache = {} if config.cache_enabled else None
 self._metrics = {
 "total_requests": 0,
 "cache_hits": 0,
 "errors": 0,
 "total_tokens": 0,
 }

 async def __aenter__(self):
 return self

 async def __aexit__(self, exc_type, exc_val, exc_tb):
 await self._client.aclose()

 def _generate_cache_key(self, query: MultimodalInput) -> str:
 """Generate deterministic cache key from query."""
 content = query.text + "".join(query.images[:2]) # Use first 2 images
 return hashlib.sha256(content.encode()).hexdigest()

 def _classify_complexity(self, text: str) -> QueryComplexity:
 """
 Classify query complexity for routing decisions.

 Simple: Single question, < 10 words
 Moderate: Multiple questions, 10-30 words
 Complex: Multi-part with conditions, > 30 words
 """
 word_count = len(text.split())
 has_conditions = any(word in text.lower() for word in 
 ["compare", "vs", "versus", "difference", "better"])
 has_multiple_questions = text.count("?") > 1

 if word_count > 30 or (has_conditions and word_count > 15):
 return QueryComplexity.COMPLEX
 elif word_count > 10 or has_multiple_questions:
 return QueryComplexity.MODERATE
 else:
 return QueryComplexity.SIMPLE

 @retry(
 stop=stop_after_attempt(3),
 wait=wait_exponential(multiplier=1, min=2, max=10),
 retry=retry_if_exception_type(
 (httpx.HTTPStatusError, httpx.TimeoutException)
 ),
 )
 async def _make_request(
 self, query: MultimodalInput
 ) -> dict:
 """
 Make the actual API request with retry logic.

 Uses exponential backoff for transient failures.
 """
 async with self._rate_limiter:
 payload = {
 "query": query.text,
 "images": query.images,
 "voice_data": query.voice_data,
 "max_tokens": self.config.max_tokens,
 "temperature": self.config.temperature,
 "stream": self.config.enable_streaming,
 }

 response = await self._client.post(
 "/search",
 json=payload,
 )
 response.raise_for_status()
 return response.json()

 async def search(
 self, query: MultimodalInput
 ) -> AIResponse:
 """
 Execute a search query against AI Mode.

 Handles caching, complexity classification, and metrics.
 """
 start_time = time.time()
 self._metrics["total_requests"] += 1

 # Check cache
 if self._cache is not None:
 cache_key = self._generate_cache_key(query)
 cached = self._cache.get(cache_key)
 if cached and (time.time() - cached["timestamp"]) < self.config.cache_ttl:
 self._metrics["cache_hits"] += 1
 return cached["response"]

 try:
 # Classify complexity for potential routing
 complexity = self._classify_complexity(query.text)

 # Make API request
 raw_response = await self._make_request(query)

 # Parse response
 response = AIResponse(
 query_id=raw_response.get("query_id", ""),
 content=raw_response.get("content", ""),
 sources=raw_response.get("sources", []),
 processing_time_ms=int((time.time() - start_time) * 1000),
 token_count=raw_response.get("token_count", 0),
 complexity=complexity,
 )

 # Update metrics
 self._metrics["total_tokens"] += response.token_count

 # Cache response
 if self._cache is not None:
 self._cache[cache_key] = {
 "response": response,
 "timestamp": time.time(),
 }

 return response

 except httpx.HTTPStatusError as e:
 self._metrics["errors"] += 1
 if e.response.status_code == 429:
 # Rate limited - implement exponential backoff
 retry_after = int(e.response.headers.get("Retry-After", 60))
 await asyncio.sleep(retry_after)
 return await self.search(query)
 elif e.response.status_code == 401:
 raise PermissionError("Invalid API key. Check your credentials.")
 elif e.response.status_code == 400:
 raise ValueError(f"Bad request: {e.response.text}")
 else:
 raise RuntimeError(f"API error {e.response.status_code}: {e.response.text}")

 except httpx.TimeoutException:
 self._metrics["errors"] += 1
 raise TimeoutError("AI Mode request timed out. Consider reducing query complexity.")

 async def stream_search(
 self, query: MultimodalInput
 ) -> AsyncGenerator[str, None]:
 """
 Stream search results for real-time display.

 Useful for long-running queries where users expect
 progressive updates.
 """
 async with self._rate_limiter:
 payload = {
 "query": query.text,
 "images": query.images,
 "max_tokens": self.config.max_tokens,
 "temperature": self.config.temperature,
 "stream": True,
 }

 async with self._client.stream(
 "POST",
 "/search",
 json=payload,
 ) as response:
 response.raise_for_status()
 async for chunk in response.aiter_text():
 if chunk.strip():
 yield chunk

 def get_metrics(self) -> dict:
 """Return current client metrics for monitoring."""
 return {
 **self._metrics,
 "cache_size": len(self._cache) if self._cache else 0,
 "rate_limiter_available": self._rate_limiter._value,
 }

Building the Query Orchestrator for Complex Queries

The real power of AI Mode lies in handling complex, multi-part queries. Let's build a query orchestrator that decomposes complex queries, routes them appropriately, and synthesizes results.

# query_orchestrator.py
import asyncio
import re
from dataclasses import dataclass
from typing import Optional

from pydantic import BaseModel, Field

class DecomposedQuery(BaseModel):
 """A single sub-query from a complex query decomposition."""
 id: str
 text: str
 dependencies: list[str] = Field(default_factory=list)
 priority: int = 0 # Higher = process first

class SynthesizedResponse(BaseModel):
 """Final synthesized response from multiple sub-queries."""
 main_answer: str
 sub_answers: dict[str, str]
 sources: list[dict]
 processing_time_ms: int

class QueryDecomposer:
 """
 Decomposes complex multi-part queries into manageable sub-queries.

 Handles:
 - Comparison queries (e.g., "Compare X vs Y")
 - Conditional queries (e.g., "If condition A, then B")
 - Multi-faceted queries (e.g., "What is X, how does Y work, and why Z?")
 """

 COMPARISON_PATTERN = re.compile(
 r"(compare|difference|vs|versus|better|worse)",
 re.IGNORECASE
 )

 CONDITIONAL_PATTERN = re.compile(
 r"(if|when|assuming|given that|provided)",
 re.IGNORECASE
 )

 MULTI_QUESTION_PATTERN = re.compile(r"\?")

 def decompose(self, query: str) -> list[DecomposedQuery]:
 """
 Decompose a complex query into sub-queries.

 Returns ordered list of sub-queries with dependency tracking.
 """
 sub_queries = []

 # Handle comparison queries
 if self.COMPARISON_PATTERN.search(query):
 sub_queries.extend(self._decompose_comparison(query))

 # Handle conditional queries
 if self.CONDITIONAL_PATTERN.search(query):
 sub_queries.extend(self._decompose_conditional(query))

 # Handle multi-question queries
 question_count = len(self.MULTI_QUESTION_PATTERN.findall(query))
 if question_count > 1:
 sub_queries.extend(self._decompose_multi_question(query))

 # If no decomposition needed, return original query
 if not sub_queries:
 sub_queries.append(
 DecomposedQuery(
 id="main",
 text=query,
 priority=0,
 )
 )

 return sub_queries

 def _decompose_comparison(self, query: str) -> list[DecomposedQuery]:
 """Decompose comparison queries into individual entity queries."""
 # Extract entities being compared
 # This is a simplified implementation; production would use NLP
 entities = re.findall(
 r"(?:compare|vs|versus)\s+(\w+(?:\s+\w+)?)",
 query,
 re.IGNORECASE
 )

 sub_queries = []
 for i, entity in enumerate(entities):
 sub_queries.append(
 DecomposedQuery(
 id=f"entity_{i}",
 text=f"Describe {entity} in detail",
 dependencies=[],
 priority=1,
 )
 )

 # Add synthesis query
 sub_queries.append(
 DecomposedQuery(
 id="synthesis",
 text=query,
 dependencies=[q.id for q in sub_queries],
 priority=0,
 )
 )

 return sub_queries

 def _decompose_conditional(self, query: str) -> list[DecomposedQuery]:
 """Decompose conditional queries into condition and result parts."""
 parts = self.CONDITIONAL_PATTERN.split(query)

 sub_queries = []
 for i, part in enumerate(parts):
 if part.strip():
 sub_queries.append(
 DecomposedQuery(
 id=f"conditional_{i}",
 text=part.strip(),
 dependencies=[],
 priority=i, # Earlier parts have higher priority
 )
 )

 return sub_queries

 def _decompose_multi_question(self, query: str) -> list[DecomposedQuery]:
 """Split multi-question queries into individual questions."""
 questions = re.split(r"\?+\s*", query)
 questions = [q.strip() + "?" for q in questions if q.strip()]

 return [
 DecomposedQuery(
 id=f"question_{i}",
 text=q,
 dependencies=[],
 priority=len(questions) - i, # Later questions have higher priority
 )
 for i, q in enumerate(questions)
 ]

class QueryOrchestrator:
 """
 Orchestrates complex queries by decomposing, routing, and synthesizing.

 Uses dependency graph to determine execution order.
 """

 def __init__(
 self,
 client: "AIModeClient",
 decomposer: Optional[QueryDecomposer] = None,
 ):
 self.client = client
 self.decomposer = decomposer or QueryDecomposer()

 async def execute_complex_query(
 self, query_text: str
 ) -> SynthesizedResponse:
 """
 Execute a complex multi-part query.

 Steps:
 1. Decompose query into sub-queries
 2. Build dependency graph
 3. Execute sub-queries in order
 4. Synthesize final response
 """
 start_time = asyncio.get_event_loop().time()

 # Step 1: Decompose
 sub_queries = self.decomposer.decompose(query_text)

 # Step 2: Build dependency graph
 dependency_graph = self._build_dependency_graph(sub_queries)

 # Step 3: Execute in dependency order
 results = {}
 for batch in self._topological_sort(dependency_graph):
 # Execute batch in parallel
 tasks = []
 for query_id in batch:
 sub_query = next(q for q in sub_queries if q.id == query_id)
 tasks.append(self._execute_sub_query(sub_query, results))

 batch_results = await asyncio.gather(*tasks)
 for query_id, result in zip(batch, batch_results):
 results[query_id] = result

 # Step 4: Synthesize
 main_answer = results.get("main", results.get("synthesis", ""))
 if not main_answer:
 # Combine all results
 main_answer = "\n\n".join(
 f"{q.text}**\n{r}" 
 for q, r in zip(sub_queries, results.values())
 )

 processing_time = int((asyncio.get_event_loop().time() - start_time) * 1000)

 return SynthesizedResponse(
 main_answer=main_answer,
 sub_answers=results,
 sources=[], # Would aggregate from all sub-queries
 processing_time_ms=processing_time,
 )

 def _build_dependency_graph(
 self, sub_queries: list[DecomposedQuery]
 ) -> dict[str, set[str]]:
 """Build adjacency list for dependency graph."""
 graph = {q.id: set(q.dependencies) for q in sub_queries}
 return graph

 def _topological_sort(
 self, graph: dict[str, set[str]]
 ) -> list[list[str]]:
 """
 Topological sort with level grouping.

 Returns batches of queries that can be executed in parallel.
 """
 # Copy graph to avoid mutation
 graph = {k: set(v) for k, v in graph.items()}

 batches = []
 while graph:
 # Find nodes with no dependencies
 ready = [node for node, deps in graph.items() if not deps]
 if not ready:
 raise ValueError("Circular dependency detected in query decomposition")

 batches.append(ready)

 # Remove ready nodes from graph
 for node in ready:
 del graph[node]
 for deps in graph.values():
 deps.difference_update(ready)

 return batches

 async def _execute_sub_query(
 self, sub_query: DecomposedQuery, previous_results: dict[str, str]
 ) -> str:
 """Execute a single sub-query, potentially using previous results."""
 # Inject previous results into query context
 context = ""
 for dep_id in sub_query.dependencies:
 if dep_id in previous_results:
 context += f"\nPrevious result for {dep_id}: {previous_results[dep_id][:500]}.."

 enhanced_query = sub_query.text
 if context:
 enhanced_query = f"{context}\n\nBased on the above, {sub_query.text}"

 multimodal_input = MultimodalInput(text=enhanced_query)
 response = await self.client.search(multimodal_input)
 return response.content

Handling Edge Cases and Production Considerations

Production deployments of AI Mode applications face several challenges. Let's address the most critical ones.

Rate Limiting and Backpressure

Google's AI Mode is experimental and likely has aggressive rate limits. Implement a token bucket algorithm for precise rate control:

# rate_limiter.py
import asyncio
import time
from collections import deque

class TokenBucketRateLimiter:
 """
 Token bucket algorithm for precise rate limiting.

 Allows burst traffic up to capacity, then throttles to steady rate.
 """

 def __init__(self, rate: float, capacity: int):
 """
 Args:
 rate: Tokens per second (steady state)
 capacity: Maximum burst size
 """
 self.rate = rate
 self.capacity = capacity
 self.tokens = capacity
 self.last_refill = time.monotonic()
 self._lock = asyncio.Lock()

 async def acquire(self, tokens: int = 1) -> float:
 """
 Acquire tokens, waiting if necessary.

 Returns wait time in seconds.
 """
 async with self._lock:
 self._refill()

 if self.tokens >= tokens:
 self.tokens -= tokens
 return 0.0

 # Calculate wait time
 deficit = tokens - self.tokens
 wait_time = deficit / self.rate

 # Wait and then consume
 await asyncio.sleep(wait_time)
 self._refill()
 self.tokens -= tokens
 return wait_time

 def _refill(self):
 """Refill tokens based on elapsed time."""
 now = time.monotonic()
 elapsed = now - self.last_refill
 self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
 self.last_refill = now

Error Recovery and Circuit Breaking

Implement a circuit breaker pattern to prevent cascading failures:

# circuit_breaker.py
import asyncio
import time
from enum import Enum

class CircuitState(Enum):
 CLOSED = "closed" # Normal operation
 OPEN = "open" # Failing, reject requests
 HALF_OPEN = "half_open" # Testing if service recovered

class CircuitBreaker:
 """
 Circuit breaker for AI Mode API calls.

 Prevents cascading failures by failing fast when
 the downstream service is unhealthy.
 """

 def __init__(
 self,
 failure_threshold: int = 5,
 recovery_timeout: float = 30.0,
 half_open_max_requests: int = 3,
 ):
 self.failure_threshold = failure_threshold
 self.recovery_timeout = recovery_timeout
 self.half_open_max_requests = half_open_max_requests

 self.state = CircuitState.CLOSED
 self.failure_count = 0
 self.last_failure_time = 0.0
 self.half_open_requests = 0
 self._lock = asyncio.Lock()

 async def call(self, coro):
 """
 Execute a coroutine with circuit breaker protection.

 Raises CircuitBreakerOpenError if circuit is open.
 """
 async with self._lock:
 if self.state == CircuitState.OPEN:
 if time.monotonic() - self.last_failure_time >= self.recovery_timeout:
 self.state = CircuitState.HALF_OPEN
 self.half_open_requests = 0
 else:
 raise CircuitBreakerOpenError("Circuit breaker is open")

 if self.state == CircuitState.HALF_OPEN:
 if self.half_open_requests >= self.half_open_max_requests:
 raise CircuitBreakerOpenError("Half-open limit reached")
 self.half_open_requests += 1

 try:
 result = await coro
 # Success - reset circuit
 async with self._lock:
 self.state = CircuitState.CLOSED
 self.failure_count = 0
 return result

 except Exception as e:
 async with self._lock:
 self.failure_count += 1
 self.last_failure_time = time.monotonic()

 if self.failure_count >= self.failure_threshold:
 self.state = CircuitState.OPEN
 raise

class CircuitBreakerOpenError(Exception):
 """Raised when circuit breaker prevents a request."""
 pass

Multimodal Input Processing

Handle image and voice inputs efficiently:

# multimodal_processor.py
import base64
from io import BytesIO
from pathlib import Path
from typing import Union

from PIL import Image

class MultimodalProcessor:
 """
 Process and optimize multimodal inputs for AI Mode.

 Handles:
 - Image resizing and compression
 - Voice data conversion
 - Input validation
 """

 MAX_IMAGE_SIZE_MB = 4
 MAX_IMAGE_DIMENSION = 2048
 SUPPORTED_IMAGE_FORMATS = {"JPEG", "PNG", "WEBP"}

 @staticmethod
 def process_image(image_path: Union[str, Path]) -> str:
 """
 Process an image file for AI Mode submission.

 Resizes if too large, converts to base64.
 """
 path = Path(image_path)
 if not path.exists():
 raise FileNotFoundError(f"Image not found: {path}")

 with Image.open(path) as img:
 # Convert to RGB if necessary
 if img.mode != "RGB":
 img = img.convert("RGB")

 # Resize if too large
 if max(img.size) > MultimodalProcessor.MAX_IMAGE_DIMENSION:
 img.thumbnail(
 (MultimodalProcessor.MAX_IMAGE_DIMENSION, 
 MultimodalProcessor.MAX_IMAGE_DIMENSION),
 Image.Resampling.LANCZOS
 )

 # Compress to meet size limit
 buffer = BytesIO()
 quality = 85
 while True:
 buffer.seek(0)
 buffer.truncate()
 img.save(buffer, format="JPEG", quality=quality, optimize=True)
 if buffer.tell() <= MultimodalProcessor.MAX_IMAGE_SIZE_MB * 1024 * 1024:
 break
 quality -= 5
 if quality < 20:
 break

 buffer.seek(0)
 return base64.b64encode(buffer.read()).decode("utf-8")

 @staticmethod
 def validate_multimodal_input(
 text: str,
 images: list[str] = None,
 voice_data: str = None,
 ) -> dict:
 """
 Validate and prepare multimodal input.

 Returns dict with warnings about potential issues.
 """
 warnings = []

 if not text.strip():
 warnings.append("Empty text query")

 if images:
 if len(images) > 5:
 warnings.append("More than 5 images may be truncated")

 for i, img in enumerate(images):
 try:
 decoded = base64.b64decode(img, validate=True)
 if len(decoded) > 10 * 1024 * 1024: # 10MB
 warnings.append(f"Image {i} exceeds 10MB")
 except Exception:
 warnings.append(f"Image {i} has invalid base64 encoding")

 if voice_data:
 try:
 base64.b64decode(voice_data, validate=True)
 except Exception:
 warnings.append("Voice data has invalid base64 encoding")

 return {"warnings": warnings, "is_valid": len(warnings) == 0}

Production Deployment and Monitoring

Here's a complete FastAPI application that ties everything together:

# main.py
import asyncio
import os
from contextlib import asynccontextmanager
from typing import AsyncGenerator

from fastapi import FastAPI, HTTPException, UploadFile, File, Form
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

from ai_mode_client import AIModeClient, AIClientConfig, MultimodalInput
from query_orchestrator import QueryOrchestrator
from rate_limiter import TokenBucketRateLimiter
from circuit_breaker import CircuitBreaker, CircuitBreakerOpenError
from multimodal_processor import MultimodalProcessor

# Configuration
config = AIClientConfig(
 api_key=os.getenv("GOOGLE_API_KEY"),
 endpoint=os.getenv("AI_MODE_ENDPOINT"),
 max_tokens=int(os.getenv("MAX_TOKENS", "4096")),
 temperature=float(os.getenv("TEMPERATURE", "0.7")),
)

# Global instances
client = AIModeClient(config)
orchestrator = QueryOrchestrator(client)
rate_limiter = TokenBucketRateLimiter(rate=10, capacity=20) # 10 req/s, burst 20
circuit_breaker = CircuitBreaker()

@asynccontextmanager
async def lifespan(app: FastAPI):
 """Manage application lifecycle."""
 # Startup
 yield
 # Shutdown
 await client._client.aclose()

app = FastAPI(
 title="AI Mode Search Engine",
 version="1.0.0",
 lifespan=lifespan,
)

class SearchRequest(BaseModel):
 query: str
 stream: bool = False

@app.post("/search")
async def search(request: SearchRequest):
 """
 Execute a search query against Google AI Mode.

 Supports both simple and complex multi-part queries.
 """
 # Rate limiting
 wait_time = await rate_limiter.acquire()
 if wait_time > 0:
 # Log rate limiting event
 pass

 try:
 # Circuit breaker protection
 response = await circuit_breaker.call(
 orchestrator.execute_complex_query(request.query)
 )

 return {
 "query": request.query,
 "response": response.main_answer,
 "sub_answers": response.sub_answers,
 "processing_time_ms": response.processing_time_ms,
 }

 except CircuitBreakerOpenError:
 raise HTTPException(
 status_code=503,
 detail="Service temporarily unavailable. Please try again later.",
 )
 except TimeoutError:
 raise HTTPException(
 status_code=504,
 detail="Query timed out. Consider simplifying your question.",
 )
 except ValueError as e:
 raise HTTPException(status_code=400, detail=str(e))
 except PermissionError as e:
 raise HTTPException(status_code=401, detail=str(e))

@app.post("/search/multimodal")
async def search_multimodal(
 query: str = Form(..),
 images: list[UploadFile] = File(None, max_length=5),
):
 """
 Search with multimodal input (text + images).

 Accepts up to 5 images alongside text query.
 """
 processed_images = []
 if images:
 for img in images:
 content = await img.read()
 # Process and optimize image
 processed = MultimodalProcessor.process_image(content)
 processed_images.append(processed)

 multimodal_input = MultimodalInput(
 text=query,
 images=processed_images,
 )

 # Validate
 validation = MultimodalProcessor.validate_multimodal_input(
 query, processed_images
 )

 try:
 response = await circuit_breaker.call(
 client.search(multimodal_input)
 )

 return {
 "query": query,
 "response": response.content,
 "sources": response.sources,
 "warnings": validation["warnings"],
 "processing_time_ms": response.processing_time_ms,
 }

 except Exception as e:
 raise HTTPException(status_code=500, detail=str(e))

@app.get("/metrics")
async def get_metrics():
 """Return client metrics for monitoring."""
 return {
 "client_metrics": client.get_metrics(),
 "rate_limiter": {
 "available_tokens": rate_limiter.tokens,
 "capacity": rate_limiter.capacity,
 },
 }

if __name__ == "__main__":
 import uvicorn
 uvicorn.run(
 "main:app",
 host="0.0.0.0",
 port=8000,
 reload=True,
 log_level="info",
 )

What's Next

Google's AI Mode represents a fundamental change in how users interact with search engines. By moving from keyword-based queries to natural language conversations with multimodal support, it enables more nuanced and productive information retrieval.

To extend this tutorial:

1. Implement caching with Redis: Replace the in-memory cache with Redis for distributed deployments
2. Add query analytics: Track which types of complex queries perform best
3. Build a feedback loop: Allow users to rate responses and use that data to improve query decomposition
4. Explore multimodal RAG: Combine AI Mode with vector database [3]s for domain-specific knowledge retrieval

The code in this tutorial is production-ready but should be adapted to your specific use case. Monitor your API usage carefully, as Google's AI Mode is still experimental and subject to change. As of May 2026, the feature continues to evolve, and staying updated with Google's developer documentation is essential for maintaining compatibility.

Remember that the true power of AI Mode lies not in replacing traditional search, but in augmenting it for complex, multi-faceted queries that previously required significant manual effort to answer.

---