CrewAI real-world example: multi-step data pipeline QA

A crewai data pipeline qa example should do more than assert row counts. You need agents that understand your schema contracts, enforce business rules at the row level, and catch silent drift before it corrupts downstream models. This guide walks through a production-grade CrewAI setup that runs as a CI gate or scheduled job, with real code you can drop into your repository.

Step 1: Define the pipeline contract

Before writing agents, codify what “correct” looks like. Create a pipeline_contract.yaml that lives alongside your dbt models or Airflow DAGs. This becomes the single source of truth for every agent.

# pipeline_contract.yaml
version: "1.3"
tables:
  - name: fact_orders
    schema:
      - column: order_id
        type: string
        nullable: false
        unique: true
      - column: customer_id
        type: string
        nullable: false
      - column: order_ts
        type: timestamp
        nullable: false
      - column: amount_usd
        type: numeric(12,2)
        nullable: false
        min: 0.01
      - column: status
        type: string
        nullable: false
        allowed_values: ["pending", "paid", "shipped", "cancelled", "refunded"]
    business_rules:
      - rule: "no_future_orders"
        sql: "order_ts <= now()"
      - rule: "refund_implies_paid"
        sql: "status = 'refunded' AND EXISTS (SELECT 1 FROM fact_orders f2 WHERE f2.order_id = fact_orders.order_id AND f2.status = 'paid')"
    freshness:
      max_lag_hours: 4
  - name: dim_customers
    schema:
      - column: customer_id
        type: string
        nullable: false
        unique: true
      - column: signup_ts
        type: timestamp
        nullable: false
      - column: tier
        type: string
        allowed_values: ["free", "pro", "enterprise"]
    freshness:
      max_lag_hours: 24

Store this in version control. Your CI pipeline fails if the contract changes without a corresponding PR.

Step 2: Scaffold the CrewAI project

Install the minimal dependencies. CrewAI 0.28+ works with any OpenAI-compatible endpoint — useful if you route through a gateway that handles fallback and usage metering across providers.

pip install crewai==0.28.10 pyyaml sqlalchemy psycopg2-binary pandas great-expectations

Create the project structure:

pipeline_qa/
├── config/
│   └── pipeline_contract.yaml
├── crew/
│   ├── __init__.py
│   ├── agents.py
│   ├── tasks.py
│   └── tools.py
├── main.py
└── requirements.txt

Step 3: Build reusable tools

Agents need deterministic tools — not LLM hallucinations — for database introspection and rule evaluation. Wrap SQLAlchemy and Great Expectations in thin tools your agents can call.

# crew/tools.py
from typing import Any, Dict, List
from sqlalchemy import create_engine, inspect, text
from sqlalchemy.engine import Engine
import pandas as pd
import great_expectations as gx
from crewai.tools import BaseTool

class DatabaseInspector(BaseTool):
    name: str = "database_inspector"

    def __init__(self, dsn: str):
        super().__init__()
        self._engine = create_engine(dsn, pool_pre_ping=True)

    @property
    def engine(self) -> Engine:
        return self._engine

    def _run(self, query: str, params: Dict[str, Any] | None = None) -> List[Dict[str, Any]]:
        with self._engine.connect() as conn:
            result = conn.execute(text(query), params or {})
            return [dict(row._mapping) for row in result]

    def get_schema(self, table: str) -> List[Dict[str, Any]]:
        insp = inspect(self._engine)
        return insp.get_columns(table)

class GreatExpectationsRunner(BaseTool):
    name: str = "ge_runner"

    def __init__(self, dsn: str):
        super().__init__()
        self._context = gx.get_context(mode="ephemeral")
        self._dsn = dsn

    def _run(self, table: str, expectations: List[Dict[str, Any]]) -> Dict[str, Any]:
        datasource = self._context.data_sources.add_sqlite(name="temp", connection_string=self._dsn)
        asset = datasource.add_table_asset(name=table, table_name=table)
        batch_request = asset.build_batch_request()
        suite = self._context.suites.add(gx.ExpectationSuite(name=f"{table}_suite"))
        for exp in expectations:
            suite.add_expectation(gx.expectations.Expectation(**exp))
        validator = self._context.get_validator(batch_request=batch_request, expectation_suite=suite)
        result = validator.validate()
        return {
            "success": result.success,
            "results": [r.to_json_dict() for r in result.results],
            "statistics": result.statistics
        }

