AI_EXTRACT 関数を Flink オペレーターおよび大規模モデルサービスと組み合わせて使用し、構造化情報を抽出します。この関数には Ververica Runtime (VVR) 11.4 以降が必要であり、モデルプラットフォームのトラフィック制限を受けます。 - Realtime Compute for Apache Flink

このトピックでは、AI_EXTRACT 関数を大規模 AI モデルと組み合わせて使用し、構造化情報を抽出する方法について説明します。

制限事項

この関数には 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 歳) は杭州在住で、電話番号は ****** です。

テスト文

次の SQL の例では、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":"杭州","age":18,"name":"シャオミン","phone":"******"}