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