These tools keep SQL execution out of the LLM loop. The agent decides what to check; the tool executes it deterministically.

Step 4: Define specialized agents

Each agent owns a distinct QA domain. Give them focused backstories and tool access so they don’t step on each other.

# crew/agents.py
from crewai import Agent
from crew.tools import DatabaseInspector, GreatExpectationsRunner

def build_agents(dsn: str) -> Dict[str, Agent]:
    inspector = DatabaseInspector(dsn)
    ge_runner = GreatExpectationsRunner(dsn)

    schema_agent = Agent(
        role="Schema Contract Validator",
        goal="Verify every column matches the pipeline contract: type, nullability, uniqueness, allowed values.",
        backstory=(
            "You are a data architect who treats schema drift as a P0 incident. "
            "You compare live information_schema against the YAML contract and "
            "report every mismatch with exact column and expected vs actual."
        ),
        tools=[inspector],
        verbose=True,
        allow_delegation=False
    )

    business_rule_agent = Agent(
        role="Business Rule Enforcer",
        goal="Execute every SQL rule in the contract and flag violations with row identifiers.",
        backstory=(
            "You know that referential integrity and state-machine rules (e.g., refund implies paid) "
            "are where silent corruption hides. You run each rule as a parameterized query and "
            "return offending primary keys so engineers can replay fixes."
        ),
        tools=[inspector],
        verbose=True,
        allow_delegation=False
    )

    freshness_agent = Agent(
        role="Freshness SLA Monitor",
        goal="Confirm each table meets its max_lag_hours SLA against the latest event timestamp.",
        backstory=(
            "You treat stale data as a reliability bug. You query the maximum event timestamp, "
            "compare to now(), and calculate lag in hours. You also check row count growth "
            "against a 7-day baseline to catch silent pipeline stalls."
        ),
        tools=[inspector],
        verbose=True,
        allow_delegation=False
    )

    drift_agent = Agent(
        role="Statistical Drift Detector",
        goal="Run Great Expectations suites to detect distribution shifts in key numeric and categorical columns.",
        backstory=(
            "You catch the slow-moving disasters: average order value creeping up, "
            "tier distribution skewing, null rates rising. You use KS-test for numerics "
            "and chi-square for categoricals against a 30-day rolling baseline."
        ),
        tools=[ge_runner],
        verbose=True,
        allow_delegation=False
    )

    return {
        "schema": schema_agent,
        "business_rules": business_rule_agent,
        "freshness": freshness_agent,
        "drift": drift_agent
    }

Step 5: Wire tasks to agents

Tasks are the executable units. Each task reads the contract, plans its queries, and emits a structured JSON result the orchestrator can aggregate.

# crew/tasks.py
from crewai import Task
from crew.agents import build_agents
import yaml
from pathlib import Path

CONTRACT_PATH = Path(__file__).parent.parent / "config" / "pipeline_contract.yaml"

def load_contract() -> dict:
    with open(CONTRACT_PATH) as f:
        return yaml.safe_load(f)

