Articles/Integrations

⬡ Integrations/2026-04-19Advanced

Antigravity × Firebase Genkit: Building Production AI Apps—From Flow Design to Deployment

A complete guide to building production AI apps with Firebase Genkit and Antigravity. Covers type-safe flow design, RAG pipelines, streaming, and deployment—all with working code examples.

firebase-genkit genkit firebase¹¹ ai-flow rag⁸ streaming⁶ typescript²⁶ production⁷¹

✦ Premium Article

Have you ever spent hours debating whether to call the Gemini API directly or use Genkit? I certainly have.

Firebase Genkit is the AI application framework Google released in 2024. It's not simply a wrapper around the Gemini API—it's designed to cover the entire lifecycle of AI features, from local development, debugging, and testing through to production deployment and monitoring. And once you actually use it alongside Antigravity, you'll find the combination works even better than expected.

This guide walks through the complete process of designing, implementing, and deploying a Genkit app while making the most of Antigravity's AI autocomplete—drawing from real experience building production apps with this stack.

The Problems Firebase Genkit Solves

Calling the Gemini API directly is perfectly valid for simple text generation. But once you start building production-grade AI features, you quickly run into common pain points.

Debugging flows is hard. Tracking which prompt produced which response, or pinpointing where a tool call failed, requires building your own logging infrastructure. Tests are difficult to write. Processes that involve LLM calls are nearly impossible to unit test without mocking. Switching models is tedious. When you want to use a lightweight model in development and a high-accuracy model in production, you often end up rewriting large swaths of code.

Genkit addresses all of this through an abstraction layer called "flows." A flow is a unit that groups together LLM calls, external API calls, and tool executions—and it comes with execution tracing, typed input/output definitions, and a local debug UI out of the box.

Why does it pair well with Antigravity? Genkit is TypeScript-first and ships with rich type information. Antigravity's tab completion accurately fills in flow options, provider configurations, and Zod schemas. The time spent on flow definition boilerplate drops significantly.

Environment Setup

Start by setting up a project to use Genkit.

# Initialize project
mkdir my-genkit-app && cd my-genkit-app
npm init -y
npm install -D typescript ts-node @types/node
 
# Genkit core and plugins
npm install genkit @genkit-ai/googleai
npm install @genkit-ai/firebase  # For Firebase deployment
 
# TypeScript config
npx tsc --init --module nodenext --moduleResolution nodenext --strict

Add the following to tsconfig.json. Specifying path settings explicitly improves how accurately Antigravity resolves types.

{
  "compilerOptions": {
    "target": "ES2022",
    "module": "nodenext",
    "moduleResolution": "nodenext",
    "strict": true,
    "outDir": "./dist",
    "rootDir": "./src",
    "resolveJsonModule": true,
    "esModuleInterop": true
  },
  "include": ["src/**/*"]
}

Initializing the Genkit Instance

Create a Genkit instance in src/genkit.ts. Sharing this file across your app centralizes configuration.

// src/genkit.ts
import { genkit } from "genkit";
import { googleAI } from "@genkit-ai/googleai";
 
// Always read the API key from an environment variable—never hardcode it
const GOOGLE_API_KEY = process.env.GOOGLE_AI_API_KEY;
if (\!GOOGLE_API_KEY) {
  throw new Error("GOOGLE_AI_API_KEY environment variable is required");
}
 
export const ai = genkit({
  plugins: [
    googleAI({ apiKey: GOOGLE_API_KEY }),
  ],
  promptDir: "./prompts",
});
 
// Define models as constants for easy swapping later
export const MODELS = {
  fast: "googleai/gemini-2.0-flash",
  balanced: "googleai/gemini-2.5-pro",
  local: "googleai/gemma-4-9b-it", // For local testing
} as const;

Manage .env files through dotenv and verify .env is in .gitignore before committing. Asking the Antigravity agent to check .gitignore becomes a useful habit that guards against secret leaks.

# .env (never commit this)
GOOGLE_AI_API_KEY=YOUR_GOOGLE_AI_API_KEY
NODE_ENV=development

✦

Thank you for reading this far.

Continue Reading

