add prevResponesId end-to-end flow

Author: sutigit

src/client/components/ChatV2/ChatBox.tsx

@@ -19,9 +19,11 @@ export const ChatBox = ({ disabled, onSubmit }: { disabled: boolean; onSubmit: (
       refetchStatus()
     }
   }
-  useEffect(() => {
-    console.log('userStatus', userStatus)
-  })
+
+  // Work in progress from Matias. Commented by Ben
+  // useEffect(() => {
+  //   console.log('userStatus', userStatus)
+  // })
 
   if (statusLoading) {
     return <p>loading</p>

src/client/components/ChatV2/ChatV2.tsx

@@ -6,6 +6,7 @@ import useLocalStorageState from '../../hooks/useLocalStorageState'
 import { DEFAULT_MODEL } from '../../../config'
 import useInfoTexts from '../../hooks/useInfoTexts'
 import { Message } from '../../types'
+import { ResponseStreamValue } from '../../../shared/types'
 import useRetryTimeout from '../../hooks/useRetryTimeout'
 import { useTranslation } from 'react-i18next'
 import { handleCompletionStreamError } from './error'
@@ -33,6 +34,9 @@ export const ChatV2 = () => {
   const [system, setSystem] = useLocalStorageState<{ content: string }>('general-chat-system', { content: '' })
   const [message, setMessage] = useLocalStorageState<{ content: string }>('general-chat-current', { content: '' })
   const [messages, setMessages] = useLocalStorageState<Message[]>('general-chat-messages', [])
+  const [prevResponse, setPrevResponse] = useLocalStorageState<{
+    id: string
+  }>('general-prev-response', { id: '' })
 
   const appContainerRef = useContext(AppContext)
   const chatContainerRef = useRef<HTMLDivElement>(null)
@@ -55,21 +59,43 @@ export const ChatV2 = () => {
   const disclaimerInfo = infoTexts?.find((infoText) => infoText.name === 'disclaimer')?.text[language] ?? null
   const systemMessageInfo = infoTexts?.find((infoText) => infoText.name === 'systemMessage')?.text[language] ?? null
 
+  const decoder = new TextDecoder()
+
   const processStream = async (stream: ReadableStream) => {
     try {
       const reader = stream.getReader()
 
       let content = ''
-      const decoder = new TextDecoder()
 
       while (true) {
         const { value, done } = await reader.read()
         if (done) break
 
-        const text = decoder.decode(value)
-        setCompletion((prev) => prev + text)
-        content += text
-        console.log(text)
+        const data = decoder.decode(value)
+
+        for (const chunk of data.split('\n')) {
+          if (!chunk || chunk.trim().length === 0) continue
+
+          const parsedChunk: ResponseStreamValue = JSON.parse(chunk)
+
+          switch (parsedChunk.status) {
+            case 'writing':
+              setCompletion((prev) => prev + parsedChunk.text)
+              content += parsedChunk.text
+              break
+
+            case 'complete':
+              setPrevResponse({ id: parsedChunk.prevResponseId })
+              break
+
+            case 'error':
+              console.error('Somehing went wrong when streaming responses')
+              break
+
+            default:
+              break
+          }
+        }
       }
 
       setMessages((prev: Message[]) => prev.concat({ role: 'assistant', content }))
@@ -89,6 +115,7 @@ export const ChatV2 = () => {
     const newMessages = messages.concat({ role: 'user', content: message })
     setMessages(newMessages)
     setMessage({ content: '' })
+    setPrevResponse({ id: '' })
     setCompletion('')
     setStreamController(new AbortController())
     setRetryTimeout(() => {
@@ -108,6 +135,7 @@ export const ChatV2 = () => {
         courseId,
         abortController: streamController,
         saveConsent,
+        prevResponseId: prevResponse.id,
       })
 
       if (tokenUsageAnalysis && tokenUsageAnalysis.message) {

src/client/components/ChatV2/util.ts

@@ -9,6 +9,7 @@ interface GetCompletoinStreamProps {
   userConsent: boolean
   modelTemperature: number
   courseId?: string
+  prevResponseId?: string
   abortController?: AbortController
   saveConsent: boolean
 }
@@ -20,6 +21,7 @@ export const getCompletionStream = async ({
   userConsent,
   modelTemperature,
   courseId,
+  prevResponseId,
   abortController,
   saveConsent,
 }: GetCompletoinStreamProps) => {
@@ -37,6 +39,7 @@ export const getCompletionStream = async ({
       userConsent,
       modelTemperature,
       saveConsent,
+      prevResponseId,
     },
   }
 
@@ -54,8 +57,9 @@ interface GetCourseCompletionStreamProps {
   model: string
   courseId: string
   abortController?: AbortController
+  prevResponseId?: string
 }
-export const getCourseCompletionStream = async ({ id, system, messages, model, courseId, abortController }: GetCourseCompletionStreamProps) => {
+export const getCourseCompletionStream = async ({ id, system, messages, model, courseId, abortController, prevResponseId }: GetCourseCompletionStreamProps) => {
   const data = {
     id,
     courseId,
@@ -68,6 +72,7 @@ export const getCourseCompletionStream = async ({ id, system, messages, model, c
         ...messages,
       ],
       model,
+      prevResponseId,
     },
   }
   const formData = new FormData()

src/server/routes/openai.ts

@@ -106,7 +106,8 @@ openaiRouter.post('/stream/:version?', upload.single('file'), async (r, res) =>
 
   let events
   if (version === 'v2') {
-    events = await responsesClient.createResponse({ input: options.messages })
+    const latestMessage = options.messages[options.messages.length - 1] // Adhoc to input only the latest message
+    events = await responsesClient.createResponse({ input: [latestMessage], prevResponseId: options.prevResponseId })
   } else {
     events = await getCompletionEvents(options)
   }
@@ -224,7 +225,8 @@ openaiRouter.post('/stream/:courseId/:version?', upload.single('file'), async (r
 
   let events
   if (version === 'v2') {
-    events = await responsesClient.createResponse({ input: options.messages })
+    const latestMessage = options.messages[options.messages.length - 1] // Adhoc to input only the latest message
+    events = await responsesClient.createResponse({ input: [latestMessage], prevResponseId: options.prevResponseId })
   } else {
     events = await getCompletionEvents(options)
   }

src/server/util/azure/ResponsesAPI.ts

@@ -16,6 +16,8 @@ import { FileSearchTool, FunctionTool, ResponseInput, ResponseInputItem, Respons
 import { courseAssistants, type CourseAssistant } from './courseAssistants'
 import { createFileSearchTool } from './util'
 
+import type { ResponseStreamValue } from '../../../shared/types'
+
 const endpoint = `https://${AZURE_RESOURCE}.openai.azure.com/`
 
 export const getAzureOpenAIClient = (deployment: string) =>
@@ -48,9 +50,6 @@ export class ResponsesClient {
       courseAssistant = courseAssistants.find((assistant) => assistant.name === 'default')
     }
 
-    this.model = deploymentId
-    this.instructions = courseAssistant.assistant_instruction
-
     const fileSearchTool = courseId
       ? [
           createFileSearchTool({
@@ -59,14 +58,16 @@ export class ResponsesClient {
         ]
       : [] // needs to retrun empty array for null
 
+    this.model = deploymentId
+    this.instructions = courseAssistant.assistant_instruction
     this.tools = fileSearchTool
   }
 
-  async createResponse({ input }: { input: ResponseInput }): Promise<Stream<ResponseStreamEvent> | APIError> {
+  async createResponse({ input, prevResponseId }: { input: ResponseInput; prevResponseId?: string }): Promise<Stream<ResponseStreamEvent> | APIError> {
     try {
       return await client.responses.create({
         model: this.model,
-        // previous_response_id=response.id // THIS MIGHT BE IT!!!!!!1
+        previous_response_id: prevResponseId,
         instructions: this.instructions,
         input,
         stream: true,
@@ -89,7 +90,14 @@ export class ResponsesClient {
 
       switch (event.type) {
         case 'response.output_text.delta':
-          await this.writeDelta(event.delta, res)
+          await this.write(
+            {
+              status: 'writing',
+              text: event.delta,
+              prevResponseId: null,
+            },
+            res,
+          )
 
           contents.push(event.delta)
           tokenCount += encoding.encode(event.delta).length ?? 0
@@ -107,8 +115,15 @@ export class ResponsesClient {
           console.log('ANNOTATIONS ADDED', JSON.stringify(event, null, 2))
           break
 
-        case 'response.function_call_arguments.done':
-          // Listen to file_search instead
+        case 'response.completed':
+          await this.write(
+            {
+              status: 'complete',
+              text: null,
+              prevResponseId: event.response.id,
+            },
+            res,
+          )
           break
       }
     }
@@ -119,16 +134,24 @@ export class ResponsesClient {
     }
   }
 
-  private async writeDelta(text: string, res: Response) {
-    // if (!inProduction) logger.info(text)
+  private async write({ status, text, prevResponseId }: ResponseStreamValue, res: Response) {
+    // if (!inProduction) logger.info(message)
 
     await new Promise((resolve) => {
-      if (
-        !res.write(text, (err) => {
-          if (err) logger.error(`${text} ${err}`)
-        })
-      ) {
-        logger.info(`${text} res.write returned false, waiting for drain`)
+      const data: ResponseStreamValue = {
+        status,
+        text,
+        prevResponseId: prevResponseId,
+      }
+
+      const success = res.write(JSON.stringify(data) + '\n', (err) => {
+        if (err) {
+          logger.error(err)
+        }
+      })
+
+      if (!success) {
+        logger.info('res.write returned false, waiting for drain')
         res.once('drain', resolve)
       } else {
         process.nextTick(resolve)

src/shared/types.ts

@@ -2,3 +2,9 @@ export type RagIndexMetadata = {
   name: string
   dim: number
 }
+
+export type ResponseStreamValue = {
+  status: 'writing' | 'complete' | 'error'
+  text: string | null
+  prevResponseId: string | null
+}