AI_EXTRACT で Realtime Compute for Apache Flink ストリーミングパイプラインから構造化データを抽出 - Flink

このトピックでは、AI_EXTRACT を使用して大規模言語モデル (LLM) から構造化情報を抽出する方法について説明します。

制限事項

リアルタイム計算エンジン Ververica Runtime (VVR) 11.4 以降でのみサポートされています。
AI_EXTRACT 文で使用される Flink 演算子のスループットは、モデルサービスプラットフォームによって制限されます。トラフィックがプラットフォームのアクセス制限に達すると、これらの演算子でバックプレッシャーが発生します。深刻なレート制限はタイムアウトを引き起こし、Flink ジョブが再起動する原因となる可能性があります。

構文

AI_EXTRACT(
  MODEL => MODEL <MODEL NAME>, 
  INPUT => <INPUT COLUMN NAME>,
  EXTRACT_SCHEMA => <EXTRACT SCHEMA>
)

入力パラメーター

パラメーター	データ型	説明
MODEL <MODEL NAME>	MODEL	登録されたモデルサービスの名前。モデルサービスの登録方法の詳細については、「モデル設定」をご参照ください。注意：モデルの出力タイプは VARIANT である必要があります。
<INPUT COLUMN NAME>	STRING	モデルが情報を抽出する元の生テキスト。
<EXTRACT SCHEMA>	STRING	抽出されたデータのスキーマを記述する JSON 文字列。注意：このパラメーターは定数である必要があります。

出力

パラメーター	データ型	説明
extracted_json	STRING	抽出された構造化情報。

例

テストデータ

id	説明
1	田中一郎は 18 歳で、中国 (杭州) に居住しています。電話番号は ****** です。

テスト文

この例では、Qwen-Plus モデルを作成し、AI_EXTRACT を使用してユーザー情報を抽出します。

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, '田中太郎は 18 歳で、東京在住です。電話番号は ****** です。');

-- 位置引数を使用して AI_EXTRACT を呼び出す
SELECT id, extracted_json
FROM infos,
LATERAL TABLE(
  AI_EXTRACT(
    MODEL general_model, 
    description, 
    '{"name":"string","phone":"string","address":"string","age":"int"}'));  
-- 名前付き引数を使用して AI_EXTRACT を呼び出す
SELECT id, extracted_json
FROM infos,
LATERAL TABLE(
  AI_EXTRACT(
    MODEL => MODEL general_model, 
    INPUT => description, 
    EXTRACT_SCHEMA => '{"name":"string","phone":"string","address":"string","age":"int"}'));

出力結果

id	extracted_json
1	{"address":"Hangzhou","age":18,"name":"Xiao Ming","phone":"******"}