パフォーマンス向上の鍵：LangChain・Redis・QStashで実現するオンデマンドLLMキャッシング

アプリケーションによっては、ユーザーからのリクエストのたびにLLMへ問い合わせることが現実的でないケースがあります。プロンプトをAPIに送信し、レスポンスを待つ一連の処理はどうしても時間がかかりがちです。PDFや音声ファイルからデータを抽出してLLMに渡すといった、LangChainを使った複雑なタスクになると、この遅延がユーザーエクスペリエンスに与える影響はさらに深刻になります。

ストリーミングも多くの場面で有効な解決策ですが、ユーザーがアプリケーションにアクセスする前の時点で処理を完了させておく方が都合が良いこともあります。そうすることで、キャッシュ済みのレスポンスをほぼ瞬時にユーザーへ提供でき、LLMが生成を終えるのを待たせる必要がなくなります。これは、書籍やニュースの要約ツールのように、全ユーザーに対して同じ入力を使うアプリケーションで特に効果を発揮します。では、ユーザーからの入力がない状態で、どうすればLangChainにレスポンスをキャッシュさせられるのでしょうか?

答えは、オンデマンドで呼び出せるマイクロサービスです。このマイクロサービスは、プロンプトをLangChainへ送り、Upstash Redisを使ってレスポンスをキャッシュできるものでなければなりません。あわせて、Upstashのレート制限SDKを利用して、うっかりレート制限を超えてしまわないよう対策します。QStashはマイクロサービスを呼び出すための非常に柔軟な手段です。好きなタイミングでレスポンスのキャッシュを指示でき、アプリケーションの要件に応じてcronジョブを設定することも可能です。さらに、マイクロサービスの利用状況を監視できるダッシュボードも備えています。万が一エンドポイントが過負荷になったり、何らかの障害が発生したりしても、QStashがHTTPリクエストを自動的に再試行し、メッセージが確実に配信されることを保証してくれます。

マイクロサービス本体は、エッジ向けの軽量かつ高速なWebフレームワーク「Hono.js」で構築します。このデモではホスティングにCloudflare Workerを採用しますが、Upstashのおかげで、他のエッジ/サーバーレスランタイムを含む virtually あらゆる環境へデプロイ可能です。

このデモの完全なソースコードはこちらから入手できます。

前提条件

• Upstash Redisデータベース
• QStashの環境変数
• OpenAI APIキー

はじめに

プロジェクトの作成

他の多くのcreate-<package>系npmパッケージとは異なり、create-honoは空のディレクトリ内で実行する必要があります。まず、プロジェクト用の新しい空ディレクトリを作成し、そこへ移動しましょう。

mkdir langchain-qstash
cd langchain-qstash

v1.0.0のリリース以降、本プロジェクトのスキャフォールディングにはBunを使用します。ただし、npm、pnpm、yarnでもcreate-honoを利用可能です。テンプレートの選択を求められたら、必ずcloudflare-workersを選んでください。bunが独立した選択肢として表示されることがありますが、これは本デモで使うテンプレートではないので注意しましょう。

bun create hono@latest

依存パッケージのインストール

続いて、お好みのパッケージマネージャーで次のコマンドを実行し、残りの依存パッケージをインストールします。

bun install @upstash/qstash @upstash/ratelimit @upstash/redis langchain openai

プロジェクトの設定

執筆時点では、create-honoはデフォルトでロックファイルを.gitignoreに含めています。厳密には必須ではありませんが、以下のように.gitignoreを更新してロックファイルを除外しておくと良いでしょう。

node_modules
dist
.wrangler
.dev.vars
wrangler.toml

次に、前提条件として挙げた環境変数を設定します。これらは、すでにソース管理から除外されているはずのwrangler.tomlに追記できます。

[vars]
QSTASH_CURRENT_SIGNING_KEY="sig_********"
QSTASH_NEXT_SIGNING_KEY="sig_********"
UPSTASH_REDIS_REST_URL="https://********.upstash.io"
UPSTASH_REDIS_REST_TOKEN="********"
OPENAI_API_KEY="sk-********"

