A Guatemalan's Guide to Drinking Coffee Properly

My mug

I grew up surrounded by coffee farms in Guatemala and somehow spent most of my adult life drinking whatever was closest, usually instant, usually while standing over the sink. If that sounds familiar, this is for you. No lectures, no gear worship, just why this coffee is actually worth the extra five minutes, how to pick a bag based on what you like instead of what a label tells you to like, and which machines are actually worth owning depending on how deep you want to go.

Why Guatemalan coffee earns the hype

Guatemala grows coffee on the slopes of active volcanoes, which is either the most metal origin story in the coffee world or just Tuesday for Guatemalan farmers. Volcanic soil is packed with minerals, the highlands stay cool even close to the equator, and slow growth at altitude gives the bean more time to build sugar and flavor before harvest. The regions genuinely taste different from each other too. Antigua leans chocolatey with a citrus edge, Huehuetenango turns bright and almost wine-like, Cobán goes softer and more floral thanks to all the rain. This is the actual, physical reason a bag from one region tastes nothing like a bag from another, not marketing copy dressed up to sound scientific.

Pick your coffee like you pick your music, based on taste, not trends

The single biggest mistake people make buying coffee is grabbing whatever bag has the nicest packaging. What actually matters is what you like in a cup, and that comes down to roast level.

If you like brightness, something with a little zing that wakes you up before the caffeine even kicks in, go lighter roast. Light roasts preserve more of the bean's natural acidity and fruitiness, they don't taste sour, they taste alive.

If you want something sweeter, rounder, and more familiar, medium roast is the safe, correct answer for most people, and it happens to be what most of the Huehuetenango bags below are roasted to. Nobody is judging you for liking the coffee equivalent of a comfortable couch.

Bags worth actually ordering

As I live in Guatemala, the brands usually available here are different from the ones on the international market, so i tried to find some of the brands that resemble the ones I prefer:

Fresh Roasted Coffee, Guatemala Huehuetenango, medium roast, single origin, kosher, sold in a 2-pack of 2 lb bags. Good option if you drink enough coffee that buying in bulk actually makes sense.

Guatemala Coffee, Huehuetenango, whole bean, medium roast, freshly roasted, 5 lb bag. Same region, bigger bag, for the household that goes through coffee like it's water.

Two Volcanoes Guatemalan Coffee, whole bean, 1 lb. Smaller bag, good if you want to try something new without committing to five pounds of a coffee you've never tasted.

The French way, easiest option...

A Veken French Press, 34oz, no plastic touching the coffee, thickened borosilicate glass, skips the paper filter entirely. That means more body, more oils, more of that heavier mouthfeel, which suits a medium roast Huehuetenango particularly well. Cleanup is slightly more annoying, that's the tradeoff, and it doubles as a cold brew pot if you're into that.

The Italian way, the most versatile one...

The Bialetti Moka Express, 12-cup is not technically espresso, don't let anyone tell you otherwise, but it's strong, concentrated, and genuinely delicious over ice or with steamed milk. The 12-cup size means you're not making this twice for a full household. It also makes a very satisfying gurgling noise right before it's ready, which is honestly half the appeal.

The Japanese way, V60 and actual patience

The Hario V60 Pour Over Starter Set, size 02 is the classic for a reason, cheap, simple, and it makes brighter beans genuinely sing. Pair it with the Hario Buono electric gooseneck kettle for actual control over your pour instead of just dumping water and hoping, and keep a stack of Hario V60 paper filters, size 02 on hand since running out mid-morning is its own special kind of tragedy. This method forces you to stand there for three minutes doing nothing but pouring water in circles, which is either meditative or mildly annoying depending on your morning.

Going full home barista

The Ninja Luxe Café Mini is the interesting middle ground, it does both espresso and drip in one machine, with a built-in burr grinder, a precision scale, and a manual steam wand, so you get real control without needing three separate appliances on your counter.

If you'd rather push one button and get a finished drink without thinking about grind size ever again, the Jura E4 in piano black is a fully automatic machine built for exactly that, bean to cup with none of the fuss. The Jura ENA 4 in Nordic white does the same job in a smaller footprint, good if counter space is the actual constraint rather than budget.

So which one do you actually need

None of them, technically. A bag of good Guatemalan coffee and a French press will outperform bad beans in a thousand dollar machine every single time. Buy the coffee that matches what you actually like drinking, and pick whichever method fits your morning, rushed, ritualistic, or somewhere in between. The volcano already did the hard part. You're just not allowed to ruin it on your end... Also, do not listen to the purist that says coffee should be black without sugar, is your money, you choose the way to drink it: sugar, cocoa, milk, cinnamon... go ahead and try new things!

