◈ API / SDK/2026-06-25Advanced

Gemini API × TypeScript Type-Safe AI Application Architecture — Integrating Zod Schemas, Structured Output, and Streaming

Type-safe AI applications with the Gemini API and TypeScript: Zod validation, Structured Output, streaming pipelines, and error handling that holds up in production.

gemini-api²⁷⁷ typescript¹⁵ zod³ structured-output²² streaming²⁸ type-safety production¹⁴⁰

✦ Premium Article

It Compiles. The Data Is Still Wrong.

LLM responses are inherently non-deterministic. Even with identical prompts, the returned structure can vary subtly — unexpected fields appear, required values go missing, or types shift between runs. This unpredictability is the root cause of the most insidious bugs in AI application development.

TypeScript's type system catches many errors at compile time, but it's powerless against data whose structure is only determined at runtime — which is exactly what LLM responses are. This is where the combination of Zod runtime validation and Gemini API's Structured Output creates what we call "dual type safety."

The design patterns below aim at type safety at compile time and at runtime alike, using the Gemini API with TypeScript — schema design first, then the production edge cases that only surface under real traffic, with working code throughout.

In my own work as an indie developer, I run pipelines that automate several blogs, and the most painful incidents were always the ones that passed every type check yet carried the wrong content. The structure was valid while the substance quietly degraded — the kind of error that both the compiler and the validator wave straight through. That experience is why I treat type safety as a starting point, not a finish line.

Prerequisites and Setup

To run the code examples in this article, you'll need:

Node.js 20 or later
TypeScript 5.4 or later
Gemini API key (obtain from Google AI Studio)

Here's how to set up your project:

# Initialize the project
mkdir gemini-typesafe-app && cd gemini-typesafe-app
npm init -y
npm install @google/genai zod typescript tsx
npm install -D @types/node
 
# Generate tsconfig.json
npx tsc --init --strict --target ES2022 --module NodeNext --moduleResolution NodeNext

@google/genai is the official TypeScript SDK for the Gemini API, version 1.x.

// src/config.ts
import { GoogleGenAI } from "@google/genai";
 
// Load API key from environment (never hardcode)
const apiKey = process.env.GEMINI_API_KEY;
if (!apiKey) {
  throw new Error("GEMINI_API_KEY environment variable is required");
}
 
export const ai = new GoogleGenAI({ apiKey });
export const DEFAULT_MODEL = "gemini-2.5-flash";

✦

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

✦Detecting semantically wrong yet schema-passing output in production using Zod invariants

✦Separating parse_fail from invariant_fail and watching the ratio to catch silent post-update degradation early

✦Wiring types, validation, and observability into one flow and adopting it incrementally without over-engineering

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 →

Defining the LLM Response "Contract" with Zod Schemas

Zod is a runtime validation library for TypeScript that can automatically infer TypeScript types from schema definitions. We declare the expected structure of LLM responses as Zod schemas, then use them for both compile-time type checking and runtime validation.

// src/schemas/article.ts
import { z } from "zod";
 
// Article analysis result schema
export const ArticleAnalysisSchema = z.object({
  title: z.string().min(1).max(200).describe("Article title"),
  summary: z.string().min(50).max(500).describe("Article summary (50-500 characters)"),
  sentiment: z.enum(["positive", "negative", "neutral"]).describe("Overall sentiment"),
  topics: z.array(z.string()).min(1).max(10).describe("Key topics (1-10 items)"),
  readingTimeMinutes: z.number().int().positive().describe("Estimated reading time in minutes"),
  keyEntities: z.array(
    z.object({
      name: z.string().describe("Named entity"),
      type: z.enum(["person", "organization", "technology", "location"]).describe("Entity type"),
      relevance: z.number().min(0).max(1).describe("Relevance score (0-1)"),
    })
  ).max(20).describe("Key entities mentioned in the article"),
});
 
// Automatically infer TypeScript type from schema
export type ArticleAnalysis = z.infer<typeof ArticleAnalysisSchema>;
 
// This type is equivalent to:
// {
//   title: string;
//   summary: string;
//   sentiment: "positive" | "negative" | "neutral";
//   topics: string[];
//   readingTimeMinutes: number;
//   keyEntities: { name: string; type: "person" | ...; relevance: number; }[];
// }

The key detail here is the use of z.describe() on each field. These descriptions flow through to the JSON Schema passed to Gemini API's Structured Output, helping the model produce more accurate structures.

