Version: 1.0.0 Last Updated: 2026-05-14 Status: Active

Pipeline Documentation¶

This directory contains documentation for all 19 pipeline microservices in src/pipelines/.

Overview¶

The pipeline microservices provide specialized data processing capabilities for various use cases including document processing, database extraction, content generation, and knowledge base ingestion.

Pipeline Categories¶

Data Processing Pipelines¶

Pipeline	Version	Description	Tests
process_pdf	0.2.2	PDF processing, OWL generation	14
process_excel	0.1.0	Excel to JSON conversion	2
webextract	0.1.1	Web content extraction	3
annotation	0.1.1	Entity annotation	1

AI Content Generation Pipelines¶

Pipeline	Version	Description	Tests
openai	0.1.0	OpenAI content generation	6
bedrock	0.1.0	AWS Bedrock content generation	3

Database Pipelines¶

Pipeline	Version	Description	Tests
mongo	0.1.1	MongoDB extraction	1
mssql	0.1.4	MS SQL extraction/CRUD	1

Vector Database Pipelines¶

Pipeline	Version	Description	Tests
milvus	0.1.2	Milvus vector indexing	3
azure_index	0.1.0	Azure Cognitive Search	1

Knowledge Management Pipelines¶

Pipeline	Version	Description	Tests
agno_pipeline	0.1.4	Agno KB ingestion	5
intent	0.1.0	Intent mapping rules	4

Analytics Pipelines¶

Pipeline	Version	Description	Tests
analytics_etl	0.1.2	Analytics ETL with MongoDB warehouse	-

Infrastructure Pipelines¶

Pipeline	Version	Description	Tests
init	0.1.3	System initialization	4

Client-Specific Pipelines¶

Pipeline	Version	Description	Tests
`client-a`	0.1.0	[Workforce Pilot] document processing	1
`client-b`	0.1.1	[Children's Hospital] ingestion pipeline	3
`client-c`	0.1.1	[Professional Education] PDF intake	2

Templates¶

Pipeline	Version	Description	Tests
template_pipeline	0.1.0	Pipeline template	3
client_template	0.1.0	Client pipeline template	3

Data Flow Patterns¶

Document Processing Flow¶

PDF/Excel/Web Sources
        │
        ▼
┌─────────────────────┐
│  process_pdf        │
│  process_excel      │
│  webextract         │
└─────────────────────┘
        │
        ▼
┌─────────────────────┐
│  openai / bedrock   │ (Content Generation)
│  annotation         │ (Entity Extraction)
└─────────────────────┘
        │
        ▼
┌─────────────────────┐
│  milvus             │
│  azure_index        │ (Vector Storage)
│  agno_pipeline      │ (Knowledge Base)
└─────────────────────┘

Database Extraction Flow¶

Database Sources
        │
        ▼
┌─────────────────────┐
│  mssql              │
│  mongo              │
└─────────────────────┘
        │
        ▼
┌─────────────────────┐
│  annotation         │ (Entity Extraction)
└─────────────────────┘
        │
        ▼
┌─────────────────────┐
│  agno_pipeline      │ (Knowledge Base)
└─────────────────────┘

Common Architecture¶

Directory Structure¶

src/pipelines/{name}/
├── bast_{name}_pipeline/    # Module code
│   ├── __init__.py
│   ├── bp/                   # Business Process layer
│   └── svc/                  # Service layer
├── tests/                    # Unit tests
├── integration/              # Integration tests
├── BUILD                     # Pants build config
├── pyproject.toml           # Poetry config
├── README.md                # Documentation
├── CHANGELOG.md             # Version history
└── VERSION                  # Current version

Base Class Pattern¶

from bast_ai.base import BastObject

class MyPipeline(BastObject):
    def __init__(self, catname: str = None):
        BastObject.__init__(self, __name__)
        self._cat = get_bast_config().get_cat(catname)

Action Pattern¶

from typing import Final

class MyPipeline:
    ACTION_NAME: Final[str] = "action_name"

    def process(self, action: str):
        if action == self.ACTION_NAME:
            # processing logic
        else:
            raise NotImplementedError(action)

Test Coverage Summary¶

Pipeline	Unit Tests	Integration Tests
process_pdf	4	10
openai	6	0
agno_pipeline	2	3
mssql	1	0
bedrock	3	0
template_pipeline	3	0
Others	1-3 each	0-1 each

Total: 47 unit test files, 13 integration test files

Running Pipelines¶

Via Pants¶

# Run integration test
pants run src/pipelines/process_pdf/integration/pdf_conversion.py

# Run unit tests
pants test src/pipelines/process_pdf:: -- -s --no-header

Programmatic Usage¶

from bast_process_pdf.bp import ProcessPDFPipeline

pipeline = ProcessPDFPipeline(catname="mycat")
result = pipeline.process(action="convert_pdf", input_path="doc.pdf")

Issues & Improvements¶

Current Issues¶

Inconsistent error handling: Some pipelines don't follow consistent error patterns
Limited retry logic: Most pipelines lack automatic retry for transient failures
Missing metrics: No standardized performance metrics collection

Suggested Improvements¶

Unified pipeline base class: Create BastPipeline base with common functionality
Progress callbacks: Standardize progress reporting across all pipelines
Pipeline orchestration: Add workflow engine for complex multi-pipeline flows
Observability: Add OpenTelemetry tracing
Configuration validation: Validate pipeline configs at startup