home

หน้าแรก
 Computer >> บทช่วยสอนคอมพิวเตอร์ >  >> การเขียนโปรแกรม >> Redis

เพิ่มประสิทธิภาพด้วย On-Demand LLM Caching: LangChain, Redis และ QStash

ในบางแอปพลิเคชัน อาจไม่เหมาะสมที่จะ Query LLM เมื่อผู้ใช้ร้องขอ กระบวนการส่ง Prompt ไป API และรอ Response อาจใช้เวลานานมาก โดยเฉพาะงานที่ซับซ้อนและใช้ LangChain เช่น การประมวลผลข้อมูลจาก PDF หรือ Audio แล้วส่งให้ LLM ความล่าช้าจะส่งผลต่อ UX อย่างมาก

แม้ Streaming จะเป็น حلเหมาะสมในหลายกรณี แต่การทำทุกอย่างล่วงหน้าที่จุดใดจุดหนึ่งก่อนที่ผู้ใช้จะเข้ามาใช้งาน จะสะดวกกว่า ดังนั้นผู้ใช้จะได้ Response ที่ Cache ไว้ทันทีโดยไม่ต้องรอ LLM Generate นี่มีประโยชน์มากสำหรับแอปที่ใช้ Input เดียวกันทุก User เช่น สรุปหนังสือหรือข่าว แต่อย่างที่ไม่มี Input จาก User เราจะให้ LangChain Cache Response ได้อย่างไร?

เราต้องการ Microservice ที่เรียก On-demand ได้ สามารถส่ง Prompt ให้ LangChain และ Cache Response ด้วย Upstash Redis เราจะใช้ Upstash Rate Limiting SDK เพื่อป้องกันไม่ให้เกิน Rate Limit QStash เป็นวิธีที่ยืดหยุ่นที่สุดในการเรียก Microservice เราสามารถตั้งให้ Cache Response เมื่อไหร่ก็ได้ แม้ Cron Job ก็ตาม มี Dashboard สำหรับ Monitor การใช้งาน และ QStash จะ Retry HTTP Request หาก Endpoint Overloaded หรือมีปัญหา เพื่อให้ Message ถูกส่งมั่นใจ

Microservice ของเราจะสร้างด้วย Hono.js Framework เว็บที่เบาและเร็วสำหรับ Edge ใน Demo นี้ใช้ Cloudflare Worker โฮสต์ แต่ด้วย Upstash สามารถ Deploy ได้เกือบทุกที่ รวม Edge/Serverless Runtimes อื่นๆ

ซอร์สโค้ดเต็มได้ที่นี่

Prerequisites - เงื่อนไขก่อนเริ่ม

  • Upstash Redis Database
  • QStash Environment Variables
  • OpenAI API Key

Getting Started - เริ่มต้น

Creating the Project - สร้างโปรเจกต์

ต่างจาก create-<package> อื่นๆ create-hono ต้องอยู่ใน Empty Directory ให้สร้าง Directory ใหม่และเข้าไป:

mkdir langchain-qstash
cd langchain-qstash

ด้วยการ Release v1.0.0 จะใช้ Bun Scaffold โปรเจกต์ แต่ใช้ npm, pnpm, yarn ก็ได้ เลือก cloudflare-workers เมื่อถาม Template (ไม่ใช่ bun option แยกต่างหาก)

bun create hono@latest

Installing Dependencies - ติดตั้ง Dependencies

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

Configuring the Project - กำหนดค่าโปรเจกต์

ตอนนี้ create-hono ใส่ Lockfiles ใน .gitignore โดย Default แม้ไม่จำเป็น แต่สามารถ Update .gitignore ให้ Exclude Lockfiles ได้:

node_modules
dist
.wrangler
.dev.vars
wrangler.toml

ตั้ง Environment Variables จาก Prerequisites เพิ่มใน wrangler.toml (ควร Exclude จาก Source Control แล้ว):

[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 ให้เพิ่ม Typing สำหรับ Environment Variables:

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 }>();

ประโยชน์ของ Signing Keys 2 ตัว คือสามารถ Roll ได้โดยไม่ต้อง Update Env เพราะ QStash จะลอง Next Key ถ้า Current ล้มเหลว

Development - พัฒนา

ตอนนี้ wrangler ควรอ่าน Env และ Deploy ได้ รัน:

bun run dev

wrangler จะ Spin Up Hono Server ให้ Local URL Port 8787 หากต้องการ Edge Preview แทน Local ให้แก้ Script ใน package.json:

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

เพื่อดู Config และ Logs ของ Cloudflare Worker ต้อง Deploy ก่อน:

bun run deploy

จะเดินทาง Sign in Cloudflare และ Deploy อัตโนมัติหลัง Auth

Creating Middleware - สร้าง Middleware

สำหรับ Debug จะดีถ้า Log Request จาก QStash จะเห็นใน Dashboard Cloudflare Worker แท็บ Logs กด Begin log stream Hono มี Logger Middleware ให้เพิ่ม:

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

