This is an automated email from the ASF dual-hosted git repository.
wenjin272 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-agents.git
The following commit(s) were added to refs/heads/main by this push:
new da4553e80 [api][java][python] Add raw-bytes factories for media blocks
(#1207)
da4553e80 is described below
commit da4553e80b0715c7000d2a5c637b52cee5ac4e93
Author: Wenjin Xie <[email protected]>
AuthorDate: Thu Oct 8 14:14:55 2026 +0800
[api][java][python] Add raw-bytes factories for media blocks (#1207)
Generated-by: Codex CLI 0.153.4 (GPT-6)
Co-authored-by: Codex <[email protected]>
---
.../flink/agents/api/chat/messages/AudioBlock.java | 15 ++-
.../agents/api/chat/messages/Base64Source.java | 9 ++
.../agents/api/chat/messages/DocumentBlock.java | 15 ++-
.../flink/agents/api/chat/messages/ImageBlock.java | 15 ++-
.../flink/agents/api/chat/messages/MediaBlock.java | 10 +-
.../flink/agents/api/chat/messages/VideoBlock.java | 15 ++-
.../agents/api/CrossLanguageEventSnapshotTest.java | 41 ++++++++
.../api/chat/messages/MediaBlockBytesTest.java | 113 +++++++++++++++++++++
docs/content/docs/development/chat_models.md | 43 +++++++-
.../java/raw_media_chat_request_event.json | 42 ++++++++
.../python/raw_media_chat_request_event.json | 63 ++++++++++++
.../OpenAIChatCompletionsMultimodalTest.java | 31 ++++++
python/flink_agents/api/chat_message.py | 27 ++++-
python/flink_agents/api/tests/test_chat_message.py | 71 +++++++++++++
.../tests/test_cross_language_event_snapshots.py | 50 ++++++++-
.../openai/tests/test_openai_multimodal.py | 30 ++++++
16 files changed, 574 insertions(+), 16 deletions(-)
diff --git
a/api/src/main/java/org/apache/flink/agents/api/chat/messages/AudioBlock.java
b/api/src/main/java/org/apache/flink/agents/api/chat/messages/AudioBlock.java
index 503a9b611..41d78c32d 100644
---
a/api/src/main/java/org/apache/flink/agents/api/chat/messages/AudioBlock.java
+++
b/api/src/main/java/org/apache/flink/agents/api/chat/messages/AudioBlock.java
@@ -48,7 +48,20 @@ public final class AudioBlock extends MediaBlock {
super(mediaType, source, name, sizeBytes, sha256);
}
- /** Creates an audio block carrying an inline base64 payload. */
+ /**
+ * Creates a block from non-empty raw bytes, encoded as standard Base64
without line wrapping.
+ * The caller's array is not retained; subsequent changes to it do not
affect the block.
+ *
+ * @throws IllegalArgumentException if the media type or data is null or
empty
+ */
+ public static AudioBlock fromBytes(String mediaType, byte[] data) {
+ return new AudioBlock(mediaType, Base64Source.fromBytes(data), null,
null, null);
+ }
+
+ /**
+ * Creates a block from a non-empty, already Base64-encoded string. The
string is preserved
+ * without encoding or validating Base64 syntax.
+ */
public static AudioBlock fromBase64(String mediaType, String data) {
return new AudioBlock(mediaType, new Base64Source(data), null, null,
null);
}
diff --git
a/api/src/main/java/org/apache/flink/agents/api/chat/messages/Base64Source.java
b/api/src/main/java/org/apache/flink/agents/api/chat/messages/Base64Source.java
index 124036139..f279d8887 100644
---
a/api/src/main/java/org/apache/flink/agents/api/chat/messages/Base64Source.java
+++
b/api/src/main/java/org/apache/flink/agents/api/chat/messages/Base64Source.java
@@ -22,6 +22,7 @@ import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import java.util.Base64;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Objects;
@@ -46,6 +47,14 @@ public final class Base64Source extends MediaSource {
return data;
}
+ /** Encodes raw bytes immediately, without retaining the caller's array. */
+ static Base64Source fromBytes(byte[] data) {
+ if (data == null || data.length == 0) {
+ throw new IllegalArgumentException("A raw media payload requires
non-empty bytes.");
+ }
+ return new Base64Source(Base64.getEncoder().encodeToString(data));
+ }
+
@Override
public String getType() {
return "base64";
diff --git
a/api/src/main/java/org/apache/flink/agents/api/chat/messages/DocumentBlock.java
b/api/src/main/java/org/apache/flink/agents/api/chat/messages/DocumentBlock.java
index 6991355d0..87c57b733 100644
---
a/api/src/main/java/org/apache/flink/agents/api/chat/messages/DocumentBlock.java
+++
b/api/src/main/java/org/apache/flink/agents/api/chat/messages/DocumentBlock.java
@@ -48,7 +48,20 @@ public final class DocumentBlock extends MediaBlock {
super(mediaType, source, name, sizeBytes, sha256);
}
- /** Creates a document block carrying an inline base64 payload. */
+ /**
+ * Creates a block from non-empty raw bytes, encoded as standard Base64
without line wrapping.
+ * The caller's array is not retained; subsequent changes to it do not
affect the block.
+ *
+ * @throws IllegalArgumentException if the media type or data is null or
empty
+ */
+ public static DocumentBlock fromBytes(String mediaType, byte[] data) {
+ return new DocumentBlock(mediaType, Base64Source.fromBytes(data),
null, null, null);
+ }
+
+ /**
+ * Creates a block from a non-empty, already Base64-encoded string. The
string is preserved
+ * without encoding or validating Base64 syntax.
+ */
public static DocumentBlock fromBase64(String mediaType, String data) {
return new DocumentBlock(mediaType, new Base64Source(data), null,
null, null);
}
diff --git
a/api/src/main/java/org/apache/flink/agents/api/chat/messages/ImageBlock.java
b/api/src/main/java/org/apache/flink/agents/api/chat/messages/ImageBlock.java
index 9816f6ddb..7f01072b0 100644
---
a/api/src/main/java/org/apache/flink/agents/api/chat/messages/ImageBlock.java
+++
b/api/src/main/java/org/apache/flink/agents/api/chat/messages/ImageBlock.java
@@ -48,7 +48,20 @@ public final class ImageBlock extends MediaBlock {
super(mediaType, source, name, sizeBytes, sha256);
}
- /** Creates an image block carrying an inline base64 payload. */
+ /**
+ * Creates a block from non-empty raw bytes, encoded as standard Base64
without line wrapping.
+ * The caller's array is not retained; subsequent changes to it do not
affect the block.
+ *
+ * @throws IllegalArgumentException if the media type or data is null or
empty
+ */
+ public static ImageBlock fromBytes(String mediaType, byte[] data) {
+ return new ImageBlock(mediaType, Base64Source.fromBytes(data), null,
null, null);
+ }
+
+ /**
+ * Creates a block from a non-empty, already Base64-encoded string. The
string is preserved
+ * without encoding or validating Base64 syntax.
+ */
public static ImageBlock fromBase64(String mediaType, String data) {
return new ImageBlock(mediaType, new Base64Source(data), null, null,
null);
}
diff --git
a/api/src/main/java/org/apache/flink/agents/api/chat/messages/MediaBlock.java
b/api/src/main/java/org/apache/flink/agents/api/chat/messages/MediaBlock.java
index 28d351aab..a462f7b5e 100644
---
a/api/src/main/java/org/apache/flink/agents/api/chat/messages/MediaBlock.java
+++
b/api/src/main/java/org/apache/flink/agents/api/chat/messages/MediaBlock.java
@@ -32,11 +32,11 @@ import java.util.Objects;
* (RFC 6838; historically called a MIME type), and the payload location is a
typed {@link
* MediaSource}.
*
- * <p>Media blocks are immutable, and every construction path — the {@code
fromBase64}/{@code
- * fromUrl} factories, the full constructors, and Jackson deserialization —
runs the same
- * validation. Which kind of source a block carries is structural: there is
exactly one {@code
- * source}, and its {@code type} discriminator says whether it is an inline
{@link Base64Source} or
- * an externally managed {@link UrlSource}.
+ * <p>Media blocks are immutable, and every construction path — the {@code
fromBytes}/{@code
+ * fromBase64}/{@code fromUrl} factories, the full constructors, and Jackson
deserialization — runs
+ * the same validation. Which kind of source a block carries is structural:
there is exactly one
+ * {@code source}, and its {@code type} discriminator says whether it is an
inline {@link
+ * Base64Source} or an externally managed {@link UrlSource}.
*
* <p>The optional {@code name}/{@code sizeBytes}/{@code sha256} metadata also
serves the Event Log,
* which records media metadata instead of payload bytes — see {@link
#sanitize()}.
diff --git
a/api/src/main/java/org/apache/flink/agents/api/chat/messages/VideoBlock.java
b/api/src/main/java/org/apache/flink/agents/api/chat/messages/VideoBlock.java
index 56e9f9e7a..1b9083286 100644
---
a/api/src/main/java/org/apache/flink/agents/api/chat/messages/VideoBlock.java
+++
b/api/src/main/java/org/apache/flink/agents/api/chat/messages/VideoBlock.java
@@ -48,7 +48,20 @@ public final class VideoBlock extends MediaBlock {
super(mediaType, source, name, sizeBytes, sha256);
}
- /** Creates a video block carrying an inline base64 payload. */
+ /**
+ * Creates a block from non-empty raw bytes, encoded as standard Base64
without line wrapping.
+ * The caller's array is not retained; subsequent changes to it do not
affect the block.
+ *
+ * @throws IllegalArgumentException if the media type or data is null or
empty
+ */
+ public static VideoBlock fromBytes(String mediaType, byte[] data) {
+ return new VideoBlock(mediaType, Base64Source.fromBytes(data), null,
null, null);
+ }
+
+ /**
+ * Creates a block from a non-empty, already Base64-encoded string. The
string is preserved
+ * without encoding or validating Base64 syntax.
+ */
public static VideoBlock fromBase64(String mediaType, String data) {
return new VideoBlock(mediaType, new Base64Source(data), null, null,
null);
}
diff --git
a/api/src/test/java/org/apache/flink/agents/api/CrossLanguageEventSnapshotTest.java
b/api/src/test/java/org/apache/flink/agents/api/CrossLanguageEventSnapshotTest.java
index 475669570..79563880b 100644
---
a/api/src/test/java/org/apache/flink/agents/api/CrossLanguageEventSnapshotTest.java
+++
b/api/src/test/java/org/apache/flink/agents/api/CrossLanguageEventSnapshotTest.java
@@ -21,8 +21,12 @@ package org.apache.flink.agents.api;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.agents.api.agents.OutputSchema;
+import org.apache.flink.agents.api.chat.messages.AudioBlock;
import org.apache.flink.agents.api.chat.messages.ChatMessage;
+import org.apache.flink.agents.api.chat.messages.DocumentBlock;
+import org.apache.flink.agents.api.chat.messages.ImageBlock;
import org.apache.flink.agents.api.chat.messages.MessageRole;
+import org.apache.flink.agents.api.chat.messages.VideoBlock;
import org.apache.flink.agents.api.context.MemoryObject;
import org.apache.flink.agents.api.context.MemoryRef;
import org.apache.flink.agents.api.event.AgentRunBeginEvent;
@@ -195,6 +199,43 @@ class CrossLanguageEventSnapshotTest {
// ── ChatRequestEvent ───────────────────────────────────────────────────
+ private static ChatRequestEvent buildRawMediaChatRequestEvent() {
+ byte[] data = {(byte) 0xff, 0, (byte) 0x80, (byte) 0xfb};
+ Map<String, Object> attrs = new LinkedHashMap<>();
+ attrs.put("model", "test-model");
+ attrs.put(
+ "messages",
+ List.of(
+ ChatMessage.user(
+ List.of(
+ ImageBlock.fromBytes("image/png",
data),
+ AudioBlock.fromBytes("audio/wav",
data),
+ VideoBlock.fromBytes("video/mp4",
data),
+
DocumentBlock.fromBytes("application/pdf", data)))));
+ return new ChatRequestEvent(FIXED_EVENT_ID, attrs);
+ }
+
+ @Test
+ void regenerateRawMediaChatRequestEventJavaSnapshot() throws Exception {
+ assumeTrue(regenerateRequested(), "Set -Dregenerate.snapshots=true to
refresh.");
+ writeJavaSnapshot("raw_media_chat_request_event.json",
buildRawMediaChatRequestEvent());
+ }
+
+ @Test
+ void rawMediaChatRequestEventJavaSnapshotIsStable() throws Exception {
+ assertJavaSnapshotStable(
+ "raw_media_chat_request_event.json",
buildRawMediaChatRequestEvent());
+ }
+
+ @Test
+ void javaCanDeserializeRawMediaChatRequestEventFromPythonSnapshot() throws
Exception {
+ ChatRequestEvent event =
+
ChatRequestEvent.fromEvent(readPythonSnapshot("raw_media_chat_request_event.json"));
+ assertEquals(
+
buildRawMediaChatRequestEvent().getMessages().get(0).getBlocks(),
+ event.getMessages().get(0).getBlocks());
+ }
+
private static ChatRequestEvent buildChatRequestEvent() {
Map<String, Object> attrs = new LinkedHashMap<>();
attrs.put("model", "test-model");
diff --git
a/api/src/test/java/org/apache/flink/agents/api/chat/messages/MediaBlockBytesTest.java
b/api/src/test/java/org/apache/flink/agents/api/chat/messages/MediaBlockBytesTest.java
new file mode 100644
index 000000000..c24590c34
--- /dev/null
+++
b/api/src/test/java/org/apache/flink/agents/api/chat/messages/MediaBlockBytesTest.java
@@ -0,0 +1,113 @@
+/*
+ * 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.flink.agents.api.chat.messages;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.Arrays;
+import java.util.Base64;
+import java.util.List;
+import java.util.Map;
+import java.util.function.BiFunction;
+import java.util.stream.Stream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Raw-byte factory contracts shared by every media modality. */
+class MediaBlockBytesTest {
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ static Stream<BiFunction<String, byte[], MediaBlock>> factories() {
+ return Stream.of(
+ ImageBlock::fromBytes,
+ AudioBlock::fromBytes,
+ VideoBlock::fromBytes,
+ DocumentBlock::fromBytes);
+ }
+
+ @ParameterizedTest
+ @MethodSource("factories")
+ void preservesBinaryPayloadAndExistingWireShape(BiFunction<String, byte[],
MediaBlock> factory)
+ throws Exception {
+ byte[] allBytes = new byte[256];
+ for (int i = 0; i < allBytes.length; i++) {
+ allBytes[i] = (byte) i;
+ }
+ // Exercise both padding lengths, no padding, non-UTF-8 data and no
line wrapping.
+ for (byte[] data :
+ List.of(
+ new byte[] {(byte) 0xff},
+ new byte[] {(byte) 0xff, 0},
+ new byte[] {(byte) 0xff, 0, (byte) 0x80},
+ allBytes)) {
+ MediaBlock block = factory.apply("application/octet-stream", data);
+ String encoded = ((Base64Source) block.getSource()).getData();
+ assertThat(encoded).doesNotContain("\n", "\r");
+ assertThat(Base64.getDecoder().decode(encoded)).isEqualTo(data);
+ assertThat(encoded).hasSize(4 * ((data.length + 2) / 3));
+ assertThat(block.getSource().getSizeBytes()).isEqualTo((long)
data.length);
+ assertThat(block.getSizeBytes()).isNull();
+ ChatMessage message = ChatMessage.user(List.of(block));
+ Map<String, Object> expected =
+ Map.of(
+ "type",
+ block.getType(),
+ "media_type",
+ "application/octet-stream",
+ "source",
+ Map.of("type", "base64", "data", encoded));
+ JsonNode serialized = MAPPER.valueToTree(block);
+ assertThat(serialized).isEqualTo(MAPPER.valueToTree(expected));
+ assertThat(
+
MAPPER.readValue(MAPPER.writeValueAsString(message), ChatMessage.class)
+ .getBlocks())
+ .containsExactly(block);
+ ChatMessage fromMap = new ChatMessage();
+ fromMap.setBlocksFromMaps(List.of(expected));
+ assertThat(fromMap.getBlocks()).containsExactly(block);
+ assertThat(block.toString()).doesNotContain(encoded);
+
assertThat(MAPPER.writeValueAsString(block.sanitize())).doesNotContain(encoded);
+ }
+ }
+
+ @ParameterizedTest
+ @MethodSource("factories")
+ void doesNotRetainCallerArray(BiFunction<String, byte[], MediaBlock>
factory) {
+ byte[] data = {(byte) 0xff, 0, (byte) 0x80};
+ MediaBlock block = factory.apply("application/octet-stream", data);
+ Arrays.fill(data, (byte) 1);
+ assertThat(((Base64Source)
block.getSource()).getData()).isEqualTo("/wCA");
+ }
+
+ @ParameterizedTest
+ @MethodSource("factories")
+ void rejectsMissingDataAndMediaType(BiFunction<String, byte[], MediaBlock>
factory) {
+ assertThatThrownBy(() -> factory.apply("image/png", null))
+ .isInstanceOf(IllegalArgumentException.class);
+ assertThatThrownBy(() -> factory.apply("image/png", new byte[0]))
+ .isInstanceOf(IllegalArgumentException.class);
+ assertThatThrownBy(() -> factory.apply(null, new byte[] {1}))
+ .isInstanceOf(IllegalArgumentException.class);
+ assertThatThrownBy(() -> factory.apply("", new byte[] {1}))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+}
diff --git a/docs/content/docs/development/chat_models.md
b/docs/content/docs/development/chat_models.md
index ab540e41c..b4877c924 100644
--- a/docs/content/docs/development/chat_models.md
+++ b/docs/content/docs/development/chat_models.md
@@ -92,11 +92,44 @@ ChatMessage message = ChatMessage.user(List.of(
{{< /tabs >}}
Each media block requires a MIME type such as `image/png` and a media source.
-Use `from_url` (Python) or `fromUrl` (Java) for an external URL, as above. For
-inline data, use `from_base64(media_type, data)` or `fromBase64(mediaType,
data)`
-with an already Base64-encoded string. The framework does not encode the data
or
-verify that its contents match the declared media type. Choose a concrete block
-such as `ImageBlock`, rather than constructing the shared `MediaBlock` base
class.
+Use `from_url` (Python) or `fromUrl` (Java) for an externally managed URL, as
above.
+For inline data, all four media block types provide two separate entry points:
+
+- `from_bytes(media_type, data)` (Python) or `fromBytes(mediaType, data)`
(Java)
+ accepts non-empty raw `bytes` or `byte[]` and encodes them as standard Base64
+ without line wrapping at construction. Java does not retain the caller's
array.
+- `from_base64(media_type, data)` or `fromBase64(mediaType, data)` accepts a
+ non-empty, already Base64-encoded string. It preserves the string without
+ encoding it again or validating its Base64 syntax.
+
+{{< tabs "Inline media" >}}
+{{< tab "Python" >}}
+```python
+from pathlib import Path
+from flink_agents.api.chat_message import ImageBlock
+
+image = ImageBlock.from_bytes("image/png", Path("image.png").read_bytes(),
name="image.png")
+# If a service already returned Base64, pass that string directly.
+encoded_image = ImageBlock.from_base64("image/png", base64_from_service)
+```
+{{< /tab >}}
+{{< tab "Java" >}}
+```java
+import java.nio.file.Files;
+import java.nio.file.Path;
+import org.apache.flink.agents.api.chat.messages.ImageBlock;
+
+ImageBlock image = ImageBlock.fromBytes("image/png",
Files.readAllBytes(Path.of("image.png")));
+// If a service already returned Base64, pass that string directly.
+ImageBlock encodedImage = ImageBlock.fromBase64("image/png",
base64FromService);
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+The framework does not verify that the contents match the declared media type.
+Choose a concrete block such as `ImageBlock`, rather than constructing the
shared
+`MediaBlock` base class. Both inline factories produce the same Base64 source
+representation for serialization and provider conversion.
Media blocks also accept optional metadata: `name` for a file name,
`size_bytes`
for the media size in bytes, and `sha256` for a content checksum.
diff --git
a/e2e-test/cross-language-event-snapshots/java/raw_media_chat_request_event.json
b/e2e-test/cross-language-event-snapshots/java/raw_media_chat_request_event.json
new file mode 100644
index 000000000..9c6e14119
--- /dev/null
+++
b/e2e-test/cross-language-event-snapshots/java/raw_media_chat_request_event.json
@@ -0,0 +1,42 @@
+{
+ "id" : "00000000-0000-0000-0000-000000000001",
+ "attributes" : {
+ "messages" : [ {
+ "role" : "user",
+ "blocks" : [ {
+ "type" : "image",
+ "media_type" : "image/png",
+ "source" : {
+ "type" : "base64",
+ "data" : "/wCA+w=="
+ }
+ }, {
+ "type" : "audio",
+ "media_type" : "audio/wav",
+ "source" : {
+ "type" : "base64",
+ "data" : "/wCA+w=="
+ }
+ }, {
+ "type" : "video",
+ "media_type" : "video/mp4",
+ "source" : {
+ "type" : "base64",
+ "data" : "/wCA+w=="
+ }
+ }, {
+ "type" : "document",
+ "media_type" : "application/pdf",
+ "source" : {
+ "type" : "base64",
+ "data" : "/wCA+w=="
+ }
+ } ],
+ "tool_calls" : [ ],
+ "extra_args" : { }
+ } ],
+ "model" : "test-model"
+ },
+ "attachments" : { },
+ "type" : "_chat_request_event"
+}
diff --git
a/e2e-test/cross-language-event-snapshots/python/raw_media_chat_request_event.json
b/e2e-test/cross-language-event-snapshots/python/raw_media_chat_request_event.json
new file mode 100644
index 000000000..ac48aa92a
--- /dev/null
+++
b/e2e-test/cross-language-event-snapshots/python/raw_media_chat_request_event.json
@@ -0,0 +1,63 @@
+{
+ "id": "00000000-0000-0000-0000-000000000001",
+ "type": "_chat_request_event",
+ "attributes": {
+ "model": "test-model",
+ "messages": [
+ {
+ "role": "user",
+ "blocks": [
+ {
+ "media_type": "image/png",
+ "source": {
+ "type": "base64",
+ "data": "/wCA+w=="
+ },
+ "name": null,
+ "size_bytes": null,
+ "sha256": null,
+ "type": "image"
+ },
+ {
+ "media_type": "audio/wav",
+ "source": {
+ "type": "base64",
+ "data": "/wCA+w=="
+ },
+ "name": null,
+ "size_bytes": null,
+ "sha256": null,
+ "type": "audio"
+ },
+ {
+ "media_type": "video/mp4",
+ "source": {
+ "type": "base64",
+ "data": "/wCA+w=="
+ },
+ "name": null,
+ "size_bytes": null,
+ "sha256": null,
+ "type": "video"
+ },
+ {
+ "media_type": "application/pdf",
+ "source": {
+ "type": "base64",
+ "data": "/wCA+w=="
+ },
+ "name": null,
+ "size_bytes": null,
+ "sha256": null,
+ "type": "document"
+ }
+ ],
+ "tool_calls": [],
+ "extra_args": {}
+ }
+ ],
+ "prompt_args": {},
+ "output_schema": null
+ },
+ "attachments": {}
+}
diff --git
a/integrations/chat-models/openai/src/test/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAIChatCompletionsMultimodalTest.java
b/integrations/chat-models/openai/src/test/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAIChatCompletionsMultimodalTest.java
index be6266a0e..78549c3ee 100644
---
a/integrations/chat-models/openai/src/test/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAIChatCompletionsMultimodalTest.java
+++
b/integrations/chat-models/openai/src/test/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAIChatCompletionsMultimodalTest.java
@@ -54,6 +54,37 @@ class OpenAIChatCompletionsMultimodalTest {
.valueToTree(OpenAIChatCompletionsUtils.convertToOpenAIMessage(message));
}
+ @Test
+ void rawBytesPreservedInProviderContentParts() {
+ byte[] data = {(byte) 0xff, 0, (byte) 0x80, (byte) 0xfb};
+ String encoded = "/wCA+w==";
+ ChatMessage raw =
+ ChatMessage.user(
+ List.of(
+ ImageBlock.fromBytes("image/png", data),
+ AudioBlock.fromBytes("audio/wav", data),
+ DocumentBlock.fromBytes("application/pdf",
data)));
+ ChatMessage existing =
+ ChatMessage.user(
+ List.of(
+ ImageBlock.fromBase64("image/png", encoded),
+ AudioBlock.fromBase64("audio/wav", encoded),
+ DocumentBlock.fromBase64("application/pdf",
encoded)));
+ assertThat(wire(raw)).isEqualTo(wire(existing));
+ JsonNode parts = wire(raw).get("content");
+ assertThat(parts.get(0).at("/image_url/url").asText())
+ .isEqualTo("data:image/png;base64," + encoded);
+
assertThat(parts.get(1).at("/input_audio/data").asText()).isEqualTo(encoded);
+ assertThat(parts.get(2).at("/file/file_data").asText())
+ .isEqualTo("data:application/pdf;base64," + encoded);
+ assertThatThrownBy(
+ () ->
+ wire(
+ ChatMessage.user(
+
List.of(VideoBlock.fromBytes("video/mp4", data)))))
+ .isInstanceOf(UnsupportedContentBlockException.class);
+ }
+
@Test
@DisplayName("A text-only user message keeps plain string content")
void testTextOnlyUserMessageKeepsStringContent() {
diff --git a/python/flink_agents/api/chat_message.py
b/python/flink_agents/api/chat_message.py
index 7563b2add..171ed180d 100644
--- a/python/flink_agents/api/chat_message.py
+++ b/python/flink_agents/api/chat_message.py
@@ -15,6 +15,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#################################################################################
+import base64
from enum import Enum
from typing import Any, Dict, List, Literal, Sequence
@@ -135,9 +136,33 @@ class MediaBlock(BaseModel):
@classmethod
def from_base64(cls, media_type: str, data: str, **kwargs: Any) -> Self:
- """Create a block carrying an inline base64 payload."""
+ """Create a block from a non-empty, already Base64-encoded string.
+
+ The string is preserved without encoding or validating Base64 syntax.
+ """
return cls(media_type=media_type, source=Base64Source(data=data),
**kwargs)
+ @classmethod
+ def from_bytes(cls, media_type: str, data: bytes, **kwargs: Any) -> Self:
+ """Create a block from non-empty raw bytes using standard Base64.
+
+ Encoding happens once at construction, without line wrapping. Optional
+ metadata is passed through just as for ``from_base64``.
+
+ Raises:
+ TypeError: If data is not bytes.
+ ValueError: If data is empty or media fields are invalid.
+ """
+ if not isinstance(data, bytes):
+ msg = "A raw media payload must be bytes."
+ raise TypeError(msg)
+ if not data:
+ msg = "A raw media payload must not be empty."
+ raise ValueError(msg)
+ return cls.from_base64(
+ media_type, base64.b64encode(data).decode("ascii"), **kwargs
+ )
+
@classmethod
def from_url(cls, media_type: str, url: str, **kwargs: Any) -> Self:
"""Create a block referencing an externally managed URL or file URI."""
diff --git a/python/flink_agents/api/tests/test_chat_message.py
b/python/flink_agents/api/tests/test_chat_message.py
index ca8247057..5fedf59b0 100644
--- a/python/flink_agents/api/tests/test_chat_message.py
+++ b/python/flink_agents/api/tests/test_chat_message.py
@@ -15,6 +15,8 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#################################################################################
+import base64
+
import pytest
from pydantic import ValidationError
@@ -24,6 +26,7 @@ from flink_agents.api.chat_message import (
ChatMessage,
DocumentBlock,
ImageBlock,
+ MediaBlock,
MessageRole,
TextBlock,
UrlSource,
@@ -318,3 +321,71 @@ def test_factories_and_text_projection() -> None:
empty.set_text("replaced")
assert empty.text == "replaced"
assert str(ChatMessage.user("hi")) == "user: hi"
+
+
+_MEDIA_TYPES = [
+ (ImageBlock, "image/png"),
+ (AudioBlock, "audio/wav"),
+ (VideoBlock, "video/mp4"),
+ (DocumentBlock, "application/pdf"),
+]
+
+
[email protected](("block_type", "media_type"), _MEDIA_TYPES)
[email protected](
+ "data", [b"\xff", b"\xff\x00", b"\xff\x00\x80", bytes(range(256))]
+)
+def test_from_bytes_preserves_binary_and_wire_contract(
+ block_type: type[MediaBlock], media_type: str, data: bytes
+) -> None:
+ """Exercise padding, non-UTF-8 bytes, and payloads longer than a MIME
line."""
+ block = block_type.from_bytes(media_type, data)
+ encoded = block.source.data
+ assert "\n" not in encoded
+ assert "\r" not in encoded
+ assert base64.b64decode(encoded, validate=True) == data
+ assert len(encoded) == 4 * ((len(data) + 2) // 3)
+ assert block.source.size_bytes == len(data)
+ assert block.size_bytes is None
+ assert block == block_type.from_base64(
+ media_type, base64.b64encode(data).decode("ascii")
+ )
+ message = ChatMessage.user([block])
+ assert ChatMessage.model_validate_json(message.model_dump_json()) ==
message
+ assert message.model_dump(mode="json", exclude_none=True)["blocks"] == [
+ {
+ "type": block.type,
+ "media_type": media_type,
+ "source": {"type": "base64", "data": encoded},
+ }
+ ]
+ assert encoded not in repr(message)
+ assert encoded not in str(block)
+
+
[email protected](("block_type", "media_type"), _MEDIA_TYPES)
[email protected](
+ "data", [None, "aGk=", 3, [1, 2], bytearray(b"hi"), memoryview(b"hi")]
+)
+def test_from_bytes_rejects_non_bytes(block_type, media_type, data) -> None:
+ with pytest.raises(TypeError, match="must be bytes"):
+ block_type.from_bytes(media_type, data)
+
+
[email protected](("block_type", "media_type"), _MEDIA_TYPES)
+def test_from_bytes_validation_and_metadata(block_type, media_type) -> None:
+ with pytest.raises(ValueError, match="must not be empty"):
+ block_type.from_bytes(media_type, b"")
+ for invalid_media_type in (None, "", 3):
+ with pytest.raises(ValidationError):
+ block_type.from_bytes(invalid_media_type, b"hi")
+ with pytest.raises(ValidationError):
+ block_type.from_bytes(media_type, b"hi", size_bytes=-1)
+ metadata = {"name": "sample", "size_bytes": 2, "sha256": "caller-supplied"}
+ assert block_type.from_bytes(
+ media_type, b"hi", **metadata
+ ) == block_type.from_base64(media_type, "aGk=", **metadata)
+ # Existing Base64 factories neither encode again nor start validating
syntax.
+ assert (
+ block_type.from_base64(media_type, "not base64!").source.data == "not
base64!"
+ )
diff --git
a/python/flink_agents/api/tests/test_cross_language_event_snapshots.py
b/python/flink_agents/api/tests/test_cross_language_event_snapshots.py
index c14bf0f91..6cc98608a 100644
--- a/python/flink_agents/api/tests/test_cross_language_event_snapshots.py
+++ b/python/flink_agents/api/tests/test_cross_language_event_snapshots.py
@@ -25,7 +25,14 @@ from uuid import UUID
import pytest
-from flink_agents.api.chat_message import ChatMessage, MessageRole
+from flink_agents.api.chat_message import (
+ AudioBlock,
+ ChatMessage,
+ DocumentBlock,
+ ImageBlock,
+ MessageRole,
+ VideoBlock,
+)
from flink_agents.api.events.chat_event import ChatRequestEvent,
ChatResponseEvent
from flink_agents.api.events.context_retrieval_event import (
ContextRetrievalRequestEvent,
@@ -141,6 +148,47 @@ def
test_python_can_deserialize_output_event_from_java_snapshot() -> None:
# ── ChatRequestEvent ────────────────────────────────────────────────────
+def _build_raw_media_chat_request_event() -> ChatRequestEvent:
+ data = b"\xff\x00\x80\xfb"
+ event = ChatRequestEvent(
+ model="test-model",
+ messages=[
+ ChatMessage.user(
+ [
+ ImageBlock.from_bytes("image/png", data),
+ AudioBlock.from_bytes("audio/wav", data),
+ VideoBlock.from_bytes("video/mp4", data),
+ DocumentBlock.from_bytes("application/pdf", data),
+ ]
+ )
+ ],
+ )
+ return _force_id(event, _FIXED_EVENT_ID)
+
+
+def test_regenerate_raw_media_chat_request_event_python_snapshot() -> None:
+ if not _regenerate_enabled():
+ pytest.skip("Set REGENERATE_SNAPSHOTS=1 to refresh.")
+ _write_python_snapshot(
+ "raw_media_chat_request_event.json",
_build_raw_media_chat_request_event()
+ )
+
+
+def test_raw_media_chat_request_event_python_snapshot_is_stable() -> None:
+ _assert_python_snapshot_stable(
+ "raw_media_chat_request_event.json",
_build_raw_media_chat_request_event()
+ )
+
+
+def
test_python_can_deserialize_raw_media_chat_request_event_from_java_snapshot()
-> (
+ None
+):
+ event = ChatRequestEvent.from_event(
+ _read_java_snapshot("raw_media_chat_request_event.json")
+ )
+ assert event.messages == _build_raw_media_chat_request_event().messages
+
+
def _build_chat_request_event() -> ChatRequestEvent:
event = ChatRequestEvent(
model="test-model",
diff --git
a/python/flink_agents/integrations/chat_models/openai/tests/test_openai_multimodal.py
b/python/flink_agents/integrations/chat_models/openai/tests/test_openai_multimodal.py
index a2895cecf..a2f6c924e 100644
---
a/python/flink_agents/integrations/chat_models/openai/tests/test_openai_multimodal.py
+++
b/python/flink_agents/integrations/chat_models/openai/tests/test_openai_multimodal.py
@@ -141,3 +141,33 @@ def test_media_outside_user_messages_fails(role:
MessageRole) -> None:
with pytest.raises(UnsupportedContentBlockError, match="only user
messages"):
convert_to_openai_message(message)
+
+
+def test_raw_bytes_preserved_in_provider_content_parts() -> None:
+ """No extra encoding or text decoding occurs when converting binary
media."""
+ data = b"\xff\x00\x80\xfb"
+ raw = ChatMessage.user(
+ [
+ ImageBlock.from_bytes("image/png", data),
+ AudioBlock.from_bytes("audio/wav", data),
+ DocumentBlock.from_bytes("application/pdf", data),
+ ]
+ )
+ encoded = "/wCA+w=="
+ existing = ChatMessage.user(
+ [
+ ImageBlock.from_base64("image/png", encoded),
+ AudioBlock.from_base64("audio/wav", encoded),
+ DocumentBlock.from_base64("application/pdf", encoded),
+ ]
+ )
+ converted = convert_to_openai_message(raw)
+ assert converted == convert_to_openai_message(existing)
+ parts = converted["content"]
+ assert parts[0]["image_url"]["url"] == f"data:image/png;base64,{encoded}"
+ assert parts[1]["input_audio"]["data"] == encoded
+ assert parts[2]["file"]["file_data"] ==
f"data:application/pdf;base64,{encoded}"
+ with pytest.raises(UnsupportedContentBlockError):
+ convert_to_openai_message(
+ ChatMessage.user([VideoBlock.from_bytes("video/mp4", data)])
+ )
