Conversation
…ation, embeddings, multimodal, and streamRawPredict operations Implement all 5 placeholder operations in GoogleVertexAIProducer: - generateChatStreaming: uses generateContentStream for streaming Gemini responses - generateImage: uses generateImages for Imagen model image generation - generateEmbeddings: uses embedContent for single/batch text embeddings - generateMultimodal: uses Content/Part for text+image/video/audio input - streamRawPredict: uses streamRawPredictCallable for streaming partner models Added 10 new header constants for operation-specific parameters. Added 7 unit tests and 5 integration tests. Updated component documentation with examples for all new operations. Signed-off-by: Andrea Cosentino <ancosen@gmail.com>
…ation, embeddings, multimodal, and streamRawPredict operations Signed-off-by: Andrea Cosentino <ancosen@gmail.com>
|
🌟 Thank you for your contribution to the Apache Camel project! 🌟 🐫 Apache Camel Committers, please review the following items:
|
|
🧪 CI tested the following changed modules:
All tested modules (72 modules)
|
gnodet
left a comment
gnodet
left a comment
There was a problem hiding this comment.
Claude Code on behalf of Guillaume Nodet
Overall good work implementing the 5 missing operations. The documentation and integration tests are appreciated. However, there are several issues that need to be addressed before merging:
Critical Issues
-
Duplicate constant:
STREAMING_CHUNK_COUNT(line 91) and the newCHUNK_COUNTboth resolve to the same header nameCamelGoogleVertexAIChunkCount. This causes issues in the generated metadata (index 20 is missing in the generated JSON). One of them should be removed. -
Dead code in
generateChatStreaming: Theif ("chunks".equals(streamOutputMode)) / elsebranches do exactly the same thing. Either implement differentiated behavior for "chunks" mode or remove the branching. -
ServerStreamnot closed instreamRawPredict: UnlikegenerateChatStreamingwhich properly uses try-with-resources onResponseStream, thestreamRawPredictmethod iteratesServerStreamwithout closing it.
Minor Issues
- Inconsistent
@Metadatalabel onGENERATED_IMAGES: Uses"generateImage"while all other new constants use"producer generateImage". - Silently swallowed exceptions in
extractAnthropicStreamingTextContent: Thecatch (Exception e)block should at least log at TRACE level. - Unchecked cast without validation in
generateEmbeddings:(List<String>) bodycould fail with confusing ClassCastException. generateMultimodalbody fallback logic is hard to follow: The condition is complex, consider simplifying.- Hardcoded
us-east5fallback: Should be documented or made consistent with the existingrawPredictmethod.
| @Metadata(label = "producer generateChatStreaming", | ||
| description = "The number of streaming chunks received", javaType = "Integer") | ||
| public static final String CHUNK_COUNT = "CamelGoogleVertexAIChunkCount"; | ||
|
|
There was a problem hiding this comment.
Claude Code on behalf of Guillaume Nodet
This new CHUNK_COUNT constant has the same header value (CamelGoogleVertexAIChunkCount) as the existing STREAMING_CHUNK_COUNT on line 91 (visible just above in this diff). This creates a duplicate constant for the same header name, which causes issues in the generated metadata (index 20 is skipped in the generated JSON).
Since STREAMING_CHUNK_COUNT was introduced in the same development cycle and hasn't been released yet, I'd suggest removing STREAMING_CHUNK_COUNT entirely and keeping only CHUNK_COUNT. If it was already released, deprecate STREAMING_CHUNK_COUNT and point to CHUNK_COUNT.
| public static final String IMAGE_ASPECT_RATIO = "CamelGoogleVertexAIImageAspectRatio"; | ||
|
|
||
| @Metadata(label = "generateImage", | ||
| description = "The generated images from an image generation operation", |
There was a problem hiding this comment.
Claude Code on behalf of Guillaume Nodet
Inconsistent @Metadata label: this uses "generateImage" while all other new constants use "producer <operationName>" (e.g., "producer generateImage", "producer generateEmbeddings"). Should be "producer generateImage" for consistency.
This also shows up in the generated JSON where the group/label is "generateImage" without the "producer" prefix.
| Message message = getMessageForResponse(exchange); | ||
|
|
||
| if ("chunks".equals(streamOutputMode)) { | ||
| // In chunks mode, return the full accumulated text but include chunk count |
There was a problem hiding this comment.
Claude Code on behalf of Guillaume Nodet
Both branches of this if/else do exactly the same thing: message.setBody(fullText.toString()). The comment says "In chunks mode, return the full accumulated text but include chunk count" but the default mode also returns the same thing.
Either:
- Implement differentiated behavior (e.g., chunks mode returns
List<String>of individual chunks, or uses Camel streaming/reactive patterns), or - Remove the
streamOutputModecheck entirely since it's dead code.
As-is, this is misleading — users will set streamOutputMode=chunks expecting different behavior but get the exact same result.
| if (body instanceof List) { | ||
| // Batch embeddings | ||
| List<String> texts = (List<String>) body; | ||
| response = client.models.embedContent(modelId, texts, embedConfig); |
There was a problem hiding this comment.
Claude Code on behalf of Guillaume Nodet
Unchecked cast (List<String>) body — if the body is a List of non-String objects (e.g., List<Integer>), this will compile fine but throw a confusing ClassCastException later when the SDK tries to use the elements. Consider:
List<?> list = (List<?>) body;
List<String> texts = list.stream()
.map(Object::toString)
.collect(Collectors.toList());or at least validate the element types.
|
|
||
| // If body is byte[] and no media header was set, treat body as media | ||
| if (parts.isEmpty() || (prompt != null && mediaData == null && gcsUri == null)) { | ||
| Object body = exchange.getIn().getBody(); |
There was a problem hiding this comment.
Claude Code on behalf of Guillaume Nodet
This condition is hard to follow:
if (parts.isEmpty() || (prompt != null && mediaData == null && gcsUri == null))The second clause means "prompt was given but no media headers were set" — so we check if body can serve as media. But if parts.isEmpty() is true, it means no prompt AND no media headers, so we also check body as a fallback.
This would be clearer split into explicit cases:
// If no media was provided via headers, check if body contains media
if (mediaData == null && gcsUri == null) {
Object body = exchange.getIn().getBody();
if (body instanceof byte[] bodyBytes && mediaMimeType != null) {
parts.add(Part.fromBytes(bodyBytes, mediaMimeType));
} else if (body instanceof String bodyStr && prompt == null) {
parts.add(Part.fromText(bodyStr));
}
}| String location = config.getLocation(); | ||
| if ("global".equalsIgnoreCase(location)) { | ||
| location = "us-east5"; | ||
| } |
There was a problem hiding this comment.
Claude Code on behalf of Guillaume Nodet
Hardcoding "us-east5" as a fallback for "global" location is fragile. This behavior should be documented (either in the code with a comment explaining why, or in the component docs). Is there a reason us-central1 isn't used as the default like elsewhere? Partner models may have different regional availability, but the choice should be transparent to the user.
| .build(); | ||
|
|
||
| ServerStream<HttpBody> stream = predictionClient.streamRawPredictCallable().call(request); | ||
|
|
There was a problem hiding this comment.
Claude Code on behalf of Guillaume Nodet
Unlike generateChatStreaming() which properly uses try-with-resources on ResponseStream, the ServerStream<HttpBody> here is never closed. If the iteration is interrupted (e.g., by an exception during append), the underlying gRPC stream won't be cleaned up.
Consider wrapping in a try block or calling stream.cancel() in a finally block.
| } | ||
| } catch (Exception e) { | ||
| // Skip non-JSON or unparseable lines | ||
| } |
There was a problem hiding this comment.
Claude Code on behalf of Guillaume Nodet
Silently swallowing all exceptions with an empty catch block. If a line starts with data: but contains malformed JSON, this will be silently ignored which could lead to truncated text output with no indication of the problem.
At minimum, add LOG.trace("Skipping unparseable SSE line: {}", jsonStr, e); so there's some visibility when debugging.
…ation, embeddings, multimodal, and streamRawPredict operations Signed-off-by: Andrea Cosentino <ancosen@gmail.com>
|
It should be ok now. |
2 participants
Description
Target
mainbranch)Tracking
Apache Camel coding standards and style
mvn clean install -DskipTestslocally from root folder and I have committed all auto-generated changes.