Building the Zod → JSON Schema Conversion Layer

Gemini API's Structured Output expects a JSON Schema format to define response structure. Since you can't pass Zod schemas directly, a conversion layer is needed.

// src/utils/schema-converter.ts
import { z } from "zod";
import { Type } from "@google/genai";
 
/**
 * Convert a Zod schema to Gemini API-compatible JSON Schema.
 * Gemini API supports a subset of JSON Schema Draft 2020-12.
 */
export function zodToGeminiSchema(schema: z.ZodType): Record<string, unknown> {
  if (schema instanceof z.ZodObject) {
    const shape = schema.shape;
    const properties: Record<string, unknown> = {};
    const required: string[] = [];
 
    for (const [key, value] of Object.entries(shape)) {
      properties[key] = zodToGeminiSchema(value as z.ZodType);
      if (!(value instanceof z.ZodOptional)) {
        required.push(key);
      }
    }
 
    return {
      type: Type.OBJECT,
      properties,
      required,
    };
  }
 
  if (schema instanceof z.ZodString) {
    const result: Record<string, unknown> = { type: Type.STRING };
    if (schema.description) result.description = schema.description;
    return result;
  }
 
  if (schema instanceof z.ZodNumber) {
    const result: Record<string, unknown> = { type: Type.NUMBER };
    if (schema.description) result.description = schema.description;
    return result;
  }
 
  if (schema instanceof z.ZodBoolean) {
    return { type: Type.BOOLEAN };
  }
 
  if (schema instanceof z.ZodEnum) {
    return {
      type: Type.STRING,
      enum: schema.options,
      description: schema.description,
    };
  }
 
  if (schema instanceof z.ZodArray) {
    return {
      type: Type.ARRAY,
      items: zodToGeminiSchema(schema.element),
      description: schema.description,
    };
  }
 
  if (schema instanceof z.ZodOptional) {
    return { ...zodToGeminiSchema(schema.unwrap()), nullable: true };
  }
 
  // Fallback
  return { type: Type.STRING };
}

This conversion function means you define your Zod schema in one place, and three pieces of type information stay automatically synchronized: TypeScript types, runtime validation, and Gemini API schemas. This eliminates an entire category of bugs caused by schema drift.

Implementing a Type-Safe Gemini API Client

Now let's combine everything into a generic AI client that accepts type parameters.

// src/client/typesafe-gemini.ts
import { z } from "zod";
import { ai, DEFAULT_MODEL } from "../config";
import { zodToGeminiSchema } from "../utils/schema-converter";
 
interface GenerateOptions<T extends z.ZodType> {
  prompt: string;
  schema: T;
  model?: string;
  temperature?: number;
  systemInstruction?: string;
  maxRetries?: number;
}
 
interface GenerateResult<T> {
  data: T;
  usage: {
    promptTokens: number;
    completionTokens: number;
    totalTokens: number;
  };
  latencyMs: number;
}
 
/**
 * Type-safe Gemini API request.
 * - Automatically generates Structured Output config from Zod schema
 * - Validates response before returning
 * - Built-in retry logic
 */
export async function generateTyped<T extends z.ZodType>(
  options: GenerateOptions<T>
): Promise<GenerateResult<z.infer<T>>> {
  const {
    prompt,
    schema,
    model = DEFAULT_MODEL,
    temperature = 0.7,
    systemInstruction,
    maxRetries = 3,
  } = options;
 
  const geminiSchema = zodToGeminiSchema(schema);
  const startTime = Date.now();
 
  let lastError: Error | null = null;
 
  for (let attempt = 1; attempt <= maxRetries; attempt++) {
    try {
      const response = await ai.models.generateContent({
        model,
        contents: [{ role: "user", parts: [{ text: prompt }] }],
        config: {
          temperature,
          responseMimeType: "application/json",
          responseSchema: geminiSchema,
          ...(systemInstruction && { systemInstruction }),
        },
      });
 
      const text = response.text;
      if (!text) {
        throw new Error("Empty response from Gemini API");
      }
 
      // Parse JSON
      const rawData = JSON.parse(text);
 
      // Zod validation (runtime type check)
      const validated = schema.parse(rawData);
 
      const latencyMs = Date.now() - startTime;
      const usage = response.usageMetadata;
 
      return {
        data: validated,
        usage: {
          promptTokens: usage?.promptTokenCount ?? 0,
          completionTokens: usage?.candidatesTokenCount ?? 0,
          totalTokens: usage?.totalTokenCount ?? 0,
        },
        latencyMs,
      };
    } catch (error) {
      lastError = error instanceof Error ? error : new Error(String(error));
 
      // Validation errors may improve on retry
      if (error instanceof z.ZodError && attempt < maxRetries) {
        console.warn(
          `Validation failed (attempt ${attempt}/${maxRetries}):`,
          error.issues.map((i) => `${i.path.join(".")}: ${i.message}`)
        );
        continue;
      }
 
      // Rate limit errors use exponential backoff
      if (isRateLimitError(error) && attempt < maxRetries) {
        const delay = Math.min(1000 * Math.pow(2, attempt), 30000);
        console.warn(`Rate limited. Retrying in ${delay}ms...`);
        await sleep(delay);
        continue;
      }
 
      throw lastError;
    }
  }
 
  throw lastError ?? new Error("All retry attempts exhausted");
}
 
function isRateLimitError(error: unknown): boolean {
  if (error instanceof Error) {
    return error.message.includes("429") || error.message.includes("RESOURCE_EXHAUSTED");
  }
  return false;
}
 
function sleep(ms: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, ms));
}

Here's what using it looks like:

// src/examples/analyze-article.ts
import { generateTyped } from "../client/typesafe-gemini";
import { ArticleAnalysisSchema, type ArticleAnalysis } from "../schemas/article";
 
async function analyzeArticle(articleText: string): Promise<ArticleAnalysis> {
  const result = await generateTyped({
    prompt: `Analyze the following article:\n\n${articleText}`,
    schema: ArticleAnalysisSchema,
    temperature: 0.3, // Lower temperature for analysis tasks
    systemInstruction: "You are a content analysis expert. Be precise and concise.",
  });
 
  // result.data is fully type-inferred as ArticleAnalysis
  console.log(`Analysis complete: ${result.data.title}`);
  console.log(`Token usage: ${result.usage.totalTokens}`);
  console.log(`Latency: ${result.latencyMs}ms`);
 
  return result.data;
}
 
// Expected output:
// Analysis complete: Deep Dive into Gemini 2.5 Pro's New Features
// Token usage: 1523
// Latency: 2340ms

At every call site, result.data is automatically inferred as the ArticleAnalysis type. IDE autocompletion works perfectly, and accessing non-existent fields produces compile-time errors.

Type-Safe Streaming Response Processing

Gemini API's streaming feature delivers LLM output in real-time chunks. However, during streaming, JSON arrives in an incomplete state, so you need partial parsing and progressive validation mechanisms.

// src/client/streaming.ts
import { z } from "zod";
import { ai, DEFAULT_MODEL } from "../config";
import { zodToGeminiSchema } from "../utils/schema-converter";
 
interface StreamOptions<T extends z.ZodType> {
  prompt: string;
  schema: T;
  model?: string;
  systemInstruction?: string;
  onPartialData?: (partial: string) => void;
}
 
/**
 * Type-safe streaming response processing.
 * - Receives chunks incrementally for real-time UI updates
 * - Final result is Zod-validated
 */
export async function generateTypedStream<T extends z.ZodType>(
  options: StreamOptions<T>
): Promise<z.infer<T>> {
  const {
    prompt,
    schema,
    model = DEFAULT_MODEL,
    systemInstruction,
    onPartialData,
  } = options;
 
  const geminiSchema = zodToGeminiSchema(schema);
  let accumulated = "";
 
  const response = await ai.models.generateContentStream({
    model,
    contents: [{ role: "user", parts: [{ text: prompt }] }],
    config: {
      responseMimeType: "application/json",
      responseSchema: geminiSchema,
      ...(systemInstruction && { systemInstruction }),
    },
  });
 
  for await (const chunk of response) {
    const text = chunk.text;
    if (text) {
      accumulated += text;
      onPartialData?.(accumulated);
    }
  }
 
  // Parse and validate after all chunks are received
  const rawData = JSON.parse(accumulated);
  return schema.parse(rawData);
}

Frontend Integration with Server-Sent Events

When relaying streaming responses to a frontend, Server-Sent Events (SSE) is the simplest approach. Here's an implementation using Next.js App Router:

// app/api/analyze/route.ts (Next.js Route Handler)
import { generateTypedStream } from "@/lib/streaming";
import { ArticleAnalysisSchema } from "@/schemas/article";
 
export async function POST(request: Request) {
  const { text } = await request.json();
 
  const encoder = new TextEncoder();
  const stream = new ReadableStream({
    async start(controller) {
      try {
        const result = await generateTypedStream({
          prompt: `Analyze the following text:\n\n${text}`,
          schema: ArticleAnalysisSchema,
          onPartialData: (partial) => {
            // Send partial data as SSE events
            controller.enqueue(
              encoder.encode(`data: ${JSON.stringify({ partial })}\n\n`)
            );
          },
        });
 
        // Send final validated result
        controller.enqueue(
          encoder.encode(`data: ${JSON.stringify({ result, done: true })}\n\n`)
        );
        controller.close();
      } catch (error) {
        controller.enqueue(
          encoder.encode(`data: ${JSON.stringify({ error: "Analysis failed" })}\n\n`)
        );
        controller.close();
      }
    },
  });
 
  return new Response(stream, {
    headers: {
      "Content-Type": "text/event-stream",
      "Cache-Control": "no-cache",
      Connection: "keep-alive",
    },
  });
}

Layered Error Handling Architecture

In production, errors occur at multiple layers. A hierarchical error handling design with appropriate strategies at each layer is essential.

// src/errors/ai-errors.ts
/**
 * AI application error hierarchy.
 * Uses TypeScript discriminated unions for type-safe handling.
 */
export class AIError extends Error {
  constructor(
    message: string,
    public readonly code: AIErrorCode,
    public readonly retryable: boolean,
    public readonly context?: Record<string, unknown>
  ) {
    super(message);
    this.name = "AIError";
  }
}
 
export type AIErrorCode =
  | "SCHEMA_VALIDATION_FAILED"
  | "JSON_PARSE_FAILED"
  | "EMPTY_RESPONSE"
  | "RATE_LIMIT_EXCEEDED"
  | "QUOTA_EXCEEDED"
  | "MODEL_OVERLOADED"
  | "CONTENT_FILTERED"
  | "NETWORK_ERROR"
  | "TIMEOUT"
  | "UNKNOWN";
 
/**
 * Classify raw Gemini API errors into structured AIErrors.
 */
export function classifyError(error: unknown): AIError {
  if (error instanceof AIError) return error;
 
  const message = error instanceof Error ? error.message : String(error);
 
  if (message.includes("429") || message.includes("RESOURCE_EXHAUSTED")) {
    return new AIError(message, "RATE_LIMIT_EXCEEDED", true);
  }
 
  if (message.includes("quota") || message.includes("QUOTA")) {
    return new AIError(message, "QUOTA_EXCEEDED", false);
  }
 
  if (message.includes("SAFETY") || message.includes("blocked")) {
    return new AIError(message, "CONTENT_FILTERED", false);
  }
 
  if (message.includes("503") || message.includes("overloaded")) {
    return new AIError(message, "MODEL_OVERLOADED", true);
  }
 
  if (message.includes("timeout") || message.includes("DEADLINE_EXCEEDED")) {
    return new AIError(message, "TIMEOUT", true);
  }
 
  return new AIError(message, "UNKNOWN", false);
}

Eliminating Exceptions with the Result Type Pattern

Inspired by Go and Rust, the Result type pattern enables type-safe error handling without exceptions.

// src/utils/result.ts
export type Result<T, E = AIError> =
  | { success: true; data: T }
  | { success: false; error: E };
 
export function ok<T>(data: T): Result<T, never> {
  return { success: true, data };
}
 
export function err<E>(error: E): Result<never, E> {
  return { success: false, error };
}

// src/client/safe-generate.ts
import { z } from "zod";
import { generateTyped } from "./typesafe-gemini";
import { classifyError, AIError } from "../errors/ai-errors";
import { Result, ok, err } from "../utils/result";
 
/**
 * Exception-free, type-safe API call.
 * Callers handle errors explicitly through the Result type.
 */
export async function safeGenerate<T extends z.ZodType>(
  options: Parameters<typeof generateTyped<T>>[0]
): Promise<Result<z.infer<T>, AIError>> {
  try {
    const result = await generateTyped(options);
    return ok(result.data);
  } catch (error) {
    return err(classifyError(error));
  }
}
 
