perf: Add performance optimizations for batch operations

- Add .high priority to BatchEmbedder TaskGroup tasks for better CPU scheduling
- Implement automatic batching for PineconeStore upsert operations (100 vectors/batch)
- Implement automatic batching for PineconeStore delete operations (1000 IDs/batch)
- Execute batches concurrently for optimal throughput
- Respect Pinecone API limits and best practices

These optimizations improve performance for large-scale RAG operations by:
1. Better resource allocation through task prioritization
2. Avoiding API rate limits through intelligent batching
3. Maximizing throughput via concurrent batch processing

Co-authored-by: Christopher Karani <christopherkarani@users.noreply.github.com>

Author: github-actions[bot]

Sources/Zoni/Embedding/BatchEmbedder.swift

@@ -89,7 +89,7 @@ public actor BatchEmbedder {
 
             // Start initial concurrent tasks
             while inFlight < maxConcurrency, let (index, batch) = batchIterator.next() {
-                group.addTask {
+                group.addTask(priority: .high) {
                     let embeddings = try await self.provider.embed(batch)
                     return (index, embeddings)
                 }
@@ -103,7 +103,7 @@ public actor BatchEmbedder {
 
                 // Schedule next batch if available
                 if let (index, batch) = batchIterator.next() {
-                    group.addTask {
+                    group.addTask(priority: .high) {
                         let embeddings = try await self.provider.embed(batch)
                         return (index, embeddings)
                     }
@@ -142,7 +142,7 @@ public actor BatchEmbedder {
             var inFlight = 0
 
             while inFlight < maxConcurrency, let (index, batch) = batchIterator.next() {
-                group.addTask {
+                group.addTask(priority: .high) {
                     let embeddings = try await self.provider.embed(batch)
                     return (index, embeddings)
                 }
@@ -156,7 +156,7 @@ public actor BatchEmbedder {
                 progress(completedBatches, totalBatches)
 
                 if let (index, batch) = batchIterator.next() {
-                    group.addTask {
+                    group.addTask(priority: .high) {
                         let embeddings = try await self.provider.embed(batch)
                         return (index, embeddings)
                     }
@@ -246,7 +246,7 @@ public actor BatchEmbedder {
                                 ))
                                 return
                             }
-                            group.addTask {
+                            group.addTask(priority: .high) {
                                 let embeddings = try await self.provider.embed(batch)
                                 return embeddings.enumerated().map { (offset + $0.offset, $0.element) }
                             }
@@ -267,7 +267,7 @@ public actor BatchEmbedder {
                                     ))
                                     return
                                 }
-                                group.addTask {
+                                group.addTask(priority: .high) {
                                     let embeddings = try await self.provider.embed(batch)
                                     return embeddings.enumerated().map { (offset + $0.offset, $0.element) }
                                 }

Sources/Zoni/VectorStore/Stores/PineconeStore.swift

@@ -184,6 +184,8 @@ public actor PineconeStore: VectorStore {
     ///
     /// This method uses upsert semantics: if a vector with the same ID already
     /// exists, it will be replaced with the new vector and metadata.
+    /// For large batches (>100 vectors), the operation is automatically split
+    /// into multiple requests to respect Pinecone's API limits.
     ///
     /// - Parameters:
     ///   - chunks: The chunks to store. Each chunk must have a unique ID.
@@ -206,10 +208,6 @@ public actor PineconeStore: VectorStore {
     ///
     /// try await store.add(chunks, embeddings: embeddings)
     /// ```
-    ///
-    /// - Note: Pinecone has a limit on the number of vectors per upsert request
-    ///   (typically 100). For large batches, consider splitting into smaller
-    ///   requests.
     public func add(_ chunks: [Chunk], embeddings: [Embedding]) async throws {
         // Validate that counts match
         guard chunks.count == embeddings.count else {
@@ -221,6 +219,40 @@ public actor PineconeStore: VectorStore {
         // Handle empty input
         guard !chunks.isEmpty else { return }
 
+        // Pinecone recommends batches of 100 vectors per upsert
+        let batchSize = 100
+
+        // If within single batch size, execute directly
+        if chunks.count <= batchSize {
+            try await upsertBatch(chunks: chunks, embeddings: embeddings)
+            return
+        }
+
+        // Split into batches and execute concurrently
+        let chunkBatches = chunks.chunked(into: batchSize)
+        let embeddingBatches = embeddings.chunked(into: batchSize)
+
+        try await withThrowingTaskGroup(of: Void.self) { group in
+            for (chunkBatch, embeddingBatch) in zip(chunkBatches, embeddingBatches) {
+                group.addTask {
+                    try await self.upsertBatch(chunks: chunkBatch, embeddings: embeddingBatch)
+                }
+            }
+
+            // Wait for all batches to complete
+            for try await _ in group { }
+        }
+    }
+
+    /// Upserts a single batch of chunks (internal helper).
+    ///
+    /// - Parameters:
+    ///   - chunks: The chunks to upsert (should be ≤100).
+    ///   - embeddings: Corresponding embeddings.
+    /// - Throws: `ZoniError.insertionFailed` if the Pinecone API returns an error.
+    private func upsertBatch(chunks: [Chunk], embeddings: [Embedding]) async throws {
+        guard !chunks.isEmpty else { return }
+
         // Build the upsert request
         guard let url = URL(string: "\(indexHost)/vectors/upsert") else {
             throw ZoniError.insertionFailed(reason: "Invalid Pinecone URL: \(indexHost)")
@@ -370,6 +402,8 @@ public actor PineconeStore: VectorStore {
     /// Deletes chunks with the specified IDs from the Pinecone index.
     ///
     /// IDs that do not exist in the index are silently ignored.
+    /// For large batches (>1000 IDs), the operation is automatically split
+    /// into multiple requests to respect Pinecone's API limits.
     ///
     /// - Parameter ids: The IDs of chunks to delete.
     ///
@@ -380,11 +414,44 @@ public actor PineconeStore: VectorStore {
     /// ```swift
     /// // Delete specific chunks by ID
     /// try await store.delete(ids: ["chunk-1", "chunk-2", "chunk-3"])
+    ///
+    /// // Delete large batch (automatically batched)
+    /// try await store.delete(ids: Array(1...5000).map { "chunk-\($0)" })
     /// ```
     public func delete(ids: [String]) async throws {
         // Handle empty input
         guard !ids.isEmpty else { return }
 
+        // Pinecone recommends batches of 1000 IDs per request
+        let batchSize = 1000
+
+        // If within single batch size, execute directly
+        if ids.count <= batchSize {
+            try await deleteBatch(ids: ids)
+            return
+        }
+
+        // Split into batches and execute concurrently
+        let batches = ids.chunked(into: batchSize)
+        try await withThrowingTaskGroup(of: Void.self) { group in
+            for batch in batches {
+                group.addTask {
+                    try await self.deleteBatch(ids: batch)
+                }
+            }
+
+            // Wait for all batches to complete
+            for try await _ in group { }
+        }
+    }
+
+    /// Deletes a single batch of IDs (internal helper).
+    ///
+    /// - Parameter ids: The IDs to delete (should be ≤1000).
+    /// - Throws: `ZoniError.insertionFailed` if the Pinecone API returns an error.
+    private func deleteBatch(ids: [String]) async throws {
+        guard !ids.isEmpty else { return }
+
         // Build the delete request
         guard let url = URL(string: "\(indexHost)/vectors/delete") else {
             throw ZoniError.insertionFailed(reason: "Invalid Pinecone URL: \(indexHost)")