def build_tasks(dsn: str) -> list[Task]:
    agents = build_agents(dsn)
    contract = load_contract()

    schema_task = Task(
        description=(
            "For each table in the contract, retrieve the live schema from information_schema. "
            "Compare every column: name, type, nullable, unique. For columns with allowed_values, "
            "run a distinct query and verify the set is a subset. Output JSON: "
            '{"table": "", "mismatches": [{"column": "", "expected": "", "actual": "", "severity": "error|warn"}]}'
        ),
        expected_output="Valid JSON array of mismatch objects, empty if clean.",
        agent=agents["schema"]
    )

    business_rule_task = Task(
        description=(
            "For each table, iterate its business_rules. Execute the SQL as a SELECT that returns "
            "violating primary keys (order_id, customer_id). Parameterize with the contract values. "
            'Output JSON: {"table": "", "rule": "", "violating_keys": [], "row_count": 0}'
        ),
        expected_output="Valid JSON array of violation objects, empty if clean.",
        agent=agents["business_rules"]
    )

    freshness_task = Task(
        description=(
            "For each table, find the max(event_timestamp_column) — infer from columns ending in _ts or _at. "
            "Compute lag_hours = EXTRACT(EPOCH FROM (now() - max_ts))/3600. Compare to max_lag_hours. "
            "Also compute row count vs 7-day average (SELECT count(*) FROM table WHERE event_ts > now() - interval '7 days'). "
            'Flag if lag > SLA or row count < 50% of baseline. Output JSON with lag_hours, baseline_count, current_count.'
        ),
        expected_output="Valid JSON array of freshness reports per table.",
        agent=agents["freshness"]
    )

    drift_task = Task(
        description=(
            "For each table, build a Great Expectations suite from the contract's numeric and categorical columns. "
            "Use expect_column_kl_divergence_to_be_less_than (threshold 0.1) for numerics, "
            "expect_column_chisquare_test_p_value_to_be_greater_than (threshold 0.05) for categoricals "
            "against a 30-day reference window. Output the GE validation result JSON."
        ),
        expected_output="Valid GE validation result JSON per table.",
        agent=agents["drift"]
    )

    return [schema_task, business_rule_task, freshness_task, drift_task]

Step 6: Orchestrate and aggregate results

The main entry point runs the crew, collects structured outputs, and exits non-zero if any agent reports failures. This makes it a first-class CI gate.

# main.py
import os
import json
import sys
from pathlib import Path
from crewai import Crew, Process
from crew.tasks import build_tasks

DSN = os.getenv("PIPELINE_QA_DSN", "postgresql://user:pass@localhost:5432/analytics")

def main() -> int:
    tasks = build_tasks(DSN)

    crew = Crew(
        agents=[t.agent for t in tasks],
        tasks=tasks,
        process=Process.sequential,  # deterministic order; parallel is fine if tools are thread-safe
        verbose=True,
        memory=False
    )

    print("Starting pipeline QA crew...", file=sys.stderr)
    results = crew.kickoff()

    # results is a list of TaskOutput objects; each .raw contains the agent's final string
    all_clean = True
    report = {"schema": [], "business_rules": [], "freshness": [], "drift": []}

    for i, task_output in enumerate(results):
        key = ["schema", "business_rules", "freshness", "drift"][i]
        try:
            parsed = json.loads(task_output.raw)
            report[key] = parsed
            # Heuristic: non-empty array or success=false means failure
            if isinstance(parsed, list) and len(parsed) > 0:
                all_clean = False
            elif isinstance(parsed, dict) and parsed.get("success") is False:
                all_clean = False
        except json.JSONDecodeError:
            print(f"WARNING: {key} agent returned non-JSON: {task_output.raw[:200]}", file=sys.stderr)
            all_clean = False

    # Write machine-readable artifact for downstream steps
    artifact_path = Path("pipeline_qa_report.json")
    artifact_path.write_text(json.dumps(report, indent=2, default=str))
    print(f"Report written to {artifact_path}", file=sys.stderr)

    # Human summary
    print("\n=== PIPELINE QA SUMMARY ===")
    for k, v in report.items():
        status = "PASS" if (isinstance(v, list) and len(v) == 0) or (isinstance(v, dict) and v.get("success")) else "FAIL"
        print(f"  {k}: {status}")

    return 0 if all_clean else 1

if __name__ == "__main__":
    sys.exit(main())

Step 7: Wire into CI/CD

Add a GitHub Actions workflow that runs on every PR touching the models/ or dags/ directories. The job spins up a test database (or uses a dedicated QA schema), loads the contract, and runs the crew.

# .github/workflows/pipeline-qa.yml
name: Pipeline QA
on:
  pull_request:
    paths:
      - 'models/**'
      - 'dags/**'
      - 'pipeline_qa/**'
      - 'pipeline_contract.yaml'

