AI_SUMMARIZE で Realtime Compute for Apache Flink AI テキスト要約パイプラインを構築 - Flink

このトピックでは、大規模言語モデル（LLM）を用いて AI_SUMMARIZE を活用し、テキストの要約を生成する方法について説明します。

制限事項

Ververica Runtime（VVR）11.4 以降でのみ利用可能です。
AI_SUMMARIZE で使用される Flink 演算子のスループットは、モデルプラットフォーム側のレート制限によって制約されます。トラフィック制限に達すると、Flink ジョブでバックプレッシャーが発生し、当該演算子がボトルネックとなります。深刻な場合は、タイムアウトエラーが発生し、ジョブが再起動される場合があります。

構文

AI_SUMMARIZE(
  MODEL => MODEL <MODEL NAME>, 
  INPUT => <INPUT COLUMN NAME>,
  MAX_LENGTH => <MAX LENGTH>
)

入力パラメーター

パラメーター	データ型	説明
MODEL <MODEL NAME>	MODEL	登録済みのモデルサービスの名前です。モデルサービスの登録手順については、「モデル設定」をご参照ください。注：現時点では、モデルが VARIANT 型の出力を返す必要があります。
<INPUT COLUMN NAME>	STRING	モデルが分析対象とするデータです。
<MAX LENGTH>	INTEGER	モデル出力の最大長です。注：現時点では、この入力パラメーターは定数である必要があります。

出力

パラメーター	データ型	説明
summary	STRING	要約内容

例

テストデータ

description

Flink とは何ですか？

Apache Flink は、リアルタイムデータストリームおよびバッチデータに対するステートフルな計算を実行するための、オープンソースの分散型ストリーム処理フレームワークです。

簡単に言うと：

Flink は、リアルタイムデータを処理するためのコンピュートエンジンです。

ウェブサイトのクリック、IoT センサーデータ、株式取引などの継続的なデータストリームを処理します。

低遅延、高スループット、および 1 回限りのセマンティクスを提供します。

ストリーム処理およびバッチ処理の両方をサポートします。

テストステートメント

以下の SQL の例では、Qwen-Plus モデルを作成し、AI_SUMMARIZE を使用して入力データを要約します。

CREATE TEMPORARY MODEL general_model
INPUT (`input` STRING)
OUTPUT (`content` VARIANT)
WITH (
    'provider' = 'openai-compat',
    'endpoint'='<YOUR ENDPOINT>',
    'apiKey' = '<YOUR KEY>',
    'model' = 'qwen-plus'
);

CREATE TEMPORARY VIEW infos(id, description)
AS VALUES (1, '
What is Flink?
Apache Flink is an open source distributed stream processing framework for stateful computation over real-time data streams and batch data.
In simple terms:
Flink is a compute engine for processing real-time data.
It handles continuous data streams such as website clicks, Internet of Things sensor data, and stock trades.
It provides low latency, high throughput, and exactly-once semantics.
It supports both stream processing and batch processing.
');

-- 位置引数を使用して AI_SUMMARIZE を呼び出す
SELECT id, summary 
FROM infos, LATERAL TABLE(
  AI_SUMMARIZE(MODEL general_model, description, 10));
-- 名前付き引数を使用して AI_SUMMARIZE を呼び出す
SELECT id, summary 
FROM infos, LATERAL TABLE(
  AI_SUMMARIZE(
    MODEL => MODEL general_model, 
    INPUT => description, 
    MAX_LENGTH => 10));

出力結果

id	summary
1	リアルタイムストリーム処理エンジン