This post contains affiliate links. As an Amazon Associate, studyyourdata.com earns from qualifying purchases at no extra cost to you.

Narrow Task Decomposition in AI Pipeline Agents

A nightly load into sales_fact fails. Ask an orchestrating agent to break down the diagnosis and it will usually stop at the first two subtasks that come to mind: check the logs, check the schema. That decomposition is narrow, and a fix built on it can miss the real cause. The problem is not the model, it is that the coordinator never questioned its own first list before delegating.

1. See what narrow decomposition misses

A first-pass breakdown of "why did sales_fact fail" often lands on log-analyzer and schema-inspector and stops there. Both are reasonable, and both are incomplete. Neither one checks whether the upstream extract delivered bad or late data, whether another job holds a lock on the same table, or which downstream dashboards are now stale because the load never completed. A fix based on two subtasks treats the visible half of the failure as the whole failure.

2. Add a self-critique step to the coordinator prompt

The fix is a coordinator prompt that forces a second pass before any delegation happens: generate an initial subtask list, ask what is missing, add subtasks to cover the gap, and only then hand work to subagents.

You are a coordinator diagnosing an ETL failure. When
decomposing the task:

1. Generate an initial list of subtasks.
2. Ask yourself: what systems, data sources, or failure
   modes are missing from that list?
3. Add subtasks to cover those gaps.
4. Only then begin delegating to subagents.

For an ETL failure specifically, consider:
- the failing job AND its upstream dependencies
- a structural cause (schema) AND a data cause (quality)
- this run AND whether it is a recurring pattern
- the technical root cause AND the downstream impact
  

Run the sales_fact example through that loop and the initial two-item list grows to four: log-analyzer and schema-inspector, plus a source-data-validator that checks the upstream extract for null spikes or row-count drops, and a downstream-impact-checker that lists which dashboards or scheduled jobs depend on the table. The self-critique step costs one extra generation and catches the two subtasks that a narrow first pass always skips.

3. Define each subtask as a subagent

With Claude's Agent SDK, each subtask becomes an AgentDefinition, with a description Claude matches against the task and a prompt scoped to that one job.

from claude_agent_sdk import AgentDefinition

log_analyzer = AgentDefinition(
    description="Diagnoses ETL failures from pipeline logs. "
                "Use when a nightly load job fails and the "
                "cause is unclear.",
    prompt="Find the failing step and report the exact "
           "error message and line number. Do not propose "
           "a fix.",
    tools=["Read", "Grep", "Bash"],
    model="haiku",
)

schema_inspector = AgentDefinition(
    description="Checks table structure for drift against "
                "the last known good schema.",
    prompt="Compare current columns, types, and "
           "constraints to the last successful load.",
    tools=["Read", "Bash"],
)

source_data_validator = AgentDefinition(
    description="Checks the upstream extract for the failed "
                "load. Use for row-count drops or null spikes.",
    prompt="Compare today's source row count and null "
           "rate to the 7-day average.",
    tools=["Read", "Bash"],
    model="haiku",
)

downstream_impact_checker = AgentDefinition(
    description="Lists dashboards and jobs that depend on "
                "the table that failed to load.",
    prompt="Find consumers of sales_fact and flag which "
           "are now stale.",
    tools=["Read", "Grep"],
)
  

AgentDefinition also takes a model field, so a subtask that is closer to pattern matching, like reading a log file or comparing two row counts, can run on a cheaper model while one that needs judgment, like assessing schema drift or downstream impact, stays on the default. That is the same idea covered in matching the model to the task, not the app, applied at the subagent level instead of a single orchestrator.

4. Wire the coordinator to call the agents

The coordinator prompt from step 2 becomes the system_prompt, and the four AgentDefinition objects from step 3 go into agents, keyed by the name Claude will use to call each one. Claude decides on its own which of the four to invoke, based on how the task matches each description.

from claude_agent_sdk import query, ClaudeAgentOptions

AGENTS = {
    "log-analyzer": log_analyzer,
    "schema-inspector": schema_inspector,
    "source-data-validator": source_data_validator,
    "downstream-impact-checker": downstream_impact_checker,
}

async for message in query(
    prompt="The sales_fact load failed last night, find out why.",
    options=ClaudeAgentOptions(
        system_prompt=COORDINATOR_PROMPT,
        # "Agent" auto-approves each subagent call
        allowed_tools=["Read", "Grep", "Bash", "Agent"],
        agents=AGENTS,
    ),
):
    for block in getattr(message, "content", None) or []:
        if getattr(block, "name", None) == "Agent":
            print("routed to:", block.input.get("subagent_type"))
  

