Streaming LLM Thời Gian Thực Trong Next.js Sử Dụng Server-Sent Events (SSE)

Các ứng dụng hỗ trợ AI phản hồi nhanh nâng cao trải nghiệm người dùng bằng cách khiến họ cảm thấy được lắng nghe và nhìn thấy, tức thì. Với streaming, bạn có thể tạo chatbot như vậy để giải quyết truy vấn tức thì hoặc cung cấp đề xuất cá nhân hóa theo thời gian thực. Không chỉ về tốc độ—đó là về đáp ứng nhu cầu người dùng nhanh chóng, tăng giá trị cảm nhận.

Streaming liên quan đến việc gửi dữ liệu trong các khối nhỏ, liên tục thay vì trong một khối lớn. Trong ngữ cảnh ứng dụng AI, như chatbot hoặc hệ thống đề xuất, streaming nghĩa là gửi phản hồi một phần đến người dùng ngay khi chúng có sẵn, thay vì chờ biên soạn và gửi toàn bộ phản hồi cùng lúc. Cách tiếp cận này đảm bảo người dùng nhận được phản hồi hoặc thông tin ngay lập tức, tạo trải nghiệm người dùng động và phản hồi hơn.

Trong hướng dẫn này, bạn sẽ học cách tận dụng Server-Sent Events (SSE) để triển khai streaming trong endpoint Next.js với LangChain và mô hình ngôn ngữ OpenAI. Thêm vào đó, bạn sẽ học cách cache các phản hồi streaming với Upstash.

Điều Kiện Tiên Quyết

Bạn sẽ cần những điều sau:

• Node.js 18 hoặc mới hơn
• Tài khoản OpenAI
• Tài khoản Upstash
• Tài khoản Vercel

Technology Stack

Tạo Token OpenAI

Sử dụng OpenAI API, bạn có thể tạo phản hồi chatbot sử dụng AI. Bất kỳ yêu cầu nào đến OpenAI API đều yêu cầu token ủy quyền. Để lấy token, điều hướng đến API Keys trong tài khoản OpenAI của bạn, và nhấp nút Create new secret key. Sao chép và lưu trữ an toàn token này để dùng sau làm biến môi trường OPENAI_API_KEY.

Thiết Lập Upstash Redis

Khi bạn đã tạo tài khoản Upstash và đăng nhập, bạn sẽ đi đến tab Redis và tạo cơ sở dữ liệu.

Cuộn xuống cho đến khi tìm thấy phần REST API và chọn nút .env. Sao chép nội dung và lưu ở nơi an toàn.

Tạo Ứng Dụng Next.js Mới

Hãy bắt đầu bằng cách tạo dự án Next.js mới. Mở terminal và chạy lệnh sau:

npx create-next-app@latest my-app

Khi được nhắc, hãy chọn:

• Yes khi được nhắc sử dụng TypeScript.
• No khi được nhắc sử dụng ESLint.
• Yes khi được nhắc sử dụng Tailwind CSS.
• No khi được nhắc sử dụng thư mục src/.
• Yes khi được nhắc sử dụng App Router.
• No khi được nhắc tùy chỉnh alias nhập mặc định (@/*).

Sau khi xong, di chuyển vào thư mục dự án và khởi động ứng dụng ở chế độ phát triển bằng lệnh sau:

cd my-app npm run dev

Ứng dụng sẽ chạy trên localhost:3000. Dừng máy chủ phát triển để cài đặt phụ thuộc LangChain trong ứng dụng bằng lệnh sau:

npm install @langchain/openai @langchain/community

Lệnh đã cài đặt thư viện sau:

• @langchain/openai: Gói LangChain để giao diện với loạt mô hình OpenAI.
• @langchain/community: Bộ sưu tập tích hợp bên thứ ba cho plug-n-play với LangChain core.

Bây giờ, tạo tệp .env tại thư mục gốc dự án. Bạn sẽ thêm OPENAI_API_KEY bạn đã lấy trước đó.

Nó nên trông giống như sau:

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

Để tạo API endpoint trong Next.js, bạn sẽ dùng Next.js Route Handlers cho phép phục vụ phản hồi qua Web Request và Response APIs. Để bắt đầu tạo API route trong Next.js stream phản hồi đến người dùng, thực hiện lệnh sau:

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

Cờ -p tạo thư mục cha cho thư mục chat nếu chúng thiếu.

Điều này thiết lập dự án Next.js của chúng ta. Bây giờ, hãy chuyển sang tạo cache handler với Upstash.

Khởi Tạo Client Cache Upstash Redis

Với lớp UpstashRedisCache, bạn có thể khởi tạo client cache Redis để cache phản hồi OpenAI API trong vài dòng mã. Tạo tệp upstashCache.tsx bên trong thư mục app/lib:

Sử dụng cache handler, việc cache các cuộc hội thoại hoặc phản hồi completion được tạo bởi OpenAI API trở nên siêu dễ dàng. Hãy chuyển sang hiểu cách bạn có thể bắt đầu sử dụng Server-Sent Events trong Next.js.

Tạo API Server-Sent Events Trong Next.js App Router

Server-Sent Events (SSE) cho phép bạn truyền cập nhật dữ liệu thời gian thực từ server đến client mà không cần polling liên tục. SSE cho phép luồng dữ liệu một chiều qua một kết nối HTTP đơn lẻ, tồn tại lâu.

Tạo tệp tên route.ts với mã sau trong thư mục app/api/stream để hiểu thiết lập tối thiểu cần thiết để stream phản hồi sử dụng 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", }, }) }

Mã trên làm những điều sau:

• Hằng số dynamic được đặt thành force-dynamic, ngăn chặn caching phản hồi trên nền tảng Vercel. Điều này đảm bảo mỗi yêu cầu cho SSE stream nhận dữ liệu mới và không được phục vụ từ cache.
• Một ReadableStream được tạo để tạo luồng dữ liệu gửi đến client. Trong phương thức start của stream, một tin nhắn được mã hóa và đưa vào hàng đợi của controller của stream. Tin nhắn này sẽ được gửi đến client như một phần của SSE stream.
• Đối tượng Response được tạo với header đặc thù cho Server-Sent Events. Các header này bao gồm:
  • Connection: keep-alive để duy trì kết nối mở cho streaming.
  • Cache-Control: no-cache, no-transform để ngăn chặn caching trong trình duyệt.
  • Content-Type: text/event-stream; charset=utf-8 để chỉ định loại nội dung là Server-Sent Events.
• Đối tượng Response chứa custom ReadableStream được trả về từ endpoint. Phản hồi này sẽ được gửi đến các client yêu cầu SSE stream.

Với điều này, bạn đã tạo một endpoint có thể stream phản hồi. Trong các phần tiếp theo, bạn sẽ hiểu cách stream phản hồi Completion và Chat Completion API từ OpenAI với LangChain callbacks.

Streaming Phản Hồi OpenAI Completion API Với LangChain Callbacks

OpenAI Completion API cung cấp công cụ mạnh mẽ để tạo văn bản dựa trên prompt cho trước, hữu ích trong nhiều kịch bản thực tế như tạo nội dung và dịch ngôn ngữ. Tích hợp LangChain callbacks cho phép streaming phản hồi theo thời gian thực, tăng khả năng phản hồi cảm nhận trong ứng dụng.

Tạo tệp tên completionModel.tsx trong thư mục app/lib, định nghĩa hàm khởi tạo instance OpenAI với streaming được bật và LangChain callbacks, tạo văn bản theo thời gian thực.

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

Mã trên làm những điều sau:

• Nhập lớp OpenAI từ gói @langchain/openai.
• Xuất hàm tên completionModel nhận hai tham số:
  • controller: Đối tượng ReadableStreamDefaultController cho phép bạn đưa tin nhắn vào hàng đợi của stream.
  • encoder: Đối tượng TextEncoder mã hóa tin nhắn thành đối tượng Uint8Array.
• Tạo instance mới của lớp OpenAI với các tùy chọn cấu hình sau:
  • streaming: Đặt thành true, chỉ ra API nên trả về phản hồi streaming.
  • callbacks: Mảng chứa một đối tượng với hai hàm callback, dùng để chặn tạo phản hồi bởi LLM ở mỗi giai đoạn.
• handleLLMNewToken: Callback được gọi khi token mới được tạo bởi mô hình ngôn ngữ. Nó đưa token đã mã hóa vào hàng đợi của controller.
• handleLLMEnd: Callback bất đồng bộ được gọi khi tạo ngôn ngữ hoàn tất. Nó đóng controller.

Tiếp theo, tạo tệp tên route.ts với mã sau trong thư mục app/api/stream/completion. Những thay đổi từ route handler streaming ban đầu bạn đã tạo đã được làm nổi bật bên dưới:

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

Ngoài những gì bạn đã làm trước để tạo endpoint streaming trong Next.js, các bổ sung mã trên làm những điều sau:

• Nhập hàm completionModel.
• Trích xuất tin nhắn người dùng từ body request sử dụng request.json().
• Bên trong phương thức start của ReadableStream, hàm completionModel được gọi bất đồng bộ với tin nhắn lấy từ request. Hàm này tạo phản hồi streaming từ OpenAI với LangChain.

Để cache truy vấn người dùng và phản hồi của họ với Upstash, nhập client cache Upstash Redis vào hàm completionModel như sau:

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

Với những thay đổi tối thiểu, bạn đã bật caching phản hồi OpenAI Completion API với Upstash.

Bây giờ, hãy chuyển sang học cách phản hồi OpenAI Chat Completion API có thể được stream trong endpoint Next.js.

Streaming Phản Hồi OpenAI Chat Completion API Với LangChain Callbacks

OpenAI Chat Completion API cung cấp công cụ mạnh mẽ để tạo phản hồi hội thoại thời gian thực dựa trên tin nhắn được cung cấp. Ứng dụng trải từ xây dựng chatbot có khả năng tham gia đối thoại tự nhiên đến nâng cao hệ thống hỗ trợ khách hàng với khả năng tương tác AI. Nó cho phép nhà phát triển tạo giao diện hội thoại động và phản hồi.

Tạo tệp tên chatCompletionModel.tsx trong thư mục app/lib, định nghĩa hàm chatCompletionModel tạo giao diện streaming cho OpenAI Chat Completion API với LangChain callbacks, tạo hội thoại thời gian thực dựa trên tin nhắn được cung cấp.

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

Mã trên làm những điều sau:

• Nhập lớp ChatOpenAI từ gói @langchain/openai.
• Tạo các định nghĩa kiểu sau:
  • ConversationMessage: Định nghĩa kiểu cho một tin nhắn đơn lẻ trong hội thoại. Nó chứa hai thuộc tính: role (chỉ ra vai trò của người gửi) và content (nội dung tin nhắn).
  • ConversationMessages: Định nghĩa kiểu cho mảng tin nhắn hội thoại.
• Xuất hàm tên chatCompletionModel nhận hai tham số:
  • controller: Đối tượng ReadableStreamDefaultController cho phép bạn đưa tin nhắn vào hàng đợi stream.
  • encoder: Đối tượng TextEncoder mã hóa tin nhắn thành đối tượng Uint8Array.
• Tạo instance mới của lớp ChatOpenAI với các tùy chọn cấu hình sau:
  • streaming: Đặt thành true, chỉ ra API nên trả về phản hồi streaming.
  • callbacks: Mảng chứa một đối tượng với hai hàm callback, dùng để chặn tạo phản hồi bởi LLM ở mỗi giai đoạn.
• handleLLMNewToken: Callback được gọi khi token mới được tạo bởi mô hình ngôn ngữ. Nó đưa token đã mã hóa vào hàng đợi controller.
• handleLLMEnd: Callback bất đồng bộ được gọi khi tạo ngôn ngữ hoàn tất. Nó đóng controller.

Tiếp theo, tạo tệp tên route.ts với mã sau trong thư mục app/api/stream/completion/chat. Những thay đổi từ route handler streaming ban đầu bạn đã tạo đã được làm nổi bật bên dưới:

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

Ngoài những gì bạn đã làm trước để tạo endpoint streaming trong Next.js, các bổ sung mã trên làm những điều sau:

• Nhập hàm chatCompletionModel và định nghĩa kiểu ConversationMessage.
• Trích xuất tin nhắn trong hội thoại từ body request sử dụng request.json().
• Bên trong phương thức start của ReadableStream, hàm chatCompletionModel được gọi bất đồng bộ sau khi biến đổi mảng tin nhắn. Mỗi tin nhắn được biến đổi thành mảng chứa role và content của tin nhắn. Hàm này tạo phản hồi streaming từ OpenAI với LangChain.

Để cache lịch sử chat với Upstash, nhập client cache Upstash Redis vào hàm chatCompletionModel như sau:

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

Với những thay đổi tối thiểu, bạn đã bật caching toàn bộ lịch sử chat với Upstash.

Bây giờ, hãy chuyển sang tạo component React phía client để tiêu thụ các endpoint streaming.

Thiết Lập Frontend Next.js Để Lắng Nghe Server-Sent Events

Trong phần này, bạn sẽ học cách thiết lập một listener tối thiểu cho tin nhắn API Server-Sent Events, và một cách tiếp cận để quản lý state của tin nhắn được trao đổi giữa người dùng và AI.

Lắng Nghe Server-Sent Events API Trong React

Để lắng nghe SSE API trong component client-side trong React, bạn sẽ khởi tạo request POST đến API route được chỉ định. Khi nhận được phản hồi, bạn sẽ giải mã dữ liệu đến dưới dạng chuỗi sử dụng phương thức TextDecoderStream, và liên tục đọc dữ liệu từ 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 <></> }

Quản Lý State Của Hội Thoại Trong React

Để quản lý state của hội thoại giữa người dùng và chatbot, bạn có thể dùng biến state. Bạn sẽ cập nhật biến state messages và latestMessage để lưu lịch sử hội thoại và tin nhắn mới nhất, tương ứng. Bằng cách cập nhật các biến state này bên trong vòng lặp xử lý dữ liệu đến từ SSE API, bạn có thể đảm bảo cập nhật thời gian thực cho hiển thị hội thoại. Các bổ sung mã để đạt được điều đó như sau:

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

Các bổ sung mã trên làm những điều sau:

• Một biến state được khai báo để lưu luồng dữ liệu đến nhận được từ SSE API.
• Khi nhận được tin nhắn hoàn chỉnh (được chỉ ra bởi done là true), tin nhắn mới nhận được được thêm vào mảng state messages. Tin nhắn này được định dạng với role assistant để phân biệt nó với tin nhắn người dùng. Cũng vậy, state tin nhắn mới nhất được reset thành chuỗi rỗng để chuẩn bị cho tin nhắn đến tiếp theo.
• Khi dữ liệu được nhận tăng dần (chưa hình thành tin nhắn hoàn chỉnh), nó được nối thêm vào biến incomingMessage.
• State tin nhắn mới nhất được cập nhật với dữ liệu đến được nối thêm, đảm bảo cập nhật thời gian thực khi dữ liệu mới đến.

Tuyệt vời! Với các biến latestMessage và messages phản ứng, bạn giờ có thể tạo giao diện người dùng động đại diện cho tin nhắn được trao đổi trong hội thoại và phản hồi AI mới nhất được tạo.

Đó là rất nhiều kiến thức! Bạn đã xong rồi ✨

Triển Khai Lên Vercel

Kho lưu trữ giờ đã sẵn sàng để triển khai lên Vercel. Sử dụng các bước sau để triển khai:

• Bắt đầu bằng cách tạo kho GitHub chứa mã ứng dụng của bạn.
• Sau đó, điều hướng đến Vercel Dashboard và tạo New Project.
• Liên kết dự án mới với kho GitHub bạn vừa tạo.
• Trong Settings, cập nhật Environment Variables để khớp với những cái trong tệp .env cục bộ của bạn.
• Triển khai! 🚀

Thông Tin Thêm

Để có thêm chi tiết, hãy khám phá các tham khảo được trích dẫn trong bài viết này.

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