pdf parsing with job

Author: tfhuhtal

compose.yaml

@@ -42,21 +42,26 @@ services:
       - redis_data:/data
 
   # Comment out if you use local ollama. Set OLLAMA_URL=host.docker.internal:11434 if using recent Docker Desktop.
-  # ollama:
-  #   image: ollama/ollama
-  #   ports:
-  #     - 11434:11434 # expose the Ollama API to the host
-  #   volumes:
-  #     - ollama_data:/root/.ollama # persistent model storage
+  ollama:
+    image: ollama/ollama
+    ports:
+      - 11434:11434 # expose the Ollama API to the host
+    volumes:
+      - ollama_data:/root/.ollama # persistent model storage
+    entrypoint: ["/bin/bash", "-c", "\
+      ollama serve & \
+      sleep 5 && \
+      ollama pull qwen2.5vl:7b && \
+      wait"]
 
   dalai:
-    # image: toska/dalai:latest
+  #   # image: toska/dalai:latest
     build:
       context: ../dalai
       dockerfile: ../dalai/dev.Dockerfile
     environment:
-      - OLLAMA_URL=http://host.docker.internal:11434
-      #- OLLAMA_URL=http://ollama:11434
+    # - OLLAMA_URL=http://host.docker.internal:11434
+      - OLLAMA_URL=http://ollama:11434
       - REDIS_HOST=redis
       - REDIS_PORT=6379
       - S3_HOST=http://minio:9000
@@ -67,7 +72,7 @@ services:
       - dalai_data:/app
     container_name: gptwrapper_dalai
     depends_on:
-      - redis
+    - redis
 
   minio:
     image: minio/minio:latest
@@ -95,7 +100,7 @@ services:
       /bin/sh -c "
       set -e;
       until (mc alias set local http://minio:9000 minioadmin minioadmin); do
-        echo 'Waiting for MinIO...'; sleep 10;
+        echo 'Waiting for MinIO...'; sleep 2;
       done;
       mc mb --ignore-existing local/mybucket;
       tail -f /dev/null

src/server/services/jobs/pdfParsing.job.ts

@@ -31,9 +31,8 @@ const queue = new Queue('llama-scan-queue', {
 export const pdfQueueEvents = new QueueEvents('llama-scan-queue', { connection })
 
 /**
- * Adds a pdf parsing job to the queue. The file must be uploaded to S3 beforehand. The jobId is based on the filename and prefix - resubmitting with the same jobId while the previous job is running has no effect.
- * @param filename
- * @param prefix
+ * Adds a pdf parsing job to the queue. The file must be uploaded to S3 beforehand. The jobId is based on the ragFile - resubmitting with the same jobId while the previous job is running has no effect.
+ * @param ragFile
  * @returns the job
  */
 export const submitPdfParsingJob = async (ragFile: RagFile) => {

src/server/services/rag/fileStore.ts

@@ -8,7 +8,6 @@ import {
 } from '@aws-sdk/client-s3'
 import type { RagFile, RagIndex } from '../../db/models'
 import { ApplicationError } from '../../util/ApplicationError'
-import { pdfToText } from '../../util/pdfToText'
 import { S3_BUCKET } from '../../util/config'
 import { s3Client } from '../../util/s3client'
 
@@ -129,12 +128,3 @@ const streamToString = (stream: any): Promise<string> => {
     stream.on('end', () => resolve(Buffer.concat(chunks).toString('utf-8')))
   })
 }
-
-const streamToBuffer = (stream: any): Promise<Buffer> => {
-  return new Promise((resolve, reject) => {
-    const chunks: any[] = []
-    stream.on('data', (chunk: any) => chunks.push(chunk))
-    stream.on('error', reject)
-    stream.on('end', () => resolve(Buffer.concat(chunks)))
-  })
-}

src/server/services/rag/ingestion.ts

@@ -34,17 +34,26 @@ export const ingestRagFiles = async (ragIndex: RagIndex) => {
       console.time(`Ingestion ${ragFile.filename}`)
 
       await ragFile.save()
-
-      const job = await submitPdfParsingJob(ragFile)
+      let needToParse = false
 
       try {
-        await job.waitUntilFinished(pdfQueueEvents)
-      } catch (error: any) {
-        console.error('Error waiting for PDF parsing job to finish:', error)
-        ragFile.pipelineStage = 'error'
-        ragFile.error = 'PDF parsing failed'
-        await ragFile.save()
-        return
+        await FileStore.readRagFileTextContent(ragFile)
+      } catch (error) {
+        needToParse = true
+      }
+
+      if (needToParse) {
+        const job = await submitPdfParsingJob(ragFile)
+
+        try {
+          await job.waitUntilFinished(pdfQueueEvents)
+        } catch (error: any) {
+          console.error('Error waiting for PDF parsing job to finish:', error)
+          ragFile.pipelineStage = 'error'
+          ragFile.error = 'PDF parsing failed'
+          await ragFile.save()
+          return
+        }
       }
 
       const text = await FileStore.readRagFileTextContent(ragFile)