Claude does not call a Python function directly, it emits a tool_use block named Agent, and block.input["subagent_type"] holds the name of the agent it picked, one of the four keys in AGENTS. Logging that value is how you confirm the coordinator actually routed to source-data-validator or downstream-impact-checker instead of quietly falling back to the two obvious subtasks. To force a specific agent instead of letting Claude choose, name it in the prompt: "use the downstream-impact-checker agent to list what depends on sales_fact."

5. The same routing in OpenAI's Agents SDK

OpenAI's Agents SDK expresses the same coordinator through handoffs. The triage agent's instructions carry the same four-step self-critique, and each of the four specialists is passed in its handoffs list, exposed to the model as a callable tool named transfer_to_<agent_name>. When Runner.run finishes, result.new_items contains a HandoffOutputItem for any handoff that fired, and its target_agent.name is the concrete signal, the OpenAI equivalent of reading subagent_type off Claude's Agent tool call.

6. One failure, traced end to end

Take one run instead of a list of what each agent can do. At 02:14, the nightly load into sales_fact fails, and the engineer on call prompts the coordinator with "the sales_fact load failed last night, find out why." The coordinator's first move is always the same: call log-analyzer, since nothing else can be scoped correctly before the actual error is known.

routed to: log-analyzer
  MERGE INTO sales_fact failed at 02:14:07 with:
  duplicate key value violates unique constraint
  "sales_fact_pkey" (order_id, sale_date)
  

A duplicate key on a merge has two plausible explanations: something changed about the table that now lets duplicates in, or the source data itself contains rows that collide on that key. Both are live possibilities under the self-critique step from earlier, so the coordinator calls schema-inspector and source-data-validator next. It does not call downstream-impact-checker yet, since nothing so far says anything about who reads this table.

routed to: schema-inspector
  no DDL changes on sales_fact in the last 14 days

routed to: source-data-validator
  source row count: 41,209 (7-day avg: 68,450, -40%)
  null rate on customer_id: 6.2% (7-day avg: 0.1%)
  

schema-inspector rules out a structural cause. source-data-validator explains the collision: a partial extract with a null spike on customer_id delivered rows that collapsed onto the same order_id and sale_date, tripping the unique constraint on merge. The coordinator stops here and never calls downstream-impact-checker, because the merge failed and rolled back, so nothing downstream ever read the bad batch. That agent earns its place in a run where a bad load succeeds silently instead of failing loudly, a different failure mode from this one.

Three of the four candidate agents ran, in two rounds instead of one batch, and the proposed fix follows directly from what they found: quarantine the extract that arrived overnight rather than retrying the load as is, and add a row-count and null-rate check ahead of the merge so a batch shaped like this one gets rejected before it reaches sales_fact, not after.

Before You Trust the Decomposition

Test the coordinator prompt against a case where you already know the full list of causes, and check whether its self-critique step actually surfaces the ones a naive first pass would miss. If it keeps landing on the same two or three subtasks regardless of the failure, the gap-checking questions in the prompt are too generic and need to name the specific systems in your pipeline.

Not every failure needs any of this. A connection timeout, a transient throttling error, a job that fails because another job it depends on is still running, these have deterministic fixes: retry with backoff, wait and requeue, alert and stop. Route the alert through a cheap classifier first, or a plain if-statement on the error code, and reserve the coordinator and its agents for failures that pattern-matching cannot already explain. Calling four agents to conclude "retry it" is not decomposition, it is waste.

Programmatic Tool Calling with the Claude SDK

Every morning somebody on the team runs the same check. Thirty tables in the staging schema, and the only thing anyone wants to know is which ones did not get loaded overnight. It is a five minute job, it is boring, and it is exactly the sort of thing you would hand to an agent. So you do. And then you watch it take thirty separate trips to the model, one table at a time, dragging every result set through the context window on the way. The answer was three table names. You paid for thirty round trips to get them.

The LLMCompiler paper put a name on this in 2024. Treat the tool calls like a compiler treats instructions: work out what depends on what, then run everything else at the same time. Two years later the idea is in the API, and it arrived in a shape nobody quite predicted. There is no plan format to parse. Claude just writes the Python.

1. One field changes everything

Add allowed_callers to a tool and switch on code execution. Your tool stops being something the model asks for one call at a time and becomes an async function sitting inside a sandbox, taking a dict, returning a string. From there, loops and conditionals and asyncio.gather come free. They are just Python.

import anthropic

client = anthropic.Anthropic()

TOOLS = [
    {"type": "code_execution_20260120",
     "name": "code_execution"},
    {
        "name": "run_sql",
        "description": ("Run a read-only query on the warehouse. "
                        "Returns a JSON list of row objects."),
        "input_schema": {
            "type": "object",
            "properties": {"sql": {"type": "string"}},
            "required": ["sql"],
        },
        "allowed_callers": ["code_execution_20260120"],
    },
]