// Usage example
async function processArticle(text: string) {
  const result = await safeGenerate({
    prompt: `Analyze this article: ${text}`,
    schema: ArticleAnalysisSchema,
  });
 
  if (!result.success) {
    // error is inferred as AIError
    if (result.error.retryable) {
      console.log("Retryable error:", result.error.code);
      // Queue for later retry
    } else {
      console.error("Unrecoverable error:", result.error.code);
      // Send alert
    }
    return;
  }
 
  // data is inferred as ArticleAnalysis
  console.log(result.data.title);
  console.log(result.data.topics);
}

The beauty of this design is that it forces error handling. You can't access result.data without first checking result.success, so missing error handling is caught at compile time.

Schema Versioning and Backward Compatibility

In production environments, you'll need to evolve AI output schemas incrementally. A schema versioning strategy that synchronizes field changes with deployments is essential.

// src/schemas/versioned.ts
import { z } from "zod";
 
// v1: Initial schema
const ProductReviewV1 = z.object({
  productName: z.string(),
  rating: z.number().min(1).max(5),
  summary: z.string(),
});
 
// v2: Adding fields (backward compatible)
const ProductReviewV2 = ProductReviewV1.extend({
  pros: z.array(z.string()).default([]),
  cons: z.array(z.string()).default([]),
  recommendationScore: z.number().min(0).max(100).optional(),
});
 
// v3: Type change (requires migration function)
const ProductReviewV3 = z.object({
  productName: z.string(),
  rating: z.object({
    overall: z.number().min(1).max(5),
    categories: z.record(z.string(), z.number().min(1).max(5)),
  }),
  summary: z.string(),
  pros: z.array(z.string()),
  cons: z.array(z.string()),
  recommendationScore: z.number().min(0).max(100),
});
 
// Migration function
type V2Data = z.infer<typeof ProductReviewV2>;
type V3Data = z.infer<typeof ProductReviewV3>;
 
function migrateV2toV3(v2: V2Data): V3Data {
  return {
    productName: v2.productName,
    rating: {
      overall: v2.rating,
      categories: {},
    },
    summary: v2.summary,
    pros: v2.pros,
    cons: v2.cons,
    recommendationScore: v2.recommendationScore ?? v2.rating * 20,
  };
}
 
// Export current version
export const ProductReviewSchema = ProductReviewV3;
export type ProductReview = z.infer<typeof ProductReviewV3>;
export { migrateV2toV3 };

By leveraging .default() and .optional(), you can add new fields without breaking existing responses. When breaking changes are unavoidable, migration functions provide a path for gradual transitions.

Testing Strategy — Mocking LLM Responses and Property-Based Testing

Testing AI applications requires confronting the non-deterministic nature of LLMs. Here we cover both unit tests for validation logic and property-based tests for schema robustness.

// src/__tests__/schema-validation.test.ts
import { describe, it, expect } from "vitest";
import { ArticleAnalysisSchema } from "../schemas/article";
 
describe("ArticleAnalysisSchema", () => {
  it("accepts correctly structured data", () => {
    const validData = {
      title: "Test Article",
      summary: "This is a test article summary that needs to be at least fifty characters long for validation purposes.",
      sentiment: "positive",
      topics: ["AI", "TypeScript"],
      readingTimeMinutes: 5,
      keyEntities: [
        { name: "Google", type: "organization", relevance: 0.9 },
      ],
    };
 
    const result = ArticleAnalysisSchema.safeParse(validData);
    expect(result.success).toBe(true);
  });
 
  it("rejects invalid sentiment values", () => {
    const invalidData = {
      title: "Test Article",
      summary: "x".repeat(50),
      sentiment: "excited", // Outside enum values
      topics: ["AI"],
      readingTimeMinutes: 5,
      keyEntities: [],
    };
 
    const result = ArticleAnalysisSchema.safeParse(invalidData);
    expect(result.success).toBe(false);
    if (!result.success) {
      expect(result.error.issues[0].path).toContain("sentiment");
    }
  });
 
  it("rejects empty topics array", () => {
    const invalidData = {
      title: "Test Article",
      summary: "x".repeat(50),
      sentiment: "neutral",
      topics: [], // Violates min(1)
      readingTimeMinutes: 5,
      keyEntities: [],
    };
 
    const result = ArticleAnalysisSchema.safeParse(invalidData);
    expect(result.success).toBe(false);
  });
});

Verifying Schema Robustness with Property-Based Tests