最後に、src/index.tsを修正して、先ほどの環境変数に対応する型定義を追加します。

type Bindings = {
 QSTASH_CURRENT_SIGNING_KEY: string;
 QSTASH_NEXT_SIGNING_KEY: string;
 UPSTASH_REDIS_REST_URL: string;
 UPSTASH_REDIS_REST_TOKEN: string;
 OPENAI_API_KEY: string;
};
 
const app = new Hono<{ Bindings: Bindings }>();

署名鍵を2種類持つメリットは、一度のローテーションで環境変数を更新せずに済む点にあります。現在の署名鍵での検証に失敗した場合、QStashが自動的に次の署名鍵を試してくれるためです。

開発環境の起動

ここまで進めれば、wranglerが環境変数を読み取り、プロジェクトをデプロイできる状態になっています。お好みのパッケージマネージャーで以下のコマンドを実行してください。

bun run dev

wranglerがHono.jsサーバーを起動し、ポート8787でローカルURLを提供してくれるので、そこでプロジェクトをテストできます。ローカルではなく直接エッジのプレビューセッションを開始したい場合は、package.jsonのdevスクリプトを次のように変更します。

"dev": "wrangler dev src/index.ts --remote",

Cloudflare Workerの設定内容やログを確認するには、まずWorkerをデプロイする必要があります。

bun run deploy

初回実行時にはCloudflareへのサインイン手順が案内され、認証完了後に自動的にプロジェクトがデプロイされます。

ミドルウェアの作成

デバッグのためには、QStashから受け取るリクエストをログに出力できると便利です。ログはCloudflare Workerのダッシュボード「Logs」タブで、Begin log streamを押すと確認できます。Hono.jsにはロガーミドルウェアが用意されており、ルーターに簡単に組み込めます。

import { Hono } from "hono";
import { logger } from "hono/logger";
 
// snip
 
const app = new Hono<{ Bindings: Bindings }>();
app.use("*", logger());
 
// snip

QStashとCloudflare Workersの接続

Hono.jsでAPIエンドポイントの開発を始める前に、QStashが送信するメッセージを傍受し、無効な署名を持つリクエストを破棄する仕組みが必要です。幸いHono.jsにはカスタムミドルウェアを実装する仕組みがあり、ミドルウェアは常にハンドラーより先に実行されます。さらに柔軟性が高く、必要に応じてミドルウェアを個別のファイルに整理することもできます。

このミドルウェアでは、QStashのReceiver(レシーバー)を活用します。新しくsrc/middleware/verify.tsというファイルを作成し、MiddlewareHandler型の関数をエクスポートしましょう。

import { Receiver } from "@upstash/qstash";
import { type MiddlewareHandler } from "hono";
 
declare global {
 interface Response {
 locals: {
 query: string;
 };
 }
}
 
export const verify: MiddlewareHandler = async (ctx, next) => {
 const receiver = new Receiver({
 currentSigningKey: ctx.env.QSTASH_CURRENT_SIGNING_KEY,
 nextSigningKey: ctx.env.QSTASH_NEXT_SIGNING_KEY,
 });
};

Hono.jsは各ミドルウェアとハンドラーにctx(コンテキスト)オブジェクトを渡します。これはCloudflare WorkersのExecutionContextそのものではありませんが、同等の情報を含んでいます。Honoのctxオブジェクトは、標準的なCloudflare Workersのfetchハンドラーに渡されるリクエストオブジェクト・環境・実行コンテキストを、1つのオブジェクトにまとめたものとほぼ等価です。

また、グローバルなResponseインターフェースを拡張して、独自のlocalsを追加しています。Expressと違い、Honoはデフォルトでres.localsオブジェクトを作らないためです。後ほどハンドラーへクエリを引き渡す際にこれを使用します。続いて、先ほど型定義した環境変数にアクセスし、Receiverを構築します。

それでは、Receiverを使ってリクエストの署名を検証してみましょう。

// snip
 
const body = await ctx.req.text();
ctx.res.locals = {
 query: JSON.parse(body).query,
};
 
const isValid = await receiver
 .verify({
 signature: ctx.req.headers.get("Upstash-Signature")!,
 body,
 })
 .catch((err) => {
 console.error(err);
 return false;
 });
 
