Create a Pipeline

A pipeline connects your plugins and ontologies into a complete document processing workflow. This guide walks you through creating, executing, and monitoring a pipeline end to end.

Last updated: 2026-07-21

A pipeline connects your plugins and ontologies into a complete document processing workflow. This guide walks you through creating, executing, and monitoring a pipeline end to end.

Prerequisites

Before creating a pipeline, ensure you have:

A registered Source plugin (to ingest documents).
A registered Classification plugin (to assign document labels).
A registered Extraction plugin (to extract structured data).
A registered Ontology (to define the fields for extraction).
A connected source, if your Source plugin requires authenticated access.

ℹ️Note

You can list your registered plugins and ontologies at any time using the GET endpoints described below. Source credentials are configured when you connect a source, not passed inline when creating the pipeline.

1. Gather Your Components

First, identify the IDs of the resources you will wire together.

List your plugins

bash

GET /api/v1/plugins
Authorization: Bearer <your-jwt-token>

json

{
  "count": 3,
  "plugins": [
    { "plugin_id": "01HQZX1A2B3C4D5E6F7G8H9J0K", "name": "Gmail Source Plugin", "plugin_type": "source" },
    { "plugin_id": "01HQZX3K4M2N5P7Q8R9S0T1U2V", "name": "Invoice Classifier", "plugin_type": "classification" },
    { "plugin_id": "01HQZX5W6X7Y8Z9A0B1C2D3E4F", "name": "Invoice Data Extractor", "plugin_type": "extraction" }
  ]
}

List your ontologies

bash

GET /api/v1/ontologies
Authorization: Bearer <your-jwt-token>

json

{
  "count": 1,
  "ontologies": [
    { "ontology_id": "01HQZY1M2N3P4Q5R6S7T8U9V0W", "name": "Invoice Ontology" }
  ]
}

Plugin and ontology IDs are ULIDs — 26-character, lexicographically sortable identifiers. Copy them exactly; you will reference them when creating the pipeline.

2. Create the Pipeline

Create a new pipeline by posting the plugin and ontology IDs you gathered. Plugins are supplied as a flat plugin_ids list — the platform sequences them by type at execution time, so ordering within the list does not matter.

bash

POST /api/v1/pipelines
Authorization: Bearer <your-jwt-token>
Content-Type: application/json

{
  "name": "Invoice Processing Pipeline",
  "description": "Ingest, classify, and extract invoice data from Gmail.",
  "plugin_ids": [
    "01HQZX1A2B3C4D5E6F7G8H9J0K",
    "01HQZX3K4M2N5P7Q8R9S0T1U2V",
    "01HQZX5W6X7Y8Z9A0B1C2D3E4F"
  ],
  "ontology_catalogs_ids": [
    "01HQZY1M2N3P4Q5R6S7T8U9V0W"
  ]
}

json

{
  "pipeline_id": "01HR001A2B3C4D5E6F7G8H9J0K",
  "name": "Invoice Processing Pipeline",
  "message": "Pipeline created successfully"
}

Save the pipeline_id — you will use it to execute the pipeline.

3. Execute the Pipeline

Trigger a pipeline execution to create one or more jobs:

bash

POST /api/v1/jobs/from-pipeline
Authorization: Bearer <your-jwt-token>
Content-Type: application/json

{
  "pipeline_id": "01HR001A2B3C4D5E6F7G8H9J0K"
}

json

{
  "job_ids": ["01HQZZ8P9Q0R1S2T3U4V5W6X7Y"],
  "message": "Execution started for 1 job(s)"
}

Save the job_id to monitor progress.

4. Monitor the Job

Poll the job status endpoint to track progress:

bash

GET /api/v1/jobs/01HQZZ8P9Q0R1S2T3U4V5W6X7Y
Authorization: Bearer <your-jwt-token>

Running:

json

{
  "job_id": "01HQZZ8P9Q0R1S2T3U4V5W6X7Y",
  "status": "running",
  "current_plugin": "Invoice Classifier",
  "progress": "2/3 plugins executed",
  "started_at": "2025-10-28T10: 30: 00Z"
}

