ai-pipelines

YAML-driven AI/LLM pipeline executor. Define multi-step workflows in YAML, run them with a single Python call.

What it does

• Describe pipelines as YAML files: read files, chunk text, call LLMs, loop, transform data, evaluate outputs
• Steps reference each other by name using JSONata expressions
• LLM prompt templates use Jinja2
• Built-in LLM-as-judge evaluation with 7 scoring strategies
• Returns per-step timing and total cost

Installation

Requires Python 3.12+ and uv.

uv sync

Quick Start

import asyncio
from ai_pipelines import load_pipeline, run_pipeline

async def main():
    pipeline = load_pipeline("my_pipeline.yaml")
    result = await run_pipeline(pipeline, {"text": "Hello world"})
    print(result.output)
    print(f"Cost: ${result.total_cost_usd:.4f}")

asyncio.run(main())

Minimal pipeline YAML

input:
  type: object
  properties:
    text: { type: string }
  required: [text]

steps:
  - kind: prompt
    name: summary
    model: haiku
    arguments: '{"text": input.text}'
    template: "Summarize this: {{ args.text }}"

Pipeline YAML Reference

Every pipeline has input (JSON Schema) and steps (ordered list). Each step requires kind and name. All prior step results are available by name via JSONata expressions.

Step types

Data flow

• input.field — pipeline input fields
• step_name.field — prior step output
• Inside for_each, item is the current element and item_index its position
• prompt templates use {{ args.field }} where args is the resolved arguments dict

Structured output

Add output with a JSON Schema to a prompt step to get a typed dict back:

- kind: prompt
  name: result
  model: sonnet
  arguments: '{"doc": input.text}'
  template: "Extract key points from: {{ args.doc }}"
  output:
    type: object
    properties:
      points: { type: array, items: { type: string } }
    required: [points]

Evaluate strategies

Running Tests

just test
# or
uv run pytest

Project Structure

src/ai_pipelines/
  __init__.py         Public API
  models.py           Pydantic models for all step types + discriminated union
  executor.py         Async orchestrator (run_pipeline)
  loader.py           YAML parsing + JSON Schema validation
  context.py          PipelineContext: scoped dict for step results
  expressions.py      JSONata evaluator
  templates.py        Jinja2 renderer
  validator.py        Static pre-flight validation
  errors.py           Exception hierarchy
  pipeline_logger.py  Structured JSON-lines logging
  steps/              One file per step kind
e2e/                  End-to-end example scripts and pipelines
tests/                pytest suite

Public API

from ai_pipelines import (
    configure_logging,
    load_pipeline,
    validate_pipeline,
    load_and_validate_pipeline,
    run_pipeline,
    validate_input,
)

All exceptions inherit from PipelineError: