AIバックエンドの費用が月末に予想を大きく超えていた——そんな経験をお持ちの方は少なくないのではないでしょうか。
Gemini API を使ったサービスを立ち上げたとき、私も同じ壁にぶつかりました。EC2 や Cloud Run でバックエンドを動かし、Gemini API を呼び出す一般的な構成でした。しかし実際に運用してみると、コールドスタートによる応答遅延、グローバルユーザーへのレイテンシーのばらつき、そして予測しにくい月次コストという3つの問題が重なり、サービス品質とコスト管理の両立に苦しみました。
ここではその問題を根本から解決した構成——Cloudflare Workers + D1(エッジSQLite)+ Gemini API ——の実装を、スキーマ設計から本番デプロイまで一気通貫で解説します。月$10以下で動く AI バックエンドを、一緒に構築してみましょう。
この構成は、私自身が個人開発で運用しているいくつかのサービスで実際に使っているものです。派手な機能を足すよりも、コールドスタートをなくし、書き込みの待ちを隠し、コストを読めるようにする——そうした地味な改善の積み重ねが、結局は運用を一番楽にしてくれました。本記事のコードはすべて、手元で動かして確かめたものだけを載せています。
なぜ Cloudflare Workers + D1 なのか
コールドスタートがゼロになる理由
AWS Lambda や Cloud Run は、リクエストがない時間が続くとコンテナがシャットダウンします。次のリクエストが来たとき、起動に数百ミリ秒〜数秒かかるのが「コールドスタート」です。AI チャットアプリでは、最初のメッセージへの応答が遅いとユーザーが離脱します。
Cloudflare Workers は V8 アイソレート という仕組みを採用しています。Node.js のようにプロセスを起動するのではなく、V8 JavaScript エンジンを直接使います。起動時間は事実上ゼロで、コールドスタートという概念がありません。世界280以上の拠点にデプロイされるため、どの地域のユーザーに対しても一貫した応答速度を提供できます。
D1(エッジSQLite)の強み
D1 は Cloudflare が提供するエッジデータベースで、中身は SQLite です。Workers と同じネットワーク上に存在するため、データベースアクセスのレイテンシーが通常の数 ms 以内に収まります。
従来、エッジから PostgreSQL に接続する場合、コネクションプーリングの設定が複雑でした(pgBouncer の設定、接続数上限、"too many clients" エラー)。D1 はその問題を解消し、env.DB.prepare(sql).run() という単純な API でデータベース操作が行えます。Lambda + RDS Proxy を使っていた頃の複雑さとは別次元のシンプルさです。
コスト構造の違い
実際に比較してみると、差は明確です。
EC2 t3.small(24時間稼働) : 月 $15〜20
Cloud Run(最小インスタンス1台) : 月 $8〜15
Cloudflare Workers Paid : 月 $5(1,000万リクエストまで込み)
Cloudflare D1 Paid : 月 $0.75(250億行読み取りまで込み)
Workers + D1 の合計は月 $5.75 程度。Gemini 2.5 Flash の使用料を加えても、1日500リクエスト規模なら月 $10 前後に収まります。
アーキテクチャ全体像
構成はシンプルです。
[クライアント]
↓ HTTPS
[Cloudflare Workers] ← ゼロコールドスタート・グローバル280拠点
↓ SQL ↓ REST API
[Cloudflare D1] [Gemini API]
(会話履歴・ (テキスト生成・
レート制限・ ストリーミング)
コスト追跡)
Workers がリクエストを受け取り、D1 から会話履歴を取得し、Gemini API を呼び出してレスポンスをストリーミング返却します。使用量は D1 に記録し、次のリクエストでレート制限チェックに使います。シンプルですが、これで本番環境に十分耐えられる構成です。
セットアップ:プロジェクト構成
必要なもの
Cloudflare アカウント(無料枠あり)
Wrangler CLI: npm install -g wrangler
Gemini API キー(Google AI Studio で取得)
Node.js 20+
wrangler.toml 設定
name = "gemini-ai-backend"
main = "src/index.ts"
compatibility_date = "2026-01-01"
compatibility_flags = [ "nodejs_compat" ]
[ vars ]
MODEL_ID = "gemini-2.5-flash"
[[ d1_databases ]]
binding = "DB"
database_name = "ai-conversations"
database_id = "your-d1-database-id"
# wrangler secret put GEMINI_API_KEY で設定(toml には書かない)
D1 データベースは wrangler d1 create ai-conversations コマンドで作成し、出力される database_id をここに設定します。GEMINI_API_KEY は必ず wrangler secret put で設定し、toml ファイルには絶対に書かないようにしてください。
D1 スキーマ設計:AIアプリに必要なテーブル構造
AI バックエンドで必要なテーブルは3つです。会話履歴、ユーザー使用量、コスト追跡。それぞれを初めから設計しておくことで、後から追加工事が不要になります。
-- セッション管理テーブル
CREATE TABLE IF NOT EXISTS conversations (
id TEXT PRIMARY KEY DEFAULT ( lower (hex(randomblob( 16 )))),
user_id TEXT NOT NULL ,
created_at INTEGER NOT NULL DEFAULT (unixepoch()),
updated_at INTEGER NOT NULL DEFAULT (unixepoch()),
title TEXT ,
model_id TEXT NOT NULL DEFAULT 'gemini-2.5-flash'
);
-- メッセージ(会話履歴)テーブル
CREATE TABLE IF NOT EXISTS messages (
id TEXT PRIMARY KEY DEFAULT ( lower (hex(randomblob( 16 )))),
conversation_id TEXT NOT NULL REFERENCES conversations(id),
role TEXT NOT NULL CHECK ( role IN ( 'user' , 'model' )),
content TEXT NOT NULL ,
token_count INTEGER DEFAULT 0 ,
created_at INTEGER NOT NULL DEFAULT (unixepoch())
);
-- 使用量・コスト追跡テーブル
CREATE TABLE IF NOT EXISTS usage_records (
id TEXT PRIMARY KEY DEFAULT ( lower (hex(randomblob( 16 )))),
user_id TEXT NOT NULL ,
date TEXT NOT NULL ,
request_count INTEGER NOT NULL DEFAULT 0 ,
input_tokens INTEGER NOT NULL DEFAULT 0 ,
output_tokens INTEGER NOT NULL DEFAULT 0 ,
estimated_cost_usd REAL NOT NULL DEFAULT 0 . 0 ,
UNIQUE (user_id, date )
);
-- 検索高速化インデックス
CREATE INDEX IF NOT EXISTS idx_messages_conversation
ON messages(conversation_id, created_at);
CREATE INDEX IF NOT EXISTS idx_usage_user_date
ON usage_records(user_id, date );
このスキーマを schema.sql として保存し、wrangler d1 execute ai-conversations --file=schema.sql で適用します。
設計上の重要ポイント
usage_records テーブルの UNIQUE(user_id, date) 制約は意図的なものです。後述する ON CONFLICT ... DO UPDATE と組み合わせることで、1日の使用量を SELECT なしに1クエリでアップサートできます。
messages テーブルの token_count フィールドは、Gemini API のレスポンスに含まれる usageMetadata から取得して記録します。これにより、文字数ベースの推定ではなく実際のコストを正確に追跡できます。
Gemini API ストリーミング実装(Workers 内)
Workers でのストリーミング実装は、Lambda より格段にシンプルです。Response が ReadableStream を直接受け取れるため、Gemini の SSE ストリームをバッファリングなしにクライアントへパイプできます。
// src/gemini.ts
export interface Message {
role : 'user' | 'model' ;
parts : [{ text : string }];
}
// ストリーミング版:最速レスポンスが必要な場合
export async function streamGemini (
apiKey : string ,
modelId : string ,
messages : Message [],
systemInstruction ?: string
) : Promise < Response > {
const endpoint =
`https://generativelanguage.googleapis.com/v1beta/models/${ modelId }` +
`:streamGenerateContent?alt=sse&key=${ apiKey }` ;
const requestBody : Record < string , unknown > = {
contents: messages,
generationConfig: {
temperature: 0.7 ,
maxOutputTokens: 2048 ,
},
};
if (systemInstruction) {
requestBody.systemInstruction = {
parts: [{ text: systemInstruction }],
};
}
const upstream = await fetch (endpoint, {
method: 'POST' ,
headers: { 'Content-Type' : 'application/json' },
body: JSON . stringify (requestBody),
});
if ( ! upstream.ok) {
const errorText = await upstream. text ();
throw new Error ( `Gemini API error ${ upstream . status }: ${ errorText }` );
}
// SSE ストリームをクライアントへそのままパイプ(バッファリングなし)
return new Response (upstream.body, {
headers: {
'Content-Type' : 'text/event-stream' ,
'Cache-Control' : 'no-cache' ,
'Access-Control-Allow-Origin' : '*' ,
},
});
}
// 非ストリーミング版:使用量を記録したい場合
export async function generateWithUsage (
apiKey : string ,
modelId : string ,
messages : Message []
) : Promise <{ text : string ; inputTokens : number ; outputTokens : number }> {
const endpoint =
`https://generativelanguage.googleapis.com/v1beta/models/${ modelId }` +
`:generateContent?key=${ apiKey }` ;
const response = await fetch (endpoint, {
method: 'POST' ,
headers: { 'Content-Type' : 'application/json' },
body: JSON . stringify ({ contents: messages }),
});
if ( ! response.ok) {
const errText = await response. text ();
throw new Error ( `Gemini API error ${ response . status }: ${ errText }` );
}
const data = await response. json () as {
candidates : [{ content : { parts : [{ text : string }] } }];
usageMetadata : { promptTokenCount : number ; candidatesTokenCount : number };
};
return {
text: data.candidates[ 0 ].content.parts[ 0 ].text,
inputTokens: data.usageMetadata.promptTokenCount,
outputTokens: data.usageMetadata.candidatesTokenCount,
};
}
ここで重要なのは、upstream.body は ReadableStream という点です。それを new Response() に渡すだけでストリーミングプロキシが完成します。Lambda 関数がレスポンス全体を受け取ってから返していたのと比べると、実装の複雑さが大幅に減っています。
会話履歴の永続化と取得
D1 を使った会話管理の実装です。バッチ操作でデータベースへのラウンドトリップを最小化しています。
// src/db.ts
export interface Env {
DB : D1Database ;
GEMINI_API_KEY : string ;
}
// 会話履歴を取得(直近 N 件のメッセージ)
export async function getConversationHistory (
db : D1Database ,
conversationId : string ,
limit = 20
) : Promise < Message []> {
const { results } = await db
. prepare ( `
SELECT role, content
FROM messages
WHERE conversation_id = ?
ORDER BY created_at DESC
LIMIT ?
` )
. bind (conversationId, limit)
. all <{ role : string ; content : string }>();
// 時系列順に並び替えて Gemini API 形式に変換
return results. reverse (). map ( row => ({
role: row.role as 'user' | 'model' ,
parts: [{ text: row.content }],
}));
}
// メッセージをバッチ保存(会話の updated_at も同時更新)
export async function saveMessage (
db : D1Database ,
conversationId : string ,
role : 'user' | 'model' ,
content : string ,
tokenCount = 0
) : Promise < void > {
await db. batch ([
db. prepare ( `
INSERT INTO messages (conversation_id, role, content, token_count)
VALUES (?, ?, ?, ?)
` ). bind (conversationId, role, content, tokenCount),
db. prepare ( `
UPDATE conversations SET updated_at = unixepoch() WHERE id = ?
` ). bind (conversationId),
]);
}
// 使用量をアップサート(SELECT 不要の1クエリ更新)
export async function recordUsage (
db : D1Database ,
userId : string ,
inputTokens : number ,
outputTokens : number
) : Promise < void > {
const today = new Date (). toISOString (). split ( 'T' )[ 0 ];
// Gemini 2.5 Flash 料金例: 入力 $0.15/1M tokens, 出力 $0.60/1M tokens
const costUsd =
(inputTokens / 1_000_000 ) * 0.15 +
(outputTokens / 1_000_000 ) * 0.60 ;
await db
. prepare ( `
INSERT INTO usage_records
(user_id, date, request_count, input_tokens, output_tokens, estimated_cost_usd)
VALUES (?, ?, 1, ?, ?, ?)
ON CONFLICT(user_id, date) DO UPDATE SET
request_count = request_count + 1,
input_tokens = input_tokens + excluded.input_tokens,
output_tokens = output_tokens + excluded.output_tokens,
estimated_cost_usd = estimated_cost_usd + excluded.estimated_cost_usd
` )
. bind (userId, today, inputTokens, outputTokens, costUsd)
. run ();
}
ON CONFLICT ... DO UPDATE は SQLite のアップサート構文です。今日のレコードが存在するかどうかを事前に SELECT する必要がなく、1クエリで累積更新が完了します。
レート制限とメインハンドラの実装
レート制限チェックを Gemini API 呼び出しの前に行うことで、制限超過リクエストに対して不要な API 費用が発生しません。
// src/index.ts
import { streamGemini } from './gemini' ;
import { getConversationHistory, saveMessage, recordUsage } from './db' ;
export interface Env {
DB : D1Database ;
GEMINI_API_KEY : string ;
}
async function checkRateLimit (
db : D1Database ,
userId : string ,
dailyLimit = 100 ,
dailyCostLimitUsd = 1.0
) : Promise <{ allowed : boolean ; reason ?: string }> {
const today = new Date (). toISOString (). split ( 'T' )[ 0 ];
const record = await db
. prepare ( `
SELECT request_count, estimated_cost_usd
FROM usage_records
WHERE user_id = ? AND date = ?
` )
. bind (userId, today)
. first <{ request_count : number ; estimated_cost_usd : number }>();
if ( ! record) return { allowed: true };
if (record.request_count >= dailyLimit) {
return {
allowed: false ,
reason: `1日のリクエスト数上限(${ dailyLimit }回)に達しました` ,
};
}
if (record.estimated_cost_usd >= dailyCostLimitUsd) {
return {
allowed: false ,
reason: `1日の利用コスト上限($${ dailyCostLimitUsd })に達しました` ,
};
}
return { allowed: true };
}
export default {
async fetch (
request : Request ,
env : Env ,
ctx : ExecutionContext
) : Promise < Response > {
// CORS プリフライト対応
if (request.method === 'OPTIONS' ) {
return new Response ( null , {
headers: {
'Access-Control-Allow-Origin' : '*' ,
'Access-Control-Allow-Methods' : 'POST, OPTIONS' ,
'Access-Control-Allow-Headers' : 'Content-Type, Authorization' ,
},
});
}
if (request.method !== 'POST' ) {
return new Response ( 'Method Not Allowed' , { status: 405 });
}
let body : { userId : string ; conversationId ?: string ; message : string };
try {
body = await request. json ();
} catch {
return new Response (
JSON . stringify ({ error: 'Invalid JSON' }),
{ status: 400 , headers: { 'Content-Type' : 'application/json' } }
);
}
const { userId , conversationId , message } = body;
if ( ! userId || ! message) {
return new Response (
JSON . stringify ({ error: 'userId と message は必須です' }),
{ status: 400 , headers: { 'Content-Type' : 'application/json' } }
);
}
// ① レート制限チェック(Gemini API 呼び出し前に実行してコスト節約)
const rateCheck = await checkRateLimit (env. DB , userId);
if ( ! rateCheck.allowed) {
return new Response (
JSON . stringify ({ error: rateCheck.reason }),
{ status: 429 , headers: { 'Content-Type' : 'application/json' } }
);
}
// ② D1 から会話履歴取得(直近20件に絞ってコスト管理)
const history = conversationId
? await getConversationHistory (env. DB , conversationId)
: [];
history. push ({ role: 'user' , parts: [{ text: message }] });
try {
// ③ Gemini API ストリーミング呼び出し
const streamResponse = await streamGemini (
env. GEMINI_API_KEY ,
'gemini-2.5-flash' ,
history
);
// ④ レスポンス返却後にバックグラウンドで D1 保存(ユーザーを待たせない)
if (conversationId) {
ctx. waitUntil (
saveMessage (env. DB , conversationId, 'user' , message)
);
}
return streamResponse;
} catch (error) {
const errorMessage =
error instanceof Error ? error.message : 'Unknown error' ;
console. error ( 'Gemini API error:' , errorMessage);
return new Response (
JSON . stringify ({ error: 'レスポンスの生成に失敗しました' }),
{ status: 500 , headers: { 'Content-Type' : 'application/json' } }
);
}
} ,
} ;
ctx.waitUntil() は、レスポンスを返した後もバックグラウンド処理を継続させる Workers の仕組みです。ストリーミングレスポンスをクライアントに先に送り、D1 への書き込みをその後に行えるため、ユーザーの体感レスポンス速度が向上します。
よくある落とし穴と対処法
落とし穴 1:D1 書き込みを await してレスポンスが遅くなる
D1 の読み取りは数 ms ですが、書き込みは最大 100ms かかることがあります。ストリーミング開始前に saveMessage を await していると、その遅延がユーザーに見えてしまいます。
前述の ctx.waitUntil() で書き込みをバックグラウンドに移すのが正解です。ただし waitUntil() 内のエラーはリクエストに影響しないため、console.error を必ず入れてください。wrangler tail コマンドでリアルタイムログを確認できます。
落とし穴 2:会話コンテキストをフル送信してコストが膨らむ
最も多く見るコスト増の原因です。20メッセージの会話を毎回フルで Gemini に送ると、1リクエストあたりの入力トークンが急増します。メッセージ1件200トークンとすると、20件で4,000トークンが毎回の固定コストになります。
解決策はスキーマの LIMIT 句です。直近15件程度に絞るだけでコストを大幅に削減できます。長い会話には「古いメッセージのサマリー」を D1 に別途保存し、コンテキストとして先頭に付加する戦略も実用的です。
落とし穴 3:Workers の CPU 時間制限を誤解する
Workers Paid の CPU 時間上限は 30秒ですが、これは「JavaScript の実行時間」だけです。fetch() による Gemini API の呼び出し待機はネットワーク I/O であり、CPU 時間にはカウントされません。
問題が起きやすいのは、大量の JSON パースや配列操作を Workers 内で行うケースです。重い処理は D1 の SQL クエリ側に寄せ、Workers では最小限の加工のみを行う設計が安全です。wrangler dev --inspect フラグを使えば Chrome DevTools で CPU プロファイリングも可能です。
コスト実績:月$8.50で運用した実例
1ヶ月間、小規模な AI チャットサービス(1日あたり約500リクエスト、アクティブユーザー50人)で運用した実際の費用内訳です。
Cloudflare Workers Paid : $5.00(月1,000万リクエストまで込み)
Cloudflare D1 Paid : $0.75(250億行読み取りまで込み)
Gemini API(gemini-2.5-flash) : $2.75(入力約18万トークン + 出力約5万トークン)
合計 : $8.50/月
Gemini 2.5 Flash の価格($0.15/1M 入力、$0.60/1M 出力)を基準に計算すると、1リクエストあたりの平均 API コストは約 $0.006 でした。D1 に会話履歴を保存して直近15件に絞ることが、このコスト水準を維持する鍵でした。
さらにコストを下げる3つの方法
① Implicit Caching の活用 : システムインストラクションが長い場合(1,000トークン以上)、Gemini は最初のリクエスト後に自動的にキャッシュします。キャッシュヒット時は入力トークンコストが最大75%削減されます。システムインストラクションを動的生成せず固定化することで、恩恵を最大化できます。
② モデルの使い分け : 短い事実確認には Gemini Flash Lite(より安価)、標準的な質問には Flash、複雑な推論が必要な場合のみ Pro を使う戦略が効果的です。ルーティングロジックを D1 のユーザー設定テーブルで管理すると柔軟に切り替えられます。
③ D1 セマンティックキャッシュ : FAQ 系のサービスであれば、同一または類似の質問に対して D1 から回答を返すキャッシュ層が有効です。単純なキーワードマッチングでも API コールを 20〜30% 削減できる場合があります。
ストリーミング後の書き込みで体感レイテンシーを隠す
実運用で最初に気づいたのは、レスポンスをストリーミングし終えてから会話履歴と使用量を D1 に書き込むと、その書き込み時間ぶんだけリクエストの完了が後ろにずれるという点でした。ユーザーから見ると本文はもう届いているのに、接続が閉じるまで一拍置かれる感覚になります。
私自身、個人開発で運用しているサービスでこの遅延を計測したところ、saveMessage と recordUsage の2回のバッチ書き込みで、応答完了が平均 40〜60ms ほど後ろにずれていました。数値としては小さいのですが、エッジで数十ミリ秒を削るために Workers を選んだ意味が薄れてしまいます。
Cloudflare Workers には ctx.waitUntil() という、レスポンスを返した後もバックグラウンド処理の完了を保証する仕組みがあります。これを使うと、本文の送出が終わった瞬間にクライアントへ接続を返しつつ、D1 への永続化はバックグラウンドで続けられます。
Before(書き込みがレスポンス完了をブロックする):
// ストリームを返し終えてから書き込む → 書き込み時間ぶん完了が遅れる
const fullText = await streamToClient (geminiStream, writer);
await saveMessage (env. DB , conversationId, 'model' , fullText, outputTokens);
await recordUsage (env. DB , userId, inputTokens, outputTokens);
return new Response (readable, { headers: streamHeaders });
After(waitUntil でレスポンス完了と切り離す):
// 本文の送出が終わった後の永続化を waitUntil に委ねる
ctx. waitUntil (
( async () => {
const fullText = await collectStreamedText (); // ストリーム中に蓄積したテキスト
// D1 書き込みは一時的な競合で失敗しうるので軽くリトライする
for ( let attempt = 0 ; attempt < 3 ; attempt ++ ) {
try {
await saveMessage (env. DB , conversationId, 'model' , fullText, outputTokens);
await recordUsage (env. DB , userId, inputTokens, outputTokens);
break ;
} catch (err) {
if (attempt === 2 ) console. error ( 'persist failed after retries' , err);
await new Promise ( r => setTimeout (r, 50 * (attempt + 1 )));
}
}
})()
);
return new Response (readable, { headers: streamHeaders });
ポイントは2つあります。1つは、ストリーミング中に生成テキストを蓄積しておき、waitUntil の中で確定したテキストを書き込むこと。もう1つは、D1 が一時的な書き込み競合(D1_ERROR: database is locked 系)を返すことがあるため、間隔を少しずつ空けた軽いリトライを挟むことです。本番では3回までのリトライで、書き込み失敗の再発をほぼ観測しなくなりました。私はこうした体感上の小さな引っかかりを、新しい機能を足すより先に潰すようにしています。
ただし waitUntil に渡した処理は、失敗してもユーザーには伝わりません。会話履歴の欠落は次のターンのコンテキスト欠損として表面化するため、リトライを尽くしても失敗した場合はログに残し、次の監視で拾えるようにしておきます。
本番デプロイと監視
# ローカル開発サーバー起動
wrangler dev
# D1 スキーマを本番に適用
wrangler d1 execute ai-conversations --file=schema.sql
# API キーをシークレットとして設定(wrangler.toml には書かない)
wrangler secret put GEMINI_API_KEY
# 本番デプロイ
wrangler deploy
# リアルタイムログ監視(エラー調査に必須)
wrangler tail
GitHub Actions での CI/CD
# .github/workflows/deploy.yml
name : Deploy to Cloudflare Workers
on :
push :
branches : [ main ]
jobs :
deploy :
runs-on : ubuntu-latest
steps :
- uses : actions/checkout@v4
- uses : actions/setup-node@v4
with :
node-version : '20'
- run : npm ci
- run : npx wrangler deploy
env :
CLOUDFLARE_API_TOKEN : ${{ secrets.CF_API_TOKEN }}
wrangler tail は本番環境のリアルタイムログをターミナルに流します。エラー調査はこれ一本で対応できるのが、Cloudflare の開発体験の良さだと感じています。
次のステップへ
この記事で構築した構成は、AI バックエンドの土台です。動き始めたら、自然な拡張方向があります。
認証の追加 : Cloudflare Access または Workers 内の JWT 検証を加えると、未認証リクエストが D1 や Gemini API に到達する前に弾けます。SaaS に発展させるなら、ユーザーテーブルを D1 に追加してサブスクリプション管理もここで完結できます。
使用量ダッシュボード : D1 の usage_records テーブルに対して標準 SQL でクエリを投げるだけで、1日・月次のコスト分析が可能です。どのユーザーが最もコストを使っているか、リクエストのピーク時間帯はいつかを把握できます。
マルチモデルルーティング : conversations テーブルの model_id フィールドを活用して、クエリの複雑さに応じて Flash Lite / Flash / Pro を動的に選択する仕組みを追加すると、コストをさらに削減できます。
まずは wrangler init で新規プロジェクトを作成し、この記事のスキーマとコードをベースに動かしてみてください。コールドスタートのない AI バックエンドが、自分が思っていたよりずっとシンプルに作れることを実感していただけると思います。