Examples

Examples#

パイプラインは、 EDB Postgres® AIでAIワークフローを定義する主な方法です。各パイプラインは、データが線形に流れる1つ以上の連続したステップで構成されています。ステップ1の出力はステップ2の入力などになります。

パイプラインの作成テーブルソース#

標準のPostgresテーブルをデータソースとして使用してパイプラインを作成するには、 aidb.create_pipeline() ファンクションを使用します。次の例は、シングルステップとマルチステップのパイプラインの両方をセットアップする方法を示しています。

シングルステップパイプライン

この例では、 Postgresテーブルからナレッジベースを作成します。パイプラインを実行した後、セマンティック検索のために結果の埋め込みを照会できます。

SELECT aidb.create_pipeline(
  name => kb_pipeline_table,
  source => source_table,
  source_key_column => id,
  source_data_column => content,
  step_1 => KnowledgeBase,
  step_1_options => aidb.knowledge_base_config(
     model => bert,  -- this is a pre-defined locally running model
     data_format => Text
   )
);

マルチステップパイプライン 複数の操作をチェーンできます。この例では、最初にHTMLを解析し、結果のテキストをチャンクに分割します。

SELECT aidb.create_pipeline(
   name => html_processing_pipeline,
   source => web_data_table,
   source_key_column => id,
   source_data_column => html_content,
   step_1 => ParseHtml,
   step_2 => ChunkText
);

パイプラインの作成 ボリュームソース#

データが外部ストレージS3バケットなどにある場合、最初にストレージの場所とボリュームを定義する必要があります。

ステップ1ストレージとボリュームを定義する

PGFSストレージの場所とAIDBボリュームを作成します。

SELECT pgfs.create_storage_location(s3_bucket_location, s3://my-ai-data,
 options => {"region": "us-east-1", "skip_signature": "true"}
 )
SELECT aidb.create_volume(source_volume, s3_bucket_location, /, Text);

ステップ2パイプラインを作成する

SELECT aidb.create_pipeline(
   name => pipeline_from_s3,
   source => source_volume,
   step_1 => KnowledgeBase,
   step_1_options => aidb.knowledge_base_config(
     model => dummy,
     data_format => Text
   )
);

パイプラインの実行と更新#

パイプラインは、作成時にデフォルトで無効になっています。それらは手動でトリガーするか、自動処理を有効にすることができます。

手動実行

既存のすべてのデータに対してパイプラインを1回実行します。

SELECT aidb.run_pipeline(kb_pipeline_table);

自動処理を有効にする

ソースデータの変更に応じてパイプラインを最新の状態に保ちます。

SELECT aidb.update_pipeline(
  name => kb_pipeline_table,
  auto_processing => Live
);

モニタリングと削除#

既存のパイプラインとその構成を表示するには、 aidb.pipelines ビューを照会します。

SELECT * FROM aidb.pipelines;
__OUTPUT__
   name      | source_type | source_schema |       source       | source_key_column | source_data_column | destination_type | destination_schema |       destination       | destination_key_column | destination_data_column |                                                 steps                                                  | auto_processing | batch_size | background_sync_interval |              owner_role
- ---------------+-------------+---------------+--------------------+-------------------+--------------------+------------------+--------------------+-------------------------+------------------------+-------------------------+--------------------------------------------------------------------------------------------------------+-----------------+------------+--------------------------+--------------------------------------
pipeline__8989 | Table       | public        | source_table__8989 | id                | content            | Table            | public             | pipeline_pipeline__8989 | source_id              | value                   | [{"options": {"max_length": null, "desired_length": 1000}, "operation": "ChunkText", "step_order": 1}] | Live            |         42 | @ 42 mins                | role_pipeline_management_single_step
(1 row)

SELECT name, source, auto_processing FROM aidb.pipelines;
__OUTPUT__
  name       |     source.         | auto_processing
- ---------------+-------------------+-----------------
pipeline__8989 | source_table__8989 | Live
(1 row)

パイプラインを削除すると、パイプラインステップによって生成された宛先テーブルもドロップします。パイプラインを削除するには

SELECT aidb.delete_pipeline(kb_pipeline_table);

エンドツーエンドの例 中間ストレージを使用したマルチステップパイプライン#

このエンドツーエンドの例では、中間ストレージを利用するEDB Postgres AIでマルチステップパイプラインを構成する方法を示します。

intermediate_destination を定義することにより、個々のステップの結果テキストチャンクなどを次のステージ要約などに渡す前に永続化できます。これは、処理されたデータのデバッグまたは再利用に役立ちます。

この例の主な機能

  • マルチステップワークフロー ChunkText ステップ1をSummarizeText ステップ2にチェーンします。

  • 自動処理 新しいデータがソーステーブルに挿入されるたびに、パイプラインは自動的にトリガーされます。

  • 中間ストレージ 中間テーブルを明示的に名前付ける方法と、システムに中間テーブルを自動生成させる方法を示します。

例1名前付けた中間宛先#

このシナリオでは、チャンクステップの出力を保存するテーブルに手動で名前を付けます。

  • 生テキストのソーステーブルを作成します。

CREATE TABLE source_table_demo(
  id INT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  content TEXT NOT NULL
);

  • カスタム中間テーブルを使用してパイプラインを定義します。

SELECT * from aidb.create_pipeline(
  name => complex_pipeline_v1,
  source => source_table_demo,
  source_key_column => id,
  source_data_column => content,
  auto_processing => Live,
  step_1 => ChunkText,
  step_1_options => jsonb_build_object(
      desired_length, 20,
      intermediate_destination, jsonb_build_object(
          enabled, true,
          destination, my_custom_chunk_storage -- Explicitly named
      )
  ),
  step_2 => SummarizeText
 );

  • 自動処理をトリガーするデータを挿入します。

INSERT INTO source_table_demo (content)
VALUES (This is a long text example that will be split into segments for easier processing.);

  • チャンクと最終的なサマリーを表示します。

-- View intermediate chunked output
SELECT * FROM my_custom_chunk_storage;

-- View the final summarized output
SELECT * FROM pipeline_complex_pipeline_v1;

例2 自動中間宛先#

中間ストレージを有効にしても名前を指定しない場合、システムはパターンpipeline_[name]_step_[n] に従ってテーブルを生成します。

  • 自動名前付けの中間ストレージを使用してパイプラインを作成します。

SELECT * from aidb.create_pipeline(
  name => auto_named_pipeline,
  source => source_table_demo_2,
  source_key_column => id,
  source_data_column => content,
  auto_processing => Live,
  step_1 => ChunkText,
  step_1_options => jsonb_build_object(
      desired_length, 20,
      intermediate_destination, jsonb_build_object(enabled, true)
  ),
  step_2 => SummarizeText
);

  • 自動生成された中間テーブルにアクセスします。

-- View intermediate chunked output (auto-generated table name)
SELECT * FROM pipeline_auto_named_pipeline_step_1;
目次