Tag: ai

The Slack message came in at 9:14 on a Tuesday: “why is yesterday’s Snowflake bill $4,200 over budget and who approved it.” Nobody had approved anything. A dashboard refresh job had been silently retrying every five minutes since Saturday after a schema change broke one of its filters, and by the time anyone noticed, it had burned three days of a warehouse running at full tilt for no reason. The fix took ten minutes once someone looked. The problem was that someone had to look, and by Tuesday the money was already gone.

That’s the gap autonomous agents are actually good for closing — not “AI does your job,” but “something is watching at 3 a.m. so a $4,200 mistake gets caught in twenty minutes instead of three days.” And it’s worth being honest about where the industry actually is on this: Gartner expects more than 40% of agentic AI projects to be canceled by the end of 2027, and its stated reasons are almost never about the model being too weak — they’re escalating costs, unclear value, and inadequate risk controls. The pattern I’ve seen up close matches that: teams skip scoping, skip guardrails, and skip deployment, then wonder why the “agent” never left someone’s laptop. This is a practical walkthrough of a small, real agent — one that watches Snowflake spend and flags anomalies — built the way that actually survives contact with production. If you want the higher-level argument for why this kind of role shift is happening at all, I’ve made that case in why automation, not AI, is what’s really changing this job.

TL;DR

• → Write down the agent’s one job, what success looks like, and what it’s never allowed to do — before opening an editor. Skipping this is the single biggest cause of stalled agent projects.
• → LangGraph has become the closest thing to a 2026 production default for stateful agents — multiple independent sources put it at 30–90M+ monthly downloads with Klarna, Uber, and LinkedIn running it live — while Microsoft has moved AutoGen into maintenance mode.
• → The core of any agent is a loop: the model reasons, calls a tool if needed, reads the result, and repeats — and that loop needs a hard step cap or it can run away and burn your API budget.
• → Memory (checkpointing) is what lets an agent handle a follow-up like “now show me last week too” without starting from zero.
• → Guardrails — input validation, a recursion limit, and a bounded retry — are what separate a demo from something safe to leave running unattended.
• → Deployment is not optional polish: wrapping the agent in a small API and a container is what turns “it worked on my machine” into something a dashboard, a Slack bot, or another service can actually call.

Step 1: Decide what it does — and what it’s never allowed to do

Before any code, write three sentences: the one job, what success looks like, and the hard boundary it can’t cross without a human. For a cost-anomaly agent, that’s:

The job: on a schedule, pull the last 24 hours of Snowflake warehouse spend, compare it against the trailing 14-day baseline, and flag any warehouse running more than 2x its typical cost.
Success looks like: a short written alert naming the warehouse, the dollar delta, and a plausible cause pulled from the query history — posted to Slack.
The hard boundary: it can query cost and query-history tables freely, but it never suspends a warehouse, kills a query, or changes a resource monitor without a human approving first.

That boundary is doing real work. An agent that can only look and report is low-risk to leave running unattended; one that can act on what it finds needs a different level of trust entirely. Skipping this step is exactly the kind of ambiguity Gartner points to when it names unclear scope and inadequate risk controls as the top reasons agentic projects get killed before they ever prove their value.

Step 2: Pick the framework — and don’t pick AutoGen

Two choices matter here: which model reasons, and which framework runs the loop of thinking, acting, and checking the result. For the model, any current frontier model with reliable tool-calling works; the code below uses Claude. For the framework, here’s where 2026 has actually settled:

LangGraph models an agent as nodes and edges in a graph with built-in checkpointing, so a failed step can resume instead of restarting from scratch. Independent industry write-ups through mid-2026 consistently cite it running in production at companies like Klarna, Uber, and LinkedIn, with monthly download figures that vary by source but are unambiguously in the tens of millions. CrewAI gets a prototype running faster — often under 20 lines — and has real traction of its own, but teams commonly outgrow its coordination model once a workflow gets non-trivial and migrate to LangGraph. AutoGen, once a default for multi-agent conversation patterns, is worth naming only as a warning: Microsoft has shifted it into maintenance mode in favor of a unified Microsoft Agent Framework, so it’s not where you want to start something new.

For a cost-anomaly agent that runs unattended on a schedule, the checkpointing is the deciding factor — a Snowflake query that times out shouldn’t mean the whole run starts over — so this build uses LangGraph.

Step 3: Set up the project

# Create and enter the project folder
mkdir cost-anomaly-agent && cd cost-anomaly-agent

# Isolate dependencies in a virtual environment
python3 -m venv venv
source venv/bin/activate   # Windows: venv\Scripts\activate

# langgraph        - orchestrates the reasoning loop
# langchain-anthropic - connects LangGraph to Claude
# snowflake-connector-python - lets the agent query Snowflake directly
# python-dotenv    - loads credentials from .env, never hardcoded
pip install langgraph langchain-anthropic snowflake-connector-python python-dotenv