if (!isValid) {
 return new Response("Invalid signature", { status: 401 });
}
 
await next();

まず、コンテキストからリクエストオブジェクトを取得します。Hono.jsはfetch、URL、Request、ResponseといったWeb標準APIのみを使用しているのが特徴です。このため、本デモではCloudflare Workers上で動作させていますが、エッジ/サーバーレス環境を含む数多くのランタイムで実行できるのです。

リクエストオブジェクトからは、本文をテキストとして読み取るとともに、独自のJWTを格納したUpstash-Signatureヘッダーを取得します。Receiverに署名とリクエスト本文を渡すことで、JWTに含まれる署名を検証できます。.catchハンドラーでは、エラーをログ出力するとともにfalseを返し、署名が無効であることを示すようにしています。

ここでリクエスト本文を消費してしまうため、後続の実際のハンドラーでは本文にアクセスできなくなる点に注意してください。bodyは一度しか読み取れないReadableStreamだからです。代わりに、クエリをハンドラーへ渡すために、レスポンスのlocalsオブジェクトに格納しています。最後に、署名が無効な場合は401 Unauthorizedレスポンスを返し、有効ならnext()を呼び出して次のミドルウェアまたはハンドラーへ処理を引き継ぎます。

レート制限の追加

このエンドポイントは自分自身のOpenAI APIキーと接続するため、うっかりレート制限を超えないようにすることが重要です。幸い、Upstashはレート制限SDKを提供しており、これを使えばエンドポイントへ容易にレート制限を導入できます。リクエスト数が閾値を超えた場合は、429 Too Many Requestsレスポンスを返すことにします。

まず、src/middleware/ratelimit.tsに新しいミドルウェアを作成しましょう。

import { Ratelimit } from "@upstash/ratelimit";
import { Redis } from "@upstash/redis/cloudflare";
import { type MiddlewareHandler } from "hono";
 
export const ratelimit: MiddlewareHandler = async (ctx, next) => {
 const redis = new Redis({
 url: ctx.env.UPSTASH_REDIS_REST_URL,
 token: ctx.env.UPSTASH_REDIS_REST_TOKEN,
 });
 
 const ratelimit = new Ratelimit({
 redis,
 limiter: Ratelimit.slidingWindow(10, "10 s"),
 analytics: true,
 });
 
 await next();
};

ここでは、Upstash Redisデータベースをレート制限SDKに接続しています。Cloudflare Workers上ではRedis.fromEnv()が環境変数を自動的に読み取れないため、ctx.envの環境変数を使ってRedisインスタンスを手動で生成する必要があります。analyticsをtrueにすると、SDKは識別子ごとの呼び出し履歴をキャッシュするために自動的にRedisを呼び出します。デフォルトのプレフィックスは@upstash/ratelimitです。

SDKは、Redisの代わりにMapベースの一時キャッシュ(ephemeral cache)を使用することもサポートしており、極端な高負荷状況下で時間とリソースを節約できます。

const cache = new Map(); // ミドルウェアハンドラーの外側で定義
 
// snip
 
const ratelimit = new Ratelimit({
 ephemeralCache: cache,
 // snip
});

さらにSDKにはlimiterを指定します。これはリクエストをどのように制限するかをSDKに伝える関数です。ここではslidingWindowリミッターを使用し、10秒間につき最大10リクエストまで許可するよう設定しています。

SDKでは各リクエストに識別子を渡すこともでき、一定期間内のリクエスト数を自動的に追跡してくれます。実運用ではユーザーのIPアドレスを識別子に使えますが、このデモでは定数文字列を使い、すべてのリクエストを単一のレート制限対象としています。

// snip
 
const identifier = "openai";
const { success } = await ratelimit.limit(identifier);
 
if (!success) {
 return new Response("Too many requests", { status: 429 });
}
 
await next();
// snip

レート制限に達したリクエストには429 Too Many Requestsレスポンスを返します。そうでなければnext()を呼び出し、次のミドルウェアまたはハンドラーへ処理を続けます。なお、複数のRedisデータベースをSDKに接続した場合、それらの間で同期処理が発生します。Vercel EdgeやCloudflare Workersでは、これが未解決のまま残るプロミス(dangling promises)の原因となるため、次のように対処します。

