Before and after¶
Four pipelines that teams write all the time, each in two versions: the script people usually write (pandas, a loop calling the model, sqlite3) and the agentl pipeline. Both run on the same data with the same model and must write the same result; the harness checks it.
python -m cases # all four, offline, about 10 seconds
python -m cases --only tickets --json cases.json
The model is a stand-in with a fixed 20 ms per call and deterministic answers, so results can be compared exactly and every call counted. With a real model, the cost follows the call counts.
Results¶
| Case | Records | Lines before | Lines with agentl | Model calls | Calls to recover from a failed write | Calls to re-run unchanged data |
|---|---|---|---|---|---|---|
| Support ticket triage | 40 | 8 | 17 | 40 | 80 → 40 | 40 → 0 |
| Invoice field extraction | 40 | 15 | 21 | 40 | 80 → 40 | 40 → 0 |
| Web pages to summaries | 6 | 19 | 19 | 6 | 12 → 6 | 6 → 0 |
| Daily sales report to a shared folder | 4 | 18 | 30 (YAML 22 + pandas 8) | 0 | 0 → 0 | 0 → 0 |
"A failed write" means the output was unavailable on the first try (a locked file, a share that is down, a full disk), then fixed, and the job run again. The hand-written script pays every model call again; agentl only re-runs what did not finish, because each answer is cached per record. "Re-run unchanged data" is the same job run again on the same input: the script calls the model for every record, agentl for none.
What it does not show¶
- Lines of code are not lower with agentl. The YAML is about as long as a minimal script, or longer. The scripts here are the short version: no retries, no rate limiting, no budget, no log of what ran. Adding those to the script is where it grows; with agentl they are already there.
- Speed: the scripts call the model in a plain loop, agentl with
concurrency: 8, so agentl finishes first (2x to 5x in the LLM cases here). A script with a thread pool would close that gap; the benchmarks measure it. - The sales report has no model calls, so the cache does not apply. What agentl adds there is the check before publishing and the versioned output, which the script also does by hand in fewer lines. Pick agentl for it if the rest of your pipelines are agentl; on its own, the script is fine.
What each version handles¶
| Script | agentl | |
|---|---|---|
| Same result | yes | yes |
| Re-run only pays for what changed | no | yes, per-record cache |
| Retries on 429 / 5xx from the model | no | yes |
| Calls in parallel | no (a loop) | concurrency |
| Stop at a spend limit | no | budget |
| Validated output shape from the model | by hand | output schema |
Run history, dashboard, agentl stats |
no | yes |
| Data check before delivery | by hand (sales report) | check steps |
The code¶
Support ticket triage¶
A CSV of tickets, one model call per ticket for category and urgency, an Excel file for the team.
"""Support ticket triage, by hand: read the CSV, ask the model about each ticket, save an Excel report."""
import pandas as pd
from cases.fake_llm import complete # stands in for client.messages.create(...) plus parsing the answer
def main(work: str) -> None:
df = pd.read_csv(f"{work}/data/tickets.csv")
answers = [complete("tickets", text) for text in df["text"]]
df["category"] = [a["category"] for a in answers]
df["urgent"] = [a["urgent"] for a in answers]
df.to_excel(f"{work}/out/triage.xlsx", index=False)
name: ticket-triage
llm: { provider: case-fake, model: fake, options: { task: tickets } }
steps:
- id: tickets
use: csv.read
with: { path: "${WORK}/data/tickets.csv" }
- id: triage
agent: >
Classify the support ticket. category is one of billing, bug, account, feature_request,
other. urgent is true only when something critical is blocked right now.
input: tickets
concurrency: 8
output: { category: str, urgent: bool }
- id: report
use: excel.write
input: triage
with: { path: "${WORK}/out/triage.xlsx" }
Invoice field extraction¶
Invoice texts, the model extracts tax id, total and due date, loaded into a database.
"""Invoice extraction, by hand: read each invoice text, ask the model for the fields, load SQLite."""
import glob
import sqlite3
from cases.fake_llm import complete # stands in for client.messages.create(...) plus parsing the answer
def main(work: str) -> None:
rows = []
for path in sorted(glob.glob(f"{work}/data/invoices/*.txt")):
with open(path, encoding="utf-8") as f:
fields = complete("invoices", f.read())
rows.append((path, fields["tax_id"], fields["total"], fields["due_date"]))
conn = sqlite3.connect(f"{work}/out/finance.db")
with conn:
conn.execute("DROP TABLE IF EXISTS invoices")
conn.execute("CREATE TABLE invoices (path, tax_id, total, due_date)")
conn.executemany("INSERT INTO invoices VALUES (?, ?, ?, ?)", rows)
conn.close()
name: invoice-extraction
llm: { provider: case-fake, model: fake, options: { task: invoices } }
steps:
- id: invoices
use: txt.read
with: { path: "${WORK}/data/invoices/*.txt" }
- id: fields
agent: >
Extract tax_id (the CNPJ as written), total (amount due as a number) and due_date
(YYYY-MM-DD) from the invoice.
input: invoices
concurrency: 8
output: { tax_id: str, total: float, due_date: str }
- id: rows
use: select
input: fields
with: { fields: [path, tax_id, total, due_date] }
- id: save
use: sqlite.write
input: rows
with: { db: "${WORK}/out/finance.db", table: invoices, mode: replace }
Web pages to summaries¶
HTML pages stripped to their main text, one summary per page, saved as JSON.
"""Web page summaries, by hand: strip each page to its main text, ask the model for a summary,
save JSON."""
import glob
import json
import trafilatura
from cases.fake_llm import complete # stands in for client.messages.create(...) plus parsing the answer
def main(work: str) -> None:
out = []
for path in sorted(glob.glob(f"{work}/data/pages/*.html")):
with open(path, encoding="utf-8") as f:
html = f.read()
text = trafilatura.extract(
html,
output_format="markdown",
include_tables=True,
include_links=True,
include_formatting=True,
favor_recall=True,
)
if not text:
continue
meta = trafilatura.extract_metadata(html)
out.append({"source": path, "title": meta.title if meta else "", **complete("summary", text.strip())})
with open(f"{work}/out/summaries.json", "w", encoding="utf-8") as f:
json.dump(out, f, ensure_ascii=False, indent=2)
name: web-summaries
llm: { provider: case-fake, model: fake, options: { task: summary } }
steps:
- id: pages
use: html.read
with: { path: "${WORK}/data/pages/*.html", format: markdown }
- id: summarized
agent: Summarize the page in one sentence.
input: pages
concurrency: 8
output: { summary: str }
- id: rows
use: select
input: summarized
with: { fields: [_source, title, summary], rename: { _source: source } }
- id: save
use: json.write
input: rows
with: { path: "${WORK}/out/summaries.json" }
Daily sales report to a shared folder¶
A messy daily export: drop blanks and repeated orders, check it, total by region, publish a dated Excel file with a latest copy and the last 7 versions. No LLM.
"""Daily sales report, by hand: clean the export, check it, total by region, save a dated Excel
file plus a `latest` copy in the shared folder, and keep the last 7 versions."""
import datetime as dt
import glob
import os
import shutil
import pandas as pd
def main(work: str) -> None:
df = pd.read_csv(f"{work}/data/sales.csv", dtype=str, keep_default_na=False)
df = df[df["amount"] != ""].drop_duplicates(subset="order_id")
if len(df) < 10 or (df["region"] == "").any():
raise ValueError(f"sales export looks wrong: {len(df)} rows")
df["amount"] = df["amount"].astype(float)
report = df.groupby("region", as_index=False)["amount"].sum().rename(columns={"amount": "total"})
report["total"] = report["total"].round(2)
path = f"{work}/out/sales_{dt.date.today().isoformat()}.xlsx"
report.to_excel(path, index=False)
shutil.copyfile(path, f"{work}/out/sales_latest.xlsx")
for old in sorted(glob.glob(f"{work}/out/sales_????-??-??.xlsx"))[:-7]:
os.remove(old)
name: daily-sales
steps:
- id: export
use: csv.read
with: { path: "${WORK}/data/sales.csv" }
- id: clean
use: filter
input: export
with: { not_empty: amount }
- id: unique
use: dedupe
input: clean
with: { on: order_id }
- id: verify
check: unique
expect: { rows: { min: 10 }, not_null: [region] }
- id: report
fn: cases.sales_report.steps:by_region
input: verify
- id: save
use: excel.write
input: report
with: { path: "${WORK}/out/sales_{date}.xlsx", latest: "${WORK}/out/sales_latest.xlsx", keep: 7 }