Create a .env file for credentials, and make sure it’s in .gitignore alongside venv/ before you write a single line of agent logic:

# .env — never commit this file
ANTHROPIC_API_KEY=your-anthropic-key-here
SNOWFLAKE_ACCOUNT=your-account-locator
SNOWFLAKE_USER=your-service-user
SNOWFLAKE_PASSWORD=your-password
SLACK_WEBHOOK_URL=your-slack-webhook

Keep agent.py (the agent logic) separate from app.py (the API wrapper), the same separation of concerns that makes any pipeline easier to reason about — the same instinct behind structuring a dbt project into clean layers.

Step 4: Build the core reasoning loop

This is the heart of it: the model reads the task, decides whether it needs a tool, calls it, reads the result, and decides what to do next.

The loop that powers every tool-using agent. Without a hard cap on step count, a stuck tool call can spin this loop indefinitely — and each spin is a billed model call.

# agent.py
from dotenv import load_dotenv
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent
import snowflake.connector
import os

load_dotenv()

# Low temperature: we want a consistent read of the numbers, not creative variation
model = ChatAnthropic(model="claude-sonnet-4-6", temperature=0.1, max_tokens=1200)

@tool
def get_warehouse_spend(lookback_days: int = 14) -> str:
    """Returns daily credit usage per warehouse for the given lookback window,
    most recent day first. Use this to compare yesterday against the baseline."""
    conn = snowflake.connector.connect(
        account=os.environ["SNOWFLAKE_ACCOUNT"],
        user=os.environ["SNOWFLAKE_USER"],
        password=os.environ["SNOWFLAKE_PASSWORD"],
    )
    cur = conn.cursor()
    cur.execute("""
        SELECT warehouse_name, start_time::date AS usage_date,
               SUM(credits_used) AS credits
        FROM snowflake.account_usage.warehouse_metering_history
        WHERE start_time >= DATEADD('day', -%s, CURRENT_DATE())
        GROUP BY warehouse_name, usage_date
        ORDER BY usage_date DESC
    """, (lookback_days,))
    rows = cur.fetchall()
    conn.close()
    return "\n".join(f"{r[0]}, {r[1]}, {r[2]:.2f} credits" for r in rows)

agent = create_react_agent(model, tools=[get_warehouse_spend])

def run_triage() -> str:
    """Asks the agent to review recent spend and flag anomalies."""
    result = agent.invoke({
        "messages": [
            ("system",
             "You monitor Snowflake warehouse spend. Compare yesterday's "
             "credit usage per warehouse against its trailing 14-day average. "
             "Flag any warehouse running more than 2x its baseline. For each "
             "flag, state the warehouse, the percentage over baseline, and "
             "the dollar impact at $3/credit. If nothing is anomalous, say so."),
            ("user", "Review the last 24 hours of warehouse spend."),
        ]
    })
    return result["messages"][-1].content

if __name__ == "__main__":
    print(run_triage())

What’s doing the real work here: get_warehouse_spend is a plain Python function turned into a callable tool by the @tool decorator — and its docstring isn’t documentation, it’s the description the model reads to decide when to call it. create_react_agent is LangGraph’s shortcut for the classic reason-act-observe loop (ReAct) without hand-writing graph nodes. The system prompt is what turns a generic tool-calling agent into a specific one: it names the exact comparison, the exact threshold, and the exact output shape, which is the difference between a useful alert and a vague paragraph.

Step 5: Add memory and a second tool

Right now the agent forgets everything between runs, which is fine for a scheduled job but breaks the moment someone asks a natural follow-up like “what caused that spike on the ETL_WH warehouse?” Add a second tool that pulls query history, and LangGraph’s built-in checkpointer to persist state across a conversation thread:

# agent.py (additions)
from langgraph.checkpoint.memory import MemorySaver

@tool
def get_top_queries(warehouse_name: str, hours: int = 24) -> str:
    """Returns the most expensive queries run on a specific warehouse
    in the given window, to help explain a cost spike."""
    conn = snowflake.connector.connect(
        account=os.environ["SNOWFLAKE_ACCOUNT"],
        user=os.environ["SNOWFLAKE_USER"],
        password=os.environ["SNOWFLAKE_PASSWORD"],
    )
    cur = conn.cursor()
    cur.execute("""
        SELECT query_text, user_name, total_elapsed_time / 1000 AS seconds
        FROM snowflake.account_usage.query_history
        WHERE warehouse_name = %s
          AND start_time >= DATEADD('hour', -%s, CURRENT_TIMESTAMP())
        ORDER BY total_elapsed_time DESC
        LIMIT 5
    """, (warehouse_name, hours))
    rows = cur.fetchall()
    conn.close()
    return "\n".join(f"{r[1]}: {r[0][:80]}... ({r[2]:.0f}s)" for r in rows)

