This is an automated email from the ASF dual-hosted git repository.
yuxiqian pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new 53fe0f92a [FLINK-40411][runtime] Introduce OpenAI-compatible model
client (#4507)
53fe0f92a is described below
commit 53fe0f92aa0e6d7306c19c1d74bfd080e41efb30
Author: haruki <[email protected]>
AuthorDate: Tue Aug 18 18:45:43 2026 +0800
[FLINK-40411][runtime] Introduce OpenAI-compatible model client (#4507)
---
.github/workflows/modules.py | 1 +
docs/content.zh/docs/core-concept/ai-model.md | 161 +++++++++
docs/content.zh/docs/core-concept/transform.md | 71 +---
docs/content/docs/core-concept/ai-model.md | 161 +++++++++
docs/content/docs/core-concept/transform.md | 71 +---
.../cdc/composer/flink/FlinkPipelineComposer.java | 3 +-
.../flink/translator/TransformTranslator.java | 12 +-
.../flink/FlinkPipelineAiFunctionITCase.java | 89 +++++
.../pom.xml | 132 ++++++++
.../cdc/models/openai/ErrorHandlingStrategy.java | 25 ++
.../models/openai/OpenAiCompatibleModelClient.java | 358 +++++++++++++++++++++
.../openai/OpenAiCompatibleModelClientFactory.java | 90 ++++++
.../openai/OpenAiCompatibleModelOptions.java | 225 +++++++++++++
.../cdc/models/openai/OpenAiRequestParams.java | 194 +++++++++++
.../cdc/models/openai/RetryBackoffStrategy.java | 54 ++++
.../src/main/resources/META-INF/NOTICE | 34 ++
.../org.apache.flink.cdc.common.factories.Factory | 16 +
.../OpenAiCompatibleModelClientFactoryTest.java | 147 +++++++++
.../openai/OpenAiCompatibleModelClientTest.java | 292 +++++++++++++++++
.../cdc/models/openai/OpenAiRequestParamsTest.java | 125 +++++++
.../src/test/resources/log4j2-test.properties | 22 ++
flink-cdc-pipeline-model/pom.xml | 1 +
22 files changed, 2140 insertions(+), 144 deletions(-)
diff --git a/.github/workflows/modules.py b/.github/workflows/modules.py
index e80b2ccad..d92a4fc65 100755
--- a/.github/workflows/modules.py
+++ b/.github/workflows/modules.py
@@ -13,6 +13,7 @@ MODULES_CORE = [
"flink-cdc-common",
"flink-cdc-composer",
"flink-cdc-runtime",
+ "flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible",
"flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base",
"flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-values"
]
diff --git a/docs/content.zh/docs/core-concept/ai-model.md
b/docs/content.zh/docs/core-concept/ai-model.md
new file mode 100644
index 000000000..9b8029326
--- /dev/null
+++ b/docs/content.zh/docs/core-concept/ai-model.md
@@ -0,0 +1,161 @@
+---
+title: "AI 模型"
+weight: 9
+type: docs
+---
+<!--
+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.
+-->
+
+# AI 模型
+
+AI 模型可用于 transform 表达式中的文本补全和 embedding。
+
+## OpenAI-compatible 模型客户端
+
+AI 模型客户端可供 transform 中的 `AI_COMPLETE` 和 `AI_EMBED` 函数引用。使用时，需要通过 `--jar` 将模型实现
JAR（例如 `flink-cdc-pipeline-model-openai-compatible`）添加到 Pipeline 命令中。
+
+OpenAI-compatible 客户端支持调用实现 OpenAI Chat Completions 和 Embeddings REST API 的服务。
+
+system prompt、函数 prompt 和输入文本均支持英文或中文内容。
+
+```yaml
+transform:
+ - source-table: db.\.*
+ projection: >-
+ *,
+ AI_COMPLETE('completion_model', content, '总结输入内容') AS summary,
+ AI_EMBED('embedding_model', content) AS embedding
+
+pipeline:
+ model:
+ - name: completion_model
+ type: openai-compatible
+ options:
+ model: gpt-4o-mini
+ endpoint: https://api.example.com/v1
+ api-key: <api-key>
+ system-prompt: 你是一个简洁的助手。
+ temperature: 0.2
+ max-tokens: 256
+ - name: embedding_model
+ type: openai-compatible
+ options:
+ model: text-embedding-3-small
+ endpoint: https://api.example.com/v1
+ api-key: <api-key>
+ dimension: 768
+```
+
+不要将 API Key 提交到代码仓库中，请通过部署环境的密钥管理机制提供。
+
+### OpenAI-compatible 配置项
+
+| 配置项 | 是否必填 | 说明 |
+|--------|----------|------|
+| `model` | 是 | 发送给服务端的模型名称；`model-name` 作为废弃别名仍可使用。 |
+| `endpoint` | 是 | OpenAI-compatible 服务的 Base URL。 |
+| `api-key` | 是 | 请求认证使用的 Bearer Token。 |
+| `system-prompt` | 否 | 添加在 `AI_COMPLETE` 生成的 system prompt 之前。 |
+| `user-prompt` | 否 | 在输入之后追加一条 user message。 |
+| `temperature`、`top-p`、`stop`、`max-tokens` | 否 | 常用文本生成参数。 |
+| `presence-penalty`、`frequency-penalty`、`n`、`seed` | 否 | 其他文本生成参数。 |
+| `response-format` | 否 | 支持 `json_object`；AI completion 的结果必须是合法 JSON。 |
+| `content-type` | 否 | `text`（默认）或 `image_url`。 |
+| `dimension` | 否 | 请求的 embedding 维度。 |
+| `extra-header`、`extra-body` | 否 | JSON 对象格式的厂商自定义请求头或请求体字段。 |
+| `error-handling-strategy` | 否 | `retry`（默认）、`failover` 或 `ignore`。 |
+| `retry-num` | 否 | 最大尝试次数，默认 `100`。 |
+| `retry-fallback-strategy` | 否 | 重试耗尽后的策略，可选 `failover`（默认）或 `ignore`。 |
+| `retry-backoff-strategy` | 否 | `fixed`（默认）或 `exponential`。 |
+| `retry-backoff-base-interval` | 否 | 重试基础间隔，默认 `1 s`。 |
+
+## 旧版 Embedding AI 模型（已废弃）
+
+> **已废弃：** 基于 `model-name` 和 `class-name` 的旧版模型 API 已废弃，并计划在未来移除。新 Pipeline
请使用上面基于 Factory 的 OpenAI-compatible 模型客户端。
+
+旧版 Embedding AI 模型可以在 transform 规则中使用。使用时，需要下载内置模型 JAR，并在 `flink-cdc.sh` 命令中添加
`--jar {$BUILT_IN_MODEL_PATH}`。
+
+如何定义一个 Embedding AI 模型：
+
+```yaml
+pipeline:
+ model:
+ - model-name: CHAT
+ class-name: OpenAIChatModel
+ openai.model: text-embedding-3-small
+ openai.host: https://xxxx
+ openai.apikey: abcd1234
+ openai.chat.prompt: please summary this
+ - model-name: GET_EMBEDDING
+ class-name: OpenAIEmbeddingModel
+ openai.model: text-embedding-3-small
+ openai.host: https://xxxx
+ openai.apikey: abcd1234
+```
+
+注意：
+
+* `model-name` 是所有支持模型通用的必填参数，表示在 `projection` 或 `filter` 中调用的函数名称。
+* `class-name` 是所有支持模型通用的必填参数，可用值参见[所有支持的模型](#所有支持的模型)。
+* `openai.model`、`openai.host`、`openai.apikey` 和 `openai.chat.prompt`
是由具体模型定义的配置项。
+
+如何使用一个 Embedding AI 模型：
+
+```yaml
+transform:
+ - source-table: db.\.*
+ projection: "*, inc(inc(inc(id))) as inc_id, GET_EMBEDDING(page) as emb,
CHAT(page) as summary"
+ filter: inc(id) < 100
+pipeline:
+ model:
+ - model-name: CHAT
+ class-name: OpenAIChatModel
+ openai.model: gpt-4o-mini
+ openai.host: http://langchain4j.dev/demo/openai/v1
+ openai.apikey: demo
+ openai.chat.prompt: please summary this
+ - model-name: GET_EMBEDDING
+ class-name: OpenAIEmbeddingModel
+ openai.model: text-embedding-3-small
+ openai.host: http://langchain4j.dev/demo/openai/v1
+ openai.apikey: demo
+```
+
+这里，`GET_EMBEDDING` 通过 `model-name` 在 `pipeline` 中定义。
+
+### 所有支持的模型
+
+下面列出了所有支持的内置模型：
+
+#### OpenAIChatModel
+
+| 参数 | 类型 | 是否必填 | 含义 |
+|------|------|----------|------|
+| `openai.model` | STRING | 是 | 要调用的模型名称，例如 `gpt-4o-mini`。可用选项有
`gpt-4o-mini`、`gpt-4o`、`gpt-4-32k` 和 `gpt-3.5-turbo`。 |
+| `openai.host` | STRING | 是 | 模型服务器地址，例如
`http://langchain4j.dev/demo/openai/v1`。 |
+| `openai.apikey` | STRING | 是 | 模型服务器验证使用的 API Key，例如 `demo`。 |
+| `openai.chat.prompt` | STRING | 否 | 与 OpenAI 聊天的提示词，例如 `Please summarize
this`。 |
+
+#### OpenAIEmbeddingModel
+
+| 参数 | 类型 | 是否必填 | 含义 |
+|------|------|----------|------|
+| `openai.model` | STRING | 是 | 要调用的模型名称，例如 `text-embedding-3-small`。可用选项有
`text-embedding-3-small`、`text-embedding-3-large` 和 `text-embedding-ada-002`。 |
+| `openai.host` | STRING | 是 | 模型服务器地址，例如
`http://langchain4j.dev/demo/openai/v1`。 |
+| `openai.apikey` | STRING | 是 | 模型服务器验证使用的 API Key，例如 `demo`。 |
diff --git a/docs/content.zh/docs/core-concept/transform.md
b/docs/content.zh/docs/core-concept/transform.md
index 8636567cf..0409d7a6a 100644
--- a/docs/content.zh/docs/core-concept/transform.md
+++ b/docs/content.zh/docs/core-concept/transform.md
@@ -527,73 +527,4 @@ transform:
filter: inc(id) < 100
```
-## Embedding AI 模型
-
-内置 AI 模型可以在 transform 规则中使用。
-为了使用内置 AI 模型，你需要下载内置模型的 jar ，然后在 `flink-cdc.sh` 命令中添加 `--jar
{$BUILT_IN_MODEL_PATH}`。
-
-如何定义一个 Embedding AI 模型：
-
-```yaml
-pipeline:
- model:
- - model-name: CHAT
- class-name: OpenAIChatModel
- openai.model: gpt-4o-mini
- openai.host: https://xxxx
- openai.apikey: abcd1234
- openai.chat.prompt: please summary this
- - model-name: GET_EMBEDDING
- class-name: OpenAIEmbeddingModel
- openai.model: text-embedding-3-small
- openai.host: https://xxxx
- openai.apikey: abcd1234
-```
-注意：
-* `model-name` 是一个通用的必填参数，用于所有支持的模型，表示在 `projection` 或 `filter` 中调用的函数名称。
-* `class-name` 是一个通用的必填参数，用于所有支持的模型，可用值可以在[所有支持的模型](#all-support-models)中找到。
-* `openai.model`，`openai.host`，`openai.apiKey` 和 `openai.chat.prompt`
是在各个模型中特别的可选参数。
-
-如何使用一个 Embedding AI 模型：
-
-```yaml
-transform:
- - source-table: db.\.*
- projection: "*, inc(inc(inc(id))) as inc_id, GET_EMBEDDING(page) as emb,
CHAT(page) as summary"
- filter: inc(id) < 100
-pipeline:
- model:
- - model-name: CHAT
- class-name: OpenAIChatModel
- openai.model: gpt-4o-mini
- openai.host: http://langchain4j.dev/demo/openai/v1
- openai.apikey: demo
- openai.chat.prompt: please summary this
- - model-name: GET_EMBEDDING
- class-name: OpenAIEmbeddingModel
- openai.model: text-embedding-3-small
- openai.host: http://langchain4j.dev/demo/openai/v1
- openai.apikey: demo
-```
-这里，GET_EMBEDDING 是通过 `model-name` 在 `pipeline` 中定义的。
-
-### 所有支持的模型
-
-下面列出了所有支持的模型：
-
-#### OpenAIChatModel
-
-| 参数 | 类型 | 是否必填 | 含义
|
-|--------------------|--------|----------|-----------------------------------------------------------------------------------------------------|
-| openai.model | STRING | 必填 | 要调用的模型名称，例如："gpt-4o-mini"，可用选项有
"gpt-4o-mini"、"gpt-4o"、"gpt-4-32k"、"gpt-3.5-turbo"。 |
-| openai.host | STRING | 必填 |
要连接的模型服务器地址，例如：`http://langchain4j.dev/demo/openai/v1`。
|
-| openai.apikey | STRING | 必填 | 模型服务器验证的 API Key，例如："demo"。
|
-| openai.chat.prompt | STRING | 可选 | 与 OpenAI 聊天的提示词，例如："Please summary
this"。 |
-
-#### OpenAIEmbeddingModel
-
-| 参数 | 类型 | 是否必填 | 含义
|
-|---------------|--------|----------|----------------------------------------------------------------------------------------------------------------------|
-| openai.model | STRING | 必填 |
要调用的模型名称，例如："text-embedding-3-small"，可用选项有
"text-embedding-3-small"、"text-embedding-3-large"、"text-embedding-ada-002"。 |
-| openai.host | STRING | 必填 |
要连接的模型服务器地址，例如：`http://langchain4j.dev/demo/openai/v1`。
|
-| openai.apikey | STRING | 必填 | 模型服务器验证的 API Key，例如："demo"。
|
+有关 AI 模型函数和配置，请参见 [AI 模型]({{< ref "docs/core-concept/ai-model" >}})。
diff --git a/docs/content/docs/core-concept/ai-model.md
b/docs/content/docs/core-concept/ai-model.md
new file mode 100644
index 000000000..da3ef0f00
--- /dev/null
+++ b/docs/content/docs/core-concept/ai-model.md
@@ -0,0 +1,161 @@
+---
+title: "AI Model"
+weight: 9
+type: docs
+---
+<!--
+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.
+-->
+
+# AI Model
+
+AI models can be used in transform expressions for text completion and
embedding.
+
+## OpenAI-compatible Model Client
+
+AI model clients can be referenced by the `AI_COMPLETE` and `AI_EMBED`
transform functions. 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.
+
+System prompts, function prompts, and input text may contain either English or
Chinese content.
+
+```yaml
+transform:
+ - source-table: db.\.*
+ projection: >-
+ *,
+ AI_COMPLETE('completion_model', content, 'Summarize the input') AS
summary,
+ AI_EMBED('embedding_model', content) AS embedding
+
+pipeline:
+ model:
+ - name: completion_model
+ type: openai-compatible
+ options:
+ model: gpt-4o-mini
+ endpoint: https://api.example.com/v1
+ api-key: <api-key>
+ system-prompt: You are a concise assistant.
+ temperature: 0.2
+ max-tokens: 256
+ - name: embedding_model
+ type: openai-compatible
+ options:
+ model: text-embedding-3-small
+ endpoint: https://api.example.com/v1
+ api-key: <api-key>
+ dimension: 768
+```
+
+Do not store API keys in source control. Supply them through the
secret-management mechanism of your deployment environment.
+
+### OpenAI-compatible Options
+
+| Option | Required | Description |
+|--------|----------|-------------|
+| `model` | Yes | Model name sent to the endpoint. `model-name` is accepted as
a deprecated alias. |
+| `endpoint` | Yes | Base URL of the OpenAI-compatible endpoint. |
+| `api-key` | Yes | Bearer token used to authenticate requests. |
+| `system-prompt` | No | Prompt prepended to the system prompt generated by
`AI_COMPLETE`. |
+| `user-prompt` | No | Additional user message appended after the input. |
+| `temperature`, `top-p`, `stop`, `max-tokens` | No | Common generation
parameters. |
+| `presence-penalty`, `frequency-penalty`, `n`, `seed` | No | Additional
generation parameters. |
+| `response-format` | No | `json_object` is supported. AI completion results
must be valid JSON. |
+| `content-type` | No | `text` (default) or `image_url`. |
+| `dimension` | No | Requested embedding dimension. |
+| `extra-header`, `extra-body` | No | Provider-specific headers or body fields
encoded as JSON objects. |
+| `error-handling-strategy` | No | `retry` (default), `failover`, or `ignore`.
|
+| `retry-num` | No | Maximum number of attempts. Defaults to `100`. |
+| `retry-fallback-strategy` | No | `failover` (default) or `ignore` after
retries are exhausted. |
+| `retry-backoff-strategy` | No | `fixed` (default) or `exponential`. |
+| `retry-backoff-base-interval` | No | Base retry interval. Defaults to `1 s`.
|
+
+## Legacy Embedding AI Model (Deprecated)
+
+> **Deprecated:** The legacy model API based on `model-name` and `class-name`
is deprecated and planned for removal. Use the factory-based OpenAI-compatible
model client above for new pipelines.
+
+The legacy Embedding AI Model can be used in transform rules. To use it,
download the built-in model JAR and add `--jar {$BUILT_IN_MODEL_PATH}` to your
`flink-cdc.sh` command.
+
+How to define an Embedding AI Model:
+
+```yaml
+pipeline:
+ model:
+ - model-name: CHAT
+ class-name: OpenAIChatModel
+ openai.model: gpt-4o-mini
+ openai.host: https://xxxx
+ openai.apikey: abcd1234
+ openai.chat.prompt: please summary this
+ - model-name: GET_EMBEDDING
+ class-name: OpenAIEmbeddingModel
+ openai.model: text-embedding-3-small
+ openai.host: https://xxxx
+ openai.apikey: abcd1234
+```
+
+Note:
+
+* `model-name` is a common required parameter for all supported models. It
represents the function name called in `projection` or `filter`.
+* `class-name` is a common required parameter for all supported models.
Available values can be found in [All Supported Models](#all-supported-models).
+* `openai.model`, `openai.host`, `openai.apikey`, and `openai.chat.prompt` are
options defined by a specific model.
+
+How to use an Embedding AI Model:
+
+```yaml
+transform:
+ - source-table: db.\.*
+ projection: "*, inc(inc(inc(id))) as inc_id, GET_EMBEDDING(page) as emb,
CHAT(page) as summary"
+ filter: inc(id) < 100
+pipeline:
+ model:
+ - model-name: CHAT
+ class-name: OpenAIChatModel
+ openai.model: gpt-4o-mini
+ openai.host: http://langchain4j.dev/demo/openai/v1
+ openai.apikey: demo
+ openai.chat.prompt: please summary this
+ - model-name: GET_EMBEDDING
+ class-name: OpenAIEmbeddingModel
+ openai.model: text-embedding-3-small
+ openai.host: http://langchain4j.dev/demo/openai/v1
+ openai.apikey: demo
+```
+
+Here, `GET_EMBEDDING` is defined through `model-name` in `pipeline`.
+
+### All Supported Models
+
+The following built-in models are provided:
+
+#### OpenAIChatModel
+
+| Parameter | Type | Required | Description |
+|-----------|------|----------|-------------|
+| `openai.model` | STRING | Yes | Name of the model to call, for example,
`gpt-4o-mini`. Available options are `gpt-4o-mini`, `gpt-4o`, `gpt-4-32k`, and
`gpt-3.5-turbo`. |
+| `openai.host` | STRING | Yes | Model server address, for example,
`http://langchain4j.dev/demo/openai/v1`. |
+| `openai.apikey` | STRING | Yes | API key for authenticating with the model
server, for example, `demo`. |
+| `openai.chat.prompt` | STRING | No | Prompt for chatting with OpenAI, for
example, `Please summarize this`. |
+
+#### OpenAIEmbeddingModel
+
+| Parameter | Type | Required | Description |
+|-----------|------|----------|-------------|
+| `openai.model` | STRING | Yes | Name of the model to call, for example,
`text-embedding-3-small`. Available options are `text-embedding-3-small`,
`text-embedding-3-large`, and `text-embedding-ada-002`. |
+| `openai.host` | STRING | Yes | Model server address, for example,
`http://langchain4j.dev/demo/openai/v1`. |
+| `openai.apikey` | STRING | Yes | API key for authenticating with the model
server, for example, `demo`. |
diff --git a/docs/content/docs/core-concept/transform.md
b/docs/content/docs/core-concept/transform.md
index a9d217ae7..8e2f9815c 100644
--- a/docs/content/docs/core-concept/transform.md
+++ b/docs/content/docs/core-concept/transform.md
@@ -532,73 +532,4 @@ transform:
filter: inc(id) < 100
```
-## Embedding AI Model
-
-Embedding AI Model can be used in transform rules.
-To use Embedding AI Model, you need to download the jar of build-in model, and
then add `--jar {$BUILT_IN_MODEL_PATH}` to your flink-cdc.sh command.
-
-How to define a Embedding AI Model:
-
-```yaml
-pipeline:
- model:
- - model-name: CHAT
- class-name: OpenAIChatModel
- openai.model: text-embedding-3-small
- openai.host: https://xxxx
- openai.apikey: abcd1234
- openai.chat.prompt: please summary this
- - model-name: GET_EMBEDDING
- class-name: OpenAIEmbeddingModel
- openai.model: text-embedding-3-small
- openai.host: https://xxxx
- openai.apikey: abcd1234
-```
-Note:
-* `model-name` is a common required parameter for all support models, which
represent the function name called in `projection` or `filter`.
-* `class-name` is a common required parameter for all support models,
available values can be found in [All Support models](#all-support-models).
-* `openai.model`, `openai.host`, `openai.apiKey` and `openai.chat.prompt` is
option parameters that defined in specific model.
-
-How to use a Embedding AI Model:
-
-```yaml
-transform:
- - source-table: db.\.*
- projection: "*, inc(inc(inc(id))) as inc_id, GET_EMBEDDING(page) as emb,
CHAT(page) as summary"
- filter: inc(id) < 100
-pipeline:
- model:
- - model-name: CHAT
- class-name: OpenAIChatModel
- openai.model: gpt-4o-mini
- openai.host: http://langchain4j.dev/demo/openai/v1
- openai.apikey: demo
- openai.chat.prompt: please summary this
- - model-name: GET_EMBEDDING
- class-name: OpenAIEmbeddingModel
- openai.model: text-embedding-3-small
- openai.host: http://langchain4j.dev/demo/openai/v1
- openai.apikey: demo
-```
-Here, GET_EMBEDDING is defined though `model-name` in `pipeline`.
-
-### All Support models
-
-The following built-in models are provided:
-
-#### OpenAIChatModel
-
-| parameter | type | optional/required | meaning
|
-|--------------------|--------|-------------------|--------------------------------------------------------------------------------------------------------------------------------------|
-| openai.model | STRING | required | Name of model to be
called, for example: "gpt-4o-mini", Available options are "gpt-4o-mini",
"gpt-4o", "gpt-4-32k", "gpt-3.5-turbo". |
-| openai.host | STRING | required | Host of the Model server
to be connected, for example: `http://langchain4j.dev/demo/openai/v1`.
|
-| openai.apikey | STRING | required | Api Key for verification
of the Model server, for example, "demo".
|
-| openai.chat.prompt | STRING | optional | Prompt for chatting with
OpenAI, for example: "Please summary this ".
|
-
-#### OpenAIEmbeddingModel
-
-| parameter | type | optional/required | meaning
|
-|---------------|--------|-------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| openai.model | STRING | required | Name of model to be called, for
example: "text-embedding-3-small", Available options are
"text-embedding-3-small", "text-embedding-3-large", "text-embedding-ada-002". |
-| openai.host | STRING | required | Host of the Model server to be
connected, for example: `http://langchain4j.dev/demo/openai/v1`.
|
-| openai.apikey | STRING | required | Api Key for verification of the
Model server, for example, "demo".
|
+For AI model functions and configuration, see [AI Model]({{< ref
"docs/core-concept/ai-model" >}}).
diff --git
a/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposer.java
b/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposer.java
index 4d843ac87..71fc865f9 100644
---
a/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposer.java
+++
b/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/FlinkPipelineComposer.java
@@ -200,7 +200,8 @@ public class FlinkPipelineComposer implements
PipelineComposer {
pipelineDef.getUdfs(),
pipelineDef.getModels(),
dataSource.supportedMetadataColumns(),
- operatorUidGenerator);
+ operatorUidGenerator,
+ env);
if (isParallelMetadataSource) {
// Translate a distributed topology for sources with distributed
tables
diff --git
a/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/TransformTranslator.java
b/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/TransformTranslator.java
index 0ccb3afd8..f24e1a99a 100644
---
a/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/TransformTranslator.java
+++
b/flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/translator/TransformTranslator.java
@@ -28,6 +28,7 @@ import
org.apache.flink.cdc.common.source.SupportedMetadataColumn;
import org.apache.flink.cdc.composer.definition.ModelDef;
import org.apache.flink.cdc.composer.definition.TransformDef;
import org.apache.flink.cdc.composer.definition.UdfDef;
+import org.apache.flink.cdc.composer.flink.FlinkEnvironmentUtils;
import org.apache.flink.cdc.composer.utils.FactoryDiscoveryUtils;
import org.apache.flink.cdc.runtime.operators.transform.PostTransformOperator;
import
org.apache.flink.cdc.runtime.operators.transform.PostTransformOperatorBuilder;
@@ -36,6 +37,7 @@ import
org.apache.flink.cdc.runtime.operators.transform.PreTransformOperatorBuil
import org.apache.flink.cdc.runtime.parser.TransformParser;
import org.apache.flink.cdc.runtime.typeutils.EventTypeInfo;
import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import java.util.Collections;
import java.util.LinkedHashMap;
@@ -111,7 +113,8 @@ public class TransformTranslator {
List<UdfDef> udfFunctions,
List<ModelDef> models,
SupportedMetadataColumn[] supportedMetadataColumns,
- OperatorUidGenerator operatorUidGenerator) {
+ OperatorUidGenerator operatorUidGenerator,
+ StreamExecutionEnvironment env) {
if (transforms.isEmpty()) {
return input;
}
@@ -138,7 +141,7 @@ public class TransformTranslator {
.filter(ModelDef::isLegacy)
.map(this::modelToUDFTuple)
.collect(Collectors.toList()));
- postTransformFunctionBuilder.addModelClients(loadModelClients(models));
+ postTransformFunctionBuilder.addModelClients(loadModelClients(models,
env));
return input.transform(
"Transform:Data", new EventTypeInfo(),
postTransformFunctionBuilder.build())
.uid(operatorUidGenerator.generateUid("post-transform"));
@@ -151,7 +154,8 @@ public class TransformTranslator {
model.getParameters());
}
- private Map<String, AiModelClient> loadModelClients(List<ModelDef> models)
{
+ private Map<String, AiModelClient> loadModelClients(
+ List<ModelDef> models, StreamExecutionEnvironment env) {
List<ModelDef> clientModels =
models.stream().filter(model ->
!model.isLegacy()).collect(Collectors.toList());
if (clientModels.isEmpty()) {
@@ -164,6 +168,8 @@ public class TransformTranslator {
AiModelClientFactory factory =
FactoryDiscoveryUtils.getFactoryByIdentifier(
model.getType(), AiModelClientFactory.class);
+ FactoryDiscoveryUtils.getJarPathByIdentifier(factory)
+ .ifPresent(jar -> FlinkEnvironmentUtils.addJar(env, jar));
FactoryHelper.createFactoryHelper(
factory,
new FactoryHelper.DefaultContext(
diff --git
a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineAiFunctionITCase.java
b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineAiFunctionITCase.java
index 8aabfaffb..f01777540 100644
---
a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineAiFunctionITCase.java
+++
b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineAiFunctionITCase.java
@@ -39,6 +39,8 @@ import
org.apache.flink.cdc.connectors.values.factory.ValuesDataFactory;
import org.apache.flink.cdc.connectors.values.sink.ValuesDataSinkOptions;
import org.apache.flink.cdc.connectors.values.source.ValuesDataSourceHelper;
import org.apache.flink.cdc.connectors.values.source.ValuesDataSourceOptions;
+import org.apache.flink.cdc.models.dummy.DummyModelClient;
+import org.apache.flink.cdc.models.dummy.DummyModelClientFactory;
import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator;
import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
import org.apache.flink.test.junit5.MiniClusterExtension;
@@ -47,14 +49,25 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
import java.io.PrintStream;
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import java.util.jar.JarEntry;
+import java.util.jar.JarOutputStream;
import static
org.apache.flink.configuration.CoreOptions.ALWAYS_PARENT_FIRST_LOADER_PATTERNS_ADDITIONAL;
+import static org.apache.flink.configuration.PipelineOptions.JARS;
import static org.assertj.core.api.Assertions.assertThat;
/** Integration test for AI functions in the Flink pipeline. */
@@ -83,6 +96,8 @@ class FlinkPipelineAiFunctionITCase {
private final PrintStream standardOut = System.out;
private final ByteArrayOutputStream outCaptor = new
ByteArrayOutputStream();
+ @TempDir Path tempDir;
+
@BeforeEach
void init() {
System.setOut(new PrintStream(outCaptor));
@@ -125,6 +140,19 @@ class FlinkPipelineAiFunctionITCase {
}
private String[] runAiFunctionTest(String projection, List<ModelDef>
models) throws Exception {
+ URL modelJar = createDummyModelJar().toUri().toURL();
+ ClassLoader originalClassLoader =
Thread.currentThread().getContextClassLoader();
+ try (URLClassLoader modelClassLoader =
+ new DummyModelClassLoader(modelJar, originalClassLoader)) {
+ Thread.currentThread().setContextClassLoader(modelClassLoader);
+ return runAiFunctionTest(projection, models, modelJar);
+ } finally {
+ Thread.currentThread().setContextClassLoader(originalClassLoader);
+ }
+ }
+
+ private String[] runAiFunctionTest(String projection, List<ModelDef>
models, URL modelJar)
+ throws Exception {
FlinkPipelineComposer composer = FlinkPipelineComposer.ofMiniCluster();
// Source: one table with a single row
@@ -190,8 +218,69 @@ class FlinkPipelineAiFunctionITCase {
// Execute & capture output
PipelineExecution execution = composer.compose(pipelineDef);
+ assertThat(composer.getEnv().getConfiguration().get(JARS))
+ .as("AI model provider JARs uploaded with the JobGraph")
+ .contains(modelJar.toString());
execution.execute();
return outCaptor.toString().trim().split("\n");
}
+
+ private Path createDummyModelJar() throws IOException {
+ Path modelJar = tempDir.resolve("dummy-model.jar");
+ try (JarOutputStream output = new
JarOutputStream(Files.newOutputStream(modelJar))) {
+ addClassToJar(DummyModelClient.class, output);
+ addClassToJar(DummyModelClientFactory.class, output);
+
+ output.putNextEntry(
+ new JarEntry(
+
"META-INF/services/org.apache.flink.cdc.common.factories.Factory"));
+ output.write(
+ (DummyModelClientFactory.class.getName() + "\n")
+ .getBytes(StandardCharsets.UTF_8));
+ output.closeEntry();
+ }
+ return modelJar;
+ }
+
+ private static void addClassToJar(Class<?> clazz, JarOutputStream output)
throws IOException {
+ String resourceName = clazz.getName().replace('.', '/') + ".class";
+ try (InputStream input =
clazz.getClassLoader().getResourceAsStream(resourceName)) {
+ assertThat(input).as("class resource %s",
resourceName).isNotNull();
+ output.putNextEntry(new JarEntry(resourceName));
+ input.transferTo(output);
+ output.closeEntry();
+ }
+ }
+
+ private static final class DummyModelClassLoader extends URLClassLoader {
+
+ private static final String DUMMY_MODEL_PACKAGE =
"org.apache.flink.cdc.models.dummy.";
+
+ private DummyModelClassLoader(URL modelJar, ClassLoader parent) {
+ super(new URL[] {modelJar}, parent);
+ }
+
+ @Override
+ protected Class<?> loadClass(String name, boolean resolve) throws
ClassNotFoundException {
+ if (!name.startsWith(DUMMY_MODEL_PACKAGE)) {
+ return super.loadClass(name, resolve);
+ }
+
+ synchronized (getClassLoadingLock(name)) {
+ Class<?> clazz = findLoadedClass(name);
+ if (clazz == null) {
+ try {
+ clazz = findClass(name);
+ } catch (ClassNotFoundException e) {
+ clazz = super.loadClass(name, false);
+ }
+ }
+ if (resolve) {
+ resolveClass(clazz);
+ }
+ return clazz;
+ }
+ }
+ }
}
diff --git
a/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/pom.xml
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/pom.xml
new file mode 100644
index 000000000..3ece6e81d
--- /dev/null
+++
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/pom.xml
@@ -0,0 +1,132 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0";
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <parent>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-cdc-pipeline-model-parent</artifactId>
+ <version>${revision}</version>
+ </parent>
+
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>flink-cdc-pipeline-model-openai-compatible</artifactId>
+ <packaging>jar</packaging>
+
+ <properties>
+ <jackson.version>2.18.2</jackson.version>
+ <openai-java.version>4.32.0</openai-java.version>
+ <okhttp.version>4.12.0</okhttp.version>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-cdc-common</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>com.openai</groupId>
+ <artifactId>openai-java</artifactId>
+ <version>${openai-java.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>com.squareup.okhttp3</groupId>
+ <artifactId>mockwebserver</artifactId>
+ <version>${okhttp.version}</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>shade-flink</id>
+ <configuration>
+ <artifactSet>
+ <includes combine.children="append">
+ <include>com.fasterxml:*</include>
+
<include>com.fasterxml.jackson.core:*</include>
+
<include>com.fasterxml.jackson.datatype:*</include>
+
<include>com.fasterxml.jackson.module:*</include>
+ <include>com.github.victools:*</include>
+ <include>com.google.errorprone:*</include>
+ <include>com.openai:*</include>
+ <include>com.squareup.okhttp3:*</include>
+ <include>com.squareup.okio:*</include>
+ <include>io.swagger.core.v3:*</include>
+ <include>org.jetbrains:*</include>
+ <include>org.jetbrains.kotlin:*</include>
+ </includes>
+ </artifactSet>
+ <filters combine.children="append">
+ <filter>
+ <artifact>*:*</artifact>
+ <excludes>
+
<exclude>META-INF/versions/**/module-info.class</exclude>
+ </excludes>
+ </filter>
+ </filters>
+ <transformers>
+ <transformer
implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
+ </transformers>
+ <relocations>
+ <relocation>
+ <pattern>com.fasterxml</pattern>
+
<shadedPattern>org.apache.flink.cdc.models.openai.shaded.com.fasterxml</shadedPattern>
+ </relocation>
+ <relocation>
+ <pattern>com.github.victools</pattern>
+
<shadedPattern>org.apache.flink.cdc.models.openai.shaded.com.github.victools</shadedPattern>
+ </relocation>
+ <relocation>
+ <pattern>com.google.errorprone</pattern>
+
<shadedPattern>org.apache.flink.cdc.models.openai.shaded.com.google.errorprone</shadedPattern>
+ </relocation>
+ <relocation>
+ <pattern>io.swagger</pattern>
+
<shadedPattern>org.apache.flink.cdc.models.openai.shaded.io.swagger</shadedPattern>
+ </relocation>
+ <relocation>
+ <pattern>okhttp3</pattern>
+
<shadedPattern>org.apache.flink.cdc.models.openai.shaded.okhttp3</shadedPattern>
+ </relocation>
+ <relocation>
+ <pattern>okio</pattern>
+
<shadedPattern>org.apache.flink.cdc.models.openai.shaded.okio</shadedPattern>
+ </relocation>
+ <relocation>
+
<pattern>org.jetbrains.annotations</pattern>
+
<shadedPattern>org.apache.flink.cdc.models.openai.shaded.org.jetbrains.annotations</shadedPattern>
+ </relocation>
+ </relocations>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>
diff --git
a/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/src/main/java/org/apache/flink/cdc/models/openai/ErrorHandlingStrategy.java
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/src/main/java/org/apache/flink/cdc/models/openai/ErrorHandlingStrategy.java
new file mode 100644
index 000000000..e4cf37c70
--- /dev/null
+++
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/src/main/java/org/apache/flink/cdc/models/openai/ErrorHandlingStrategy.java
@@ -0,0 +1,25 @@
+/*
+ * 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.models.openai;
+
+/** Strategy for handling model request failures. */
+public enum ErrorHandlingStrategy {
+ RETRY,
+ FAILOVER,
+ IGNORE
+}
diff --git
a/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/src/main/java/org/apache/flink/cdc/models/openai/OpenAiCompatibleModelClient.java
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/src/main/java/org/apache/flink/cdc/models/openai/OpenAiCompatibleModelClient.java
new file mode 100644
index 000000000..5a1e9041c
--- /dev/null
+++
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/src/main/java/org/apache/flink/cdc/models/openai/OpenAiCompatibleModelClient.java
@@ -0,0 +1,358 @@
+/*
+ * 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.models.openai;
+
+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.SupportsTextGeneration;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.openai.client.OpenAIClient;
+import com.openai.client.okhttp.OpenAIOkHttpClient;
+import com.openai.core.JsonValue;
+import com.openai.errors.OpenAIIoException;
+import com.openai.errors.OpenAIRetryableException;
+import com.openai.errors.OpenAIServiceException;
+import com.openai.models.chat.completions.ChatCompletion;
+import com.openai.models.chat.completions.ChatCompletionContentPart;
+import com.openai.models.chat.completions.ChatCompletionContentPartImage;
+import com.openai.models.chat.completions.ChatCompletionCreateParams;
+import com.openai.models.embeddings.CreateEmbeddingResponse;
+import com.openai.models.embeddings.Embedding;
+import com.openai.models.embeddings.EmbeddingCreateParams;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Supplier;
+
+/** AI model client that connects to an OpenAI-compatible endpoint. */
+public class OpenAiCompatibleModelClient
+ implements AiModelClient, SupportsTextGeneration, SupportsEmbedding {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(OpenAiCompatibleModelClient.class);
+
+ private static final long serialVersionUID = 1L;
+
+ private final String endpoint;
+ private final String apiKey;
+ private final String model;
+ @Nullable private final String configuredSystemPrompt;
+ private final OpenAiRequestParams params;
+
+ private transient OpenAIClient client;
+ private transient Map<String, List<String>> additionalHeaders;
+ private transient Map<String, JsonValue> additionalBody;
+
+ OpenAiCompatibleModelClient(
+ String endpoint,
+ String apiKey,
+ String model,
+ @Nullable String configuredSystemPrompt,
+ OpenAiRequestParams params) {
+ this.endpoint = endpoint;
+ this.apiKey = apiKey;
+ this.model = model;
+ this.configuredSystemPrompt = configuredSystemPrompt;
+ this.params = params;
+ }
+
+ @Override
+ public void open() {
+ additionalHeaders = parseHeaders(params.extraHeader);
+ additionalBody = parseBody(params.extraBody);
+ client =
+
OpenAIOkHttpClient.builder().baseUrl(endpoint).apiKey(apiKey).maxRetries(0).build();
+ LOG.info("Opened OpenAI-compatible model client. Endpoint: {} Model:
{}", endpoint, model);
+ }
+
+ @Override
+ public void close() {
+ if (client != null) {
+ client.close();
+ client = null;
+ }
+ }
+
+ @Override
+ public String generate(String systemPrompt, String userInput) {
+ return executeWithRetry("text completion", () ->
complete(systemPrompt, userInput));
+ }
+
+ @Override
+ public float[] embed(String text) {
+ return executeWithRetry("embedding", () -> createEmbedding(text));
+ }
+
+ private String complete(String systemPrompt, String userInput) {
+ ChatCompletionCreateParams.Builder builder =
+ ChatCompletionCreateParams.builder().model(model);
+ String effectiveSystemPrompt =
+ configuredSystemPrompt == null
+ ? systemPrompt
+ : configuredSystemPrompt + "\n" + systemPrompt;
+ builder.addSystemMessage(effectiveSystemPrompt);
+ if (params.contentType ==
OpenAiCompatibleModelOptions.ContentType.IMAGE_URL) {
+ ChatCompletionContentPartImage.ImageUrl imageUrl =
+
ChatCompletionContentPartImage.ImageUrl.builder().url(userInput).build();
+ ChatCompletionContentPartImage image =
+
ChatCompletionContentPartImage.builder().imageUrl(imageUrl).build();
+ builder.addUserMessageOfArrayOfContentParts(
+
Collections.singletonList(ChatCompletionContentPart.ofImageUrl(image)));
+ } else {
+ builder.addUserMessage(userInput);
+ }
+ if (params.userPrompt != null) {
+ builder.addUserMessage(params.userPrompt);
+ }
+ applyCompletionParams(builder);
+ builder.putAllAdditionalHeaders(headersOrEmpty());
+ builder.putAllAdditionalBodyProperties(bodyOrEmpty());
+
+ ChatCompletion completion =
currentClient().chat().completions().create(builder.build());
+ if (completion.choices().isEmpty()) {
+ throw new IllegalStateException(
+ "OpenAI-compatible text completion returned no choices.");
+ }
+ return completion
+ .choices()
+ .get(0)
+ .message()
+ .content()
+ .orElseThrow(
+ () ->
+ new IllegalStateException(
+ "OpenAI-compatible text completion
returned no text content."));
+ }
+
+ private void applyCompletionParams(ChatCompletionCreateParams.Builder
builder) {
+ if (params.temperature != null) {
+ builder.temperature(params.temperature);
+ }
+ if (params.topP != null) {
+ builder.topP(params.topP);
+ }
+ if (params.stop != null) {
+ builder.stop(params.stop);
+ }
+ if (params.maxTokens != null) {
+ builder.maxTokens(params.maxTokens);
+ }
+ if (params.presencePenalty != null) {
+ builder.presencePenalty(params.presencePenalty);
+ }
+ if (params.frequencyPenalty != null) {
+ builder.frequencyPenalty(params.frequencyPenalty);
+ }
+ if (params.n != null) {
+ builder.n(params.n);
+ }
+ if (params.seed != null) {
+ builder.seed(params.seed);
+ }
+ if (params.responseFormat != null) {
+ builder.responseFormat(params.responseFormat.toResponseFormat());
+ }
+ }
+
+ private float[] createEmbedding(String text) {
+ EmbeddingCreateParams.Builder builder =
+ EmbeddingCreateParams.builder().model(model).input(text);
+ if (params.dimension != null) {
+ builder.dimensions(params.dimension);
+ }
+ builder.putAllAdditionalHeaders(headersOrEmpty());
+ builder.putAllAdditionalBodyProperties(bodyOrEmpty());
+
+ CreateEmbeddingResponse response =
currentClient().embeddings().create(builder.build());
+ List<Embedding> data = response.data();
+ if (data.isEmpty()) {
+ return new float[0];
+ }
+ List<Float> embedding = data.get(0).embedding();
+ float[] result = new float[embedding.size()];
+ for (int i = 0; i < result.length; i++) {
+ result[i] = embedding.get(i);
+ }
+ return result;
+ }
+
+ private <T> T executeWithRetry(String operation, Supplier<T> action) {
+ int maximumAttempts =
+ params.errorHandlingStrategy == ErrorHandlingStrategy.RETRY ?
params.retryNum : 1;
+ long intervalMillis = params.retryBackoffBaseIntervalMillis;
+ RuntimeException lastException = null;
+ int attempts = 0;
+ for (; attempts < maximumAttempts; attempts++) {
+ try {
+ return action.get();
+ } catch (RuntimeException e) {
+ lastException = e;
+ boolean hasAnotherAttempt = attempts + 1 < maximumAttempts;
+ if (!hasAnotherAttempt || !isRetryable(e)) {
+ attempts++;
+ break;
+ }
+ LOG.warn(
+ "OpenAI-compatible {} request failed on attempt {}.
Retrying in {} ms. Cause: {}",
+ operation,
+ attempts + 1,
+ intervalMillis,
+ e.toString());
+ sleepBeforeRetry(intervalMillis);
+ intervalMillis =
params.retryBackoffStrategy.nextInterval(intervalMillis);
+ }
+ }
+
+ ErrorHandlingStrategy finalStrategy =
+ params.errorHandlingStrategy == ErrorHandlingStrategy.RETRY
+ ? params.retryFallbackStrategy
+ : params.errorHandlingStrategy;
+ if (finalStrategy == ErrorHandlingStrategy.IGNORE) {
+ LOG.warn(
+ "OpenAI-compatible {} request failed after {} attempt(s).
Ignoring the input. Cause: {}",
+ operation,
+ attempts,
+ lastException);
+ return null;
+ }
+ throw new RuntimeException(
+ String.format(
+ "OpenAI-compatible %s request failed after %s
attempt(s).",
+ operation, attempts),
+ lastException);
+ }
+
+ private static void sleepBeforeRetry(long intervalMillis) {
+ try {
+ Thread.sleep(intervalMillis);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted while backing off an
OpenAI request.", e);
+ }
+ }
+
+ private static boolean isRetryable(RuntimeException exception) {
+ if (hasThrowable(exception, IOException.class)
+ || hasThrowable(exception, OpenAIIoException.class)
+ || hasThrowable(exception, OpenAIRetryableException.class)) {
+ return true;
+ }
+ for (Throwable throwable = exception; throwable != null; throwable =
throwable.getCause()) {
+ if (throwable instanceof OpenAIServiceException) {
+ int statusCode = ((OpenAIServiceException)
throwable).statusCode();
+ return statusCode == 408
+ || statusCode == 409
+ || statusCode == 429
+ || (statusCode >= 500 && statusCode < 600);
+ }
+ }
+ return false;
+ }
+
+ private OpenAIClient currentClient() {
+ if (client == null) {
+ throw new IllegalStateException("OpenAI-compatible model client
has not been opened.");
+ }
+ return client;
+ }
+
+ private Map<String, List<String>> headersOrEmpty() {
+ return additionalHeaders == null ? Collections.emptyMap() :
additionalHeaders;
+ }
+
+ private Map<String, JsonValue> bodyOrEmpty() {
+ return additionalBody == null ? Collections.emptyMap() :
additionalBody;
+ }
+
+ private static Map<String, List<String>> parseHeaders(@Nullable String
headerJson) {
+ if (headerJson == null || headerJson.trim().isEmpty()) {
+ return Collections.emptyMap();
+ }
+ JsonNode root = parseJsonObject(headerJson, "extra-header");
+ Map<String, List<String>> headers = new HashMap<>();
+ for (Iterator<Map.Entry<String, JsonNode>> fields = root.fields();
fields.hasNext(); ) {
+ Map.Entry<String, JsonNode> field = fields.next();
+ List<String> values = new ArrayList<>();
+ if (field.getValue().isArray()) {
+ for (JsonNode value : field.getValue()) {
+ values.add(headerValue(field.getKey(), value));
+ }
+ } else {
+ values.add(headerValue(field.getKey(), field.getValue()));
+ }
+ headers.put(field.getKey(), values);
+ }
+ return headers;
+ }
+
+ private static String headerValue(String name, JsonNode value) {
+ if (!value.isValueNode() || value.isNull()) {
+ throw new IllegalArgumentException(
+ String.format("Header '%s' must contain a scalar JSON
value.", name));
+ }
+ return value.asText();
+ }
+
+ private static Map<String, JsonValue> parseBody(@Nullable String bodyJson)
{
+ if (bodyJson == null || bodyJson.trim().isEmpty()) {
+ return Collections.emptyMap();
+ }
+ JsonNode root = parseJsonObject(bodyJson, "extra-body");
+ Map<String, JsonValue> body = new HashMap<>();
+ for (Iterator<Map.Entry<String, JsonNode>> fields = root.fields();
fields.hasNext(); ) {
+ Map.Entry<String, JsonNode> field = fields.next();
+ body.put(field.getKey(), JsonValue.fromJsonNode(field.getValue()));
+ }
+ return body;
+ }
+
+ private static JsonNode parseJsonObject(String json, String option) {
+ try {
+ JsonNode root = new ObjectMapper().readTree(json);
+ if (root == null || !root.isObject()) {
+ throw new IllegalArgumentException(
+ String.format("Option '%s' must be a JSON object.",
option));
+ }
+ return root;
+ } catch (JsonProcessingException e) {
+ throw new IllegalArgumentException(
+ String.format("Option '%s' contains invalid JSON.",
option), e);
+ }
+ }
+
+ private static boolean hasThrowable(
+ Throwable exception, Class<? extends Throwable> targetClass) {
+ for (Throwable throwable = exception; throwable != null; throwable =
throwable.getCause()) {
+ if (targetClass.isInstance(throwable)) {
+ return true;
+ }
+ }
+ return false;
+ }
+}
diff --git
a/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/src/main/java/org/apache/flink/cdc/models/openai/OpenAiCompatibleModelClientFactory.java
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/src/main/java/org/apache/flink/cdc/models/openai/OpenAiCompatibleModelClientFactory.java
new file mode 100644
index 000000000..576488b43
--- /dev/null
+++
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/src/main/java/org/apache/flink/cdc/models/openai/OpenAiCompatibleModelClientFactory.java
@@ -0,0 +1,90 @@
+/*
+ * 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.models.openai;
+
+import org.apache.flink.cdc.common.configuration.ConfigOption;
+import org.apache.flink.cdc.common.configuration.Configuration;
+import org.apache.flink.cdc.common.factories.FactoryHelper;
+import org.apache.flink.cdc.common.model.AiModelClient;
+import org.apache.flink.cdc.common.model.AiModelClientFactory;
+import org.apache.flink.cdc.common.model.ModelContext;
+
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/** SPI factory for {@link OpenAiCompatibleModelClient}. */
+public class OpenAiCompatibleModelClientFactory implements
AiModelClientFactory {
+
+ private static final Set<ConfigOption<?>> REQUIRED_OPTIONS =
+ Set.of(
+ OpenAiCompatibleModelOptions.MODEL,
+ OpenAiCompatibleModelOptions.ENDPOINT,
+ OpenAiCompatibleModelOptions.API_KEY);
+
+ private static final Set<ConfigOption<?>> OPTIONAL_OPTIONS =
+ OpenAiCompatibleModelOptions.ALL_OPTIONS.stream()
+ .filter(option -> !REQUIRED_OPTIONS.contains(option))
+ .collect(Collectors.toSet());
+
+ @Override
+ public String identifier() {
+ return "openai-compatible";
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ return REQUIRED_OPTIONS;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ return OPTIONAL_OPTIONS;
+ }
+
+ @Override
+ public AiModelClient createClient(ModelContext context) {
+ validate(context);
+ Configuration options = Configuration.fromMap(context.getOptions());
+ String endpoint =
+
requireNonBlank(options.get(OpenAiCompatibleModelOptions.ENDPOINT), "endpoint");
+ String apiKey =
+
requireNonBlank(options.get(OpenAiCompatibleModelOptions.API_KEY), "api-key");
+ String model =
requireNonBlank(options.get(OpenAiCompatibleModelOptions.MODEL), "model");
+ String systemPrompt =
options.get(OpenAiCompatibleModelOptions.SYSTEM_PROMPT);
+ OpenAiRequestParams params = OpenAiRequestParams.fromOptions(options);
+ return new OpenAiCompatibleModelClient(endpoint, apiKey, model,
systemPrompt, params);
+ }
+
+ void validate(ModelContext context) {
+ FactoryHelper.createFactoryHelper(
+ this,
+ new FactoryHelper.DefaultContext(
+ Configuration.fromMap(context.getOptions()),
+ new Configuration(),
+ context.getClassLoader()))
+ .validate();
+ }
+
+ private static String requireNonBlank(String value, String option) {
+ if (value == null || value.trim().isEmpty()) {
+ throw new IllegalArgumentException(
+ String.format("Option '%s' must not be blank.", option));
+ }
+ return value;
+ }
+}
diff --git
a/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/src/main/java/org/apache/flink/cdc/models/openai/OpenAiCompatibleModelOptions.java
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/src/main/java/org/apache/flink/cdc/models/openai/OpenAiCompatibleModelOptions.java
new file mode 100644
index 000000000..3f71787cf
--- /dev/null
+++
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/src/main/java/org/apache/flink/cdc/models/openai/OpenAiCompatibleModelOptions.java
@@ -0,0 +1,225 @@
+/*
+ * 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.models.openai;
+
+import org.apache.flink.cdc.common.configuration.ConfigOption;
+import org.apache.flink.cdc.common.configuration.ConfigOptions;
+
+import com.openai.models.ResponseFormatJsonObject;
+import com.openai.models.ResponseFormatText;
+import com.openai.models.chat.completions.ChatCompletionCreateParams;
+
+import java.time.Duration;
+import java.util.Set;
+
+/** Config options accepted by the {@code openai-compatible} pipeline model. */
+public class OpenAiCompatibleModelOptions {
+
+ public static final ConfigOption<String> MODEL =
+ ConfigOptions.key("model")
+ .stringType()
+ .noDefaultValue()
+ .withDeprecatedKeys("model-name")
+ .withDescription("Name of the model to invoke.");
+
+ public static final ConfigOption<String> ENDPOINT =
+ ConfigOptions.key("endpoint")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Base URL of the OpenAI-compatible
endpoint.");
+
+ public static final ConfigOption<String> API_KEY =
+ ConfigOptions.key("api-key")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("API key used to authenticate against the
endpoint.");
+
+ public static final ConfigOption<String> SYSTEM_PROMPT =
+ ConfigOptions.key("system-prompt")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("System prompt prepended to every text
completion request.");
+
+ public static final ConfigOption<String> USER_PROMPT =
+ ConfigOptions.key("user-prompt")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Additional user prompt appended after
the input.");
+
+ public static final ConfigOption<Double> TEMPERATURE =
+ ConfigOptions.key("temperature")
+ .doubleType()
+ .noDefaultValue()
+ .withDescription("Sampling temperature.");
+
+ public static final ConfigOption<Double> TOP_P =
+ ConfigOptions.key("top-p")
+ .doubleType()
+ .noDefaultValue()
+ .withDescription("Nucleus sampling probability mass.");
+
+ public static final ConfigOption<String> STOP =
+ ConfigOptions.key("stop")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Stop sequence that ends generation.");
+
+ public static final ConfigOption<Integer> MAX_TOKENS =
+ ConfigOptions.key("max-tokens")
+ .intType()
+ .noDefaultValue()
+ .withDescription("Maximum number of tokens to generate.");
+
+ public static final ConfigOption<Double> PRESENCE_PENALTY =
+ ConfigOptions.key("presence-penalty")
+ .doubleType()
+ .noDefaultValue()
+ .withDescription("Presence penalty applied during
generation.");
+
+ public static final ConfigOption<Double> FREQUENCY_PENALTY =
+ ConfigOptions.key("frequency-penalty")
+ .doubleType()
+ .noDefaultValue()
+ .withDescription("Frequency penalty applied during
generation.");
+
+ public static final ConfigOption<Long> N =
+ ConfigOptions.key("n")
+ .longType()
+ .noDefaultValue()
+ .withDescription("Number of chat completion choices to
generate.");
+
+ public static final ConfigOption<Long> SEED =
+ ConfigOptions.key("seed")
+ .longType()
+ .noDefaultValue()
+ .withDescription("Seed for deterministic sampling.");
+
+ public static final ConfigOption<ResponseFormat> RESPONSE_FORMAT =
+ ConfigOptions.key("response-format")
+ .enumType(ResponseFormat.class)
+ .noDefaultValue()
+ .withDescription("Response format of the chat
completion.");
+
+ public static final ConfigOption<ContentType> CONTENT_TYPE =
+ ConfigOptions.key("content-type")
+ .enumType(ContentType.class)
+ .defaultValue(ContentType.TEXT)
+ .withDescription("Content type of the model input.");
+
+ public static final ConfigOption<String> EXTRA_HEADER =
+ ConfigOptions.key("extra-header")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Additional HTTP headers as a JSON
object.");
+
+ public static final ConfigOption<String> EXTRA_BODY =
+ ConfigOptions.key("extra-body")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Additional request-body properties as a
JSON object.");
+
+ public static final ConfigOption<Integer> DIMENSION =
+ ConfigOptions.key("dimension")
+ .intType()
+ .noDefaultValue()
+ .withDescription("Number of dimensions for the embedding
output.");
+
+ public static final ConfigOption<ErrorHandlingStrategy>
ERROR_HANDLING_STRATEGY =
+ ConfigOptions.key("error-handling-strategy")
+ .enumType(ErrorHandlingStrategy.class)
+ .defaultValue(ErrorHandlingStrategy.RETRY)
+ .withDescription("Strategy applied when a request fails.");
+
+ public static final ConfigOption<Integer> RETRY_NUM =
+ ConfigOptions.key("retry-num")
+ .intType()
+ .defaultValue(100)
+ .withDescription("Maximum number of attempts when retry is
enabled.");
+
+ public static final ConfigOption<ErrorHandlingStrategy>
RETRY_FALLBACK_STRATEGY =
+ ConfigOptions.key("retry-fallback-strategy")
+ .enumType(ErrorHandlingStrategy.class)
+ .defaultValue(ErrorHandlingStrategy.FAILOVER)
+ .withDescription("Strategy applied after retries are
exhausted.");
+
+ public static final ConfigOption<RetryBackoffStrategy>
RETRY_BACKOFF_STRATEGY =
+ ConfigOptions.key("retry-backoff-strategy")
+ .enumType(RetryBackoffStrategy.class)
+ .defaultValue(RetryBackoffStrategy.FIXED)
+ .withDescription("Backoff strategy between retry
attempts.");
+
+ public static final ConfigOption<Duration> RETRY_BACKOFF_BASE_INTERVAL =
+ ConfigOptions.key("retry-backoff-base-interval")
+ .durationType()
+ .defaultValue(Duration.ofSeconds(1))
+ .withDescription("Base interval between retry attempts.");
+
+ public static final Set<ConfigOption<?>> ALL_OPTIONS =
+ Set.of(
+ MODEL,
+ ENDPOINT,
+ API_KEY,
+ SYSTEM_PROMPT,
+ USER_PROMPT,
+ TEMPERATURE,
+ TOP_P,
+ STOP,
+ MAX_TOKENS,
+ PRESENCE_PENALTY,
+ FREQUENCY_PENALTY,
+ N,
+ SEED,
+ RESPONSE_FORMAT,
+ CONTENT_TYPE,
+ EXTRA_HEADER,
+ EXTRA_BODY,
+ DIMENSION,
+ ERROR_HANDLING_STRATEGY,
+ RETRY_NUM,
+ RETRY_FALLBACK_STRATEGY,
+ RETRY_BACKOFF_STRATEGY,
+ RETRY_BACKOFF_BASE_INTERVAL);
+
+ /** Format of a text completion response. */
+ public enum ResponseFormat {
+ TEXT {
+ @Override
+ ChatCompletionCreateParams.ResponseFormat toResponseFormat() {
+ return ChatCompletionCreateParams.ResponseFormat.ofText(
+ ResponseFormatText.builder().build());
+ }
+ },
+ JSON_OBJECT {
+ @Override
+ ChatCompletionCreateParams.ResponseFormat toResponseFormat() {
+ return ChatCompletionCreateParams.ResponseFormat.ofJsonObject(
+ ResponseFormatJsonObject.builder().build());
+ }
+ };
+
+ abstract ChatCompletionCreateParams.ResponseFormat toResponseFormat();
+ }
+
+ /** Content type of the user input. */
+ public enum ContentType {
+ TEXT,
+ IMAGE_URL
+ }
+
+ private OpenAiCompatibleModelOptions() {}
+}
diff --git
a/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/src/main/java/org/apache/flink/cdc/models/openai/OpenAiRequestParams.java
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/src/main/java/org/apache/flink/cdc/models/openai/OpenAiRequestParams.java
new file mode 100644
index 000000000..cfb5bd967
--- /dev/null
+++
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/src/main/java/org/apache/flink/cdc/models/openai/OpenAiRequestParams.java
@@ -0,0 +1,194 @@
+/*
+ * 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.models.openai;
+
+import org.apache.flink.cdc.common.configuration.Configuration;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+import java.time.Duration;
+
+/** Parsed request and retry parameters for the OpenAI-compatible model
client. */
+class OpenAiRequestParams implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ @Nullable final String userPrompt;
+ @Nullable final Double temperature;
+ @Nullable final Double topP;
+ @Nullable final String stop;
+ @Nullable final Integer maxTokens;
+ @Nullable final Double presencePenalty;
+ @Nullable final Double frequencyPenalty;
+ @Nullable final Long n;
+ @Nullable final Long seed;
+ @Nullable final OpenAiCompatibleModelOptions.ResponseFormat responseFormat;
+ final OpenAiCompatibleModelOptions.ContentType contentType;
+ @Nullable final String extraHeader;
+ @Nullable final String extraBody;
+ @Nullable final Integer dimension;
+ final ErrorHandlingStrategy errorHandlingStrategy;
+ final int retryNum;
+ final ErrorHandlingStrategy retryFallbackStrategy;
+ final RetryBackoffStrategy retryBackoffStrategy;
+ final long retryBackoffBaseIntervalMillis;
+
+ private OpenAiRequestParams(
+ @Nullable String userPrompt,
+ @Nullable Double temperature,
+ @Nullable Double topP,
+ @Nullable String stop,
+ @Nullable Integer maxTokens,
+ @Nullable Double presencePenalty,
+ @Nullable Double frequencyPenalty,
+ @Nullable Long n,
+ @Nullable Long seed,
+ @Nullable OpenAiCompatibleModelOptions.ResponseFormat
responseFormat,
+ OpenAiCompatibleModelOptions.ContentType contentType,
+ @Nullable String extraHeader,
+ @Nullable String extraBody,
+ @Nullable Integer dimension,
+ ErrorHandlingStrategy errorHandlingStrategy,
+ int retryNum,
+ ErrorHandlingStrategy retryFallbackStrategy,
+ RetryBackoffStrategy retryBackoffStrategy,
+ long retryBackoffBaseIntervalMillis) {
+ this.userPrompt = userPrompt;
+ this.temperature = temperature;
+ this.topP = topP;
+ this.stop = stop;
+ this.maxTokens = maxTokens;
+ this.presencePenalty = presencePenalty;
+ this.frequencyPenalty = frequencyPenalty;
+ this.n = n;
+ this.seed = seed;
+ this.responseFormat = responseFormat;
+ this.contentType = contentType;
+ this.extraHeader = extraHeader;
+ this.extraBody = extraBody;
+ this.dimension = dimension;
+ this.errorHandlingStrategy = errorHandlingStrategy;
+ this.retryNum = retryNum;
+ this.retryFallbackStrategy = retryFallbackStrategy;
+ this.retryBackoffStrategy = retryBackoffStrategy;
+ this.retryBackoffBaseIntervalMillis = retryBackoffBaseIntervalMillis;
+ }
+
+ static OpenAiRequestParams fromOptions(Configuration options) {
+ String userPrompt =
options.get(OpenAiCompatibleModelOptions.USER_PROMPT);
+ Double temperature =
options.get(OpenAiCompatibleModelOptions.TEMPERATURE);
+ Double topP = options.get(OpenAiCompatibleModelOptions.TOP_P);
+ String stop = options.get(OpenAiCompatibleModelOptions.STOP);
+ Integer maxTokens =
options.get(OpenAiCompatibleModelOptions.MAX_TOKENS);
+ Double presencePenalty =
options.get(OpenAiCompatibleModelOptions.PRESENCE_PENALTY);
+ Double frequencyPenalty =
options.get(OpenAiCompatibleModelOptions.FREQUENCY_PENALTY);
+ Long n = options.get(OpenAiCompatibleModelOptions.N);
+ Long seed = options.get(OpenAiCompatibleModelOptions.SEED);
+ OpenAiCompatibleModelOptions.ResponseFormat responseFormat =
+ options.get(OpenAiCompatibleModelOptions.RESPONSE_FORMAT);
+ OpenAiCompatibleModelOptions.ContentType contentType =
+ options.get(OpenAiCompatibleModelOptions.CONTENT_TYPE);
+ String extraHeader =
options.get(OpenAiCompatibleModelOptions.EXTRA_HEADER);
+ String extraBody =
options.get(OpenAiCompatibleModelOptions.EXTRA_BODY);
+ Integer dimension =
options.get(OpenAiCompatibleModelOptions.DIMENSION);
+ ErrorHandlingStrategy errorHandlingStrategy =
+
options.get(OpenAiCompatibleModelOptions.ERROR_HANDLING_STRATEGY);
+ int retryNum = options.get(OpenAiCompatibleModelOptions.RETRY_NUM);
+ ErrorHandlingStrategy retryFallbackStrategy =
+
options.get(OpenAiCompatibleModelOptions.RETRY_FALLBACK_STRATEGY);
+ RetryBackoffStrategy retryBackoffStrategy =
+
options.get(OpenAiCompatibleModelOptions.RETRY_BACKOFF_STRATEGY);
+ long retryBackoffBaseIntervalMillis =
+
options.get(OpenAiCompatibleModelOptions.RETRY_BACKOFF_BASE_INTERVAL).toMillis();
+
+ validateRange("temperature", temperature, 0.0d, 2.0d);
+ validateRange("top-p", topP, 0.0d, 1.0d);
+ validateRange("presence-penalty", presencePenalty, -2.0d, 2.0d);
+ validateRange("frequency-penalty", frequencyPenalty, -2.0d, 2.0d);
+ validatePositive("max-tokens", maxTokens);
+ validatePositive("n", n);
+ validatePositive("dimension", dimension);
+ if (responseFormat ==
OpenAiCompatibleModelOptions.ResponseFormat.TEXT) {
+ throw new IllegalArgumentException(
+ "Only 'json_object' is supported for option
'response-format', because the "
+ + "built-in AI completion function parses model
output as JSON.");
+ }
+ if (retryNum < 1) {
+ throw new IllegalArgumentException("Option 'retry-num' must be at
least 1.");
+ }
+ if (retryFallbackStrategy == ErrorHandlingStrategy.RETRY) {
+ throw new IllegalArgumentException("Option
'retry-fallback-strategy' cannot be retry.");
+ }
+ if (retryBackoffBaseIntervalMillis < 0) {
+ throw new IllegalArgumentException(
+ "Option 'retry-backoff-base-interval' cannot be
negative.");
+ }
+ validateRetryDelay(retryBackoffStrategy,
retryBackoffBaseIntervalMillis, retryNum);
+
+ return new OpenAiRequestParams(
+ userPrompt,
+ temperature,
+ topP,
+ stop,
+ maxTokens,
+ presencePenalty,
+ frequencyPenalty,
+ n,
+ seed,
+ responseFormat,
+ contentType,
+ extraHeader,
+ extraBody,
+ dimension,
+ errorHandlingStrategy,
+ retryNum,
+ retryFallbackStrategy,
+ retryBackoffStrategy,
+ retryBackoffBaseIntervalMillis);
+ }
+
+ private static void validateRange(
+ String option, @Nullable Double value, double minimum, double
maximum) {
+ if (value != null && (value < minimum || value > maximum)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Option '%s' must be between %s and %s.", option,
minimum, maximum));
+ }
+ }
+
+ private static void validatePositive(String option, @Nullable Number
value) {
+ if (value != null && value.longValue() < 1) {
+ throw new IllegalArgumentException(
+ String.format("Option '%s' must be at least 1.", option));
+ }
+ }
+
+ private static void validateRetryDelay(
+ RetryBackoffStrategy strategy, long baseIntervalMillis, int
retryNum) {
+ try {
+ strategy.minimumTotalDelay(baseIntervalMillis, retryNum);
+ } catch (ArithmeticException e) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Total retry delay is too large. Base interval:
%s, strategy: %s, attempts: %s.",
+ Duration.ofMillis(baseIntervalMillis), strategy,
retryNum),
+ e);
+ }
+ }
+}
diff --git
a/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/src/main/java/org/apache/flink/cdc/models/openai/RetryBackoffStrategy.java
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/src/main/java/org/apache/flink/cdc/models/openai/RetryBackoffStrategy.java
new file mode 100644
index 000000000..9a6c5307d
--- /dev/null
+++
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/src/main/java/org/apache/flink/cdc/models/openai/RetryBackoffStrategy.java
@@ -0,0 +1,54 @@
+/*
+ * 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.models.openai;
+
+/** Strategy for calculating the delay between retry attempts. */
+public enum RetryBackoffStrategy {
+ FIXED {
+ @Override
+ long nextInterval(long currentIntervalMillis) {
+ return currentIntervalMillis;
+ }
+
+ @Override
+ long minimumTotalDelay(long baseIntervalMillis, int attempts) {
+ return Math.multiplyExact(baseIntervalMillis, Math.max(0, attempts
- 1));
+ }
+ },
+
+ EXPONENTIAL {
+ @Override
+ long nextInterval(long currentIntervalMillis) {
+ return Math.multiplyExact(currentIntervalMillis, 2L);
+ }
+
+ @Override
+ long minimumTotalDelay(long baseIntervalMillis, int attempts) {
+ int delays = Math.max(0, attempts - 1);
+ if (delays >= Long.SIZE - 1) {
+ throw new ArithmeticException("Too many exponential retry
attempts");
+ }
+ long multiplier = Math.subtractExact(1L << delays, 1L);
+ return Math.multiplyExact(baseIntervalMillis, multiplier);
+ }
+ };
+
+ abstract long nextInterval(long currentIntervalMillis);
+
+ abstract long minimumTotalDelay(long baseIntervalMillis, int attempts);
+}
diff --git
a/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/src/main/resources/META-INF/NOTICE
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/src/main/resources/META-INF/NOTICE
new file mode 100644
index 000000000..b5ee44f11
--- /dev/null
+++
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/src/main/resources/META-INF/NOTICE
@@ -0,0 +1,34 @@
+flink-cdc-pipeline-model-openai-compatible
+Copyright 2024-2026 The Apache Software Foundation
+
+This product includes software developed at
+The Apache Software Foundation (http://www.apache.org/).
+
+This project bundles the following dependencies under the Apache Software
License 2.0
+(https://www.apache.org/licenses/LICENSE-2.0):
+
+- com.openai:openai-java:4.32.0
+- com.openai:openai-java-client-okhttp:4.32.0
+- com.openai:openai-java-core:4.32.0
+- com.fasterxml.jackson.core:jackson-annotations:2.18.2
+- com.fasterxml.jackson.core:jackson-core:2.18.2
+- com.fasterxml.jackson.core:jackson-databind:2.18.2
+- com.fasterxml.jackson.datatype:jackson-datatype-jdk8:2.18.2
+- com.fasterxml.jackson.datatype:jackson-datatype-jsr310:2.18.2
+- com.fasterxml.jackson.module:jackson-module-kotlin:2.18.2
+- com.fasterxml:classmate:1.7.0
+- com.github.victools:jsonschema-generator:4.38.0
+- com.github.victools:jsonschema-module-jackson:4.38.0
+- com.github.victools:jsonschema-module-swagger-2:4.38.0
+- com.google.errorprone:error_prone_annotations:2.33.0
+- com.squareup.okhttp3:logging-interceptor:4.12.0
+- com.squareup.okhttp3:okhttp:4.12.0
+- com.squareup.okio:okio:3.6.0
+- com.squareup.okio:okio-jvm:3.6.0
+- io.swagger.core.v3:swagger-annotations:2.2.31
+- org.jetbrains:annotations:13.0
+- org.jetbrains.kotlin:kotlin-stdlib:1.8.0
+- org.jetbrains.kotlin:kotlin-stdlib-common:1.8.0
+- org.jetbrains.kotlin:kotlin-stdlib-jdk7:1.8.0
+- org.jetbrains.kotlin:kotlin-stdlib-jdk8:1.8.0
+- org.jetbrains.kotlin:kotlin-reflect:1.8.10
diff --git
a/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/src/main/resources/META-INF/services/org.apache.flink.cdc.common.factories.Factory
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/src/main/resources/META-INF/services/org.apache.flink.cdc.common.factories.Factory
new file mode 100644
index 000000000..017156db4
--- /dev/null
+++
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/src/main/resources/META-INF/services/org.apache.flink.cdc.common.factories.Factory
@@ -0,0 +1,16 @@
+# 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.
+
+org.apache.flink.cdc.models.openai.OpenAiCompatibleModelClientFactory
diff --git
a/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/src/test/java/org/apache/flink/cdc/models/openai/OpenAiCompatibleModelClientFactoryTest.java
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/src/test/java/org/apache/flink/cdc/models/openai/OpenAiCompatibleModelClientFactoryTest.java
new file mode 100644
index 000000000..9265683d9
--- /dev/null
+++
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/src/test/java/org/apache/flink/cdc/models/openai/OpenAiCompatibleModelClientFactoryTest.java
@@ -0,0 +1,147 @@
+/*
+ * 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.models.openai;
+
+import org.apache.flink.cdc.common.factories.Factory;
+import org.apache.flink.cdc.common.model.AiModelClient;
+import org.apache.flink.cdc.common.model.ModelContext;
+import org.apache.flink.cdc.common.model.abilities.SupportsEmbedding;
+import org.apache.flink.cdc.common.model.abilities.SupportsTextGeneration;
+import org.apache.flink.table.api.ValidationException;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.ServiceLoader;
+import java.util.stream.StreamSupport;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link OpenAiCompatibleModelClientFactory}. */
+class OpenAiCompatibleModelClientFactoryTest {
+
+ private final OpenAiCompatibleModelClientFactory factory =
+ new OpenAiCompatibleModelClientFactory();
+
+ @Test
+ void testIdentifierAndOptions() {
+ assertThat(factory.identifier()).isEqualTo("openai-compatible");
+ assertThat(factory.requiredOptions())
+ .extracting(option -> option.key())
+ .containsExactlyInAnyOrder("model", "endpoint", "api-key");
+ assertThat(factory.optionalOptions())
+ .extracting(option -> option.key())
+ .contains(
+ "system-prompt",
+ "temperature",
+ "top-p",
+ "retry-num",
+ "retry-backoff-strategy",
+ "dimension")
+ .doesNotContain("model", "endpoint", "api-key");
+ }
+
+ @Test
+ void testFactoryIsDiscoverable() {
+ assertThat(
+
StreamSupport.stream(ServiceLoader.load(Factory.class).spliterator(), false)
+
.filter(OpenAiCompatibleModelClientFactory.class::isInstance))
+ .hasSize(1);
+ }
+
+ @Test
+ void testCreateClient() {
+ AiModelClient client = factory.createClient(context(validOptions()));
+
+ assertThat(client)
+ .isInstanceOf(OpenAiCompatibleModelClient.class)
+ .isInstanceOf(SupportsTextGeneration.class)
+ .isInstanceOf(SupportsEmbedding.class);
+ }
+
+ @Test
+ void testDeprecatedModelNameAlias() {
+ Map<String, String> options = validOptions();
+ options.put("model-name", options.remove("model"));
+
+ assertThat(factory.createClient(context(options)))
+ .isInstanceOf(OpenAiCompatibleModelClient.class);
+ }
+
+ @Test
+ void testMissingRequiredOptionIsRejected() {
+ Map<String, String> options = validOptions();
+ options.remove("api-key");
+
+ assertThatThrownBy(() -> factory.createClient(context(options)))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("required options")
+ .hasMessageContaining("api-key");
+ }
+
+ @Test
+ void testUnknownOptionIsRejected() {
+ Map<String, String> options = validOptions();
+ options.put("unknown-option", "value");
+
+ assertThatThrownBy(() -> factory.createClient(context(options)))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("Unsupported options")
+ .hasMessageContaining("unknown-option");
+ }
+
+ @Test
+ void testBlankOptionIsRejected() {
+ Map<String, String> options = validOptions();
+ options.put("endpoint", " ");
+
+ assertThatThrownBy(() -> factory.createClient(context(options)))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("endpoint")
+ .hasMessageContaining("must not be blank");
+ }
+
+ private static Map<String, String> validOptions() {
+ Map<String, String> options = new HashMap<>();
+ options.put("endpoint", "https://api.example.com/v1";);
+ options.put("api-key", "sk-test");
+ options.put("model", "gpt-test");
+ return options;
+ }
+
+ private static ModelContext context(Map<String, String> options) {
+ return new ModelContext() {
+ @Override
+ public String getModelName() {
+ return "test-model";
+ }
+
+ @Override
+ public Map<String, String> getOptions() {
+ return options;
+ }
+
+ @Override
+ public ClassLoader getClassLoader() {
+ return Thread.currentThread().getContextClassLoader();
+ }
+ };
+ }
+}
diff --git
a/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/src/test/java/org/apache/flink/cdc/models/openai/OpenAiCompatibleModelClientTest.java
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/src/test/java/org/apache/flink/cdc/models/openai/OpenAiCompatibleModelClientTest.java
new file mode 100644
index 000000000..e8352a94e
--- /dev/null
+++
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/src/test/java/org/apache/flink/cdc/models/openai/OpenAiCompatibleModelClientTest.java
@@ -0,0 +1,292 @@
+/*
+ * 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.models.openai;
+
+import org.apache.flink.cdc.common.model.ModelContext;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import okhttp3.mockwebserver.MockResponse;
+import okhttp3.mockwebserver.MockWebServer;
+import okhttp3.mockwebserver.RecordedRequest;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** HTTP-level tests for {@link OpenAiCompatibleModelClient}. */
+class OpenAiCompatibleModelClientTest {
+
+ private static final String COMPLETION_RESPONSE =
+ "{"
+ + "\"id\":\"chatcmpl-test\","
+ + "\"object\":\"chat.completion\","
+ + "\"created\":123,"
+ + "\"model\":\"test-model\","
+ + "\"choices\":[{"
+ + "\"index\":0,"
+ + "\"message\":{\"role\":\"assistant\","
+ + "\"content\":\"{\\\"result\\\":\\\"done\\\"}\"},"
+ + "\"finish_reason\":\"stop\"}],"
+ +
"\"usage\":{\"prompt_tokens\":5,\"completion_tokens\":3,\"total_tokens\":8}"
+ + "}";
+
+ private static final String EMBEDDING_RESPONSE =
+ "{"
+ + "\"object\":\"list\","
+ + "\"data\":[{\"object\":\"embedding\",\"index\":0,"
+ + "\"embedding\":[0.1,-0.2,0.3]}],"
+ + "\"model\":\"test-model\","
+ + "\"usage\":{\"prompt_tokens\":3,\"total_tokens\":3}"
+ + "}";
+
+ private static final String ERROR_RESPONSE =
+ "{\"error\":{\"message\":\"request failed\","
+ +
"\"type\":\"test_error\",\"param\":null,\"code\":\"test\"}}";
+
+ private final ObjectMapper objectMapper = new ObjectMapper();
+
+ private MockWebServer server;
+ private OpenAiCompatibleModelClient client;
+
+ @BeforeEach
+ void setUp() throws IOException {
+ server = new MockWebServer();
+ server.start();
+ }
+
+ @AfterEach
+ void tearDown() throws IOException {
+ if (client != null) {
+ client.close();
+ }
+ server.shutdown();
+ }
+
+ @Test
+ void testTextCompletionWithCommonRequestParameters() throws Exception {
+ server.enqueue(jsonResponse(200, COMPLETION_RESPONSE));
+ Map<String, String> options = baseOptions();
+ options.put("system-prompt", "configured prompt");
+ options.put("user-prompt", "follow up");
+ options.put("temperature", "0.7");
+ options.put("top-p", "0.8");
+ options.put("stop", "END");
+ options.put("max-tokens", "256");
+ options.put("presence-penalty", "1.2");
+ options.put("frequency-penalty", "0.4");
+ options.put("n", "2");
+ options.put("seed", "123");
+ options.put("response-format", "json_object");
+ options.put("extra-header", "{\"X-Test\":\"header-value\"}");
+ options.put("extra-body", "{\"vendor_flag\":true}");
+ client = createAndOpenClient(options);
+
+ assertThat(client.generate("runtime prompt", "input text"))
+ .isEqualTo("{\"result\":\"done\"}");
+
+ RecordedRequest request = server.takeRequest();
+ assertThat(request.getPath()).isEqualTo("/v1/chat/completions");
+ assertThat(request.getHeader("Authorization")).isEqualTo("Bearer
sk-test");
+ assertThat(request.getHeader("X-Test")).isEqualTo("header-value");
+ JsonNode body = objectMapper.readTree(request.getBody().readUtf8());
+ assertThat(body.path("model").asText()).isEqualTo("test-model");
+ assertThat(body.at("/messages/0/role").asText()).isEqualTo("system");
+ assertThat(body.at("/messages/0/content").asText())
+ .isEqualTo("configured prompt\nruntime prompt");
+ assertThat(body.at("/messages/1/content").asText()).isEqualTo("input
text");
+ assertThat(body.at("/messages/2/content").asText()).isEqualTo("follow
up");
+ assertThat(body.path("temperature").asDouble()).isEqualTo(0.7d);
+ assertThat(body.path("top_p").asDouble()).isEqualTo(0.8d);
+ assertThat(body.path("stop").asText()).isEqualTo("END");
+ assertThat(body.path("max_tokens").asInt()).isEqualTo(256);
+ assertThat(body.path("presence_penalty").asDouble()).isEqualTo(1.2d);
+ assertThat(body.path("frequency_penalty").asDouble()).isEqualTo(0.4d);
+ assertThat(body.path("n").asLong()).isEqualTo(2L);
+ assertThat(body.path("seed").asLong()).isEqualTo(123L);
+
assertThat(body.at("/response_format/type").asText()).isEqualTo("json_object");
+ assertThat(body.path("vendor_flag").asBoolean()).isTrue();
+ }
+
+ @Test
+ void testTextCompletionPreservesChinesePromptsAndInput() throws Exception {
+ server.enqueue(jsonResponse(200, COMPLETION_RESPONSE));
+ Map<String, String> options = baseOptions();
+ options.put("system-prompt", "你是一个简洁的助手。");
+ options.put("user-prompt", "请直接回答。");
+ client = createAndOpenClient(options);
+
+ assertThat(client.generate("总结输入内容",
"包含中文的输入文本")).isEqualTo("{\"result\":\"done\"}");
+
+ JsonNode body =
objectMapper.readTree(server.takeRequest().getBody().readUtf8());
+
assertThat(body.at("/messages/0/content").asText()).isEqualTo("你是一个简洁的助手。\n总结输入内容");
+
assertThat(body.at("/messages/1/content").asText()).isEqualTo("包含中文的输入文本");
+
assertThat(body.at("/messages/2/content").asText()).isEqualTo("请直接回答。");
+ }
+
+ @Test
+ void testImageUrlCompletion() throws Exception {
+ server.enqueue(jsonResponse(200, COMPLETION_RESPONSE));
+ Map<String, String> options = baseOptions();
+ options.put("content-type", "image_url");
+ client = createAndOpenClient(options);
+
+ assertThat(client.generate("describe the image",
"https://example.com/image.png";))
+ .contains("done");
+
+ JsonNode body =
objectMapper.readTree(server.takeRequest().getBody().readUtf8());
+
assertThat(body.at("/messages/1/content/0/type").asText()).isEqualTo("image_url");
+ assertThat(body.at("/messages/1/content/0/image_url/url").asText())
+ .isEqualTo("https://example.com/image.png";);
+ }
+
+ @Test
+ void testTextEmbedding() throws Exception {
+ server.enqueue(jsonResponse(200, EMBEDDING_RESPONSE));
+ Map<String, String> options = baseOptions();
+ options.put("dimension", "3");
+ options.put("extra-header", "{\"X-Embedding\":\"enabled\"}");
+ options.put("extra-body", "{\"vendor_flag\":true}");
+ client = createAndOpenClient(options);
+
+ assertThat(client.embed("embed this")).containsExactly(0.1f, -0.2f,
0.3f);
+
+ RecordedRequest request = server.takeRequest();
+ assertThat(request.getPath()).isEqualTo("/v1/embeddings");
+ assertThat(request.getHeader("X-Embedding")).isEqualTo("enabled");
+ JsonNode body = objectMapper.readTree(request.getBody().readUtf8());
+ assertThat(body.path("model").asText()).isEqualTo("test-model");
+ assertThat(body.path("input").asText()).isEqualTo("embed this");
+ assertThat(body.path("dimensions").asInt()).isEqualTo(3);
+ assertThat(body.path("vendor_flag").asBoolean()).isTrue();
+ }
+
+ @Test
+ void testRetryableHttpErrorIsRetried() {
+ server.enqueue(jsonResponse(429, ERROR_RESPONSE));
+ server.enqueue(jsonResponse(200, COMPLETION_RESPONSE));
+ Map<String, String> options = baseOptions();
+ options.put("retry-num", "2");
+ options.put("retry-backoff-base-interval", "1 ms");
+ client = createAndOpenClient(options);
+
+ assertThat(client.generate("system", "input")).contains("done");
+ assertThat(server.getRequestCount()).isEqualTo(2);
+ }
+
+ @Test
+ void testNonRetryableHttpErrorFailsImmediately() {
+ server.enqueue(jsonResponse(400, ERROR_RESPONSE));
+ Map<String, String> options = baseOptions();
+ options.put("retry-num", "3");
+ options.put("retry-backoff-base-interval", "1 ms");
+ client = createAndOpenClient(options);
+
+ assertThatThrownBy(() -> client.generate("system", "input"))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("failed after 1 attempt");
+ assertThat(server.getRequestCount()).isEqualTo(1);
+ }
+
+ @Test
+ void testIgnoreFallbackReturnsNullAfterRetries() {
+ server.enqueue(jsonResponse(500, ERROR_RESPONSE));
+ server.enqueue(jsonResponse(500, ERROR_RESPONSE));
+ Map<String, String> options = baseOptions();
+ options.put("retry-num", "2");
+ options.put("retry-backoff-base-interval", "1 ms");
+ options.put("retry-fallback-strategy", "ignore");
+ client = createAndOpenClient(options);
+
+ assertThat(client.generate("system", "input")).isNull();
+ assertThat(server.getRequestCount()).isEqualTo(2);
+ }
+
+ @Test
+ void testInvalidAdditionalHeadersFailOnOpen() {
+ Map<String, String> options = baseOptions();
+ options.put("extra-header", "not-json");
+ client = createClient(options);
+
+ assertThatThrownBy(client::open)
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("extra-header")
+ .hasMessageContaining("invalid JSON");
+ }
+
+ @Test
+ void testRequestBeforeOpenFailsWithContext() {
+ client = createClient(baseOptions());
+
+ assertThatThrownBy(() -> client.embed("input"))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("embedding request failed")
+ .hasRootCauseMessage("OpenAI-compatible model client has not
been opened.");
+ }
+
+ private OpenAiCompatibleModelClient createAndOpenClient(Map<String,
String> options) {
+ OpenAiCompatibleModelClient result = createClient(options);
+ result.open();
+ return result;
+ }
+
+ private OpenAiCompatibleModelClient createClient(Map<String, String>
options) {
+ return (OpenAiCompatibleModelClient)
+ new
OpenAiCompatibleModelClientFactory().createClient(context(options));
+ }
+
+ private Map<String, String> baseOptions() {
+ Map<String, String> options = new HashMap<>();
+ options.put("endpoint", server.url("/v1").toString());
+ options.put("api-key", "sk-test");
+ options.put("model", "test-model");
+ return options;
+ }
+
+ private static ModelContext context(Map<String, String> options) {
+ return new ModelContext() {
+ @Override
+ public String getModelName() {
+ return "test-model-definition";
+ }
+
+ @Override
+ public Map<String, String> getOptions() {
+ return options;
+ }
+
+ @Override
+ public ClassLoader getClassLoader() {
+ return Thread.currentThread().getContextClassLoader();
+ }
+ };
+ }
+
+ private static MockResponse jsonResponse(int statusCode, String body) {
+ return new MockResponse()
+ .setResponseCode(statusCode)
+ .addHeader("Content-Type", "application/json")
+ .setBody(body);
+ }
+}
diff --git
a/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/src/test/java/org/apache/flink/cdc/models/openai/OpenAiRequestParamsTest.java
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/src/test/java/org/apache/flink/cdc/models/openai/OpenAiRequestParamsTest.java
new file mode 100644
index 000000000..95f2d00a9
--- /dev/null
+++
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/src/test/java/org/apache/flink/cdc/models/openai/OpenAiRequestParamsTest.java
@@ -0,0 +1,125 @@
+/*
+ * 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.models.openai;
+
+import org.apache.flink.cdc.common.configuration.Configuration;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link OpenAiRequestParams}. */
+class OpenAiRequestParamsTest {
+
+ @Test
+ void testParseRequestAndRetryOptions() {
+ Map<String, String> options = new HashMap<>();
+ options.put("user-prompt", "follow up");
+ options.put("temperature", "0.5");
+ options.put("top-p", "0.7");
+ options.put("stop", "END");
+ options.put("max-tokens", "256");
+ options.put("presence-penalty", "1.2");
+ options.put("frequency-penalty", "0.4");
+ options.put("n", "2");
+ options.put("seed", "123");
+ options.put("response-format", "json_object");
+ options.put("content-type", "image_url");
+ options.put("extra-header", "{\"x-test\":\"value\"}");
+ options.put("extra-body", "{\"vendor\":true}");
+ options.put("dimension", "768");
+ options.put("error-handling-strategy", "retry");
+ options.put("retry-num", "3");
+ options.put("retry-fallback-strategy", "ignore");
+ options.put("retry-backoff-strategy", "exponential");
+ options.put("retry-backoff-base-interval", "2 s");
+
+ OpenAiRequestParams params =
+
OpenAiRequestParams.fromOptions(Configuration.fromMap(options));
+
+ assertThat(params.userPrompt).isEqualTo("follow up");
+ assertThat(params.temperature).isEqualTo(0.5d);
+ assertThat(params.topP).isEqualTo(0.7d);
+ assertThat(params.stop).isEqualTo("END");
+ assertThat(params.maxTokens).isEqualTo(256);
+ assertThat(params.presencePenalty).isEqualTo(1.2d);
+ assertThat(params.frequencyPenalty).isEqualTo(0.4d);
+ assertThat(params.n).isEqualTo(2L);
+ assertThat(params.seed).isEqualTo(123L);
+ assertThat(params.responseFormat)
+
.isEqualTo(OpenAiCompatibleModelOptions.ResponseFormat.JSON_OBJECT);
+ assertThat(params.contentType)
+ .isEqualTo(OpenAiCompatibleModelOptions.ContentType.IMAGE_URL);
+ assertThat(params.extraHeader).contains("x-test");
+ assertThat(params.extraBody).contains("vendor");
+ assertThat(params.dimension).isEqualTo(768);
+
assertThat(params.errorHandlingStrategy).isEqualTo(ErrorHandlingStrategy.RETRY);
+ assertThat(params.retryNum).isEqualTo(3);
+
assertThat(params.retryFallbackStrategy).isEqualTo(ErrorHandlingStrategy.IGNORE);
+
assertThat(params.retryBackoffStrategy).isEqualTo(RetryBackoffStrategy.EXPONENTIAL);
+ assertThat(params.retryBackoffBaseIntervalMillis).isEqualTo(2000L);
+ }
+
+ @Test
+ void testRetryDefaults() {
+ OpenAiRequestParams params =
+ OpenAiRequestParams.fromOptions(Configuration.fromMap(new
HashMap<>()));
+
+
assertThat(params.contentType).isEqualTo(OpenAiCompatibleModelOptions.ContentType.TEXT);
+
assertThat(params.errorHandlingStrategy).isEqualTo(ErrorHandlingStrategy.RETRY);
+ assertThat(params.retryNum).isEqualTo(100);
+
assertThat(params.retryFallbackStrategy).isEqualTo(ErrorHandlingStrategy.FAILOVER);
+
assertThat(params.retryBackoffStrategy).isEqualTo(RetryBackoffStrategy.FIXED);
+ assertThat(params.retryBackoffBaseIntervalMillis).isEqualTo(1000L);
+ }
+
+ @Test
+ void testInvalidParameterRangesAreRejected() {
+ assertThatThrownBy(() -> parseOption("temperature", "2.1"))
+ .hasMessageContaining("temperature");
+ assertThatThrownBy(() -> parseOption("top-p",
"-0.1")).hasMessageContaining("top-p");
+ assertThatThrownBy(() -> parseOption("dimension",
"0")).hasMessageContaining("dimension");
+ assertThatThrownBy(() -> parseOption("retry-num",
"0")).hasMessageContaining("retry-num");
+ }
+
+ @Test
+ void testInvalidRetrySettingsAreRejected() {
+ assertThatThrownBy(() -> parseOption("retry-fallback-strategy",
"retry"))
+ .hasMessageContaining("cannot be retry");
+
+ Map<String, String> options = new HashMap<>();
+ options.put("retry-backoff-strategy", "exponential");
+ options.put("retry-num", "100");
+ assertThatThrownBy(() ->
OpenAiRequestParams.fromOptions(Configuration.fromMap(options)))
+ .hasMessageContaining("Total retry delay is too large");
+ }
+
+ @Test
+ void testTextResponseFormatIsRejected() {
+ assertThatThrownBy(() -> parseOption("response-format", "text"))
+ .hasMessageContaining("Only 'json_object' is supported");
+ }
+
+ private static void parseOption(String key, String value) {
+ OpenAiRequestParams.fromOptions(Configuration.fromMap(Map.of(key,
value)));
+ }
+}
diff --git
a/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/src/test/resources/log4j2-test.properties
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/src/test/resources/log4j2-test.properties
new file mode 100644
index 000000000..4a3eb1046
--- /dev/null
+++
b/flink-cdc-pipeline-model/flink-cdc-pipeline-model-openai-compatible/src/test/resources/log4j2-test.properties
@@ -0,0 +1,22 @@
+# 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.
+
+rootLogger.level = INFO
+rootLogger.appenderRef.console.ref = ConsoleAppender
+
+appender.console.name = ConsoleAppender
+appender.console.type = CONSOLE
+appender.console.layout.type = PatternLayout
+appender.console.layout.pattern = %d{HH:mm:ss,SSS} %-5p %-60c %x - %m%n
diff --git a/flink-cdc-pipeline-model/pom.xml b/flink-cdc-pipeline-model/pom.xml
index 1a3e398b7..02b7d4fa0 100644
--- a/flink-cdc-pipeline-model/pom.xml
+++ b/flink-cdc-pipeline-model/pom.xml
@@ -31,6 +31,7 @@ limitations under the License.
<modules>
<module>flink-cdc-pipeline-model-legacy</module>
<module>flink-cdc-pipeline-model-dummy</module>
+ <module>flink-cdc-pipeline-model-openai-compatible</module>
</modules>
</project>
![https://example.com/image.png"](https://example.com/image.png"){kind=link}