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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
