genkit_ops

Caching, terminal diagnostics, redaction, and operational middleware for Genkit Dart generate calls.

genkit_ops is for teams building Genkit Dart apps that are moving from experiments to real product workflows. It helps answer operational questions that show up quickly in AI apps:

• Did this request call the model or return from cache?
• Why did my local AI feature feel slow?
• How many prompt/completion tokens did this response report?
• Can I inspect AI calls in the terminal without leaking secrets?
• Can repeated prompts avoid repeated model cost?

It complements Genkit's Developer UI and tracing tools. It is not a model provider, vector database, retry replacement, or agent tool library. Its job is repeat-call cost control and safe local diagnostics around generate() calls.

How it differs from genkit_middleware

genkit_middleware and genkit_ops solve different problems in the same Genkit middleware ecosystem.

genkit_middleware is about agent capabilities: filesystem access, reusable skills, and tool approval. Use it when the model needs to read files, load instructions, or request human approval before running sensitive tools.

genkit_ops is about operations: caching repeated model calls, seeing cache hit/miss status, printing terminal diagnostics, redacting secrets, and tracking latency/token usage while you build and debug.

They can be used together. A coding assistant might use genkit_middleware for filesystem and approval workflows, while genkit_ops handles repeat-call cost control and safe logs around the same generate() calls.

Features

• Async CacheStore<T> abstraction.
• TTL-aware InMemoryCacheStore<T> with LRU-style max entry eviction.
• Deterministic cache keys from Genkit ModelRequest payloads.
• Genkit GenerateMiddleware for model response caching.
• Pretty terminal logger with token usage, latency, cache hit/miss, and errors.
• Redaction-first logging defaults.

Install

dependencies:
  genkit_ops: ^0.1.2

Usage with Genkit

Register the plugin, then attach middleware refs to generate.

import 'package:genkit/genkit.dart';
import 'package:genkit_ops/genkit_ops.dart';

final ai = Genkit(
  plugins: [
    GenkitOpsPlugin(
      cacheStore: InMemoryCacheStore<ModelResponse>(
        defaultTtl: const Duration(minutes: 10),
        maxEntries: 500,
      ),
    ),
  ],
);

final response = await ai.generate(
  prompt: 'Summarize the release notes.',
  use: genkitOps(
    policy: const CachePolicy(
      namespace: 'docs',
      keyVersion: 'v1',
      ttl: Duration(minutes: 10),
    ),
    logger: const GenkitOpsLogger(
      level: GenkitOpsLogLevel.info,
    ),
  ),
);

print(response.text);

Real-world scenario: customer support replies

A support app often receives repeated questions such as refund policy, password reset steps, delivery tracking, or subscription cancellation. Without caching, each repeated question can trigger another paid model request. With genkit_ops, the same prompt/model/config response can be returned from memory for a short period while the terminal still shows whether the response came from cache or the model.

The example below keeps support answers isolated by namespace, invalidates old answers when policy text changes with keyVersion, includes selected context in the cache key, and keeps raw prompt/output hidden in terminal logs by default.

import 'package:genkit/genkit.dart';
import 'package:genkit_ops/genkit_ops.dart';

final supportCache = InMemoryCacheStore<ModelResponse>(
  defaultTtl: const Duration(minutes: 30),
  maxEntries: 1000,
);

final ai = Genkit(
  plugins: [
    // Register your model provider plugin here.
    // Example: googleAI(),
    GenkitOpsPlugin(cacheStore: supportCache),
  ],
);

Future<String> answerSupportTicket({
  required String customerQuestion,
  required String planName,
  required String locale,
  required String tenantId,
}) async {
  final response = await ai.generate(
    prompt: '''
You are a concise support assistant for an ecommerce app.
Customer plan: $planName
Locale: $locale
Question: $customerQuestion

Answer with clear next steps and avoid inventing policy details.
''',
    context: {
      'locale': locale,
      'tenantId': tenantId,
    },
    use: genkitOps(
      policy: CachePolicy(
        namespace: 'support-replies-$tenantId',
        keyVersion: 'policy-2026-01',
        ttl: const Duration(minutes: 30),
        contextKeys: const ['locale', 'tenantId'],
      ),
      logger: const GenkitOpsLogger(
        level: GenkitOpsLogLevel.info,
      ),
    ),
  );

  return response.text;
}

For a complete copyable version of this pattern, see example/main.dart.

Direct Middleware

For APIs that accept middleware instances directly:

final cache = GenkitSmartCacheMiddleware(
  store: InMemoryCacheStore<ModelResponse>(),
  policy: const CachePolicy(ttl: Duration(minutes: 5)),
);

final logger = GenkitAILoggerMiddleware(
  logger: const GenkitOpsLogger(level: GenkitOpsLogLevel.verbose),
);

Safe Defaults

Tool-enabled and streaming model requests are not cached by default because they may depend on external state or partial output. Raw prompt/output payloads are not printed unless verbose logging and includePayloads are explicitly enabled.

Roadmap

• Disk and database stores will live in companion packages.
• Cost estimation hooks will be added without hard-coding provider pricing.
• Error diagnostics will stay separate from retry behavior; use Genkit's retry middleware for retries.