feat(RAG): initial beta with preprocessing, embedding, semantic retrieval, and ctx passage

Author: jakeaturner

admin/app/controllers/ollama_controller.ts

@@ -41,16 +41,17 @@ export default class OllamaController {
 
     if (lastUserMessage) {
       // Search for relevant context in the knowledge base
+      // Using lower threshold (0.3) with improved hybrid search
       const relevantDocs = await this.ragService.searchSimilarDocuments(
         lastUserMessage.content,
         5, // Retrieve top 5 most relevant chunks
-        0.7 // Minimum similarity score of 0.7
+        0.3 // Minimum similarity score of 0.3 (lowered from 0.7 for better recall)
       )
 
       // If relevant context is found, inject as a system message
       if (relevantDocs.length > 0) {
         const contextText = relevantDocs
-          .map((doc, idx) => `[Context ${idx + 1}]\n${doc.text}`)
+          .map((doc, idx) => `[Context ${idx + 1}] (Relevance: ${(doc.score * 100).toFixed(1)}%)\n${doc.text}`)
           .join('\n\n')
 
         const systemMessage = {

admin/app/controllers/rag_controller.ts

@@ -1,9 +1,12 @@
 import { RagService } from '#services/rag_service'
+import { EmbedFileJob } from '#jobs/embed_file_job'
 import { inject } from '@adonisjs/core'
 import type { HttpContext } from '@adonisjs/core/http'
 import app from '@adonisjs/core/services/app'
 import { randomBytes } from 'node:crypto'
 import { sanitizeFilename } from '../utils/fs.js'
+import { stat } from 'node:fs/promises'
+import { getJobStatusSchema } from '#validators/rag'
 
 @inject()
 export default class RagController {
@@ -19,21 +22,50 @@ export default class RagController {
     const sanitizedName = sanitizeFilename(uploadedFile.clientName)
 
     const fileName = `${sanitizedName}-${randomSuffix}.${uploadedFile.extname || 'txt'}`
-    const fullPath = app.makePath('storage/uploads', fileName)
+    const fullPath = app.makePath(RagService.UPLOADS_STORAGE_PATH, fileName)
 
-    await uploadedFile.move(app.makePath('storage/uploads'), {
+    await uploadedFile.move(app.makePath(RagService.UPLOADS_STORAGE_PATH), {
       name: fileName,
     })
 
-    // Don't await this - process in background
-    this.ragService.processAndEmbedFile(fullPath)
+    // Get file size for tracking
+    let fileSize: number | undefined = undefined
+    try {
+      const stats = await stat(fullPath)
+      fileSize = stats.size
+    } catch (error) {
+      // Not critical if we can't get file size, just swallow the error
+    }
+
+    // Dispatch background job for embedding
+    const result = await EmbedFileJob.dispatch({
+      filePath: fullPath,
+      fileName,
+      fileSize,
+    })
 
-    return response.status(200).json({
-      message: 'File has been uploaded and queued for processing.',
-      file_path: `/uploads/${fileName}`,
+    return response.status(202).json({
+      message: result.message,
+      jobId: result.jobId,
+      fileName,
+      filePath: `/${RagService.UPLOADS_STORAGE_PATH}/${fileName}`,
+      alreadyProcessing: !result.created,
     })
   }
 
+  public async getJobStatus({ request, response }: HttpContext) {
+    const reqData = await request.validateUsing(getJobStatusSchema)
+
+    const fullPath = app.makePath(RagService.UPLOADS_STORAGE_PATH, reqData.filePath)
+    const status = await EmbedFileJob.getStatus(fullPath)
+
+    if (!status.exists) {
+      return response.status(404).json({ error: 'Job not found for this file' })
+    }
+
+    return response.status(200).json(status)
+  }
+
   public async getStoredFiles({ response }: HttpContext) {
     const files = await this.ragService.getStoredFiles()
     return response.status(200).json({ files })

admin/app/jobs/embed_file_job.ts

@@ -0,0 +1,161 @@
+import { Job } from 'bullmq'
+import { QueueService } from '#services/queue_service'
+import { RagService } from '#services/rag_service'
+import { DockerService } from '#services/docker_service'
+import { OllamaService } from '#services/ollama_service'
+import { createHash } from 'crypto'
+import logger from '@adonisjs/core/services/logger'
+
+export interface EmbedFileJobParams {
+  filePath: string
+  fileName: string
+  fileSize?: number
+}
+
+export class EmbedFileJob {
+  static get queue() {
+    return 'file-embeddings'
+  }
+
+  static get key() {
+    return 'embed-file'
+  }
+
+  static getJobId(filePath: string): string {
+    return createHash('sha256').update(filePath).digest('hex').slice(0, 16)
+  }
+
+  async handle(job: Job) {
+    const { filePath, fileName } = job.data as EmbedFileJobParams
+
+    logger.info(`[EmbedFileJob] Starting embedding process for: ${fileName}`)
+
+    const dockerService = new DockerService()
+    const ollamaService = new OllamaService()
+    const ragService = new RagService(dockerService, ollamaService)
+
+    try {
+      // Update progress starting
+      await job.updateProgress(0)
+      await job.updateData({
+        ...job.data,
+        status: 'processing',
+        startedAt: Date.now(),
+      })
+
+      logger.info(`[EmbedFileJob] Processing file: ${filePath}`)
+
+      // Process and embed the file
+      const result = await ragService.processAndEmbedFile(filePath)
+
+      if (!result.success) {
+        logger.error(`[EmbedFileJob] Failed to process file ${fileName}: ${result.message}`)
+        throw new Error(result.message)
+      }
+
+      // Update progress complete
+      await job.updateProgress(100)
+      await job.updateData({
+        ...job.data,
+        status: 'completed',
+        completedAt: Date.now(),
+        chunks: result.chunks,
+      })
+
+      logger.info(
+        `[EmbedFileJob] Successfully embedded ${result.chunks} chunks from file: ${fileName}`
+      )
+
+      return {
+        success: true,
+        fileName,
+        filePath,
+        chunks: result.chunks,
+        message: `Successfully embedded ${result.chunks} chunks`,
+      }
+    } catch (error) {
+      logger.error(`[EmbedFileJob] Error embedding file ${fileName}:`, error)
+
+      await job.updateData({
+        ...job.data,
+        status: 'failed',
+        failedAt: Date.now(),
+        error: error instanceof Error ? error.message : 'Unknown error',
+      })
+
+      throw error
+    }
+  }
+
+  static async getByFilePath(filePath: string): Promise<Job | undefined> {
+    const queueService = new QueueService()
+    const queue = queueService.getQueue(this.queue)
+    const jobId = this.getJobId(filePath)
+    return await queue.getJob(jobId)
+  }
+
+  static async dispatch(params: EmbedFileJobParams) {
+    const queueService = new QueueService()
+    const queue = queueService.getQueue(this.queue)
+    const jobId = this.getJobId(params.filePath)
+
+    try {
+      const job = await queue.add(this.key, params, {
+        jobId,
+        attempts: 3,
+        backoff: {
+          type: 'exponential',
+          delay: 5000, // Delay 5 seconds before retrying
+        },
+        removeOnComplete: { count: 50 }, // Keep last 50 completed jobs for history
+        removeOnFail: { count: 20 } // Keep last 20 failed jobs for debugging
+      })
+
+      logger.info(`[EmbedFileJob] Dispatched embedding job for file: ${params.fileName}`)
+
+      return {
+        job,
+        created: true,
+        jobId,
+        message: `File queued for embedding: ${params.fileName}`,
+      }
+    } catch (error) {
+      if (error.message && error.message.includes('job already exists')) {
+        const existing = await queue.getJob(jobId)
+        logger.info(`[EmbedFileJob] Job already exists for file: ${params.fileName}`)
+        return {
+          job: existing,
+          created: false,
+          jobId,
+          message: `Embedding job already exists for: ${params.fileName}`,
+        }
+      }
+      throw error
+    }
+  }
+
+  static async getStatus(filePath: string): Promise<{
+    exists: boolean
+    status?: string
+    progress?: number
+    chunks?: number
+    error?: string
+  }> {
+    const job = await this.getByFilePath(filePath)
+
+    if (!job) {
+      return { exists: false }
+    }
+
+    const state = await job.getState()
+    const data = job.data
+
+    return {
+      exists: true,
+      status: data.status || state,
+      progress: typeof job.progress === 'number' ? job.progress : undefined,
+      chunks: data.chunks,
+      error: data.error,
+    }
+  }
+}