配置並調用 AI 推理模型 - Realtime Compute for Apache Flink

本文介紹如何在Flink資料攝入作業中使用內建的 AI 推理模型，實現對資料流的即時智能處理（如文本摘要產生、向量化等）。

前提條件

Realtime Compute引擎：VVR 11.9（含 Preview 預覽版本）及更高版本。
附加依賴包：將 flink-cdc-pipeline-model-openai-compatible JAR 包作為作業附加依賴上傳。

工作原理

AI 推理模型的整合分為兩個步驟：

定義模型：在 YAML 作業草稿的 Pipeline 塊中定義模型，也可以直接引用中繼資料中心中建立的模型。
使用模型：在 YAML 作業草稿的 Transform 塊中調用 AI 推理或向量產生模型。

每個模型需要指定唯一名稱，該名稱將作為函數名在資料轉換中調用。

配置並調用 Flink AI 內建模型

步驟一：先決條件準備

開始前，請先參照 Flink AI服務（內建模型）文檔的說明開通 Flink AI 服務。

此外，您還需要將 flink-cdc-pipeline-model-openai-compatible JAR 包作為作業附加依賴上傳，並附加到作業草稿中。

步驟二：在 Pipeline 塊中定義模型

為了定義 AI 模型，您需要在 YAML 作業草稿的 pipeline 模組中添加 model 欄位列表。您需要為每個模型提供唯一的名稱，且需要符合一般標識符的規則（必須以字母開頭，且只能包含字母、底線及數字）。

必須填寫的參數包括：

模型標識符 name，可任意選擇，需要為合法的標識符
模型類型 type，固定為 openai-compatible
調用百鍊服務中的模型名稱 model

pipeline:
  model:
    - name: CHAT
      type: openai-compatible
      model: qwen-plus
      user-prompt: Please summarize the input.

    - name: GET_EMBEDDING
      type: openai-compatible
      model: text-embedding-v4

以下是您可以指定的完整參數列表：

參數	類型	是否必填	預設值	說明
name	STRING	是	無	Pipeline 中的模型別名，供 AI 函數引用；必須唯一，且符合標識符命名規則。
type	STRING	是	無	模型類型，這裡固定填寫 openai-compatible。
model	STRING	是	無	Flink AI 服務用於推理的實際模型名稱。提供的模型情況請參見內建模型列表。
task	STRING	否	cdc-yaml-ai-task	模型任務名稱，作為監控指標中的 task 分組。
error-handling-strategy	STRING	否	retry	請求失敗後的處理策略；可選值：retry、failover、ignore。
retry-num	INT	否	100	當 error-handling-strategy 為 retry 時，單次模型調用的最大嘗試次數。
retry-fallback-strategy	STRING	否	failover	重試耗盡後的處理策略；可選值：failover、ignore，不能配置為 retry。
retry-backoff-strategy	STRING	否	fixed	稍候再試策略；可選值：fixed、exponential。
retry-backoff-base-interval	DURATION	否	1 s	重試基礎間隔，例如 500 ms、1 s、2 min。
system-prompt	STRING	否	無	添加到請求中的系統提示詞；常規文本產生時會放在 AI 函數自身的系統提示詞之前。
user-prompt	STRING	否	無	在模型輸入之後追加一條使用者提示詞。
temperature	DOUBLE	否	由模型服務決定	採樣溫度。
top-p	DOUBLE	否	由模型服務決定	核採樣機率品質。
stop	STRING	否	無	遇到指定字串時停止產生。
max-tokens	INT	否	由模型服務決定	單次請求允許產生的最大 Token 數。
presence-penalty	DOUBLE	否	由模型服務決定	Presence penalty，用於降低模型重複已有主題的傾向。
n	LONG	否	由模型服務決定	請求產生的候選結果數量；當前用戶端只返回第一條結果。
seed	LONG	否	無	隨機種子；模型服務支援時可用於提高結果的可複現性。
response-format	STRING	否	json_object	響應格式；當前只支援 json_object。
content-type	STRING	否	text	模型輸入類型；可選值：text、image_url。image_url 會將輸入字串作為圖片 URL 發送。
extra-header	STRING	否	無	以 JSON 對象形式聲明附加到請求中的 HTTP Header。
extra-body	STRING	否	無	以 JSON 對象形式聲明附加到請求體中的廠商擴充參數。
dimension	INT	否	由模型服務決定	AI_EMBED 返迴向量的維度；僅在對應模型支援自訂維度時生效。

步驟三：在 Transform 模組中調用 AI Function

按照模態分類，下面是您可以在 Transform 模組中調用的內建 AI 函數。

說明

每個內建 AI 函數的第一個參數均為您在 Pipeline 塊中定義的模型標識符（model 中 name 參數的值）。

文本產生類函數

