LangChain, Redis और QStash के साथ ऑन-डिमांड LLM कैशिंग से प्रदर्शन बढ़ाएं

कुछ एप्लिकेशन में, जब कोई उपयोगकर्ता अनुरोध करता है तो LLM को क्वेरी करना संभव नहीं हो सकता है। एक प्रॉम्प्ट को API पर भेजने और उसकी प्रतिक्रिया की प्रतीक्षा करने की प्रक्रिया विशेष रूप से समय लेने वाली हो सकती है। LangChain जैसे अधिक जटिल कार्यों के लिए, जैसे PDF या ऑडियो फ़ाइल से डेटा प्रोसेस करना और इसे LLM को खिलाना, देरी उपयोगकर्ता अनुभव के लिए और भी बड़ा झटका बन जाती है।

जबकि स्ट्रीमिंग है कई मामलों में एक उपयुक्त समाधान, उपयोगकर्ता द्वारा आपके एप्लिकेशन पर जाने से पहले किसी मनमाने बिंदु पर सब कुछ करना अधिक सुविधाजनक हो सकता है। इस तरह, उपयोगकर्ता को LLM द्वारा इसे उत्पन्न करने की प्रतीक्षा किए बिना कैश्ड प्रतिक्रिया लगभग तुरंत परोसी जा सकती है। यह उन एप्लिकेशन के लिए विशेष रूप से उपयोगी है जो हर उपयोगकर्ता के लिए समान इनपुट का उपयोग करते हैं, जैसे पुस्तक या समाचार सारांशकर्ता। लेकिन उपयोगकर्ता से इनपुट के बिना, हम LangChain को प्रतिक्रिया कैश करने के लिए कैसे प्राप्त कर सकते हैं?

हमें एक माइक्रोसर्विस की आवश्यकता होगी जिसे हम ऑन-डिमांड कॉल कर सकें, जो LangChain को प्रॉम्प्ट भेजने और Upstash Redis का उपयोग करके प्रतिक्रिया को कैश करने में सक्षम हो। हम यह सुनिश्चित करने के लिए Upstash के रेट लिमिटिंग SDK का भी उपयोग करेंगे कि हम गलती से अपनी रेट लिमिट से अधिक न हो जाएं। QStash हमारे माइक्रोसर्विस को कॉल करने का सबसे बहुमुखी तरीका है—हम इसे जब चाहें प्रतिक्रियाओं को कैश करने के लिए सेट कर सकते हैं, यहां तक कि एक क्रॉन जॉब पास कर सकते हैं यदि हमारे एप्लिकेशन को इसकी आवश्यकता है। यह एक डैशबोर्ड भी प्रदान करता है जिसका उपयोग हम अपने माइक्रोसर्विस के उपयोग की निगरानी के लिए कर सकते हैं। यदि हमारा एंडपॉइंट ओवरलोडेड है या कुछ और गलत होता है, तो QStash HTTP अनुरोध को पुनः प्रयास करेगा और यह सुनिश्चित करेगा कि हमारा संदेश वितरित हो।

हमारा माइक्रोसर्विस 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 का उपयोग किया जाएगा। हालांकि, 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 }>();

दो साइनिंग कुंजियाँ होने का लाभ यह है कि आप अपने पर्यावरण चर को अपडेट किए बिना उन्हें एक बार रोल कर सकते हैं, क्योंकि QStash स्वचालित रूप से अगली साइनिंग कुंजी को आज़माएगा और उपयोग करेगा यदि वर्तमान विफल हो जाती है।

विकास

इस चरण में, wrangler आपके पर्यावरण चर को पढ़ने और आपके प्रोजेक्ट को डिप्लॉय करने में सक्षम होना चाहिए। ऐसा करने के लिए, आप अपनी पसंद के पैकेज मैनेजर के साथ निम्न कमांड चला सकते हैं:

bun run dev

wrangler आपके Hono.js सर्वर को स्पिन अप करेगा और आपके प्रोजेक्ट का परीक्षण करने के लिए पोर्ट 8787 पर एक स्थानीय URL प्रदान करेगा। यदि आप स्थानीय रूप से परीक्षण करने के बजाय सीधे एक एज प्रीव्यू सत्र शुरू करना चाहते हैं, तो अपने देव package.json स्क्रिप्ट को इस प्रकार संशोधित करें:

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

अपने Cloudflare Worker के कॉन्फ़िगरेशन और लॉग देखने के लिए, आपको पहले अपने वर्कर को डिप्लॉय करना होगा:

bun run deploy

यह आपको प्रमाणीकरण के बाद पहली बार स्वचालित रूप से अपने प्रोजेक्ट को डिप्लॉय करने के चरणों के माध्यम से चलना चाहिए।

मिडलवेयर बनाना

डीबगिंग उद्देश्यों के लिए, QStash से प्राप्त अनुरोधों को लॉग करना उपयोगी होगा। वे हमारे Cloudflare Worker के डैशबोर्ड में "लॉग्स" टैब के तहत दिखाई देंगे जब आप 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 के रिसीवर का उपयोग कर सकते हैं। आइए 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 हैंडलर को पास किए गए एक्जीक्यूशन कॉन्टेक्स्ट के संयोजन के लगभग बराबर है, सभी एक ऑब्जेक्ट में संयुक्त।

हम वैश्विक 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 सुविधाजनक रूप से केवल वेब स्टैंडर्ड API का उपयोग करता है, जैसे fetch, URL, Request, और Response। यद्यपि हम इस डेमो में इसे Cloudflare Workers पर उपयोग कर रहे हैं, यह इसे अनगिनत अन्य वातावरणों पर चलाने में सक्षम बनाता है, जिसमें एज/सर्वरलेस वातावरण शामिल हैं।