const { pending, success } = await ratelimit.limit(identifier);
 
ctx.event.waitUntil(pending);

この対処方法は使用するライブラリによって異なります。HonoはWeb標準APIを使用しているため、event.waitUntilメソッドでプロミスの解決を待機できます。ただし本デモでは単一のRedisデータベースのみを使用するため、dangling promisesを気にする必要はありません。

QStashからのメッセージ受信

QStashへHTTPリクエストを送る際、宛先として指定するのはCloudflare WorkersエンドポイントのURLです。Hono.jsを使えば、このエンドポイント用のハンドラーを作成できます。src/index.tsに新しいハンドラーを追加し、先ほど作成したミドルウェアを有効化しましょう。

// snip
import { ratelimit } from "./middleware/ratelimit";
import { verify } from "./middleware/verify";
 
// snip
 
app.post("/api/announce", ratelimit, verify, async (ctx) => {});
 
// snip

このハンドラー内で、LangChainを使って与えられたプロンプトへの応答を生成し、Upstash Redisで結果をキャッシュします。

// snip
 
import { Redis } from "@upstash/redis/cloudflare";
import { UpstashRedisCache } from "langchain/cache/upstash_redis";
import { OpenAI } from "langchain/llms/openai";
 
// snip
 
app.post("/api/announce", ratelimit, verify, async (ctx) => {
 const redis = new Redis({
 url: ctx.env.UPSTASH_REDIS_REST_URL,
 token: ctx.env.UPSTASH_REDIS_REST_TOKEN,
 });
 
 const cache = new UpstashRedisCache({ client: redis });
 const model = new OpenAI({
 cache,
 openAIApiKey: ctx.env.OPENAI_API_KEY,
 });
 
 const query = ctx.res.locals.query;
 const result = await model
 .call(query)
 .then((result) => {
 console.log(result);
 return result;
 })
 .catch((err) => console.error(err));
 
 return new Response(result ?? "", { status: 200 });
});
 
// snip

ここでは、LangChainのキャッシュ設定に必要なクラス群をインポートしています。その後、OpenAIモデルとUpstash Redisキャッシュを使ってチェーンを構築し、Upstash Redisインスタンスをキャッシュへ渡します。前述の通り、Cloudflare Workers上ではAPIキーを自動読み込みできないため、OpenAIモデルにも手動でAPIキーを渡しています。

続いて、res.localsに保存しておいたクエリにアクセスし、それを使ってモデルを呼び出して結果をログ出力します。最後に、結果をレスポンスとして返します。何らかの問題が発生した場合は空文字列を返すようにしています。この段階で、POSTリクエストを送信してレスポンスを確認すれば、エンドポイントのテストが可能です。まず、package.jsonのdeployスクリプトを再実行します。

bun run deploy

Cloudflare WorkersエンドポイントのURLを取得したら、curlでPOSTリクエストを送ってみましょう。

curl -XPOST \
 "https://qstash.upstash.io/v2/publish/https://<YOUR_API_URL>.workers.dev/api/announce" \
 -H "Authorization: Bearer <YOUR_QSTASH_TOKEN>" \
 -H "Content-Type: application/json" \
 -d "{ \"query\": \"What's the derivative of e^x?\" }"

Upstashはこうしたリクエストをより手軽に送信できるQStashコンソールを提供しています。また、QStashにcronジョブを設定してリクエストを繰り返し実行させるのも同様に簡単です。QStashはリクエスト本文をそのままエンドポイントへ渡します。今回は次のようなJSONペイロードを送信しており、queryがモデルへのプロンプトとなります。

{
 "query": "What's the derivative of e^x?"
}

まとめ

レート制限SDKは、識別子ごとの呼び出し回数を正常にキャッシュできています。

同様に、LangChainが生成したコンテンツもUpstash Redisデータベースへ問題なくキャッシュされています。

そして最後に、レスポンスがCloudflare Workerのログにも記録されていることが確認できます。