Why this matters

AI engineers spend significant time not building applications but running them: ingesting data, running eval suites, batch-processing prompts, exporting results, comparing prompt variants. A poorly-built script for any of these tasks becomes a reliability problem — a pipeline that crashes without saving partial results, an eval runner that gives no progress feedback on a 20-minute run, a data prep tool that silently skips malformed inputs.

This lesson teaches you to build these tools properly: structured CLI interfaces, progress tracking, partial failure handling, and machine-readable output.

Core concepts

Building CLI tools with typer

typer generates a full CLI from Python type annotations. No argparse boilerplate, automatic --help generation, and type validation on inputs:

import typer
from pathlib import Path

app = typer.Typer(name="eval-runner", help="Run evaluation suites against LLM outputs.")


@app.command()
def run(
    dataset: Path = typer.Argument(..., help="Path to golden dataset JSON file"),
    output: Path = typer.Option(Path("eval-results.json"), help="Output file for results"),
    model: str = typer.Option("claude-3-5-sonnet-20241022", help="Model ID to evaluate"),
    concurrency: int = typer.Option(5, min=1, max=20, help="Max concurrent requests"),
    pass_threshold: float = typer.Option(0.90, min=0.0, max=1.0, help="Minimum pass rate to exit 0"),
    verbose: bool = typer.Option(False, "--verbose", "-v"),
):
    import asyncio
    import json

    if not dataset.exists():
        typer.echo(f"Error: dataset not found at {dataset}", err=True)
        raise typer.Exit(1)

    typer.echo(f"Running eval on {dataset.name} with model {model}...")
    results, pass_rate = asyncio.run(_run_eval(dataset, model, concurrency, verbose))

    report = {
        "model": model,
        "dataset": str(dataset),
        "pass_rate": pass_rate,
        "total": len(results),
        "passed": sum(1 for r in results if r["passed"]),
        "failed": sum(1 for r in results if not r["passed"]),
        "cases": results,
    }
    output.write_text(json.dumps(report, indent=2))
    typer.echo(f"Pass rate: {pass_rate:.1%} ({report['passed']}/{report['total']})")
    typer.echo(f"Report written to {output}")

    if pass_rate < pass_threshold:
        typer.echo(f"FAILED: pass rate {pass_rate:.1%} below threshold {pass_threshold:.1%}", err=True)
        raise typer.Exit(1)


if __name__ == "__main__":
    app()

raise typer.Exit(1) on eval failure lets CI systems detect regressions automatically.

Progress tracking with rich

Long-running batch jobs need progress feedback. The rich library provides progress bars without custom terminal code:

from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn, TaskProgressColumn
from rich.console import Console

console = Console()


def process_batch_with_progress(items: list[dict], process_fn) -> list[dict]:
    results = []
    failed = []

    with Progress(
        SpinnerColumn(),
        TextColumn("[progress.description]{task.description}"),
        BarColumn(),
        TaskProgressColumn(),
        console=console,
    ) as progress:
        task = progress.add_task("Processing...", total=len(items))
        for item in items:
            try:
                result = process_fn(item)
                results.append(result)
            except Exception as exc:
                failed.append({"id": item.get("id"), "error": str(exc)})
            finally:
                progress.advance(task)

    console.print(f"[green]Done:[/green] {len(results)} processed, {len(failed)} failed")
    return results

Batch processing with partial failure handling

Batch scripts must not abort on individual item failures. Capture failures per item, continue processing, report at the end:

import json
import logging
from dataclasses import dataclass, field
from pathlib import Path

logger = logging.getLogger(__name__)


@dataclass
class BatchResult:
    processed: list[dict] = field(default_factory=list)
    failed: list[dict] = field(default_factory=list)

    @property
    def total(self) -> int:
        return len(self.processed) + len(self.failed)

    @property
    def success_rate(self) -> float:
        return len(self.processed) / self.total if self.total > 0 else 0.0

    def summary(self) -> str:
        return f"{len(self.processed)}/{self.total} succeeded ({self.success_rate:.1%}), {len(self.failed)} failed"


def run_batch(input_path: Path, output_path: Path, process_fn, overwrite: bool = False) -> BatchResult:
    if output_path.exists() and not overwrite:
        raise FileExistsError(f"{output_path} already exists. Pass overwrite=True to replace.")

    items = [json.loads(line) for line in input_path.read_text().splitlines() if line.strip()]
    logger.info("batch_start", extra={"total_items": len(items)})

    result = BatchResult()
    for item in items:
        try:
            result.processed.append(process_fn(item))
        except Exception as exc:
            result.failed.append({"id": item.get("id"), "error": str(exc)})
            logger.warning("item_failed", extra={"id": item.get("id"), "error": str(exc)})

    output_path.write_text("\n".join(json.dumps(r) for r in result.processed))
    logger.info("batch_complete", extra={"summary": result.summary()})
    return result

Machine-readable report output

Scripts that produce JSON output compose into pipelines. Support both human-readable and machine-readable output:

import json
from pathlib import Path


def write_report(results: dict, output_path: Path | None, json_output: bool) -> None:
    if json_output or output_path:
        serialized = json.dumps(results, indent=2)
        if output_path:
            output_path.write_text(serialized)
        else:
            print(serialized)
    else:
        print(f"Pass rate: {results['pass_rate']:.1%}")
        print(f"Total: {results['total']} | Passed: {results['passed']} | Failed: {results['failed']}")
        if results.get("failed_cases"):
            print("\nFailed cases:")
            for case in results["failed_cases"]:
                print(f"  {case['id']}: {', '.join(case['errors'])}")

Common mistakes

Scripts that abort on any single failure. A batch of 1000 items should not stop at item 42. Capture per-item errors and continue.

No overwrite protection. A script that silently overwrites its own output on rerun deletes the previous results. Require an explicit --overwrite flag.

Missing exit code conventions. A script that prints "FAILED" but exits with code 0 cannot be detected by CI. Exit with a non-zero code on any meaningful failure.

Progress output mixed with result output. Use stderr for progress and logging; use stdout or a file for machine-readable results.

Try it yourself

1. Build a typer CLI for an eval runner. Include arguments for the dataset path, output file, and model. Verify --help is informative and the tool exits with code 1 when the pass rate is below threshold.
2. Add a rich progress bar to a batch processing loop. Verify it shows remaining item count and updates in real time on a synthetic batch of 20 items.
3. Modify a batch script to capture per-item failures in a BatchResult dataclass. Verify that a batch with 2 intentionally broken items processes all items and reports 2 failures without aborting.