AI_COMPLETE
- 用途：通用的文本產生函數，可根據給出的提示詞處理輸入文本，適用於內容產生、文本改寫和通用問答。
- 使用樣本：AI_COMPLETE('chat_model', content_string, '提示：請概括輸入內容')
- 返回樣本：{"result":"輸入內容的摘要"}
AI_CLASSIFY
- 用途：將輸入文本分類到指定的候選類別中。多個類別之間使用逗號分隔。
- 使用樣本：AI_CLASSIFY('chat_model', content_string, '新聞,體育,娛樂')
- 返回樣本：{"category":"體育","confidence":0.95}
AI_TRANSLATE
- 用途：將輸入文本從源語言翻譯為目標語言。源語言填寫 auto 時，由模型自動識別。
- 使用樣本：AI_TRANSLATE('chat_model', content_string, 'auto', 'en')
- 返回樣本：{"translated_text":"Hello, world.","detected_language":"zh"}
AI_SUMMARIZE
- 用途：為輸入文本產生摘要。第三個參數表示摘要允許包含的最大字元數。
- 使用樣本：AI_SUMMARIZE('chat_model', content_string, 200)
- 返回樣本：{"summary":"輸入內容的摘要"}
AI_SENTIMENT
- 用途：分析輸入文本的情感傾向。情感分數 score 的取值範圍為 -1.0～1.0。
- 使用樣本：AI_SENTIMENT('chat_model', review_content)
- 返回樣本：{"score":0.8,"label":"positive","confidence":0.92}
AI_EXTRACT
- 用途：按照指定的 Schema 從非結構化文本中提取結構化資訊。
- 使用樣本：AI_EXTRACT('chat_model', content_string, '{"name":"string","age":"integer"}')
- 返回樣本：{"extracted_json":"{\"name\":\"張三\",\"age\":25}"}
AI_MASK
- 用途：識別並遮蓋指定類型的敏感資訊。多個實體類型之間使用逗號分隔。
- 使用樣本：AI_MASK('chat_model', content_string, 'name,phone,email')
- 返回樣本：{"masked_text":"連絡人：張*，手機：138****8000","detected_entities":"name,phone"}

文本向量化函數

AI_EMBED
- 用途：將輸入文本轉換為浮點數向量，可用於語義檢索、相似性計算和文本聚類。
- 使用樣本：AI_EMBED('embedding_model', content_string)
- 返回樣本：[0.0123, -0.0876, 0.2345, ...]

多模態產生類函數

AI_IMAGE_COMPLETE
- 用途：根據圖片位元據和提示詞產生文本，可用於圖片描述、內容識別和視覺問答。
- 使用樣本：AI_IMAGE_COMPLETE('vision_model', image_bytes, '請描述圖片中的商品')
- 返回樣本："圖片中是一雙白色運動鞋。"

多模態向量化函數

AI_IMAGE_EMBED
- 用途：將圖片位元據轉換為浮點數向量，可用於以圖搜圖、圖片聚類和多模態檢索。
- 使用樣本：AI_IMAGE_EMBED('image_embedding_model', image_bytes)
- 返回樣本：[0.0312, -0.1024, 0.2867, ...]

完整配置樣本

以下樣本從 MySQL 讀取文章及圖片資料，調用文本摘要和向量化函數，並寫入 DLF Paimon 資料湖。

重要

預設情況下，AI 推理模型會順序串列求值，受限於網路 I/O 速率可能導致吞吐下降、Checkpoint 過慢等問題。

您可以開啟 transform.async-execution.enabled參數啟用非同步 AI 推理最佳化（實驗性功能），具體的參數功能及限制參見 Pipeline配置參數。

source:
  type: mysql
  using.built-in-catalog: mysql_catalog
  tables: inventory.articles

sink:
  type: paimon
  using.built-in-catalog: dlf_catalog


transform:
  - source-table: inventory.articles
    projection: >-
      id, 
      AI_SUMMARIZE('chat_model', content, 200) AS summary,
      AI_EMBED('embedding_model', content) AS text_embedding,
      AI_IMAGE_COMPLETE('chat_model', image_bytes, '請描述圖片內容') AS image_description,
      AI_IMAGE_EMBED('embedding_model', image_bytes) AS image_embedding

pipeline:
  # 啟用非同步 AI 推理最佳化實驗性特性
  transform.async-execution.enabled: true

  parallelism: 4
  model:
    - name: chat_model
      type: openai-compatible
      model: qwen3.7-max
    - name: embedding_model
      type: openai-compatible
      model: qwen3-vl-embedding
      dimension: 1024

複用 Flink SQL 的 AI 模型定義

重要

目前只支援 Flink SQL 定義的 openai-compat、dashscope 類型的 AI 模型。

建立推理模型

VVR 11.7 及以上版本支援通過 CREATE MODEL 在中繼資料中心建立持久化模型。使用內建模型時，無需配置 endpoint 和 api-key。

CREATE MODEL qwen_chat
USING openai_compatible
WITH (
  'provider' = 'openai-compat',
  'task' = 'chat/completions',
  'model' = 'qwen3.7-max'
);

CREATE MODEL multimodal_embedding
USING openai_compatible
WITH (
  'provider' = 'dashscope',
  'task' = 'multimodal-embedding',
  'model' = 'qwen3-vl-embedding',
  'content-type' = 'image_url',
  'dimension' = '512'
);

複用已有模型

建立完成後，可以在 YAML Pipeline 中通過完整名稱引用 Flink SQL 模型：

pipeline:
  model:
    - name: chat_model
      reuse-model: vvp.default.qwen_chat

    - name: image_embedding_model
      reuse-model: vvp.default.multimodal_embedding