gnodet commented on code in PR #26106:
URL: https://github.com/apache/camel/pull/26106#discussion_r3931470723
##########
components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIEmbeddingsProducer.java:
##########
@@ -127,6 +151,10 @@ private void processInternal(Exchange exchange) throws
Exception {
calculateSimilarityIfRequested(exchange, embeddings);
}
+ private static Integer toTokenCount(long tokens) {
Review Comment:
`Math.toIntExact(tokens)` will throw `ArithmeticException` if the token
count exceeds `Integer.MAX_VALUE`. Since this is called inside the `try` block
that catches `Exception`, a hypothetical overflow would abort the entire
exchange even though the API call itself succeeded.
More importantly, this narrowing is unnecessary. `GenAiUsage` accepts `Long`
directly — the existing `createChatCompletion` in `OpenAIProducer` passes
`usage.promptTokens()` (a `long`) straight to `GenAiUsage.of(Long, Long, ...)`
without any conversion:
```suggestion
private static Long toTokenCount(long tokens) {
return tokens;
}
```
Or just inline the `long` values directly and drop `toTokenCount` entirely,
matching the `OpenAIProducer` pattern.
##########
components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIResponsesProducer.java:
##########
@@ -136,28 +142,89 @@ private void processInternal(Exchange exchange) throws
Exception {
Class<?> responseClass = resolveOutputClass(in, outputClass);
if (responseClass != null) {
- processStructured(exchange, config, paramsBuilder, responseClass);
+ processStructured(exchange, config, paramsBuilder, responseClass,
model);
return;
}
if (ObjectHelper.isNotEmpty(jsonSchema)) {
OpenAIResponsesSupport.applyJsonSchemaTextFormat(paramsBuilder,
jsonSchema);
}
ResponseCreateParams params = paramsBuilder.build();
- Response response =
getEndpoint().getClient().responses().create(params);
+ Response response = createResponse(exchange, model, params);
finishExchange(exchange, config, response,
OpenAIResponsesSupport.extractAssistantText(response));
}
private void processStructured(
Exchange exchange, OpenAIConfiguration config,
ResponseCreateParams.Builder paramsBuilder,
- Class<?> responseClass)
+ Class<?> responseClass, String model)
throws Exception {
StructuredResponseCreateParams<?> structuredParams =
paramsBuilder.text(responseClass).build();
- StructuredResponse<?> structured =
getEndpoint().getClient().responses().create(structuredParams);
- Response raw = structured.rawResponse();
+ Response raw = createStructuredResponse(exchange, model,
structuredParams);
finishExchange(exchange, config, raw,
OpenAIResponsesSupport.extractAssistantText(raw));
}
+ private Response createResponse(Exchange exchange, String model,
ResponseCreateParams params) throws Exception {
+ GenAiObservationContext observationContext =
GenAiObservationContext.builder()
+ .operationName(GenAiOperationName.CHAT)
+ .system("openai")
+ .requestModel(model)
+ .componentScheme("openai")
+ .build();
+ GenAiObservation observation = GenAiObservability.start(exchange,
observationContext);
+ try {
+ Response response =
getEndpoint().getClient().responses().create(params);
+ recordResponseSuccess(observation, response);
+ return response;
+ } catch (Exception e) {
+ GenAiErrorSupport.apply(exchange, e);
+ observation.recordError(e);
+ throw e;
+ } finally {
+ observation.close();
+ }
+ }
+
+ private Response createStructuredResponse(
+ Exchange exchange, String model, StructuredResponseCreateParams<?>
structuredParams)
+ throws Exception {
+ GenAiObservationContext observationContext =
GenAiObservationContext.builder()
+ .operationName(GenAiOperationName.CHAT)
+ .system("openai")
+ .requestModel(model)
+ .componentScheme("openai")
+ .build();
+ GenAiObservation observation = GenAiObservability.start(exchange,
observationContext);
+ try {
+ StructuredResponse<?> structured =
getEndpoint().getClient().responses().create(structuredParams);
+ Response raw = structured.rawResponse();
+ recordResponseSuccess(observation, raw);
+ return raw;
+ } catch (Exception e) {
+ GenAiErrorSupport.apply(exchange, e);
+ observation.recordError(e);
+ throw e;
+ } finally {
+ observation.close();
+ }
+ }
+
+ private static void recordResponseSuccess(GenAiObservation observation,
Response response) {
+ String finishReason =
OpenAIResponsesSupport.extractFinishStatus(response)
+ .map(OpenAIResponsesProducer::mapFinishReason)
+ .orElse(null);
+ response.usage().ifPresentOrElse(
+ usage -> observation.recordSuccess(GenAiUsage.of(
+ toTokenCount(usage.inputTokens()),
+ toTokenCount(usage.outputTokens()),
+ finishReason,
+ response.model().toString())),
+ () -> observation.recordSuccess(GenAiUsage.of(null, null,
finishReason, response.model().toString())));
+ }
Review Comment:
Same issue as in `OpenAIEmbeddingsProducer` — `Math.toIntExact` is both
unnecessary (the `Long` overload of `GenAiUsage.of` exists) and risky (throws
`ArithmeticException` on overflow, which the catch block would treat as a
failed operation).
```suggestion
private static Long toTokenCount(long tokens) {
return tokens;
}
```
##########
components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIModerationProducer.java:
##########
@@ -87,19 +93,32 @@ private void processInternal(Exchange exchange) throws
Exception {
paramsBuilder.inputOfStrings(inputs);
}
- ModerationCreateResponse response = getEndpoint().getClient()
- .moderations().create(paramsBuilder.build());
+ GenAiObservationContext observationContext =
GenAiObservationContext.builder()
+ .operationName(GenAiOperationName.MODERATION)
+ .system("openai")
+ .requestModel(model)
+ .componentScheme("openai")
+ .build();
+ GenAiObservation observation = GenAiObservability.start(exchange,
observationContext);
+ ModerationCreateResponse response;
+ try {
+ response =
getEndpoint().getClient().moderations().create(paramsBuilder.build());
+ observation.recordSuccess(GenAiUsage.of(null, null, null,
response.model()));
+ } catch (Exception e) {
+ GenAiErrorSupport.apply(exchange, e);
+ observation.recordError(e);
+ throw e;
+ } finally {
+ observation.close();
Review Comment:
The comment that was here ("this operation is used to gate untrusted
content, so a missing verdict must fail the exchange...") explained a security
invariant — *why* a mismatched result count throws rather than silently
proceeding. Worth keeping; it's not redundant with the observability changes.
##########
components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIResponsesProducer.java:
##########
@@ -136,28 +142,89 @@ private void processInternal(Exchange exchange) throws
Exception {
Class<?> responseClass = resolveOutputClass(in, outputClass);
if (responseClass != null) {
- processStructured(exchange, config, paramsBuilder, responseClass);
+ processStructured(exchange, config, paramsBuilder, responseClass,
model);
return;
}
if (ObjectHelper.isNotEmpty(jsonSchema)) {
OpenAIResponsesSupport.applyJsonSchemaTextFormat(paramsBuilder,
jsonSchema);
}
ResponseCreateParams params = paramsBuilder.build();
- Response response =
getEndpoint().getClient().responses().create(params);
+ Response response = createResponse(exchange, model, params);
finishExchange(exchange, config, response,
OpenAIResponsesSupport.extractAssistantText(response));
}
private void processStructured(
Exchange exchange, OpenAIConfiguration config,
ResponseCreateParams.Builder paramsBuilder,
- Class<?> responseClass)
+ Class<?> responseClass, String model)
throws Exception {
StructuredResponseCreateParams<?> structuredParams =
paramsBuilder.text(responseClass).build();
- StructuredResponse<?> structured =
getEndpoint().getClient().responses().create(structuredParams);
- Response raw = structured.rawResponse();
+ Response raw = createStructuredResponse(exchange, model,
structuredParams);
finishExchange(exchange, config, raw,
OpenAIResponsesSupport.extractAssistantText(raw));
}
+ private Response createResponse(Exchange exchange, String model,
ResponseCreateParams params) throws Exception {
+ GenAiObservationContext observationContext =
GenAiObservationContext.builder()
+ .operationName(GenAiOperationName.CHAT)
+ .system("openai")
+ .requestModel(model)
+ .componentScheme("openai")
+ .build();
Review Comment:
`createResponse` and `createStructuredResponse` are nearly identical — same
observation context, same error handling, same `finally`. The only difference
is the SDK call. Consider extracting a common helper, e.g.:
```java
private Response observedCall(Exchange exchange, String model,
ThrowingSupplier<Response> call) throws Exception {
GenAiObservationContext ctx = GenAiObservationContext.builder()
.operationName(GenAiOperationName.CHAT)
.system("openai").requestModel(model)
.componentScheme("openai").build();
GenAiObservation observation = GenAiObservability.start(exchange, ctx);
try {
Response response = call.get();
recordResponseSuccess(observation, response);
return response;
} catch (Exception e) {
GenAiErrorSupport.apply(exchange, e);
observation.recordError(e);
throw e;
} finally {
observation.close();
}
}
```
Then `createResponse` becomes `observedCall(exchange, model, () ->
getEndpoint().getClient().responses().create(params))` and
`createStructuredResponse` becomes `observedCall(exchange, model, () ->
getEndpoint().getClient().responses().create(structuredParams).rawResponse())`.
Not blocking, but reduces ~40 lines of duplication to ~2.
--
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]