resp = client.messages.create(
    model="claude-sonnet-5",
    max_tokens=4096,
    tools=TOOLS,
    messages=[{"role": "user", "content":
        "For every table in the stage schema, get the latest "
        "load timestamp and list only those older than 6 hours."}],
)
  

Sonnet 5 is the right call here. Plan, fan out, summarize, nothing that needs the expensive model, which is the same reasoning behind matching the model to the task rather than the app. The floor is the code execution version, not the tier, so any Sonnet from 4.5 forward will do.

2. Somebody else writes the loop

Back comes a server_tool_use block with the generated script inside it. The first time you read one of these it is a little uncanny, because it is roughly the code you would have written yourself on a Tuesday afternoon.

# written in the container, not by you
import json, asyncio

tables = json.loads(await run_sql({"sql": LIST_SQL}))
rows = await asyncio.gather(*[
    run_sql({"sql": f"select max(load_ts) ts from {t['name']}"})
    for t in tables
])

stale = [(t["name"], r) for t, r in zip(tables, rows)
         if hours_since(r) > 6]
print(stale)
  

One query to find the tables, then all thirty timestamps at once, then a filter. Three names get printed. Three names is all the model ever sees. The other twenty seven results lived and died in the sandbox, which is the whole point.

3. You still answer the phone

Your database is not in that container, so every time the script calls the tool, everything stops and waits for you. The response shows up with stop_reason of tool_use, a container id, and one block per pending call. Run the query, hand back the rows, the script picks up mid-line.

while resp.stop_reason == "tool_use":
    results = [
        {"type": "tool_result",
         "tool_use_id": b.id,
         "content": execute(b.name, b.input)}
        for b in resp.content if b.type == "tool_use"
    ]
    # tool_result blocks only on this turn, nothing else
    messages += [{"role": "assistant", "content": resp.content},
                 {"role": "user", "content": results}]
    resp = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=4096,
        container=resp.container.id,  # required while calls pend
        tools=TOOLS,
        messages=messages,
    )
  

Two things will trip you on the first attempt. That user turn takes tool results and absolutely nothing else, not even a stray line of text. And the full tools array has to ride along on every continuation, code execution included, otherwise the paused script has nothing to wake up into.

4. Is it worth it

Depends entirely on the shape of your work, and the documentation is refreshingly blunt about it. On a 75-tool agent benchmark, about 38 percent fewer billed input tokens with accuracy unchanged. Across production traffic declaring 10 to 49 tools, savings of 20 to 40 percent. And on a benchmark where each turn made one or two sequential calls, it cost 8 percent more. Fan-out wins. Short chains lose.

Before you point this at production

Keep an eye on the clock. A pending call gives up after about four minutes and throws a timeout inside the running script, and idle containers get reclaimed after about five, so a query that takes ten minutes needs chunking, not optimism. Availability is uneven at the moment: Claude API, Claude Platform on AWS and Microsoft Foundry yes, Bedrock and Google Cloud not yet. And the obvious one, which is worth saying anyway. Every query in that fan-out is generated text. Give the tool a read-only connection and let the database enforce it, because allowed_callers is a hint to Claude about how to use the tool, not a wall around it.

Generate Embeddings in SQL with Aurora and Bedrock

Most embedding pipelines on AWS have the same shape: a job reads rows out of the database, calls Amazon Bedrock, and writes the vectors back. That is a second service to deploy, monitor, and pay for, and it exists only to move text a few hundred milliseconds away and bring numbers back. Aurora PostgreSQL can do the whole thing in one statement. The Aurora machine learning extension calls Bedrock from inside SQL, and pgvector stores and searches the result in the same table.

1. Wire the cluster to Bedrock

Aurora ML needs an IAM role carrying bedrock:InvokeModel, attached to the cluster under the Bedrock feature in the Connectivity and security tab. Model access has to be requested in Bedrock separately. After that, two extensions in the target database. The Aurora machine learning documentation walks through the console steps.

-- aws_ml pulls in aws_commons and creates the aws_bedrock schema
CREATE EXTENSION IF NOT EXISTS aws_ml CASCADE;
CREATE EXTENSION IF NOT EXISTS vector;

SELECT extname, extversion FROM pg_extension
WHERE extname IN ('aws_ml', 'vector');
  

You want aws_ml at 2.0, which is the version that adds the Bedrock functions, and vector at 0.8.0. Version 2.0 of the extension ships with Aurora PostgreSQL 16.1, 15.5, and 14.10 and higher. pgvector 0.8.0 needs 17.4, 16.8, 15.12, 14.17, or 13.20 and higher.

2. Call Bedrock from a SELECT

