ডিসকানেক্ট, রিফ্রেশ, ক্র্যাশের মধ্যেও টিকে থাকা রেজিলিয়েন্ট LLM স্ট্রিম তৈরি করুন

আমরা কী বিল্ড করছি

এই আর্টিকেলে, আমরা অত্যন্ত টেকসই LLM স্ট্রিম বিল্ড করব যা সহজেই মোকাবিলা করতে পারে:

• নেটওয়ার্ক আউটেজ
• পেজ রিফ্রেশ
• ওয়েবসাইট বন্ধ করা
• ল্যাপটপের ঢাকনা বন্ধ করা

বোনাস: আপনি একই স্ট্রিম একাধিক ডিভাইসে (যেমন ফোন এবং ল্যাপটপ) একই সাথে দেখতে পারেন।

আপনি স্ট্রিম ভাঙতে কতটুকু চেষ্টা করুন না কেন, এটি পটভূমিতে চলতে থাকে যতক্ষণ আপনি ডিসকানেক্টেড থাকেন, এবং আপনি ফিরে আসার সাথে সাথে সুন্দরভাবে চালিয়ে যায়। এটি একটি অভব ইউজার এক্সপেরিয়েন্স।

টেকসই LLM স্ট্রিম ডেমো 👇

প্রেরণা

AI দিয়ে বিল্ড করার সময়, রিয়েল-টাইমে AI রেসপন্স স্ট্রিম করা একটি বেস্ট প্র্যাকটিস।

পুরো রেসপন্সের জন্য অপেক্ষা করার বদলে, আপনার ইউজার কন্টেন্ট জেনারেট হওয়ার সাথে সাথে রিয়েল-টাইমে দেখে — যা UX-এর জন্য চমৎকার। Vercel-এর AI SDK-এর মতো টুলগুলো এটা খুব সহজ করে দিয়েছে:

import { openai } from "@ai-sdk/openai"; import { streamText } from "ai"; const { textStream } = streamText({ model: openai("gpt-4o"), prompt: "একটি নতুন ছুটি উদ্ভাবন করুন এবং এর রীতিনীতি বর্ণনা করুন।", });

রিয়েল-টাইম LLM স্ট্রিম টেকনিক্যাল লেভেলে কাজ করার জন্য, আপনি একটি ক্লায়েন্টকে একটি API-র সাথে কানেক্ট করেন এবং Server Sent Events (SSE) এর মতো প্রোটোকল ব্যবহার করে ডেটা স্ট্রিম করে ফেরত পাঠান:

কিন্তু: এই সেটআপে একটি সমস্যা আছে।

স্ট্রিম চলার সময় যদি কিছুই হয়, যেন আপনার ইন্টারনেট ডিসকানেক্ট হয়, ল্যাপটপের ঢাকনা বন্ধ হয়, বা নেটওয়ার্ক হিকাপ — পুরো জেনারেশন হারিয়ে যায়। আপনাকে আবার শুরু করতে হয় এবং পুরো জেনারেশনের জন্য অপেক্ষা করতে হয়। এটি বিশেষ করে লম্বা জেনারেশনের জন্য (যেমন O1-এর মতো ব্যয়বহুল মডেল চালানোর সময়) অত্যন্ত বিরক্তিকর।

স্পষ্টতই এই সমস্যা মানুষের রডারে আছে। വിശ্বস্ত রিয়েল-টাইম LLM স্ট্রিমিং-এর জন্য আসল চাহিদা আছে, এবং আরও ডেভেলপার এটি কাজ করার উপায় পরীক্ষা করছেন:

উচ্চ-টেকসই LLM স্ট্রিম বিল্ড করা

সত্যিকারের টেকসই, রিজিউমেবল LLM স্ট্রিম তৈরির গোপন উপাদান হলো ক্লায়েন্টকে জেনারেশন এনভায়রনমেন্ট থেকে আলাদা করা। ক্লায়েন্ট কানেকশন অস্থির এবং অনেক কারণেই ডিসকানেক্ট হতে পারে, যেন ল্যাপটপ বন্ধ করা, নেটওয়ার্ক সমস্যা বা পেজ রিফ্রেশ।

ক্লায়েন্ট এবং জেনারেশন প্রসেসকে আলাদা রাখলে, জেনারেশন সর্বদা অনবিচ্ছিন্ন চলতে থাকে। ক্লায়েন্ট যেকোনো সময় রিকানেক্ট করতে পারে — চলমান জেনারেশন বাধা ছাড়া।

খারাপ ধারণা: স্থায়ী, সরাসরি কানেকশন:

ভালো ধারণা: রিপ্লেসেবল, ইন্টাররপ্টিবল স্ট্রিম কানেকশন:

এবং হ্যাঁ — এই আর্কিটেকচার একটি সিম্পল AI স্ট্রিমের জন্য доста জটিল দেখাতে পারে। কিন্তু কোড দেখলে বুঝবেন, এটি কয়েক লাইন কোড আর কয়েক মিনিটেই ইমপ্লিমেন্ট করা যায়।

টেকসই স্ট্রিম সেটআপ করা

একটি অত্যন্ত নির্ভরযোগ্য LLM স্ট্রিম সেটআপের তিনটি অংশ আছে:

• দ্য ক্লায়েন্ট (ফ্রন্টএন্ড)
• দ্য স্ট্রিম জেনারেটর (একটি API রাউট)
• দ্য স্ট্রিম কনজিউমার (একটি API রাউট)

ক্লায়েন্টের সাথে সরাসরি সকল কানেকশন যেকোনো সময় ইন্টাররপ্ট বা পজ হতে পারে। LLM আউটপুট স্ট্রিম জেনারেট করার দায়িত্বে থাকা লজিক (স্ট্রিম জেনারেটর) তাই একটি স্বাধীন API হওয়া উচিত যার ক্লায়েন্টের সাথে কখনো এক্টিভ কানেকশন থাকে না।

বদলে, আমরা ক্লায়েন্টের সাথে কনজিউমারের মাধ্যমে কানেক্ট করব — যা শুধু রেডিস থেকে ডেটা পড়ে এবং অন্যথায় די "বুদ্ধিহীন"। এর একমাত্র উদ্দেশ্য হলো জেনারেটরের আউটপুট পড়ে ক্লায়েন্ট যেসব LLM চান্ক এখনও দেখেনি, সেগুলো সরবরাহ করা যখনই কোনো ক্লায়েন্ট কানেক্ট করে। এটাই সারবস্তু।

দ্রুত সারসংক্ষেপ — প্রতিটি অংশ কী করে:

• ক্লায়েন্ট: স্ট্রিম জেনারেটর ট্রিগার করে (কিন্তু কখনো ওপেন কানেকশন মেইনটেইন করে না) এবং রিয়েল-টাইম স্ট্রিম রেন্ডার করে
• স্ট্রিম জেনারেটর: রিয়েল-টাইমে LLM আউটপুট জেনারেট করে এবং রেডিসে পাবলিশ করে
• স্ট্রিম কনজিউমার: জেনারেটরের স্ট্রিম পড়ে এবং চান্কগুলো ক্লায়েন্টে পুশ করে

জেনারেটর শুধু একটি LLM স্ট্রিম পড়ে সেটিকে রিয়েল-টাইমে রেডিসে পাবলিশ করার জন্য দায়বদ্ধ। আমরা ক্লায়েন্ট থেকে স্ট্রিম কনজিউমারের একটি রিপ্লেসেবল কানেকশন পাই যা এন্ড, রিকানেক্ট করা যায় — কিছুই স্ট্রিম জেনারেটরকে প্রভাবিত করে না।

কোড উদাহরণ

এই সেকশনে, আমরা কোডে নজর দিব। নীতিগুলো খুব স্পষ্ট করতে, আমরা এক्चুয়াল, কমপ্লিট প্রডাকশন কোড ইমপ্লিমেন্টেশন শেষে দেখব।

এখনতک, কোর স্নিপেটগুলো এবং তাদের উদ্দেশ্য দেখে কোড বোঝা অনেক সহজ — পুরো কোড ফাইল দেখার বদলে।

১. ক্লায়েন্ট

ক্লায়েন্টের শুধু ৩টি দায়িত্ব আছে:

• সেশন আইডি জেনারেট করা
• জেনারেটর ট্রিগার করা
• জেনারেশন স্ট্রিম রেন্ডার করা

প্রতিটি দেখি:

ক্লায়েন্ট: সেশন আইডি জেনারেট করা

যখন একটি ক্লায়েন্ট একটি স্ট্রিমের সাথে কানেক্ট বা রিকানেক্ট করে, আমরা সেই সকল মেসেজ পাঠাতে চাই যা ক্লায়েন্ট এখনও দেখেনি। অর্থাৎ একটি এক্টিভ স্ট্রিমের সময়, প্রতিটি মেসেজ শুধু সटিক ডেল্টা ধারণ করে যা ক্লায়েন্টকে দেখাতে হবে, পুরো স্ট্রিম নয়।

রিকানেক্ট করার সময়, বর্তমান জেনারেশন পয়েন্ট পর্যন্ত পুরো স্ট্রিম পাঠানো হয় এবং ভবিষ্যতের সকল ইভেন্টের সাবস্ক্রিপশন কোনো মিসিং চান্ক ছাড়া সম্পূর্ণ সিমলেস হয়।

