基於YAML CDC實現AI智能文本標註 - Realtime Compute for Apache Flink

方案架構

企業每天產生大量非結構化文本資料（客服工單、使用者評論、投訴記錄等），傳統方式依賴人工逐條標註或T+1離線批處理，成本高、延遲大、標準不一致。

通過CDC YAML的Transform模組調用AI Function，資料在從Kafka流入Paimon資料湖的過程中同步完成智能標註，實現秒級延遲的流式打標。

該方案涵蓋以下典型情境：

客訴風險智能標註：使用AI_COMPLETE對客服工單進行投訴情緒等級判定、曝光意圖識別等多維度風險打標。
內容分類智能標註：使用AI_CLASSIFY對使用者評論或反饋進行預定義類別的自動分類。
敏感資訊自動脫敏：使用AI_MASK對入湖資料中的個人隱私資訊（PII）進行識別和脫敏。

前提條件

已開通Realtime ComputeFlink版並建立工作空間，詳情請參見開通Realtime ComputeFlink版。
已開通Flink AI服務（內建模型）。
僅Realtime Compute引擎版本為VVR 11.9（含preview）及以上支援。
已建立Kafka執行個體，並準備好待標註的文本資料Topic。

使用限制

AI Function在CDC YAML中僅支援在Transform模組的projection運算式中使用。
AI Function的輸送量受阿里雲百鍊平台限流限制。觸及資料傳輸量上限時，Flink作業可能出現反壓或逾時重啟。詳情請參見百鍊平台限流。

目前的版本支援的內建百鍊模型如下：

模型名稱	說明
qwen3.6-flash	通用文本產生，推薦用於標註情境
qwen3.6-plus	更強推理能力
qwen3.5-flash	通用文本產生
qwen3.5-plus	更強推理能力
text-embedding-v4	文本向量化

AI Function速查

CDC YAML的Transform模組支援以下AI Function。

函數	簽名	返回欄位	用途
AI_COMPLETE	`AI_COMPLETE('modelName', input, systemPrompt)`	`result`	通用文本產生
AI_CLASSIFY	`AI_CLASSIFY('modelName', input, labels)`	`category`，`confidence`	文本分類
AI_TRANSLATE	`AI_TRANSLATE('modelName', input, sourceLang, targetLang)`	`translated_text`，`detected_language`	文本翻譯
AI_SUMMARIZE	`AI_SUMMARIZE('modelName', input, maxLength)`	`summary`	文本摘要
AI_SENTIMENT	`AI_SENTIMENT('modelName', input)`	`score`，`label`，`confidence`	情感分析
AI_EXTRACT	`AI_EXTRACT('modelName', input, schema)`	`extracted_json`	結構化資訊提取
AI_MASK	`AI_MASK('modelName', input, entities)`	`masked_text`，`detected_entities`	敏感資訊脫敏

在projection運算式中，通過['欄位名']訪問返回欄位。例如：AI_COMPLETE('mymodel', content, '')['result']。

情境一：客訴風險智能標註

使用AI_COMPLETE對客服工單文本進行多維度風險打標，輸出投訴情緒等級、曝光意圖、品牌抹黑傾向等判定結果。

作業樣本

說明

部署時需要額外的依賴檔案與配置，詳情請參見部署與啟動。

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: <YOUR-KAFKA-BROKER>:9092
  topic: customer_complaints
  value.format: json
  scan.startup.mode: earliest-offset

# 將測試結果輸出到控制台日誌中
sink:
  type: values

# sink:
#  type: paimon
#  catalog.properties.metastore: rest
#  catalog.properties.uri: <YOUR-DLF-ENDPOINT>
#  catalog.properties.warehouse: <YOUR-CATALOG-NAME>
#  catalog.properties.token.provider: dlf
#  table.properties.deletion-vectors.enabled: true

transform:
  - source-table: customer_complaints
    projection: >
      comment,
      AI_COMPLETE('mymodel', comment, '')['result'] as risk_label

pipeline:
  parallelism: 2
  model:
  - name: mymodel
    type: openai-compatible
    model-name: qwen3.6-flash
    system-prompt: '提示詞：
      #你是電商客訴風險判定專家，根據使用者評價內容進行客觀判標。
      #判標規則
      1.投訴情緒等級（三選一）：正常/不滿/憤怒
      2.是否有明確曝光意圖（二選一）：是/否
      3.是否有抹黑品牌傾向（二選一）：是/否
      #輸出（嚴格按以下結構，一行一個）
      投訴情緒等級：xxx
      是否有明確曝光意圖：xxx
      是否有抹黑品牌傾向：xxx
      信賴度：0.xx
      打標依據：xxx
      '

輸入輸出樣本

輸入資料（JSON格式）：

{
    "comment": "買了不到一個月就壞了，聯絡客服一直讓等，等了兩周沒人管。再不處理我，就發到網上曝光投訴你們。"
}

輸出資料：

