[FLINK-40513][runtime] Introduce multimodal AI functions for image understanding

docs/content.zh/docs/core-concept/ai-model.md

@@ -24,11 +24,11 @@ under the License.
 
 # AI 模型
 
-AI 模型可用于 transform 表达式中的文本生成、文本分析和 embedding。
+AI 模型可用于 transform 表达式中的文本生成、文本分析、embedding 和图片理解。
 
 ## AI Functions
 
-模型名称必须是字符串常量，并引用 `pipeline.model` 中声明的模型。文本函数要求模型客户端支持文本生成，`AI_EMBED` 要求模型客户端支持 embedding；Pipeline 会在执行前校验引用模型的 capability 是否匹配。
+模型名称必须是字符串常量，并引用 `pipeline.model` 中声明的模型。文本、embedding 和图片函数分别要求模型客户端实现对应的 capability；Pipeline 会在执行前校验引用模型的 capability 是否匹配。
 
 所有文本函数都会将模型返回的 JSON 解析为 `VARIANT`。
 
@@ -43,13 +43,24 @@ AI 模型可用于 transform 表达式中的文本生成、文本分析和 embed
 | `AI_MASK(model, input, entities)` | 对指定实体类型进行脱敏。 | `masked_text`、`detected_entities` |
 | `AI_EMBED(model, input)` | 生成 embedding 向量。 | 不返回 JSON，而是返回 `ARRAY<FLOAT>`。 |
 
+以下多模态函数从 `BYTES` 字段读取图片数据：
+
+| 函数 | 说明 | 返回类型 |
+|------|------|----------|
+| `AI_IMAGE_COMPLETE(model, image, prompt)` | 根据图片和自然语言 prompt 生成文本。 | `STRING` |
+| `AI_IMAGE_EMBED(model, image)` | 将图片转换为 embedding 向量。 | `ARRAY<FLOAT>` |
+
+OpenAI-compatible 模型客户端通过标准 vision chat 支持 `AI_IMAGE_COMPLETE`。客户端会识别 PNG、JPEG、GIF 和 WebP 图片，并将图片编码为 Base64 data URL。图片为 `NULL` 时直接返回 `NULL`，且不会调用模型；图片为空或格式无法识别时，会在发送请求前报错。
+
+`AI_IMAGE_EMBED` 当前只提供框架函数和 provider capability。OpenAI API 目前没有定义标准的图片向量化协议，因此图片 embedding 需要由具体 provider 单独实现。OpenAI-compatible 模型客户端不实现图片 embedding，社区发行包目前也没有可用于生产的图片 embedding provider。需要图片向量化的用户需要等待后续 provider 实现。
+
 六个专用文本函数使用内置英文 prompt 模板，但输入文本可以是任意语言。输入为 `NULL` 时直接返回 `NULL`，且不会调用模型；模型返回 `NULL` 时也返回 `NULL`。非空文本响应必须是语法合法的 JSON，否则当前记录处理失败，错误信息会标明具体 AI 函数。运行时只校验 JSON 语法，不校验响应字段是否存在或字段类型是否匹配。
 
 ## OpenAI-compatible 模型客户端
 
 AI 模型客户端可供上述 AI Functions 引用。使用时，需要通过 `--jar` 将模型实现 JAR（例如 `flink-cdc-pipeline-model-openai-compatible`）添加到 Pipeline 命令中。
 
-OpenAI-compatible 客户端支持调用实现 OpenAI Chat Completions 和 Embeddings REST API 的服务。
+OpenAI-compatible 客户端支持调用实现 OpenAI Chat Completions、vision chat 和 Embeddings REST API 的服务。
 
 system prompt、函数 prompt 和输入文本均支持英文或中文内容。
 
@@ -60,6 +71,7 @@ transform:
       *,
       AI_COMPLETE('completion_model', content, '总结输入内容') AS summary,
       AI_SENTIMENT('completion_model', content) AS sentiment,
+      AI_IMAGE_COMPLETE('vision_model', image, '描述这张图片') AS image_description,
       AI_EMBED('embedding_model', content) AS embedding
 
 pipeline:
@@ -80,6 +92,12 @@ pipeline:
         endpoint: https://api.example.com/v1
         api-key: <api-key>
         dimension: 768