কীভাবে?

রেডিস স্ট্রিমস, রিয়েল-টাইম ডেটা কার্যকরভাবে সংরক্ষণ ও রিট্রিভ করার একটি উপায়, এই not-yet-seen ফাংশনালিটি বিল্ট-ইন রেখেছে কনসিউমার গ্রুপস নামক একটি জিনিসের মাধ্যমে। আমাদের যা করতে হবে: প্রতিটি ক্লায়েন্টের একটি ইউনিক সেশন নিশ্চিত করা — অর্থাৎ প্রতিটি জেনারেশনের জন্য একটি ইউনিক আইডি অ্যাসাইন করা।

কনসিউমার গ্রুপস সম্পর্কে আমরা স্ট্রিম কনজিউমার দেখার সময় আরো জানব। এগুলো এমন দেখায়:

await redis.xgroup("redis-key", { type: "CREATE", group: "my-group-name", id: "0", });

কোন ক্লায়েন্ট কোন স্ট্রিম কোথায় পর্যন্ত দেখেছে এবং কোন পিসগুলো মিসিং — এই সম্পূর্ণ লজিক রেডিস স্ট্রিমস দ্বারা নিশ্চিত দক্ষতার সাথে হ্যান্ডেল করা হয়। আমরা কখনো কোনো মিসিং LLM চান্ক পাই না এবং সর্বদা সঠিক ডেটা পাঠাই যা ক্লায়েন্টের দরকার।

এখন ক্লায়েন্টের যা করতে হবে: প্রতিটি জেনারেশনের জন্য একটি আইডি অ্যাসাইন করা। আমরা সিম্পলি nanoid ব্যবহার করি:

import { customAlphabet } from "nanoid" const nanoid = customAlphabet("0123456789", 6);

ক্লায়েন্ট: জেনারেশন স্ট্রিম ট্রিগার করা

জেনারেশন ইঞ্জিনের সাথে ক্লায়েন্টের একমাত্র ইন্টারঅ্যাকশন হলো এটি ট্রিগার করা। টেকনিক্যালি বলতে, আপনি জেনারেশন যেকোনো জায়গা থেকে ট্রিগার করতে পারেন (যেমন CRON জব, অটোমেটেড পাইপলাইন)।

এটির সিম্পলেস্ট ফর্ম হলো জেনারেশন API রাউটে একটি ফेच কল:

// 👇 স্ট্রিম জেনারেটর ট্রিগার করুন await fetch("/api/llm-stream", { method: "POST", headers: { "Content-Type": "application/json", }, body: JSON.stringify({ prompt, sessionId }), });

ক্লায়েন্ট: জেনারেশন স্ট্রিম পড়া

জেনারেশন ট্রিগার করার পর, জেনারেটর রিয়েল-টাইমে LLM আউটপুট একটি সেন্ট্রালাইজড রেডিস স্টোরে স্ট্রিম করতে শুরু করে — ক্লায়েন্ট থেকে সম্পূর্ণ ডিকাপল্ড। আসুন স্ট্রিম কনজিউমারের সাথে কানেক্ট হই জেনারেশন স্ট্রিম পড়ি:

// 👇 স্ট্রিম কনজিউমারের সাথে কানেক্ট করুন const res = await fetch(`/api/check-stream?sessionId=${sessionId}`, { headers: { "Content-Type": "text/event-stream" }, });

ততটুকুই!

এই তিনটি দায়িত্বই ক্লায়েন্টের। অবশ্যই আমরা কাস্টম হুকস ফর আইডি জেনারেশন, রিয়েক্ট-কুয়েরি ফর অ্যাডিশনাল রিলায়েবিলিটি ইত্যাদি দিয়ে আরও ফ্যান্সি হতে পারি — কমপ্লিট কোড উদাহরণে সেটা দেখব।

২. স্ট্রিম জেনারেটর

স্ট্রিম জেনারেটর একটি LLM স্ট্রিম খুলে এবং প্রতিটি চান্ক রেডিস স্ট্রিমে লিখে। এটি প্রতিটি চান্ক লেখার জন্য একটি মেসেজ পাবলিশ করে যাতে স্ট্রিম কনজিউমারকে নতুন ডেটার বিষয়ে রিয়েল-টাইম আপডেটের জন্য সতর্ক করা যায়।

নোট: আবার, এটি ইন্টেনশনালি ফুল কোড উদাহরণ নয়। শেষে ফুল কোড দেখব — এখন কনসেপ্ট বোঝার জন্য।

