upload UX

Author: Veikkosuhonen

src/client/Router.tsx

@@ -11,7 +11,7 @@ import Discussion from './components/Courses/Course/Discussion'
 import NoAccess from './components/NoAccess'
 import Chats from './components/Chats'
 import Statistics from './components/Statistics'
-import Rag from './components/Rag'
+import Rag from './components/Rag/Rag'
 import { ChatV2 } from './components/ChatV2/ChatV2'
 
 const router = createBrowserRouter(

src/client/components/Rag/ProgressReporter.tsx

@@ -0,0 +1,125 @@
+import { Box, Typography } from '@mui/material'
+import { useEffect, useReducer } from 'react'
+
+type ProgressEvent = {
+  stage: string
+  item?: string
+  done?: boolean
+  error?: string
+}
+
+type ProgressState = {
+  [fileName: string]: {
+    [stage: string]: {
+      count: number
+      done: boolean
+    }
+  }
+}
+
+type Action = { type: 'UPDATE'; payload: ProgressEvent } | { type: 'RESET' }
+
+const progressReducer = (state: ProgressState, action: Action): ProgressState => {
+  switch (action.type) {
+    case 'UPDATE': {
+      const { stage, item, done } = action.payload
+      if (done) {
+        // mark all items at this stage as done
+        const newState = { ...state }
+        for (const file in newState) {
+          if (newState[file][stage]) {
+            newState[file][stage].done = true
+          }
+        }
+        return newState
+      }
+      if (!item) return state
+
+      const fileStages = state[item] || {}
+      const stageData = fileStages[stage] || { count: 0, done: false }
+
+      return {
+        ...state,
+        [item]: {
+          ...fileStages,
+          [stage]: {
+            count: stageData.count + 1,
+            done: stageData.done || false,
+          },
+        },
+      }
+    }
+
+    case 'RESET':
+      return {}
+
+    default:
+      return state
+  }
+}
+
+export const ProgressReporter: React.FC<{ stream: ReadableStream | null }> = ({ stream }) => {
+  const [progress, dispatch] = useReducer(progressReducer, {})
+
+  useEffect(() => {
+    let reader: ReadableStreamDefaultReader<Uint8Array> | null = null
+    if (stream) {
+      // Check if the stream is readable
+      try {
+        reader = stream.getReader()
+      } catch (error) {
+        console.error('Error getting reader from stream:', error)
+        return
+      }
+
+      const decoder = new TextDecoder('utf-8')
+
+      const readStream = async () => {
+        while (true) {
+          const { value, done } = await reader.read()
+          if (done) break
+
+          const chunkText = decoder.decode(value, { stream: true })
+          const lines = chunkText.split('\n')
+
+          for (const line of lines) {
+            if (line.trim()) {
+              console.log('Received chunk:', line)
+              try {
+                const jsonChunk = JSON.parse(line)
+                dispatch({ type: 'UPDATE', payload: jsonChunk })
+              } catch (err) {
+                console.error('Invalid chunk:', line, err)
+              }
+            }
+          }
+        }
+      }
+
+      readStream()
+
+      return () => {
+        if (reader) {
+          reader.releaseLock()
+        }
+      }
+    }
+  }, [stream])
+
+  return (
+    <Box>
+      {Object.entries(progress).map(([file, stages]) => (
+        <div key={file} className="border p-2 rounded shadow">
+          <h3 className="font-bold text-lg">{file}</h3>
+          <ul className="ml-4 list-disc">
+            {Object.entries(stages).map(([stage, { count, done }]) => (
+              <li key={stage}>
+                <span className="font-medium">{stage}:</span> {done ? "✅ Done" : `🔄 Processing (${count}x)`}
+              </li>
+            ))}
+          </ul>
+        </div>
+      ))}
+    </Box>
+  )
+}

src/client/components/Rag.tsx renamed to src/client/components/Rag/Rag.tsx

@@ -1,10 +1,11 @@
 import React, { useState } from 'react'
 import { TextField, Button, Box, Typography, Table, TableHead, TableBody, TableRow, TableCell, Paper, IconButton, Dialog, DialogTitle, styled } from '@mui/material'
-import apiClient from '../util/apiClient'
+import apiClient, { postAbortableStream } from '../../util/apiClient'
 import { useMutation, useQuery } from '@tanstack/react-query'
 import { CloudUpload, Settings } from '@mui/icons-material'
-import Markdown from './Banner/Markdown'
+import Markdown from '../Banner/Markdown'
 import { useSnackbar } from 'notistack'
+import { ProgressReporter } from './ProgressReporter'
 
 type RagResponse = {
   id: string
@@ -69,13 +70,12 @@ const useUploadMutation = (index: RagIndexAttributes | null) => {
         formData.append('files', file)
       })
 
-      const response = await apiClient.put(`/rag/indices/${index.id}/upload`, formData, {
-        headers: {
-          'Content-Type': 'multipart/form-data',
-        },
-      })
+      const { stream } = await postAbortableStream(`/rag/indices/${index.id}/upload`, formData)
+      if (!stream) {
+        throw new Error('No stream returned from server')
+      }
 
-      return response.data
+      return stream
     },
   })
   return mutation
