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