The function aws_bedrock.invoke_model_get_embeddings takes the model payload as JSON and pulls the vector out by key. Titan Text Embeddings V2 returns 1,024 dimensions by default and puts them under embedding. Aurora hands the result back as float8[], so cast it.

SELECT aws_bedrock.invoke_model_get_embeddings(
    model_id     := 'amazon.titan-embed-text-v2:0',
    content_type := 'application/json',
    json_key     := 'embedding',
    model_input  := '{"inputText": "tempdb contention on a heap"}'
)::vector AS embedding;
  

3. Materialize the vector, do not compute it at query time

Every invocation is billed, so the embedding belongs in a stored column that you fill once. Backfill in bounded batches rather than one enormous UPDATE, because each row is a separate synchronous call and a single failure rolls the whole statement back.

CREATE TABLE documents (
    id        bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    tenant_id text NOT NULL,
    content   text NOT NULL,
    embedding vector(1024)
);

-- backfill 500 rows at a time
UPDATE documents d
SET embedding = aws_bedrock.invoke_model_get_embeddings(
        'amazon.titan-embed-text-v2:0',
        'application/json',
        'embedding',
        json_build_object('inputText', d.content)::text
    )::vector
WHERE d.id IN (
    SELECT id FROM documents
    WHERE embedding IS NULL ORDER BY id LIMIT 500
);
  

4. Index it, then filter without losing rows

Titan V2 normalizes its output by default, which means inner product ranks identically to cosine while skipping the norm division on every comparison. Pair the <#> operator with vector_ip_ops. The m and ef_construction values below are the AWS starting points from the pgvector production guidance.

CREATE INDEX documents_embedding_ip_idx
ON documents USING hnsw (embedding vector_ip_ops)
WITH (m = 16, ef_construction = 128);

BEGIN;
SET LOCAL hnsw.iterative_scan = relaxed_order;
SET LOCAL hnsw.ef_search = 100;

SELECT id, content
FROM documents
WHERE tenant_id = 'acme'
ORDER BY embedding <#> '<query embedding>'
LIMIT 10;
COMMIT;
  

Those two settings are the reason to be on pgvector 0.8.0. Before it, a WHERE clause combined with a vector search regularly returned fewer rows than the LIMIT asked for, because the filter ran after the index had already handed back its top candidates. Iterative scans keep pulling from the index until the query is satisfied, and relaxed_order is the mode to reach for in production. The ef_search default of 40 is usually too low once real traffic arrives. Worth confirming with EXPLAIN that the planner actually picks the HNSW index here and has not quietly fallen back to a sequential scan, which is the kind of thing PlanTrace makes obvious at a glance.

Before you point this at a large table

The appeal here is the missing infrastructure. No Lambda, no queue, no separate deployment, and the vector never leaves the transaction that produced it. The limit is that Bedrock calls through Aurora ML are one row at a time: the UDFs support neither batching nor parallel execution, unlike the Comprehend and SageMaker functions in the same extension. That makes this a good fit for scheduled backfills through pg_cron and for low-volume writes, and a poor fit for an INSERT trigger on a high-throughput table. Size the instance so the HNSW graph stays in memory, watch BufferCacheHitRatio, and treat a drop there as the first sign the index has outgrown the box.

Match the Model to the Task, Not the App: An Orchestrator-Worker Pattern in Python

Most apps that call an LLM send every request to the same model. That works until the bill arrives. A classification job that returns one word costs the same as a multi-step reasoning job, because you picked one model at design time and never revisited it. The fix is an orchestrator-worker setup: a small central router reads the incoming task, decides which tier of model it needs, and delegates to a worker. The router sits at the hub, each model tier is a spoke, and no worker talks to another. This post builds that in Python using both the Anthropic and OpenAI SDKs, so you can mix providers in the same pipeline.

1. Define the tiers before writing any code

Three tiers cover almost everything. Fast handles extraction, classification, tagging, and format conversion. Balanced handles summarization, SQL generation, and short analysis. Deep handles multi-step reasoning, code review, and anything where a wrong answer is expensive. Map each tier to a model from each provider so you can fail over between them.

# tiers.py
TIERS = {
    "fast":     {"anthropic": "claude-haiku-4-5", "openai": "gpt-5.6-luna"},
    "balanced": {"anthropic": "claude-sonnet-5",  "openai": "gpt-5.6-terra"},
    "deep":     {"anthropic": "claude-opus-4-8",  "openai": "gpt-5.6-sol"},
}
  

2. The orchestrator itself belongs on the cheap tier

The orchestrator runs on every request, before any real work happens. Whatever it costs gets multiplied across your entire traffic, and its latency lands in front of every response. Putting a deep model there means paying premium prices to decide what to do. Routing is short classification against a closed label set, which is precisely what the small models handle well.

