diff --git a/internal/controller/chunksgenerator_controller.go b/internal/controller/chunksgenerator_controller.go index 3259640a..74078b50 100644 --- a/internal/controller/chunksgenerator_controller.go +++ b/internal/controller/chunksgenerator_controller.go @@ -189,14 +189,14 @@ func (r *ChunksGeneratorReconciler) processConvertedFile(ctx context.Context, co } // chunk the file - chunksFile, err := r.chunkFile(ctx, convertedFilePath, chunksGeneratorCR) + chunkRows, err := r.chunkFile(ctx, convertedFilePath, chunksGeneratorCR) if err != nil { logger.Error(err, "failed to chunk file") return false, err } // store the chunks in the filestore - chunksFileBytes, err := json.Marshal(chunksFile) + chunksFileBytes, err := json.Marshal(chunkRows) if err != nil { logger.Error(err, "failed to marshal chunks file") return false, err @@ -215,21 +215,12 @@ func (r *ChunksGeneratorReconciler) needsChunking(ctx context.Context, converted chunksFilePath := unstructured.RemapToOutputDir(convertedFilePath, inputPath, outputPath) - // fetch the converted file from the filestore - // this will also make sure that the converted file exists in the filestore - convertedFileRaw, err := r.fileStore.Retrieve(ctx, convertedFilePath) - if err != nil { - return false, err - } - - convertedFile := unstructured.ConvertedFile{} - err = json.Unmarshal(convertedFileRaw, &convertedFile) + // fetch the converted file metadata + _, convertedFileMetadata, err := r.readConvertedFile(ctx, convertedFilePath) if err != nil { return false, err } - convertedFileMetadata := convertedFile.ConvertedDocument.Metadata - // check if the chunked file does not exist in the filestore then return true chunksFileExists, err := r.fileStore.Exists(ctx, chunksFilePath) if err != nil { @@ -245,18 +236,31 @@ func (r *ChunksGeneratorReconciler) needsChunking(ctx context.Context, converted return false, err } - chunksFile := unstructured.ChunksFile{} - err = json.Unmarshal(chunksFileRaw, &chunksFile) - if err != nil { - return false, err - } - - // now the chunks file should be the same as the current chunks file in filestore newChunksFileMetadata := unstructured.ChunksFileMetadata{ ConvertedFileMetadata: convertedFileMetadata, ChunkingTool: unstructured.LangchainChunkingTool, ChunksGeneratorConfig: chunksGeneratorCR.Spec.ChunksGeneratorConfig, } + + // try new array format first + var chunkRows []unstructured.ChunkRow + if err := json.Unmarshal(chunksFileRaw, &chunkRows); err == nil && len(chunkRows) > 0 && chunkRows[0].Metadata != nil { + if chunkRows[0].Metadata.Equal(&newChunksFileMetadata) { + return false, nil + } + logger.Info("chunks file config has changed, re-chunking needed", "file", convertedFilePath) + return true, nil + } + + // fall back to old single-object format + chunksFile := unstructured.ChunksFile{} + if parseErr := json.Unmarshal(chunksFileRaw, &chunksFile); parseErr != nil { + logger.Info("chunks file exists but cannot be parsed, re-chunking needed", "file", convertedFilePath) + return true, nil //nolint:nilerr // unparseable file means re-chunking is needed + } + if chunksFile.ChunksDocument == nil || chunksFile.ChunksDocument.Metadata == nil { + return true, nil + } if !chunksFile.ChunksDocument.Metadata.Equal(&newChunksFileMetadata) { logger.Info("chunks file config has changed, re-chunking needed", "file", convertedFilePath) return true, nil @@ -265,18 +269,31 @@ func (r *ChunksGeneratorReconciler) needsChunking(ctx context.Context, converted return false, nil } -func (r *ChunksGeneratorReconciler) chunkFile(ctx context.Context, convertedFilePath string, chunksGeneratorCR *operatorv1alpha1.ChunksGenerator) (*unstructured.ChunksFile, error) { - logger := log.FromContext(ctx) - logger.Info("chunking file", "file", convertedFilePath) - - // read the converted file from the filestore +func (r *ChunksGeneratorReconciler) readConvertedFile(ctx context.Context, convertedFilePath string) (string, *unstructured.ConvertedFileMetadata, error) { convertedFileRaw, err := r.fileStore.Retrieve(ctx, convertedFilePath) if err != nil { - return nil, err + return "", nil, err } + // try new array format first + var convertedRows []unstructured.ConvertedRow + if err := json.Unmarshal(convertedFileRaw, &convertedRows); err == nil && len(convertedRows) > 0 && convertedRows[0].Metadata != nil { + return convertedRows[0].Markdown, convertedRows[0].Metadata, nil + } + + // fall back to old single-object format convertedFile := unstructured.ConvertedFile{} - err = json.Unmarshal(convertedFileRaw, &convertedFile) + if err := json.Unmarshal(convertedFileRaw, &convertedFile); err != nil { + return "", nil, err + } + return convertedFile.ConvertedDocument.Content.Markdown, convertedFile.ConvertedDocument.Metadata, nil +} + +func (r *ChunksGeneratorReconciler) chunkFile(ctx context.Context, convertedFilePath string, chunksGeneratorCR *operatorv1alpha1.ChunksGenerator) ([]unstructured.ChunkRow, error) { + logger := log.FromContext(ctx) + logger.Info("chunking file", "file", convertedFilePath) + + markdown, convertedMetadata, err := r.readConvertedFile(ctx, convertedFilePath) if err != nil { return nil, err } @@ -321,24 +338,27 @@ func (r *ChunksGeneratorReconciler) chunkFile(ctx context.Context, convertedFile return nil, fmt.Errorf("invalid strategy: %s", chunksGeneratorCR.Spec.ChunksGeneratorConfig.Strategy) } - chunks, err := chunker.Chunk(convertedFile.ConvertedDocument.Content.Markdown) + chunks, err := chunker.Chunk(markdown) if err != nil { return nil, err } - return &unstructured.ChunksFile{ - ConvertedDocument: convertedFile.ConvertedDocument, - ChunksDocument: &unstructured.ChunksDocument{ - Metadata: &unstructured.ChunksFileMetadata{ - ChunkingTool: unstructured.LangchainChunkingTool, - ChunksGeneratorConfig: chunksGeneratorCR.Spec.ChunksGeneratorConfig, - ConvertedFileMetadata: convertedFile.ConvertedDocument.Metadata, - }, - Chunks: &unstructured.Chunks{ - Text: chunks, - }, - }, - }, nil + fileID := convertedMetadata.FileIdentifier + metadata := &unstructured.ChunksFileMetadata{ + ChunkingTool: unstructured.LangchainChunkingTool, + ChunksGeneratorConfig: chunksGeneratorCR.Spec.ChunksGeneratorConfig, + ConvertedFileMetadata: convertedMetadata, + } + rows := make([]unstructured.ChunkRow, len(chunks)) + for i, text := range chunks { + rows[i] = unstructured.ChunkRow{ + FileID: fileID, + ChunkIndex: i, + Text: text, + Metadata: metadata, + } + } + return rows, nil } func (r *ChunksGeneratorReconciler) findDependents(ctx context.Context, obj client.Object) []reconcile.Request { diff --git a/internal/controller/documentprocessor_controller.go b/internal/controller/documentprocessor_controller.go index db84b755..3b312f89 100644 --- a/internal/controller/documentprocessor_controller.go +++ b/internal/controller/documentprocessor_controller.go @@ -232,16 +232,13 @@ func (r *DocumentProcessorReconciler) reconcileJob(ctx context.Context, job oper DocumentConverter: unstructured.DocumentConverterDocling, DoclingConfig: documentProcessorCR.Spec.DocumentProcessorConfig.DoclingConfig, } - convertedFile := unstructured.ConvertedFile{ - ConvertedDocument: &unstructured.ConvertedDocument{ - Metadata: &convertedFileMetadata, - Content: &unstructured.Content{ - Markdown: doclingResponse.Document.MDContent, - }, - }, - } + convertedRows := []unstructured.ConvertedRow{{ + FileID: job.FileIdentifier, + Markdown: doclingResponse.Document.MDContent, + Metadata: &convertedFileMetadata, + }} - convertedFileBytes, err := json.Marshal(convertedFile) + convertedFileBytes, err := json.Marshal(convertedRows) if err != nil { return err } @@ -400,13 +397,6 @@ func (r *DocumentProcessorReconciler) needsConversion(ctx context.Context, rawFi return false, err } - convertedFile := unstructured.ConvertedFile{} - err = json.Unmarshal(convertedFileRaw, &convertedFile) - if err != nil { - return false, err - } - currentConvertedFileMetadata := convertedFile.ConvertedDocument.Metadata - fileToConvertMetadata := unstructured.ConvertedFileMetadata{ RawFilePath: rawFilePath, FileIdentifier: fileUID, @@ -414,9 +404,22 @@ func (r *DocumentProcessorReconciler) needsConversion(ctx context.Context, rawFi DoclingConfig: documentProcessorCR.Spec.DocumentProcessorConfig.DoclingConfig, } - if currentConvertedFileMetadata.Equal(&fileToConvertMetadata) { - logger.Info("converted file has the same configuration, no conversion needed", "filePath", rawFilePath) - return false, nil + // try new array format first + var convertedRows []unstructured.ConvertedRow + if err := json.Unmarshal(convertedFileRaw, &convertedRows); err == nil && len(convertedRows) > 0 && convertedRows[0].Metadata != nil { + if convertedRows[0].Metadata.Equal(&fileToConvertMetadata) { + logger.Info("converted file has the same configuration, no conversion needed", "filePath", rawFilePath) + return false, nil + } + } else { + // fall back to old single-object format + convertedFile := unstructured.ConvertedFile{} + if err := json.Unmarshal(convertedFileRaw, &convertedFile); err == nil && convertedFile.ConvertedDocument != nil && convertedFile.ConvertedDocument.Metadata != nil { + if convertedFile.ConvertedDocument.Metadata.Equal(&fileToConvertMetadata) { + logger.Info("converted file has the same configuration, no conversion needed", "filePath", rawFilePath) + return false, nil + } + } } } diff --git a/internal/controller/vectorembeddingsgenerator_controller.go b/internal/controller/vectorembeddingsgenerator_controller.go index 7fef8ee4..abbe71c8 100644 --- a/internal/controller/vectorembeddingsgenerator_controller.go +++ b/internal/controller/vectorembeddingsgenerator_controller.go @@ -163,31 +163,17 @@ func (r *VectorEmbeddingsGeneratorReconciler) processChunkedFile(ctx context.Con return false, nil } - logger.Info("retrieving chunked file from filestore", "file", chunksFilePath) - chunkedFileRaw, err := r.fileStore.Retrieve(ctx, chunksFilePath) + texts, fileID, convertedMeta, chunksMeta, err := r.readChunksFile(ctx, chunksFilePath) if err != nil { - logger.Error(err, "failed to retrieve chunked file") + logger.Error(err, "failed to read chunked file") return false, err } - chunkedFile := &unstructured.ChunksFile{} - if err := json.Unmarshal(chunkedFileRaw, &chunkedFile); err != nil { - logger.Error(err, "failed to unmarshal chunked file") - return false, err - } - - // Validate chunked file structure - if chunkedFile.ConvertedDocument == nil || chunkedFile.ChunksDocument == nil { - return false, errors.New("invalid chunks file structure: missing required fields") - } - if chunkedFile.ChunksDocument.Chunks == nil || len(chunkedFile.ChunksDocument.Chunks.Text) == 0 { + if len(texts) == 0 { logger.Info("chunks file has no text chunks, skipping", "file", chunksFilePath) return false, nil } - texts := make([]string, len(chunkedFile.ChunksDocument.Chunks.Text)) - copy(texts, chunkedFile.ChunksDocument.Chunks.Text) - vegConfig := vectorEmbeddingsGeneratorCR.Spec.VectorEmbeddingsGeneratorConfig modelName := vegConfig.ModelName @@ -204,8 +190,8 @@ func (r *VectorEmbeddingsGeneratorReconciler) processChunkedFile(ctx context.Con }) embeddingFileMetadata := &unstructured.EmbeddingFileMetadata{ - ConvertedFileMetadata: chunkedFile.ConvertedDocument.Metadata, - ChunkFileMetadata: chunkedFile.ChunksDocument.Metadata, + ConvertedFileMetadata: convertedMeta, + ChunkFileMetadata: chunksMeta, ModelName: modelName, NomicEmbedTextV15Config: vegConfig.NomicEmbedTextV15Config, GeminiEmbedding2Config: vegConfig.GeminiEmbedding2Config, @@ -252,26 +238,18 @@ func (r *VectorEmbeddingsGeneratorReconciler) processChunkedFile(ctx context.Con logger.Info("successfully generated embeddings", "file", chunksFilePath, "embeddingCount", len(allEmbeddings)) - // rearrange the embeddings - embeddings := make([]*unstructured.Embeddings, len(allEmbeddings)) + embeddingRows := make([]unstructured.EmbeddingRow, len(allEmbeddings)) for i, embeddingVector := range allEmbeddings { - embeddings[i] = &unstructured.Embeddings{ - Text: texts[i], - Embedding: embeddingVector, - } - } - - // Create the complete embeddings file structure - embeddingsFile := &unstructured.EmbeddingsFile{ - ConvertedDocument: chunkedFile.ConvertedDocument, - ChunksDocument: chunkedFile.ChunksDocument, - EmbeddingDocument: &unstructured.EmbeddingDocument{ + embeddingRows[i] = unstructured.EmbeddingRow{ + FileID: fileID, + ChunkIndex: i, + Text: texts[i], + Embedding: embeddingVector, Metadata: embeddingFileMetadata, - Embeddings: embeddings, - }, + } } - embeddingsFileBytes, err := json.Marshal(embeddingsFile) + embeddingsFileBytes, err := json.Marshal(embeddingRows) if err != nil { logger.Error(err, "failed to marshal embeddings file") return false, err @@ -288,6 +266,38 @@ func (r *VectorEmbeddingsGeneratorReconciler) processChunkedFile(ctx context.Con return true, nil } +func (r *VectorEmbeddingsGeneratorReconciler) readChunksFile(ctx context.Context, chunksFilePath string) (texts []string, fileID string, convertedMeta *unstructured.ConvertedFileMetadata, chunksMeta *unstructured.ChunksFileMetadata, err error) { + chunkedFileRaw, err := r.fileStore.Retrieve(ctx, chunksFilePath) + if err != nil { + return nil, "", nil, nil, err + } + + // try new array format first + var chunkRows []unstructured.ChunkRow + if err := json.Unmarshal(chunkedFileRaw, &chunkRows); err == nil && len(chunkRows) > 0 && chunkRows[0].Metadata != nil { + texts = make([]string, len(chunkRows)) + for i, row := range chunkRows { + texts[i] = row.Text + } + return texts, chunkRows[0].FileID, chunkRows[0].Metadata.ConvertedFileMetadata, chunkRows[0].Metadata, nil + } + + // fall back to old single-object format + chunkedFile := &unstructured.ChunksFile{} + if err := json.Unmarshal(chunkedFileRaw, chunkedFile); err != nil { + return nil, "", nil, nil, err + } + if chunkedFile.ConvertedDocument == nil || chunkedFile.ChunksDocument == nil { + return nil, "", nil, nil, errors.New("invalid chunks file structure: missing required fields") + } + if chunkedFile.ChunksDocument.Chunks == nil { + return nil, "", nil, nil, nil + } + texts = make([]string, len(chunkedFile.ChunksDocument.Chunks.Text)) + copy(texts, chunkedFile.ChunksDocument.Chunks.Text) + return texts, chunkedFile.ConvertedDocument.Metadata.FileIdentifier, chunkedFile.ConvertedDocument.Metadata, chunkedFile.ChunksDocument.Metadata, nil +} + func (r *VectorEmbeddingsGeneratorReconciler) needsEmbedding(ctx context.Context, chunksFilePath string, vectorEmbeddingsGeneratorCR *operatorv1alpha1.VectorEmbeddingsGenerator, inputPath, outputPath string) (bool, error) { logger := log.FromContext(ctx) logger.Info("checking if file needs embedding", "file", chunksFilePath) @@ -302,16 +312,11 @@ func (r *VectorEmbeddingsGeneratorReconciler) needsEmbedding(ctx context.Context return false, err } - chunkedFileRaw, err := r.fileStore.Retrieve(ctx, chunksFilePath) + _, _, convertedMeta, chunksMeta, err := r.readChunksFile(ctx, chunksFilePath) if err != nil { return false, err } - chunkedFile := &unstructured.ChunksFile{} - if err := json.Unmarshal(chunkedFileRaw, &chunkedFile); err != nil { - return false, err - } - embeddingsFilePath := unstructured.RemapToOutputDir(chunksFilePath, inputPath, outputPath) logger.Info("embeddings file path", "embeddingsFilePath", embeddingsFilePath) embeddingsFileExists, err := r.fileStore.Exists(ctx, embeddingsFilePath) @@ -325,26 +330,35 @@ func (r *VectorEmbeddingsGeneratorReconciler) needsEmbedding(ctx context.Context return false, err } + fileToEmbedMetadata := &unstructured.EmbeddingFileMetadata{ + ConvertedFileMetadata: convertedMeta, + ChunkFileMetadata: chunksMeta, + ModelName: vectorEmbeddingsGeneratorCR.Spec.VectorEmbeddingsGeneratorConfig.ModelName, + NomicEmbedTextV15Config: vectorEmbeddingsGeneratorCR.Spec.VectorEmbeddingsGeneratorConfig.NomicEmbedTextV15Config, + GeminiEmbedding2Config: vectorEmbeddingsGeneratorCR.Spec.VectorEmbeddingsGeneratorConfig.GeminiEmbedding2Config, + } + + // try new array format first + var embeddingRows []unstructured.EmbeddingRow + if err := json.Unmarshal(embeddingsFileRaw, &embeddingRows); err == nil && len(embeddingRows) > 0 && embeddingRows[0].Metadata != nil { + if embeddingRows[0].Metadata.Equal(fileToEmbedMetadata) { + logger.Info("embeddings file has the same configuration, no embedding needed", "file", chunksFilePath) + return false, nil + } + logger.Info("embeddings file exists but with different configuration, will re-embed", "file", chunksFilePath) + return true, nil + } + + // fall back to old single-object format currentEmbeddedFile := &unstructured.EmbeddingsFile{} - if err := json.Unmarshal(embeddingsFileRaw, ¤tEmbeddedFile); err != nil { + if err := json.Unmarshal(embeddingsFileRaw, currentEmbeddedFile); err != nil { logger.Info("embeddings file exists but cannot be parsed, will re-embed", "file", chunksFilePath, "error", err) return true, nil } - - // Check if the embedded file structure is valid if currentEmbeddedFile.EmbeddingDocument == nil || currentEmbeddedFile.EmbeddingDocument.Metadata == nil { logger.Info("embeddings file exists but has invalid structure, will re-embed", "file", chunksFilePath) return true, nil } - - fileToEmbedMetadata := &unstructured.EmbeddingFileMetadata{ - ConvertedFileMetadata: chunkedFile.ConvertedDocument.Metadata, - ChunkFileMetadata: chunkedFile.ChunksDocument.Metadata, - ModelName: vectorEmbeddingsGeneratorCR.Spec.VectorEmbeddingsGeneratorConfig.ModelName, - NomicEmbedTextV15Config: vectorEmbeddingsGeneratorCR.Spec.VectorEmbeddingsGeneratorConfig.NomicEmbedTextV15Config, - GeminiEmbedding2Config: vectorEmbeddingsGeneratorCR.Spec.VectorEmbeddingsGeneratorConfig.GeminiEmbedding2Config, - } - if currentEmbeddedFile.EmbeddingDocument.Metadata.Equal(fileToEmbedMetadata) { logger.Info("embeddings file has the same configuration, no embedding needed", "file", chunksFilePath) return false, nil diff --git a/pkg/unstructured/chunks_file.go b/pkg/unstructured/chunks_file.go index 5a7f7ee2..751bd367 100644 --- a/pkg/unstructured/chunks_file.go +++ b/pkg/unstructured/chunks_file.go @@ -31,6 +31,13 @@ type ChunksFile struct { ChunksDocument *ChunksDocument `json:"chunksDocument"` } +type ChunkRow struct { + FileID string `json:"fileId"` + ChunkIndex int `json:"chunkIndex"` + Text string `json:"text"` + Metadata *ChunksFileMetadata `json:"metadata"` +} + func (c *ChunksFileMetadata) Equal(other *ChunksFileMetadata) bool { if !c.ConvertedFileMetadata.Equal(other.ConvertedFileMetadata) { return false diff --git a/pkg/unstructured/converted_file.go b/pkg/unstructured/converted_file.go index 182d1809..144dedc4 100644 --- a/pkg/unstructured/converted_file.go +++ b/pkg/unstructured/converted_file.go @@ -47,6 +47,12 @@ type ConvertedFile struct { ConvertedDocument *ConvertedDocument `json:"convertedDocument"` } +type ConvertedRow struct { + FileID string `json:"fileId"` + Markdown string `json:"markdown"` + Metadata *ConvertedFileMetadata `json:"metadata"` +} + func (c *ConvertedFileMetadata) Equal(other *ConvertedFileMetadata) bool { if c.RawFilePath != other.RawFilePath { return false diff --git a/pkg/unstructured/embeddings_file.go b/pkg/unstructured/embeddings_file.go index 5b47d95d..3ff11597 100644 --- a/pkg/unstructured/embeddings_file.go +++ b/pkg/unstructured/embeddings_file.go @@ -29,6 +29,14 @@ type EmbeddingsFile struct { EmbeddingDocument *EmbeddingDocument `json:"embeddingDocument"` } +type EmbeddingRow struct { + FileID string `json:"fileId"` + ChunkIndex int `json:"chunkIndex"` + Text string `json:"text"` + Embedding []float64 `json:"embedding"` + Metadata *EmbeddingFileMetadata `json:"metadata"` +} + func (c *EmbeddingFileMetadata) Equal(other *EmbeddingFileMetadata) bool { if !cmp.Equal(c.ConvertedFileMetadata, other.ConvertedFileMetadata) { return false