Merge pull request Nutlope#10 from smaeda-ks/smaeda-ks

Move parsing logic to server-side

package.json

@@ -11,6 +11,7 @@
     "@heroicons/react": "^2.0.13",
     "@tailwindcss/forms": "^0.5.3",
     "@vercel/analytics": "^0.1.8",
+    "eventsource-parser": "^0.0.5",
     "framer-motion": "^8.4.3",
     "next": "latest",
     "react": "18.2.0",

pages/api/generate.ts

@@ -1,4 +1,9 @@
 import type { NextRequest } from "next/server";
+import {
+  createParser,
+  ParsedEvent,
+  ReconnectInterval,
+} from "eventsource-parser";
 
 if (!process.env.OPENAI_API_KEY) {
   throw new Error("Missing env var from OpenAI");
@@ -17,6 +22,9 @@ const handler = async (req: NextRequest): Promise<Response> => {
     return new Response("No prompt in the request", { status: 400 });
   }
 
+  const encoder = new TextEncoder();
+  const decoder = new TextDecoder();
+
   const payload = {
     model: "text-davinci-003",
     prompt,
@@ -38,11 +46,46 @@ const handler = async (req: NextRequest): Promise<Response> => {
     body: JSON.stringify(payload),
   });
 
-  const data = res.body;
+  let counter = 0;
+  const stream = new ReadableStream({
+    async start(controller) {
+      // callback
+      function onParse(event: ParsedEvent | ReconnectInterval) {
+        if (event.type === "event") {
+          const data = event.data;
+          // https://beta.openai.com/docs/api-reference/completions/create#completions/create-stream
+          if (data === "[DONE]") {
+            controller.close();
+            return;
+          }
+          try {
+            const json = JSON.parse(data);
+            const text = json.choices[0].text;
+            if (counter < 2 && (text.match(/\n/) || []).length) {
+              // this is a prefix character (i.e., "\n\n"), do nothing
+              return;
+            }
+            const queue = encoder.encode(text);
+            controller.enqueue(queue);
+            counter++;
+          } catch (e) {
+            // maybe parse error
+            controller.error(e);
+          }
+        }
+      }
 
-  return new Response(data, {
-    headers: { "Content-Type": "application/json; charset=utf-8" },
+      // stream response (SSE) from OpenAI may be fragmented into multiple chunks
+      // this ensures we properly read chunks and invoke an event for each SSE event stream
+      const parser = createParser(onParse);
+      // https://web.dev/streams/#asynchronous-iteration
+      for await (const chunk of res.body as any) {
+        parser.feed(decoder.decode(chunk));
+      }
+    },
   });
+
+  return new Response(stream);
 };
 
 export default handler;

pages/index.tsx

@@ -59,51 +59,11 @@ const Home: NextPage = () => {
     const decoder = new TextDecoder();
 
     let done = false;
-    let tempState = "";
-
     while (!done) {
       const { value, done: doneReading } = await reader.read();
       done = doneReading;
-      const newValue = decoder
-        .decode(value)
-        .replaceAll("data: ", "")
-        .split("\n\n")
-        .filter(Boolean);
-
-      if (tempState) {
-        newValue[0] = tempState + newValue[0];
-        tempState = "";
-      }
-
-      newValue.forEach((newVal) => {
-        if (newVal === "[DONE]") {
-          return;
-        }
-
-        try {
-          const json = JSON.parse(newVal) as {
-            id: string;
-            object: string;
-            created: number;
-            choices?: {
-              text: string;
-              index: number;
-              logprobs: null;
-              finish_reason: null | string;
-            }[];
-            model: string;
-          };
-
-          if (!json.choices?.length) {
-            throw new Error("Something went wrong.");
-          }
-
-          const choice = json.choices[0];
-          setGeneratedBios((prev) => prev + choice.text);
-        } catch (error) {
-          tempState = newVal;
-        }
-      });
+      const chunkValue = decoder.decode(value);
+      setGeneratedBios((prev) => prev + chunkValue);
     }
 
     setLoading(false);