अनुरोध ऑब्जेक्ट से, हम अनुरोध के शरीर को टेक्स्ट के रूप में पढ़ते हैं साथ ही Upstash-Signature हेडर, जिसमें एक कस्टम JWT होता है। हम अनुरोध के हस्ताक्षर और शरीर को पास करके रिसीवर का उपयोग करके JWT में निहित हस्ताक्षर को सत्यापित कर सकते हैं। .catch हैंडलर में, हम त्रुटि को लॉग करना सुनिश्चित करते हैं साथ ही false लौटाते हैं यह इंगित करने के लिए कि हस्ताक्षर अमान्य है।

क्योंकि हम अनुरोध के शरीर का उपभोग कर रहे हैं, हमारे पास बाद में अपने वास्तविक हैंडलर में इसकी पहुंच नहीं होगी। ऐसा इसलिए है क्योंकि शरीर एक ReadableStream है जिसे केवल एक बार उपभोग किया जा सकता है। इसके बजाय, हैंडलर को क्वेरी पास करने के लिए, हम इसे प्रतिक्रिया पर locals ऑब्जेक्ट में जोड़ सकते हैं। अंत में, यदि हस्ताक्षर अमान्य है, तो हम 401 अनधिकृत प्रतिक्रिया लौटाते हैं। अन्यथा, हम अगले मिडलवेयर या हैंडलर पर जारी रखने के लिए next() कॉल करते हैं।

रेट-लिमिटिंग जोड़ना

क्योंकि हम इस एंडपॉइंट को अपने व्यक्तिगत OpenAI API कुंजियों से जोड़ने जा रहे हैं, यह सुनिश्चित करना महत्वपूर्ण है कि हम गलती से अपनी रेट लिमिट से अधिक न हो जाएं। शुक्र है, Upstash एक रेट लिमिटिंग SDK प्रदान करता है जिसका उपयोग हम अपने एंडपॉइंट पर आसानी से रेट-लिमिटिंग जोड़ने के लिए कर सकते हैं। यदि अनुरोधों की संख्या एक दी गई सीमा से अधिक हो जाती है, तो हम 429 टू मेनी रिक्वेस्ट प्रतिक्रिया लौटाएंगे।

आइए 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 से जोड़ रहे हैं। हमें ctx.env से पर्यावरण चर का उपयोग करके Redis इंस्टेंस को मैन्युअल रूप से बनाना होगा, क्योंकि Redis.fromEnv() Cloudflare Workers का उपयोग करते समय उन्हें स्वचालित रूप से पढ़ने में विफल रहता है। जब analytics सत्य होता है, तो SDK स्वचालित रूप से प्रत्येक पहचानकर्ता के लिए कॉल का कैश रखने के लिए Redis को कॉल करेगा। यह डिफ़ॉल्ट रूप से @upstash/ratelimit को उपसर्ग के रूप में उपयोग करता है।

SDK एक एफेमेरल कैश का उपयोग करने का भी समर्थन करता है जो Redis के बजाय Map का उपयोग करता है, जो चरम लोड के तहत समय और संसाधनों को बचा सकता है:

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 टू मेनी रिक्वेस्ट प्रतिक्रिया लौटाते हैं। अन्यथा, हम अगले मिडलवेयर या हैंडलर पर जारी रखने के लिए next() कॉल करते हैं। यदि हमने SDK से कई Redis डेटाबेस जोड़े होते, तो उसे उनके बीच सिंक्रनाइज़ेशन करना पड़ता। इससे Vercel Edge और Cloudflare Workers पर डैंगलिंग प्रॉमिस हो जातीं, जिनका हम इस तरह ख्याल रख सकते हैं:

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

इसे करने का तरीका आपके द्वारा उपयोग की जा रही लाइब्रेरी के आधार पर भिन्न हो सकता है। चूंकि Hono वेब स्टैंडर्ड API का उपयोग करता है, हम प्रॉमिस के हल होने की प्रतीक्षा करने के लिए event.waitUntil विधि का उपयोग कर सकते हैं। हालांकि, इस डेमो के लिए, हम केवल एक ही Redis डेटाबेस का उपयोग करेंगे, इसलिए हमें डैंगलिंग प्रॉमिस के बारे में चिंता करने की ज़रूरत नहीं है।

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 इंस्टेंस पास करते हुए। फिर से, हमें अपने OpenAI API कुंजी को मॉडल को मैन्युअल रूप से पास करना होगा क्योंकि इसे Cloudflare Workers पर स्वचालित रूप से पढ़ा नहीं जा सकता।

फिर, हम उस क्वेरी को एक्सेस करते हैं जिसे पहले 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 को हमारे अनुरोध को बार-बार चलाने के लिए एक क्रॉन जॉब देना भी उतना ही सरल है। QStash अनुरोध के शरीर को जैसा है वैसा ही हमारे एंडपॉइंट पर पास करता है। हम निम्नलिखित JSON पेलोड भेज रहे हैं, जहां query मॉडल के लिए प्रॉम्प्ट है:

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

निष्कर्ष

रेट लिमिटिंग SDK प्रति पहचानकर्ता कॉल की संख्या को सफलतापूर्वक कैश करता है:

इसी तरह, LangChain से उत्पन्न सामग्री हमारे Upstash Redis डेटाबेस में सफलतापूर्वक कैश हो जाती है:

और अंत में, प्रतिक्रिया हमारे Cloudflare Worker के लॉग में लॉग होती है: