Creating pipelines
==================

A pipeline defines how data moves from a source, through one or more
transformation steps, to an AI-ready destination. Creating a pipeline is
the first step in making your data searchable — without one, there is no
process to generate embeddings or keep your knowledge base in sync with
your source data.

Use ``aidb.create_pipeline()`` to define a pipeline. Its parameters fall
into four groups:

.. csv-table::
  :header: Parameter group,Parameters,More information
  :widths: 12,25,15
  :align: left
  :class: longtable

  **Source** ,"`source`, `source_key_column`, `source_data_column`",This page
  **Steps** ,"`step_1` … `step_10`, `step_N_options`",:ref:`Pipeline steps<Pipeline steps>` 
  **Orchestration** ,"`auto_processing`, `background_sync_interval`, `batch_size`",:ref:`Orchestration<Orchestration>` 
  **Destination** ,"`destination`, `destination_key_column`, `destination_data_column`","Set automatically by the `KnowledgeBase` step, or specify explicitly"

Table sources
-------------

To read from a Postgres table, set the ``source`` parameter to the table
name, ``source_key_column`` to the unique key column, and
``source_data_column`` to the column containing the data to process:

.. code:: sql

   SELECT aidb.create_pipeline(
       name               => my_pipeline,
       source             => my_table,
       source_key_column  => id,
       source_data_column => content,
       step_1             => KnowledgeBase,
       step_1_options     => aidb.knowledge_base_config(my_model, Text)
   );

Volume sources (PGFS)
---------------------

To process files stored in external cloud storage (S3, GCS, or Azure),
pipelines use :ref:`Postgres File System (PGFS) </pg_extensions/pgfs/>`  . PGFS mounts external object storage as a
volume that the pipeline can scan for files. For more information on
volumes, see :ref:`Configuring external storage <Configuring external storage>`  .

Step 1: Create a storage location
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

Define the external storage connection using
``pgfs.create_storage_location`` :

.. code:: sql

   SELECT pgfs.create_storage_location(
       name     => my_s3_location,
       uri      => s3://my-bucket/my-folder,
       options  => {"region": "us-east-1"}
   );

Step 2: Create a volume
^^^^^^^^^^^^^^^^^^^^^^^

Create an :ref:`Configuring external storage <Configuring external storage>`  that references the storage location. The
pipeline will scan this volume for new or changed files:

.. code:: sql

   SELECT aidb.create_volume(
       name             => my_volume,
       server_name => my_s3_location,
       path => ,
       data_format => Text
   );

Step 3: Reference the volume as the pipeline source
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

Set the ``source`` parameter to the volume name:

.. code:: sql

   SELECT aidb.create_pipeline(
       name       => my_pdf_pipeline,
       source     => my_volume,
       step_1     => ParsePdf,
       step_2     => KnowledgeBase,
       step_2_options => aidb.knowledge_base_config(my_model, Text)
   );

For full PGFS reference, see :ref:`PGFS documentation </pg_extensions/pgfs/>`  .

For a complete walkthrough combining a table source, pipeline steps, and
retrieval, see :ref:`Examples <Examples>`  .