เพิ่มประสิทธิภาพด้วย On-Demand LLM Caching: LangChain, Redis และ QStash

Connecting QStash to Cloudflare Workers

ก่อนสร้าง API Endpoint ต้องมีวิธี Intercept Message จาก QStash และทิ้ง Request ที่ Signature ไม่ถูกต้อง Hono มี Custom Middleware ที่รันก่อน Handler พอที่จะจัดแยกไฟล์ได้

ใน Middleware ใช้ QStash Receiver สร้างไฟล์ src/middleware/verify.ts Export Function Typed 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 ส่ง ctx (Context) ให้ Middleware และ Handler มีข้อมูลเท่า Cloudflare Workers ExecutionContext แต่รวม Request, Env, Execution Context ใน Object เดียว

เราเพิ่ม locals ใน Global Response Interface เพื่อส่ง Query ให้ Handler ต่อไป (Hono ไม่สร้าง res.locals Default เหมือน Express) ใช้ Env Variables ที่ Typed ไว้สร้าง Receiver

Verify Signature ของ Request:

// 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();

อ่าน Body เป็น Text และ Header Upstash-Signature (JWT Custom) Verify ด้วย Receiver ส่ง Signature และ Body ใน .catch Log Error และ Return False หาก Invalid

เนื่องจาก Consume Body แล้ว Handler ต่อไปจะเข้าถึงไม่ได้ (ReadableStream อ่านครั้งเดียว) จึงเก็บ Query ไว้ใน locals บน Response หาก Signature Invalid Return 401 ไม่งั้น next() ต่อ

Adding Rate Limiting - เพิ่ม Rate Limiting

เนื่องจากเชื่อมต่อ OpenAI API Key สำคัญที่จะไม่เกิน Rate Limit Upstash มี Rate Limiting SDK ใช้ง่าย หาก Request เกิน Threshold Return 429

สร้าง Middleware ใหม่ 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 กับ Rate Limiting SDK สร้าง Redis Instance เองจาก ctx.env เพราะ Redis.fromEnv() อ่านไม่ได้ Auto บน Cloudflare Workers เมื่อ analytics True SDK จะ Cache Calls ต่อ Identifier ใช้ Prefix @upstash/ratelimit

SDK รองรับ Ephemeral Cache ใช้ Map แทน Redis ประหยัด Resource ภายใต้ Load สูง:

const cache = new Map(); // นอก Middleware Handler
// snip
const ratelimit = new Ratelimit({
  ephemeralCache: cache,
  // snip
});

ให้ Limiter แบบ slidingWindow อนุญาต 10 Requests ทุก 10 วินาที

SDK ให้ Identifier ต่อ Request ติดตามจำนวน Request ในช่วงเวลา ใน Real-world ใช้ IP User แต่ Demo ใช้ Constant String Limit ทั้งหมด:

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

Rate Limited Return 429 ไม่งั้น next() หากหลาย Redis Databases ต้อง Sync จะมี Dangling Promises บน Vercel Edge และ Cloudflare Workers จัดการได้โดย:

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

Demo ใช้ Single Redis จึงไม่ต้องกังวล

Receiving Messages from QStash - รับข้อความจาก QStash

ส่ง HTTP Request ไป QStash Destination คือ URL Cloudflare Workers Endpoint สร้าง Handler ด้วย Hono ใช้ Middleware ก่อนหน้า:

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

ใน Handler ใช้ LangChain Generate Response จาก Prompt Cache ด้วย 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

Import Classes ตั้งค่า Cache LangChain สร้าง Chain ด้วย OpenAI Model และ Upstash Redis Cache ส่ง Redis Instance ให้ Cache อีกครั้ง ต้องส่ง OpenAI API Key Manual เพราะอ่าน Auto ไม่ได้บน Cloudflare Workers

เข้าถึง Query จาก res.locals เรียก Model Log Result Return Response หรือ Empty String หาก Error Deploy ใหม่:

bun run deploy

ดึง URL Cloudflare Workers ส่ง POST ด้วย curl:

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 Console ส่ง Request ได้ง่าย ตั้ง Cron Job ให้รันซ้ำได้ QStash ส่ง Body As-is ไป Endpoint Payload JSON:

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

เพิ่มประสิทธิภาพด้วย On-Demand LLM Caching: LangChain, Redis และ QStash

Conclusion - สรุป

Rate Limiting SDK Cache จำนวน Calls ต่อ Identifier สำเร็จ:

เพิ่มประสิทธิภาพด้วย On-Demand LLM Caching: LangChain, Redis และ QStash

Content จาก LangChain Cache ใน Upstash Redis สำเร็จ:

เพิ่มประสิทธิภาพด้วย On-Demand LLM Caching: LangChain, Redis และ QStash

Response Log ใน Cloudflare Worker Logs:

เพิ่มประสิทธิภาพด้วย On-Demand LLM Caching: LangChain, Redis และ QStash