パフォーマンスの向上:LangChain、Redis、QStash を使用したオンデマンド LLM キャッシング

アプリケーションによっては、ユーザーが LLM を要求したときに LLM をクエリできない場合があります。 API にプロンプ​​トを送信してその応答を待つプロセスは、特に時間がかかることがあります。 PDF や音声ファイルのデータを処理して LLM に供給するなど、LangChain が関係するより複雑なタスクの場合、遅延はユーザー エクスペリエンスにさらに大きな打撃を与えます。

ストリーミング中に 多くの場合、適切なソリューションですが、ユーザーがアプリケーションにアクセスする前の任意の時点ですべてを実行する方が便利な場合があります。このようにして、ユーザーは、LLM が生成するのを待つことなく、キャッシュされた応答をほぼ瞬時に受け取ることができます。これは、書籍やニュースの要約など、すべてのユーザーに対して同じ入力を使用するアプリケーションに特に役立ちます。しかし、ユーザーからの入力がなければ、どうやって LangChain に応答をキャッシュさせることができるのでしょうか?

オンデマンドで呼び出すことができ、LangChain にプロンプトを送信し、Upstash Redis を使用して応答をキャッシュできるマイクロサービスが必要です。また、誤ってレート制限を超えないようにするために、Upstash のレート制限 SDK も使用します。 QStash は、マイクロサービスを呼び出すための最も汎用性の高い方法です。いつでも必要なときに応答をキャッシュするように設定でき、アプリケーションが必要に応じて cron ジョブを渡すこともできます。また、マイクロサービスの使用状況を監視するために使用できるダッシュボードも提供します。エンドポイントが過負荷になったり、何か問題が発生した場合、QStash は HTTP リクエストを再試行し、メッセージが確実に配信されるようにします。

私たちのマイクロサービスは、エッジ用の軽量で高速な Web フレームワークである Hono.js を使用して構築されます。このデモでは、Cloudflare Worker を使用してマイクロサービスをホストします。ただし、Upstash のおかげで、他のエッジ/サーバーレス ランタイムを含め、事実上どこにでもデプロイできます。

このデモの完全なソース コードはここで見つけることができます。

前提条件

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

はじめに

プロジェクトの作成

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

mkdir langchain-qstash
cd langchain-qstash

最近の v1.0.0 リリースでは、このプロジェクトの足場として Bun が使用されます。ただし、create-hono を使用することもできます。 npm を使用 、pnpm 、および yarn 。必ず 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 つの署名キーを持つことの利点は、環境変数を更新する必要なく、それらを 1 回ロールアウトできるという事実にあります。これは、現在の署名キーが失敗した場合、QStash が自動的に次の署名キーの使用を試行するためです。

開発

この段階では、wrangler 環境変数を読み取り、プロジェクトをデプロイできる必要があります。これを行うには、好みのパッケージ マネージャーで次のコマンドを実行します。

bun run dev

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

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

Cloudflare ワーカーの設定とログを表示するには、まずワーカーをデプロイする必要があります。

bun run deploy

これにより、Cloudflare にサインインし、認証後に初めてプロジェクトを自動的にデプロイする手順が説明されます。

ミドルウェアの作成

デバッグの目的で、QStash から受信したリクエストをログに記録すると便利です。 Begin log stream を押すと、これらは Cloudflare Worker のダッシュボードの [ログ] タブに表示されます。 。 Hono.js は、ルーターに追加できるロガー ミドルウェアを提供します。

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

QStash を Cloudflare ワーカーに接続する

Hono.js を使用して API エンドポイントの開発を開始する前に、QStash によって送信されたメッセージをインターセプトし、無効な署名が含まれている場合はリクエストを破棄する方法が必要です。ありがたいことに、Hono.js は、常にハンドラーの前に実行される独自のカスタム ミドルウェアを実装する方法を提供します。これは十分に堅牢であるため、必要に応じてミドルウェアを個別のファイルに整理できます。

私たちのミドルウェアでは、QStash のレシーバーを利用できます。 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 と直接同じではありません。 —しかし、同じ情報が含まれています。ほののctx オブジェクトは、デフォルトのCloudflare Workers fetchに渡されるリクエストオブジェクト、環境、および実行コンテキストとほぼ同等です。 ハンドラ、すべてが 1 つのオブジェクトに結合されます。

グローバル Response も変更します。 カスタム locals を含めるインターフェース 。 Express とは異なり、Hono は res.locals を作成しません。 デフォルトではオブジェクトです。後でこれを使用してクエリをハンドラーに渡します。次に、先ほどの型指定された環境変数にアクセスして、レシーバーを構築します。

これで、受信者を使用してリクエストの署名を検証できるようになりました。

// 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 などの Web 標準 API のみを便利に使用します。 、URL 、Request 、および Response 。このデモでは Cloudflare Workers で使用していますが、これにより、エッジ/サーバーレス環境を含む他の無数の環境でも実行できるようになります。

リクエスト オブジェクトから、リクエストの本文と Upstash-Signature をテキストとして読み取ります。 ヘッダーにはカスタム JWT が含まれます。レシーバーにリクエストの署名と本文を渡すことで、レシーバーを使用して JWT に含まれる署名を検証できます。 .catch 内 ハンドラーでは、エラーをログに記録し、false を返すようにします。 署名が無効であることを示します。

リクエストの本文を消費しているため、後で実際のハンドラーでそれにアクセスすることはできません。これは、本文が 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 に接続しています。 Redis を作成する必要があります。 ctx.env の環境変数を使用して手動でインスタンスを作成します 、Redis.fromEnv() のため Cloudflare Workers を使用する場合、それらを自動的に読み取ることができません。 analytics の場合 true の場合、SDK は Redis を自動的に呼び出して、各識別子の呼び出しのキャッシュを保持します。 @upstash/ratelimit を使用します デフォルトではプレフィックスとして使用されます。

SDK は、Map を使用する一時キャッシュの使用もサポートしています。 Redis の代わりに、極度の負荷の下で時間とリソースを節約できます。

const cache = new Map(); // outside of the middleware handler
 
// snip
 
const ratelimit = new Ratelimit({
 ephemeralCache: cache,
 // snip
});

limiter も提供します SDKに。これは、レート制限リクエストの方法を 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 に未解決の Promise が発生する可能性がありますが、これは次のように処理できます。

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

これを行う方法は、使用しているライブラリによって異なる場合があります。 Hono は Web 標準 API を使用しているため、event.waitUntil を使用できます。 Promise が解決されるまで待機するメソッド。ただし、このデモでは単一の Redis データベースのみを使用するため、未解決の Promise について心配する必要はありません。

QStash からのメッセージの受信

HTTP リクエストを QStash に送信するとき、指定する宛先は 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 インスタンスをキャッシュに渡します。繰り返しになりますが、OpenAI API キーは Cloudflare Workers で自動的に読み取ることができないため、手動でモデルに渡す必要があります。

次に、以前に res.locals に保存したクエリにアクセスします。 、それを使用してモデルを呼び出し、結果を記録します。最後に、結果を応答として返すか、何か問題があった場合は空の文字列を返します。この段階では、エンドポイントに POST リクエストを送信し、応答を確認することでエンドポイントをテストできます。まず、deploy を再実行します。 package.json 内のスクリプト:

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 のログに記録されます。