What follows includes implementation code, benchmarks, and practical content we hope you'll find useful. This site runs without ads — server and development costs are supported entirely by members like you. If it's been helpful, we'd be truly grateful for your support.

WHAT YOU'LL LEARN

✦If you've been stuck on Genkit flow design, you'll walk away able to build type-safe AI pipelines today—leveraging Antigravity's autocomplete to cut boilerplate dramatically

✦Firestore Vector Search integration, chunking strategy, and error handling for RAG pipelines are all demystified with working, production-ready code

✦You'll acquire a clear mental model for designing retry logic, streaming, and cost control in production—knowledge you can apply directly to your own products

Secure payment via Stripe · Cancel anytime

✦

Unlock This Article

Get full access to the rest of this article. Buy once, read anytime. This site is ad-free — your support goes directly toward keeping it running.

Unlock all articles with Membership →

Implementing Basic Flows

The central concept in Genkit is the "flow"—a type-safe async processing unit with Zod-defined input and output schemas.

Simple Text Generation Flow

// src/flows/summarize.ts
import { z } from "genkit";
import { ai, MODELS } from "../genkit";
 
const SummarizeInput = z.object({
  text: z.string().min(1).max(10000).describe("Text to summarize"),
  language: z.enum(["ja", "en"]).default("en").describe("Output language"),
  maxLength: z.number().int().min(50).max(500).default(200).describe("Max character count"),
});
 
const SummarizeOutput = z.object({
  summary: z.string().describe("Generated summary"),
  keyPoints: z.array(z.string()).describe("Key takeaways"),
  wordCount: z.number().describe("Character count of original text"),
});
 
export const summarizeFlow = ai.defineFlow(
  {
    name: "summarizeFlow",
    inputSchema: SummarizeInput,
    outputSchema: SummarizeOutput,
  },
  async (input) => {
    const { output } = await ai.generate({
      model: MODELS.fast,
      output: { schema: SummarizeOutput },
      prompt: `Summarize the following text in ${input.language === "ja" ? "Japanese" : "English"}.
Keep the summary under ${input.maxLength} characters and extract up to 3 key points.
 
Text:
${input.text}`,
    });
 
    if (\!output) {
      throw new Error("LLM returned an empty response");
    }
 
    return { ...output, wordCount: input.text.length };
  }
);

To run the flow, start Genkit's development server.

# Launch the local debug UI at http://localhost:4000
npx genkit start -- npx ts-node src/index.ts

Opening http://localhost:4000 gives you a development UI where you can run flows from a GUI. It also records traces of every LLM call, which makes debugging far easier. This UI was the first thing that made me think, "okay, this is genuinely better."

Entry Point

// src/index.ts
import { summarizeFlow } from "./flows/summarize";
 
// Export flows so Genkit can discover them
export { summarizeFlow };
 
if (process.env.NODE_ENV === "development") {
  console.log("🚀 Genkit development server starting...");
}

Structured Output and Tool Integration

In real products, you often need more than simple text generation—you want data in a specific format, or you want to call external APIs and incorporate their results.

Defining Tools (Function Calling)

Tool definitions in Genkit combine a Zod schema with a handler function.

// src/tools/search.ts
import { z } from "genkit";
import { ai } from "../genkit";
 
export const searchProductsTool = ai.defineTool(
  {
    name: "searchProducts",
    description: "Searches the product database and returns matching items",
    inputSchema: z.object({
      query: z.string().describe("Search keywords"),
      category: z.string().optional().describe("Category filter"),
      maxResults: z.number().int().min(1).max(20).default(5),
    }),
    outputSchema: z.object({
      products: z.array(
        z.object({
          id: z.string(),
          name: z.string(),
          price: z.number(),
          category: z.string(),
        })
      ),
      totalCount: z.number(),
    }),
  },
  async (input) => {
    // Replace with your actual database call
    try {
      const mockProducts = [
        { id: "prod-001", name: `${input.query} Product A`, price: 19.80, category: input.category ?? "general" },
        { id: "prod-002", name: `${input.query} Product B`, price: 32.80, category: input.category ?? "general" },
      ].slice(0, input.maxResults);
 
      return { products: mockProducts, totalCount: mockProducts.length };
    } catch (error) {
      const message = error instanceof Error ? error.message : "Database error";
      throw new Error(`Product search failed: ${message}`);
    }
  }
);