The risk is that router errors are asymmetric. Sending a fast task to the deep tier wastes money and you see it in the bill. Sending a deep task to the fast tier produces a plausible wrong answer that flows downstream unnoticed. Three controls keep that in check: a closed label set instead of free-form output, a hard cap on max_tokens so the router cannot drift into explanation, and the validator gate covered further down, which catches the cases where the cheap tier was not enough.

import anthropic

client = anthropic.Anthropic()

ROUTER_PROMPT = """Classify the task into exactly one tier.
fast: extraction, classification, tagging, formatting
balanced: summarization, query generation, short analysis
deep: multi-step reasoning, code review, planning
Answer with the tier name only."""

def route(task_text):
    msg = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=8,
        system=ROUTER_PROMPT,
        messages=[{"role": "user", "content": task_text}],
    )
    tier = msg.content[0].text.strip().lower()
    return tier if tier in TIERS else "balanced"
  

The fallback on the last line matters. Routers misfire, and an unrecognized label should land on a safe middle tier rather than raise. And when your app already knows the task type because the user clicked a specific button or the pipeline stage is fixed, skip the router entirely. A dictionary lookup costs nothing and never misclassifies.

3. Wrap each worker behind one function signature

The two SDKs differ in shape. Anthropic uses messages.create with a separate system argument and a required max_tokens. OpenAI uses responses.create with an input list and a reasoning effort setting. Normalize both to take a prompt and return a string, and the orchestrator stops caring which vendor answered. The Anthropic Python SDK and the OpenAI model catalog document the current identifiers for each family.

import anthropic
from openai import OpenAI

anthropic_client = anthropic.Anthropic()
openai_client = OpenAI()

def call_anthropic(model, system, prompt, max_tokens=1024):
    msg = anthropic_client.messages.create(
        model=model,
        max_tokens=max_tokens,
        system=system,
        messages=[{"role": "user", "content": prompt}],
    )
    return msg.content[0].text

def call_openai(model, system, prompt, effort="low"):
    resp = openai_client.responses.create(
        model=model,
        instructions=system,
        input=prompt,
        reasoning={"effort": effort},
    )
    return resp.output_text
  

4. Dispatch, and allow one escalation

The orchestrator is now small. Pick the tier, pick the provider, run the call. Add a single escalation path: if the fast tier returns something your validator rejects, retry once on the next tier up. That ladder is a model cascade, and it should stay short. One retry, not a loop. Unbounded escalation turns a cost optimization into a cost multiplier.

LADDER = ["fast", "balanced", "deep"]

def run_task(prompt, system, validator, provider="anthropic"):
    tier = route(prompt)
    idx = LADDER.index(tier)

    for attempt_tier in LADDER[idx:idx + 2]:
        model = TIERS[attempt_tier][provider]
        if provider == "anthropic":
            out = call_anthropic(model, system, prompt)
        else:
            effort = "low" if attempt_tier == "fast" else "high"
            out = call_openai(model, system, prompt, effort)

        if validator(out):
            return {"tier": attempt_tier, "model": model, "output": out}

    raise ValueError("no tier produced a valid result")
  

The validator is the piece people skip, and it is what makes the whole design safe. For extraction, check that the JSON parses and the required keys exist. For SQL generation, run EXPLAIN against the target database. Without a validator you have no signal that the cheap tier was good enough, and the routing decision becomes a guess.

5. Log the tier, not just the answer

Store the tier, model, token counts, and validator result for every call. After a week you will see which task types actually escalate and which ones you over-provisioned from the start. That log is the only honest way to tune the routing prompt, and it usually reveals that a large share of traffic never needed anything above the fast tier.

6. A real pipeline: invoice ingestion

Consider a service that receives scanned invoices and has to produce clean rows in a database. Four distinct jobs happen per document, and only one of them is hard. Classifying the document type is a label from a short list. Field extraction is bounded and schema-driven. Totals reconciliation is arithmetic plus judgment about rounding and tax lines. Resolving an ambiguous line item against a product catalog is where reasoning actually pays for itself. Hardcode the first three, and let the orchestrator decide only on the last one.

# pipeline.py

STAGE_TIERS = {
    "classify_doc":   "fast",
    "extract_fields": "fast",
    "reconcile":      "balanced",
}

def run_stage(stage, system, prompt, validator, provider="anthropic"):
    # known stage, no router call needed
    tier = STAGE_TIERS[stage]
    model = TIERS[tier][provider]
    ...  # same call plus escalation logic as run_task