memory = MemorySaver()
agent = create_react_agent(
    model,
    tools=[get_warehouse_spend, get_top_queries],
    checkpointer=memory,
)

def run_triage(thread_id: str = "daily-triage") -> str:
    config = {"configurable": {"thread_id": thread_id}}
    result = agent.invoke(
        {"messages": [("user", "Review the last 24 hours of warehouse spend.")]},
        config=config,
    )
    return result["messages"][-1].content

MemorySaver is what lets a follow-up question in the same thread_id reuse everything the agent already found, instead of re-querying from zero — the same reason warehouse result caching saves you from redoing work that hasn’t changed.

Step 6: Guardrails — the step most tutorials skip

An agent that only reads cost data is low risk. But even a read-only agent needs bounds around runaway loops and bad state, and this is exactly the gap Gartner’s cancellation numbers point back to — not model quality, but the absence of operational discipline around it.

# agent.py (guardrails)
import time

MAX_RETRIES = 2
RECURSION_LIMIT = 12   # caps reasoning/tool-call steps in a single run

def run_triage_safely(thread_id: str = "daily-triage") -> str:
    config = {
        "configurable": {"thread_id": thread_id},
        "recursion_limit": RECURSION_LIMIT,
    }
    for attempt in range(1, MAX_RETRIES + 1):
        try:
            result = agent.invoke(
                {"messages": [("user", "Review the last 24 hours of warehouse spend.")]},
                config=config,
            )
            return result["messages"][-1].content
        except Exception as e:
            if attempt == MAX_RETRIES:
                return f"Triage failed after {MAX_RETRIES} attempts: {e}"
            time.sleep(3)

recursion_limit is the single most important line in this block. Without it, a Snowflake connection hiccup or a confusing result can send the agent into extra reasoning steps that quietly rack up model calls — the agentic equivalent of the retrying dashboard job that started this article. The bounded retry handles the ordinary case of a transient network blip without masking a real failure.

Step 7: Ship it somewhere real

A script that runs when you remember to run it isn’t monitoring anything. Wrap it in a small API and a scheduled container so it runs whether or not you’re watching.

Guardrails make the agent safe to run unattended; the container and API are what let something else — a scheduler, a Slack bot, a dashboard — actually trigger it.

# app.py
from fastapi import FastAPI
from agent import run_triage_safely

app = FastAPI(title="Cost Anomaly Agent")

@app.get("/health")
def health():
    return {"status": "ok"}

@app.post("/triage")
def triage():
    return {"report": run_triage_safely()}

# Dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["sh", "-c", "uvicorn app:app --host 0.0.0.0 --port ${PORT:-8000}"]

Push it to a container host, point a scheduler (a cron trigger, an Airflow task, or the host’s own scheduled jobs) at POST /triage every morning, and pipe the response into Slack. If you’re already orchestrating pipelines, wiring this into the same system you use for everything else is straightforward — the pattern is no different from triggering any other scheduled job against Snowflake. And because this agent only reads account usage data, giving it credentials safely is worth doing properly — see the broader pattern in giving agents metadata access without opening security holes.

The gotchas nobody warns you about

A read-only agent still needs a recursion limit. “It can’t do damage, it only reads” is not the same as “it can’t run forever.” Every reasoning step is a billed model call.

The system prompt is the actual product. The framework, the tools, and the code are plumbing. The threshold, the comparison window, and the exact output format live in the prompt — get that vague and the agent produces vague alerts no matter how solid the code is.

Account usage views lag. Snowflake’s ACCOUNT_USAGE schema can trail real-time by up to a few hours. An agent triaging “the last hour” against that view will occasionally miss the very spike it was built to catch — know the latency of your data source before you trust the silence.

A demo that works once is not a deployed agent. The gap between “it worked in my terminal” and “it’s live and something else can call it” is exactly steps 6 and 7 — and it’s the gap most abandoned agent projects never cross.

Framework churn is real; the concepts aren’t. AutoGen’s shift to maintenance mode is a reminder that frameworks move fast. The scoping, loop, memory, and guardrail concepts in this article transfer to whatever framework wins next.

The one principle

An autonomous agent is not a smarter script — it’s a script with a loop, a memory, and a leash, and the leash is what makes it safe to leave running. The model reasoning is the easy 20%; the scoping, the guardrails, and the deployment are the 80% that decide whether this becomes something that catches a $4,200 mistake at 3 a.m., or one more repo nobody ever pushed past a terminal window.

Related reading: It’s not AI you should worry about — it’s automation · Giving agents metadata access safely · Tools vs subagents: don’t over-build · MCP explained in 3 levels · Gartner: 40% of agentic AI projects canceled by 2027 · LangGraph production case studies