Multi-Turn Conversation Flow

Here's an example of a shopping assistant that maintains conversational context.

// src/flows/shopping-assistant.ts
import { z } from "genkit";
import { ai, MODELS } from "../genkit";
import { searchProductsTool } from "../tools/search";
 
const AssistantInput = z.object({
  message: z.string().describe("User message"),
  conversationHistory: z
    .array(z.object({ role: z.enum(["user", "model"]), content: z.string() }))
    .default([])
    .describe("Prior conversation history"),
});
 
const AssistantOutput = z.object({
  reply: z.string(),
  recommendedProducts: z
    .array(z.object({ id: z.string(), name: z.string(), price: z.number() }))
    .default([]),
  requiresHumanSupport: z.boolean().default(false),
});
 
export const shoppingAssistantFlow = ai.defineFlow(
  {
    name: "shoppingAssistantFlow",
    inputSchema: AssistantInput,
    outputSchema: AssistantOutput,
  },
  async (input) => {
    const history = input.conversationHistory.map((msg) => ({
      role: msg.role,
      content: [{ text: msg.content }],
    }));
 
    const { output } = await ai.generate({
      model: MODELS.balanced,
      tools: [searchProductsTool],
      output: { schema: AssistantOutput },
      messages: [
        ...history,
        { role: "user" as const, content: [{ text: input.message }] },
      ],
      system: `You are a shopping assistant. Use the product search tool to suggest relevant items.
Respond in a friendly, helpful tone.
Set requiresHumanSupport to true for questions you cannot handle (complaints, legal issues, etc.).`,
    });
 
    if (\!output) {
      throw new Error("Assistant returned an empty response");
    }
 
    return output;
  }
);

With tool-using flows, the LLM autonomously decides which tools to call and when. Genkit handles the tool call loop automatically, so you don't need to manually re-feed tool results back into the LLM.

Building a RAG Pipeline

RAG (Retrieval-Augmented Generation) retrieves relevant information from an external database and passes it to the LLM. Pairing Genkit with Firestore Vector Search yields a scalable RAG pipeline.

Setting Up Firestore Vector Search

npm install -g firebase-tools
firebase init firestore
npm install @genkit-ai/firebase firebase-admin

Indexing Documents

// src/rag/indexer.ts
import { z } from "genkit";
import { ai } from "../genkit";
import { getFirestore, FieldValue } from "firebase-admin/firestore";
import { initializeApp, getApps } from "firebase-admin/app";
 
if (\!getApps().length) {
  initializeApp();
}
 
const db = getFirestore();
 
// Split text into overlapping chunks
function chunkText(text: string, chunkSize = 800, overlap = 100): string[] {
  const chunks: string[] = [];
  let start = 0;
 
  while (start < text.length) {
    const end = Math.min(start + chunkSize, text.length);
    chunks.push(text.slice(start, end));
    start = end - overlap;
    if (start >= text.length) break;
  }
 
  return chunks;
}
 
export const indexDocumentFlow = ai.defineFlow(
  {
    name: "indexDocumentFlow",
    inputSchema: z.object({
      documentId: z.string(),
      content: z.string(),
      metadata: z.record(z.string()).default({}),
    }),
    outputSchema: z.object({
      chunksIndexed: z.number(),
      status: z.string(),
    }),
  },
  async (input) => {
    const chunks = chunkText(input.content);
    let chunksIndexed = 0;
 
    for (const chunk of chunks) {
      try {
        const embeddingResult = await ai.embed({
          embedder: "googleai/text-embedding-004",
          content: chunk,
        });
 
        await db.collection("document_chunks").add({
          documentId: input.documentId,
          content: chunk,
          embedding: FieldValue.vector(embeddingResult.embedding),
          metadata: input.metadata,
          createdAt: FieldValue.serverTimestamp(),
        });
 
        chunksIndexed++;
      } catch (error) {
        // Log chunk-level errors and continue processing
        console.error(`Chunk ${chunksIndexed} indexing failed:`, error);
        if (chunksIndexed === 0) throw error; // Fail fast if nothing indexed yet
      }
    }
 
    return { chunksIndexed, status: chunksIndexed > 0 ? "success" : "failed" };
  }
);