DataChangeEvent{
  tableId=customer_complaints,
  after=[
    使用者來電反映，2024年3月購買的某品牌空調...,
    "投訴情緒等級：憤怒
     是否有明確曝光意圖：是
     是否有抹黑品牌傾向：是
     信賴度：0.95
     打標依據：再不處理我，就發到網上曝光投訴你們。"
  ],
  op=INSERT, meta=()
}

說明

AI Function輸出欄位的資料類型為VARIANT，便於以JSON格式訪問結構化內容。如果下遊需要特定類型（如STRING或INT），需通過CAST顯式轉換，例如CAST(risk_label AS STRING)。

情境二：內容分類智能標註

使用AI_CLASSIFY對使用者評論或反饋文本進行預定義類別分類。

作業樣本

說明

部署時需要額外的依賴檔案與配置，詳情請參見部署與啟動。

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: <YOUR-KAFKA-BROKER>:9092
  topic: user_reviews
  value.format: json
  scan.startup.mode: earliest-offset
  
# 將測試結果輸出到控制台日誌中  
sink:
  type: values

# sink:
#  type: paimon
#  catalog.properties.metastore: rest
#  catalog.properties.uri: <YOUR-DLF-ENDPOINT>
#  catalog.properties.warehouse: <YOUR-CATALOG-NAME>
#  catalog.properties.token.provider: dlf
#  table.properties.deletion-vectors.enabled: true

transform:
  - source-table: user_reviews
    projection: >
      content,
      AI_CLASSIFY('mymodel', content, 'ARRAY[''產品品質'',''售後服務'',''物流問題'',''價格異議'']')['category'] as category,
      AI_CLASSIFY('mymodel', content, 'ARRAY[''產品品質'',''售後服務'',''物流問題'',''價格異議'']')['confidence'] as confidence

pipeline:
  parallelism: 2
  model:
  - name: mymodel
    type: openai-compatible
    model-name: qwen3.6-flash

輸入輸出樣本

輸入資料：

{"content": "空調噪音太大了，開機就嗡嗡響，完全沒法睡覺"}

輸出資料：

DataChangeEvent{
  tableId=user_reviews, 
  before=[], 
  after=[空調噪音太大了，開機就嗡嗡響，完全沒法睡覺, "產品品質", 0.95], 
  op=INSERT, meta=()
}

情境三：敏感資訊自動脫敏

使用AI_MASK對入湖資料中的個人隱私資訊（姓名、手機號、地址等）進行自動識別和脫敏。

作業樣本

說明

部署時需要額外的依賴檔案與配置，詳情請參見部署與啟動。

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: <YOUR-KAFKA-BROKER>:9092
  topic: user_feedbacks
  value.format: json
  scan.startup.mode: earliest-offset

# 將測試結果輸出到控制台日誌中
sink:
  type: values

# sink:
#  type: paimon
#  catalog.properties.metastore: rest
#  catalog.properties.uri: <YOUR-DLF-ENDPOINT>
#  catalog.properties.warehouse: <YOUR-CATALOG-NAME>
#  catalog.properties.token.provider: dlf
#  table.properties.deletion-vectors.enabled: true

transform:
  - source-table: user_feedbacks
    projection: >
      AI_MASK('mymodel', content, 'ARRAY[''姓名'',''手機號'',''社會安全號碼'',''地址'']')['masked_text'] as masked_content,
      AI_MASK('mymodel', content, 'ARRAY[''姓名'',''手機號'',''社會安全號碼'',''地址'']')['detected_entities'] as detected_entities

pipeline:
  parallelism: 2
  model:
  - name: mymodel
    type: openai-compatible
    model-name: qwen3.6-flash

輸入輸出樣本

輸入資料：

{"content": "客戶張三，手機號13812345678，地址為北京市朝陽區建國路88號，反映冰箱不製冷"}

輸出資料：

DataChangeEvent{
  tableId=user_feedbacks, 
  before=[], 
  after=["客戶張*，手機號138****5678，地址為北京市朝陽區**，反映冰箱不製冷", [
  {"type":"姓名","value":"張三"},
  {"type":"手機號","value":"13812345678"},
  {"type":"地址","value":"北京市朝陽區建國路88號"}
  ]], 
  op=INSERT, meta=()
}

部署與啟動

登入Realtime Compute控制台，建立CDC YAML資料攝入作業，將上述任一情境的YAML配置粘貼到編輯器中。詳情請參見Flink CDC資料攝入作業開發。
上傳附加依賴檔案。在作業草稿的附加依賴檔案地區，上傳以下JAR包：

flink-cdc-pipeline-model-openai-compatible
部署並啟動作業。

監控與限流

目前的版本僅通過Flink Web UI監控AI Function的Token用量。在 Flink Dashboard 中，單擊目標作業，選擇 Metrics 頁簽。在 Add Metric 下拉式清單中選擇目標運算元的指標（例如 ai_model_completion_tokens），查看當前指標值。