+better stream decoder

.gitignore

@@ -1 +1,2 @@
-.vscode
+.vscode
+.env

deno.json

@@ -1,4 +1,6 @@
 {
-  "tasks": {},
+  "tasks": {
+    "test": "YOUR_API_KEY=YOUR_API_KEY_HERE deno test --allow-env --allow-net"
+  },
   "lock": false
-}
+}

src/openai.ts

@@ -1,5 +1,5 @@
 import { basename } from "https://deno.land/std@0.189.0/path/mod.ts";
-import { decodeStream, throwError } from "./util.ts";
+import { decodeStream, throwErrorIfNeeded } from "./util.ts";
 import type {
   ChatCompletion,
   ChatCompletionOptions,
@@ -70,7 +70,7 @@ export class OpenAI {
     );
     const data = await response.json();
 
-    throwError(data);
+    throwErrorIfNeeded(data);
 
     return data;
   }

src/types.ts

@@ -773,3 +773,22 @@ export interface Moderation {
     flagged: boolean;
   }[];
 }
+
+export interface ErrorResponse {
+  error: Error;
+}
+
+export interface ModelError {
+  type: string;
+  message: string;
+  param: string | null;
+  code: string | null;
+}
+
+export class RequiredError extends Error {
+  public name = "RequiredError";
+
+  constructor(public readonly field: string, message?: string) {
+    super(message);
+  }
+}

src/util.ts

@@ -1,40 +1,46 @@
-import { TextDelimiterStream } from "https://deno.land/std@0.189.0/streams/mod.ts";
+import { TextDelimiterStream } from "https://deno.land/std@0.189.0/streams/text_delimiter_stream.ts";
+import { ModelError, RequiredError } from "./types.ts";
 
-export function throwError(
-  data: { error?: { type: string; message: string; code: string } },
-) {
-  if (data.error) {
-    let errorMessage = `${data.error.type}`;
-    if (data.error.message) {
-      errorMessage += ": " + data.error.message;
-    }
-    if (data.error.code) {
-      errorMessage += ` (${data.error.code})`;
-    }
-    // console.log(data.error);
-    throw new Error(errorMessage);
+export function throwErrorIfNeeded(response: unknown) {
+  // deno-lint-ignore ban-types
+  if ("error" in (response as object)) {
+    const {
+      error: { type, message },
+    } = response as { error: ModelError };
+
+    throw new RequiredError(type, message);
   }
 }
 
-// deno-lint-ignore no-explicit-any
-export async function decodeStream(
-  res: Response,
-  callback: (data: any) => void,
+export async function decodeStream<T>(
+  { body: stream }: Response,
+  callback: (data: T) => void
 ) {
-  const chunks = res.body!
+  if (stream === null || stream.locked) {
+    throw new Error(`The stream is ${stream === null ? "null" : "locked"}.`);
+  }
+
+  const chunks = stream
     .pipeThrough(new TextDecoderStream())
-    .pipeThrough(new TextDelimiterStream("\n\n"));
-
-  for await (const chunk of chunks) {
-    let data;
-    try {
-      data = JSON.parse(chunk);
-    } catch {
-      // no-op (just checking if error message)
-    }
-    if (data) throwError(data);
+    .pipeThrough(new TextDelimiterStream("\n\n"))
+    .getReader();
+
+  try {
+    for (;;) {
+      const { done, value: chunk } = await chunks.read();
+
+      if (done || chunk === "data: [DONE]") {
+        break;
+      }
 
-    if (chunk === "data: [DONE]") break;
-    callback(JSON.parse(chunk.slice(6)));
+      const argument = JSON.parse(
+        chunk.startsWith("data: ") ? chunk.slice(6) : chunk
+      );
+
+      throwErrorIfNeeded(argument);
+      callback(argument);
+    }
+  } finally {
+    chunks.releaseLock();
   }
 }

tests/createChatCompletionStream.test.ts

@@ -0,0 +1,32 @@
+import { assertEquals } from "https://deno.land/std@0.192.0/testing/asserts.ts";
+import { OpenAI } from "../mod.ts";
+
+const permissions = {
+  net: true,
+  env: true,
+} satisfies Deno.PermissionOptions;
+
+const openai = new OpenAI(Deno.env.get("YOUR_API_KEY")!);
+
+Deno.test(
+  "createChatCompletionStream",
+  { permissions },
+  async () =>
+    await openai.createChatCompletionStream(
+      {
+        model: "gpt-3.5-turbo",
+        messages: [
+          { role: "system", content: "You are a helpful assistant." },
+          { role: "user", content: "Who won the world series in 2020?" },
+          {
+            role: "assistant",
+            content: "The Los Angeles Dodgers won the World Series in 2020.",
+          },
+          { role: "user", content: "Where was it played?" },
+        ],
+      },
+      (chunk) => {
+        assertEquals(chunk.object, "chat.completion.chunk");
+      }
+    )
+);