Retrieval and Generation Flow

// src/rag/query.ts
import { z } from "genkit";
import { ai, MODELS } from "../genkit";
import { getFirestore, FieldValue } from "firebase-admin/firestore";
 
const db = getFirestore();
 
export const ragQueryFlow = ai.defineFlow(
  {
    name: "ragQueryFlow",
    inputSchema: z.object({
      question: z.string().min(1),
      topK: z.number().int().min(1).max(10).default(3),
    }),
    outputSchema: z.object({
      answer: z.string(),
      sources: z.array(
        z.object({
          documentId: z.string(),
          excerpt: z.string(),
          relevanceScore: z.number(),
        })
      ),
    }),
  },
  async (input) => {
    const queryEmbedding = await ai.embed({
      embedder: "googleai/text-embedding-004",
      content: input.question,
    });
 
    const vectorQuery = db
      .collection("document_chunks")
      .findNearest("embedding", FieldValue.vector(queryEmbedding.embedding), {
        limit: input.topK,
        distanceMeasure: "COSINE",
      });
 
    const snapshot = await vectorQuery.get();
 
    const sources = snapshot.docs.map((doc) => ({
      documentId: doc.data().documentId as string,
      excerpt: doc.data().content as string,
      relevanceScore: 1 - (doc.data().distance ?? 0),
    }));
 
    if (sources.length === 0) {
      return { answer: "No relevant documents were found.", sources: [] };
    }
 
    const context = sources
      .map((s, i) => `[Source ${i + 1}] ${s.excerpt}`)
      .join("\n\n");
 
    const { output } = await ai.generate({
      model: MODELS.balanced,
      output: { schema: z.object({ answer: z.string() }) },
      prompt: `Answer the question accurately and concisely based only on the provided sources.
If the answer isn't in the sources, say "That information was not found."
 
Sources:
${context}
 
Question: ${input.question}`,
    });
 
    return { answer: output?.answer ?? "Could not generate an answer", sources };
  }
);

Chunk size design directly impacts RAG accuracy. In practice, the right size depends on the nature of your documents—structured technical docs often benefit from smaller chunks (400–600 chars), while narrative prose can handle larger ones (800–1200 chars). Adding overlap prevents context from being severed at chunk boundaries.

Implementing Streaming Responses

For chat interfaces where you want to display responses incrementally, streaming is essential. Genkit's streaming flows are designed to pair naturally with Server-Sent Events (SSE).

// src/flows/streaming-chat.ts
import { z } from "genkit";
import { ai, MODELS } from "../genkit";
import type { IncomingMessage, ServerResponse } from "http";
 
const StreamingChatInput = z.object({
  message: z.string(),
  sessionId: z.string(),
});
 
export const streamingChatFlow = ai.defineStreamingFlow(
  {
    name: "streamingChatFlow",
    inputSchema: StreamingChatInput,
    outputSchema: z.string(),
    streamSchema: z.string(),
  },
  async (input, streamingCallback) => {
    const { stream, response } = await ai.generateStream({
      model: MODELS.fast,
      prompt: input.message,
      config: { maxOutputTokens: 2048, temperature: 0.7 },
    });
 
    try {
      for await (const chunk of stream) {
        const text = chunk.text;
        if (text) streamingCallback(text);
      }
    } finally {
      // Ensure the stream is always closed to prevent memory leaks
      await response.catch(() => {});
    }
 
    return (await response).text;
  }
);
 