jobs:
  qa:
    runs-on: ubuntu-latest
    services:
      postgres:
        image: postgres:16
        env:
          POSTGRES_DB: analytics
          POSTGRES_USER: test
          POSTGRES_PASSWORD: test
        ports: ["5432:5432"]
        options: >-
          --health-cmd "pg_isready -U test"
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - name: Install deps
        run: |
          pip install -r pipeline_qa/requirements.txt
      - name: Load test fixtures
        run: |
          PGPASSWORD=test psql -h localhost -U test -d analytics -f tests/fixtures/seed.sql
        env:
          PGPASSWORD: test
      - name: Run CrewAI QA
        env:
          PIPELINE_QA_DSN: postgresql://test:test@localhost:5432/analytics
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
          OPENAI_API_BASE: ${{ secrets.OPENAI_API_BASE }}  # optional: route via n4n.ai for fallback/metering
        run: |
          cd pipeline_qa && python main.py
      - name: Upload QA report
        if: always()
        uses: actions/upload-artifact@v4
        with:
          name: pipeline-qa-report
          path: pipeline_qa/pipeline_qa_report.json

The OPENAI_API_BASE override lets you route through a gateway that handles provider fallback and per-token metering without changing agent code.

Step 8: Verify locally before pushing

Run the full stack against a local Postgres with a known-good seed. This catches contract drift early.

# Start Postgres
docker run -d --name qa-db \
  -e POSTGRES_DB=analytics -e POSTGRES_USER=test -e POSTGRES_PASSWORD=test \
  -p 5432:5432 postgres:16

# Wait for readiness
sleep 5

# Load schema + seed data
PGPASSWORD=test psql -h localhost -U test -d analytics -f tests/fixtures/schema.sql
PGPASSWORD=test psql -h localhost -U test -d analytics -f tests/fixtures/seed.sql

# Run crew
export PIPELINE_QA_DSN="postgresql://test:test@localhost:5432/analytics"
export OPENAI_API_KEY="sk-..."
cd pipeline_qa && python main.py

Expected output on clean data:

=== PIPELINE QA SUMMARY ===
  schema: PASS
  business_rules: PASS
  freshness: PASS
  drift: PASS

Exit code 0. The pipeline_qa_report.json artifact contains empty arrays for each check.

Step 9: Inject a failure to confirm detection

Modify seed.sql to violate a rule — e.g., insert an order with amount_usd = -5.00 or a future order_ts. Re-run. You should see:

=== PIPELINE QA SUMMARY ===
  schema: PASS
  business_rules: FAIL
  freshness: PASS
  drift: PASS

And the report JSON will include the violating order_id under business_rules. Fix the seed, re-run, confirm green.

Step 10: Extend for production realities

Three patterns make this survive real workloads:

Partition-aware freshness — For partitioned tables, query the max timestamp per partition and require all recent partitions to meet SLA, not just the global max.

-- Freshness agent can generate this dynamically
SELECT partition_name, max(event_ts) as max_ts
FROM information_schema.partitions p
JOIN fact_orders f ON f.partition = p.partition_name
WHERE p.table_name = 'fact_orders'
GROUP BY partition_name

Incremental drift baselines — Store the 30-day reference statistics in a drift_baselines table updated nightly. The drift agent reads from there instead of recomputing the full window every run.

Slack/ PagerDuty alerting — Add a final aggregator task that posts a formatted summary to your incident channel on any FAIL. Keep the crew focused on detection; let your alerting layer handle routing.

# crew/tasks.py (addition)
alert_task = Task(
    description=(
        "Receive the aggregated report from previous tasks. If any check failed, "
        "format a Slack message with table, check type, and top 5 violations. "
        "Post to the webhook URL in SLACK_WEBHOOK_URL env var."
    ),
    expected_output="Posted to Slack or no-op if all clean.",
    agent=Agent(
        role="Alert Dispatcher",
        goal="Notify on-call only when actionable failures exist.",
        backstory="You hate alert fatigue. You only ping humans when there are concrete rows to fix.",
        tools=[SlackWebhookTool()],  # implement similarly to DatabaseInspector
        allow_delegation=False
    ),
    context=[schema_task, business_rule_task, freshness_task, drift_task]
)

What this buys you

A crewai data pipeline qa example built this way catches three failure modes that unit tests miss: schema drift from upstream migrations, business logic violations that only appear in production data, and silent distribution shifts that degrade model performance weeks before anyone notices. The agents are replaceable — swap the LLM, swap the tools, keep the contract as the invariant. Your CI gate becomes a living document of what “correct” means, enforced by code that reads the same YAML your engineers review.