Examples ======== Pipelines are the primary way to define AI workflows in EDB Postgres AI. Each pipeline consists of one or more sequential steps where data flows linearly: the output of step 1 becomes the input for step 2, and so on. Creating a pipeline (table source) ---------------------------------- To create a pipeline using a standard Postgres table as your data source, use the ``aidb.create_pipeline()`` function. The following examples show how to set up both a single-step and a multi-step pipeline. Single-step pipeline ^^^^^^^^^^^^^^^^^^^^ This example creates a knowledge base from a Postgres table. After running the pipeline, you can query the resulting embeddings for semantic search. .. code:: sql 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 ) ); Multi-step pipeline ^^^^^^^^^^^^^^^^^^^ You can chain multiple operations. This example first parses HTML and then chunks the resulting text. .. code:: sql 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 ); Creating a pipeline (volume source) ----------------------------------- If your data lives in external storage (like an S3 bucket), you must first define a storage location and a volume. Step 1: Define storage and volume ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ Create the PGFS storage location and AIDB volume. .. code:: sql 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); Step 2: Create the pipeline ^^^^^^^^^^^^^^^^^^^^^^^^^^^ .. code:: sql 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 ) ); Running and updating pipelines ------------------------------ Pipelines are disabled by default upon creation. You can trigger them manually or enable automatic processing. Manual execution ^^^^^^^^^^^^^^^^ Run the pipeline once on all existing data: .. code:: sql SELECT aidb.run_pipeline(kb_pipeline_table); Enable auto-processing ^^^^^^^^^^^^^^^^^^^^^^ Keep the pipeline up-to-date as source data changes: .. code:: sql SELECT aidb.update_pipeline( name => kb_pipeline_table, auto_processing => Live ); Monitoring and deletion ----------------------- To view your existing pipelines and their configurations, query the ``aidb.pipelines`` view: .. code:: sql 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) .. code:: sql SELECT name, source, auto_processing FROM aidb.pipelines; __OUTPUT__ name | source | auto_processing - ---------------+--------------------+----------------- pipeline__8989 | source_table__8989 | Live (1 row) Deleting a pipeline will also drop any destination tables generated by the pipeline steps. To delete a pipeline: .. code:: sql SELECT aidb.delete_pipeline(kb_pipeline_table); End-to-end example: Knowledge base for RAG ------------------------------------------ This example builds a complete retrieval-augmented generation (RAG) flow over an internal support knowledge base, entirely in SQL and entirely local: vector **retrieval** finds candidate articles, a cross-encoder **reranker** refines their order, and a **generation** model synthesizes an answer grounded in the retrieved articles. A reduced version of this example appears on the :ref:`AI pipelines landing page <./>` . Step 1: Register the models ^^^^^^^^^^^^^^^^^^^^^^^^^^^ Each stage uses a different kind of model. The embedding model, ``bge-small-en-v1.5-f16`` , is pre-registered on every AIDB install, so only the reranking and generation models need registering. ``aidb.create_model()`` downloads and validates each GGUF file from Hugging Face (the reranker is ~420 MB, Llama-3.2-3B ~3.4 GB); the ``revision`` pins make the example reproducible — without them, the model repository’s mutable ``main`` branch is used. .. code:: sql - - Reranking model (cross-encoder) SELECT aidb.create_model( name => bge-reranker-v2-m3-Q4, provider => llamacpp_reranking, config => jsonb_build_object( model, gpustack/bge-reranker-v2-m3-GGUF, model_file, bge-reranker-v2-m3-Q4_K_M.gguf, revision, 3093af03b1a635e67b084b1d8c03c5f5e020fd05, n_ctx, 2048 ) ); - - Generation model (instruct LLM) SELECT aidb.create_model( name => llama-3.2-3b-instruct-Q8_0, provider => llamacpp_generate, config => jsonb_build_object( model, unsloth/Llama-3.2-3B-Instruct-GGUF, model_file, Llama-3.2-3B-Instruct-Q8_0.gguf, revision, e7d0997e49c9cb00d88b4c1a6a16aa894b0bbc31, n_ctx, 8192, temperature, 0.0 ) ); !!!note Model sizing for the generation stage 1B-class generation models (including the pre-registered ``llama-3.2-1b-instruct-Q8_0`` ) aren’t reliable for the generation stage of this example — in testing, they recommended actions that contradict the question’s constraints. Llama-3.2-**3B**-Instruct answers correctly and is still a modest local model. Step 2: Create the source data ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ Sixteen short support articles about database operations. Several deliberately share vocabulary (disk space, vacuum, backups) — that’s what makes the reranking stage visibly matter later: .. code:: sql CREATE TABLE support_articles ( id INT PRIMARY KEY GENERATED ALWAYS AS IDENTITY, title TEXT NOT NULL, body TEXT NOT NULL ); INSERT INTO support_articles (title, body) VALUES (Reclaiming disk space with VACUUM FULL, VACUUM FULL rewrites a table into a new file and returns unused disk space to the operating system. It takes an ACCESS EXCLUSIVE lock, so all reads and writes to the table are blocked until it completes.), (Reducing table bloat without downtime, The pg_repack extension rebuilds bloated tables and indexes online. It holds only brief locks, so applications can continue reading and writing while space is reclaimed. It requires free disk space roughly equal to the table size.), (Monitoring database disk usage, Use pg_database_size() and pg_total_relation_size() to track how much disk space databases and tables occupy. Configure alerts well before storage volumes reach capacity.), (Resolving "too many connections" errors, The error "FATAL: sorry, too many clients already" means the max_connections limit is reached. Find idle sessions in pg_stat_activity and terminate them with pg_terminate_backend(), or introduce a connection pooler.), (Connection pooling with PgBouncer, PgBouncer multiplexes thousands of client connections onto a small pool of database sessions. Transaction pooling mode provides the highest connection density for typical web application workloads.), (Diagnosing streaming replication lag, Compare pg_current_wal_lsn() on the primary with pg_last_wal_replay_lsn() on the standby to measure replication lag. Common causes include network saturation and long-running queries on the standby holding back WAL replay.), (Creating a hot standby with pg_basebackup, Run pg_basebackup to clone the primary server, then start the clone with a standby.signal file. The standby serves read-only queries while continuously replaying WAL from the primary.), (Replicating selected tables with logical replication, Logical replication publishes row changes for selected tables to subscriber databases and keeps them continuously synchronized. Unlike streaming replication, it can replicate a subset of tables and works across major versions.), (Point-in-time recovery with WAL archiving, With continuous WAL archiving enabled, you can restore a base backup and set recovery_target_time to roll the database forward to a specific moment, such as just before an accidental data deletion.), (Logical backups with pg_dump, pg_dump exports a database or single tables to a portable archive that can be restored into newer PostgreSQL versions. Dumps are consistent snapshots but offer no incremental or point-in-time capability.), (Reading query plans with EXPLAIN ANALYZE, EXPLAIN ANALYZE executes a query and reports the actual time spent in each plan node. Look for row-estimate mismatches and sequential scans on large tables to find missing indexes.), (Finding and removing unused indexes, Query pg_stat_user_indexes for indexes with zero scans. Unused indexes slow down every write and consume disk space; dropping them is a safe, immediate win.), (Tuning work_mem for large sorts, When sorts or hash joins exceed work_mem, they spill to temporary files on disk and slow down dramatically. Raise work_mem session-locally for reporting queries rather than globally.), (Restricting data access with row-level security, Row-level security policies filter which rows each role can see or modify. Policies are enforced by the server itself, so they apply uniformly to every client and application.), (Auditing database activity with pgaudit, The pgaudit extension produces detailed audit logs of reads, writes, and DDL for compliance requirements. Configure log classes narrowly to control log volume.), (Upgrading major versions with pg_upgrade, pg_upgrade migrates a cluster to a new major version by relinking data files instead of dumping and reloading. With the --link option, downtime is typically minutes even for large clusters.); Step 3: Create and run the knowledge base pipeline ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ .. code:: sql SELECT aidb.create_pipeline( name => support_kb, source => support_articles, source_key_column => id, source_data_column => body, step_1 => KnowledgeBase, step_1_options => aidb.knowledge_base_config( model => bge-small-en-v1.5-f16, data_format => Text ), auto_processing => Disabled ); SELECT aidb.run_pipeline(support_kb); Step 4: Retrieval ^^^^^^^^^^^^^^^^^ Semantic search finds the right article even when the query shares few or no keywords with it. Paste raw error text, get the resolution: .. code:: sql SELECT a.title, r.distance::numeric(5,3) AS distance FROM aidb.retrieve_key(support_kb, application logs show: FATAL: sorry, too many clients already, 3) r JOIN support_articles a ON a.id = r.key::int ORDER BY r.distance; __OUTPUT__ title | distance - ----------------------------------------+---------- Resolving "too many connections" errors | 0.507 Tuning work_mem for large sorts | 0.844 Connection pooling with PgBouncer | 0.908 (3 rows) Or describe a symptom that has no tool names or keywords in common with the answer: .. code:: sql SELECT a.title, r.distance::numeric(5,3) AS distance FROM aidb.retrieve_key(support_kb, A developer accidentally deleted rows from a table yesterday. How do we get the data back?, 3) r JOIN support_articles a ON a.id = r.key::int ORDER BY r.distance; __OUTPUT__ title | distance - ------------------------------------------+---------- Point-in-time recovery with WAL archiving | 0.795 Reducing table bloat without downtime | 0.806 Finding and removing unused indexes | 0.846 (3 rows) (``aidb.retrieve_text()`` returns the article body directly, with no join needed; the join variant is shown because titles read better in output tables.) Step 5: Reranking ^^^^^^^^^^^^^^^^^ Demo question: *“The database is running out of disk space, but we cannot take the application offline to fix it.”* Vector search alone ranks *Point-in-time recovery* — clearly unhelpful for a disk-space problem — at #1, with the correct article (*pg_repack*) second in a near-tie: .. code:: sql SELECT a.title, r.distance::numeric(5,3) AS distance FROM aidb.retrieve_key(support_kb, The database is running out of disk space, but we cannot take the application offline to fix it., 5) r JOIN support_articles a ON a.id = r.key::int ORDER BY r.distance; __OUTPUT__ title | distance - ------------------------------------------+---------- Point-in-time recovery with WAL archiving | 0.808 ← wrong Reducing table bloat without downtime | 0.838 ← correct answer, near-tie Tuning work_mem for large sorts | 0.851 Monitoring database disk usage | 0.865 Resolving "too many connections" errors | 0.883 (5 rows) The cross-encoder reads the question and each article together and reorders decisively. ``aidb.rerank_text()`` returns ``(text, logit_score, id)`` , where ``id`` is the 0-based position in the input array and a higher ``logit_score`` means more relevant — the ``id + 1`` maps that index back into the 1-based SQL arrays: .. code:: sql WITH candidates AS ( SELECT array_agg(a.title ORDER BY r.distance) AS titles, array_agg(a.body ORDER BY r.distance) AS bodies FROM aidb.retrieve_key(support_kb, The database is running out of disk space, but we cannot take the application offline to fix it., 5) r JOIN support_articles a ON a.id = r.key::int ) SELECT c.titles[r.id + 1] AS title, r.logit_score::numeric(7,3) AS logit_score FROM candidates c, aidb.rerank_text(bge-reranker-v2-m3-Q4, The database is running out of disk space, but we cannot take the application offline to fix it., c.bodies) r ORDER BY r.logit_score DESC; __OUTPUT__ title | logit_score - ------------------------------------------+------------- Reducing table bloat without downtime | -2.742 ← promoted to #1, clear margin Monitoring database disk usage | -3.116 Point-in-time recovery with WAL archiving | -7.037 ← demoted Tuning work_mem for large sorts | -8.836 Resolving "too many connections" errors | -9.216 (5 rows) This is the classic two-stage retrieval pattern: over-fetch candidates cheaply with the vector index (5 here; 20–100 in production), then let the more accurate but more expensive cross-encoder pick the final order. A reranker runs a full model forward pass per candidate — never rerank a whole table. Step 6: Generation ^^^^^^^^^^^^^^^^^^ Full RAG in a single statement: retrieve 5 candidates, rerank and keep the best 3, and generate a grounded answer. The first two CTEs are exactly the Step-4 and Step-5 commands above. (``\set`` is a psql feature — in other clients, inline the question text directly.) .. code:: sql \set question The database is running out of disk space, but we cannot take the application offline to fix it. What should we do? WITH candidates AS ( -- Stage 1: vector search over the knowledge base (over-fetch 5 candidates) SELECT array_agg(a.title ORDER BY r.distance) AS titles, array_agg(a.body ORDER BY r.distance) AS bodies FROM aidb.retrieve_key(support_kb, :question, 5) r JOIN support_articles a ON a.id = r.key::int ), reranked AS ( -- Stage 2: cross-encoder reranking, keep the best 3 SELECT c.titles[r.id + 1] AS title, c.bodies[r.id + 1] AS body, r.logit_score FROM candidates c, aidb.rerank_text(bge-reranker-v2-m3-Q4, :question, c.bodies) r ORDER BY r.logit_score DESC LIMIT 3 ), context AS ( SELECT string_agg(### || title || E\n || body, E\n\n ORDER BY logit_score DESC) AS articles FROM reranked ) - - Stage 3: grounded answer generation SELECT aidb.generate_text( llama-3.2-3b-instruct-Q8_0, EKnowledge base articles:\n\n || context.articles || E\n\nQuestion: || :question || E\n\nAnswer the question using only the facts in the articles above. Recommend one course of action and mention its caveats., { "system_prompt": "You are a database support assistant. Be accurate and concise; do not invent facts that are not in the provided articles.", "max_tokens": 250, "temperature": 0.0 }::json ) AS answer FROM context; The output (deterministic at ``temperature: 0.0`` — repeat runs produce identical answers): :: Based on the provided articles, I recommend using the `pg_repack` extension to reduce table bloat without downtime. According to the article, `pg_repack` rebuilds bloated tables and indexes online, holding only brief locks, allowing applications to continue reading and writing while space is reclaimed. This approach does not require taking the application offline. However, there is a caveat: `pg_repack` requires free disk space roughly equal to the table size. This means that you need to ensure there is sufficient free disk space available before running `pg_repack`, which may require some planning and preparation. No single article contains this answer: the model combined the constraint from the question (“cannot go offline”) with facts from two competing articles (VACUUM FULL blocks reads and writes; pg_repack works online) and carried over the free-disk-space caveat — a synthesis, not a verbatim return. The same statement works with any registered generation model, including hosted ones — only the model name changes. One caveat for hosted GPT-5.x reasoning models: omit ``"temperature"`` from the inference config; they reject the parameter. Step 7: Teardown ^^^^^^^^^^^^^^^^ .. code:: sql SELECT aidb.delete_pipeline(support_kb); DROP TABLE support_articles; SELECT aidb.delete_model(llama-3.2-3b-instruct-Q8_0); SELECT aidb.delete_model(bge-reranker-v2-m3-Q4); (The embedding model isn’t deleted — ``bge-small-en-v1.5-f16`` is one of the pre-registered defaults.) End-to-end example: Multi-step pipeline with intermediate storage ----------------------------------------------------------------- This end-to-end example demonstrates how to configure a multi-step pipeline in EDB Postgres AI that utilizes intermediate storage. By defining an ``intermediate_destination`` , you can persist the results of individual steps (like text chunking) before they pass to the next stage (like summarization), which is useful for debugging or reusing processed data. It’s off by default, and no step config helper (``aidb.chunk_text_config()`` , and so on) has a parameter for it — both examples below add it as plain JSON alongside the helper’s own options. See :ref:`End-to-end example: Multi-step pipeline with intermediate storage ` for the full reference, including how this interacts with ``deduplicate`` at query time. **Key features of this example**: - **Multi-step workflow**: Chains ``ChunkText`` (Step 1) into ``SummarizeText`` (Step 2). - **Auto-processing**: The pipeline automatically triggers whenever new data is inserted into the source table. - **Intermediate storage**: Demonstrates how to explicitly name an intermediate table versus allowing the system to auto-generate one. Example 1: Named intermediate destination ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ In this scenario, we will manually name the table that stores the output of the chunking step. - Create a source table for your raw text: .. code:: sql CREATE TABLE source_table_demo( id INT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, content TEXT NOT NULL ); - Define the pipeline with a custom intermediate table: .. code:: sql 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 data to trigger auto-processing: .. code:: sql INSERT INTO source_table_demo (content) VALUES (This is a long text example that will be split into segments for easier processing.); - View the chunks and final summaries: .. code:: sql -- View intermediate chunked output SELECT * FROM my_custom_chunk_storage; -- View the final summarized output SELECT * FROM pipeline_complex_pipeline_v1; Example 2: Automatic intermediate destination ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ If you enable intermediate storage but don’t provide a name, the system generates a table following the pattern ``pipeline_[name]_step_[n]`` . - Create a pipeline with auto-named intermediate storage: .. code:: sql 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 ); - Access the auto-generated intermediate table: .. code:: sql -- View intermediate chunked output (auto-generated table name) SELECT * FROM pipeline_auto_named_pipeline_step_1; End-to-end example: Scanned PDFs to a searchable knowledge base --------------------------------------------------------------- This example shows how to build a full pipeline that turns a collection of scanned PDF documents into a vector-searchable knowledge base. It uses three sequential steps: 1. **PdfToImage**: renders each PDF page as an image. 2. **PerformOcr**: extracts text from each rendered image using an OCR model. 3. **KnowledgeBase**: embeds the extracted text and stores it for semantic search. This pattern is ideal for scanned documents, legacy PDFs without a text layer, and any content where **ParsePdf** returns empty or low-quality text. By leveraging intermediate storage, you can inspect the OCR output before it feeds into the knowledge base, ensuring data quality and debugging any issues in the workflow. Prerequisites ^^^^^^^^^^^^^ You need a registered NVIDIA NIM PaddleOCR model and a registered embedding model. If you don’t have these yet, register them first: .. code:: sql - - Register the OCR model SELECT aidb.create_model( my_paddle_ocr_model, nim_paddle_ocr, credentials => {"api_key": ""}::JSONB -- or credentials_env to read this from an environment variable instead ); The embedding model (bert) is pre-installed and needs no registration. If you’re using a different model, register it here. Step 1: Create the source table ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ Create a table to hold the raw PDF data: .. code:: sql CREATE TABLE scanned_pdfs ( id INT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, name TEXT NOT NULL, pdf_data BYTEA NOT NULL ); Step 2: Load PDFs into the table ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ Insert PDF documents as binary data. The example below uses ``pg_read_binary_file()`` to load files from the server’s file system: .. code:: sql INSERT INTO scanned_pdfs (name, pdf_data) VALUES (quarterly_report.pdf, pg_read_binary_file(/path/to/quarterly_report.pdf)); You can also load PDFs from an S3-compatible volume — see :ref:`Configuring external storage ` and :ref:`Creating a pipeline (volume source) ` . Step 3: Create the pipeline ^^^^^^^^^^^^^^^^^^^^^^^^^^^ .. code:: sql 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 ); With ``auto_processing`` set to ``Live`` , the pipeline will automatically run whenever new PDFs are inserted into the ``scanned_pdfs`` table. Step 4: Run the pipeline on existing data ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ If you inserted rows before enabling the pipeline, process them now: .. code:: sql SELECT aidb.run_pipeline(scanned_pdf_kb); Step 5: Query the knowledge base ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ After the pipeline finishes, query the results using semantic search: .. code:: sql SELECT * FROM aidb.retrieve_text( scanned_pdf_kb, quarterly revenue figures, 5 -- return top 5 matches ); Each row in the result corresponds to a text block extracted from a PDF page, ranked by semantic similarity to your query.