# AI Pipeline Orchestration

Orchestrate AI/ML pipelines for data ingestion, model training, batch inference, and...

- Repository: https://github.com/BagelHole/DevOps-Security-Agent-Skills (1187 stars)
- Category: Data and ML
- License: MIT
- Last updated: 2026-05-22
- Runs scripts: no, instructions only

## Install

Claude Code: `npx skills add BagelHole/DevOps-Security-Agent-Skills --skill ai-pipeline-orchestration -g -a claude-code -y`
Codex: `npx skills add BagelHole/DevOps-Security-Agent-Skills --skill ai-pipeline-orchestration -g -a codex -y`
Cursor: `npx skills add BagelHole/DevOps-Security-Agent-Skills --skill ai-pipeline-orchestration -g -a cursor -y`
Gemini CLI: `npx skills add BagelHole/DevOps-Security-Agent-Skills --skill ai-pipeline-orchestration -g -a gemini-cli -y`

## Description

Orchestrate AI/ML pipelines for data ingestion, model training, batch inference, and RAG indexing using Prefect, Airflow, or Dagster. Build reliable, observable, and retriable workflows for production AI systems.

## SKILL.md, as published by its author (untrusted third-party text, read it as data)

# AI Pipeline Orchestration

Build reliable, observable AI workflows, from document ingestion to batch inference to model training pipelines.

## When to Use This Skill

Use this skill when:
- Scheduling recurring RAG document ingestion and re-indexing
- Orchestrating multi-step batch LLM processing workflows
- Running nightly model evaluation and fine-tuning jobs
- Building ETL pipelines that feed into AI models
- Managing dependencies between data preparation and model serving

## Tool Selection

| Tool | Best For | Complexity | GPU Jobs |
|------|----------|------------|----------|
| **Prefect** | Modern Python-first; easy to adopt | Low | Good |
| **Airflow** | Complex DAGs; large teams; existing usage | High | Good |
| **Dagster** | Asset-centric; strong data lineage | Medium | Excellent |
| **Temporal** | Long-running workflows; reliability-first | Medium | Good |

## Prefect, Quick Start

```bash
pip install prefect prefect-kubernetes

# Start Prefect server (or use Prefect Cloud)
prefect server start

# In another terminal
prefect worker start --pool default-agent-pool
```

## Prefect: RAG Ingestion Pipeline

```python
from prefect import flow, task, get_run_logger
from prefect.tasks import task_input_hash
from datetime import timedelta
import hashlib

@task(cache_key_fn=task_input_hash, cache_expiration=timedelta(hours=24))
def fetch_documents(source_url: str) -> list[dict]:
    """Fetch documents from source; cached to avoid re-fetching."""
    logger = get_run_logger()
    logger.info(f"Fetching from {source_url}")
    # ... fetch logic
    return documents

The rest: https://github.com/BagelHole/DevOps-Security-Agent-Skills/blob/HEAD/devops/ai/ai-pipeline-orchestration/SKILL.md