chore: improve ux and add progress statuses

Author: manekinekko

api/src/functions/assistant.js

@@ -2,7 +2,6 @@ require("dotenv/config");
 
 const { Readable } = require("node:stream");
 const { app } = require("@azure/functions");
-app.setup({ enableHttpStream: true });
 
 const { AzureOpenAI } = require("openai");
 const { DefaultAzureCredential, getBearerTokenProvider } = require("@azure/identity");
@@ -19,7 +18,7 @@ const {
 // please add proper error handling.
 
 async function initAzureOpenAI(context) {
-  context.log("Using Azure OpenAI (w/ Microsoft Entra ID) ...");
+  console.log("Using Azure OpenAI (w/ Microsoft Entra ID) ...");
   const credential = new DefaultAzureCredential();
   const azureADTokenProvider = getBearerTokenProvider(credential, "https://cognitiveservices.azure.com/.default");
   return new AzureOpenAI({
@@ -30,7 +29,8 @@ async function initAzureOpenAI(context) {
 const assistantDefinition = {
   name: "Finance Assistant",
   instructions:
-    "You are a personal finance assistant. Retrieve the latest closing price of a stock using its ticker symbol. You also know how to generate a full body email in both plain text and html.",
+    "You are a personal finance assistant. Retrieve the latest closing price of a stock using its ticker symbol. "
+    + "You also know how to generate a full body email in both plain text and html. Only use the functions you have been provideded with",
   tools: [
     {
       type: "function",
@@ -80,54 +80,73 @@ const assistantDefinition = {
   model: AZURE_DEPLOYMENT_NAME,
 };
 
-async function* processQuery(userQuery, context) {
-  context.log('Step 0: Connect and acquire an OpenAI instance');
-  const openai = await initAzureOpenAI(context);
+async function* processQuery(userQuery) {
+  console.log('Step 0: Connect and acquire an OpenAI instance');
+  const openai = await initAzureOpenAI();
 
-  context.log('Step 1: Retrieve or Create an Assistant');
+  console.log('Step 1: Retrieve or Create an Assistant');
   const assistant = ASSISTANT_ID
     ? await openai.beta.assistants.retrieve(ASSISTANT_ID)
     : await openai.beta.assistants.create(assistantDefinition);
 
-  context.log('Step 2: Create a Thread');
+  console.log('Step 2: Create a Thread');
   const thread = await openai.beta.threads.create();
 
-  context.log('Step 3: Add a Message to the Thread');
+  console.log('Step 3: Add a Message to the Thread');
   const message = await openai.beta.threads.messages.create(thread.id, {
     role: "user",
     content: userQuery,
   });
 
-  context.log('Step 4: Create a Run (and stream the response)');
+  console.log('Step 4: Create a Run (and stream the response)');
   const run = openai.beta.threads.runs.stream(thread.id, {
     assistant_id: assistant.id,
     stream: true,
-    tool_choice: { "type": "function", "function": { "name": "getStockPrice" } }
   });
 
-  context.log('Step 5: Read streamed response', { run });
+  console.log('Step 5: Read streamed response', { run });
   for await (const chunk of run) {
     const { event, data } = chunk;
 
-    if (event === "thread.message.delta") {
+    console.log('processing event', { event, data });
+
+    if (event === "thread.run.created") {
+      yield "@created";
+      console.log('Processed thread.run.created');
+    }
+    else if (event === "thread.run.queued") {
+      yield "@queued";
+      console.log('Processed thread.run.queued');
+    }
+    else if (event === "thread.run.in_progress") {
+      yield "@in_progress";
+      console.log('Processed thread.run.in_progress');
+    }
+    else if (event === "thread.message.delta") {
       const delta = data.delta;
 
       if (delta) {
         const value = delta.content[0]?.text?.value || "";
         yield value;
-        context.log('Processed thread.message.delta', { value });
+        console.log('Processed thread.message.delta', { value });
       }
-    } else if (event === "thread.run.requires_action") {
-      yield* handleRequiresAction(openai, data, data.id, data.thread_id, context);
+    }
+    else if (event === "thread.run.failed") {
+      const value = data.last_error.message;
+      yield value;
+      console.log('Processed thread.run.failed', { value });
+    }
+    else if (event === "thread.run.requires_action") {
+      yield* handleRequiresAction(openai, data, data.id, data.thread_id);
     }
     // else if ... handle the other events as needed
   }
 
-  context.log('Done!');
+  console.log('Done!');
 }
 
-async function* handleRequiresAction(openai, run, runId, threadId, context) {
-  context.log('Handle Function Calling', { required_action: run.required_action.submit_tool_outputs.tool_calls });
+async function* handleRequiresAction(openai, run, runId, threadId) {
+  console.log('Handle Function Calling', { required_action: run.required_action.submit_tool_outputs.tool_calls });
   try {
     const toolOutputs = await Promise.all(
       run.required_action.submit_tool_outputs.tool_calls.map(
@@ -156,36 +175,38 @@ async function* handleRequiresAction(openai, run, runId, threadId, context) {
     );
 
     // Submit all the tool outputs at the same time
-    yield* submitToolOutputs(openai, toolOutputs, runId, threadId, context);
+    yield* submitToolOutputs(openai, toolOutputs, runId, threadId);
   } catch (error) {
-    context.error("Error processing required action:", error);
+    console.error("Error processing required action:", error);
   }
 }
 
-async function* submitToolOutputs(openai, toolOutputs, runId, threadId, context) {
+async function* submitToolOutputs(openai, toolOutputs, runId, threadId) {
   try {
     // Use the submitToolOutputsStream helper
-    context.log('Call Tool output and stream the response');
+    console.log('Call Tool output and stream the response');
     const asyncStream = openai.beta.threads.runs.submitToolOutputsStream(
       threadId,
       runId,
       { tool_outputs: toolOutputs }
     );
     for await (const chunk of asyncStream) {
-      if (chunk.event === "thread.message.delta") {
+      const { event, data } = chunk;
+      // console.log({ event, data });
+      if (event === "thread.message.delta") {
         // stream message back to UI
-        const { delta } = chunk.data;
+        const { delta } = data;
 
         if (delta) {
           const value = delta.content[0]?.text?.value || "";
           yield value;
-          context.log('Processed thread.message.delta (tool output)', { value });
+          console.log('Processed thread.message.delta (tool output)', { value });
         }
       }
       // else if ... handle the other events as needed
     }
   } catch (error) {
-    context.error("Error submitting tool outputs:", error);
+    console.error("Error submitting tool outputs:", error);
   }
 }
 
@@ -204,19 +225,19 @@ async function writeAndSendEmail(subject, text, html) {
 }
 
 // API definition
-
+app.setup({ enableHttpStream: true });
 app.http("assistant", {
   methods: ["POST"],
   authLevel: "anonymous",
-  handler: async (request, context) => {
-    context.log(`Http function processed request for url "${request.url}"`);
+  handler: async (request) => {
+    console.log(`Http function processed request for url "${request.url}"`);
     const query = await request.text();
 
     return {
       headers: {
         'Content-Type': 'text/plain',
         "Transfer-Encoding": "chunked"
-      }, body: Readable.from(processQuery(query, context))
+      }, body: Readable.from(processQuery(query))
     };
   },
 });

src/index.html

@@ -29,7 +29,7 @@ <h1>Azure OpenAI / Finance Assistant Demo</h1>
     <section class="output__container">
       <div id="loadingRef" class="loader__container hidden">
         <span class="spinner"><span class="spinner__tail"></span></span>
-        Working on it ...
+        <p>Working on it (<span id="statusLabelRef">waiting</span>)</p>
       </div>
       <div id="outputRef" class="hidden"></div>
     </section>

src/script.js

@@ -6,6 +6,7 @@ const loadingRef = document.querySelector("#loadingRef");
 const userQueryRef = document.querySelector("#userQueryRef");
 const cancelQueryRef = document.querySelector("#cancelQueryRef");
 const submitQueryRef = document.querySelector("#submitQueryRef");
+const statusLabelRef = document.querySelector("#statusLabelRef");
 
 userQueryRef.value =
   "Based on the latest financial data and current stock market trends, can you provide a detailed analysis of Microsoft's current state? Please include insights into their recent performance, market position, and future outlook. Additionally, retrieve and include the latest closing price of Microsoft's stock using its ticker symbol (MSFT). Send me the full analysis by email.";
@@ -22,69 +23,68 @@ submitQueryRef
   .addEventListener("click", async (event) => {
     const { value } = userQueryRef;
     if (value) {
+      statusLabelRef.innerHTML = "waiting";
+      outputRef.innerHTML = "";
 
       loadingRef.classList.remove("hidden");
       outputRef.classList.add("hidden");
       cancelQueryRef.classList.remove("hidden");
       submitQueryRef.classList.add("hidden");
 
       autoTimeout();
-      submitQuery(value, insertText);
+      submitQuery(value);
     }
   });
 
 function autoTimeout() {
   autoAbortTimeout = setTimeout(() => {
     cancelQueryRef.click();
-
+    outputRef.classList.remove("hidden");
     if (outputRef.innerHTML === "") {
       outputRef.innerHTML = "Your Assistant could not fetch data. Please try again!"
     }
-
-  }, 30_000); // in case, cancel request after 30 of timeout
-
+  }, 60_000); // cancel request if it times out
 }
 
-function insertText(chunk) {
-  const delta = new TextDecoder().decode(chunk);
-  outputRef.innerHTML += delta;
-  outputRef.scrollTop = outputRef.scrollHeight; // scroll to bottom
-};
-
-
-function submitQuery(userQuery, cb) {
-
+function submitQuery(body) {
   const { API_URL = 'http://localhost:7071' } = import.meta.env;
-
   fetch(`${API_URL}/api/assistant`, {
+    body,
     method: "POST",
-    body: userQuery,
     signal: aborter.signal
   }).then(response => response.body)
-    .then(rs => processReadableStream(rs, cb));
+    .then(processReadableStream);
 }
 
-function processReadableStream(rs, cb) {
-
-  rs.pipeTo(new WritableStream({
+function processReadableStream(stream) {
+  stream.pipeTo(new WritableStream({
     write(chunk, controller) {
-      cb(chunk);
-    },
-    start(controller) {
-      outputRef.innerHTML = "";
+      const value = new TextDecoder().decode(chunk);
+      if (value.startsWith('@')) {
+        statusLabelRef.innerHTML = value.replace('@', '');
+        return;
+      }
+
       loadingRef.classList.add("hidden");
       outputRef.classList.remove("hidden");
+
+      outputRef.innerHTML += value;
+      outputRef.scrollTop = outputRef.scrollHeight; // scroll to bottom
+    },
+    start(controller) {
       clearTimeout(autoAbortTimeout); // cancel 
     },
     close(controller) {
       cancelQueryRef.classList.add("hidden");
       submitQueryRef.classList.remove("hidden");
       loadingRef.classList.add("hidden");
+      outputRef.classList.remove("hidden");
+      if (outputRef.innerHTML === "") {
+        outputRef.innerHTML = "Whoops, something went wrong. Please try again!"
+      }
     },
     abort(reason) {
       console.log(reason);
     },
   })).catch(console.error);
-
-  return rs;
 }