ai-pipeline-orchestration

Name: ai-pipeline-orchestration
Author: BagelHole

42stars4forks

How to install ai-pipeline-orchestration?

bagelhole/devops-security-agent-skills/ai-pipeline-orchestration

$npx -y skills add BagelHole/DevOps-Security-Agent-Skills --skill ai-pipeline-orchestration

Installs into the current project.

Summary

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.

Files · 1

View on GitHub

SKILL.md

1	# AI Pipeline Orchestration
2
3	Build reliable, observable AI workflows — from document ingestion to batch inference to model training pipelines.
4
5	## When to Use This Skill
6
7	Use this skill when:
8	- Scheduling recurring RAG document ingestion and re-indexing
9	- Orchestrating multi-step batch LLM processing workflows
10	- Running nightly model evaluation and fine-tuning jobs
11	- Building ETL pipelines that feed into AI models
12	- Managing dependencies between data preparation and model serving
13
14	## Tool Selection
15
16	\| Tool \| Best For \| Complexity \| GPU Jobs \|
17	\|------\|----------\|------------\|----------\|
18	\| Prefect \| Modern Python-first; easy to adopt \| Low \| Good \|
19	\| Airflow \| Complex DAGs; large teams; existing usage \| High \| Good \|
20	\| Dagster \| Asset-centric; strong data lineage \| Medium \| Excellent \|
21	\| Temporal \| Long-running workflows; reliability-first \| Medium \| Good \|
22
23	## Prefect — Quick Start
24
25	```bash
26	pip install prefect prefect-kubernetes
27
28	# Start Prefect server (or use Prefect Cloud)
29	prefect server start
30
31	# In another terminal
32	prefect worker start --pool default-agent-pool
33	```
34
35	## Prefect: RAG Ingestion Pipeline
36
37	```python
38	from prefect import flow, task, get_run_logger
39	from prefect.tasks import task_input_hash
40	from datetime import timedelta
41	import hashlib
42
43	@task(cache_key_fn=task_input_hash, cache_expiration=timedelta(hours=24))
44	def fetch_documents(source_url: str) -> list[dict]:
45	"""Fetch documents from source; cached to avoid re-fetching."""
46	logger = get_run_logger()
47	logger.info(f"Fetching from {source_url}")
48	# ... fetch logic
49	return documents
50
51	@task(retries=3, retry_delay_seconds=30)
52	def chunk_and_embed(documents: list[dict]) -> list[dict]:
53	"""Chunk documents and generate embeddings with retry on failure."""
54	from sentence_transformers import SentenceTransformer
55	model = SentenceTransformer("BAAI/bge-large-en-v1.5")
56	chunks = []
57	for doc in documents:
58	doc_chunks = chunk_text(doc["content"])
59	embeddings = model.encode(doc_chunks, batch_size=64)
60	for chunk, emb in zip(doc_chunks, embeddings):
61	chunks.append({"text": chunk, "embedding": emb.tolist(),
62	"source": doc["url"], "doc_hash": doc["hash"]})
63	return chunks
64
65	@task(retries=2)
66	def upsert_to_vector_store(chunks: list[dict]) -> int:
67	"""Upsert embeddings to Qdrant, skip unchanged documents."""
68	from qdrant_client import QdrantClient
69	client = QdrantClient("http://qdrant:6333")
70	client.upsert(collection_name="knowledge-base", points=[...])
71	return len(chunks)
72
73	@flow(name="rag-ingestion", log_prints=True)
74	def rag_ingestion_pipeline(sources: list[str]):
75	"""Full RAG ingestion flow — runs daily."""
76	logger = get_run_logger()
77	total = 0
78	for source in sources:
79	docs = fetch_documents(source)
80	chunks = chunk_and_embed(docs)
81	count = upsert_to_vector_store(chunks)
82	total += count
83	logger.info(f"Ingested {count} chunks from {source}")
84	logger.info(f"Pipeline complete: {total} total chunks indexed")
85
86	if __name__ == "__main__":
87	rag_ingestion_pipeline.serve(
88	name="daily-rag-ingestion",
89	cron="0 2 * * *", # 2 AM daily
90	parameters={"sources": ["https://docs.myapp.com", "https://api.myapp.com/kb"]},
91	)
92	```
93
94	## Prefect: Batch LLM Inference Pipeline
95
96	```python
97	from prefect import flow, task
98	from prefect.concurrency.sync import concurrency
99	import asyncio
100	from openai import AsyncOpenAI
101
102	@task(retries=3, retry_delay_seconds=60)
103	async def process_batch(items: list[dict], model: str = "gpt-4o-mini") -> list[dict]:
104	"""Process a batch of items through LLM with rate limit protection."""
105	client = AsyncOpenAI()
106	async with concurrency("openai-api", occupy=len(items)): # rate limit
107	tasks = [
108	client.chat.completions.create(
109	model=model,
110	messages=[{"role": "user", "content": item["prompt"]}],
111	max_tokens=256,
112	)
113	for item in items
114	]
115	responses = await asyncio.gather(*tasks, return_exceptions=True)
116
117	results = []
118	for item, response in zip(items, responses):
119	if isinstance(response, Exception):
120	results.append({**item, "error": str(response), "output": None})
121	else:
122	results.append({**item, "output": response.choices[0].message.content})
123	return results
124
125	@flow(name="batch-llm-inference")
126	async def batch_inference_flow(input_file: str, output_file: str, batch_size: int = 50):
127	import json
128	items = [json.loads(line) for line in open(input_file)]
129	batches = [items[i:i+batch_size] for i in range

People who liked this also viewed

6 picks

Type

ai-pipeline-orchestration

42stars4forks

View on GitHub

How to install ai-pipeline-orchestration?

bagelhole/devops-security-agent-skills/ai-pipeline-orchestration

$npx -y skills add BagelHole/DevOps-Security-Agent-Skills --skill ai-pipeline-orchestration

Installs into the current project.

Summary

Files · 1

View on GitHub

SKILL.md

1	# AI Pipeline Orchestration
2
3	Build reliable, observable AI workflows — from document ingestion to batch inference to model training pipelines.
4
5	## When to Use This Skill
6
7	Use this skill when:
8	- Scheduling recurring RAG document ingestion and re-indexing
9	- Orchestrating multi-step batch LLM processing workflows
10	- Running nightly model evaluation and fine-tuning jobs
11	- Building ETL pipelines that feed into AI models
12	- Managing dependencies between data preparation and model serving
13
14	## Tool Selection
15
16	\| Tool \| Best For \| Complexity \| GPU Jobs \|
17	\|------\|----------\|------------\|----------\|
18	\| Prefect \| Modern Python-first; easy to adopt \| Low \| Good \|
19	\| Airflow \| Complex DAGs; large teams; existing usage \| High \| Good \|
20	\| Dagster \| Asset-centric; strong data lineage \| Medium \| Excellent \|
21	\| Temporal \| Long-running workflows; reliability-first \| Medium \| Good \|
22
23	## Prefect — Quick Start
24
25	```bash
26	pip install prefect prefect-kubernetes
27
28	# Start Prefect server (or use Prefect Cloud)
29	prefect server start
30
31	# In another terminal
32	prefect worker start --pool default-agent-pool
33	```
34
35	## Prefect: RAG Ingestion Pipeline
36
37	```python
38	from prefect import flow, task, get_run_logger
39	from prefect.tasks import task_input_hash
40	from datetime import timedelta
41	import hashlib
42
43	@task(cache_key_fn=task_input_hash, cache_expiration=timedelta(hours=24))
44	def fetch_documents(source_url: str) -> list[dict]:
45	"""Fetch documents from source; cached to avoid re-fetching."""
46	logger = get_run_logger()
47	logger.info(f"Fetching from {source_url}")
48	# ... fetch logic
49	return documents
50
51	@task(retries=3, retry_delay_seconds=30)
52	def chunk_and_embed(documents: list[dict]) -> list[dict]:
53	"""Chunk documents and generate embeddings with retry on failure."""
54	from sentence_transformers import SentenceTransformer
55	model = SentenceTransformer("BAAI/bge-large-en-v1.5")
56	chunks = []
57	for doc in documents:
58	doc_chunks = chunk_text(doc["content"])
59	embeddings = model.encode(doc_chunks, batch_size=64)
60	for chunk, emb in zip(doc_chunks, embeddings):
61	chunks.append({"text": chunk, "embedding": emb.tolist(),
62	"source": doc["url"], "doc_hash": doc["hash"]})
63	return chunks
64
65	@task(retries=2)
66	def upsert_to_vector_store(chunks: list[dict]) -> int:
67	"""Upsert embeddings to Qdrant, skip unchanged documents."""
68	from qdrant_client import QdrantClient
69	client = QdrantClient("http://qdrant:6333")
70	client.upsert(collection_name="knowledge-base", points=[...])
71	return len(chunks)
72
73	@flow(name="rag-ingestion", log_prints=True)
74	def rag_ingestion_pipeline(sources: list[str]):
75	"""Full RAG ingestion flow — runs daily."""
76	logger = get_run_logger()
77	total = 0
78	for source in sources:
79	docs = fetch_documents(source)
80	chunks = chunk_and_embed(docs)
81	count = upsert_to_vector_store(chunks)
82	total += count
83	logger.info(f"Ingested {count} chunks from {source}")
84	logger.info(f"Pipeline complete: {total} total chunks indexed")
85
86	if __name__ == "__main__":
87	rag_ingestion_pipeline.serve(
88	name="daily-rag-ingestion",
89	cron="0 2 * * *", # 2 AM daily
90	parameters={"sources": ["https://docs.myapp.com", "https://api.myapp.com/kb"]},
91	)
92	```
93
94	## Prefect: Batch LLM Inference Pipeline
95
96	```python
97	from prefect import flow, task
98	from prefect.concurrency.sync import concurrency
99	import asyncio
100	from openai import AsyncOpenAI
101
102	@task(retries=3, retry_delay_seconds=60)
103	async def process_batch(items: list[dict], model: str = "gpt-4o-mini") -> list[dict]:
104	"""Process a batch of items through LLM with rate limit protection."""
105	client = AsyncOpenAI()
106	async with concurrency("openai-api", occupy=len(items)): # rate limit
107	tasks = [
108	client.chat.completions.create(
109	model=model,
110	messages=[{"role": "user", "content": item["prompt"]}],
111	max_tokens=256,
112	)
113	for item in items
114	]
115	responses = await asyncio.gather(*tasks, return_exceptions=True)
116
117	results = []
118	for item, response in zip(items, responses):
119	if isinstance(response, Exception):
120	results.append({**item, "error": str(response), "output": None})
121	else:
122	results.append({**item, "output": response.choices[0].message.content})
123	return results
124
125	@flow(name="batch-llm-inference")
126	async def batch_inference_flow(input_file: str, output_file: str, batch_size: int = 50):
127	import json
128	items = [json.loads(line) for line in open(input_file)]
129	batches = [items[i:i+batch_size] for i in range

People who liked this also viewed

6 picks

Type