def process_invoice(ocr_text):
    doc_type = run_stage(
        "classify_doc",
        system="Answer with one of: invoice, receipt, credit_note, other.",
        prompt=ocr_text[:2000],
        validator=is_known_doc_type,
    )
    if doc_type != "invoice":
        return route_to_manual_queue(ocr_text, doc_type)

    fields = run_stage(
        "extract_fields",
        system=EXTRACTION_SCHEMA_PROMPT,
        prompt=ocr_text,
        validator=parses_as_invoice_json,
    )

    if not totals_match(fields):
        fields = run_stage(
            "reconcile",
            system=RECONCILE_PROMPT,
            prompt=describe_mismatch(fields),
            validator=totals_match,
        )

    # only unresolved items reach the orchestrator and the expensive tiers
    for item in unmatched_items(fields):
        item["sku"] = run_task(
            prompt=catalog_match_prompt(item),
            system=CATALOG_PROMPT,
            validator=sku_exists,
        )["output"]

    return persist(fields)
  

Notice what the structure buys you. Every invoice pays for two fast calls. The balanced tier only runs when the arithmetic disagrees, and the deep tier only touches line items the catalog lookup could not resolve on its own. On a clean batch, most documents never leave the cheap tier, and the expensive models are reserved for the handful of cases that genuinely earn them.

Wrapping Up

Orchestrator-worker does not have to mean an elaborate framework. A tier table, a cheap orchestrator, one wrapper per SDK, and a validator gate get you most of the benefit in under a hundred lines. Both providers expose a low-cost tier that handles far more than people expect once the prompt is narrow enough.

One practical note before you ship: pin the versioned model identifiers in production rather than the family aliases. Aliases move when a vendor releases a new generation, and a routing table that silently changes tiers under you is a debugging session nobody enjoys.

Moving On-Prem PostgreSQL to the Cloud: Picking the Right Path for Big Tables

Every PostgreSQL migration eventually hits the same fork in the road. The database is small enough to dump and restore in a maintenance window, or it isn't. Once you cross into terabyte territory, or the business can't afford more than a few minutes of downtime, the simple approach stops being an option and you have to think harder. Here are the three paths that actually work, and when each one makes sense.

1. Dump and restore, the reliable baseline

For anything under a couple hundred GB, pg_dump piped into pg_restore is still the cleanest way to move a database. It carries the full schema, data, indexes, and sequences, and on decent hardware a mid-size database finishes in well under an hour. The one flag people forget is parallelism.

# dump in the directory format so restore can run in parallel
pg_dump -Fd -j 8 -f /dump/appdb sourcedb

# restore with the same worker count on the target
pg_restore -Fd -j 8 -d appdb /dump/appdb
  

The catch shows up with one dominant table. Parallelism here is per-table, so a single 2 TB table becomes one long serial stream while the other seven workers sit idle. That is the moment dump and restore starts feeling slow, and the reason the next two options exist.

2. Native logical replication for a live cutover

When downtime is the real constraint, PostgreSQL's built-in logical replication lets the target keep pace with the source while the application stays online. You publish on the old server, subscribe on the new one, and Postgres copies the initial snapshot and then streams every change until you decide to switch over.

-- on the source (wal_level must be 'logical')
CREATE PUBLICATION migration_pub FOR ALL TABLES;

-- on the target
CREATE SUBSCRIPTION migration_sub
  CONNECTION 'host=old-db port=5432 dbname=appdb user=repl password=...'
  PUBLICATION migration_pub;
  

Two things save you real pain on big datasets. First, tables without a primary key need a REPLICA IDENTITY FULL or a unique index, otherwise updates and deletes won't replicate. Second, that same large table problem returns. Logical replication syncs each table with its own worker, so your biggest table still copies as a single stream. A common workaround is to pre-load the huge table with pg_restore, then attach it to the subscription with copy_data = false so replication only handles the incremental changes.

3. Managed DMS when you want the plumbing handled

Both AWS and Google Cloud offer a managed migration service that wraps logical replication for PostgreSQL, including from on-premises sources. Google's Database Migration Service supports CDC via pglogical and can open multiple parallel subscriptions across tables during the initial load and the ongoing CDC phase. AWS DMS does the same using logical replication slots with the test_decoding or pglogical plugin. Both handle slot management and give you a controlled cutover step.

The shared limitation is worth knowing before you commit: for tables without a primary key, both services replicate the initial snapshot and INSERT operations only. UPDATE and DELETE changes during CDC are dropped and need to be handled separately. Beyond that, parallelism is still table-level on both platforms, so the single dominant large table remains the bottleneck either way.

How to actually choose

Small database and a maintenance window you can spare? Dump and restore, done. Need the source online with minimal downtime and comfortable managing slots yourself? Native logical replication. Want the same low-downtime result without babysitting the replication setup? Reach for the managed DMS on your target cloud. In all three cases the single dominant table is what decides your timeline, so measure it first and plan to pre-seed it on its own rather than hoping a migration tool will parallelize what the engine simply won't.

