From 098639e146bfa8443264f9165a629b745cd89834 Mon Sep 17 00:00:00 2001 From: Markus Hartung Date: Wed, 22 Apr 2026 02:10:38 +0200 Subject: [PATCH] fix: add batch item dispatching to specific handlers based on URL (#1495) Integrated into release/v3.7.0 --- open-sse/services/batchProcessor.ts | 67 +++++++++++++++++++++++------ tests/manual/batch.http | 29 +++++++++++++ tests/unit/batch_api.test.ts | 59 +++++++++++++++++++++++++ 3 files changed, 141 insertions(+), 14 deletions(-) diff --git a/open-sse/services/batchProcessor.ts b/open-sse/services/batchProcessor.ts index 315ba94a26..7452b917c6 100644 --- a/open-sse/services/batchProcessor.ts +++ b/open-sse/services/batchProcessor.ts @@ -1,17 +1,21 @@ import { v4 as uuidv4 } from "uuid"; import { - getPendingBatches, - getTerminalBatches, - updateBatch, - getFileContent, createFile, + deleteFile, getApiKeyById, getBatch, + getFileContent, + getPendingBatches, + getTerminalBatches, listFiles, - deleteFile, + updateBatch, updateFileStatus, } from "@/lib/localDb"; import { handleChat } from "@/sse/handlers/chat"; +import { POST as handleEmbeddingsRoute } from "@/app/api/v1/embeddings/route"; +import { POST as handleModerationsRoute } from "@/app/api/v1/moderations/route"; +import { POST as handleImagesGenerationsRoute } from "@/app/api/v1/images/generations/route"; +import { POST as handleVideosGenerationsRoute } from "@/app/api/v1/videos/generations/route"; let isProcessing = false; let pollInterval: NodeJS.Timeout | null = null; @@ -175,6 +179,49 @@ async function startBatch(batch: any) { } } +/** + * Dispatch a single batch item to the correct handler based on the URL. + * Each batch item may target a different API endpoint (chat, embeddings, etc.). + */ +async function dispatchBatchItem( + url: string, + headers: Headers, + body: Record +): Promise { + const request = new Request(`http://localhost${url}`, { + method: "POST", + headers, + body: JSON.stringify(body), + }); + + switch (url) { + case "/v1/embeddings": + return handleEmbeddingsRoute(request); + case "/v1/moderations": + return handleModerationsRoute(request); + case "/v1/images/generations": + case "/v1/images/edits": + return handleImagesGenerationsRoute(request); + case "/v1/videos": + case "/v1/videos/generations": + return handleVideosGenerationsRoute(request); + // /v1/chat/completions, /v1/completions, /v1/responses + default: { + return handleChat( + new Request(`http://localhost${url}`, { + method: "POST", + headers, + body: JSON.stringify({ + ...body, + // BATCH-SPECIFIC: Force stream: false — batches don't support SSE responses + stream: body.stream ?? false, + }), + }) + ); + } + } +} + async function processBatchItems(batch: any, lines: string[]) { const results: any[] = []; const errors: any[] = []; @@ -204,15 +251,7 @@ async function processBatchItems(batch: any, lines: string[]) { } headers.set("Content-Type", "application/json"); - // BATCH-SPECIFIC: Force stream: false — batches don't support SSE responses - const batchItemBody = { ...body, stream: false }; - - const response = await handleChat({ - json: async () => batchItemBody, - url: `http://localhost${url}`, - headers, - method: "POST", - } as any); + const response = await dispatchBatchItem(url, headers, body); let responseData: { error: any; id?: any; usage?: any; model?: any }; let statusCode = 200; diff --git a/tests/manual/batch.http b/tests/manual/batch.http index 131ae14f05..46d7e7b837 100644 --- a/tests/manual/batch.http +++ b/tests/manual/batch.http @@ -21,6 +21,35 @@ batch client.global.set("file_id", response.body.id); %} +### +# @name Upload a JSONL file for batch processing - test embedding endpoint +POST {{omniroute-address}}/v1/files +Authorization: Bearer {{OMNIROUTE_API_KEY}} +Content-Type: multipart/form-data; boundary=boundary + +--boundary +Content-Disposition: form-data; name="file"; filename="batch_input.jsonl" +Content-Type: application/jsonl + +{"custom_id": "request-1", "method": "POST", "url": "/v1/embeddings", "body": {"model": "pinecone/llama-text-embed-v2", "input": "The food was delicious and the waiter was very attentive."}} +{"custom_id": "request-2", "method": "POST", "url": "/v1/embeddings", "body": {"model": "mistral/mistral-embed", "input": "The food was delicious and the waiter was very attentive."}} +{"custom_id": "request-2", "method": "POST", "url": "/v1/embeddings", "body": {"model": "mistral/mistral-embed", "input": "I had a terrible experience at the restaurant. The food was cold and the service was slow."}} +{"custom_id": "request-2", "method": "POST", "url": "/v1/embeddings", "body": {"model": "mistral/mistral-embed", "input": "Too bad! The food was cold and the service was slow."}} +{"custom_id": "request-2", "method": "POST", "url": "/v1/embeddings", "body": {"model": "mistral/mistral-embed", "input": "Best dining experience ever! The food was amazing and the waiter was super friendly."}} +{"custom_id": "request-2", "method": "POST", "url": "/v1/embeddings", "body": {"model": "mistral/mistral-embed", "input": "Lorem ipsum dolor sit amet, consectetur adipiscing elit. Sed do eiusmod tempor incididunt ut labore et dolore magna aliqua."}} +{"custom_id": "request-2", "method": "POST", "url": "/v1/embeddings", "body": {"model": "mistral/mistral-embed", "input": "All you can eat buffet was a disaster. The food was stale and the staff was rude."}} +{"custom_id": "request-2", "method": "POST", "url": "/v1/embeddings", "body": {"model": "mistral/mistral-embed", "input": "All your base are belong to us."}} + +--boundary +Content-Disposition: form-data; name="purpose" + +batch +--boundary-- + +> {% + client.global.set("file_id", response.body.id); +%} + ### # @name List all uploaded files GET {{omniroute-address}}/v1/files diff --git a/tests/unit/batch_api.test.ts b/tests/unit/batch_api.test.ts index 30adc29fa3..bff2c3075e 100644 --- a/tests/unit/batch_api.test.ts +++ b/tests/unit/batch_api.test.ts @@ -703,6 +703,65 @@ test("Retrieve file content spec compliance", async () => { assert.ok(unauthorized); }); +test("Batch dispatches to embeddings handler for /v1/embeddings URL", async () => { + initBatchProcessor(); + try { + const batchItems = [ + JSON.stringify({ + custom_id: "embed-request-1", + method: "POST", + url: "/v1/embeddings", + body: { model: "mistral/mistral-embed", input: "The food was delicious." } + }) + ].join("\n"); + + const file = createFile({ + bytes: Buffer.byteLength(batchItems), + filename: "embed_batch.jsonl", + purpose: "batch", + content: Buffer.from(batchItems), + apiKeyId: null + }); + + const batch = createBatch({ + endpoint: "/v1/embeddings", + completionWindow: "24h", + inputFileId: file.id, + apiKeyId: null + }); + + let maxAttempts = 20; + let currentBatch = getBatch(batch.id); + while (maxAttempts > 0 && currentBatch?.status !== "completed" && currentBatch?.status !== "failed") { + await new Promise(resolve => setTimeout(resolve, 1000)); + currentBatch = getBatch(batch.id); + maxAttempts--; + } + + assert.ok( + currentBatch?.status === "completed" || currentBatch?.status === "failed", + "Batch should reach a terminal state" + ); + assert.strictEqual(currentBatch?.requestCountsTotal, 1); + + // Verify the batch item was dispatched to the embeddings handler, not the chat handler. + // The chat handler would return errors about missing "messages", "Missing model", etc. + // The embeddings handler returns errors about missing credentials or invalid embedding models. + const outputFileId = currentBatch?.outputFileId || currentBatch?.errorFileId; + assert.ok(outputFileId, "Should have an output or error file"); + const outputContent = getFileContent(outputFileId!); + assert.ok(outputContent, "Output file should have content"); + const result = JSON.parse(outputContent.toString()); + const errorMsg = result.response?.body?.error?.message || ""; + assert.ok( + !errorMsg.includes("messages") && !errorMsg.includes("Missing model"), + `Error should not be a chat-specific error. Got: ${errorMsg}` + ); + } finally { + stopBatchProcessor(); + } +}); + test("getTerminalBatches returns only terminal statuses ordered oldest first", async () => { const apiKey = await createApiKey("Terminal Batches Test Key", "test-machine");