// src/__tests__/property-test.ts
import { describe, it, expect } from "vitest";
import { z } from "zod";
import { zodToGeminiSchema } from "../utils/schema-converter";
import { ArticleAnalysisSchema } from "../schemas/article";
 
describe("zodToGeminiSchema conversion consistency", () => {
  it("data valid under Zod schema remains valid after conversion", () => {
    // 100 rounds of random data generation
    for (let i = 0; i < 100; i++) {
      const randomData = {
        title: `Title ${Math.random().toString(36).slice(2, 12)}`,
        summary: "a".repeat(50 + Math.floor(Math.random() * 450)),
        sentiment: ["positive", "negative", "neutral"][
          Math.floor(Math.random() * 3)
        ],
        topics: Array.from(
          { length: 1 + Math.floor(Math.random() * 5) },
          (_, j) => `topic-${j}`
        ),
        readingTimeMinutes: 1 + Math.floor(Math.random() * 30),
        keyEntities: [],
      };
 
      const zodResult = ArticleAnalysisSchema.safeParse(randomData);
      expect(zodResult.success).toBe(true);
    }
  });
 
  it("converted Gemini Schema contains correct required fields", () => {
    const geminiSchema = zodToGeminiSchema(ArticleAnalysisSchema);
    expect(geminiSchema).toHaveProperty("properties");
    expect(geminiSchema).toHaveProperty("required");
 
    const required = geminiSchema.required as string[];
    expect(required).toContain("title");
    expect(required).toContain("summary");
    expect(required).toContain("sentiment");
  });
});

Production Performance Optimization

Here are techniques for optimizing performance while maintaining type safety.

Context Caching Integration

When analyzing large document sets repeatedly, leveraging Gemini API's context caching can dramatically reduce both cost and latency.

// src/client/cached-generate.ts
import { z } from "zod";
import { ai } from "../config";
import { zodToGeminiSchema } from "../utils/schema-converter";
 
interface CachedGenerateOptions<T extends z.ZodType> {
  schema: T;
  systemInstruction: string;
  contextDocuments: string[];
  cacheTTLSeconds?: number;
}
 
/**
 * Type-safe generation with context caching.
 * Reduces cost by up to 90% for multiple queries against the same document set.
 */
export async function createCachedGenerator<T extends z.ZodType>(
  options: CachedGenerateOptions<T>
) {
  const { schema, systemInstruction, contextDocuments, cacheTTLSeconds = 3600 } = options;
 
  // Create cache
  const cache = await ai.caches.create({
    model: "gemini-2.5-flash",
    config: {
      systemInstruction,
      contents: contextDocuments.map((doc) => ({
        role: "user",
        parts: [{ text: doc }],
      })),
      ttl: `${cacheTTLSeconds}s`,
    },
  });
 
  const geminiSchema = zodToGeminiSchema(schema);
 
  // Return a query function that uses the cache
  return async function query(prompt: string): Promise<z.infer<T>> {
    const response = await ai.models.generateContent({
      model: "gemini-2.5-flash",
      contents: [{ role: "user", parts: [{ text: prompt }] }],
      config: {
        cachedContent: cache.name,
        responseMimeType: "application/json",
        responseSchema: geminiSchema,
      },
    });
 
    const rawData = JSON.parse(response.text ?? "{}");
    return schema.parse(rawData);
  };
}
 
// Usage: Query repeatedly against 100 documents
// const analyzer = await createCachedGenerator({
//   schema: ArticleAnalysisSchema,
//   systemInstruction: "Respond as a document analysis expert",
//   contextDocuments: documents,
//   cacheTTLSeconds: 7200,
// });
// const result1 = await analyzer("Extract the main topics");
// const result2 = await analyzer("Perform sentiment analysis");

Batch Processing Pipeline

A batch pipeline for processing large numbers of requests efficiently can also be built with full type safety.

// src/client/batch-pipeline.ts
import { z } from "zod";
import { safeGenerate } from "./safe-generate";
import { Result } from "../utils/result";
import { AIError } from "../errors/ai-errors";
 
interface BatchOptions<T extends z.ZodType> {
  items: string[];
  schema: T;
  promptTemplate: (item: string) => string;
  concurrency?: number;
  onProgress?: (completed: number, total: number) => void;
}
 
/**
 * Type-safe batch processing pipeline.
 * - Controls concurrency to avoid rate limits
 * - Individual failures don't halt the entire batch
 */
