Real-Time LLM Streaming ใน Next.js ด้วย Server-Sent Events (SSE)

AI-powered applications ที่ตอบสนองอย่างรวดเร็วเพิ่มประสบการณ์ผู้ใช้โดยทำให้พวกเขารู้สึกได้ยินและถูกมองเห็น ทันที ด้วย streaming คุณสามารถสร้าง chatbots ที่ตอบคำถามหรือส่ง personalized recommendations แบบ real-time ได้ ไม่ได้เกี่ยวกับความเร็วเพียงอย่างเดียว - แต่เกี่ยวกับการตอบสนองความต้องการของผู้ใช้ทันที เพิ่มค่าที่รับรู้ได้

Streaming เกี่ยวข้องกับการส่งข้อมูลใน small, continuous chunks แทนที่จะเป็น one large block ในบริบทของ AI applications เช่น chatbots หรือ recommendation systems streaming หมายถึงการส่ง partial responses ไปยังผู้ใช้ทันทีที่พร้อมใช้งาน แทนที่จะรอ compile และส่ง entire response ทีเดียว วิธีนี้รับรองว่าผู้ใช้ได้รับ feedback หรือข้อมูลทันที สร้าง user experience ที่ dynamic และ responsive มากขึ้น

ในคู่มือนี้ คุณจะได้เรียนรู้วิธีใช้ประโยชน์จาก Server-Sent Events (SSE) เพื่อ implement streaming ใน Next.js endpoints ด้วย LangChain และ OpenAI's language models เพิ่มเติม คุณจะได้เรียนรู้วิธี cache streaming responses ด้วย Upstash

ข้อกำหนดเบื้องต้น (Prerequisites)

คุณจะต้องมีสิ่งต่อไปนี้:

• Node.js 18 หรือใหม่กว่า
• บัญชี OpenAI
• บัญชี Upstash
• บัญชี Vercel

Tech Stack

สร้าง OpenAI Token

ด้วย OpenAI API คุณสามารถสร้าง chatbot responses โดยใช้ AI การ request ใดๆ ไปยัง OpenAI API ต้องการ authorization token เพื่อได้ token ให้ไปที่ API Keys ในบัญชี OpenAI ของคุณ และคลิกปุ่ม Create new secret key คัดลอกและเก็บ token นี้ไว้อย่างปลอดภัยสำหรับการใช้งานภายหลังเป็น environment variable OPENAI_API_KEY

ตั้งค่า Upstash Redis

เมื่อคุณสร้างบัญชี Upstash และ login แล้ว คุณจะไปที่แท็บ Redis และสร้างฐานข้อมูล

เลื่อนลงจนกว่าจะเจอส่วน REST API และเลือกปุ่ม .env คัดลอกเนื้อหาและเก็บไว้ที่ปลอดภัย

สร้าง Next.js Application ใหม่

มาเริ่มต้นด้วยการสร้าง Next.js project ใหม่ เปิดเทอร์มินัลและรันคำสั่งต่อไปนี้:

npx create-next-app@latest my-app

เมื่อถูกพรอมต์ ให้เลือก:

• Yes เมื่อถูกถามว่าจะใช้ TypeScript หรือไม่
• No เมื่อถูกถามว่าจะใช้ ESLint หรือไม่
• Yes เมื่อถูกถามว่าจะใช้ Tailwind CSS หรือไม่
• No เมื่อถูกถามว่าจะใช้ไดเรกทอรี src/ หรือไม่
• Yes เมื่อถูกถามว่าจะใช้ App Router หรือไม่
• No เมื่อถูกถามว่าจะปรับแต่ง default import alias (@/*) หรือไม่

เมื่อเสร็จแล้ว ให้เข้าไปในไดเรกทอรีโปรเจกต์และเริ่มแอปในโหมดพัฒนาด้วยการรันคำสั่งต่อไปนี้:

cd my-app npm run dev

แอปควรทำงานบน localhost:3000 หยุด development server เพื่อติดตั้ง LangChain dependencies ในแอปพลิเคชันด้วยการรันคำสั่งต่อไปนี้:

npm install @langchain/openai @langchain/community

คำสั่งติดตั้งไลบรารีต่อไปนี้:

• @langchain/openai: แพ็กเกจ LangChain สำหรับ interface กับ OpenAI series ของ models
• @langchain/community: ชุด third party integrations สำหรับ plug-n-play กับ LangChain core

ตอนนี้ สร้างไฟล์ .env ที่รากของโปรเจกต์ คุณจะเพิ่ม OPENAI_API_KEY ที่ได้ก่อนหน้านี้

ควรมีลักษณะประมาณนี้:

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

เพื่อสร้าง API endpoints ใน Next.js คุณจะใช้ Next.js Route Handlers ซึ่งอนุญาตให้คุณ serve responses ผ่าน Web Request และ Response APIs เพื่อเริ่มสร้าง API route ใน Next.js ที่ stream responses ไปยังผู้ใช้ ให้รันคำสั่งต่อไปนี้:

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

Flag -p สร้าง parent directories สำหรับไดเรกทอรี chat หากหายไป

นี่ตั้งค่า Next.js project ของเรา ตอนนี้ มาไปสร้าง cache handler ด้วย Upstash

สร้าง Upstash Redis Cache Client Instance

ด้วย class UpstashRedisCache คุณสามารถ initialize Redis cache client สำหรับ cache OpenAI API responses ภายในไม่กี่บรรทัดของโค้ด สร้างไฟล์ upstashCache.tsx ภายในไดเรกทอรี app/lib:

ด้วย cache handler การ cache conversations หรือ completion responses ที่สร้างโดย OpenAI API จะกลายเป็น super easy มาเข้าใจว่าคุณสามารถเริ่มใช้ Server-Sent Events ใน Next.js ได้อย่างไร

สร้าง Server-Sent Events API ใน Next.js App Router

Server-Sent Events (SSE) อนุญาตให้คุณส่ง real-time data updates จาก server ไปยัง client โดยไม่ต้อง continuous polling SSE เปิดใช้งาน unidirectional flow ของข้อมูลผ่าน single, long-lived HTTP connection

สร้างไฟล์ชื่อ route.ts ด้วยโค้ดต่อไปนี้ในไดเรกทอรี app/api/stream เพื่อเข้าใจ minimal setup ที่จำเป็นสำหรับ stream responses โดยใช้ Server-Sent Events:

// 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", }, }) }

โค้ดด้านบนทำสิ่งต่อไปนี้:

• ค่าคงที่ dynamic ถูกตั้งเป็น force-dynamic ป้องกันการ cache responses บน platform Vercel นี่รับรองว่าแต่ละ request สำหรับ SSE stream ได้รับ fresh data และไม่ได้ถูก serve จาก cache
• ReadableStream ถูกสร้างเพื่อ generate stream of data เพื่อส่งไปยัง client ใน start method ของ stream ข้อความถูก encode และ enqueue เข้า stream's controller ข้อความนี้จะถูกส่งไปยัง client เป็นส่วนหนึ่งของ SSE stream
• Object Response ถูกสร้างพร้อม headers เฉพาะสำหรับ Server-Sent Events headers เหล่านี้รวมถึง:
  • Connection: keep-alive เพื่อรักษา connection เปิดไว้สำหรับ streaming
  • Cache-Control: no-cache, no-transform เพื่อป้องกัน caching ใน browser
  • Content-Type: text/event-stream; charset=utf-8 เพื่อระบุ content type เป็น Server-Sent Events
• Response object ที่มี custom ReadableStream ถูกคืนค่าจาก endpoint Response นี้จะถูกส่งไปยัง clients ที่ request SSE stream

ด้วยสิ่งนี้ คุณได้สร้าง endpoint ที่สามารถ stream responses ได้ ในส่วนต่อไป คุณจะเข้าใจว่าคุณสามารถ stream Completion และ Chat Completion API responses จาก OpenAI ด้วย LangChain callbacks ได้อย่างไร

Streaming OpenAI Completion API responses ด้วย LangChain Callbacks

OpenAI Completion API ให้เครื่องมือทรงพลังสำหรับ generate text จาก prompt(s) ที่ให้มา มีประโยชน์ใน real-life scenarios ต่างๆ เช่น content generation และ language translation การผสานรวม LangChain callbacks เปิดใช้งาน real-time streaming ของ responses เพิ่ม perceived responsiveness ใน applications

สร้างไฟล์ชื่อ completionModel.tsx ในไดเรกทอรี app/lib กำหนดฟังก์ชันที่ initialize OpenAI instance พร้อม streaming enabled และ LangChain callbacks generate text แบบ real-time

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

โค้ดด้านบนทำสิ่งต่อไปนี้:

• Import class OpenAI จากแพ็กเกจ @langchain/openai
• Export ฟังก์ชันชื่อ completionModel ซึ่งรับสองพารามิเตอร์:
  • controller: object ReadableStreamDefaultController ซึ่งอนุญาตให้คุณ enqueue messages เข้า stream
  • encoder: object TextEncoder ที่ encode messages เป็น object Uint8Array
• สร้าง instance ใหม่ของ class OpenAI ด้วยการกำหนดค่าต่อไปนี้:
  • streaming: ตั้งเป็น true บ่งชี้ว่า API ควรคืนค่า streaming responses
  • callbacks: อาร์เรย์ที่มี object เดียวพร้อม callback functions สองตัว ใช้สำหรับ intercept response creation โดย LLM ที่แต่ละขั้นตอน
• handleLLMNewToken: Callback ที่ถูก invoke เมื่อ new token ถูก generate โดย language model มัน enqueue encoded token ไปยัง controller
• handleLLMEnd: Asynchronous callback ที่ถูก invoke เมื่อ language model generation เสร็จสิ้น มัน close controller

เพิ่มเติม สร้างไฟล์ชื่อ route.ts ด้วยโค้ดต่อไปนี้ในไดเรกทอรี app/api/stream/completion การเปลี่ยนแปลงจาก initial streaming route handler ที่คุณสร้างถูก highlight ด้านล่าง:

// 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", }, }); }

นอกจากสิ่งที่คุณทำก่อนหน้านี้เพื่อสร้าง streaming endpoint ใน Next.js การเพิ่มโค้ดด้านบนทำสิ่งต่อไปนี้:

• Import ฟังก์ชัน completionModel
• Extract user message จาก request body โดยใช้ request.json()
• ภายใน method start ของ ReadableStream ฟังก์ชัน completionModel ถูก invoke แบบ asynchronous พร้อม message ที่ได้จาก request ฟังก์ชันนี้ generate streaming response จาก OpenAI ด้วย LangChain

เพื่อ cache user queries และ responses ของพวกเขาด้วย Upstash ให้นำเข้า Upstash Redis cache client เข้าไปในฟังก์ชัน 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(); }, }, ], });

ด้วยการเปลี่ยนแปลงน้อยที่สุด คุณได้เปิดใช้งาน caching ของ OpenAI Completion API responses ด้วย Upstash

ตอนนี้ มาเรียนรู้ว่าการ stream OpenAI Chat Completion API responses ใน Next.js endpoints ทำได้อย่างไร

Streaming OpenAI Chat Completion API responses ด้วย LangChain Callbacks

OpenAI Chat Completion API ให้เครื่องมือทรงพลังสำหรับ generate real-time conversational responses จาก messages ที่ให้มา Applications ครอบคลุมจากการสร้าง chatbots ที่สามารถ engaged ใน natural dialogues ไปจนถึงการเพิ่มประสิทธิภาพ customer support systems ด้วย AI-driven interaction capabilities มันช่วยให้นักพัฒนาสร้าง conversational interfaces ที่ dynamic และ responsive

สร้างไฟล์ชื่อ chatCompletionModel.tsx ในไดเรกทอรี app/lib กำหนดฟังก์ชัน chatCompletionModel ซึ่งสร้าง streaming interface สำหรับ OpenAI's Chat Completion API ด้วย LangChain callbacks อำนวยความสะดวกในการ generate real-time conversation จาก messages ที่ให้มา

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

โค้ดด้านบนทำสิ่งต่อไปนี้:

• Import class ChatOpenAI จากแพ็กเกจ @langchain/openai
• สร้าง type definitions ต่อไปนี้:
  • ConversationMessage: กำหนด type สำหรับ single message ใน conversation ประกอบด้วยสอง properties: role (บ่งชี้ role ของ sender) และ content (message content)
  • ConversationMessages: กำหนด type สำหรับ array ของ conversation messages
• Export ฟังก์ชันชื่อ chatCompletionModel ซึ่งรับสองพารามิเตอร์:
  • controller: object ReadableStreamDefaultController ซึ่งอนุญาตให้คุณ enqueue messages เข้า stream
  • encoder: object TextEncoder ที่ encode messages เป็น object Uint8Array
• สร้าง instance ใหม่ของ class ChatOpenAI ด้วยการกำหนดค่าต่อไปนี้:
  • streaming: ตั้งเป็น true บ่งชี้ว่า API ควรคืนค่า streaming responses
  • callbacks: อาร์เรย์ที่มี object เดียวพร้อม callback functions สองตัว ใช้สำหรับ intercept response creation โดย LLM ที่แต่ละขั้นตอน
• handleLLMNewToken: Callback ที่ถูก invoke เมื่อ new token ถูก generate โดย language model มัน enqueue encoded token ไปยัง controller
• handleLLMEnd: Asynchronous callback ที่ถูก invoke เมื่อ language model generation เสร็จสิ้น มัน close controller

เพิ่มเติม สร้างไฟล์ชื่อ route.ts ด้วยโค้ดต่อไปนี้ในไดเรกทอรี app/api/stream/completion/chat การเปลี่ยนแปลงจาก initial streaming route handler ที่คุณสร้างถูก highlight ด้านล่าง:

// 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", }, }); }

นอกจากสิ่งที่คุณทำก่อนหน้านี้เพื่อสร้าง streaming endpoint ใน Next.js การเพิ่มโค้ดด้านบนทำสิ่งต่อไปนี้:

• Import ฟังก์ชัน chatCompletionModel และ type definition ConversationMessage
• Extract messages ใน conversation จาก request body โดยใช้ request.json()
• ภายใน method start ของ ReadableStream ฟังก์ชัน chatCompletionModel ถูก invoke แบบ asynchronous หลังจาก transform messages array แต่ละ message ถูก transform เป็น array ที่มี role และ content ของ message ฟังก์ชันนี้ generate streaming response จาก OpenAI ด้วย LangChain

เพื่อ cache chat history ด้วย Upstash ให้นำเข้า Upstash Redis cache client เข้าไปในฟังก์ชัน 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(); }, }, ], });

ด้วยการเปลี่ยนแปลงน้อยที่สุด คุณได้เปิดใช้งาน caching ของ chat history ทั้งหมดด้วย Upstash

ตอนนี้ มา setup Next.js frontend เพื่อ listen to Server-Sent Events

Setup Next.js frontend เพื่อ listen to Server-Sent Events

ในส่วนนี้ คุณจะได้เรียนรู้วิธี setup minimal listener สำหรับ Server-Sent Events API messages และ approach ในการจัดการ state ของ messages ที่แลกเปลี่ยนระหว่าง user และ AI

Listening to Server-Sent Events API ใน React

เพื่อ listen to SSE API ใน client-side component ของเราใน React คุณจะ initiate POST request ไปยัง API route ที่ระบุ เมื่อได้รับ response คุณจะ decode incoming data เป็น string โดยใช้ method TextDecoderStream และอ่าน data จาก stream อย่างต่อเนื่อง

"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 <></> }

จัดการ state ของ conversation ใน React

เพื่อจัดการ state ของ conversation ระหว่าง user และ chatbot คุณสามารถใช้ state variables คุณจะ update state variables messages และ latestMessage เพื่อจัดเก็บ conversation history และ latest message ตามลำดับ โดยการ update state variables เหล่านี้ภายใน loop ที่ประมวลผล incoming data จาก SSE API คุณสามารถรับรอง real-time updates ของ conversation display การเพิ่มโค้ดเพื่อให้ได้ผลลัพธ์เดียวกันมีดังนี้:

"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 <></> }

การเพิ่มโค้ดด้านบนทำสิ่งต่อไปนี้:

• State variable ถูกประกาศเพื่อจัดเก็บ incoming data stream ที่ได้รับจาก SSE API
• เมื่อได้รับ complete message (บ่งชี้โดย done เป็น true) latest received message จะถูกเพิ่มเข้าไปใน state array messages Message นี้ถูก format ด้วย role ของ assistant เพื่อแยกจาก user messages นอกจากนี้ latest message state จะถูก reset เป็น empty string เพื่อเตรียมพร้อมสำหรับ incoming message ถัดไป
• ขณะที่ data ถูกส่งมาทีละนิด (ยังไม่สร้าง complete message) มันจะถูก append เข้าไปในตัวแปร incomingMessage
• Latest message state ถูก update ด้วย concatenated incoming data รับรอง real-time updates ขณะที่ data ใหม่มาถึง

ยอดเยี่ยม! ด้วย reactive latestMessage และ messages variables คุณสามารถสร้าง dynamic user interfaces ที่แทน messages ที่แลกเปลี่ยนใน conversation และ latest AI generated response ได้แล้ว

นั่นคือการเรียนรู้เยอะมาก! คุณเสร็จสิ้นแล้ว ✨

Deploy ไปที่ Vercel

Repository ตอนนี้พร้อมที่จะ deploy ไปที่ Vercel ใช้ขั้นตอนต่อไปนี้เพื่อ deploy:

• เริ่มต้นด้วยการสร้าง GitHub repository ที่มี code ของแอปของคุณ
• จากนั้น ไปที่ Vercel Dashboard และสร้าง New Project
• Link โปรเจกต์ใหม่กับ GitHub repository ที่คุณเพิ่งสร้าง
• ใน Settings อัปเดต Environment Variables ให้ตรงกับในไฟล์ .env local ของคุณ
• Deploy! 🚀

More Information

สำหรับข้อมูลเชิงลึกเพิ่มเติม สำรวจอ้างอิงที่อ้างอิงในโพสต์นี้

• GitHub Repository
• Server Sent Events
• Next.js Streaming
• Caching layer for LLMs