Completed:

json

{
  "job_id": "01HQZZ8P9Q0R1S2T3U4V5W6X7Y",
  "status": "completed",
  "documents_processed": 15,
  "documents_with_labels": 12,
  "documents_with_data": 12,
  "execution_time_seconds": 45.2,
  "started_at": "2025-10-28T10: 30: 00Z",
  "completed_at": "2025-10-28T10: 30: 45Z"
}

Failed:

json

{
  "job_id": "01HQZZ8P9Q0R1S2T3U4V5W6X7Y",
  "status": "failed",
  "error": "Plugin 'Invoice Classifier' failed: LLM API rate limit exceeded",
  "failed_plugin": "Invoice Classifier",
  "started_at": "2025-10-28T10: 30: 00Z",
  "failed_at": "2025-10-28T10: 30: 23Z"
}

A job transitions through pending, running, and then a terminal state (completed, failed, or cancelled).

ℹ️Note

For long-running jobs, poll every 5–10 seconds. The status fields update in real time as documents move through each stage. Real-time progress is also streamed over Server-Sent Events (SSE) for clients that prefer a push model over polling.

5. Retrieve Processed Documents

Once the job completes, retrieve the processed documents with their extracted data.

Get all documents:

bash

GET /api/v1/documents
Authorization: Bearer <your-jwt-token>

Filter by labels:

bash

GET /api/v1/documents?labels=invoice
Authorization: Bearer <your-jwt-token>

Paginate deep result sets:

bash

="color:#5c6370;font-style:italic"># First page
GET /api/v1/documents?limit=50
="color:#5c6370;font-style:italic"># Subsequent pages: pass back the response's next_cursor unchanged
GET /api/v1/documents?limit=50&cursor=eyJ2IjoxLCJzIjoiMjAyNS0xMC0yOFQxMDozMDo0NSswMDowMCIsImQiOiIuLi4ifQ
Authorization: Bearer <your-jwt-token>

Response:

json

{
  "total_count": 12,
  "documents": [
    {
      "id": "01HR001A2B3C4D5E6F7G8H9J0K",
      "metadata": {
        "source": "gmail",
        "subject": "Invoice #12345",
        "sender": "billing@acme.com"
      },
      "labels": ["invoice"],
      "data": {
        "invoice_total": 1500.00,
        "invoice_date": "2025-10-15",
        "invoice_number": "INV-12345",
        "vendor_name": "Acme Corp"
      }
    }
  ],
  "next_cursor": null
}

💡Tip

next_cursor is non-null whenever the page is full. Use it for the next request rather than ?offset=; keyset pagination scales past page ~50 where OFFSET would slow down. Treat the cursor as opaque — don't parse it.

Plugin Execution Order

Plugins execute in a fixed order within every pipeline, regardless of how you list them in plugin_ids. Understanding this order matters for debugging and for designing plugins that depend on a previous stage.

Order	Type	Purpose
1	Source	Ingest documents from external sources
2	Classification	Analyze content, add labels
3	Extraction	Extract structured data for the resolved labels
4	Benchmark	Score and link related documents

Example flow:

1. Gmail Source Plugin (source)
2. Invoice Classifier (classification)
3. Invoice Extractor (extraction)

⚠️Warning

If a document fails classification, it is not passed to the extraction stage. If extraction fails for one document, that document is excluded from later stages while the others proceed normally.

Pre-Conditions

Plugins can define conditions that filter which documents they process. Pre-conditions are evaluated before the plugin runs, using the labels and metadata produced by earlier stages.

yaml

"color:#e06c75">spec:
  "color:#e06c75">type: "extraction"
  "color:#e06c75">pre_conditions:
    - key: "document.labels"
      "color:#e06c75">operator: "contains"
      "color:#e06c75">value: "invoice"

