Realtime Compute for Apache Flink:クイックスタート：LLM によるリアルタイムデータ分析

チャット/補完モデル

次のサンプル SQL で、qwen-turbo モデルを登録します。

CREATE MODEL ai_analyze_sentiment
INPUT (`input` STRING)
OUTPUT (`content` STRING)
WITH (
    'provider'='bailian',
    'endpoint'='<base_url>/compatible-mode/v1/chat/completions',    -- チャット/補完モデルタスクのエンドポイント。
    'api-key' = '<YOUR KEY>',
    'model'='qwen-turbo',                                                               -- Qwen-turbo モデル。
    'system-prompt' = 'Classify the text below into one of the following labels: [positive, negative, neutral, mixed]. Output only the label.'
);

埋め込みモデル

次のサンプル SQL で、text-embedding-v3 モデルを登録します。

CREATE MODEL embedding_model
INPUT (`input` STRING)
OUTPUT (`embeddings` ARRAY<FLOAT>)
WITH (
    'provider'='bailian',
    'endpoint'='https://dashscope.aliyuncs.com/compatible-mode/v1/embeddings',   -- 埋め込みモデルタスクのエンドポイント。
    'api-key' = '<YOUR KEY>',
    'model'='text-embedding-v3'                                                  -- text-embedding-v3 モデル。
);

チャット/補完モデル

登録済みの ai_analyze_sentiment モデルを ML_PREDICT で呼び出し、映画のレビューの感情分析を実行します。

次の SQL をエディタにコピーします。

-- 一時シンクテーブルを作成します。
CREATE TEMPORARY TABLE print_sink(
  id BIGINT,
  movie_name VARCHAR, 
  predict_label VARCHAR, 
  actual_label VARCHAR
) WITH (
  'connector' = 'print',   -- print コネクタを使用します。
  'logger' = 'true'        -- 結果をコンソールに表示します。
);
-- テストデータ用の一時ビューを作成します。
-- | id | movie_name         | user_comment                                                                                                                   | actual_label |
-- | 1  | Her Story          | My favorite part was when the kid guessed the sounds. It is one of the most romantic narratives I have seen in movies. Very gentle and loving. | positive     |
-- | 2  | The Dumpling Queen | Unremarkable.                                                                                                                  | negative     |
CREATE TEMPORARY VIEW movie_comment(id, movie_name, user_comment, actual_label)
AS VALUES (1, 'Her Story', 'My favorite part was when the kid guessed the sounds. It is one of the most romantic narratives I have seen in movies. Very gentle and loving.', 'positive'), (2, 'The Dumpling Queen', 'Unremarkable.', 'negative');
INSERT INTO print_sink
SELECT id, movie_name, content as predict_label, actual_label 
FROM ML_PREDICT(
  TABLE movie_comment, 
  MODEL ai_analyze_sentiment,  -- 登録した Qwen qwen-turbo モデル。
  DESCRIPTOR(user_comment));   

埋め込みモデル

登録済みの embedding_model を ML_PREDICT で呼び出して映画のレビューのエンベディングを生成し、結果を Milvus (パブリックプレビュー) に書き込みます。

次の SQL をエディタにコピーします。

-- milvus_sink という名前の一時シンクテーブルを作成します。
CREATE TEMPORARY TABLE milvus_sink
(
    id STRING,
    movie_name STRING,
    user_comment STRING,
    embeddings ARRAY<FLOAT>,
    PRIMARY KEY (id) NOT ENFORCED
)
WITH (
    'connector' = 'milvus',
    'endpoint' = '<YOUR-ENDPOINT>',
    'port' = '<YOUR-PORT>',
    'userName' = '<YOUR-USERNAME>',
    'password' = '<YOUR-PASSWORD>',
    'databaseName' = 'default',
    'collectionName' = 'movie-comment-embeddings'
);
-- テストデータ用の一時ビューを作成します。
-- | id | movie_name         | user_comment                                                                                                                   |
-- | 1  | Her Story          | My favorite part was when the kid guessed the sounds. It is one of the most romantic narratives I have seen in movies. Very gentle and loving. |
-- | 2  | The Dumpling Queen | Unremarkable.                                                                                                                  |
CREATE TEMPORARY VIEW movie_comment(id, movie_name,  user_comment)
AS VALUES ('1', 'Her Story', 'My favorite part was when the kid guessed the sounds. It is one of the most romantic narratives I have seen in movies. Very gentle and loving.'), ('2', 'The Dumpling Queen', 'Unremarkable.');
INSERT INTO
    milvus_sink
SELECT
    id,
    movie_name,
    user_comment,
    embeddings
FROM
    ML_PREDICT (
        TABLE movie_comment,
        MODEL embedding_model,  -- 登録した text-embedding-v3 モデル。                   
        DESCRIPTOR (user_comment)
    );

チャット/補完モデル

1. ジョブのステータスが [FINISHED] であることを確認してください。

2. [O&M] コンソールで、[デプロイメント] に移動し、対象のジョブをクリックします。

3. [ログ] タブで、[タスクマネージャー] サブタブをクリックし、[現在のタスクマネージャー] を選択します。

4. [ログ] をクリックし、PrintSinkOutputWriter を検索します。

   predict_label は actual_label と一致する必要があります。

   たとえば、+I[1, Her Story, positive, positive] と +I[2, The Dumpling Queen, negative, negative] は、予測が実際のラベルと一致していることを示します。