Gemini API × Inngest で耐障害性の高いAIワークフローを構築する本番運用ガイド

// src/inngest/functions/digest-pipeline.ts
import { inngest } from "../client";
import { GoogleGenAI } from "@google/genai";
const ai = new GoogleGenAI({ apiKey: process.env.GEMINI_API_KEY! });
 
// 子関数: 1記事の要約
const summarizeOne = inngest.createFunction(
  {
    id: "summarize-one",
    // Geminiのレート制限に合わせて全体スループットを制御
    throttle: { limit: 60, period: "1m", key: "event.data.modelTier" },
    retries: 3,
  },
  { event: "digest/summarize.requested" },
  async ({ event, step }) => {
    const { url, modelTier } = event.data;
    const html = await step.run("fetch", async () => {
      const r = await fetch(url, { signal: AbortSignal.timeout(20_000) });
      return r.text();
    });
    const summary = await step.run("summarize", async () => {
      const res = await ai.models.generateContent({
        model: modelTier === "pro" ? "gemini-2.5-pro" : "gemini-2.5-flash",
        contents: `次のHTMLから要点を3行で抽出してください:\n${html.slice(0, 50_000)}`,
      });
      return res.text ?? "";
    });
    return { url, summary };
  },
);
 
// 親関数: 100件をfan-outしてfan-in
const buildDigest = inngest.createFunction(
  { id: "build-digest", retries: 2 },
  { event: "digest/build.requested" },
  async ({ event, step }) => {
    const { urls } = event.data as { urls: string[] };
 
    // fan-out: 子関数を並列起動して結果を待つ
    const summaries = await Promise.all(
      urls.map((url, i) =>
        step.invoke(`summarize-${i}`, {
          function: summarizeOne,
          data: { url, modelTier: "flash" },
        }),
      ),
    );
 
    // fan-in: まとめてダイジェスト生成
    const digest = await step.run("compose-digest", async () => {
      const joined = summaries.map((s) => `- ${s.summary}`).join("\n");
      const res = await ai.models.generateContent({
        model: "gemini-2.5-pro",
        contents: `以下の要約からニュースダイジェストを書いてください:\n${joined}`,
      });
      return res.text ?? "";
    });
 
    await step.run("publish", async () => {
      await fetch(`${process.env.API_BASE_URL}/digest`, {
        method: "POST",
        headers: { "content-type": "application/json" },
        body: JSON.stringify({ digest }),
      });
    });
 
    return { count: summaries.length };
  },
);
 
export { summarizeOne, buildDigest };

// src/inngest/functions/article-with-approval.ts
import { inngest } from "../client";
 
export const articleWithApproval = inngest.createFunction(
  { id: "article-with-approval", retries: 2 },
  { event: "article/generate.requested" },
  async ({ event, step }) => {
    // ... outline / body 生成は前述と同じ ...
    const draftId: string = await step.run("save-draft", async () => {
      // ドラフトを保存して draftId を返す
      return crypto.randomUUID();
    });
 
    // 担当者にレビュー依頼通知（Slack等）
    await step.run("notify-reviewer", async () => {
      await fetch(`${process.env.SLACK_WEBHOOK}`, {
        method: "POST",
        body: JSON.stringify({ text: `レビュー待ち: /draft/${draftId}` }),
      });
    });
 
    // 最大3日間、承認イベントを待つ
    const approval = await step.waitForEvent("wait-approval", {
      event: "article/approval.received",
      timeout: "3d",
      // 同じdraftIdに対する承認イベントだけを受け取る
      if: `async.data.draftId == "${draftId}"`,
    });
 
    if (!approval) {
      // タイムアウト: 自動却下
      await step.run("auto-reject", async () => {
        await fetch(`${process.env.API_BASE_URL}/draft/${draftId}/reject`, {
          method: "POST",
        });
      });
      return { status: "auto-rejected", draftId };
    }
 
    // 承認された: 公開処理
    await step.run("publish", async () => {
      await fetch(`${process.env.API_BASE_URL}/draft/${draftId}/publish`, {
        method: "POST",
      });
    });
    return { status: "published", draftId };
  },
);

const safeAgent = inngest.createFunction(
  {
    id: "safe-agent",
    // concurrency: 同時実行数の上限（同時に走る関数インスタンス数）
    concurrency: [
      { scope: "fn", limit: 50 }, // 関数全体で50並列まで
      { scope: "fn", key: "event.data.userId", limit: 1 }, // ユーザーごとに1並列
    ],
    // throttle: 起動レートの上限（バーストを許容しない）
    throttle: { limit: 30, period: "1m" },
    // rateLimit: 重複イベントを単純に捨てる（厳しめ）
    rateLimit: { limit: 1, period: "10s", key: "event.data.userId" },
    retries: 3,
  },
  { event: "agent/run.requested" },
  async ({ event, step }) => {
    // 月次トークン予算をチェック
    const used = await step.run("check-budget", async () => {
      const r = await fetch(
        `${process.env.API_BASE_URL}/usage/${event.data.userId}`,
      );
      return (await r.json()) as { tokensUsedThisMonth: number };
    });
 
    if (used.tokensUsedThisMonth > 5_000_000) {
      // 予算オーバー: ここで明示的に止める
      return { status: "budget-exceeded", used: used.tokensUsedThisMonth };
    }
 
    // ... 以下、Gemini呼び出し ...
  },
);

import { NonRetriableError } from "inngest";
 
const robustGeneration = inngest.createFunction(
  {
    id: "robust-generation",
    retries: 5,
    onFailure: async ({ event, error, step }) => {
      // すべてのretryを使い切った後に呼ばれる
      await step.run("notify-deadletter", async () => {
        await fetch(`${process.env.OPS_WEBHOOK}`, {
          method: "POST",
          body: JSON.stringify({
            severity: "high",
            functionId: "robust-generation",
            originalEvent: event,
            error: error.message,
          }),
        });
      });
    },
  },
  { event: "article/generate.requested" },
  async ({ event, step }) => {
    try {
      // ... Gemini処理 ...
    } catch (e: unknown) {
      // 4xx は retry してもムダ。即座に dead-letter へ
      if (e instanceof Error && e.message.includes("INVALID_ARGUMENT")) {
        throw new NonRetriableError("client error", { cause: e });
      }
      throw e;
    }
  },
);

// ❌ 間違い: 再実行時、キャッシュからtimestampが返るので「現在時刻」にならない
await step.run("save", async () => {
  await db.save({ value, timestamp: Date.now() });
});
 
// ✅ 正しい: timestampをstep.runの戻り値として固定
const timestamp = await step.run("get-timestamp", async () => Date.now());
await step.run("save", async () => {
  await db.save({ value, timestamp });
});

const persist = await step.run("persist", async () => {
  await db.execute(
    `INSERT INTO articles (request_id, topic, body)
     VALUES ($1, $2, $3)
     ON CONFLICT (request_id) DO NOTHING`,
    [requestId, topic, body],
  );
});
 
// Stripe API への冪等性キー付き呼び出し
const charge = await step.run("charge", async () => {
  return stripe.charges.create(
    { amount: 250, currency: "jpy", source: "tok_visa" },
    { idempotencyKey: requestId },
  );
});