+    - name: vision_model
+      type: openai-compatible
+      options:
+        model: gpt-4o-mini
+        endpoint: https://api.example.com/v1
+        api-key: <api-key>
 ```
 
 不要将 API Key 提交到代码仓库中，请通过部署环境的密钥管理机制提供。

docs/content/docs/core-concept/ai-model.md

@@ -24,11 +24,12 @@ under the License.
 
 # AI Model
 
-AI models can be used in transform expressions for text generation, text analysis, and embedding.
+AI models can be used in transform expressions for text generation, text analysis, embedding, and
+image understanding.
 
 ## AI Functions
 
-The model name must be a string constant that refers to a model declared in `pipeline.model`. Text functions require a model client that implements text generation, while `AI_EMBED` requires embedding support. The pipeline validates the referenced model capability before execution.
+The model name must be a string constant that refers to a model declared in `pipeline.model`. Text functions require a model client that implements text generation, while embedding and image functions require their corresponding capabilities. The pipeline validates the referenced model capability before execution.
 
 All text functions return `VARIANT` values parsed from the model's JSON response.
 
@@ -43,13 +44,31 @@ All text functions return `VARIANT` values parsed from the model's JSON response
 | `AI_MASK(model, input, entities)` | Masks the requested entity types. | `masked_text`, `detected_entities` |
 | `AI_EMBED(model, input)` | Creates an embedding vector. | Returns `ARRAY<FLOAT>` instead of JSON. |
 
+The following multimodal functions accept image data from a `BYTES` column:
+
+| Function | Description | Return type |
+|----------|-------------|-------------|
+| `AI_IMAGE_COMPLETE(model, image, prompt)` | Generates text from an image and a natural-language prompt. | `STRING` |
+| `AI_IMAGE_EMBED(model, image)` | Converts an image into an embedding vector. | `ARRAY<FLOAT>` |
+
+The OpenAI-compatible model client supports `AI_IMAGE_COMPLETE` through standard vision chat. It
+detects PNG, JPEG, GIF, and WebP images and sends the image as a Base64 data URL. A `NULL` image
+returns `NULL` without invoking the model, while empty or unrecognized image data is rejected before
+the request is sent.
+
+`AI_IMAGE_EMBED` currently provides only the framework function and provider capability. The OpenAI
+API does not define a standard image embedding protocol, so image embedding requires a
+provider-specific implementation. The OpenAI-compatible model client does not implement image
+embedding, and the community distribution does not yet include a production provider for it. Users
+who need image embedding must wait for a follow-up provider implementation.
+
 The specialized text functions use built-in English prompt templates, but their input may be in any language. If the input is `NULL`, the function returns `NULL` without invoking the model. A `NULL` model response also produces `NULL`. A non-null text response must be syntactically valid JSON; otherwise, record processing fails with an error that identifies the AI function. The runtime validates JSON syntax but does not validate the presence or types of individual response fields.
 
 ## OpenAI-compatible Model Client
 
 AI model clients can be referenced by the AI functions above. Add the model implementation JAR, such as `flink-cdc-pipeline-model-openai-compatible`, to the pipeline command with `--jar`.
 
-The OpenAI-compatible client supports chat completions and text embeddings against endpoints that implement the corresponding OpenAI REST APIs.
+The OpenAI-compatible client supports chat completions, vision chat, and text embeddings against endpoints that implement the corresponding OpenAI REST APIs.
 
 System prompts, function prompts, and input text may contain either English or Chinese content.
 
@@ -60,6 +79,7 @@ transform:
       *,
       AI_COMPLETE('completion_model', content, 'Summarize the input') AS summary,
       AI_SENTIMENT('completion_model', content) AS sentiment,
+      AI_IMAGE_COMPLETE('vision_model', image, 'Describe the image') AS image_description,
       AI_EMBED('embedding_model', content) AS embedding
 
 pipeline:
@@ -80,6 +100,12 @@ pipeline:
         endpoint: https://api.example.com/v1
         api-key: <api-key>
         dimension: 768
+    - name: vision_model
+      type: openai-compatible
+      options:
+        model: gpt-4o-mini
+        endpoint: https://api.example.com/v1
+        api-key: <api-key>
 ```
 
 Do not store API keys in source control. Supply them through the secret-management mechanism of your deployment environment.

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/model/abilities/SupportsImageEmbedding.java

@@ -0,0 +1,29 @@
+/*
+ * 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.cdc.common.model.abilities;
+
+import org.apache.flink.cdc.common.annotation.Experimental;
+import org.apache.flink.cdc.common.model.AiModelClient;
+
+/** Ability interface for {@link AiModelClient} implementations that can embed image bytes. */
+@Experimental
+public interface SupportsImageEmbedding {
+
+    /** Converts the given image bytes into a dense float vector. */
+    float[] embedImage(byte[] image);
+}

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/model/abilities/SupportsImageTextGeneration.java

@@ -0,0 +1,32 @@
+/*
+ * 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.cdc.common.model.abilities;
+
+import org.apache.flink.cdc.common.annotation.Experimental;
+import org.apache.flink.cdc.common.model.AiModelClient;
+
+/**
+ * Ability interface for {@link AiModelClient} implementations that can generate text from an image
+ * combined with a natural-language prompt.
+ */
+@Experimental
+public interface SupportsImageTextGeneration {
+
+    /** Generates text from the given image bytes guided by a natural-language prompt. */
+    String generateTextFromImage(byte[] image, String prompt);
+}

flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineAiFunctionITCase.java

@@ -176,6 +176,32 @@ void testAiEmbedInProjection() throws Exception {
                         "DataChangeEvent{tableId=default_namespace.default_schema.mytable1, before=[], after=[1, [3.0, 1.0, 4.0, 1.0, 5.0, 9.0, 2.0, 6.0]], op=INSERT, meta=()}");
     }
 
+    @Test
+    void testAiImageCompleteInProjection() throws Exception {
+        String[] output =
+                runAiFunctionTest(
+                        "id, AI_IMAGE_COMPLETE('visionModel', image, 'Describe the image') AS description",
+                        List.of(ModelDef.of("visionModel", "dummy", Collections.emptyMap())));
+
+        assertThat(output)
+                .containsExactly(
+                        "CreateTableEvent{tableId=default_namespace.default_schema.mytable1, schema=columns={`id` INT NOT NULL,`description` STRING}, primaryKeys=id, options=()}",
+                        "DataChangeEvent{tableId=default_namespace.default_schema.mytable1, before=[], after=[1, A dummy description of the image.], op=INSERT, meta=()}");
+    }
+
+    @Test
+    void testAiImageEmbedInProjection() throws Exception {
+        String[] output =
+                runAiFunctionTest(
+                        "id, AI_IMAGE_EMBED('imageEmbedModel', image) AS embedding",
+                        List.of(ModelDef.of("imageEmbedModel", "dummy", Collections.emptyMap())));
+
+        assertThat(output)
+                .containsExactly(
+                        "CreateTableEvent{tableId=default_namespace.default_schema.mytable1, schema=columns={`id` INT NOT NULL,`embedding` ARRAY<FLOAT>}, primaryKeys=id, options=()}",
+                        "DataChangeEvent{tableId=default_namespace.default_schema.mytable1, before=[], after=[1, [2.0, 7.0, 1.0, 8.0, 2.0, 8.0]], op=INSERT, meta=()}");
+    }
+
     private String[] runAiFunctionTest(String projection, List<ModelDef> models) throws Exception {
         return runAiFunctionTest(projection, models, Collections.emptyList());
     }
@@ -204,6 +230,7 @@ private String[] runAiFunctionTest(
                 Schema.newBuilder()
                         .physicalColumn("id", DataTypes.INT())
                         .physicalColumn("content", DataTypes.STRING())
+                        .physicalColumn("image", DataTypes.BYTES())
                         .primaryKey("id")
                         .build();
         BinaryRecordDataGenerator generator =
@@ -216,7 +243,9 @@ private String[] runAiFunctionTest(
                         tableId,
                         generator.generate(
                                 new Object[] {
-                                    1, BinaryStringData.fromString("I love this product")
+                                    1,
+                                    BinaryStringData.fromString("I love this product"),
+                                    new byte[] {1, 2, 3, 4}
                                 })));
         ValuesDataSourceHelper.setSourceEvents(Collections.singletonList(events));
 

flink-cdc-pipeline-model/flink-cdc-pipeline-model-dummy/src/main/java/org/apache/flink/cdc/models/dummy/DummyModelClient.java

@@ -19,10 +19,17 @@
 
 import org.apache.flink.cdc.common.model.AiModelClient;
 import org.apache.flink.cdc.common.model.abilities.SupportsEmbedding;
+import org.apache.flink.cdc.common.model.abilities.SupportsImageEmbedding;
+import org.apache.flink.cdc.common.model.abilities.SupportsImageTextGeneration;
 import org.apache.flink.cdc.common.model.abilities.SupportsTextGeneration;
 
 /** Deterministic AI model client used by tests. */
-public class DummyModelClient implements AiModelClient, SupportsTextGeneration, SupportsEmbedding {
+public class DummyModelClient
+        implements AiModelClient,
+                SupportsTextGeneration,
+                SupportsEmbedding,
+                SupportsImageTextGeneration,
+                SupportsImageEmbedding {
 
     private static final long serialVersionUID = 1L;
 
@@ -63,6 +70,24 @@ public float[] embed(String text) {
         return new float[] {3f, 1f, 4f, 1f, 5f, 9f, 2f, 6f};
     }
 
+    @Override
+    public String generateTextFromImage(byte[] image, String prompt) {
+        if (debug) {
+            System.out.printf(
+                    "Received image of %d bytes%nPrompt: %s%n",
+                    image == null ? 0 : image.length, prompt);
+        }
+        return "A dummy description of the image.";
+    }
+
+    @Override
+    public float[] embedImage(byte[] image) {
+        if (debug) {
+            System.out.printf("Received image of %d bytes%n", image == null ? 0 : image.length);
+        }
+        return new float[] {2f, 7f, 1f, 8f, 2f, 8f};
+    }
+
     @Override
     public void open() {
         if (debug) {