This extraction plugin only runs on documents labeled invoice. Each pre-condition is an object with a key, an operator, and a value.

Operators:

contains — array contains the value
equals — exact match
in — value is one of a list
greater_than, less_than — numeric comparison

Documents that do not satisfy a plugin's pre-conditions are skipped by that plugin and do not count as failures.

Direct Execution (Alternative)

For ad-hoc processing and quick testing, you can start a job directly without creating a pipeline. Supply the plugin and ontology IDs inline, plus the source identifiers to process:

bash

POST /api/v1/jobs
Authorization: Bearer <your-jwt-token>
Content-Type: application/json

{
  "plugin_ids": [
    "01HQZX1A2B3C4D5E6F7G8H9J0K",
    "01HQZX3K4M2N5P7Q8R9S0T1U2V",
    "01HQZX5W6X7Y8Z9A0B1C2D3E4F"
  ],
  "ontology_catalogs_ids": [
    "01HQZY1M2N3P4Q5R6S7T8U9V0W"
  ],
  "source_ids": ["user@gmail.com"]
}

Pipeline Management

The API provides full CRUD operations for pipelines.

Operation	Method	Endpoint	Description
List	GET	`/api/v1/pipelines`	List all pipelines.
Get	GET	`/api/v1/pipelines/{pipeline_id}`	Retrieve a single pipeline by ID.
Create	POST	`/api/v1/pipelines`	Create a new pipeline.
Update	PUT	`/api/v1/pipelines/{pipeline_id}`	Update pipeline configuration, plugins, or ontologies.
Delete	DELETE	`/api/v1/pipelines/{pipeline_id}`	Delete a pipeline (does not delete associated jobs or documents).
Execute	POST	`/api/v1/jobs/from-pipeline`	Trigger a new job for a pipeline.

Update a Pipeline

bash

PUT /api/v1/pipelines/{pipeline_id}
Authorization: Bearer <your-jwt-token>
Content-Type: application/json

{
  "name": "Updated Invoice Pipeline",
  "plugin_ids": [...],
  "ontology_catalogs_ids": [...]
}

Job Management

List All Jobs

bash

GET /api/v1/jobs
Authorization: Bearer <your-jwt-token>

Cancel a Running Job

bash

POST /api/v1/jobs/{job_id}/cancel
Authorization: Bearer <your-jwt-token>

Common Issues

Issue	Cause	Resolution
Job stuck in "pending"	The platform is at maximum capacity — no worker is available yet.	Wait for other jobs to finish, or cancel unnecessary jobs. Jobs are picked up in order.
Job failed: "Plugin not found"	Plugin ID doesn't exist or doesn't belong to your tenant.	Verify the ID with `GET /api/v1/plugins/{plugin_id}`.
Job failed: "Credential not found"	The Source plugin requires a connected source that isn't configured.	Connect the source before running, or contact support to configure source credentials.
All documents fail classification	The classification plugin returns invalid labels, or its prompt is too vague.	Test the classifier on a single document first. Check the prompt and the valid label list.
Extraction returns empty fields	Ontology field names don't match what the LLM returns, or the classification stage didn't run.	Verify field names and descriptions, and confirm the document carries the expected labels.
No documents processed	The source found nothing, or pre-conditions filtered everything out.	Confirm the source ran, check pre-conditions, and review the job for errors.

Best Practices

Use pipelines for recurring work — create a reusable pipeline for any workflow you'll run more than once.
Test incrementally — verify each plugin individually before combining them into a pipeline.
Monitor jobs — don't assume success; poll the status endpoint or subscribe to the SSE stream.
Start small — process 1–5 documents on your first run before scaling to hundreds.
Use pre-conditions — filter documents between stages to avoid unnecessary processing.
Keep configuration in the pipeline — override plugin defaults at the pipeline level rather than editing plugin code for each use case.

Next Steps

Create an Ontology — define extraction schemas
Create a Prompt — reusable LLM instruction templates

PLUGINSCreate a Benchmark WORKFLOWSCreate an Ontology