import { streamText } from "ai" import { redis } from "@/utils" const result = await new Promise( async (resolve, reject) => { const { textStream } = streamText({ model: openai("gpt-4o"), prompt, onError: (err) => reject(err), onFinish: async () => { resolve({ // ... }), }) for await (const chunk of textStream) { if (chunk) { const chunkMessage: ChunkMessage = { type: MessageType.CHUNK, content: chunk, } // 👇 চান্কটি রেডিস স্ট্রিমে লিখুন await redis.xadd(streamKey, "*", chunkMessage) // 👇 কনজিউমারকে সতর্ক করুন যে নতুন চান্ক আছে await redis.publish(streamKey, { type: MessageType.CHUNK }) } } } )

৩. স্ট্রিম কনজিউমার

স্ট্রিম কনজিউমার রেডিসের সাথে কানেক্ট হয় এবং রেডিস পাব/সাব-এর মাধ্যমে নতুন চান্ক এলার্ট শোনে। প্রতিটি ক্লায়েন্টের নিজস্ব কনসিউমার গ্রুপ থাকে যাতে সে কী দেখেছে এবং কী দেখেনি তা ট্র্যাক করতে পারে।

নোট: পাবলিশ আসল চান্ক ট্রান্সফার করে না, শুধু স্ট্রিমে নতুন চান্ক আছে তা সতর্ক করে।

যখন একটি নতুন চান্ক উপলব্ধ হয়, স্ট্রিম কনজিউমার API সেটিকে স্ট্রিম থেকে পড়ে এবং সব কানেক্টেড ক্লায়েন্টে ফরওয়ার্ড করে। রেডিস কনসিউমার গ্রুপ ট্র্যাক করে যে প্রতিটি ক্লায়েন্ট কী দেখেছে যাতে কোনো ডুপ্লিকেট বা মিসিং চান্ক ট্রান্সফার না হয়।

কোর স্ট্রিম কনজিউমার এমন দেখায়:

const streamKey = `llm:stream:${sessionId}`; const groupName = `sse-group-${nanoid()}`; await redis.xgroup(streamKey, { type: "CREATE", group: groupName, id: "0", }); const readStreamMessages = async () => { const chunks = (await redis.xreadgroup( groupName, `consumer-1`, streamKey, // 👇 বিল্ট-ইন রেডিস স্ট্রিম ফাংশনালিটি: শুধু unseen মেসেজ পাঠান ">", )) as StreamData[]; if (chunks?.length > 0) { const [_streamKey, messages] = chunks[0]; for (const [_messageId, fields] of messages) { const rawObj = arrToObj(fields); const validatedMessage = validateMessage(rawObj); if (validatedMessage) { controller.enqueue(json(validatedMessage)); } } } }; // 👇 প্রাথমিক পড়া await readStreamMessages(); const subscription = redis.subscribe(streamKey); subscription.on("message", async () => { // 👇 স্ট্রিমে নতুন চান্ক লেখার সময় প্রতিবার পড়ুন await readStreamMessages(); });

নোট: আমরা প্রতিটি কানেকশনে একটি কনসিউমার গ্রুপ তৈরি করছি। এটি খুব ভালো কাজ করে কারণ রেডিস এই অপারেশনকে идемпোটেন্টলি হ্যান্ডেল করে, অর্থাৎ গ্রুপ ইতিমধ্যে থাকলে কিছুই হয় না।

ফুল কোড

SessionID জেনারেশন

এখন পর্যন্ত, আমরা ক্লায়েন্টের টাস্ক, স্ট্রিম জেনারেটর এবং স্ট্রিম কনজিউমার আলাদা আলাদা বুঝতে পারলাম। এখন দেখি এই টুকরোগুলো কীভাবে একসাথে ফিট হয় কমপ্লিট ইমপ্লিমেন্টেশন দেখে।

শুরু করি, nanoid() ব্যবহার করার চেয়ে সেশন আইডি তৈরি করা আরও রেজিলিয়েন্ট হওয়া উচিত। কেননা, যদি ওয়েবসাইট রিফ্রেশ বা বন্ধ হয়? রিকানেক্ট করার সময়, যদি কোথাও স্টোর না করা হয়, জেনারেটেড সেশন আইডি হারিয়ে যাবে — এটি জেনারেশন চলার পর্যন্ত পারসিস্ট করতে হবে।

সৌভাগ্যবশত localStorage এটার জন্য পারফেক্ট:

import { customAlphabet } from "nanoid"; import { useRouter } from "next/navigation"; import { useCallback, useEffect, useState } from "react"; export const useLLMSession = () => { const [sessionId, setSessionId] = useState<string>(""); const router = useRouter(); const nanoid = customAlphabet("0123456789", 6); const updateUrlWithSessionId = useCallback( (id: string) => { const url = new URL(window.location.href); url.searchParams.set("sessionId", id); router.replace(url.toString(), { scroll: false }); }, [router], ); useEffect(() => { const urlParams = new URLSearchParams(window.location.search); const urlSessionId = urlParams.get("sessionId"); const storedSessionId = localStorage.getItem("llm-session-id"); if (urlSessionId) { localStorage.setItem("llm-session-id", urlSessionId); setSessionId(urlSessionId); } else if (storedSessionId) { setSessionId(storedSessionId); updateUrlWithSessionId(storedSessionId); } else { const newSessionId = nanoid(); localStorage.setItem("llm-session-id", newSessionId); setSessionId(newSessionId); updateUrlWithSessionId(newSessionId); } // eslint-disable-next-line react-hooks/exhaustive-deps }, []); const clearSessionId = useCallback(() => { localStorage.removeItem("llm-session-id"); setSessionId(""); const url = new URL(window.location.href); url.searchParams.delete("sessionId"); router.replace(url.toString(), { scroll: false }); }, [router]); const regenerateSessionId = () => { const newSessionId = nanoid(); localStorage.setItem("llm-session-id", newSessionId); setSessionId(newSessionId); updateUrlWithSessionId(newSessionId); return newSessionId; }; return { sessionId, regenerateSessionId, clearSessionId, }; };

ক্লায়েন্ট

আমরা ইতিমধ্যে ক্লায়েন্টের দুটি সবচেয়ে গুরুত্বপূর্ণ অংশ দেখেছি: একটি স্ট্রিম শুরু করা এবং একটি স্ট্রিমের সাথে কানেক্ট করা। একবার আমাদের API থেকে কনফার্মেশন পেলে যে জেনারেটর চলছে, আমরা react-query-এর refetch ইউটিলিটি ব্যবহার করে আমাদের কানেকশন কোয়েরি আহ্বান করে স্ট্রিমের সাথে কানেক্ট হই।

সকল টুকরো কীভাবে একসাথে ফিট হয়:

"use client" import { useLLMSession } from "@/use-llm-session" import { useMutation, useQuery } from "@tanstack/react-query" import { FormEvent, useRef, useState, useEffect } from "react" import { MessageType, validateMessage, type ChunkMessage, type MetadataMessage, StreamStatus, } from "@/lib/message-schema" // precondition = স্ট্রিম পড়তে প্রস্তুত class PreconditionFailedError extends Error { constructor(message: string) { super(message) this.name = "PreconditionFailedError" } } export default function Home() { const { sessionId, regenerateSessionId, clearSessionId } = useLLMSession() const [prompt, setPrompt] = useState("") const [status, setStatus] = useState< "idle" | "loading" | "streaming" | "completed" | "error" >("idle") const [response, setResponse] = useState("") const [chunkCount, setChunkCount] = useState(0) const controller = useRef<AbortController | null>(null) const responseRef = useRef<HTMLDivElement>(null) const isInitialRequest = useRef(true) // জেনারেশনকে ভিউপোর্টে রাখুন useEffect(() => { if (responseRef.current) { responseRef.current.scrollTop = responseRef.current.scrollHeight } }, [response]) // জেনারেটর শুরু করুন const { mutate, error, isIdle } = useMutation({ mutationFn: async (newSessionId: string) => { controller.current?.abort() isInitialRequest.current = false await fetch("/api/llm-stream", { method: "POST", headers: { "Content-Type": "application/json", }, body: JSON.stringify({ prompt, sessionId: newSessionId }), }) }, onSuccess: () => { setStatus("streaming") refetch() }, }) // চলমান স্ট্রিমের সাথে কানেক্ট করুন const { refetch } = useQuery({ queryKey: ["stream", sessionId], queryFn: async () => { if (!sessionId) return null setResponse("") setChunkCount(0) const abortController = new AbortController() controller.current = abortController const res = await fetch(`/api/check-stream?sessionId=${sessionId}`, { headers: { "Content-Type": "text/event-stream" }, signal: controller.current.signal, }) if (res.status === 412) { // স্ট্রিম এখনো প্রস্তুত নয়, কানেকশন রিট্রাই করুন throw new PreconditionFailedError("Stream not ready yet") } if (!res.body) return null const reader = res.body.pipeThrough(new TextDecoderStream()).getReader() let streamContent = "" while (true) { const { value, done } = await reader.read() if (done) break if (value) { const messages = value.split("\n\n").filter(Boolean) for (const message of messages) { if (message.startsWith("data: ")) { const data = message.slice(6) try { const parsedData = JSON.parse(data) const validatedMessage = validateMessage(parsedData) if (!validatedMessage) continue switch (validatedMessage.type) { case MessageType.CHUNK: const chunkMessage = validatedMessage as ChunkMessage streamContent += chunkMessage.content setResponse((prev) => prev + chunkMessage.content) setChunkCount((prev) => prev + 1) break case MessageType.METADATA: const metadataMessage = validatedMessage as MetadataMessage if (metadataMessage.status === StreamStatus.COMPLETED) { setStatus("completed") } break case MessageType.ERROR: setStatus("error") break } } catch (e) { console.error("Failed to parse message:", e) } } } } } return streamContent }, refetchOnWindowFocus: false, refetchOnMount: false, retry(failureCount, error) { if (isInitialRequest.current === true) return false if (error instanceof PreconditionFailedError) { return failureCount < 10 } return false }, }) const handleSubmit = async (e: FormEvent) => { e.preventDefault() setStatus("loading") const newSessionId = regenerateSessionId() mutate(newSessionId) } const handleReset = () => { controller.current?.abort() clearSessionId() setPrompt("") setResponse("") setChunkCount(0) setStatus("idle") } return ( <main className="flex min-h-screen flex-col items-center justify-between p-12 sm:p-24"> <div className="z-10 max-w-5xl w-full items-center justify-between font-mono text-sm"> <h1 className="text-4xl tracking-tight font-bold mb-8 text-center"> Resumable LLM Stream </h1> <form onSubmit={handleSubmit} className="mb-8"> <div className="mb-4"> <label htmlFor="prompt" className="block text-sm font-medium mb-2"> Enter your prompt: </label> <textarea autoFocus id="prompt" value={prompt} onChange={(e) => setPrompt(e.target.value)} className="w-full p-2 border border-zinc-700 rounded-md min-h-[100px] focus:outline-none focus:ring-2 focus:ring-blue-500 focus:border-transparent transition-all duration-200" placeholder="Ask the AI something..." disabled={status === "loading" || status === "streaming"} /> </div> <div className="flex gap-4"> <button type="submit" disabled={status === "loading" || status === "streaming"} className="px-4 py-2 bg-blue-600 text-white rounded-md hover:bg-blue-700 disabled:bg-gray-400" > {status === "loading" ? "Starting..." : status === "streaming" ? "Streaming..." : "Generate Response"} </button> <button type="button" onClick={handleReset} className="px-4 py-2 bg-zinc-600 text-white rounded-md hover:bg-zinc-700" > Reset </button> </div> </form> <div className="mt-8"> <h2 className="text-xl tracking-tight font-semibold mb-2"> Response: </h2> {status === "error" ? ( <div className="p-4 bg-red-100 border border-red-300 rounded-md text-red-800"> <p className="font-bold">Error:</p> <p>{error?.message}</p> </div> ) : status === "idle" && !response ? ( <p className="text-gray-500"> Enter a prompt and click "Generate Response" to see the AI's response. </p> ) : ( <div ref={responseRef} className="flex flex-col h-96 overflow-y-auto p-4 bg-zinc-900 text-zinc-200 border border-zinc-800 rounded-md whitespace-pre-wrap [&::-webkit-scrollbar]:w-2 [&::-webkit-scrollbar-thumb]:bg-zinc-700 [&::-webkit-scrollbar-track]:bg-zinc-800"> <div>{response || "Loading..."}</div> </div> )} {(status === "streaming" || status === "completed") && ( <div className="mt-2 text-sm text-gray-500"> <p>Session ID: {sessionId}</p> <p>Status: {status}</p> <p>Chunks received: {chunkCount}</p> <p> Connection: {status === "streaming" ? "Active SSE" : "Closed"} </p> </div> )} </div> </div> </main> ) }

স্ট্রিম জেনারেটর

এখানে স্ট্রিম জেনারেটরের পুরো কোড। যদি LLM জেনারেশন কোনো পয়েন্টে ফেল করে, এটি আপস্ট্যাশ ওয়ার্কফ্লো ব্যবহার করে স্বয়ংক্রিয়ভাবে রিট্রাই করা হয় সর্বোচ্চ নির্ভরযোগ্যতার জন্য:

import { MessageType, StreamStatus, type ChunkMessage, type MetadataMessage, } from "@/lib/message-schema"; import { redis } from "@/utils"; import { openai } from "@ai-sdk/openai"; import { serve } from "@upstash/workflow/nextjs"; import { streamText } from "ai"; interface LLMStreamResponse { success: boolean; sessionId: string; totalChunks: number; fullContent: string; } export const { POST } = serve(async (context) => { const { prompt, sessionId } = context.requestPayload as { prompt?: string; sessionId?: string; }; if (!prompt || !sessionId) { throw new Error("Prompt and sessionId are required"); } const streamKey = `llm:stream:${sessionId}`; await context.run("mark-stream-start", async () => { const metadataMessage: MetadataMessage = { type: MessageType.METADATA, status: StreamStatus.STARTED, completedAt: new Date().toISOString(), totalChunks: 0, fullContent: "", }; await redis.xadd(streamKey, "*", metadataMessage); await redis.publish(streamKey, { type: MessageType.METADATA }); }); const res = await context.run("generate-llm-response", async () => { const result = await new Promise<LLMStreamResponse>( async (resolve, reject) => { let fullContent = ""; let chunkIndex = 0; const { textStream } = streamText({ model: openai("gpt-4o"), prompt, onError: (err) => reject(err), onFinish: async () => { resolve({ success: true, sessionId, totalChunks: chunkIndex, fullContent, }); }, }); for await (const chunk of textStream) { if (chunk) { fullContent += chunk; chunkIndex++; const chunkMessage: ChunkMessage = { type: MessageType.CHUNK, content: chunk, }; await redis.xadd(streamKey, "*", chunkMessage); await redis.publish(streamKey, { type: MessageType.CHUNK }); } } }, ); return result; }); await context.run("mark-stream-end", async () => { const metadataMessage: MetadataMessage = { type: MessageType.METADATA, status: StreamStatus.COMPLETED, completedAt: new Date().toISOString(), totalChunks: res.totalChunks, fullContent: res.fullContent, }; await redis.xadd(streamKey, "*", metadataMessage); await redis.publish(streamKey, { type: MessageType.METADATA }); }); });

কমপ্লিট টাইপ-সেফティর জন্য, আমি সকল মেসেজ স্কিমা Zod-এ লিখেছি:

import { z } from "zod"; export const MessageType = { CHUNK: "chunk", METADATA: "metadata", EVENT: "event", ERROR: "error", } as const; export const StreamStatus = { STARTED: "started", STREAMING: "streaming", COMPLETED: "completed", ERROR: "error", } as const; export const baseMessageSchema = z.object({ type: z.enum([ MessageType.CHUNK, MessageType.METADATA, MessageType.EVENT, MessageType.ERROR, ]), }); export const chunkMessageSchema = baseMessageSchema.extend({ type: z.literal(MessageType.CHUNK), content: z.string(), }); export const metadataMessageSchema = baseMessageSchema.extend({ type: z.literal(MessageType.METADATA), status: z.enum([ StreamStatus.STARTED, StreamStatus.STREAMING, StreamStatus.COMPLETED, StreamStatus.ERROR, ]), completedAt: z.string().optional(), totalChunks: z.number().optional(), fullContent: z.string().optional(), error: z.string().optional(), }); export const eventMessageSchema = baseMessageSchema.extend({ type: z.literal(MessageType.EVENT), }); export const errorMessageSchema = baseMessageSchema.extend({ type: z.literal(MessageType.ERROR), error: z.string(), }); export const messageSchema = z.discriminatedUnion("type", [ chunkMessageSchema, metadataMessageSchema, eventMessageSchema, errorMessageSchema, ]); export type Message = z.infer<typeof messageSchema>; export type ChunkMessage = z.infer<typeof chunkMessageSchema>; export type MetadataMessage = z.infer<typeof metadataMessageSchema>; export type EventMessage = z.infer<typeof eventMessageSchema>; export type ErrorMessage = z.infer<typeof errorMessageSchema>; export const validateMessage = (data: unknown): Message | null => { const result = messageSchema.safeParse(data); return result.success ? result.data : null; };

স্ট্রিম কনজিউমার

শেষে, ফুল স্ট্রিম কনজিউমার ইমপ্লিমেন্টেশন দেখি। এটি वह রিপ্লেসেবল কানেকশন যা অটোমেটিক্যালি সমস্ত unseen চান্ক পাঠায় যখন কোনো ক্লায়েন্ট কানেক্ট করে:

import { redis } from "@/utils" import { nanoid } from "nanoid" import { NextRequest, NextResponse } from "next/server" import { validateMessage, MessageType, type ErrorMessage, } from "@/lib/message-schema" export const dynamic = "force-dynamic" export const maxDuration = 60 export const runtime = "nodejs" type StreamField = string type StreamMessage = [string, StreamField[]] type StreamData = [string, StreamMessage[]] const arrToObj = (arr: StreamField[]) => { const obj: Record<string, string> = {} for (let i = 0; i < arr.length; i += 2) { obj[arr[i]] = arr[i + 1] } return obj } const json = (data: Record<string, unknown>) => { return new TextEncoder().encode(`data: ${JSON.stringify(data)}\n\n`) } export async function GET(req: NextRequest) { const { searchParams } = new URL(req.url) const sessionId = searchParams.get("sessionId") if (!sessionId) { return NextResponse.json( { error: "Stream key is required" }, { status: 400 } ) } const streamKey = `llm:stream:${sessionId}` const groupName = `sse-group-${nanoid()}` const keyExists = await redis.exists(streamKey) if (!keyExists) { return NextResponse.json( { error: "Stream does not (yet) exist" }, { status: 412 } ) } try { await redis.xgroup(streamKey, { type: "CREATE", group: groupName, id: "0", }) } catch (_err) {} const response = new Response( new ReadableStream({ async start(controller) { const readStreamMessages = async () => { const chunks = (await redis.xreadgroup( groupName, `consumer-1`, streamKey, ">" )) as StreamData[] if (chunks?.length > 0) { const [_streamKey, messages] = chunks[0] for (const [_messageId, fields] of messages) { const rawObj = arrToObj(fields) const validatedMessage = validateMessage(rawObj) if (validatedMessage) { controller.enqueue(json(validatedMessage)) } } } } await readStreamMessages() const subscription = redis.subscribe(streamKey) subscription.on("message", async () => { await readStreamMessages() }) subscription.on("error", (error) => { console.error(`SSE subscription error on ${streamKey}:`, error) const errorMessage: ErrorMessage = { type: MessageType.ERROR, error: error.message, } controller.enqueue(json(errorMessage)) controller.close() }) req.signal.addEventListener("abort", () => { console.log("Client disconnected, cleaning up subscription") subscription.unsubscribe() controller.close() }) }, }), { headers: { "Content-Type": "text/event-stream", "Cache-Control": "no-cache, no-transform", Connection: "keep-alive", }, } ) return response }

দ্রুত সারসংক্ষেপ এবং শেষ কথা

আমরা vừa একটি অত্যন্ত রবাস্ট LLM স্ট্রিম বিল্ড করেছি যা নেটওয়ার্ক ইন্টারাপশন, পেজ রিফ্রেশ, এবং সম্পূর্ণ ডিসকানেকশন হ্যান্ডেল করতে পারে। এখানে আমরা যা করেছি:

• জেনারেশন ডেলিভারি থেকে ডিকাপল করা: LLM জেনারেশনকে ক্লায়েন্ট কানেকশন থেকে আলাদা করে, কন্টেন্ট জেনারেশন ক্লায়েন্ট সমস্যা নির্বিশেষে চলতে থাকে।

• রেডিস স্ট্রিম ব্যবহার করে পারসিস্টেন্ট স্টোরেজ: আমরা রেডিস স্ট্রিমকে একটি পারসিস্টেন্ট মেসেজ ব্রোকার হিসেবে ব্যবহার করছি যাতে LLM রেসপন্সের প্রতিটি চান্ক জেনারেট হওয়ার সাথে সাথে সংরক্ষিত হয়।

• রেডিস পাব/সাব দিয়ে রিয়েল-টাইম আপডেট: আমরা একটি নোটিফিকেশন সিস্টেম বিল্ড করেছি রেডিস পাব/সাব ব্যবহার করে যাতে স্ট্রিম কনজিউমারকে নতুন চান্ক উপলব্ধ হওয়ার বিষয়ে জানানো যায়।

• অটোমেটিক রিকানেকশন: ক্লায়েন্ট যেকোনো সময় রিকানেক্ট করতে পারে এবং সকল কন্টেন্ট স্বয়ংক্রিয়ভাবে পায়, গ্যারান্টিডভাবে কোনো ডুপ্লিকেট বা মিসিং চান্ক ছাড়া। এতে ডিসকানেক্ট হওয়ার সময় জেনারেট হওয়া কন্টেন্টও অন্তর্ভুক্ত।

• সেশন ম্যানেজমেন্ট: আমরা একটি সেশন সিস্টেম তৈরি করেছি যা ইউজারকে একাধিক ডিভাইসে একই সাথে স্ট্রিম দেখতে দেয়।

নিচের লাইন, আমরা আমাদের ইউজারদের একটি এক্সসেপশনাল ইউজার এক্সপেরিয়েন্স (UX) প্রদান করছি। আমি এই অ্যাপ্রোচ খুব সুপারিশ করি, বিশেষ করে যদি আপনি LLM চ্যাট সার্ভিসের মতো কিছু বিল্ড করেন।

পড়ার জন্য ধন্যবাদ! আপনার কোনো ফিডব্যাক থাকলে বা আপস্ট্যাশে গেস্ট অথর হতে চান তবে josh@upstash.com-এ যোগাযোগ করুন 🙌