@@ -105,6 +105,7 @@ const Rag: React.FC = () => {
   const [response, setResponse] = useState<RagResponse[] | null>(null)
   const uploadMutation = useUploadMutation(selectedIndex)
   const [modalOpen, setModalOpen] = useState(false)
+  const [stream, setStream] = useState<ReadableStream | null>(null)
 
   const handleSubmit = async (event: React.FormEvent) => {
     event.preventDefault()
@@ -131,39 +132,40 @@ const Rag: React.FC = () => {
     <Box sx={{ display: 'flex', gap: 2 }}>
       <Dialog open={!!selectedIndex && modalOpen} onClose={() => setModalOpen(false)}>
         <DialogTitle>Edit {selectedIndex?.metadata?.name}</DialogTitle>
-        <Box sx={{ padding: 2, display: 'flex', gap: 2 }}>
-          <Button component="label" role={undefined} variant="contained" tabIndex={-1} startIcon={<CloudUpload />} disabled={uploadMutation.isPending}>
-            {uploadMutation.isPending ? 'Uploading...' : 'Upload Files'}
-            <VisuallyHiddenInput
-              type="file"
-              onChange={async (event) => {
-                const files = event.target.files
-                console.log('Files selected:', files)
-                if (files && files.length > 0) {
-                  await uploadMutation.mutateAsync(files)
+        <Box sx={{ padding: 2 }}>
+          <Box sx={{ display: 'flex', gap: 2 }}>
+            <Button component="label" role={undefined} variant="contained" tabIndex={-1} startIcon={<CloudUpload />} disabled={uploadMutation.isPending}>
+              {uploadMutation.isPending ? 'Uploading...' : 'Upload Files'}
+              <VisuallyHiddenInput
+                type="file"
+                onChange={async (event) => {
+                  const files = event.target.files
+                  console.log('Files selected:', files)
+                  if (files && files.length > 0) {
+                    const stream = await uploadMutation.mutateAsync(files)
+                    setStream(stream)
+                  }
+                }}
+                multiple
+              />
+            </Button>
+            <Button
+              variant="text"
+              color="error"
+              onClick={async () => {
+                if (selectedIndex && window.confirm(`Are you sure you want to delete index ${selectedIndex.metadata.name}?`)) {
+                  await deleteIndexMutation.mutateAsync(selectedIndex.id)
+                  setSelectedIndex(null)
                   refetch()
-                  setModalOpen(false)
-                  enqueueSnackbar('Files uploaded successfully', {
-                    variant: 'success',
-                  })
                 }
               }}
-              multiple
-            />
-          </Button>
-          <Button
-            variant="text"
-            color="error"
-            onClick={async () => {
-              if (selectedIndex && window.confirm(`Are you sure you want to delete index ${selectedIndex.metadata.name}?`)) {
-                await deleteIndexMutation.mutateAsync(selectedIndex.id)
-                setSelectedIndex(null)
-                refetch()
-              }
-            }}
-          >
-            Delete Index
-          </Button>
+            >
+              Delete Index
+            </Button>
+          </Box>
+          <Box mt={2}>
+            <ProgressReporter stream={stream} />
+          </Box>
         </Box>
       </Dialog>
       <Box>
@@ -247,7 +249,9 @@ const Rag: React.FC = () => {
             {response.map((doc) => (
               <Paper key={doc.id} sx={{ marginBottom: 2, p: 1 }} elevation={2}>
                 <Typography variant="caption">Score: {doc.value.score}</Typography>
-                <Typography variant="subtitle1" fontFamily="monospace" mb={2}>{JSON.stringify(doc.value.metadata, null, 2)}</Typography>
+                <Typography variant="subtitle1" fontFamily="monospace" mb={2}>
+                  {JSON.stringify(doc.value.metadata, null, 2)}
+                </Typography>
                 {doc.value.metadata.type === 'md' ? (
                   <Markdown>{doc.value.content}</Markdown>
                 ) : (

src/server/routes/rag.ts

@@ -9,6 +9,7 @@ import { ingestionPipeline } from '../services/rag/ingestion/pipeline'
 import { getAzureOpenAIClient } from '../util/azure'
 import multer from 'multer'
 import { mkdir, rm, stat } from 'fs/promises'
+import { Readable } from 'stream'
 
 const router = Router()
 
@@ -115,7 +116,7 @@ const indexUploadDirMiddleware = async (req: Request, _res: Response, next: Next
   next()
 }
 
-router.put('/indices/:id/upload', [indexUploadDirMiddleware, uploadMiddleware], async (req, res) => {
+router.post('/indices/:id/upload', [indexUploadDirMiddleware, uploadMiddleware], async (req, res) => {
   const { user } = req as unknown as RequestWithUser
   const id = IndexIdSchema.parse(req.params.id)
 
@@ -132,11 +133,14 @@ router.put('/indices/:id/upload', [indexUploadDirMiddleware, uploadMiddleware],
     return
   }
 
+  res.setHeader('Content-Type', 'application/x-ndjson')
+  res.setHeader('Transfer-Encoding', 'chunked')
+
   const openAiClient = getAzureOpenAIClient(EMBED_MODEL)
 
-  await ingestionPipeline(openAiClient, `uploads/rag/${id}`, ragIndex)
+  const progressReporter = await ingestionPipeline(openAiClient, `uploads/rag/${id}`, ragIndex)
 
-  res.json({ message: 'Files uploaded and processed' })
+  progressReporter.pipe(res)
 })
 
 const RagIndexQuerySchema = z.object({

src/server/services/rag/ingestion/chunker.ts

@@ -3,9 +3,11 @@ import { chunkingAlgorithms } from './chunkingAlgorithms.ts'
 import { mkdirSync } from 'node:fs'
 import { writeFile } from 'node:fs/promises'
 import { TextData } from './textExtractor.ts'
+import { StageReporter } from './progressReporter.ts'
 
 export class Chunker extends Transform {
   private cachePath: string
+  public progressReporter: StageReporter
 
   constructor(cachePath: string) {
     super({ objectMode: true })
@@ -16,7 +18,7 @@ export class Chunker extends Transform {
     mkdirSync(this.cachePath, { recursive: true })
   }
 
-  _transform(data: TextData, _encoding: BufferEncoding, callback: (error?: Error | null) => void) {
+  async _transform(data: TextData, _encoding: BufferEncoding, callback: (error?: Error | null) => void) {
     const chunkingAlgorithm = chunkingAlgorithms[data.chunkingStrategy]
 
     const chunks = chunkingAlgorithm(data)
@@ -26,13 +28,20 @@ export class Chunker extends Transform {
 
     // Save chunks to cache
 
-    Promise.all(
+    await Promise.all(
       chunks.map((chunk) => {
         const chunkPath = `${this.cachePath}/${chunk.id}.json`
         return writeFile(chunkPath, JSON.stringify(chunk, null, 2), 'utf-8')
       }),
-    ).then(() => {
-      callback()
-    })
+    )
+
+    this.progressReporter.reportProgress(data.fileName)
+
+    callback()
+  }
+
+  _flush(callback: (error?: Error | null) => void) {
+    this.progressReporter.reportDone()
+    callback()
   }
 }

src/server/services/rag/ingestion/chunkingAlgorithms.ts

@@ -3,8 +3,11 @@ import { TextData } from './textExtractor.ts'
 export type Chunk = {
   id: string
   content: string[]
-  metadata?: {
-    [key: string]: any
+  metadata: {
+    title?: string
+    titleHierarchy?: string[]
+    type: string
+    filename: string
   }
 }
 
@@ -28,6 +31,7 @@ export const createTitleChunks = (file: TextData): Chunk[] => {
           title,
           titleHierarchy: [...titleHierarchy],
           type: file.type,
+          filename: file.fileName,
         },
       })
 
@@ -62,6 +66,7 @@ export const createTitleChunks = (file: TextData): Chunk[] => {
         title,
         titleHierarchy: [...titleHierarchy],
         type: file.type,
+        filename: file.fileName,
       },
     })
   }
@@ -82,8 +87,9 @@ export const createSplittedTitleChunks = (file: TextData): Chunk[] => {
         content: section.split('\n'),
         metadata: {
           title: `${title} - ${index + 1}`,
-          titleHierarchy: [...titleHierarchy, index + 1],
+          titleHierarchy: [...titleHierarchy, index + 1 + ''],
           type: file.type,
+          filename: file.fileName,
         },
       }))
   })
@@ -103,6 +109,7 @@ export const createStaticChunks = (file: TextData, length: number = 800, overlap
         metadata: {
           title: file.fileName,
           type: file.type,
+          filename: file.fileName,
         },
       })
     }