सर्वर‑सेंटेड इवेंट (एसएसई) का उपयोग करके नेक्स्ट.जेएस में रीयल-टाइम एलएलएम स्ट्रीमिंग

आवश्यकताएँ

• Node.js 18 या बाद का संस्करण
• एक OpenAI खाता
• एक अपस्टैश खाता
• एक वर्सेल खाता

टेक स्टैक

OpenAI टोकन जेनरेट करें

अपस्टैश रेडिस की स्थापना

एक नया Next.js एप्लिकेशन बनाएं

npx create-next-app@latest my-app

• Yes जब टाइपस्क्रिप्ट का उपयोग करने के लिए कहा जाए।
• No जब ESLint का उपयोग करने के लिए कहा जाए।
• Yes जब टेलविंड सीएसएस का उपयोग करने के लिए कहा जाए।
• No जब src/ का उपयोग करने के लिए कहा जाए निर्देशिका.
• Yes जब ऐप राउटर का उपयोग करने के लिए कहा जाए।
• No जब डिफ़ॉल्ट आयात उपनाम (@/*) को अनुकूलित करने के लिए कहा जाए ).

cd my-app
npm run dev

npm install @langchain/openai @langchain/community

• @langchain/openai :मॉडलों की OpenAI श्रृंखला के साथ इंटरफेस करने के लिए एक लैंगचेन पैकेज।
• @langchain/community :लैंगचेन कोर के साथ प्लग-एन-प्ले के लिए तृतीय पक्ष एकीकरण का संग्रह।

# .env
 
# OpenAI Environment Variable
OPENAI_API_KEY="..."
 
# Upstash Redis Environment Variables
UPSTASH_REDIS_REST_URL="..."
UPSTASH_REDIS_REST_TOKEN="..."

mkdir -p app/api/stream/completion/chat
mkdir app/lib

अपस्टैश रेडिस कैश क्लाइंट को इंस्टेंट करें

नेक्स्ट.जेएस ऐप राउटर में सर्वर-सेंटेड इवेंट एपीआई बनाना

// File: app/api/stream/route.ts
 
// Prevents this route's response from being cached on Vercel
export const dynamic = "force-dynamic";
 
export async function POST(request: Request) {
 const encoder = new TextEncoder()
 // Create a streaming response
 const customReadable = new ReadableStream({
 start(controller) {
 const message = "A sample message."
 controller.enqueue(encoder.encode(`data: ${message}\n\n`))
 },
 })
 // Return the stream response and keep the connection alive
 return new Response(customReadable, {
 // Set the headers for Server-Sent Events (SSE)
 headers: {
 Connection: "keep-alive",
 "Content-Encoding": "none",
 "Cache-Control": "no-cache, no-transform",
 "Content-Type": "text/event-stream; charset=utf-8",
 },
 })
}

• गतिशील स्थिरांक force-dynamic पर सेट है , वर्सेल प्लेटफ़ॉर्म पर प्रतिक्रियाओं की कैशिंग को रोकना। यह सुनिश्चित करता है कि एसएसई स्ट्रीम के लिए प्रत्येक अनुरोध को ताज़ा डेटा मिलता है और कैश से नहीं दिया जाता है।
• ए ReadableStream क्लाइंट को भेजे जाने वाले डेटा की एक स्ट्रीम उत्पन्न करने के लिए बनाया गया है। स्ट्रीम की प्रारंभ विधि में, एक संदेश को एन्कोड किया जाता है और स्ट्रीम के नियंत्रक में सूचीबद्ध किया जाता है। यह संदेश क्लाइंट को SSE स्ट्रीम के भाग के रूप में भेजा जाएगा।
• द Response ऑब्जेक्ट सर्वर-भेजे गए इवेंट के लिए विशिष्ट हेडर के साथ बनाया गया है। इन शीर्षलेखों में निम्नलिखित शामिल हैं:
  • Connection: keep-alive स्ट्रीमिंग के लिए कनेक्शन खुला रखने के लिए।
  • Cache-Control: no-cache, no-transform ब्राउज़र में कैशिंग को रोकने के लिए.
  • Content-Type: text/event-stream; charset=utf-8 सामग्री प्रकार को सर्वर-भेजे गए ईवेंट के रूप में निर्दिष्ट करने के लिए।
• कस्टम ReadableStream युक्त रिस्पॉन्स ऑब्जेक्ट एंडपॉइंट से लौटाया जाता है। यह प्रतिक्रिया एसएसई स्ट्रीम का अनुरोध करने वाले ग्राहकों को भेजी जाएगी।

लैंगचेन कॉलबैक के साथ ओपनएआई कंप्लीशन एपीआई प्रतिक्रियाओं को स्ट्रीम करना

// File: app/lib/completionModel.tsx
 
import { OpenAI } from "@langchain/openai";
 
export const completionModel = (
 controller: ReadableStreamDefaultController,
 encoder: TextEncoder
) =>
 new OpenAI({
 streaming: true,
 callbacks: [
 {
 handleLLMNewToken(token) {
 controller.enqueue(encoder.encode(`data: ${token}\n\n`));
 },
 async handleLLMEnd(output) {
 controller.close();
 },
 },
 ],
 });

• OpenAI आयात करता है @langchain/openai से कक्षा पैकेज.
• completionModel नाम का एक फ़ंक्शन निर्यात करता है जो दो पैरामीटर लेता है:
  • नियंत्रक:A ReadableStreamDefaultController ऑब्जेक्ट जो आपको संदेशों को स्ट्रीम में पंक्तिबद्ध करने की अनुमति देता है।
  • एनकोडर:ए TextEncoder ऑब्जेक्ट जो संदेशों को Uint8Array में एन्कोड करता है वस्तु.
• OpenAI का एक नया उदाहरण बनाता है निम्न कॉन्फ़िगरेशन विकल्पों के साथ कक्षा:
  • streaming :सत्य पर सेट करें, यह दर्शाता है कि एपीआई को स्ट्रीमिंग प्रतिक्रियाएँ लौटानी चाहिए।
  • callbacks :एक सरणी जिसमें दो कॉलबैक फ़ंक्शंस के साथ एक ऑब्जेक्ट होता है, जिसका उपयोग प्रत्येक चरण में एलएलएम द्वारा प्रतिक्रिया निर्माण को रोकने के लिए किया जाता है।
• handleLLMNewToken :भाषा मॉडल द्वारा एक नया टोकन उत्पन्न होने पर कॉलबैक लागू होता है। यह एन्कोडेड टोकन को नियंत्रक के पास भेजता है।
• handleLLMEnd :भाषा मॉडल निर्माण पूरा होने पर एक एसिंक्रोनस कॉलबैक लागू किया जाता है। यह नियंत्रक को बंद कर देता है।

// File: app/api/stream/completion/route.ts
 
// Prevents this route's response from being cached on Vercel
export const dynamic = "force-dynamic";
 
+ import { completionModel } from "@/app/lib/completionModel";
 
export async function POST(request: Request) {
+ // Obtain the user message from request's body
+ const { message } = await request.json();
 const encoder = new TextEncoder();
 // Create a streaming response
 const customReadable = new ReadableStream({
 async start(controller) {
+ // Generate a streaming response from OpenAI with LangChain
+ await completionModel(controller, encoder).invoke(message);
 },
 });
 // Return the stream response and keep the connection alive
 return new Response(customReadable, {
 // Set the headers for Server-Sent Events (SSE)
 headers: {
 Connection: "keep-alive",
 "Content-Encoding": "none",
 "Cache-Control": "no-cache, no-transform",
 "Content-Type": "text/event-stream; charset=utf-8",
 },
 });
}

• completionModel आयात करता है समारोह.
• request.json() का उपयोग करके अनुरोध निकाय से उपयोगकर्ता संदेश निकालता है .
• start के अंदर ReadableStream की विधि, completionModel अनुरोध से प्राप्त संदेश के साथ फ़ंक्शन को अतुल्यकालिक रूप से लागू किया जाता है। यह फ़ंक्शन लैंगचेन के साथ ओपनएआई से स्ट्रीमिंग प्रतिक्रिया उत्पन्न करता है।

import { OpenAI } from "@langchain/openai";
+ import cache from '@/app/lib/upstashCache';
 
export const completionModel = (
 controller: ReadableStreamDefaultController,
 encoder: TextEncoder
) =>
 new OpenAI({
+ cache,
 streaming: true,
 temperature: 0.9,
 callbacks: [
 {
 handleLLMNewToken(token) {
 controller.enqueue(encoder.encode(token));
 },
 async handleLLMEnd(output) {
 console.log(output.generations[0][0].text);
 controller.close();
 },
 },
 ],
 });

लैंगचेन कॉलबैक के साथ ओपनएआई चैट समापन एपीआई प्रतिक्रियाओं को स्ट्रीम करना

// File: app/lib/chatCompletionModel.tsx
 
import { ChatOpenAI } from "@langchain/openai";
 
export type ConversationMessage = {
 role: string;
 content: string;
};
 
export type ConversationMessages = ConversationMessage[];
 
export const chatCompletionModel = (
 controller: ReadableStreamDefaultController,
 encoder: TextEncoder
) =>
 new ChatOpenAI({
 streaming: true,
 callbacks: [
 {
 handleLLMNewToken(token) {
 controller.enqueue(encoder.encode(`data: ${token}\n\n`));
 },
 async handleLLMEnd(output) {
 controller.close();
 },
 },
 ],
 });

• ChatOpenAI आयात करता है @langchain/openai से कक्षा पैकेज.
• निम्न प्रकार की परिभाषाएँ बनाता है:
  • ConversationMessage :वार्तालाप में एकल संदेश के लिए एक प्रकार को परिभाषित करता है। इसमें दो गुण शामिल हैं:भूमिका (प्रेषक की भूमिका का संकेत) और सामग्री (संदेश सामग्री).
  • ConversationMessages :वार्तालाप संदेशों की एक श्रृंखला के लिए एक प्रकार को परिभाषित करता है।
• chatCompletionModel नाम का एक फ़ंक्शन निर्यात करता है जो दो पैरामीटर लेता है:
  • नियंत्रक:ए ReadableStreamDefaultController ऑब्जेक्ट जो आपको संदेशों को स्ट्रीम में पंक्तिबद्ध करने की अनुमति देता है।
  • एनकोडर:ए TextEncoder ऑब्जेक्ट जो संदेशों को Uint8Array में एन्कोड करता है वस्तु.
• ChatOpenAI का एक नया उदाहरण बनाता है निम्न कॉन्फ़िगरेशन विकल्पों के साथ कक्षा:
  • streaming :सत्य पर सेट करें, यह दर्शाता है कि एपीआई को स्ट्रीमिंग प्रतिक्रियाएँ लौटानी चाहिए।
  • callbacks :एक सरणी जिसमें दो कॉलबैक फ़ंक्शंस के साथ एक ऑब्जेक्ट होता है, जिसका उपयोग प्रत्येक चरण में एलएलएम द्वारा प्रतिक्रिया निर्माण को रोकने के लिए किया जाता है।
• handleLLMNewToken :भाषा मॉडल द्वारा एक नया टोकन उत्पन्न होने पर कॉलबैक लागू होता है। यह एन्कोडेड टोकन को नियंत्रक के पास भेजता है।
• handleLLMEnd :भाषा मॉडल निर्माण पूरा होने पर एक एसिंक्रोनस कॉलबैक लागू किया जाता है। यह नियंत्रक को बंद कर देता है।

// File: app/api/stream/completion/chat/route.ts
 
// Prevents this route's response from being cached on Vercel
export const dynamic = "force-dynamic";
 
+ import {
+ type ConversationMessage,
+ chatCompletionModel,
+ } from "@/app/lib/chatCompletionModel";
 
export async function POST(request: Request) {
+ // Obtain the conversation messages from request's body
+ const { messages = [] } = await request.json();
 const encoder = new TextEncoder();
 // Create a streaming response
 const customReadable = new ReadableStream({
 async start(controller) {
+ // Generate a streaming response from OpenAI with LangChain
+ await chatCompletionModel(controller, encoder).invoke(
+ messages.map((i: ConversationMessage) => [i["role"], i["content"]])
+ );
 },
 });
 // Return the stream response and keep the connection alive
 return new Response(customReadable, {
 // Set the headers for Server-Sent Events (SSE)
 headers: {
 Connection: "keep-alive",
 "Content-Encoding": "none",
 "Cache-Control": "no-cache, no-transform",
 "Content-Type": "text/event-stream; charset=utf-8",
 },
 });
}

• chatCompletionModel आयात करता है फ़ंक्शन और ConversationMessage प्रकार परिभाषा.
• request.json() का उपयोग करके अनुरोध निकाय से वार्तालाप में मौजूद संदेशों को निकालता है .
• start के अंदर ReadableStream की विधि, chatCompletionModel रूपांतरित संदेश सरणी के बाद फ़ंक्शन को अतुल्यकालिक रूप से लागू किया जाता है। प्रत्येक संदेश एक सारणी में बदल जाता है जिसमें संदेश की भूमिका और सामग्री होती है। यह फ़ंक्शन लैंगचेन के साथ ओपनएआई से स्ट्रीमिंग प्रतिक्रिया उत्पन्न करता है।

+ import cache from '@/app/lib/upstashCache';
import { ChatOpenAI } from "@langchain/openai";
 
export type ConversationMessage = {
 role: string;
 content: string;
};
 
export type ConversationMessages = ConversationMessage[];
 
export const chatCompletionModel = (
 controller: ReadableStreamDefaultController,
 encoder: TextEncoder
) =>
 new ChatOpenAI({
+ cache,
 streaming: true,
 temperature: 0.9,
 callbacks: [
 {
 handleLLMNewToken(token) {
 controller.enqueue(encoder.encode(token));
 },
 async handleLLMEnd(output) {
 console.log(output.generations[0][0].text);
 controller.close();
 },
 },
 ],
 });

सर्वर द्वारा भेजे गए ईवेंट को सुनने के लिए Next.js फ्रंटएंड सेटअप करें

रिएक्ट में सर्वर द्वारा भेजे गए इवेंट एपीआई को सुनना

"use client";
 
export default function () {
 
 const obtainAPIResponse = async (apiRoute: string, apiData: any) => {
 // Initiate the first call to connect to SSE API
 const apiResponse = await fetch(apiRoute, {
 method: "POST",
 headers: {
 "Content-Type": "text/event-stream",
 },
 body: JSON.stringify(apiData),
 });
 
 if (!apiResponse.body) return;
 
 // To decode incoming data as a string
 const reader = apiResponse.body
 .pipeThrough(new TextDecoderStream())
 .getReader();
 
 while (true) {
 const { value, done } = await reader.read();
 if (done) {
 break;
 }
 if (value) {
 // Do something
 }
 }
 };
 
 return <></>
}

प्रतिक्रिया में बातचीत की स्थिति प्रबंधित करना

"use client";
 
+ import { useEffect, useState } from "react";
+ import { ConversationMessages } from "@/app/lib/chatCompletionModel";
 
export default function () {
+ const [latestMessage, setLatestMessage] = useState<string>("");
+ const [messages, setMessages] = useState<ConversationMessages>([]);
 
 const obtainAPIResponse = async (apiRoute: string, apiData: any) => {
 // Initiate the first call to connect to SSE API
 const apiResponse = await fetch(apiRoute, {
 method: "POST",
 headers: {
 "Content-Type": "text/event-stream",
 },
 body: JSON.stringify(apiData),
 });
 
 if (!apiResponse.body) return;
 
 // To decode incoming data as a string
 const reader = apiResponse.body
 .pipeThrough(new TextDecoderStream())
 .getReader();
 
+ let incomingMessage = "";
 
 while (true) {
 const { value, done } = await reader.read();
 if (done) {
+ // Insert the response received into the messages state
+ setMessages((messages) => [
+ ...messages,
+ { role: "assistant", content: incomingMessage },
+ ]);
+ // Reset the latest message's state received 
+ setLatestMessage("");
 break;
 }
 if (value) {
+ // Append the incoming data to latest message's value
+ incomingMessage += value;
+ setLatestMessage(incomingMessage);
 }
 }
 };
 
 return <></>
}

• एसएसई एपीआई से प्राप्त आने वाली डेटा स्ट्रीम को स्टोर करने के लिए एक स्टेट वेरिएबल घोषित किया गया है।
• पूर्ण संदेश प्राप्त होने पर (done द्वारा दर्शाया गया)। सत्य होने पर), नवीनतम प्राप्त संदेश messages में जोड़ा जाता है राज्य सरणी. यह संदेश सहायक की भूमिका के साथ स्वरूपित किया गया है इसे उपयोगकर्ता संदेशों से अलग करने के लिए। साथ ही, अगले आने वाले संदेश की तैयारी के लिए नवीनतम संदेश स्थिति को एक खाली स्ट्रिंग पर रीसेट कर दिया जाता है।
• चूंकि डेटा क्रमिक रूप से प्राप्त होता है (अभी तक पूर्ण संदेश नहीं बना है), इसे incomingMessage में जोड़ दिया जाता है परिवर्तनशील.
• नवीनतम संदेश स्थिति को सम्मिलित आने वाले डेटा के साथ अद्यतन किया जाता है, जिससे नए डेटा के आने पर वास्तविक समय में अपडेट सुनिश्चित किया जाता है।

वर्सेल में तैनात करें

• अपने ऐप के कोड वाली GitHub रिपॉजिटरी बनाकर शुरुआत करें।
• फिर, वर्सेल डैशबोर्ड पर नेविगेट करें और एक नया प्रोजेक्ट बनाएं .
• नए प्रोजेक्ट को आपके द्वारा अभी बनाए गए GitHub रिपॉजिटरी से लिंक करें।
• सेटिंग्समें , पर्यावरण चर को अद्यतन करें आपके स्थानीय .env से मिलान करने के लिए फ़ाइल.
• तैनाती! 🚀

अधिक जानकारी

• गिटहब रिपॉजिटरी
• सर्वर द्वारा भेजे गए इवेंट
• Next.js स्ट्रीमिंग
• एलएलएम के लिए कैशिंग परत