// SSE endpoint for Express / Hono
export function createSSEHandler() {
  return async (req: IncomingMessage, res: ServerResponse) => {
    const body = await parseBody(req);
    const input = StreamingChatInput.parse(body);
 
    res.setHeader("Content-Type", "text/event-stream");
    res.setHeader("Cache-Control", "no-cache");
    res.setHeader("Connection", "keep-alive");
    res.setHeader("Access-Control-Allow-Origin", "*");
 
    try {
      const { stream } = streamingChatFlow.stream(input);
      for await (const chunk of stream) {
        res.write(`data: ${JSON.stringify({ text: chunk })}\n\n`);
      }
      res.write("data: [DONE]\n\n");
    } catch (error) {
      const msg = error instanceof Error ? error.message : "Unknown error";
      res.write(`data: ${JSON.stringify({ error: msg })}\n\n`);
    } finally {
      res.end();
    }
  };
}
 
async function parseBody(req: IncomingMessage): Promise<unknown> {
  return new Promise((resolve, reject) => {
    let body = "";
    req.on("data", (chunk) => { body += chunk; });
    req.on("end", () => {
      try { resolve(JSON.parse(body)); } catch (e) { reject(e); }
    });
    req.on("error", reject);
  });
}

Error Handling and Retry Strategies

Handling LLM errors and rate limits gracefully is among the most important production concerns. Genkit has built-in retry behavior, but fine-grained control requires custom implementation.

// src/utils/retry.ts
 
interface RetryOptions {
  maxAttempts: number;
  initialDelayMs: number;
  maxDelayMs: number;
  backoffFactor: number;
}
 
const DEFAULT_RETRY_OPTIONS: RetryOptions = {
  maxAttempts: 3,
  initialDelayMs: 1000,
  maxDelayMs: 30000,
  backoffFactor: 2,
};
 
// Exponential backoff retry
export async function withRetry<T>(
  fn: () => Promise<T>,
  options: Partial<RetryOptions> = {}
): Promise<T> {
  const opts = { ...DEFAULT_RETRY_OPTIONS, ...options };
  let lastError: Error | null = null;
  let delay = opts.initialDelayMs;
 
  for (let attempt = 1; attempt <= opts.maxAttempts; attempt++) {
    try {
      return await fn();
    } catch (error) {
      lastError = error instanceof Error ? error : new Error(String(error));
 
      // Fail immediately for non-retryable errors
      if (isNonRetryableError(error)) throw lastError;
 
      if (attempt < opts.maxAttempts) {
        console.warn(`Attempt ${attempt} failed: ${lastError.message}. Retrying in ${delay}ms...`);
        await sleep(delay);
        delay = Math.min(delay * opts.backoffFactor, opts.maxDelayMs);
      }
    }
  }
 
  throw new Error(`All ${opts.maxAttempts} attempts failed. Last error: ${lastError?.message}`);
}
 
function isNonRetryableError(error: unknown): boolean {
  if (error instanceof Error) {
    const msg = error.message.toLowerCase();
    return (
      msg.includes("invalid_argument") ||
      msg.includes("unauthenticated") ||
      msg.includes("permission_denied")
    );
  }
  return false;
}
 
const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
 
// Automatic primary → fallback model switching
export async function withModelFallback<T>(
  primaryFn: () => Promise<T>,
  fallbackFn: () => Promise<T>
): Promise<T> {
  try {
    return await withRetry(primaryFn, { maxAttempts: 2 });
  } catch (error) {
    console.warn("Primary model failed, switching to fallback:", error);
    return await withRetry(fallbackFn, { maxAttempts: 2 });
  }
}

Common Mistakes and How to Fix Them

Here are three issues I've run into when using Genkit in production.

① Zod schema and LLM output don't match

Even with output.schema specified, the LLM may return output that doesn't match—especially when z.enum() restricts the values.

// ❌ Problematic: enum is too strict
const OutputSchema = z.object({
  category: z.enum(["tech", "lifestyle", "food"]),
  // Fails with ZodError if the LLM returns "Technology"
});
 
// ✅ Better: normalize with transform
const OutputSchema = z.object({
  category: z.string().transform((val) => {
    const normalized = val.toLowerCase();
    if (normalized.includes("tech")) return "tech";
    if (normalized.includes("life")) return "lifestyle";
    if (normalized.includes("food")) return "food";
    return "tech"; // sensible default
  }),
});

② Memory leaks from streaming flows

When an error occurs during for await...of stream consumption, the stream can be left open if not properly closed.

// ❌ Problematic: stream may leak on error
const { stream } = await ai.generateStream({ ... });
for await (const chunk of stream) {
  process(chunk); // An exception here leaves the stream open
}
 
// ✅ Correct: use try-finally to guarantee stream closure
const { stream, response } = await ai.generateStream({ ... });
try {
  for await (const chunk of stream) {
    process(chunk);
  }
} catch (error) {
  console.error("Stream processing error:", error);
  throw error;
} finally {
  await response.catch(() => {}); // Silently close the stream
}

③ Circular dependency errors between flows

When multiple flows import each other, Node.js raises a circular dependency error. Separate shared tools and utilities into dedicated files.

// ❌ Problematic: flow-a.ts imports flow-b.ts, and vice versa
 
// ✅ Correct structure:
// src/tools/shared-tools.ts   ← All shared tools go here
// src/flows/flow-a.ts         ← Imports only from shared-tools
// src/flows/flow-b.ts         ← Imports only from shared-tools

Deploying to Firebase App Hosting

Genkit apps deploy directly to Firebase App Hosting (which is backed by Cloud Run).

# apphosting.yaml
runConfig:
  cpu: 1
  memoryMiB: 512
  concurrency: 80
  minInstances: 0  # Start at zero for cost savings
  maxInstances: 10
 
env:
  - variable: GOOGLE_AI_API_KEY
    secret: google-ai-api-key  # Managed in Firebase Secret Manager
  - variable: NODE_ENV
    value: production

// src/server.ts — deployment entry point
import express from "express";
import { summarizeFlow } from "./flows/summarize";
import { ragQueryFlow } from "./rag/query";
 
const app = express();
app.use(express.json({ limit: "10mb" }));
 
app.get("/health", (_req, res) => {
  res.json({ status: "ok", timestamp: new Date().toISOString() });
});
 
app.post("/api/summarize", async (req, res) => {
  try {
    const result = await summarizeFlow(req.body);
    res.json({ success: true, data: result });
  } catch (error) {
    const message = error instanceof Error ? error.message : "Unknown error";
    res.status(500).json({ success: false, error: message });
  }
});
 
app.post("/api/rag/query", async (req, res) => {
  try {
    const result = await ragQueryFlow(req.body);
    res.json({ success: true, data: result });
  } catch (error) {
    const message = error instanceof Error ? error.message : "Unknown error";
    res.status(500).json({ success: false, error: message });
  }
});
 
const PORT = process.env.PORT ?? 8080;
app.listen(PORT, () => console.log(`Server running on port ${PORT}`));

Deploying is a single command: firebase deploy --only hosting. App Hosting automatically builds the Cloud Run container and deploys it.

Development Workflow in Antigravity

Setting up custom rules for the project pays dividends when working in Antigravity.

# .antigravity/rules.md
## Genkit Flow Development Rules
- Always define Zod schemas for both input and output
- Use the withRetry utility for error handling
- Select models from the MODELS constant (don't hardcode model strings)
- Always use try-finally in streaming flows

With these rules in place, the Antigravity agent automatically applies these patterns when generating new flows. Asking it to "write a new flow" produces complete code including type definitions and error handling.

Cost Management

Genkit makes it easy to assign different models to different flows, which is the most effective lever for controlling costs. My approach: conversational flows that need fast responses use gemini-2.0-flash; batch processing that demands high accuracy uses gemini-2.5-pro. Designing with cost in mind from the start also makes future model swaps much less disruptive.

Wrapping Up

Firebase Genkit reduces both development and operational costs compared to calling the Gemini API directly—specifically by hiding the complexity of working with LLMs while delivering the observability and type safety that production requires. Antigravity's autocomplete, which accurately infers flow type information, further speeds up initial setup.

The first thing to try is the local development UI (genkit start). Running your flows from a browser and watching LLM inputs, outputs, and traces in real time tends to dissolve the assumption that "debugging AI features is painful."

Thank You for Reading

Antigravity Lab is ad-free, supported entirely by members like you. We publish practical guides daily with implementation code, benchmarks, and production-ready patterns. If you've found it useful, we'd love to have you on board.