Generative AI Google Brain

gemini-multimodal-live-api-real-time-rag-retail-gemini-2-0

4 min read

Real-time Retrieval Augmented Generation (RAG) using the Multimodal Live API with Gemini 2.0

Source notebook

Repo path: gemini/multimodal-live-api/real_time_rag_retail_gemini_2_0.ipynb · Open on GitHub · intermediate

Builds a retail RAG pipeline with Gemini Multimodal Live API for grounded text and audio answers.

Summary

This notebook demonstrates Vertex AI Gemini API and Multimodal Live API calls, first showing that an ungrounded model cannot answer Cymbal Bikes policy and service questions reliably. It downloads Cymbal Bikes PDFs from Cloud Storage, extracts and chunks PDF text, embeds chunks with text-embedding-005, retrieves relevant chunks with cosine similarity, and prompts Gemini with retrieved context. The final async RAG pipeline supports both text and audio output using the same retrieval path.

Key code patterns

Vertex AI GenAI client

client = genai.Client(
    vertexai=True,
    project=PROJECT_ID,
    location=LOCATION,
)
MODEL_ID = "gemini-2.0-flash-live-preview-04-09"
MODEL = f"projects/{PROJECT_ID}/locations/{LOCATION}/publishers/google/models/{MODEL_ID}"
text_embedding_model = "text-embedding-005"

Configures the Google GenAI SDK to call Vertex AI endpoints and builds the Live API model resource path.

Live API text session

async def generate_content(query: str) -> str:
    config = LiveConnectConfig(response_modalities=["TEXT"])
    async with client.aio.live.connect(model=MODEL, config=config) as session:
        await session.send(input=query, end_of_turn=True)
        response = []
        async for message in session.receive():
            if message.text:
                response.append(message.text)
            if message.server_content.turn_complete:
                return "".join(str(x) for x in response)

Uses an async Live API websocket-style session for low-latency text output.

Live API audio output

async def generate_audio_content(query: str):
    config = LiveConnectConfig(response_modalities=["AUDIO"])
    async with client.aio.live.connect(model=MODEL, config=config) as session:
        await session.send(input=query, end_of_turn=True)
        audio_parts = []
        async for message in session.receive():
            if message.server_content.model_turn:
                for part in message.server_content.model_turn.parts:
                    if part.inline_data:
                        audio_parts.append(np.frombuffer(part.inline_data.data, dtype=np.int16))
            if message.server_content.turn_complete and audio_parts:
                display(Audio(np.concatenate(audio_parts), rate=24000, autoplay=True))

Requests native audio output from the Multimodal Live API and plays returned inline audio chunks.

Embedding with retry

@retry(wait=wait_random_exponential(multiplier=1, max=120), stop=stop_after_attempt(4))
def get_embeddings(embedding_client, embedding_model, text, output_dim=768):
    try:
        response = embedding_client.models.embed_content(
            model=embedding_model,
            contents=[text],
            config=types.EmbedContentConfig(output_dimensionality=output_dim),
        )
        return [response.embeddings[0].values]
    except Exception as e:
        if "RESOURCE_EXHAUSTED" in str(e):
            return None
        raise

Generates Vertex AI text embeddings with exponential backoff for quota-sensitive calls.

PDF chunk index

def build_index(document_paths, embedding_client, embedding_model, chunk_size=512):
    all_chunks = []
    for doc_path in document_paths:
        with open(doc_path, "rb") as file:
            pdf_reader = PyPDF2.PdfReader(file)
            for page_num, page in enumerate(pdf_reader.pages):
                page_text = page.extract_text()
                chunks = [page_text[i:i + chunk_size] for i in range(0, len(page_text), chunk_size)]
                for chunk_num, chunk_text in enumerate(chunks):
                    all_chunks.append({"document_name": doc_path, "page_number": page_num + 1, "chunk_number": chunk_num, "chunk_text": chunk_text, "embeddings": get_embeddings(embedding_client, embedding_model, chunk_text)})
    return pd.DataFrame(all_chunks)

Creates a small searchable RAG index with chunk text, page metadata, and embeddings.

Cosine top-k retrieval

query_embedding = get_embeddings(embedding_client, embedding_model, query)
similarities = [
    cosine_similarity(query_embedding, chunk_emb)[0][0]
    for chunk_emb in vector_db["embeddings"]
]
top_indices = np.argsort(similarities)[-top_k:]
relevant_chunks = vector_db.iloc[top_indices]
context = "\n\n".join(
    f"[Page {r.page_number}, Chunk {r.chunk_number}]: {r.chunk_text}"
    for _, r in relevant_chunks.iterrows()
)

Embeds the query with the same model as the documents and formats the most similar chunks as prompt context.

Grounded generation prompt

prompt = f"""Based on the following context, please answer the question.
 
Context:
{context}
 
Question: {query}
 
Answer:"""
if modality == "text":
    return await generate_content(prompt)
if modality == "audio":
    await generate_audio_content(prompt)

Constrains Gemini answers to retrieved context while switching only the response modality.

Models & APIs used

When to use this

Use this pattern when a low-latency retail assistant needs grounded answers from a small document set with text or audio responses.

Gotchas & caveats

  • Gemini 2.0 Flash live preview and the Google Gen AI SDK are marked experimental, and output can vary.
  • The notebook requires an existing Google Cloud project and the Vertex AI API enabled.
  • Colab users must authenticate, and the runtime must restart after installing google-genai and PyPDF2.
  • LOCATION defaults to us-central1 from GOOGLE_CLOUD_REGION when no region environment variable is set.
  • Embedding and generation calls may hit RESOURCE_EXHAUSTED; the notebook uses retry/backoff and quota fallback messages.
  • The Multimodal Live API section states the notebook uses text input and text or audio output.
  • The notebook notes that replicating real-life real-time scenarios in Colab can be challenging.

Best practices

  • Use genai.Client with vertexai=True, project, and location for Vertex AI endpoint calls.
  • Ground domain-specific answers with retrieved document context instead of relying on pretrained model knowledge.
  • Use the same embedding model for document chunks and user query embeddings.
  • Keep page number, chunk number, and document metadata with retrieved chunks.
  • Add exponential retry handling for quota-sensitive embedding and answer generation calls.
  • Optimize chunk size and evaluate chunking strategies for retrieval performance.
  • Consider managed scalable indexing services such as Vertex AI Search for larger systems.
  • Evaluate RAG applications before production using the referenced Vertex AI Gen AI evaluation service.

Graph View

Backlinks