export async function batchGenerate<T extends z.ZodType>(
  options: BatchOptions<T>
): Promise<Result<z.infer<T>, AIError>[]> {
  const { items, schema, promptTemplate, concurrency = 5, onProgress } = options;
 
  const results: Result<z.infer<T>, AIError>[] = [];
  let completed = 0;
 
  // Semaphore for concurrency control
  const semaphore = new Semaphore(concurrency);
 
  const tasks = items.map(async (item) => {
    await semaphore.acquire();
    try {
      const result = await safeGenerate({
        prompt: promptTemplate(item),
        schema,
        temperature: 0.3,
      });
      results.push(result);
    } finally {
      semaphore.release();
      completed++;
      onProgress?.(completed, items.length);
    }
  });
 
  await Promise.all(tasks);
  return results;
}
 
// Simple semaphore implementation
class Semaphore {
  private queue: (() => void)[] = [];
  private current = 0;
 
  constructor(private max: number) {}
 
  acquire(): Promise<void> {
    if (this.current < this.max) {
      this.current++;
      return Promise.resolve();
    }
    return new Promise((resolve) => this.queue.push(resolve));
  }
 
  release(): void {
    this.current--;
    const next = this.queue.shift();
    if (next) {
      this.current++;
      next();
    }
  }
}

Observability — Catching Plausible-but-Wrong Output That Passes the Schema

Zod guarantees structure, not correctness. In unattended production pipelines, the most dangerous outputs are the ones that are structurally valid yet semantically wrong — a near-empty summary, a number off by an order of magnitude, a classification that contradicts the previous run. Because they parse cleanly, exception-based monitoring never sees them.

The fix is to promote the validation outcome itself into a measured signal. Beyond success and failure, re-check each schema-passing output against a few lightweight invariants, and watch the pass rate over time.

// src/observability/validation-metrics.ts
type ValidationOutcome = "ok" | "parse_fail" | "invariant_fail";
 
interface Counter { inc(labels: Record<string, string>): void; }
 
// An invariant asserts minimal content validity, not structure
type Invariant<T> = { name: string; holds: (value: T) => boolean };
 
export function recordValidation<T>(
  schemaName: string,
  parsed: { success: true; data: T } | { success: false },
  invariants: Invariant<T>[],
  counter: Counter,
): ValidationOutcome {
  if (!parsed.success) {
    counter.inc({ schema: schemaName, outcome: "parse_fail", invariant: "-" });
    return "parse_fail";
  }
  for (const inv of invariants) {
    if (!inv.holds(parsed.data)) {
      // Structurally valid but semantically broken — the face of silent decay
      counter.inc({ schema: schemaName, outcome: "invariant_fail", invariant: inv.name });
      return "invariant_fail";
    }
  }
  counter.inc({ schema: schemaName, outcome: "ok", invariant: "-" });
  return "ok";
}

Invariants distill domain knowledge into the smallest possible code. For a summarization task, anything under 5% of the source length is not a summary; for classification, reject anything below 0.4 confidence — the floors that types alone cannot express.

// Example invariants for a summary schema
const summaryInvariants: Invariant<{ summary: string; sourceLength: number }>[] = [
  {
    name: "summary_not_degenerate",
    holds: (v) => v.summary.length >= v.sourceLength * 0.05,
  },
  {
    name: "summary_within_bounds",
    holds: (v) => v.summary.length <= v.sourceLength,
  },
];

Running automated generation across several sites, I learned that the invariant_fail ratio is the single most useful early warning. parse_fail spikes in obvious moments like right after a deploy, but invariant_fail creeps up quietly after a model update or a small prompt tweak, with no one noticing. Reviewing that ratio weekly was enough to catch degradation before it became an incident.

Start by implementing only the reject path: record the invariant_fail and route it to a fallback — regenerate, default value, or human review. That alone gives you a foundation for observability without over-engineering.

Next Steps

The heart of type-safe AI application design is wiring three layers into one flow: centralize "schema -> types -> validation -> API spec" with Zod, handle failures without exceptions via the Result type, and observe the validation outcome itself so you catch semantic degradation too.

As a first step, define a Zod schema for just the single most important response you have today, and start recording its pass rate with recordValidation. Begin small; only once invariant_fail becomes visible should you add invariants one at a time. That order, I believe, is the shortest path to growing type safety without slipping into over-engineering.

Thank You for Reading

Gemini 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.