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;
エンドツーエンドの例 スキャンされたPDFを検索可能なナレッジベースに#
この例では、スキャンされたPDFドキュメントのコレクションをベクトル検索可能なナレッジベースに変換する完全なパイプラインを構築する方法を示します。 3つの連続したステップを使用します。
PdfToImage - 各PDFページを画像としてレンダリングします。
PerformOcr - OCRモデルを使用して、各レンダリング画像からテキストを抽出します。
KnowledgeBase - 抽出されたテキストを埋め込み、セマンティック検索のために保存します。
このパターンは、スキャンされた文書、テキストレイヤーのない古いPDF、および ParsePdf が空または低品質のテキストを返すコンテンツに最適です。 中間ストレージを活用することで、ナレッジベースにフィードする前にOCR出力を検査し、データ品質を保証し、ワークフローの問題をデバッグできます。
前提条件#
登録されたNVIDIA NIM PaddleOCRモデルと登録された埋め込みモデルが必要です。 これらをまだお持ちでない場合は、最初に登録します。
- - Register the OCR model
SELECT aidb.create_model(
my_paddle_ocr_model,
nim_paddle_ocr,
credentials => {"api_key": "<NVIDIA_NIM_API_KEY>"}::JSONB
);
埋め込みモデルbertはプリインストールされており、登録は必要ありません。別のモデルを使用している場合は、ここに登録します。
ステップ1ソーステーブルを作成する
生のPDFデータを保持するテーブルを作成します。
CREATE TABLE scanned_pdfs (
id INT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
name TEXT NOT NULL,
pdf_data BYTEA NOT NULL
);
ステップ2PDFをテーブルにロードする
PDFドキュメントをバイナリデータとして挿入します。次の例では、
pg_read_binary_file()
を使用してサーバーのファイルシステムからファイルをロードします。
INSERT INTO scanned_pdfs (name, pdf_data)
VALUES (quarterly_report.pdf, pg_read_binary_file(/path/to/quarterly_report.pdf));
S3互換ボリュームからPDFをロードすることもできます。 Configuring external storage および 파이프라인(볼륨 소스) 생성 を参照してください。
ステップ3パイプラインを作成する
SELECT aidb.create_pipeline(
name => scanned_pdf_kb,
source => scanned_pdfs,
source_key_column => id,
source_data_column => pdf_data,
step_1 => PdfToImage,
step_1_options => {"dpi": 150}::jsonb,
step_2 => PerformOcr,
step_2_options => aidb.ocr_config(model => my_paddle_ocr_model),
step_3 => KnowledgeBase,
step_3_options => aidb.knowledge_base_config(
model => bert,
data_format => Text,
distance_operator => Cosine
),
auto_processing => Live
);
auto_processing をLive
に設定すると、新しいPDFがscanned_pdfs
テーブルに挿入されるたびに、パイプラインが自動的に実行されます。
ステップ4既存のデータでパイプラインを実行する
パイプラインを有効にする前に行を挿入した場合、ここで処理します。
SELECT aidb.run_pipeline(scanned_pdf_kb);
ステップ5ナレッジベースを照会する
パイプラインが完了したら、セマンティック検索を使用して結果を照会します。
SELECT * FROM aidb.retrieve_text(
scanned_pdf_kb,
quarterly revenue figures,
5 -- return top 5 matches
);
結果の各行は、PDFページから抽出されたテキストブロックに対応し、クエリーとの意味的な類似性によってランク付けされます。