Setting Up a Mac for Data Engineering and AI Work

If you work with data pipelines, SQL, notebooks, or machine learning models, a Mac with Apple Silicon is genuinely one of the best machines you can have as a daily driver. The unified memory architecture means your CPU and GPU share the same memory pool, which matters a lot when you are running Docker containers, a local Postgres instance, Jupyter, and an AWS CLI session all at once without the machine breaking a sweat. This post covers which model to pick and how to get the whole stack running from scratch. Some of the product links below are affiliate links, meaning I may earn a small commission if you purchase through them, at no extra cost to you.

1. Picking the right model

The current MacBook line runs entirely on M5 chips. For most data engineers doing cloud-first work — Redshift, Glue, dbt, Airflow — the MacBook Air M5 with 16GB is more than enough. If you run heavier local workloads like training models, spinning up multiple containers, or working with large DataFrames in memory, the MacBook Pro 14" M5 Pro with 24GB is the right step up. The Pro has active cooling, which matters when you push it hard for extended periods. The 36GB M5 Max is only worth it if local deep learning training is a regular part of your day, otherwise you are paying for headroom you will rarely use.

2. First thing: Homebrew and Xcode tools

Everything else in this setup depends on Homebrew, the package manager for macOS. Before installing it, you need Apple's command line developer tools, which also gives you Git.

# Install Xcode command line tools
xcode-select --install

# Install Homebrew
/bin/bash -c "$(curl -fsSL https://raw.githubusercontent.com/Homebrew/install/HEAD/install.sh)"

# Add Homebrew to your PATH (Apple Silicon path)
echo 'eval "$(/opt/homebrew/bin/brew shellenv)"' >> ~/.zprofile
eval "$(/opt/homebrew/bin/brew shellenv)"

3. Python environment management with uv

Forget installing Python directly or relying on conda for everything. The current best practice on Apple Silicon is uv, a fast Python package and project manager that handles Python versions and virtual environments without polluting your global install. It is significantly faster than pip and plays well with Jupyter.

# Install uv
brew install uv

# Create a new project environment
mkdir my-de-project && cd my-de-project
uv init

# Add packages (same idea as pip install)
uv add pandas sqlalchemy boto3 jupyterlab scikit-learn

# Launch Jupyter inside the project environment
uv run --with jupyter jupyter lab

4. Core tools via Homebrew

These cover the daily stack for data engineering: database clients, cloud CLIs, container runtime, and a better terminal.

# AWS CLI and tools
brew install awscli

# PostgreSQL client (psql without the full server)
brew install libpq
brew link --force libpq

# Docker via OrbStack (lighter than Docker Desktop on Apple Silicon)
brew install orbstack

# Git, jq, and wget
brew install git jq wget

# VS Code
brew install --cask visual-studio-code

5. PyTorch with Metal GPU acceleration

Apple Silicon GPUs run PyTorch through the Metal Performance Shaders (MPS) backend, which gives you real GPU acceleration for model training without needing CUDA. It works natively on macOS 12.3 or later and PyTorch picks it up automatically.

# Install PyTorch (Apple Silicon native build)
uv add torch torchvision torchaudio

# Verify MPS is available in Python
import torch
print(torch.backends.mps.is_available())  # should print True

# Move a model to the MPS device
device = "mps" if torch.backends.mps.is_available() else "cpu"
model = model.to(device)

One thing worth knowing: MPS does not support every PyTorch operation yet. If you hit an unsupported op, set the environment variable PYTORCH_ENABLE_MPS_FALLBACK=1 and PyTorch will silently fall back to CPU for that specific operation while keeping everything else on the GPU.

6. One limitation to consider

Macs do not support CUDA, and that is a real constraint if your production training runs on NVIDIA GPUs or if your team uses CUDA-specific libraries. The practical answer most engineers land on is using the Mac for development, prototyping, and running notebooks, then pushing actual training jobs to AWS SageMaker, Google Colab, or a cloud GPU instance. The unified memory architecture makes the Mac excellent for loading large quantized models locally — a 7B or 13B parameter model fits comfortably in 24GB unified memory — but for serious fine-tuning or multi-GPU training, cloud is still the right call.

A solid machine that gets out of your way

The real reason data engineers gravitate toward Macs is not any single spec — it is the combination of a Unix shell that works the way you expect, excellent battery life, and hardware that handles a full data stack locally without fan noise or thermal throttling on everyday tasks. Getting to a productive environment takes less than an hour with Homebrew, uv, and OrbStack in place. After that, you have PostgreSQL, Docker, AWS CLI, Jupyter, and PyTorch with GPU acceleration all running natively on Apple Silicon, which is a genuinely capable local setup for most data and AI workflows.