You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by "Croway (via GitHub)" <gi...@apache.org> on 2026/09/18 13:50:59 UTC
[PR] CAMEL-24674: camel-openai - support the Batch API [camel]
Croway opened a new pull request, #26601:
URL: https://github.com/apache/camel/pull/26601
Adds the OpenAI Batch API to `camel-openai` ([CAMEL-24674](https://issues.apache.org/jira/browse/CAMEL-24674)).
The Batch API runs requests offline at half the price of the synchronous API, against a separate rate limit, and returns the answers as a file within 24 hours. It fits work nobody waits for: overnight enrichment, embedding backfills, moderation backlogs and prompt regression runs. Everything between the upload and the result file is integration work, which is what this makes routable.
## Operations
| Operation | Does |
|---|---|
| `batch` | Uploads the requests as a JSONL file and creates the batch |
| `batch-retrieve` | Reports status and request counts, leaving the body untouched |
| `batch-cancel` | Cancels a batch that is still running |
| `batch-results` | Downloads the output file, or the error file with `batchResultsFile=error` |
## Building the input
The body is either the JSONL itself (`File`, `Path`, `WrappedFile`, `InputStream`, `byte[]`, String), or a `Map` keyed by `custom_id`. For a map the component writes the envelope of every line (`method`, and `url` from `batchEndpoint`), and the value decides the request body:
- a String is turned into a request built from the endpoint options (`/v1/chat/completions`, `/v1/responses`, `/v1/embeddings`, `/v1/moderations`), so a route holding prompts does not assemble API payloads itself
- a `Map` or `JsonNode` is used as the request body as it is, which covers the other endpoints and per-line differences
```java
from("direct:classify")
.setBody(constant(Map.of("ticket-1", "I was charged twice", "ticket-2", "The app crashes")))
.to("openai:batch?batchEndpoint=/v1/chat/completions&model=gpt-4o-mini&systemMessage=Classify the ticket");
from("direct:collect")
.noStreamCaching()
.to("openai:batch-results")
.split(body().tokenize("\n")).streaming()
.setHeader("ticketId", jsonpath("$.custom_id"))
.setBody(jsonpath("$.response.body.choices[0].message.content"))
.to("sql:update ticket set category = :#${body} where id = :#${header.ticketId}")
.end();
```
## Behaviour worth reviewing
- The endpoint is validated against the eight endpoints the API supports **before** anything is uploaded, so a typo leaves no orphan file.
- A create that fails deletes the file it just uploaded, attaching any delete failure as a suppressed exception.
- Map bodies are written to a temporary file, so the SDK can retry the upload; a one-shot stream could not be resent. The file is deleted afterwards.
- `batch-results` works for `completed`, `expired` and `cancelled`; it fails while the batch is still running, and reports the validation errors for a `failed` batch (also in `CamelOpenAIBatchErrors`).
- A batch without failures has no error file, so the body is `null` and splits into nothing.
- The results stream is closed when the exchange completes, and must be consumed inside the route.
- `streaming`, `conversationMemory`, `mcpServer` and `tags` are rejected rather than silently ignored, since a batch line is a single request.
- Every SDK call goes through `GenAiErrorSupport`. No GenAI span is recorded: creating a batch is not an inference call and `GenAiOperationName` has no value for it.
- Result lines are model output, so the docs unmarshal them to a `Map` and never to a type the content chooses.
## Test infra
`camel-test-infra-openai-mock` now serves the Files and Batch APIs: upload, retrieve and download of files, batch create/retrieve/cancel, a configurable status progression advancing one step per retrieve, and output/error files built from expectations matched by `custom_id`. `getBatchStore().getUploadedFile()` returns the JSONL a route produced.
## Verified
- **On the mock:** 9 tests in `OpenAIBatchMockTest` β prompts uploaded as JSONL with the envelope and options applied, raw JSONL passed through unchanged, an unsupported endpoint rejected before upload, status and counts while polling, cancel, output streaming correlated by `custom_id`, the error file, results requested too early, and a missing batch id. The full `camel-openai` suite passes.
- **Not verified:** nothing has run against the real Batch API or another implementation yet. An opt-in `OpenAIBatchExternalServiceIT` (Llama Stack in front of vLLM/Ollama, then OpenAI itself) is the next step, along with checking what partial results an expired or cancelled batch actually returns.
Pre-existing and unrelated: `OpenAIEmbeddingsIT` fails locally because the Ollama model `granite-embedding:30m` is not pulled.
## Follow-up
Generic Files API operations (`file-upload`, `file-content`, `file-delete`) are deliberately left out; they belong with Responses `input_file` and vector stores in their own ticket. Reacting to the `batch.completed` webhook instead of polling is CAMEL-24673.
_Claude Code on behalf of Federico Mariani (Croway)_
π€ Generated with [Claude Code](https://claude.com/claude-code)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-24674: camel-openai - support the Batch API [camel]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #26601:
URL: https://github.com/apache/camel/pull/26601#issuecomment-5731607756
<!-- ci-tested-modules -->
:test_tube: **CI tested the following changed modules:**
- `catalog/camel-catalog`
- `components/camel-ai/camel-openai`
- `docs`
- `dsl/camel-endpointdsl`
- `test-infra/camel-test-infra-openai-mock`
---
<details><summary>:microscope: Scalpel shadow comparison β Scalpel: 13 of 692 tested, 24 compile-only β current: 13 all tested</summary>
[Maveniverse Scalpel](https://github.com/maveniverse/scalpel) detected **40 affected modules** (current approach: 13).
**Skip-tests mode would test 13 modules** (5 direct + 11 downstream), **skip tests for 24** (generated code, meta-modules)
<details><summary>:warning: Modules only in Scalpel (27)</summary>
- `apache-camel`
- `camel-allcomponents`
- `camel-catalog`
- `camel-catalog-console`
- `camel-catalog-maven`
- `camel-catalog-suggest`
- `camel-componentdsl`
- `camel-endpointdsl`
- `camel-endpointdsl-support`
- `camel-itest`
- `camel-jbang-core`
- `camel-jbang-it`
- `camel-jbang-main`
- `camel-jbang-plugin-edit`
- `camel-jbang-plugin-generate`
- `camel-jbang-plugin-kubernetes`
- `camel-jbang-plugin-test`
- `camel-kamelet-main`
- `camel-launcher`
- `camel-report-maven-plugin`
- `camel-route-parser`
- `camel-yaml-dsl`
- `camel-yaml-dsl-deserializers`
- `camel-yaml-dsl-maven-plugin`
- `coverage`
- `docs`
- `dummy-component`
</details>
<details><summary>Modules Scalpel would test (13)</summary>
- `camel-catalog` β catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/openai.json, catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs.properties, catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/openai-batch.adoc, catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/openai-component.adoc
- `camel-endpointdsl` β dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/StaticEndpointBuilders.java, dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/OpenAIEndpointBuilderFactory.java
- `camel-jbang-mcp` β downstream of org.apache.camel:camel-catalog
- `camel-jbang-plugin-mcp` β downstream of org.apache.camel:camel-jbang-core
- `camel-jbang-plugin-route-parser` β downstream of org.apache.camel:camel-route-parser
- `camel-jbang-plugin-tui` β downstream of org.apache.camel:camel-catalog
- `camel-jbang-plugin-validate` β downstream of org.apache.camel:camel-yaml-dsl-validator
- `camel-langchain4j-agent` β downstream of org.apache.camel:camel-test-infra-openai-mock
- `camel-launcher-container` β downstream of org.apache.camel:camel-launcher
- `camel-mcp-server` β downstream of org.apache.camel:camel-openai
- `camel-openai` β components/camel-ai/camel-openai/src/generated/java/org/apache/camel/component/openai/OpenAIEndpointConfigurer.java, components/camel-ai/camel-openai/src/generated/java/org/apache/camel/component/openai/OpenAIEndpointUriFactory.java, components/camel-ai/camel-openai/src/generated/resources/META-INF/org/apache/camel/component/openai/openai.json, components/camel-ai/camel-openai/src/main/docs/openai-batch.adoc, components/camel-ai/camel-openai/src/main/docs/openai-component.adoc, components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIBatchProducer.java, components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIBatchResultsProducer.java, components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIBatchStoredProducer.java, components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIBatchSupport.java, components/camel-ai/camel-openai/src/main/java/org/
apache/camel/component/openai/OpenAIConfiguration.java, components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIConstants.java, components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIEndpoint.java, components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIOperations.java, components/camel-ai/camel-openai/src/test/java/org/apache/camel/component/openai/OpenAIBatchMockTest.java
- `camel-test-infra-all` β downstream of org.apache.camel:camel-test-infra-openai-mock
- `camel-test-infra-openai-mock` β test-infra/camel-test-infra-openai-mock/README.md, test-infra/camel-test-infra-openai-mock/src/main/java/org/apache/camel/test/infra/openai/mock/BatchExpectation.java, test-infra/camel-test-infra-openai-mock/src/main/java/org/apache/camel/test/infra/openai/mock/BatchRequestHandler.java, test-infra/camel-test-infra-openai-mock/src/main/java/org/apache/camel/test/infra/openai/mock/BatchStore.java, test-infra/camel-test-infra-openai-mock/src/main/java/org/apache/camel/test/infra/openai/mock/OpenAIMock.java, test-infra/camel-test-infra-openai-mock/src/main/java/org/apache/camel/test/infra/openai/mock/OpenAIMockBuilder.java, test-infra/camel-test-infra-openai-mock/src/main/java/org/apache/camel/test/infra/openai/mock/OpenAIMockExpectations.java, test-infra/camel-test-infra-openai-mock/src/main/java/org/apache/camel/test/infra/openai/mock/OpenAIMockServerHandler.java
- `camel-yaml-dsl-validator` β downstream of org.apache.camel:camel-catalog
- `camel-yaml-dsl-validator-maven-plugin` β downstream of org.apache.camel:camel-yaml-dsl-validator
- `docs` β docs/components/modules/others/nav.adoc, docs/components/modules/others/pages/openai-batch.adoc
</details>
<details><summary>Modules with tests skipped (24)</summary>
- `apache-camel`
- `camel-allcomponents`
- `camel-catalog-console`
- `camel-catalog-maven`
- `camel-catalog-suggest`
- `camel-componentdsl`
- `camel-endpointdsl-support`
- `camel-itest`
- `camel-jbang-core`
- `camel-jbang-it`
- `camel-jbang-main`
- `camel-jbang-plugin-edit`
- `camel-jbang-plugin-generate`
- `camel-jbang-plugin-kubernetes`
- `camel-jbang-plugin-test`
- `camel-kamelet-main`
- `camel-launcher`
- `camel-report-maven-plugin`
- `camel-route-parser`
- `camel-yaml-dsl`
- `camel-yaml-dsl-deserializers`
- `camel-yaml-dsl-maven-plugin`
- `coverage`
- `dummy-component`
</details>
> :information_source: Shadow mode β Scalpel observes but does not affect test execution. [Learn more](https://github.com/maveniverse/scalpel)
</details>
:warning: **Some tests are disabled on GitHub Actions** (`@DisabledIfSystemProperty(named = "ci.env.name")`) and require manual verification:
- `components/camel-ai/camel-openai`: 7 test(s) disabled on GitHub Actions
<details><summary>All tested modules (40 modules, 6m 6s total)</summary>
**Total reactor time:** 6m 6s
| Module | Duration | Status |
| --- | --- | --- |
| Camel :: YAML DSL | 54.0s | SUCCESS |
| Camel :: AI :: MCP Server | 43.1s | SUCCESS |
| Camel :: Launcher | 37.2s | SUCCESS |
| Camel :: AI :: OpenAI | 30.9s | SUCCESS |
| Camel :: JBang :: MCP | 29.4s | SUCCESS |
| Camel :: JBang :: Plugin :: TUI | 25.0s | SUCCESS |
| Camel :: AI :: LangChain4j :: Agent | 19.9s | SUCCESS |
| Camel :: Catalog :: Camel Catalog | 15.6s | SUCCESS |
| Camel :: Component DSL | 15.2s | SUCCESS |
| Camel :: Test Infra :: All test services | 12.2s | SUCCESS |
| Camel :: JBang :: Plugin :: Kubernetes | 10.7s | SUCCESS |
| Camel :: Docs | 9.0s | SUCCESS |
| Camel :: Catalog :: Camel Report Maven Plugin | 6.4s | SUCCESS |
| Camel :: JBang :: Plugin :: Testing | 6.0s | SUCCESS |
| Camel :: Kamelet Main | 5.9s | SUCCESS |
| Camel :: YAML DSL :: Validator | 5.6s | SUCCESS |
| Camel :: Catalog :: Camel Route Parser | 5.5s | SUCCESS |
| Camel :: Test Infra :: OpenAI Mock | 4.0s | SUCCESS |
| Camel :: All Components Sync point | 4.0s | SUCCESS |
| Camel :: JBang :: Plugin :: Validate | 3.7s | SUCCESS |
| Camel :: YAML DSL :: Deserializers | 3.6s | SUCCESS |
| Camel :: YAML DSL :: Validator Maven Plugin | 2.7s | SUCCESS |
| Camel :: Catalog :: Maven | 2.3s | SUCCESS |
| Camel :: YAML DSL :: Maven Plugins | 2.2s | SUCCESS |
| Camel :: Assembly | 1.7s | SUCCESS |
| Camel :: Catalog :: Suggest (deprecated) | 1.6s | SUCCESS |
| Camel :: Coverage | 1.4s | SUCCESS |
| Camel :: JBang :: Plugin :: Edit | 1.2s | SUCCESS |
| Camel :: Endpoint DSL :: Support | 1.0s | SUCCESS |
| Camel :: JBang :: Plugin :: Generate | 0.9s | SUCCESS |
| Camel :: JBang :: Integration tests | 0.8s | SUCCESS |
| Camel :: Catalog :: Dummy Component | 0.8s | SUCCESS |
| Camel :: JBang :: Plugin :: MCP | 0.7s | SUCCESS |
| Camel :: Launcher :: Container | 0.6s | SUCCESS |
| Camel :: Catalog :: Console | 0.6s | SUCCESS |
| Camel :: JBang :: Main | 0.6s | SUCCESS |
| Camel :: JBang :: Plugin :: Route Parser | 0.4s | SUCCESS |
| Camel :: Endpoint DSL | n/a | |
| Camel :: Integration Tests | n/a | |
| Camel :: JBang :: Core | n/a | |
**Top 20 slowest modules:**
- `Camel :: YAML DSL` (54.0s)
- `Camel :: AI :: MCP Server` (43.1s)
- `Camel :: Launcher` (37.2s)
- `Camel :: AI :: OpenAI` (30.9s)
- `Camel :: JBang :: MCP` (29.4s)
- `Camel :: JBang :: Plugin :: TUI` (25.0s)
- `Camel :: AI :: LangChain4j :: Agent` (19.9s)
- `Camel :: Catalog :: Camel Catalog` (15.6s)
- `Camel :: Component DSL` (15.2s)
- `Camel :: Test Infra :: All test services` (12.2s)
- `Camel :: JBang :: Plugin :: Kubernetes` (10.7s)
- `Camel :: Docs` (9.0s)
- `Camel :: Catalog :: Camel Report Maven Plugin` (6.4s)
- `Camel :: JBang :: Plugin :: Testing` (6.0s)
- `Camel :: Kamelet Main` (5.9s)
- `Camel :: YAML DSL :: Validator` (5.6s)
- `Camel :: Catalog :: Camel Route Parser` (5.5s)
- `Camel :: Test Infra :: OpenAI Mock` (4.0s)
- `Camel :: All Components Sync point` (4.0s)
- `Camel :: JBang :: Plugin :: Validate` (3.7s)
</details>
---
:gear: [View full build and test results](https://github.com/apache/camel/actions/runs/35354690038)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-24674: camel-openai - support the Batch API [camel]
Posted by "gnodet-bot (via GitHub)" <gi...@apache.org>.
gnodet-bot commented on code in PR #26601:
URL: https://github.com/apache/camel/pull/26601#discussion_r4048104047
##########
test-infra/camel-test-infra-openai-mock/src/main/java/org/apache/camel/test/infra/openai/mock/BatchStore.java:
##########
@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.test.infra.openai.mock;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * In-memory state behind the Files and Batch APIs of the mock: the uploaded input files, the result files built for a
+ * finished batch, and the batches themselves. Tests can read the uploaded content back to assert on the JSONL a route
+ * produced.
+ */
+public class BatchStore {
+
+ /**
+ * A file uploaded by a route or generated by the mock for a finished batch.
+ */
+ public record StoredFile(String id, String filename, String purpose, byte[] content) {
+ }
+
+ /**
+ * A created batch and the position it reached in the configured status progression.
+ */
+ public static class StoredBatch {
+ final String id;
+ final String inputFileId;
+ final String endpoint;
+ final String metadata;
+ int statusIndex;
+ boolean cancelling;
+ boolean cancelled;
+ boolean resultsBuilt;
+ String outputFileId;
+ String errorFileId;
+ int total;
+ int completed;
+ int failed;
Review Comment:
β οΈ **Data race on `StoredBatch` mutable fields (raised in previous review, not yet addressed)**
`StoredBatch` exposes nine plain, non-`volatile` instance fields (`statusIndex`, `cancelling`, `cancelled`, `resultsBuilt`, `outputFileId`, `errorFileId`, `total`, `completed`, `failed`). These are mutated by `BatchRequestHandler` methods (`retrieveBatch`, `cancelBatch`, `maybeBuildResults`) which are dispatched by `com.sun.net.httpserver.HttpServer` β a thread-pool executor. A simultaneous retrieve + cancel from a polled test will data-race on `cancelling`/`cancelled`/`statusIndex` with no synchronization beyond the `ConcurrentHashMap` holding the batch reference (which only guarantees safe publication, not ongoing field visibility).
In practice, tests run sequentially and the race window is narrow, but the mock is designed to be shared infrastructure. The fix is straightforward β either make all mutable fields `volatile`, or synchronize the `BatchRequestHandler` methods on the `StoredBatch` instance:
```suggestion
volatile int statusIndex;
volatile boolean cancelling;
volatile boolean cancelled;
volatile boolean resultsBuilt;
volatile String outputFileId;
volatile String errorFileId;
volatile int total;
volatile int completed;
volatile int failed;
```
##########
test-infra/camel-test-infra-openai-mock/src/main/java/org/apache/camel/test/infra/openai/mock/BatchStore.java:
##########
@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.test.infra.openai.mock;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * In-memory state behind the Files and Batch APIs of the mock: the uploaded input files, the result files built for a
+ * finished batch, and the batches themselves. Tests can read the uploaded content back to assert on the JSONL a route
+ * produced.
+ */
+public class BatchStore {
+
+ /**
+ * A file uploaded by a route or generated by the mock for a finished batch.
+ */
+ public record StoredFile(String id, String filename, String purpose, byte[] content) {
+ }
+
+ /**
+ * A created batch and the position it reached in the configured status progression.
+ */
+ public static class StoredBatch {
+ final String id;
+ final String inputFileId;
+ final String endpoint;
+ final String metadata;
+ int statusIndex;
+ boolean cancelling;
+ boolean cancelled;
+ boolean resultsBuilt;
+ String outputFileId;
+ String errorFileId;
+ int total;
+ int completed;
+ int failed;
+
Review Comment:
β οΈ **Data race on `StoredBatch` mutable fields**
`BatchStore` wraps batch objects in a `ConcurrentHashMap`, but that only protects map-level operations. Once a `StoredBatch` reference is retrieved, concurrent `retrieveBatch()`, `cancelBatch()`, and `maybeBuildResults()` calls in `BatchRequestHandler` race on the same object's plain `int`/`boolean` fields. `com.sun.net.httpserver` uses a thread pool, so two parallel HTTP requests against the same batch can produce corrupted state.
The most dangerous race is in `maybeBuildResults`: two concurrent retrieves that both see a batch reaching its final status will both read `resultsBuilt == false`, both set it to `true`, and both build output/error files β producing duplicate stored files and wrong counts.
**Fix:** Either declare the mutable fields `volatile` and use an `AtomicBoolean` for `resultsBuilt`, or add a `synchronized (batch)` guard in the methods that mutate it.
```suggestion
public static class StoredBatch {
final String id;
final String inputFileId;
final String endpoint;
final String metadata;
volatile int statusIndex;
volatile boolean cancelling;
volatile boolean cancelled;
volatile boolean resultsBuilt;
volatile String outputFileId;
volatile String errorFileId;
volatile int total;
volatile int completed;
volatile int failed;
```
##########
test-infra/camel-test-infra-openai-mock/src/main/java/org/apache/camel/test/infra/openai/mock/BatchRequestHandler.java:
##########
@@ -0,0 +1,404 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.test.infra.openai.mock;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.regex.Pattern;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.sun.net.httpserver.HttpExchange;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Serves the Files and Batch APIs: uploading an input file, creating a batch from it, reporting its status and serving
+ * the result files.
+ * <p>
+ * A batch walks the configured status progression, one step per retrieve, so a route that polls sees the same sequence
+ * of states as it would against the real API. A cancel moves a running batch to {@code cancelling}, and the next
+ * retrieve to {@code cancelled}. Once a final status is reached the output and error files are built from the input
+ * file, matching each request line to a {@link BatchExpectation} by its {@code custom_id}.
+ */
+public class BatchRequestHandler {
+ private static final Logger LOG = LoggerFactory.getLogger(BatchRequestHandler.class);
+
+ static final List<String> DEFAULT_STATUSES = List.of("validating", "in_progress", "finalizing", "completed");
+
+ private static final String DEFAULT_RESPONSE_BODY
+ = "{\"id\":\"chatcmpl-mock\",\"object\":\"chat.completion\",\"model\":\"openai-mock\","
+ + "\"choices\":[{\"index\":0,\"message\":{\"role\":\"assistant\",\"content\":\"mock response\"},"
+ + "\"finish_reason\":\"stop\"}]}";
+
+ private final List<BatchExpectation> expectations;
+ private final List<String> statuses;
+ private final BatchStore store;
+ private final ObjectMapper objectMapper;
+
+ public BatchRequestHandler(List<BatchExpectation> expectations, List<String> statuses, BatchStore store,
+ ObjectMapper objectMapper) {
+ this.expectations = expectations;
+ this.statuses = statuses;
+ this.store = store;
+ this.objectMapper = objectMapper;
+ }
+
+ /**
+ * Whether this handler serves the given path, so the dispatcher can route Files and Batch calls here before the
+ * handlers of the inference APIs.
+ */
+ public static boolean handles(String path) {
+ return path.endsWith("/files") || path.contains("/files/")
+ || path.endsWith("/batches") || path.contains("/batches/");
+ }
+
+ public void handleRequest(HttpExchange exchange) throws IOException {
+ String path = exchange.getRequestURI().getPath();
+ String method = exchange.getRequestMethod();
+ try {
+ if (path.endsWith("/files") && "POST".equalsIgnoreCase(method)) {
+ uploadFile(exchange);
+ } else if (path.endsWith("/content") && "GET".equalsIgnoreCase(method)) {
+ fileContent(exchange, idBefore(path, "/content"));
+ } else if (path.contains("/files/") && "GET".equalsIgnoreCase(method)) {
+ fileObject(exchange, lastSegment(path));
+ } else if (path.contains("/files/") && "DELETE".equalsIgnoreCase(method)) {
+ deleteFile(exchange, lastSegment(path));
+ } else if (path.endsWith("/batches") && "POST".equalsIgnoreCase(method)) {
+ createBatch(exchange);
+ } else if (path.endsWith("/cancel") && "POST".equalsIgnoreCase(method)) {
+ cancelBatch(exchange, idBefore(path, "/cancel"));
+ } else if (path.contains("/batches/") && "GET".equalsIgnoreCase(method)) {
+ retrieveBatch(exchange, lastSegment(path));
+ } else {
+ sendError(exchange, 404, "invalid_request_error", "Unsupported request: " + method + " " + path);
+ }
+ } catch (Exception e) {
+ LOG.error("Error processing batch request {} {}", method, path, e);
+ sendError(exchange, 500, "internal_error", String.valueOf(e.getMessage()));
+ }
+ }
+
+ private void uploadFile(HttpExchange exchange) throws IOException {
+ byte[] body;
+ try (InputStream is = exchange.getRequestBody()) {
+ body = is.readAllBytes();
+ }
+ String contentType = exchange.getRequestHeaders().getFirst("Content-Type");
+ String boundary = contentType != null && contentType.contains("boundary=")
+ ? contentType.substring(contentType.indexOf("boundary=") + "boundary=".length()).trim() : null;
+ if (boundary == null) {
+ sendError(exchange, 400, "invalid_request_error", "The file upload must be a multipart request");
+ return;
+ }
+
+ // ISO-8859-1 maps every byte to one character, so the file part survives the split unchanged
+ String raw = new String(body, StandardCharsets.ISO_8859_1);
+ String filename = "unknown";
+ String purpose = "unknown";
+ byte[] content = new byte[0];
+ // the boundary may hold regex metacharacters, so it is quoted rather than spliced into the pattern
+ for (String part : raw.split(Pattern.quote("--" + boundary))) {
+ int headerEnd = part.indexOf("\r\n\r\n");
+ if (headerEnd < 0) {
+ continue;
+ }
+ String headers = part.substring(0, headerEnd);
+ String value = part.substring(headerEnd + 4);
+ if (value.endsWith("\r\n")) {
+ value = value.substring(0, value.length() - 2);
+ }
+ if (headers.contains("name=\"file\"")) {
+ filename = between(headers, "filename=\"", "\"");
+ content = value.getBytes(StandardCharsets.ISO_8859_1);
+ } else if (headers.contains("name=\"purpose\"")) {
+ purpose = value.trim();
+ }
+ }
+
+ String id = store.addFile(filename, purpose, content);
+ LOG.debug("Stored uploaded file {} ({} bytes, purpose {})", id, content.length, purpose);
+ sendJson(exchange, 200, fileNode(id, filename, purpose, content.length).toString());
+ }
+
+ private void fileContent(HttpExchange exchange, String fileId) throws IOException {
+ byte[] content = store.getFileContent(fileId);
+ if (content == null) {
+ sendError(exchange, 404, "invalid_request_error", "No such file: " + fileId);
+ return;
+ }
+ exchange.getResponseHeaders().add("Content-Type", "application/octet-stream");
+ exchange.sendResponseHeaders(200, content.length);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(content);
+ }
+ }
+
+ private void fileObject(HttpExchange exchange, String fileId) throws IOException {
+ BatchStore.StoredFile file = store.file(fileId);
+ if (file == null) {
+ sendError(exchange, 404, "invalid_request_error", "No such file: " + fileId);
+ return;
+ }
+ sendJson(exchange, 200, fileNode(file.id(), file.filename(), file.purpose(), file.content().length).toString());
+ }
+
+ private void deleteFile(HttpExchange exchange, String fileId) throws IOException {
+ if (store.file(fileId) == null) {
+ sendError(exchange, 404, "invalid_request_error", "No such file: " + fileId);
+ return;
+ }
+ store.removeFile(fileId);
+ ObjectNode node = objectMapper.createObjectNode();
+ node.put("id", fileId);
+ node.put("object", "file");
+ node.put("deleted", true);
+ sendJson(exchange, 200, node.toString());
+ }
+
+ private void createBatch(HttpExchange exchange) throws IOException {
+ JsonNode request;
+ try (InputStream is = exchange.getRequestBody()) {
+ request = objectMapper.readTree(is.readAllBytes());
+ }
+ String inputFileId = request.path("input_file_id").asText();
+ if (store.file(inputFileId) == null) {
+ sendError(exchange, 404, "invalid_request_error", "No such file: " + inputFileId);
+ return;
+ }
+ JsonNode metadata = request.get("metadata");
+ BatchStore.StoredBatch batch = store.addBatch(inputFileId, request.path("endpoint").asText(),
+ metadata != null && !metadata.isNull() ? metadata.toString() : null);
+ maybeBuildResults(batch);
+ LOG.debug("Created batch {} for input file {}", batch.id, inputFileId);
+ sendJson(exchange, 200, batchNode(batch).toString());
+ }
+
+ private void retrieveBatch(HttpExchange exchange, String batchId) throws IOException {
+ BatchStore.StoredBatch batch = store.batch(batchId);
+ if (batch == null) {
+ sendError(exchange, 404, "invalid_request_error", "No such batch: " + batchId);
+ return;
+ }
+ if (batch.cancelling) {
+ // a cancelled batch is final, so the retrieve after the cancel reports it as such, as the API does
+ batch.cancelling = false;
+ batch.cancelled = true;
+ } else if (!batch.cancelled && batch.statusIndex < statusList().size() - 1) {
+ batch.statusIndex++;
+ }
+ maybeBuildResults(batch);
+ sendJson(exchange, 200, batchNode(batch).toString());
+ }
+
+ private void cancelBatch(HttpExchange exchange, String batchId) throws IOException {
+ BatchStore.StoredBatch batch = store.batch(batchId);
+ if (batch == null) {
+ sendError(exchange, 404, "invalid_request_error", "No such batch: " + batchId);
+ return;
+ }
+ if (isFinal(status(batch))) {
+ sendError(exchange, 400, "invalid_request_error",
+ "Cannot cancel a batch with status '" + status(batch) + "'.");
+ return;
+ }
+ batch.cancelling = true;
+ sendJson(exchange, 200, batchNode(batch).toString());
+ }
+
+ private List<String> statusList() {
+ return statuses.isEmpty() ? DEFAULT_STATUSES : statuses;
+ }
+
+ private String status(BatchStore.StoredBatch batch) {
+ if (batch.cancelled) {
+ return "cancelled";
+ }
+ if (batch.cancelling) {
+ return "cancelling";
+ }
+ return statusList().get(Math.min(batch.statusIndex, statusList().size() - 1));
+ }
+
+ private boolean isFinal(String status) {
+ return "completed".equals(status) || "expired".equals(status) || "cancelled".equals(status);
+ }
+
+ /**
+ * Builds the output and error files once the batch reaches a final status, as the API does when it finishes
+ * processing the input file.
+ */
+ private void maybeBuildResults(BatchStore.StoredBatch batch) throws IOException {
+ if (batch.resultsBuilt || !isFinal(status(batch))) {
+ return;
+ }
+ batch.resultsBuilt = true;
+
+ BatchStore.StoredFile input = store.file(batch.inputFileId);
+ List<String> outputLines = new ArrayList<>();
+ List<String> errorLines = new ArrayList<>();
+ int line = 0;
+ for (String requestLine : new String(input.content(), StandardCharsets.UTF_8).split("\n")) {
Review Comment:
β οΈ **JSONL split on `\n` breaks on Windows-style line endings**
An input file with `\r\n` line endings β which a `WrappedFile` body from a Windows filesystem, or a `String` body produced with `System.lineSeparator()` on Windows, will have β produces request lines with a trailing `\r`. `objectMapper.readTree(requestLine)` then returns a `custom_id` value like `"ticket-1\r"`, which matches no expectation, and every line lands in the error file instead of the output file. The multipart parser above already guards against this (it strips `\r\n` from part values); the JSONL reader should too.
```suggestion
for (String requestLine : new String(input.content(), StandardCharsets.UTF_8).split("\\r?\\n")) {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-24674: camel-openai - support the Batch API [camel]
Posted by "gnodet-bot (via GitHub)" <gi...@apache.org>.
gnodet-bot commented on code in PR #26601:
URL: https://github.com/apache/camel/pull/26601#discussion_r4053319678
##########
components/camel-ai/camel-openai/src/main/docs/openai-batch.adoc:
##########
@@ -0,0 +1,179 @@
+= OpenAI - Batch API Operations
+:tabs-sync-option:
+
+xref:ROOT:openai-component.adoc[Back to OpenAI Component]
+
+The Batch API runs requests offline at half the price of the synchronous API, against a separate rate limit, and
+returns the answers within 24 hours as a file. It suits work nobody waits for: overnight enrichment, embedding
+backfills, moderation backlogs and prompt regression runs.
+
+Four operations cover the flow:
+
+* `batch` - uploads the requests as a JSONL file and creates the batch
+* `batch-retrieve` - reports the status and request counts of a batch
+* `batch-cancel` - cancels a batch that is still running
+* `batch-results` - downloads the output file, or the error file
+
+== Building the input
+
+The body of the `batch` operation is either the JSONL itself (`File`, `Path`, `WrappedFile`, `InputStream`,
+`byte[]` or String), or a `Map` keyed by `custom_id`, the id that ties each answer back to its request.
+
+For a map, the component writes the envelope of every line (`method` and the `url` of `batchEndpoint`). What the
+value holds decides the request body:
+
+* a String is turned into a request built from the endpoint options, for `/v1/chat/completions`, `/v1/responses`,
+ `/v1/embeddings` and `/v1/moderations`. The request is the one the synchronous operation of that endpoint would
+ send, so the same options and headers apply: `model`, `systemMessage`, `developerMessage`, `temperature`, `topP`,
+ `maxTokens`, `outputClass`, `jsonSchema` and `additionalBodyProperty` for chat completions and responses,
+ `embeddingModel` and `dimensions` for embeddings, and `moderationModel` for moderations
+* a `Map` or `JsonNode` is used as the request body as it is, which is what the other endpoints need, and what
+ lets each line use different parameters
+
+The `CamelOpenAIBatchEndpoint` header overrides `batchEndpoint` for one message. A file body is uploaded under its
+own name with `.jsonl` appended when it has another extension, the only one the Files API accepts for a batch input.
+
+[tabs]
+====
+Java::
++
+[source,java]
+----
+// prompts: the request of every line comes from the options
+from("direct:classify")
+ .setBody(constant(Map.of("ticket-1", "I was charged twice", "ticket-2", "The app crashes")))
+ .to("openai:batch?batchEndpoint=/v1/chat/completions&model=gpt-4o-mini"
+ + "&systemMessage=Classify the ticket&batchMetadata.job=triage")
+ .log("Batch ${header.CamelOpenAIBatchId} is ${header.CamelOpenAIBatchStatus}");
+
+// full request bodies: a different model per line
+from("direct:mixed")
+ .setBody(constant(Map.of("a", Map.of("model", "gpt-4o-mini", "messages",
+ List.of(Map.of("role", "user", "content", "Summarise"))))))
+ .to("openai:batch?batchEndpoint=/v1/chat/completions");
+
+// a JSONL file produced elsewhere is uploaded unchanged
+from("file:batches/inbox?include=.*\\.jsonl")
+ .to("openai:batch?batchEndpoint=/v1/chat/completions");
+----
+
+YAML::
++
+[source,yaml]
+----
+- from:
+ uri: direct:classify
+ steps:
+ - to: "openai:batch?batchEndpoint=/v1/chat/completions&model=gpt-4o-mini&systemMessage=Classify the ticket"
+ - log: "Batch ${header.CamelOpenAIBatchId} is ${header.CamelOpenAIBatchStatus}"
+----
+====
+
+== Waiting for the batch
+
+`batch-retrieve` reads `CamelOpenAIBatchId` and reports the batch on the headers, leaving the body untouched. A
+short job can poll in place:
+
+[source,java]
+----
+from("direct:wait")
+ .loopDoWhile(simple("${header.CamelOpenAIBatchStatus} !in 'completed,failed,expired,cancelled'"))
+ .delay(60000)
+ .to("openai:batch-retrieve")
+ .end()
+ .to("direct:collect");
+----
+
+A job that must survive a restart should store `CamelOpenAIBatchId` instead, and poll it from a scheduler route:
+
+[source,java]
+----
+from("scheduler:batches?delay=600000")
+ .to("sql:select id from openai_batch where status not in ('processed', 'failed')")
+ .split(body())
+ .setHeader("CamelOpenAIBatchId", simple("${body[id]}"))
+ .to("openai:batch-retrieve")
+ .to("sql:update openai_batch set status = :#CamelOpenAIBatchStatus where id = :#CamelOpenAIBatchId")
+ .filter(header("CamelOpenAIBatchStatus").in("completed", "expired", "cancelled"))
+ .to("direct:collect")
+ .to("sql:update openai_batch set status = 'processed' where id = :#CamelOpenAIBatchId")
+ .end()
+ .end();
+----
+
+The row is marked `processed` only after `direct:collect` has read the results, so a failure while reading leaves
+it at `completed` and the next run retries it.
+
+== Reading the results
+
+`batch-results` reads `CamelOpenAIBatchId`, downloads the output file and sets it as an `InputStream`, so it can be
+split line by line. Set `batchResultsFile=error` for the failed requests instead. Each line carries the `custom_id`
+of its request.
+
+The polling routes above hand a finished batch to `direct:collect`, which reads the file. With the defaults, stream
+caching reads the file into memory when the next step runs, and the splitter works on that copy. For a batch of a
+few thousand requests this is fine, and the route needs nothing special:
+
+[source,java]
+----
+from("direct:collect")
+ .to("openai:batch-results")
+ .split(body().tokenize("\n"))
+ .setHeader("ticketId", jsonpath("$.custom_id"))
+ .setBody(jsonpath("$.response.body.choices[0].message.content"))
+ .to("sql:update ticket set category = :#${body} where id = :#${header.ticketId}")
+ .end();
+----
+
+A result file can reach hundreds of megabytes, and holding it in memory is then the problem. Turning stream
+caching off on the route keeps the body as the response stream, and a splitter in streaming mode reads it one line
+at a time as each line is processed. The body can then be read only once, which is what the splitter does:
+
+[source,java]
+----
+from("direct:collect")
+ .noStreamCaching() // keep the response as a stream instead of caching it in memory
+ .to("openai:batch-results")
+ .split(body().tokenize("\n")).streaming() // read one line at a time
+ .setHeader("ticketId", jsonpath("$.custom_id"))
+ .setBody(jsonpath("$.response.body.choices[0].message.content"))
+ .to("sql:update ticket set category = :#${body} where id = :#${header.ticketId}")
+ .end();
+----
+
+An embedding backfill works the same way, straight into a vector store. Embedding vectors make for large result
+files, so this one uses the streaming form. The batch is created by the first route, and a polling route like the
+ones above hands it to `direct:store` once it is finished:
+
+[source,java]
+----
+from("direct:embeddings")
+ .setBody(constant(Map.of("doc-1", "Camel routes messages", "doc-2", "Batches run offline")))
+ .to("openai:batch?batchEndpoint=/v1/embeddings&embeddingModel=text-embedding-3-small");
+
+from("direct:store")
+ .noStreamCaching()
+ .to("openai:batch-results")
+ .split(body().tokenize("\n")).streaming()
+ .setHeader("CamelPgVectorAction", constant("UPSERT"))
+ .setHeader("CamelPgVectorRecordId", jsonpath("$.custom_id"))
+ .setBody(jsonpath("$.response.body.data[0].embedding"))
Review Comment:
π Makes sense β the explicit `(Number) β Float` step in the example is the right pragmatic fix for now, and filing a follow-up to let vector store components accept any `Number` is the correct long-term approach. Nothing blocking here.
_generated by an AI agent, Hermès_ <!-- reviewer: gnodet-bot -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-24674: camel-openai - support the Batch API [camel]
Posted by "gnodet-bot (via GitHub)" <gi...@apache.org>.
gnodet-bot commented on code in PR #26601:
URL: https://github.com/apache/camel/pull/26601#discussion_r4053391141
##########
components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIBatchAggregationStrategy.java:
##########
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.openai;
+
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.camel.AggregationStrategy;
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.util.IOHelper;
+import org.apache.camel.util.ObjectHelper;
+
+/**
+ * Collects the messages of an aggregation into the input of a batch, so a stream of prompts becomes one
+ * {@code openai:batch} call:
+ *
+ * <pre>
+ * from("kafka:tickets")
+ * .aggregate(constant(true), new OpenAIBatchAggregationStrategy())
+ * .completionSize(1000).completionTimeout(60000)
+ * .to("openai:batch?batchEndpoint=/v1/chat/completions&model=gpt-4o-mini");
+ * </pre>
+ *
+ * The body of each message is the value of its request, as for a {@link Map} body of the {@code batch} operation: a
+ * String is turned into a request from the endpoint options, a {@code Map} or {@code JsonNode} is the request body. Its
+ * {@code custom_id} is the {@code CamelOpenAIBatchCustomId} header when set, and the message id otherwise, so the
+ * result lines can be matched back to what produced them.
+ * <p>
+ * By default the requests are collected in a {@link Map} in memory, which suits batches of a few thousand short
+ * requests. {@link #spooled()} writes them to a file instead, one line per message, so a batch of any size is
+ * aggregated with constant memory, and a persistent {@code AggregationRepository} stores a path rather than the
+ * requests. The {@code batch} operation streams either form and removes the file once the batch is created.
+ */
+public class OpenAIBatchAggregationStrategy implements AggregationStrategy {
+
+ private final Path spoolDirectory;
+ // the writer of each spool being aggregated, closed when its aggregation completes
+ private final Map<File, BufferedWriter> writers = new ConcurrentHashMap<>();
+
+ /**
+ * Collects the requests in memory.
+ */
+ public OpenAIBatchAggregationStrategy() {
+ this(null);
+ }
+
+ /**
+ * Collects the requests in a file created in the given directory.
+ */
+ public OpenAIBatchAggregationStrategy(Path spoolDirectory) {
+ this.spoolDirectory = spoolDirectory;
+ }
+
+ /**
+ * Collects the requests in a file created in the temporary directory of the JVM.
+ */
+ public static OpenAIBatchAggregationStrategy spooled() {
+ return new OpenAIBatchAggregationStrategy(Path.of(System.getProperty("java.io.tmpdir")));
+ }
+
+ public boolean isSpooled() {
+ return spoolDirectory != null;
+ }
+
+ @Override
+ public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
+ Message message = newExchange.getMessage();
+ String customId = message.getHeader(OpenAIConstants.BATCH_CUSTOM_ID, message.getMessageId(), String.class);
+ Object value = message.getBody();
+ if (ObjectHelper.isEmpty(customId) || value == null) {
+ throw new IllegalArgumentException(
+ "A message aggregated into a batch needs a custom_id and a body; got custom_id " + customId
+ + " and body " + value);
+ }
+
+ try {
+ if (oldExchange == null) {
+ if (isSpooled()) {
+ OpenAIBatchSpool spool = OpenAIBatchSpool.create(spoolDirectory);
+ OpenAIBatchSpool.append(writer(spool), customId, value);
+ message.setBody(spool);
+ } else {
+ Map<String, Object> requests = new LinkedHashMap<>();
+ requests.put(customId, value);
+ message.setBody(requests);
+ }
+ return newExchange;
+ }
+
+ Object aggregated = oldExchange.getMessage().getBody();
+ if (aggregated instanceof OpenAIBatchSpool spool) {
+ OpenAIBatchSpool.append(writer(spool), customId, value);
+ } else if (aggregated instanceof Map<?, ?> requests) {
+ @SuppressWarnings("unchecked")
+ Map<String, Object> map = (Map<String, Object>) requests;
+ map.put(customId, value);
+ } else {
+ throw new IllegalStateException("Unexpected aggregated batch body: " + aggregated);
+ }
+ return oldExchange;
+ } catch (IOException e) {
+ throw new RuntimeCamelException("Cannot spool the batch request of custom_id " + customId, e);
+ }
+ }
+
+ @Override
+ public void onCompletion(Exchange exchange) {
+ if (exchange != null && exchange.getMessage().getBody() instanceof OpenAIBatchSpool spool) {
+ BufferedWriter writer = writers.remove(spool.getFile());
+ if (writer != null) {
+ IOHelper.close(writer);
+ }
+ }
+ }
+
+ private BufferedWriter writer(OpenAIBatchSpool spool) throws IOException {
+ BufferedWriter writer = writers.get(spool.getFile());
+ if (writer == null) {
+ writer = spool.open();
+ writers.put(spool.getFile(), writer);
+ }
+ return writer;
Review Comment:
π **`writer()` still has a non-atomic check-then-act on `ConcurrentHashMap`** (raised on `06cc6a25`, not yet addressed)
```java
private BufferedWriter writer(OpenAIBatchSpool spool) throws IOException {
BufferedWriter writer = writers.get(spool.getFile());
if (writer == null) {
writer = spool.open();
writers.put(spool.getFile(), writer);
}
return writer;
}
```
The `get` + conditional `put` is not atomic: two concurrent `aggregate()` calls for the same spool could both see `null`, open two writers for the same file, and one of them is leaked β neither `onCompletion` nor a second `computeIfAbsent` will close the overwritten writer.
Camel's aggregator does serialize calls per group key in practice, so this is not a live bug today β but the `ConcurrentHashMap` field signals that concurrent access is anticipated. `computeIfAbsent` makes the intent explicit and closes the race in one line:
```suggestion
private BufferedWriter writer(OpenAIBatchSpool spool) throws IOException {
return writers.computeIfAbsent(spool.getFile(), f -> {
try {
return spool.open();
} catch (IOException e) {
throw new UncheckedIOException(e);
}
});
}
```
(`UncheckedIOException` is already imported via `OpenAIBatchSpool`'s iterator β add the import if the compiler complains.)
_generated by an AI agent, Hermès_ <!-- reviewer: gnodet-bot -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-24674: camel-openai - support the Batch API [camel]
Posted by "Croway (via GitHub)" <gi...@apache.org>.
Croway commented on code in PR #26601:
URL: https://github.com/apache/camel/pull/26601#discussion_r4053303589
##########
components/camel-ai/camel-openai/src/main/docs/openai-batch.adoc:
##########
@@ -0,0 +1,179 @@
+= OpenAI - Batch API Operations
+:tabs-sync-option:
+
+xref:ROOT:openai-component.adoc[Back to OpenAI Component]
+
+The Batch API runs requests offline at half the price of the synchronous API, against a separate rate limit, and
+returns the answers within 24 hours as a file. It suits work nobody waits for: overnight enrichment, embedding
+backfills, moderation backlogs and prompt regression runs.
+
+Four operations cover the flow:
+
+* `batch` - uploads the requests as a JSONL file and creates the batch
+* `batch-retrieve` - reports the status and request counts of a batch
+* `batch-cancel` - cancels a batch that is still running
+* `batch-results` - downloads the output file, or the error file
+
+== Building the input
+
+The body of the `batch` operation is either the JSONL itself (`File`, `Path`, `WrappedFile`, `InputStream`,
+`byte[]` or String), or a `Map` keyed by `custom_id`, the id that ties each answer back to its request.
+
+For a map, the component writes the envelope of every line (`method` and the `url` of `batchEndpoint`). What the
+value holds decides the request body:
+
+* a String is turned into a request built from the endpoint options, for `/v1/chat/completions`, `/v1/responses`,
+ `/v1/embeddings` and `/v1/moderations`. The request is the one the synchronous operation of that endpoint would
+ send, so the same options and headers apply: `model`, `systemMessage`, `developerMessage`, `temperature`, `topP`,
+ `maxTokens`, `outputClass`, `jsonSchema` and `additionalBodyProperty` for chat completions and responses,
+ `embeddingModel` and `dimensions` for embeddings, and `moderationModel` for moderations
+* a `Map` or `JsonNode` is used as the request body as it is, which is what the other endpoints need, and what
+ lets each line use different parameters
+
+The `CamelOpenAIBatchEndpoint` header overrides `batchEndpoint` for one message. A file body is uploaded under its
+own name with `.jsonl` appended when it has another extension, the only one the Files API accepts for a batch input.
+
+[tabs]
+====
+Java::
++
+[source,java]
+----
+// prompts: the request of every line comes from the options
+from("direct:classify")
+ .setBody(constant(Map.of("ticket-1", "I was charged twice", "ticket-2", "The app crashes")))
+ .to("openai:batch?batchEndpoint=/v1/chat/completions&model=gpt-4o-mini"
+ + "&systemMessage=Classify the ticket&batchMetadata.job=triage")
+ .log("Batch ${header.CamelOpenAIBatchId} is ${header.CamelOpenAIBatchStatus}");
+
+// full request bodies: a different model per line
+from("direct:mixed")
+ .setBody(constant(Map.of("a", Map.of("model", "gpt-4o-mini", "messages",
+ List.of(Map.of("role", "user", "content", "Summarise"))))))
+ .to("openai:batch?batchEndpoint=/v1/chat/completions");
+
+// a JSONL file produced elsewhere is uploaded unchanged
+from("file:batches/inbox?include=.*\\.jsonl")
+ .to("openai:batch?batchEndpoint=/v1/chat/completions");
+----
+
+YAML::
++
+[source,yaml]
+----
+- from:
+ uri: direct:classify
+ steps:
+ - to: "openai:batch?batchEndpoint=/v1/chat/completions&model=gpt-4o-mini&systemMessage=Classify the ticket"
+ - log: "Batch ${header.CamelOpenAIBatchId} is ${header.CamelOpenAIBatchStatus}"
+----
+====
+
+== Waiting for the batch
+
+`batch-retrieve` reads `CamelOpenAIBatchId` and reports the batch on the headers, leaving the body untouched. A
+short job can poll in place:
+
+[source,java]
+----
+from("direct:wait")
+ .loopDoWhile(simple("${header.CamelOpenAIBatchStatus} !in 'completed,failed,expired,cancelled'"))
+ .delay(60000)
+ .to("openai:batch-retrieve")
+ .end()
+ .to("direct:collect");
+----
+
+A job that must survive a restart should store `CamelOpenAIBatchId` instead, and poll it from a scheduler route:
+
+[source,java]
+----
+from("scheduler:batches?delay=600000")
+ .to("sql:select id from openai_batch where status not in ('processed', 'failed')")
+ .split(body())
+ .setHeader("CamelOpenAIBatchId", simple("${body[id]}"))
+ .to("openai:batch-retrieve")
+ .to("sql:update openai_batch set status = :#CamelOpenAIBatchStatus where id = :#CamelOpenAIBatchId")
+ .filter(header("CamelOpenAIBatchStatus").in("completed", "expired", "cancelled"))
+ .to("direct:collect")
+ .to("sql:update openai_batch set status = 'processed' where id = :#CamelOpenAIBatchId")
+ .end()
+ .end();
+----
+
+The row is marked `processed` only after `direct:collect` has read the results, so a failure while reading leaves
+it at `completed` and the next run retries it.
+
+== Reading the results
+
+`batch-results` reads `CamelOpenAIBatchId`, downloads the output file and sets it as an `InputStream`, so it can be
+split line by line. Set `batchResultsFile=error` for the failed requests instead. Each line carries the `custom_id`
+of its request.
+
+The polling routes above hand a finished batch to `direct:collect`, which reads the file. With the defaults, stream
+caching reads the file into memory when the next step runs, and the splitter works on that copy. For a batch of a
+few thousand requests this is fine, and the route needs nothing special:
+
+[source,java]
+----
+from("direct:collect")
+ .to("openai:batch-results")
+ .split(body().tokenize("\n"))
+ .setHeader("ticketId", jsonpath("$.custom_id"))
+ .setBody(jsonpath("$.response.body.choices[0].message.content"))
+ .to("sql:update ticket set category = :#${body} where id = :#${header.ticketId}")
+ .end();
+----
+
+A result file can reach hundreds of megabytes, and holding it in memory is then the problem. Turning stream
+caching off on the route keeps the body as the response stream, and a splitter in streaming mode reads it one line
+at a time as each line is processed. The body can then be read only once, which is what the splitter does:
+
+[source,java]
+----
+from("direct:collect")
+ .noStreamCaching() // keep the response as a stream instead of caching it in memory
+ .to("openai:batch-results")
+ .split(body().tokenize("\n")).streaming() // read one line at a time
+ .setHeader("ticketId", jsonpath("$.custom_id"))
+ .setBody(jsonpath("$.response.body.choices[0].message.content"))
+ .to("sql:update ticket set category = :#${body} where id = :#${header.ticketId}")
+ .end();
+----
+
+An embedding backfill works the same way, straight into a vector store. Embedding vectors make for large result
+files, so this one uses the streaming form. The batch is created by the first route, and a polling route like the
+ones above hands it to `direct:store` once it is finished:
+
+[source,java]
+----
+from("direct:embeddings")
+ .setBody(constant(Map.of("doc-1", "Camel routes messages", "doc-2", "Batches run offline")))
+ .to("openai:batch?batchEndpoint=/v1/embeddings&embeddingModel=text-embedding-3-small");
+
+from("direct:store")
+ .noStreamCaching()
+ .to("openai:batch-results")
+ .split(body().tokenize("\n")).streaming()
+ .setHeader("CamelPgVectorAction", constant("UPSERT"))
+ .setHeader("CamelPgVectorRecordId", jsonpath("$.custom_id"))
+ .setBody(jsonpath("$.response.body.data[0].embedding"))
Review Comment:
Confirmed, thanks: camel-jsonpath yields `Double` and `PgVectorProducer.toFloatArray` casts to `Float`. In f08c4bee the example converts the vector to `Float` before `pgvector:`, with a sentence saying why. The proper fix is for the vector store components to accept any `Number`, which affects pgvector, qdrant, milvus and weaviate alike; I will file that separately so JSON-sourced vectors work without the conversion step.
_Claude Code on behalf of Federico Mariani (Croway)_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-24674: camel-openai - support the Batch API [camel]
Posted by "davsclaus (via GitHub)" <gi...@apache.org>.
davsclaus merged PR #26601:
URL: https://github.com/apache/camel/pull/26601
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-24674: camel-openai - support the Batch API [camel]
Posted by "Croway (via GitHub)" <gi...@apache.org>.
Croway commented on code in PR #26601:
URL: https://github.com/apache/camel/pull/26601#discussion_r4048117625
##########
test-infra/camel-test-infra-openai-mock/src/main/java/org/apache/camel/test/infra/openai/mock/BatchStore.java:
##########
@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.test.infra.openai.mock;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * In-memory state behind the Files and Batch APIs of the mock: the uploaded input files, the result files built for a
+ * finished batch, and the batches themselves. Tests can read the uploaded content back to assert on the JSONL a route
+ * produced.
+ */
+public class BatchStore {
+
+ /**
+ * A file uploaded by a route or generated by the mock for a finished batch.
+ */
+ public record StoredFile(String id, String filename, String purpose, byte[] content) {
+ }
+
+ /**
+ * A created batch and the position it reached in the configured status progression.
+ */
+ public static class StoredBatch {
+ final String id;
+ final String inputFileId;
+ final String endpoint;
+ final String metadata;
+ int statusIndex;
+ boolean cancelling;
+ boolean cancelled;
+ boolean resultsBuilt;
Review Comment:
The mock serves requests on `Executors.newSingleThreadExecutor()` (`OpenAIMock` line 127), so `StoredBatch` is only ever touched by one thread and there is no race to guard. The `ConcurrentHashMap` covers the test thread reading the store while the server thread writes. Leaving as is.
_Claude Code on behalf of Federico Mariani (Croway)_
##########
test-infra/camel-test-infra-openai-mock/src/main/java/org/apache/camel/test/infra/openai/mock/BatchRequestHandler.java:
##########
@@ -0,0 +1,404 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.test.infra.openai.mock;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.regex.Pattern;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.sun.net.httpserver.HttpExchange;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Serves the Files and Batch APIs: uploading an input file, creating a batch from it, reporting its status and serving
+ * the result files.
+ * <p>
+ * A batch walks the configured status progression, one step per retrieve, so a route that polls sees the same sequence
+ * of states as it would against the real API. A cancel moves a running batch to {@code cancelling}, and the next
+ * retrieve to {@code cancelled}. Once a final status is reached the output and error files are built from the input
+ * file, matching each request line to a {@link BatchExpectation} by its {@code custom_id}.
+ */
+public class BatchRequestHandler {
+ private static final Logger LOG = LoggerFactory.getLogger(BatchRequestHandler.class);
+
+ static final List<String> DEFAULT_STATUSES = List.of("validating", "in_progress", "finalizing", "completed");
+
+ private static final String DEFAULT_RESPONSE_BODY
+ = "{\"id\":\"chatcmpl-mock\",\"object\":\"chat.completion\",\"model\":\"openai-mock\","
+ + "\"choices\":[{\"index\":0,\"message\":{\"role\":\"assistant\",\"content\":\"mock response\"},"
+ + "\"finish_reason\":\"stop\"}]}";
+
+ private final List<BatchExpectation> expectations;
+ private final List<String> statuses;
+ private final BatchStore store;
+ private final ObjectMapper objectMapper;
+
+ public BatchRequestHandler(List<BatchExpectation> expectations, List<String> statuses, BatchStore store,
+ ObjectMapper objectMapper) {
+ this.expectations = expectations;
+ this.statuses = statuses;
+ this.store = store;
+ this.objectMapper = objectMapper;
+ }
+
+ /**
+ * Whether this handler serves the given path, so the dispatcher can route Files and Batch calls here before the
+ * handlers of the inference APIs.
+ */
+ public static boolean handles(String path) {
+ return path.endsWith("/files") || path.contains("/files/")
+ || path.endsWith("/batches") || path.contains("/batches/");
+ }
+
+ public void handleRequest(HttpExchange exchange) throws IOException {
+ String path = exchange.getRequestURI().getPath();
+ String method = exchange.getRequestMethod();
+ try {
+ if (path.endsWith("/files") && "POST".equalsIgnoreCase(method)) {
+ uploadFile(exchange);
+ } else if (path.endsWith("/content") && "GET".equalsIgnoreCase(method)) {
+ fileContent(exchange, idBefore(path, "/content"));
+ } else if (path.contains("/files/") && "GET".equalsIgnoreCase(method)) {
+ fileObject(exchange, lastSegment(path));
+ } else if (path.contains("/files/") && "DELETE".equalsIgnoreCase(method)) {
+ deleteFile(exchange, lastSegment(path));
+ } else if (path.endsWith("/batches") && "POST".equalsIgnoreCase(method)) {
+ createBatch(exchange);
+ } else if (path.endsWith("/cancel") && "POST".equalsIgnoreCase(method)) {
+ cancelBatch(exchange, idBefore(path, "/cancel"));
+ } else if (path.contains("/batches/") && "GET".equalsIgnoreCase(method)) {
+ retrieveBatch(exchange, lastSegment(path));
+ } else {
+ sendError(exchange, 404, "invalid_request_error", "Unsupported request: " + method + " " + path);
+ }
+ } catch (Exception e) {
+ LOG.error("Error processing batch request {} {}", method, path, e);
+ sendError(exchange, 500, "internal_error", String.valueOf(e.getMessage()));
+ }
+ }
+
+ private void uploadFile(HttpExchange exchange) throws IOException {
+ byte[] body;
+ try (InputStream is = exchange.getRequestBody()) {
+ body = is.readAllBytes();
+ }
+ String contentType = exchange.getRequestHeaders().getFirst("Content-Type");
+ String boundary = contentType != null && contentType.contains("boundary=")
+ ? contentType.substring(contentType.indexOf("boundary=") + "boundary=".length()).trim() : null;
+ if (boundary == null) {
+ sendError(exchange, 400, "invalid_request_error", "The file upload must be a multipart request");
+ return;
+ }
+
+ // ISO-8859-1 maps every byte to one character, so the file part survives the split unchanged
+ String raw = new String(body, StandardCharsets.ISO_8859_1);
+ String filename = "unknown";
+ String purpose = "unknown";
+ byte[] content = new byte[0];
+ // the boundary may hold regex metacharacters, so it is quoted rather than spliced into the pattern
+ for (String part : raw.split(Pattern.quote("--" + boundary))) {
+ int headerEnd = part.indexOf("\r\n\r\n");
+ if (headerEnd < 0) {
+ continue;
+ }
+ String headers = part.substring(0, headerEnd);
+ String value = part.substring(headerEnd + 4);
+ if (value.endsWith("\r\n")) {
+ value = value.substring(0, value.length() - 2);
+ }
+ if (headers.contains("name=\"file\"")) {
+ filename = between(headers, "filename=\"", "\"");
+ content = value.getBytes(StandardCharsets.ISO_8859_1);
+ } else if (headers.contains("name=\"purpose\"")) {
+ purpose = value.trim();
+ }
+ }
+
+ String id = store.addFile(filename, purpose, content);
+ LOG.debug("Stored uploaded file {} ({} bytes, purpose {})", id, content.length, purpose);
+ sendJson(exchange, 200, fileNode(id, filename, purpose, content.length).toString());
+ }
+
+ private void fileContent(HttpExchange exchange, String fileId) throws IOException {
+ byte[] content = store.getFileContent(fileId);
+ if (content == null) {
+ sendError(exchange, 404, "invalid_request_error", "No such file: " + fileId);
+ return;
+ }
+ exchange.getResponseHeaders().add("Content-Type", "application/octet-stream");
+ exchange.sendResponseHeaders(200, content.length);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(content);
+ }
+ }
+
+ private void fileObject(HttpExchange exchange, String fileId) throws IOException {
+ BatchStore.StoredFile file = store.file(fileId);
+ if (file == null) {
+ sendError(exchange, 404, "invalid_request_error", "No such file: " + fileId);
+ return;
+ }
+ sendJson(exchange, 200, fileNode(file.id(), file.filename(), file.purpose(), file.content().length).toString());
+ }
+
+ private void deleteFile(HttpExchange exchange, String fileId) throws IOException {
+ if (store.file(fileId) == null) {
+ sendError(exchange, 404, "invalid_request_error", "No such file: " + fileId);
+ return;
+ }
+ store.removeFile(fileId);
+ ObjectNode node = objectMapper.createObjectNode();
+ node.put("id", fileId);
+ node.put("object", "file");
+ node.put("deleted", true);
+ sendJson(exchange, 200, node.toString());
+ }
+
+ private void createBatch(HttpExchange exchange) throws IOException {
+ JsonNode request;
+ try (InputStream is = exchange.getRequestBody()) {
+ request = objectMapper.readTree(is.readAllBytes());
+ }
+ String inputFileId = request.path("input_file_id").asText();
+ if (store.file(inputFileId) == null) {
+ sendError(exchange, 404, "invalid_request_error", "No such file: " + inputFileId);
+ return;
+ }
+ JsonNode metadata = request.get("metadata");
+ BatchStore.StoredBatch batch = store.addBatch(inputFileId, request.path("endpoint").asText(),
+ metadata != null && !metadata.isNull() ? metadata.toString() : null);
+ maybeBuildResults(batch);
+ LOG.debug("Created batch {} for input file {}", batch.id, inputFileId);
+ sendJson(exchange, 200, batchNode(batch).toString());
+ }
+
+ private void retrieveBatch(HttpExchange exchange, String batchId) throws IOException {
+ BatchStore.StoredBatch batch = store.batch(batchId);
+ if (batch == null) {
+ sendError(exchange, 404, "invalid_request_error", "No such batch: " + batchId);
+ return;
+ }
+ if (batch.cancelling) {
+ // a cancelled batch is final, so the retrieve after the cancel reports it as such, as the API does
+ batch.cancelling = false;
+ batch.cancelled = true;
+ } else if (!batch.cancelled && batch.statusIndex < statusList().size() - 1) {
+ batch.statusIndex++;
+ }
+ maybeBuildResults(batch);
+ sendJson(exchange, 200, batchNode(batch).toString());
+ }
+
+ private void cancelBatch(HttpExchange exchange, String batchId) throws IOException {
+ BatchStore.StoredBatch batch = store.batch(batchId);
+ if (batch == null) {
+ sendError(exchange, 404, "invalid_request_error", "No such batch: " + batchId);
+ return;
+ }
+ if (isFinal(status(batch))) {
+ sendError(exchange, 400, "invalid_request_error",
+ "Cannot cancel a batch with status '" + status(batch) + "'.");
+ return;
+ }
+ batch.cancelling = true;
+ sendJson(exchange, 200, batchNode(batch).toString());
+ }
+
+ private List<String> statusList() {
+ return statuses.isEmpty() ? DEFAULT_STATUSES : statuses;
+ }
+
+ private String status(BatchStore.StoredBatch batch) {
+ if (batch.cancelled) {
+ return "cancelled";
+ }
+ if (batch.cancelling) {
+ return "cancelling";
+ }
+ return statusList().get(Math.min(batch.statusIndex, statusList().size() - 1));
+ }
+
+ private boolean isFinal(String status) {
+ return "completed".equals(status) || "expired".equals(status) || "cancelled".equals(status);
+ }
+
+ /**
+ * Builds the output and error files once the batch reaches a final status, as the API does when it finishes
+ * processing the input file.
+ */
+ private void maybeBuildResults(BatchStore.StoredBatch batch) throws IOException {
+ if (batch.resultsBuilt || !isFinal(status(batch))) {
+ return;
+ }
+ batch.resultsBuilt = true;
+
+ BatchStore.StoredFile input = store.file(batch.inputFileId);
+ List<String> outputLines = new ArrayList<>();
+ List<String> errorLines = new ArrayList<>();
+ int line = 0;
+ for (String requestLine : new String(input.content(), StandardCharsets.UTF_8).split("\n")) {
Review Comment:
Applied in d2a51a238e8b: the split is now `\\r?\\n`. The `newLine()` trigger no longer exists (lines are joined with an explicit `\\n` since ddafa304), but a raw JSONL body produced on Windows still can carry CRLF.
_Claude Code on behalf of Federico Mariani (Croway)_
##########
components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIBatchResultsProducer.java:
##########
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.openai;
+
+import java.util.Optional;
+
+import com.openai.core.http.HttpResponse;
+import com.openai.models.batches.Batch;
+import org.apache.camel.CamelExchangeException;
+import org.apache.camel.Exchange;
+import org.apache.camel.component.ai.observability.GenAiErrorSupport;
+import org.apache.camel.support.DefaultProducer;
+import org.apache.camel.support.SynchronizationAdapter;
+import org.apache.camel.util.ObjectHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * OpenAI producer for the {@code batch-results} operation, which downloads the output or error file of a finished
+ * batch.
+ * <p>
+ * The body is the file as an {@link java.io.InputStream}, so a large result can be split line by line with the splitter
+ * in streaming mode. The HTTP response is closed when the exchange completes. Stream caching reads such a body into
+ * memory before the next step, so a route handling large results should turn it off.
+ */
+public class OpenAIBatchResultsProducer extends DefaultProducer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(OpenAIBatchResultsProducer.class);
+ private static final String ERROR_FILE = "error";
+
+ public OpenAIBatchResultsProducer(OpenAIEndpoint endpoint) {
+ super(endpoint);
+ }
+
+ @Override
+ public OpenAIEndpoint getEndpoint() {
+ return (OpenAIEndpoint) super.getEndpoint();
+ }
+
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ OpenAIConfiguration config = getEndpoint().getConfiguration();
+ String batchId = exchange.getIn().getHeader(OpenAIConstants.BATCH_ID, String.class);
+ if (ObjectHelper.isEmpty(batchId)) {
+ throw new IllegalArgumentException(
+ "The " + OpenAIConstants.BATCH_ID + " header must hold the id of the batch to read the results of");
+ }
+
+ String resultsFile = exchange.getIn().getHeader(OpenAIConstants.BATCH_RESULTS_FILE,
+ config.getBatchResultsFile(), String.class);
+ if (!"output".equals(resultsFile) && !ERROR_FILE.equals(resultsFile)) {
+ throw new IllegalArgumentException(
+ "Unsupported batch results file: " + resultsFile + ". Supported: output, error");
+ }
+
+ Batch batch;
+ try {
+ batch = getEndpoint().getClient().batches().retrieve(batchId);
+ } catch (RuntimeException e) {
+ GenAiErrorSupport.apply(exchange, e);
+ throw e;
+ }
+ OpenAIBatchSupport.setBatchHeaders(exchange.getMessage(), batch);
+
+ // the API sets the file ids once it has stopped processing, so their presence, not the status, says whether
+ // the results can be read; the status only explains why a file is missing
+ Optional<String> fileId = ERROR_FILE.equals(resultsFile) ? batch.errorFileId() : batch.outputFileId();
+ if (fileId.isEmpty()) {
+ switch (batch.status().value()) {
+ case FAILED ->
+ // the input was rejected, so no result file exists: the reason is in the errors of the batch
+ throw new CamelExchangeException(
+ "Batch " + batchId + " failed: " + OpenAIBatchSupport.errors(batch), exchange);
Review Comment:
Applied in d2a51a238e8b: a failed batch with no `errors` now reports "the API reported no validation errors" instead of `[]`. Worded neutrally since the component also targets compatible servers.
_Claude Code on behalf of Federico Mariani (Croway)_
##########
test-infra/camel-test-infra-openai-mock/src/main/java/org/apache/camel/test/infra/openai/mock/BatchStore.java:
##########
@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.test.infra.openai.mock;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * In-memory state behind the Files and Batch APIs of the mock: the uploaded input files, the result files built for a
+ * finished batch, and the batches themselves. Tests can read the uploaded content back to assert on the JSONL a route
+ * produced.
+ */
+public class BatchStore {
+
+ /**
+ * A file uploaded by a route or generated by the mock for a finished batch.
+ */
+ public record StoredFile(String id, String filename, String purpose, byte[] content) {
+ }
+
+ /**
+ * A created batch and the position it reached in the configured status progression.
+ */
+ public static class StoredBatch {
+ final String id;
+ final String inputFileId;
+ final String endpoint;
+ final String metadata;
+ int statusIndex;
+ boolean cancelling;
+ boolean cancelled;
+ boolean resultsBuilt;
+ String outputFileId;
+ String errorFileId;
+ int total;
+ int completed;
+ int failed;
Review Comment:
The mock serves requests on `Executors.newSingleThreadExecutor()` (`OpenAIMock` line 127), so `StoredBatch` is only ever touched by one thread and there is no race to guard. The `ConcurrentHashMap` covers the test thread reading the store while the server thread writes. Leaving as is.
_Claude Code on behalf of Federico Mariani (Croway)_
##########
test-infra/camel-test-infra-openai-mock/src/main/java/org/apache/camel/test/infra/openai/mock/BatchStore.java:
##########
@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.test.infra.openai.mock;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * In-memory state behind the Files and Batch APIs of the mock: the uploaded input files, the result files built for a
+ * finished batch, and the batches themselves. Tests can read the uploaded content back to assert on the JSONL a route
+ * produced.
+ */
+public class BatchStore {
+
+ /**
+ * A file uploaded by a route or generated by the mock for a finished batch.
+ */
+ public record StoredFile(String id, String filename, String purpose, byte[] content) {
+ }
+
+ /**
+ * A created batch and the position it reached in the configured status progression.
+ */
+ public static class StoredBatch {
+ final String id;
+ final String inputFileId;
+ final String endpoint;
+ final String metadata;
+ int statusIndex;
+ boolean cancelling;
+ boolean cancelled;
+ boolean resultsBuilt;
+ String outputFileId;
+ String errorFileId;
+ int total;
+ int completed;
+ int failed;
+
Review Comment:
The mock serves requests on `Executors.newSingleThreadExecutor()` (`OpenAIMock` line 127), so `StoredBatch` is only ever touched by one thread and there is no race to guard. The `ConcurrentHashMap` covers the test thread reading the store while the server thread writes. Leaving as is.
_Claude Code on behalf of Federico Mariani (Croway)_
##########
test-infra/camel-test-infra-openai-mock/src/main/java/org/apache/camel/test/infra/openai/mock/BatchRequestHandler.java:
##########
@@ -0,0 +1,404 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.test.infra.openai.mock;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.regex.Pattern;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.sun.net.httpserver.HttpExchange;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Serves the Files and Batch APIs: uploading an input file, creating a batch from it, reporting its status and serving
+ * the result files.
+ * <p>
+ * A batch walks the configured status progression, one step per retrieve, so a route that polls sees the same sequence
+ * of states as it would against the real API. A cancel moves a running batch to {@code cancelling}, and the next
+ * retrieve to {@code cancelled}. Once a final status is reached the output and error files are built from the input
+ * file, matching each request line to a {@link BatchExpectation} by its {@code custom_id}.
+ */
+public class BatchRequestHandler {
+ private static final Logger LOG = LoggerFactory.getLogger(BatchRequestHandler.class);
+
+ static final List<String> DEFAULT_STATUSES = List.of("validating", "in_progress", "finalizing", "completed");
+
+ private static final String DEFAULT_RESPONSE_BODY
+ = "{\"id\":\"chatcmpl-mock\",\"object\":\"chat.completion\",\"model\":\"openai-mock\","
+ + "\"choices\":[{\"index\":0,\"message\":{\"role\":\"assistant\",\"content\":\"mock response\"},"
+ + "\"finish_reason\":\"stop\"}]}";
+
+ private final List<BatchExpectation> expectations;
+ private final List<String> statuses;
+ private final BatchStore store;
+ private final ObjectMapper objectMapper;
+
+ public BatchRequestHandler(List<BatchExpectation> expectations, List<String> statuses, BatchStore store,
+ ObjectMapper objectMapper) {
+ this.expectations = expectations;
+ this.statuses = statuses;
+ this.store = store;
+ this.objectMapper = objectMapper;
+ }
+
+ /**
+ * Whether this handler serves the given path, so the dispatcher can route Files and Batch calls here before the
+ * handlers of the inference APIs.
+ */
+ public static boolean handles(String path) {
+ return path.endsWith("/files") || path.contains("/files/")
+ || path.endsWith("/batches") || path.contains("/batches/");
+ }
+
+ public void handleRequest(HttpExchange exchange) throws IOException {
+ String path = exchange.getRequestURI().getPath();
+ String method = exchange.getRequestMethod();
+ try {
+ if (path.endsWith("/files") && "POST".equalsIgnoreCase(method)) {
+ uploadFile(exchange);
+ } else if (path.endsWith("/content") && "GET".equalsIgnoreCase(method)) {
+ fileContent(exchange, idBefore(path, "/content"));
+ } else if (path.contains("/files/") && "GET".equalsIgnoreCase(method)) {
+ fileObject(exchange, lastSegment(path));
+ } else if (path.contains("/files/") && "DELETE".equalsIgnoreCase(method)) {
+ deleteFile(exchange, lastSegment(path));
+ } else if (path.endsWith("/batches") && "POST".equalsIgnoreCase(method)) {
+ createBatch(exchange);
+ } else if (path.endsWith("/cancel") && "POST".equalsIgnoreCase(method)) {
+ cancelBatch(exchange, idBefore(path, "/cancel"));
+ } else if (path.contains("/batches/") && "GET".equalsIgnoreCase(method)) {
+ retrieveBatch(exchange, lastSegment(path));
+ } else {
+ sendError(exchange, 404, "invalid_request_error", "Unsupported request: " + method + " " + path);
+ }
+ } catch (Exception e) {
+ LOG.error("Error processing batch request {} {}", method, path, e);
+ sendError(exchange, 500, "internal_error", String.valueOf(e.getMessage()));
+ }
+ }
+
+ private void uploadFile(HttpExchange exchange) throws IOException {
+ byte[] body;
+ try (InputStream is = exchange.getRequestBody()) {
+ body = is.readAllBytes();
+ }
+ String contentType = exchange.getRequestHeaders().getFirst("Content-Type");
+ String boundary = contentType != null && contentType.contains("boundary=")
+ ? contentType.substring(contentType.indexOf("boundary=") + "boundary=".length()).trim() : null;
+ if (boundary == null) {
+ sendError(exchange, 400, "invalid_request_error", "The file upload must be a multipart request");
+ return;
+ }
+
+ // ISO-8859-1 maps every byte to one character, so the file part survives the split unchanged
+ String raw = new String(body, StandardCharsets.ISO_8859_1);
+ String filename = "unknown";
+ String purpose = "unknown";
+ byte[] content = new byte[0];
+ // the boundary may hold regex metacharacters, so it is quoted rather than spliced into the pattern
+ for (String part : raw.split(Pattern.quote("--" + boundary))) {
+ int headerEnd = part.indexOf("\r\n\r\n");
+ if (headerEnd < 0) {
+ continue;
+ }
+ String headers = part.substring(0, headerEnd);
+ String value = part.substring(headerEnd + 4);
+ if (value.endsWith("\r\n")) {
+ value = value.substring(0, value.length() - 2);
+ }
+ if (headers.contains("name=\"file\"")) {
+ filename = between(headers, "filename=\"", "\"");
+ content = value.getBytes(StandardCharsets.ISO_8859_1);
+ } else if (headers.contains("name=\"purpose\"")) {
+ purpose = value.trim();
+ }
+ }
+
+ String id = store.addFile(filename, purpose, content);
+ LOG.debug("Stored uploaded file {} ({} bytes, purpose {})", id, content.length, purpose);
+ sendJson(exchange, 200, fileNode(id, filename, purpose, content.length).toString());
+ }
+
+ private void fileContent(HttpExchange exchange, String fileId) throws IOException {
+ byte[] content = store.getFileContent(fileId);
+ if (content == null) {
+ sendError(exchange, 404, "invalid_request_error", "No such file: " + fileId);
+ return;
+ }
+ exchange.getResponseHeaders().add("Content-Type", "application/octet-stream");
+ exchange.sendResponseHeaders(200, content.length);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(content);
+ }
+ }
+
+ private void fileObject(HttpExchange exchange, String fileId) throws IOException {
+ BatchStore.StoredFile file = store.file(fileId);
+ if (file == null) {
+ sendError(exchange, 404, "invalid_request_error", "No such file: " + fileId);
+ return;
+ }
+ sendJson(exchange, 200, fileNode(file.id(), file.filename(), file.purpose(), file.content().length).toString());
+ }
+
+ private void deleteFile(HttpExchange exchange, String fileId) throws IOException {
+ if (store.file(fileId) == null) {
+ sendError(exchange, 404, "invalid_request_error", "No such file: " + fileId);
+ return;
+ }
+ store.removeFile(fileId);
+ ObjectNode node = objectMapper.createObjectNode();
+ node.put("id", fileId);
+ node.put("object", "file");
+ node.put("deleted", true);
+ sendJson(exchange, 200, node.toString());
+ }
+
+ private void createBatch(HttpExchange exchange) throws IOException {
+ JsonNode request;
+ try (InputStream is = exchange.getRequestBody()) {
+ request = objectMapper.readTree(is.readAllBytes());
+ }
+ String inputFileId = request.path("input_file_id").asText();
+ if (store.file(inputFileId) == null) {
+ sendError(exchange, 404, "invalid_request_error", "No such file: " + inputFileId);
+ return;
+ }
+ JsonNode metadata = request.get("metadata");
+ BatchStore.StoredBatch batch = store.addBatch(inputFileId, request.path("endpoint").asText(),
+ metadata != null && !metadata.isNull() ? metadata.toString() : null);
+ maybeBuildResults(batch);
+ LOG.debug("Created batch {} for input file {}", batch.id, inputFileId);
+ sendJson(exchange, 200, batchNode(batch).toString());
+ }
+
+ private void retrieveBatch(HttpExchange exchange, String batchId) throws IOException {
+ BatchStore.StoredBatch batch = store.batch(batchId);
+ if (batch == null) {
+ sendError(exchange, 404, "invalid_request_error", "No such batch: " + batchId);
+ return;
+ }
+ if (batch.cancelling) {
+ // a cancelled batch is final, so the retrieve after the cancel reports it as such, as the API does
+ batch.cancelling = false;
+ batch.cancelled = true;
+ } else if (!batch.cancelled && batch.statusIndex < statusList().size() - 1) {
+ batch.statusIndex++;
+ }
+ maybeBuildResults(batch);
+ sendJson(exchange, 200, batchNode(batch).toString());
+ }
+
+ private void cancelBatch(HttpExchange exchange, String batchId) throws IOException {
+ BatchStore.StoredBatch batch = store.batch(batchId);
+ if (batch == null) {
+ sendError(exchange, 404, "invalid_request_error", "No such batch: " + batchId);
+ return;
+ }
+ if (isFinal(status(batch))) {
+ sendError(exchange, 400, "invalid_request_error",
+ "Cannot cancel a batch with status '" + status(batch) + "'.");
+ return;
+ }
+ batch.cancelling = true;
+ sendJson(exchange, 200, batchNode(batch).toString());
+ }
+
+ private List<String> statusList() {
+ return statuses.isEmpty() ? DEFAULT_STATUSES : statuses;
+ }
+
+ private String status(BatchStore.StoredBatch batch) {
+ if (batch.cancelled) {
+ return "cancelled";
+ }
+ if (batch.cancelling) {
+ return "cancelling";
+ }
+ return statusList().get(Math.min(batch.statusIndex, statusList().size() - 1));
+ }
+
+ private boolean isFinal(String status) {
+ return "completed".equals(status) || "expired".equals(status) || "cancelled".equals(status);
+ }
+
+ /**
+ * Builds the output and error files once the batch reaches a final status, as the API does when it finishes
+ * processing the input file.
+ */
+ private void maybeBuildResults(BatchStore.StoredBatch batch) throws IOException {
+ if (batch.resultsBuilt || !isFinal(status(batch))) {
+ return;
+ }
+ batch.resultsBuilt = true;
+
+ BatchStore.StoredFile input = store.file(batch.inputFileId);
+ List<String> outputLines = new ArrayList<>();
+ List<String> errorLines = new ArrayList<>();
+ int line = 0;
+ for (String requestLine : new String(input.content(), StandardCharsets.UTF_8).split("\n")) {
Review Comment:
Applied in d2a51a238e8b: the split is now `\\r?\\n`. The `newLine()` trigger no longer exists (lines are joined with an explicit `\\n` since ddafa304), but a raw JSONL body produced on Windows still can carry CRLF.
_Claude Code on behalf of Federico Mariani (Croway)_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-24674: camel-openai - support the Batch API [camel]
Posted by "davsclaus (via GitHub)" <gi...@apache.org>.
davsclaus commented on code in PR #26601:
URL: https://github.com/apache/camel/pull/26601#discussion_r4048765273
##########
components/camel-ai/camel-openai/src/main/docs/openai-batch.adoc:
##########
@@ -0,0 +1,179 @@
+= OpenAI - Batch API Operations
+:tabs-sync-option:
+
+xref:ROOT:openai-component.adoc[Back to OpenAI Component]
+
+The Batch API runs requests offline at half the price of the synchronous API, against a separate rate limit, and
+returns the answers within 24 hours as a file. It suits work nobody waits for: overnight enrichment, embedding
+backfills, moderation backlogs and prompt regression runs.
+
+Four operations cover the flow:
+
+* `batch` - uploads the requests as a JSONL file and creates the batch
+* `batch-retrieve` - reports the status and request counts of a batch
+* `batch-cancel` - cancels a batch that is still running
+* `batch-results` - downloads the output file, or the error file
+
+== Building the input
+
+The body of the `batch` operation is either the JSONL itself (`File`, `Path`, `WrappedFile`, `InputStream`,
+`byte[]` or String), or a `Map` keyed by `custom_id`, the id that ties each answer back to its request.
+
+For a map, the component writes the envelope of every line (`method` and the `url` of `batchEndpoint`). What the
+value holds decides the request body:
+
+* a String is turned into a request built from the endpoint options, for `/v1/chat/completions`, `/v1/responses`,
+ `/v1/embeddings` and `/v1/moderations`. The request is the one the synchronous operation of that endpoint would
+ send, so the same options and headers apply: `model`, `systemMessage`, `developerMessage`, `temperature`, `topP`,
+ `maxTokens`, `outputClass`, `jsonSchema` and `additionalBodyProperty` for chat completions and responses,
+ `embeddingModel` and `dimensions` for embeddings, and `moderationModel` for moderations
+* a `Map` or `JsonNode` is used as the request body as it is, which is what the other endpoints need, and what
+ lets each line use different parameters
+
+The `CamelOpenAIBatchEndpoint` header overrides `batchEndpoint` for one message. A file body is uploaded under its
+own name with `.jsonl` appended when it has another extension, the only one the Files API accepts for a batch input.
+
+[tabs]
+====
+Java::
++
+[source,java]
+----
+// prompts: the request of every line comes from the options
+from("direct:classify")
+ .setBody(constant(Map.of("ticket-1", "I was charged twice", "ticket-2", "The app crashes")))
+ .to("openai:batch?batchEndpoint=/v1/chat/completions&model=gpt-4o-mini"
+ + "&systemMessage=Classify the ticket&batchMetadata.job=triage")
+ .log("Batch ${header.CamelOpenAIBatchId} is ${header.CamelOpenAIBatchStatus}");
+
+// full request bodies: a different model per line
+from("direct:mixed")
+ .setBody(constant(Map.of("a", Map.of("model", "gpt-4o-mini", "messages",
+ List.of(Map.of("role", "user", "content", "Summarise"))))))
+ .to("openai:batch?batchEndpoint=/v1/chat/completions");
+
+// a JSONL file produced elsewhere is uploaded unchanged
+from("file:batches/inbox?include=.*\\.jsonl")
+ .to("openai:batch?batchEndpoint=/v1/chat/completions");
+----
+
+YAML::
++
+[source,yaml]
+----
+- from:
+ uri: direct:classify
+ steps:
+ - to: "openai:batch?batchEndpoint=/v1/chat/completions&model=gpt-4o-mini&systemMessage=Classify the ticket"
+ - log: "Batch ${header.CamelOpenAIBatchId} is ${header.CamelOpenAIBatchStatus}"
+----
+====
+
+== Waiting for the batch
+
+`batch-retrieve` reads `CamelOpenAIBatchId` and reports the batch on the headers, leaving the body untouched. A
+short job can poll in place:
+
+[source,java]
+----
+from("direct:wait")
+ .loopDoWhile(simple("${header.CamelOpenAIBatchStatus} !in 'completed,failed,expired,cancelled'"))
+ .delay(60000)
+ .to("openai:batch-retrieve")
+ .end()
+ .to("direct:collect");
+----
+
+A job that must survive a restart should store `CamelOpenAIBatchId` instead, and poll it from a scheduler route:
+
+[source,java]
+----
+from("scheduler:batches?delay=600000")
+ .to("sql:select id from openai_batch where status not in ('processed', 'failed')")
+ .split(body())
+ .setHeader("CamelOpenAIBatchId", simple("${body[id]}"))
+ .to("openai:batch-retrieve")
+ .to("sql:update openai_batch set status = :#CamelOpenAIBatchStatus where id = :#CamelOpenAIBatchId")
+ .filter(header("CamelOpenAIBatchStatus").in("completed", "expired", "cancelled"))
+ .to("direct:collect")
+ .to("sql:update openai_batch set status = 'processed' where id = :#CamelOpenAIBatchId")
+ .end()
+ .end();
+----
+
+The row is marked `processed` only after `direct:collect` has read the results, so a failure while reading leaves
+it at `completed` and the next run retries it.
+
+== Reading the results
+
+`batch-results` reads `CamelOpenAIBatchId`, downloads the output file and sets it as an `InputStream`, so it can be
+split line by line. Set `batchResultsFile=error` for the failed requests instead. Each line carries the `custom_id`
+of its request.
+
+The polling routes above hand a finished batch to `direct:collect`, which reads the file. With the defaults, stream
+caching reads the file into memory when the next step runs, and the splitter works on that copy. For a batch of a
+few thousand requests this is fine, and the route needs nothing special:
+
+[source,java]
+----
+from("direct:collect")
+ .to("openai:batch-results")
+ .split(body().tokenize("\n"))
+ .setHeader("ticketId", jsonpath("$.custom_id"))
+ .setBody(jsonpath("$.response.body.choices[0].message.content"))
+ .to("sql:update ticket set category = :#${body} where id = :#${header.ticketId}")
+ .end();
+----
+
+A result file can reach hundreds of megabytes, and holding it in memory is then the problem. Turning stream
+caching off on the route keeps the body as the response stream, and a splitter in streaming mode reads it one line
+at a time as each line is processed. The body can then be read only once, which is what the splitter does:
+
+[source,java]
+----
+from("direct:collect")
+ .noStreamCaching() // keep the response as a stream instead of caching it in memory
+ .to("openai:batch-results")
+ .split(body().tokenize("\n")).streaming() // read one line at a time
+ .setHeader("ticketId", jsonpath("$.custom_id"))
+ .setBody(jsonpath("$.response.body.choices[0].message.content"))
+ .to("sql:update ticket set category = :#${body} where id = :#${header.ticketId}")
+ .end();
+----
+
+An embedding backfill works the same way, straight into a vector store. Embedding vectors make for large result
+files, so this one uses the streaming form. The batch is created by the first route, and a polling route like the
+ones above hands it to `direct:store` once it is finished:
+
+[source,java]
+----
+from("direct:embeddings")
+ .setBody(constant(Map.of("doc-1", "Camel routes messages", "doc-2", "Batches run offline")))
+ .to("openai:batch?batchEndpoint=/v1/embeddings&embeddingModel=text-embedding-3-small");
+
+from("direct:store")
+ .noStreamCaching()
+ .to("openai:batch-results")
+ .split(body().tokenize("\n")).streaming()
+ .setHeader("CamelPgVectorAction", constant("UPSERT"))
+ .setHeader("CamelPgVectorRecordId", jsonpath("$.custom_id"))
+ .setBody(jsonpath("$.response.body.data[0].embedding"))
Review Comment:
This example will fail at runtime. camel-jsonpath uses `JacksonJsonProvider`, so the embedding array arrives as a `List<Double>`, but `PgVectorProducer.upsert()` reads the body as `List<Float>` and `toFloatArray()` does `Float value = vector.get(i)` β that checkcast throws `ClassCastException: Double cannot be cast to Float` on the first element.
The sync `openai:embeddings` producer emits `List<Float>`, which is why the existing openai β pgvector example in `openai-operations.adoc` works and this one will not. Either add a conversion step before `pgvector:` (a small processor mapping the list to `Float`), or use a sink that accepts the JSON numbers as they are. Since this is the example users will copy for the embedding backfill use case, it is worth getting right in this PR.
_Claude Code on behalf of davsclaus_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-24674: camel-openai - support the Batch API [camel]
Posted by "gnodet-bot (via GitHub)" <gi...@apache.org>.
gnodet-bot commented on code in PR #26601:
URL: https://github.com/apache/camel/pull/26601#discussion_r4053381828
##########
components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIBatchAggregationStrategy.java:
##########
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.openai;
+
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.camel.AggregationStrategy;
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.util.IOHelper;
+import org.apache.camel.util.ObjectHelper;
+
+/**
+ * Collects the messages of an aggregation into the input of a batch, so a stream of prompts becomes one
+ * {@code openai:batch} call:
+ *
+ * <pre>
+ * from("kafka:tickets")
+ * .aggregate(constant(true), new OpenAIBatchAggregationStrategy())
+ * .completionSize(1000).completionTimeout(60000)
+ * .to("openai:batch?batchEndpoint=/v1/chat/completions&model=gpt-4o-mini");
+ * </pre>
+ *
+ * The body of each message is the value of its request, as for a {@link Map} body of the {@code batch} operation: a
+ * String is turned into a request from the endpoint options, a {@code Map} or {@code JsonNode} is the request body. Its
+ * {@code custom_id} is the {@code CamelOpenAIBatchCustomId} header when set, and the message id otherwise, so the
+ * result lines can be matched back to what produced them.
+ * <p>
+ * By default the requests are collected in a {@link Map} in memory, which suits batches of a few thousand short
+ * requests. {@link #spooled()} writes them to a file instead, one line per message, so a batch of any size is
+ * aggregated with constant memory, and a persistent {@code AggregationRepository} stores a path rather than the
+ * requests. The {@code batch} operation streams either form and removes the file once the batch is created.
+ */
+public class OpenAIBatchAggregationStrategy implements AggregationStrategy {
+
+ private final Path spoolDirectory;
+ // the writer of each spool being aggregated, closed when its aggregation completes
+ private final Map<File, BufferedWriter> writers = new ConcurrentHashMap<>();
+
+ /**
+ * Collects the requests in memory.
+ */
+ public OpenAIBatchAggregationStrategy() {
+ this(null);
+ }
+
+ /**
+ * Collects the requests in a file created in the given directory.
+ */
+ public OpenAIBatchAggregationStrategy(Path spoolDirectory) {
+ this.spoolDirectory = spoolDirectory;
+ }
+
+ /**
+ * Collects the requests in a file created in the temporary directory of the JVM.
+ */
+ public static OpenAIBatchAggregationStrategy spooled() {
+ return new OpenAIBatchAggregationStrategy(Path.of(System.getProperty("java.io.tmpdir")));
+ }
+
+ public boolean isSpooled() {
+ return spoolDirectory != null;
+ }
+
+ @Override
+ public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
+ Message message = newExchange.getMessage();
+ String customId = message.getHeader(OpenAIConstants.BATCH_CUSTOM_ID, message.getMessageId(), String.class);
+ Object value = message.getBody();
+ if (ObjectHelper.isEmpty(customId) || value == null) {
+ throw new IllegalArgumentException(
+ "A message aggregated into a batch needs a custom_id and a body; got custom_id " + customId
+ + " and body " + value);
+ }
+
+ try {
+ if (oldExchange == null) {
+ if (isSpooled()) {
+ OpenAIBatchSpool spool = OpenAIBatchSpool.create(spoolDirectory);
+ OpenAIBatchSpool.append(writer(spool), customId, value);
+ message.setBody(spool);
+ } else {
+ Map<String, Object> requests = new LinkedHashMap<>();
+ requests.put(customId, value);
+ message.setBody(requests);
+ }
+ return newExchange;
+ }
+
+ Object aggregated = oldExchange.getMessage().getBody();
+ if (aggregated instanceof OpenAIBatchSpool spool) {
+ OpenAIBatchSpool.append(writer(spool), customId, value);
+ } else if (aggregated instanceof Map<?, ?> requests) {
+ @SuppressWarnings("unchecked")
+ Map<String, Object> map = (Map<String, Object>) requests;
+ map.put(customId, value);
+ } else {
+ throw new IllegalStateException("Unexpected aggregated batch body: " + aggregated);
+ }
+ return oldExchange;
+ } catch (IOException e) {
+ throw new RuntimeCamelException("Cannot spool the batch request of custom_id " + customId, e);
+ }
+ }
+
+ @Override
+ public void onCompletion(Exchange exchange) {
+ if (exchange != null && exchange.getMessage().getBody() instanceof OpenAIBatchSpool spool) {
+ BufferedWriter writer = writers.remove(spool.getFile());
+ if (writer != null) {
+ IOHelper.close(writer);
+ }
+ }
+ }
+
+ private BufferedWriter writer(OpenAIBatchSpool spool) throws IOException {
+ BufferedWriter writer = writers.get(spool.getFile());
+ if (writer == null) {
+ writer = spool.open();
+ writers.put(spool.getFile(), writer);
+ }
+ return writer;
Review Comment:
π ** has a non-atomic check-then-act on **
The + conditional is not atomic: two concurrent calls for the same spool could both see , open two writers for the same file, and one would be leaked (neither nor a second will close the overwritten writer).
In practice, Camel's aggregator serializes calls per group key, so this is not a live bug β but the implies concurrent access is expected. Use to make the intent explicit and close the race:
_generated by an AI agent, Hermès_ <!-- reviewer: gnodet-bot -->
##########
components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIBatchProducer.java:
##########
@@ -0,0 +1,467 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.openai;
+
+import java.io.ByteArrayInputStream;
+import java.io.Closeable;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.SequenceInputStream;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.util.Enumeration;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.openai.core.JsonValue;
+import com.openai.core.MultipartField;
+import com.openai.core.ObjectMappers;
+import com.openai.models.ResponseFormatJsonSchema;
+import com.openai.models.batches.Batch;
+import com.openai.models.batches.BatchCreateParams;
+import com.openai.models.chat.completions.ChatCompletionCreateParams;
+import com.openai.models.embeddings.EmbeddingCreateParams;
+import com.openai.models.files.FileCreateParams;
+import com.openai.models.files.FilePurpose;
+import com.openai.models.moderations.ModerationCreateParams;
+import com.openai.models.responses.ResponseCreateParams;
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.WrappedFile;
+import org.apache.camel.component.ai.observability.GenAiErrorSupport;
+import org.apache.camel.component.openai.OpenAIResponsesInputBuilder.InputSpec;
+import org.apache.camel.support.DefaultProducer;
+import org.apache.camel.util.ObjectHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * OpenAI producer for the {@code batch} operation, which uploads the requests of a batch as a JSONL file and creates
+ * the batch that runs them.
+ * <p>
+ * The body is either the JSONL itself, as a {@link File}, {@link Path}, {@link WrappedFile}, {@link InputStream},
+ * {@code byte[]} or String, or a {@link Map} keyed by {@code custom_id}. A map value that is itself a map is used as
+ * the request body as it is, while a String value is turned into a request built from the endpoint options, so a route
+ * that only has prompts does not have to assemble the API payload itself. An {@link Iterable} body, such as the
+ * {@link OpenAIBatchSpool} of an aggregation, is read the same way one item at a time. Such a request is built with the
+ * same SDK parameters as the synchronous operation of the endpoint, so the options and headers of that operation apply.
+ */
+public class OpenAIBatchProducer extends DefaultProducer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(OpenAIBatchProducer.class);
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+ private static final String INPUT_FILE_SUFFIX = ".jsonl";
+ private static final String INPUT_FILE_NAME = "camel-openai-batch" + INPUT_FILE_SUFFIX;
+
+ private Class<?> outputClassResolved;
+
+ public OpenAIBatchProducer(OpenAIEndpoint endpoint) {
+ super(endpoint);
+ }
+
+ @Override
+ public OpenAIEndpoint getEndpoint() {
+ return (OpenAIEndpoint) super.getEndpoint();
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+ String outputClass = getEndpoint().getConfiguration().getOutputClass();
+ if (ObjectHelper.isNotEmpty(outputClass)) {
+ outputClassResolved = getEndpoint().getCamelContext().getClassResolver().resolveMandatoryClass(outputClass);
+ }
+ }
+
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ OpenAIConfiguration config = getEndpoint().getConfiguration();
+ Message in = exchange.getIn();
+ // the options a batch cannot honour are refused when the endpoint starts; the header can still ask for them
+ if (Boolean.TRUE.equals(in.getHeader(OpenAIConstants.STREAMING, Boolean.class))) {
+ throw new IllegalArgumentException(
+ "The batch operation cannot stream responses; remove the " + OpenAIConstants.STREAMING + " header");
+ }
+
+ String endpoint = OpenAIBatchSupport.resolveEndpoint(in, config);
+
+ Object body = in.getBody();
+ String inputFileId;
+ if (body instanceof Map<?, ?> map) {
+ inputFileId = upload(exchange, requestLines(map.entrySet(), endpoint, in, config), INPUT_FILE_NAME);
+ } else if (body instanceof Iterable<?> items) {
+ inputFileId = upload(exchange, requestLines(items, endpoint, in, config), INPUT_FILE_NAME);
+ } else {
+ InputStream stream = in.getBody(InputStream.class);
+ if (stream == null) {
+ throw new IllegalArgumentException(
+ "Unsupported body type for the batch operation: "
+ + (body != null ? body.getClass().getName() : "null")
+ + ". Supported: File, Path, InputStream, byte[], String, a Map "
+ + "keyed by custom_id, or an Iterable");
+ }
+ inputFileId = upload(exchange, stream, inputFileName(body));
+ }
+ in.setHeader(OpenAIConstants.BATCH_INPUT_FILE_ID, inputFileId);
+
+ Batch batch = create(exchange, inputFileId, endpoint, in, config);
+ if (body instanceof OpenAIBatchSpool spool) {
+ // the spool of an aggregation is consumed by this batch, so it is not left behind in the directory
+ spool.delete();
Review Comment:
β οΈ ** failure orphans a successfully-created batch (still outstanding from prior review)**
returned a valid β the batch is running on OpenAI's servers. If β throws (read-only filesystem, network-mounted temp dir, concurrent deletion, etc.) the propagates out of as an exchange failure. at line 134 is never reached: the caller receives no , the batch runs untracked, and there is no way to retrieve its results or cancel it.
The fix is straightforward β catch the , log a warning, and continue:
_generated by an AI agent, Hermès_ <!-- reviewer: gnodet-bot -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-24674: camel-openai - support the Batch API [camel]
Posted by "Croway (via GitHub)" <gi...@apache.org>.
Croway commented on PR #26601:
URL: https://github.com/apache/camel/pull/26601#issuecomment-5732206649
Thanks for the review. Two of the three findings are applied in d2a51a238e8b; the data-race one does not apply because the mock server runs single-threaded (reply inline). Note the temp-file handling praised in the summary was removed in ddafa304: a stream part is not repeatable in the SDK, so the file never enabled a retry, and redelivery is left to Camel's error handling.
_Claude Code on behalf of Federico Mariani (Croway)_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-24674: camel-openai - support the Batch API [camel]
Posted by "gnodet-bot (via GitHub)" <gi...@apache.org>.
gnodet-bot commented on code in PR #26601:
URL: https://github.com/apache/camel/pull/26601#discussion_r4053384946
##########
components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIBatchProducer.java:
##########
@@ -0,0 +1,467 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.openai;
+
+import java.io.ByteArrayInputStream;
+import java.io.Closeable;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.SequenceInputStream;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.util.Enumeration;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.openai.core.JsonValue;
+import com.openai.core.MultipartField;
+import com.openai.core.ObjectMappers;
+import com.openai.models.ResponseFormatJsonSchema;
+import com.openai.models.batches.Batch;
+import com.openai.models.batches.BatchCreateParams;
+import com.openai.models.chat.completions.ChatCompletionCreateParams;
+import com.openai.models.embeddings.EmbeddingCreateParams;
+import com.openai.models.files.FileCreateParams;
+import com.openai.models.files.FilePurpose;
+import com.openai.models.moderations.ModerationCreateParams;
+import com.openai.models.responses.ResponseCreateParams;
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.WrappedFile;
+import org.apache.camel.component.ai.observability.GenAiErrorSupport;
+import org.apache.camel.component.openai.OpenAIResponsesInputBuilder.InputSpec;
+import org.apache.camel.support.DefaultProducer;
+import org.apache.camel.util.ObjectHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * OpenAI producer for the {@code batch} operation, which uploads the requests of a batch as a JSONL file and creates
+ * the batch that runs them.
+ * <p>
+ * The body is either the JSONL itself, as a {@link File}, {@link Path}, {@link WrappedFile}, {@link InputStream},
+ * {@code byte[]} or String, or a {@link Map} keyed by {@code custom_id}. A map value that is itself a map is used as
+ * the request body as it is, while a String value is turned into a request built from the endpoint options, so a route
+ * that only has prompts does not have to assemble the API payload itself. An {@link Iterable} body, such as the
+ * {@link OpenAIBatchSpool} of an aggregation, is read the same way one item at a time. Such a request is built with the
+ * same SDK parameters as the synchronous operation of the endpoint, so the options and headers of that operation apply.
+ */
+public class OpenAIBatchProducer extends DefaultProducer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(OpenAIBatchProducer.class);
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+ private static final String INPUT_FILE_SUFFIX = ".jsonl";
+ private static final String INPUT_FILE_NAME = "camel-openai-batch" + INPUT_FILE_SUFFIX;
+
+ private Class<?> outputClassResolved;
+
+ public OpenAIBatchProducer(OpenAIEndpoint endpoint) {
+ super(endpoint);
+ }
+
+ @Override
+ public OpenAIEndpoint getEndpoint() {
+ return (OpenAIEndpoint) super.getEndpoint();
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+ String outputClass = getEndpoint().getConfiguration().getOutputClass();
+ if (ObjectHelper.isNotEmpty(outputClass)) {
+ outputClassResolved = getEndpoint().getCamelContext().getClassResolver().resolveMandatoryClass(outputClass);
+ }
+ }
+
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ OpenAIConfiguration config = getEndpoint().getConfiguration();
+ Message in = exchange.getIn();
+ // the options a batch cannot honour are refused when the endpoint starts; the header can still ask for them
+ if (Boolean.TRUE.equals(in.getHeader(OpenAIConstants.STREAMING, Boolean.class))) {
+ throw new IllegalArgumentException(
+ "The batch operation cannot stream responses; remove the " + OpenAIConstants.STREAMING + " header");
+ }
+
+ String endpoint = OpenAIBatchSupport.resolveEndpoint(in, config);
+
+ Object body = in.getBody();
+ String inputFileId;
+ if (body instanceof Map<?, ?> map) {
+ inputFileId = upload(exchange, requestLines(map.entrySet(), endpoint, in, config), INPUT_FILE_NAME);
+ } else if (body instanceof Iterable<?> items) {
+ inputFileId = upload(exchange, requestLines(items, endpoint, in, config), INPUT_FILE_NAME);
+ } else {
+ InputStream stream = in.getBody(InputStream.class);
+ if (stream == null) {
+ throw new IllegalArgumentException(
+ "Unsupported body type for the batch operation: "
+ + (body != null ? body.getClass().getName() : "null")
+ + ". Supported: File, Path, InputStream, byte[], String, a Map "
+ + "keyed by custom_id, or an Iterable");
+ }
+ inputFileId = upload(exchange, stream, inputFileName(body));
+ }
+ in.setHeader(OpenAIConstants.BATCH_INPUT_FILE_ID, inputFileId);
+
+ Batch batch = create(exchange, inputFileId, endpoint, in, config);
+ if (body instanceof OpenAIBatchSpool spool) {
+ // the spool of an aggregation is consumed by this batch, so it is not left behind in the directory
+ spool.delete();
Review Comment:
β οΈ **`spool.delete()` failure orphans a successfully-created batch (still outstanding from prior review)**
`create()` returned a valid `Batch` β the batch is running on OpenAI's servers. If `spool.delete()` β `Files.deleteIfExists()` throws (read-only filesystem, network-mounted temp dir, concurrent deletion, etc.) the `IOException` propagates out of `process()` as an exchange failure. `setBatchHeaders()` at line 134 is never reached: the caller receives no `CamelOpenAIBatchId`, the batch runs untracked, and there is no way to retrieve its results or cancel it.
The fix is straightforward β catch the `IOException`, log a warning, and continue:
```java
if (body instanceof OpenAIBatchSpool spool) {
try {
spool.delete();
} catch (IOException e) {
LOG.warn("Could not delete batch spool file {}; the batch {} is running normally",
spool.getFile(), batch.id(), e);
}
}
```
_generated by an AI agent, Hermès_ <!-- reviewer: gnodet-bot -->
##########
components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIBatchAggregationStrategy.java:
##########
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.openai;
+
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.camel.AggregationStrategy;
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.util.IOHelper;
+import org.apache.camel.util.ObjectHelper;
+
+/**
+ * Collects the messages of an aggregation into the input of a batch, so a stream of prompts becomes one
+ * {@code openai:batch} call:
+ *
+ * <pre>
+ * from("kafka:tickets")
+ * .aggregate(constant(true), new OpenAIBatchAggregationStrategy())
+ * .completionSize(1000).completionTimeout(60000)
+ * .to("openai:batch?batchEndpoint=/v1/chat/completions&model=gpt-4o-mini");
+ * </pre>
+ *
+ * The body of each message is the value of its request, as for a {@link Map} body of the {@code batch} operation: a
+ * String is turned into a request from the endpoint options, a {@code Map} or {@code JsonNode} is the request body. Its
+ * {@code custom_id} is the {@code CamelOpenAIBatchCustomId} header when set, and the message id otherwise, so the
+ * result lines can be matched back to what produced them.
+ * <p>
+ * By default the requests are collected in a {@link Map} in memory, which suits batches of a few thousand short
+ * requests. {@link #spooled()} writes them to a file instead, one line per message, so a batch of any size is
+ * aggregated with constant memory, and a persistent {@code AggregationRepository} stores a path rather than the
+ * requests. The {@code batch} operation streams either form and removes the file once the batch is created.
+ */
+public class OpenAIBatchAggregationStrategy implements AggregationStrategy {
+
+ private final Path spoolDirectory;
+ // the writer of each spool being aggregated, closed when its aggregation completes
+ private final Map<File, BufferedWriter> writers = new ConcurrentHashMap<>();
+
+ /**
+ * Collects the requests in memory.
+ */
+ public OpenAIBatchAggregationStrategy() {
+ this(null);
+ }
+
+ /**
+ * Collects the requests in a file created in the given directory.
+ */
+ public OpenAIBatchAggregationStrategy(Path spoolDirectory) {
+ this.spoolDirectory = spoolDirectory;
+ }
+
+ /**
+ * Collects the requests in a file created in the temporary directory of the JVM.
+ */
+ public static OpenAIBatchAggregationStrategy spooled() {
+ return new OpenAIBatchAggregationStrategy(Path.of(System.getProperty("java.io.tmpdir")));
+ }
+
+ public boolean isSpooled() {
+ return spoolDirectory != null;
+ }
+
+ @Override
+ public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
+ Message message = newExchange.getMessage();
+ String customId = message.getHeader(OpenAIConstants.BATCH_CUSTOM_ID, message.getMessageId(), String.class);
+ Object value = message.getBody();
+ if (ObjectHelper.isEmpty(customId) || value == null) {
+ throw new IllegalArgumentException(
+ "A message aggregated into a batch needs a custom_id and a body; got custom_id " + customId
+ + " and body " + value);
+ }
+
+ try {
+ if (oldExchange == null) {
+ if (isSpooled()) {
+ OpenAIBatchSpool spool = OpenAIBatchSpool.create(spoolDirectory);
+ OpenAIBatchSpool.append(writer(spool), customId, value);
+ message.setBody(spool);
+ } else {
+ Map<String, Object> requests = new LinkedHashMap<>();
+ requests.put(customId, value);
+ message.setBody(requests);
+ }
+ return newExchange;
+ }
+
+ Object aggregated = oldExchange.getMessage().getBody();
+ if (aggregated instanceof OpenAIBatchSpool spool) {
+ OpenAIBatchSpool.append(writer(spool), customId, value);
+ } else if (aggregated instanceof Map<?, ?> requests) {
+ @SuppressWarnings("unchecked")
+ Map<String, Object> map = (Map<String, Object>) requests;
+ map.put(customId, value);
+ } else {
+ throw new IllegalStateException("Unexpected aggregated batch body: " + aggregated);
+ }
+ return oldExchange;
+ } catch (IOException e) {
+ throw new RuntimeCamelException("Cannot spool the batch request of custom_id " + customId, e);
+ }
+ }
+
+ @Override
+ public void onCompletion(Exchange exchange) {
+ if (exchange != null && exchange.getMessage().getBody() instanceof OpenAIBatchSpool spool) {
+ BufferedWriter writer = writers.remove(spool.getFile());
+ if (writer != null) {
+ IOHelper.close(writer);
+ }
+ }
+ }
+
+ private BufferedWriter writer(OpenAIBatchSpool spool) throws IOException {
+ BufferedWriter writer = writers.get(spool.getFile());
+ if (writer == null) {
+ writer = spool.open();
+ writers.put(spool.getFile(), writer);
+ }
+ return writer;
Review Comment:
π **`writer()` has a non-atomic check-then-act on `ConcurrentHashMap`**
```java
private BufferedWriter writer(OpenAIBatchSpool spool) throws IOException {
BufferedWriter writer = writers.get(spool.getFile());
if (writer == null) {
writer = spool.open();
writers.put(spool.getFile(), writer);
}
return writer;
}
```
The `get` + conditional `put` is not atomic: two concurrent `aggregate()` calls for the same spool could both see `null`, open two writers for the same file, and one would be leaked (neither `onCompletion()` nor a second `writers.remove()` will close the overwritten writer).
In practice, Camel's aggregator serializes calls per group key, so this is not a live bug β but the `ConcurrentHashMap` implies concurrent access is expected. Use `computeIfAbsent()` to make the intent explicit and close the race:
```java
private BufferedWriter writer(OpenAIBatchSpool spool) throws IOException {
try {
return writers.computeIfAbsent(spool.getFile(), f -> {
try {
return spool.open();
} catch (IOException e) {
throw new UncheckedIOException(e);
}
});
} catch (UncheckedIOException e) {
throw e.getCause();
}
}
```
_generated by an AI agent, Hermès_ <!-- reviewer: gnodet-bot -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-24674: camel-openai - support the Batch API [camel]
Posted by "gnodet-bot (via GitHub)" <gi...@apache.org>.
gnodet-bot commented on code in PR #26601:
URL: https://github.com/apache/camel/pull/26601#discussion_r4053342663
##########
components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIBatchProducer.java:
##########
@@ -0,0 +1,467 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.openai;
+
+import java.io.ByteArrayInputStream;
+import java.io.Closeable;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.SequenceInputStream;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.util.Enumeration;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.openai.core.JsonValue;
+import com.openai.core.MultipartField;
+import com.openai.core.ObjectMappers;
+import com.openai.models.ResponseFormatJsonSchema;
+import com.openai.models.batches.Batch;
+import com.openai.models.batches.BatchCreateParams;
+import com.openai.models.chat.completions.ChatCompletionCreateParams;
+import com.openai.models.embeddings.EmbeddingCreateParams;
+import com.openai.models.files.FileCreateParams;
+import com.openai.models.files.FilePurpose;
+import com.openai.models.moderations.ModerationCreateParams;
+import com.openai.models.responses.ResponseCreateParams;
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.WrappedFile;
+import org.apache.camel.component.ai.observability.GenAiErrorSupport;
+import org.apache.camel.component.openai.OpenAIResponsesInputBuilder.InputSpec;
+import org.apache.camel.support.DefaultProducer;
+import org.apache.camel.util.ObjectHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * OpenAI producer for the {@code batch} operation, which uploads the requests of a batch as a JSONL file and creates
+ * the batch that runs them.
+ * <p>
+ * The body is either the JSONL itself, as a {@link File}, {@link Path}, {@link WrappedFile}, {@link InputStream},
+ * {@code byte[]} or String, or a {@link Map} keyed by {@code custom_id}. A map value that is itself a map is used as
+ * the request body as it is, while a String value is turned into a request built from the endpoint options, so a route
+ * that only has prompts does not have to assemble the API payload itself. An {@link Iterable} body, such as the
+ * {@link OpenAIBatchSpool} of an aggregation, is read the same way one item at a time. Such a request is built with the
+ * same SDK parameters as the synchronous operation of the endpoint, so the options and headers of that operation apply.
+ */
+public class OpenAIBatchProducer extends DefaultProducer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(OpenAIBatchProducer.class);
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+ private static final String INPUT_FILE_SUFFIX = ".jsonl";
+ private static final String INPUT_FILE_NAME = "camel-openai-batch" + INPUT_FILE_SUFFIX;
+
+ private Class<?> outputClassResolved;
+
+ public OpenAIBatchProducer(OpenAIEndpoint endpoint) {
+ super(endpoint);
+ }
+
+ @Override
+ public OpenAIEndpoint getEndpoint() {
+ return (OpenAIEndpoint) super.getEndpoint();
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+ String outputClass = getEndpoint().getConfiguration().getOutputClass();
+ if (ObjectHelper.isNotEmpty(outputClass)) {
+ outputClassResolved = getEndpoint().getCamelContext().getClassResolver().resolveMandatoryClass(outputClass);
+ }
+ }
+
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ OpenAIConfiguration config = getEndpoint().getConfiguration();
+ Message in = exchange.getIn();
+ // the options a batch cannot honour are refused when the endpoint starts; the header can still ask for them
+ if (Boolean.TRUE.equals(in.getHeader(OpenAIConstants.STREAMING, Boolean.class))) {
+ throw new IllegalArgumentException(
+ "The batch operation cannot stream responses; remove the " + OpenAIConstants.STREAMING + " header");
+ }
+
+ String endpoint = OpenAIBatchSupport.resolveEndpoint(in, config);
+
+ Object body = in.getBody();
+ String inputFileId;
+ if (body instanceof Map<?, ?> map) {
+ inputFileId = upload(exchange, requestLines(map.entrySet(), endpoint, in, config), INPUT_FILE_NAME);
+ } else if (body instanceof Iterable<?> items) {
+ inputFileId = upload(exchange, requestLines(items, endpoint, in, config), INPUT_FILE_NAME);
+ } else {
+ InputStream stream = in.getBody(InputStream.class);
+ if (stream == null) {
+ throw new IllegalArgumentException(
+ "Unsupported body type for the batch operation: "
+ + (body != null ? body.getClass().getName() : "null")
+ + ". Supported: File, Path, InputStream, byte[], String, a Map "
+ + "keyed by custom_id, or an Iterable");
+ }
+ inputFileId = upload(exchange, stream, inputFileName(body));
+ }
+ in.setHeader(OpenAIConstants.BATCH_INPUT_FILE_ID, inputFileId);
+
+ Batch batch = create(exchange, inputFileId, endpoint, in, config);
+ if (body instanceof OpenAIBatchSpool spool) {
+ // the spool of an aggregation is consumed by this batch, so it is not left behind in the directory
+ spool.delete();
Review Comment:
β οΈ **`spool.delete()` failure orphans a successfully-created batch (raised on `f3d018e6`, not yet addressed)**
`create()` has returned a valid `Batch` at this point. If `spool.delete()` throws an `IOException` β read-only filesystem, network-mounted temp dir, etc. β the exception propagates out of `process()` as an exchange failure. `setBatchHeaders()` below is never reached: the caller gets no `CamelOpenAIBatchId`, the batch runs untracked on OpenAI's servers, and there is no way to retrieve its results or cancel it.
Spool deletion is cleanup. It must never mask a successful batch creation.
```suggestion
if (body instanceof OpenAIBatchSpool spool) {
// the spool of an aggregation is consumed by this batch, so it is not left behind in the directory
try {
spool.delete();
} catch (IOException e) {
LOG.warn("Could not delete the batch spool file; the batch {} was created successfully", batch.id(), e);
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-24674: camel-openai - support the Batch API [camel]
Posted by "gnodet-bot (via GitHub)" <gi...@apache.org>.
gnodet-bot commented on code in PR #26601:
URL: https://github.com/apache/camel/pull/26601#discussion_r4048045884
##########
test-infra/camel-test-infra-openai-mock/src/main/java/org/apache/camel/test/infra/openai/mock/BatchStore.java:
##########
@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.test.infra.openai.mock;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * In-memory state behind the Files and Batch APIs of the mock: the uploaded input files, the result files built for a
+ * finished batch, and the batches themselves. Tests can read the uploaded content back to assert on the JSONL a route
+ * produced.
+ */
+public class BatchStore {
+
+ /**
+ * A file uploaded by a route or generated by the mock for a finished batch.
+ */
+ public record StoredFile(String id, String filename, String purpose, byte[] content) {
+ }
+
+ /**
+ * A created batch and the position it reached in the configured status progression.
+ */
+ public static class StoredBatch {
+ final String id;
+ final String inputFileId;
+ final String endpoint;
+ final String metadata;
+ int statusIndex;
+ boolean cancelling;
+ boolean cancelled;
+ boolean resultsBuilt;
Review Comment:
β οΈ **Data race on `StoredBatch` mutable fields**
`StoredBatch` fields (`statusIndex`, `cancelling`, `cancelled`, `resultsBuilt`, `outputFileId`, `errorFileId`, etc.) are plain, non-`volatile` instance fields. `BatchStore` correctly uses a `ConcurrentHashMap` for the batch registry, but that only guards the lookup β once you have the `StoredBatch` reference, concurrent requests from multiple threads can race on its fields without any memory visibility guarantee.
`HttpServer.create` without `setExecutor` uses `sun.net.httpserver.ServerImpl`'s default thread pool, which **is not single-threaded**. A concurrent polling route that fires two retrieve calls close together will have two threads writing `batch.statusIndex`, `batch.cancelling`, and `batch.cancelled` simultaneously with no synchronization. The mock is public test-infra API; other teams that use it with parallel routes will hit this.
The simplest fix: make `retrieveBatch` and `cancelBatch` in `BatchRequestHandler` `synchronized` on `batch` (or make `StoredBatch` hold a `ReentrantLock`), or mark the mutated fields `volatile`. The minimal safe approach here is to synchronize on the `StoredBatch` instance in both handlers:
```java
synchronized (batch) {
if (batch.cancelling) {
batch.cancelling = false;
batch.cancelled = true;
} else if (!batch.cancelled && batch.statusIndex < statusList().size() - 1) {
batch.statusIndex++;
}
}
```
##########
components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIBatchResultsProducer.java:
##########
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.openai;
+
+import java.util.Optional;
+
+import com.openai.core.http.HttpResponse;
+import com.openai.models.batches.Batch;
+import org.apache.camel.CamelExchangeException;
+import org.apache.camel.Exchange;
+import org.apache.camel.component.ai.observability.GenAiErrorSupport;
+import org.apache.camel.support.DefaultProducer;
+import org.apache.camel.support.SynchronizationAdapter;
+import org.apache.camel.util.ObjectHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * OpenAI producer for the {@code batch-results} operation, which downloads the output or error file of a finished
+ * batch.
+ * <p>
+ * The body is the file as an {@link java.io.InputStream}, so a large result can be split line by line with the splitter
+ * in streaming mode. The HTTP response is closed when the exchange completes. Stream caching reads such a body into
+ * memory before the next step, so a route handling large results should turn it off.
+ */
+public class OpenAIBatchResultsProducer extends DefaultProducer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(OpenAIBatchResultsProducer.class);
+ private static final String ERROR_FILE = "error";
+
+ public OpenAIBatchResultsProducer(OpenAIEndpoint endpoint) {
+ super(endpoint);
+ }
+
+ @Override
+ public OpenAIEndpoint getEndpoint() {
+ return (OpenAIEndpoint) super.getEndpoint();
+ }
+
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ OpenAIConfiguration config = getEndpoint().getConfiguration();
+ String batchId = exchange.getIn().getHeader(OpenAIConstants.BATCH_ID, String.class);
+ if (ObjectHelper.isEmpty(batchId)) {
+ throw new IllegalArgumentException(
+ "The " + OpenAIConstants.BATCH_ID + " header must hold the id of the batch to read the results of");
+ }
+
+ String resultsFile = exchange.getIn().getHeader(OpenAIConstants.BATCH_RESULTS_FILE,
+ config.getBatchResultsFile(), String.class);
+ if (!"output".equals(resultsFile) && !ERROR_FILE.equals(resultsFile)) {
+ throw new IllegalArgumentException(
+ "Unsupported batch results file: " + resultsFile + ". Supported: output, error");
+ }
+
+ Batch batch;
+ try {
+ batch = getEndpoint().getClient().batches().retrieve(batchId);
+ } catch (RuntimeException e) {
+ GenAiErrorSupport.apply(exchange, e);
+ throw e;
+ }
+ OpenAIBatchSupport.setBatchHeaders(exchange.getMessage(), batch);
+
+ // the API sets the file ids once it has stopped processing, so their presence, not the status, says whether
+ // the results can be read; the status only explains why a file is missing
+ Optional<String> fileId = ERROR_FILE.equals(resultsFile) ? batch.errorFileId() : batch.outputFileId();
+ if (fileId.isEmpty()) {
+ switch (batch.status().value()) {
+ case FAILED ->
+ // the input was rejected, so no result file exists: the reason is in the errors of the batch
+ throw new CamelExchangeException(
+ "Batch " + batchId + " failed: " + OpenAIBatchSupport.errors(batch), exchange);
Review Comment:
π **Misleading error message when `FAILED` batch has no errors populated**
```java
case FAILED ->
throw new CamelExchangeException(
"Batch " + batchId + " failed: " + OpenAIBatchSupport.errors(batch), exchange);
```
`errors(batch)` returns an empty `List` when `batch.errors()` is absent or `data` is empty β the OpenAI API does not guarantee that validation errors are always populated on a `failed` batch (e.g. the batch failed at the file-parsing stage before individual errors are attributed). The message becomes `"Batch X failed: []"`, which tells the user nothing.
Consider:
```java
case FAILED -> {
List<Map<String, Object>> errors = OpenAIBatchSupport.errors(batch);
String detail = errors.isEmpty() ? "check the batch in the OpenAI dashboard" : errors.toString();
throw new CamelExchangeException("Batch " + batchId + " failed: " + detail, exchange);
}
```
##########
test-infra/camel-test-infra-openai-mock/src/main/java/org/apache/camel/test/infra/openai/mock/BatchRequestHandler.java:
##########
@@ -0,0 +1,404 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.test.infra.openai.mock;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.regex.Pattern;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.sun.net.httpserver.HttpExchange;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Serves the Files and Batch APIs: uploading an input file, creating a batch from it, reporting its status and serving
+ * the result files.
+ * <p>
+ * A batch walks the configured status progression, one step per retrieve, so a route that polls sees the same sequence
+ * of states as it would against the real API. A cancel moves a running batch to {@code cancelling}, and the next
+ * retrieve to {@code cancelled}. Once a final status is reached the output and error files are built from the input
+ * file, matching each request line to a {@link BatchExpectation} by its {@code custom_id}.
+ */
+public class BatchRequestHandler {
+ private static final Logger LOG = LoggerFactory.getLogger(BatchRequestHandler.class);
+
+ static final List<String> DEFAULT_STATUSES = List.of("validating", "in_progress", "finalizing", "completed");
+
+ private static final String DEFAULT_RESPONSE_BODY
+ = "{\"id\":\"chatcmpl-mock\",\"object\":\"chat.completion\",\"model\":\"openai-mock\","
+ + "\"choices\":[{\"index\":0,\"message\":{\"role\":\"assistant\",\"content\":\"mock response\"},"
+ + "\"finish_reason\":\"stop\"}]}";
+
+ private final List<BatchExpectation> expectations;
+ private final List<String> statuses;
+ private final BatchStore store;
+ private final ObjectMapper objectMapper;
+
+ public BatchRequestHandler(List<BatchExpectation> expectations, List<String> statuses, BatchStore store,
+ ObjectMapper objectMapper) {
+ this.expectations = expectations;
+ this.statuses = statuses;
+ this.store = store;
+ this.objectMapper = objectMapper;
+ }
+
+ /**
+ * Whether this handler serves the given path, so the dispatcher can route Files and Batch calls here before the
+ * handlers of the inference APIs.
+ */
+ public static boolean handles(String path) {
+ return path.endsWith("/files") || path.contains("/files/")
+ || path.endsWith("/batches") || path.contains("/batches/");
+ }
+
+ public void handleRequest(HttpExchange exchange) throws IOException {
+ String path = exchange.getRequestURI().getPath();
+ String method = exchange.getRequestMethod();
+ try {
+ if (path.endsWith("/files") && "POST".equalsIgnoreCase(method)) {
+ uploadFile(exchange);
+ } else if (path.endsWith("/content") && "GET".equalsIgnoreCase(method)) {
+ fileContent(exchange, idBefore(path, "/content"));
+ } else if (path.contains("/files/") && "GET".equalsIgnoreCase(method)) {
+ fileObject(exchange, lastSegment(path));
+ } else if (path.contains("/files/") && "DELETE".equalsIgnoreCase(method)) {
+ deleteFile(exchange, lastSegment(path));
+ } else if (path.endsWith("/batches") && "POST".equalsIgnoreCase(method)) {
+ createBatch(exchange);
+ } else if (path.endsWith("/cancel") && "POST".equalsIgnoreCase(method)) {
+ cancelBatch(exchange, idBefore(path, "/cancel"));
+ } else if (path.contains("/batches/") && "GET".equalsIgnoreCase(method)) {
+ retrieveBatch(exchange, lastSegment(path));
+ } else {
+ sendError(exchange, 404, "invalid_request_error", "Unsupported request: " + method + " " + path);
+ }
+ } catch (Exception e) {
+ LOG.error("Error processing batch request {} {}", method, path, e);
+ sendError(exchange, 500, "internal_error", String.valueOf(e.getMessage()));
+ }
+ }
+
+ private void uploadFile(HttpExchange exchange) throws IOException {
+ byte[] body;
+ try (InputStream is = exchange.getRequestBody()) {
+ body = is.readAllBytes();
+ }
+ String contentType = exchange.getRequestHeaders().getFirst("Content-Type");
+ String boundary = contentType != null && contentType.contains("boundary=")
+ ? contentType.substring(contentType.indexOf("boundary=") + "boundary=".length()).trim() : null;
+ if (boundary == null) {
+ sendError(exchange, 400, "invalid_request_error", "The file upload must be a multipart request");
+ return;
+ }
+
+ // ISO-8859-1 maps every byte to one character, so the file part survives the split unchanged
+ String raw = new String(body, StandardCharsets.ISO_8859_1);
+ String filename = "unknown";
+ String purpose = "unknown";
+ byte[] content = new byte[0];
+ // the boundary may hold regex metacharacters, so it is quoted rather than spliced into the pattern
+ for (String part : raw.split(Pattern.quote("--" + boundary))) {
+ int headerEnd = part.indexOf("\r\n\r\n");
+ if (headerEnd < 0) {
+ continue;
+ }
+ String headers = part.substring(0, headerEnd);
+ String value = part.substring(headerEnd + 4);
+ if (value.endsWith("\r\n")) {
+ value = value.substring(0, value.length() - 2);
+ }
+ if (headers.contains("name=\"file\"")) {
+ filename = between(headers, "filename=\"", "\"");
+ content = value.getBytes(StandardCharsets.ISO_8859_1);
+ } else if (headers.contains("name=\"purpose\"")) {
+ purpose = value.trim();
+ }
+ }
+
+ String id = store.addFile(filename, purpose, content);
+ LOG.debug("Stored uploaded file {} ({} bytes, purpose {})", id, content.length, purpose);
+ sendJson(exchange, 200, fileNode(id, filename, purpose, content.length).toString());
+ }
+
+ private void fileContent(HttpExchange exchange, String fileId) throws IOException {
+ byte[] content = store.getFileContent(fileId);
+ if (content == null) {
+ sendError(exchange, 404, "invalid_request_error", "No such file: " + fileId);
+ return;
+ }
+ exchange.getResponseHeaders().add("Content-Type", "application/octet-stream");
+ exchange.sendResponseHeaders(200, content.length);
+ try (OutputStream os = exchange.getResponseBody()) {
+ os.write(content);
+ }
+ }
+
+ private void fileObject(HttpExchange exchange, String fileId) throws IOException {
+ BatchStore.StoredFile file = store.file(fileId);
+ if (file == null) {
+ sendError(exchange, 404, "invalid_request_error", "No such file: " + fileId);
+ return;
+ }
+ sendJson(exchange, 200, fileNode(file.id(), file.filename(), file.purpose(), file.content().length).toString());
+ }
+
+ private void deleteFile(HttpExchange exchange, String fileId) throws IOException {
+ if (store.file(fileId) == null) {
+ sendError(exchange, 404, "invalid_request_error", "No such file: " + fileId);
+ return;
+ }
+ store.removeFile(fileId);
+ ObjectNode node = objectMapper.createObjectNode();
+ node.put("id", fileId);
+ node.put("object", "file");
+ node.put("deleted", true);
+ sendJson(exchange, 200, node.toString());
+ }
+
+ private void createBatch(HttpExchange exchange) throws IOException {
+ JsonNode request;
+ try (InputStream is = exchange.getRequestBody()) {
+ request = objectMapper.readTree(is.readAllBytes());
+ }
+ String inputFileId = request.path("input_file_id").asText();
+ if (store.file(inputFileId) == null) {
+ sendError(exchange, 404, "invalid_request_error", "No such file: " + inputFileId);
+ return;
+ }
+ JsonNode metadata = request.get("metadata");
+ BatchStore.StoredBatch batch = store.addBatch(inputFileId, request.path("endpoint").asText(),
+ metadata != null && !metadata.isNull() ? metadata.toString() : null);
+ maybeBuildResults(batch);
+ LOG.debug("Created batch {} for input file {}", batch.id, inputFileId);
+ sendJson(exchange, 200, batchNode(batch).toString());
+ }
+
+ private void retrieveBatch(HttpExchange exchange, String batchId) throws IOException {
+ BatchStore.StoredBatch batch = store.batch(batchId);
+ if (batch == null) {
+ sendError(exchange, 404, "invalid_request_error", "No such batch: " + batchId);
+ return;
+ }
+ if (batch.cancelling) {
+ // a cancelled batch is final, so the retrieve after the cancel reports it as such, as the API does
+ batch.cancelling = false;
+ batch.cancelled = true;
+ } else if (!batch.cancelled && batch.statusIndex < statusList().size() - 1) {
+ batch.statusIndex++;
+ }
+ maybeBuildResults(batch);
+ sendJson(exchange, 200, batchNode(batch).toString());
+ }
+
+ private void cancelBatch(HttpExchange exchange, String batchId) throws IOException {
+ BatchStore.StoredBatch batch = store.batch(batchId);
+ if (batch == null) {
+ sendError(exchange, 404, "invalid_request_error", "No such batch: " + batchId);
+ return;
+ }
+ if (isFinal(status(batch))) {
+ sendError(exchange, 400, "invalid_request_error",
+ "Cannot cancel a batch with status '" + status(batch) + "'.");
+ return;
+ }
+ batch.cancelling = true;
+ sendJson(exchange, 200, batchNode(batch).toString());
+ }
+
+ private List<String> statusList() {
+ return statuses.isEmpty() ? DEFAULT_STATUSES : statuses;
+ }
+
+ private String status(BatchStore.StoredBatch batch) {
+ if (batch.cancelled) {
+ return "cancelled";
+ }
+ if (batch.cancelling) {
+ return "cancelling";
+ }
+ return statusList().get(Math.min(batch.statusIndex, statusList().size() - 1));
+ }
+
+ private boolean isFinal(String status) {
+ return "completed".equals(status) || "expired".equals(status) || "cancelled".equals(status);
+ }
+
+ /**
+ * Builds the output and error files once the batch reaches a final status, as the API does when it finishes
+ * processing the input file.
+ */
+ private void maybeBuildResults(BatchStore.StoredBatch batch) throws IOException {
+ if (batch.resultsBuilt || !isFinal(status(batch))) {
+ return;
+ }
+ batch.resultsBuilt = true;
+
+ BatchStore.StoredFile input = store.file(batch.inputFileId);
+ List<String> outputLines = new ArrayList<>();
+ List<String> errorLines = new ArrayList<>();
+ int line = 0;
+ for (String requestLine : new String(input.content(), StandardCharsets.UTF_8).split("\n")) {
Review Comment:
β οΈ **JSONL split on `\n` will break on Windows**
```java
for (String requestLine : new String(input.content(), StandardCharsets.UTF_8).split("\n")) {
```
The producer writes the temp file with `BufferedWriter.newLine()`, which uses the **platform line separator** β `\r\n` on Windows. If the mock runs on Windows (or the input was produced there), splitting on `\n` leaves a trailing `\r` in each line. `objectMapper.readTree(requestLine).path("custom_id").asText()` then returns `"ticket-1\r"` instead of `"ticket-1"`, no expectation matches, and all lines fall to the `mock_no_expectation` error path.
Fix:
```suggestion
for (String requestLine : new String(input.content(), StandardCharsets.UTF_8).split("\r?\n")) {
```
The output-file joins are `String.join("\n", ...)` (always `\n`), which is fine. Only the input split needs the fix.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-24674: camel-openai - support the Batch API [camel]
Posted by "Croway (via GitHub)" <gi...@apache.org>.
Croway commented on code in PR #26601:
URL: https://github.com/apache/camel/pull/26601#discussion_r4053371785
##########
components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIBatchProducer.java:
##########
@@ -0,0 +1,467 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.openai;
+
+import java.io.ByteArrayInputStream;
+import java.io.Closeable;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.SequenceInputStream;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.util.Enumeration;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.openai.core.JsonValue;
+import com.openai.core.MultipartField;
+import com.openai.core.ObjectMappers;
+import com.openai.models.ResponseFormatJsonSchema;
+import com.openai.models.batches.Batch;
+import com.openai.models.batches.BatchCreateParams;
+import com.openai.models.chat.completions.ChatCompletionCreateParams;
+import com.openai.models.embeddings.EmbeddingCreateParams;
+import com.openai.models.files.FileCreateParams;
+import com.openai.models.files.FilePurpose;
+import com.openai.models.moderations.ModerationCreateParams;
+import com.openai.models.responses.ResponseCreateParams;
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.WrappedFile;
+import org.apache.camel.component.ai.observability.GenAiErrorSupport;
+import org.apache.camel.component.openai.OpenAIResponsesInputBuilder.InputSpec;
+import org.apache.camel.support.DefaultProducer;
+import org.apache.camel.util.ObjectHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * OpenAI producer for the {@code batch} operation, which uploads the requests of a batch as a JSONL file and creates
+ * the batch that runs them.
+ * <p>
+ * The body is either the JSONL itself, as a {@link File}, {@link Path}, {@link WrappedFile}, {@link InputStream},
+ * {@code byte[]} or String, or a {@link Map} keyed by {@code custom_id}. A map value that is itself a map is used as
+ * the request body as it is, while a String value is turned into a request built from the endpoint options, so a route
+ * that only has prompts does not have to assemble the API payload itself. An {@link Iterable} body, such as the
+ * {@link OpenAIBatchSpool} of an aggregation, is read the same way one item at a time. Such a request is built with the
+ * same SDK parameters as the synchronous operation of the endpoint, so the options and headers of that operation apply.
+ */
+public class OpenAIBatchProducer extends DefaultProducer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(OpenAIBatchProducer.class);
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+ private static final String INPUT_FILE_SUFFIX = ".jsonl";
+ private static final String INPUT_FILE_NAME = "camel-openai-batch" + INPUT_FILE_SUFFIX;
+
+ private Class<?> outputClassResolved;
+
+ public OpenAIBatchProducer(OpenAIEndpoint endpoint) {
+ super(endpoint);
+ }
+
+ @Override
+ public OpenAIEndpoint getEndpoint() {
+ return (OpenAIEndpoint) super.getEndpoint();
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+ String outputClass = getEndpoint().getConfiguration().getOutputClass();
+ if (ObjectHelper.isNotEmpty(outputClass)) {
+ outputClassResolved = getEndpoint().getCamelContext().getClassResolver().resolveMandatoryClass(outputClass);
+ }
+ }
+
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ OpenAIConfiguration config = getEndpoint().getConfiguration();
+ Message in = exchange.getIn();
+ // the options a batch cannot honour are refused when the endpoint starts; the header can still ask for them
+ if (Boolean.TRUE.equals(in.getHeader(OpenAIConstants.STREAMING, Boolean.class))) {
+ throw new IllegalArgumentException(
+ "The batch operation cannot stream responses; remove the " + OpenAIConstants.STREAMING + " header");
+ }
+
+ String endpoint = OpenAIBatchSupport.resolveEndpoint(in, config);
+
+ Object body = in.getBody();
+ String inputFileId;
+ if (body instanceof Map<?, ?> map) {
+ inputFileId = upload(exchange, requestLines(map.entrySet(), endpoint, in, config), INPUT_FILE_NAME);
+ } else if (body instanceof Iterable<?> items) {
+ inputFileId = upload(exchange, requestLines(items, endpoint, in, config), INPUT_FILE_NAME);
+ } else {
+ InputStream stream = in.getBody(InputStream.class);
+ if (stream == null) {
+ throw new IllegalArgumentException(
+ "Unsupported body type for the batch operation: "
+ + (body != null ? body.getClass().getName() : "null")
+ + ". Supported: File, Path, InputStream, byte[], String, a Map "
+ + "keyed by custom_id, or an Iterable");
+ }
+ inputFileId = upload(exchange, stream, inputFileName(body));
+ }
+ in.setHeader(OpenAIConstants.BATCH_INPUT_FILE_ID, inputFileId);
+
+ Batch batch = create(exchange, inputFileId, endpoint, in, config);
+ if (body instanceof OpenAIBatchSpool spool) {
+ // the spool of an aggregation is consumed by this batch, so it is not left behind in the directory
+ spool.delete();
Review Comment:
Agreed, applied in 0f78bc3d3db4: the delete is wrapped, a failure is logged at WARN with the spool path and the batch id, and the headers are set regardless.
_Claude Code on behalf of Federico Mariani (Croway)_
##########
components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIBatchProducer.java:
##########
@@ -0,0 +1,467 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.openai;
+
+import java.io.ByteArrayInputStream;
+import java.io.Closeable;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.SequenceInputStream;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.util.Enumeration;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.openai.core.JsonValue;
+import com.openai.core.MultipartField;
+import com.openai.core.ObjectMappers;
+import com.openai.models.ResponseFormatJsonSchema;
+import com.openai.models.batches.Batch;
+import com.openai.models.batches.BatchCreateParams;
+import com.openai.models.chat.completions.ChatCompletionCreateParams;
+import com.openai.models.embeddings.EmbeddingCreateParams;
+import com.openai.models.files.FileCreateParams;
+import com.openai.models.files.FilePurpose;
+import com.openai.models.moderations.ModerationCreateParams;
+import com.openai.models.responses.ResponseCreateParams;
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.WrappedFile;
+import org.apache.camel.component.ai.observability.GenAiErrorSupport;
+import org.apache.camel.component.openai.OpenAIResponsesInputBuilder.InputSpec;
+import org.apache.camel.support.DefaultProducer;
+import org.apache.camel.util.ObjectHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * OpenAI producer for the {@code batch} operation, which uploads the requests of a batch as a JSONL file and creates
+ * the batch that runs them.
+ * <p>
+ * The body is either the JSONL itself, as a {@link File}, {@link Path}, {@link WrappedFile}, {@link InputStream},
+ * {@code byte[]} or String, or a {@link Map} keyed by {@code custom_id}. A map value that is itself a map is used as
+ * the request body as it is, while a String value is turned into a request built from the endpoint options, so a route
+ * that only has prompts does not have to assemble the API payload itself. An {@link Iterable} body, such as the
+ * {@link OpenAIBatchSpool} of an aggregation, is read the same way one item at a time. Such a request is built with the
+ * same SDK parameters as the synchronous operation of the endpoint, so the options and headers of that operation apply.
+ */
+public class OpenAIBatchProducer extends DefaultProducer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(OpenAIBatchProducer.class);
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+ private static final String INPUT_FILE_SUFFIX = ".jsonl";
+ private static final String INPUT_FILE_NAME = "camel-openai-batch" + INPUT_FILE_SUFFIX;
+
+ private Class<?> outputClassResolved;
+
+ public OpenAIBatchProducer(OpenAIEndpoint endpoint) {
+ super(endpoint);
+ }
+
+ @Override
+ public OpenAIEndpoint getEndpoint() {
+ return (OpenAIEndpoint) super.getEndpoint();
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+ String outputClass = getEndpoint().getConfiguration().getOutputClass();
+ if (ObjectHelper.isNotEmpty(outputClass)) {
+ outputClassResolved = getEndpoint().getCamelContext().getClassResolver().resolveMandatoryClass(outputClass);
+ }
+ }
+
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ OpenAIConfiguration config = getEndpoint().getConfiguration();
+ Message in = exchange.getIn();
+ // the options a batch cannot honour are refused when the endpoint starts; the header can still ask for them
+ if (Boolean.TRUE.equals(in.getHeader(OpenAIConstants.STREAMING, Boolean.class))) {
+ throw new IllegalArgumentException(
+ "The batch operation cannot stream responses; remove the " + OpenAIConstants.STREAMING + " header");
+ }
+
+ String endpoint = OpenAIBatchSupport.resolveEndpoint(in, config);
+
+ Object body = in.getBody();
+ String inputFileId;
+ if (body instanceof Map<?, ?> map) {
+ inputFileId = upload(exchange, requestLines(map.entrySet(), endpoint, in, config), INPUT_FILE_NAME);
+ } else if (body instanceof Iterable<?> items) {
+ inputFileId = upload(exchange, requestLines(items, endpoint, in, config), INPUT_FILE_NAME);
+ } else {
+ InputStream stream = in.getBody(InputStream.class);
+ if (stream == null) {
+ throw new IllegalArgumentException(
+ "Unsupported body type for the batch operation: "
+ + (body != null ? body.getClass().getName() : "null")
+ + ". Supported: File, Path, InputStream, byte[], String, a Map "
+ + "keyed by custom_id, or an Iterable");
+ }
+ inputFileId = upload(exchange, stream, inputFileName(body));
+ }
+ in.setHeader(OpenAIConstants.BATCH_INPUT_FILE_ID, inputFileId);
+
+ Batch batch = create(exchange, inputFileId, endpoint, in, config);
+ if (body instanceof OpenAIBatchSpool spool) {
+ // the spool of an aggregation is consumed by this batch, so it is not left behind in the directory
+ spool.delete();
+ }
Review Comment:
Agreed, applied in 0f78bc3d3db4: the delete is wrapped, a failure is logged at WARN with the spool path and the batch id, and the headers are set regardless.
_Claude Code on behalf of Federico Mariani (Croway)_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-24674: camel-openai - support the Batch API [camel]
Posted by "gnodet-bot (via GitHub)" <gi...@apache.org>.
gnodet-bot commented on code in PR #26601:
URL: https://github.com/apache/camel/pull/26601#discussion_r4053314642
##########
components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIBatchProducer.java:
##########
@@ -0,0 +1,467 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.openai;
+
+import java.io.ByteArrayInputStream;
+import java.io.Closeable;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.SequenceInputStream;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.util.Enumeration;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.openai.core.JsonValue;
+import com.openai.core.MultipartField;
+import com.openai.core.ObjectMappers;
+import com.openai.models.ResponseFormatJsonSchema;
+import com.openai.models.batches.Batch;
+import com.openai.models.batches.BatchCreateParams;
+import com.openai.models.chat.completions.ChatCompletionCreateParams;
+import com.openai.models.embeddings.EmbeddingCreateParams;
+import com.openai.models.files.FileCreateParams;
+import com.openai.models.files.FilePurpose;
+import com.openai.models.moderations.ModerationCreateParams;
+import com.openai.models.responses.ResponseCreateParams;
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.WrappedFile;
+import org.apache.camel.component.ai.observability.GenAiErrorSupport;
+import org.apache.camel.component.openai.OpenAIResponsesInputBuilder.InputSpec;
+import org.apache.camel.support.DefaultProducer;
+import org.apache.camel.util.ObjectHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * OpenAI producer for the {@code batch} operation, which uploads the requests of a batch as a JSONL file and creates
+ * the batch that runs them.
+ * <p>
+ * The body is either the JSONL itself, as a {@link File}, {@link Path}, {@link WrappedFile}, {@link InputStream},
+ * {@code byte[]} or String, or a {@link Map} keyed by {@code custom_id}. A map value that is itself a map is used as
+ * the request body as it is, while a String value is turned into a request built from the endpoint options, so a route
+ * that only has prompts does not have to assemble the API payload itself. An {@link Iterable} body, such as the
+ * {@link OpenAIBatchSpool} of an aggregation, is read the same way one item at a time. Such a request is built with the
+ * same SDK parameters as the synchronous operation of the endpoint, so the options and headers of that operation apply.
+ */
+public class OpenAIBatchProducer extends DefaultProducer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(OpenAIBatchProducer.class);
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+ private static final String INPUT_FILE_SUFFIX = ".jsonl";
+ private static final String INPUT_FILE_NAME = "camel-openai-batch" + INPUT_FILE_SUFFIX;
+
+ private Class<?> outputClassResolved;
+
+ public OpenAIBatchProducer(OpenAIEndpoint endpoint) {
+ super(endpoint);
+ }
+
+ @Override
+ public OpenAIEndpoint getEndpoint() {
+ return (OpenAIEndpoint) super.getEndpoint();
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+ String outputClass = getEndpoint().getConfiguration().getOutputClass();
+ if (ObjectHelper.isNotEmpty(outputClass)) {
+ outputClassResolved = getEndpoint().getCamelContext().getClassResolver().resolveMandatoryClass(outputClass);
+ }
+ }
+
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ OpenAIConfiguration config = getEndpoint().getConfiguration();
+ Message in = exchange.getIn();
+ // the options a batch cannot honour are refused when the endpoint starts; the header can still ask for them
+ if (Boolean.TRUE.equals(in.getHeader(OpenAIConstants.STREAMING, Boolean.class))) {
+ throw new IllegalArgumentException(
+ "The batch operation cannot stream responses; remove the " + OpenAIConstants.STREAMING + " header");
+ }
+
+ String endpoint = OpenAIBatchSupport.resolveEndpoint(in, config);
+
+ Object body = in.getBody();
+ String inputFileId;
+ if (body instanceof Map<?, ?> map) {
+ inputFileId = upload(exchange, requestLines(map.entrySet(), endpoint, in, config), INPUT_FILE_NAME);
+ } else if (body instanceof Iterable<?> items) {
+ inputFileId = upload(exchange, requestLines(items, endpoint, in, config), INPUT_FILE_NAME);
+ } else {
+ InputStream stream = in.getBody(InputStream.class);
+ if (stream == null) {
+ throw new IllegalArgumentException(
+ "Unsupported body type for the batch operation: "
+ + (body != null ? body.getClass().getName() : "null")
+ + ". Supported: File, Path, InputStream, byte[], String, a Map "
+ + "keyed by custom_id, or an Iterable");
+ }
+ inputFileId = upload(exchange, stream, inputFileName(body));
+ }
+ in.setHeader(OpenAIConstants.BATCH_INPUT_FILE_ID, inputFileId);
+
+ Batch batch = create(exchange, inputFileId, endpoint, in, config);
+ if (body instanceof OpenAIBatchSpool spool) {
+ // the spool of an aggregation is consumed by this batch, so it is not left behind in the directory
+ spool.delete();
+ }
Review Comment:
β οΈ **`spool.delete()` failure orphans a successfully-created batch.**
`create()` has returned a valid `Batch` at this point. If `Files.deleteIfExists()` throws (read-only filesystem, network mount, etc.) the `IOException` propagates out of `process()` as an exchange failure β but `setBatchHeaders()` on line 134 is never reached. The caller receives no `CamelOpenAIBatchId`, the batch runs untracked on OpenAI's servers, and there is no way to retrieve its results or cancel it.
Spool deletion is cleanup; it must never mask a successful batch creation.
```suggestion
if (body instanceof OpenAIBatchSpool spool) {
// the spool of an aggregation is consumed by this batch, so it is not left behind in the directory
try {
spool.delete();
} catch (IOException e) {
LOG.warn("Could not delete the batch spool file {}; the batch {} was created", spool.getFile(), batch.id(), e);
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-24674: camel-openai - support the Batch API [camel]
Posted by "Croway (via GitHub)" <gi...@apache.org>.
Croway commented on PR #26601:
URL: https://github.com/apache/camel/pull/26601#issuecomment-5742240320
Thanks for the review. The doc example is fixed in f08c4bee (inline reply) and the build log line counts the batch expectations since f3d018e6. `BATCH_METADATA` as `Map<String, String>` is deliberate: the API stores string values and the producer stringifies whatever it gets.
f08c4bee also changes two shapes worth a look: `batch-results` now returns an iterator of parsed lines instead of the raw stream and the input is streamed line by line, so neither side holds the file in memory; and `OpenAIBatchAggregationStrategy` (in memory, or spooled to a file) turns a stream of messages into a batch with the aggregator.
_Claude Code on behalf of Federico Mariani (Croway)_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-24674: camel-openai - support the Batch API [camel]
Posted by "Croway (via GitHub)" <gi...@apache.org>.
Croway commented on code in PR #26601:
URL: https://github.com/apache/camel/pull/26601#discussion_r4053403080
##########
components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIBatchAggregationStrategy.java:
##########
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.openai;
+
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.camel.AggregationStrategy;
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.util.IOHelper;
+import org.apache.camel.util.ObjectHelper;
+
+/**
+ * Collects the messages of an aggregation into the input of a batch, so a stream of prompts becomes one
+ * {@code openai:batch} call:
+ *
+ * <pre>
+ * from("kafka:tickets")
+ * .aggregate(constant(true), new OpenAIBatchAggregationStrategy())
+ * .completionSize(1000).completionTimeout(60000)
+ * .to("openai:batch?batchEndpoint=/v1/chat/completions&model=gpt-4o-mini");
+ * </pre>
+ *
+ * The body of each message is the value of its request, as for a {@link Map} body of the {@code batch} operation: a
+ * String is turned into a request from the endpoint options, a {@code Map} or {@code JsonNode} is the request body. Its
+ * {@code custom_id} is the {@code CamelOpenAIBatchCustomId} header when set, and the message id otherwise, so the
+ * result lines can be matched back to what produced them.
+ * <p>
+ * By default the requests are collected in a {@link Map} in memory, which suits batches of a few thousand short
+ * requests. {@link #spooled()} writes them to a file instead, one line per message, so a batch of any size is
+ * aggregated with constant memory, and a persistent {@code AggregationRepository} stores a path rather than the
+ * requests. The {@code batch} operation streams either form and removes the file once the batch is created.
+ */
+public class OpenAIBatchAggregationStrategy implements AggregationStrategy {
+
+ private final Path spoolDirectory;
+ // the writer of each spool being aggregated, closed when its aggregation completes
+ private final Map<File, BufferedWriter> writers = new ConcurrentHashMap<>();
+
+ /**
+ * Collects the requests in memory.
+ */
+ public OpenAIBatchAggregationStrategy() {
+ this(null);
+ }
+
+ /**
+ * Collects the requests in a file created in the given directory.
+ */
+ public OpenAIBatchAggregationStrategy(Path spoolDirectory) {
+ this.spoolDirectory = spoolDirectory;
+ }
+
+ /**
+ * Collects the requests in a file created in the temporary directory of the JVM.
+ */
+ public static OpenAIBatchAggregationStrategy spooled() {
+ return new OpenAIBatchAggregationStrategy(Path.of(System.getProperty("java.io.tmpdir")));
+ }
+
+ public boolean isSpooled() {
+ return spoolDirectory != null;
+ }
+
+ @Override
+ public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
+ Message message = newExchange.getMessage();
+ String customId = message.getHeader(OpenAIConstants.BATCH_CUSTOM_ID, message.getMessageId(), String.class);
+ Object value = message.getBody();
+ if (ObjectHelper.isEmpty(customId) || value == null) {
+ throw new IllegalArgumentException(
+ "A message aggregated into a batch needs a custom_id and a body; got custom_id " + customId
+ + " and body " + value);
+ }
+
+ try {
+ if (oldExchange == null) {
+ if (isSpooled()) {
+ OpenAIBatchSpool spool = OpenAIBatchSpool.create(spoolDirectory);
+ OpenAIBatchSpool.append(writer(spool), customId, value);
+ message.setBody(spool);
+ } else {
+ Map<String, Object> requests = new LinkedHashMap<>();
+ requests.put(customId, value);
+ message.setBody(requests);
+ }
+ return newExchange;
+ }
+
+ Object aggregated = oldExchange.getMessage().getBody();
+ if (aggregated instanceof OpenAIBatchSpool spool) {
+ OpenAIBatchSpool.append(writer(spool), customId, value);
+ } else if (aggregated instanceof Map<?, ?> requests) {
+ @SuppressWarnings("unchecked")
+ Map<String, Object> map = (Map<String, Object>) requests;
+ map.put(customId, value);
+ } else {
+ throw new IllegalStateException("Unexpected aggregated batch body: " + aggregated);
+ }
+ return oldExchange;
+ } catch (IOException e) {
+ throw new RuntimeCamelException("Cannot spool the batch request of custom_id " + customId, e);
+ }
+ }
+
+ @Override
+ public void onCompletion(Exchange exchange) {
+ if (exchange != null && exchange.getMessage().getBody() instanceof OpenAIBatchSpool spool) {
+ BufferedWriter writer = writers.remove(spool.getFile());
+ if (writer != null) {
+ IOHelper.close(writer);
+ }
+ }
+ }
+
+ private BufferedWriter writer(OpenAIBatchSpool spool) throws IOException {
+ BufferedWriter writer = writers.get(spool.getFile());
+ if (writer == null) {
+ writer = spool.open();
+ writers.put(spool.getFile(), writer);
+ }
+ return writer;
Review Comment:
Applied in 2e3ddae45db8 with `computeIfAbsent`, unwrapping the `UncheckedIOException` so the method keeps its checked signature. As you note, the aggregator serialises calls per correlation key, so no two threads share a spool; the map is concurrent because different aggregations run at once, each on its own file.
_Claude Code on behalf of Federico Mariani (Croway)_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-24674: camel-openai - support the Batch API [camel]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #26601:
URL: https://github.com/apache/camel/pull/26601#issuecomment-5730981107
:star2: Thank you for your contribution to the Apache Camel project! :star2:
:robot: CI automation will test this PR automatically.
:camel: Apache Camel Committers, please review the following items:
* First-time contributors **require MANUAL approval** for the GitHub Actions to run
* You can use the command `/component-test (camel-)component-name1 (camel-)component-name2..` to request a test from the test bot although they are normally detected and executed by CI.
* You can label PRs using `skip-tests` and `test-dependents` to fine-tune the checks executed by this PR.
* Build and test logs are available in the summary page. **Only** [Apache Camel committers](https://camel.apache.org/community/team/#committers) have access to the summary.
:warning: Be careful when sharing logs. Review their contents before sharing them publicly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-24674: camel-openai - support the Batch API [camel]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #26601:
URL: https://github.com/apache/camel/pull/26601#issuecomment-5731185961
<!-- regen-patch -->
### :warning: Generated files are out of date
The [CI build](https://github.com/apache/camel/actions/runs/35352640831) regenerated files that are not committed on this PR.
If *Allow edits from maintainers* is enabled, a maintainer bot will automatically
push a "Regen" commit to this branch shortly.
Otherwise, you can apply the patch manually β the missing changes are available
as the `regen-patch` artifact of that run:
```bash
gh pr checkout 26601 --repo apache/camel
gh run download 35352640831 --repo apache/camel -n regen-patch
git apply --index regen.patch && rm regen.patch pr-number
git commit -m "Regen" && git push
```
Alternatively, run `./etc/scripts/regen.sh` locally and commit the resulting changes.
If the patch does not apply cleanly, update the branch from the base branch first β
the patch was generated against the merge with the base branch.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org