Skip to content

Week 11 — Production: cost, latency, reliability, AWS

Why this week: once a demo works, interviewers ask three questions. What does one check cost? How long does it take? What happens when the model provider goes down? This week you answer all three with numbers from your own service, and you show you can run it on a second cloud.

By Friday you have: every check traced in Langfuse, with tokens, cost and time; cost and time stored with each check in Postgres; before-and-after numbers for prompt caching and for routing easy messages down a cheaper path; a fallback model, a time limit and a general-advice answer when Claude is unavailable; /health endpoints and one-line JSON logs; the API running on AWS; a one-page Google Cloud to AWS map; and docs/cost-report.md.

Main online course: none this week; the docs below are the course. If you want one long read, Reducing latency covers most of the week's ideas in one place.

Concepts

Trace and span. A trace is the record of one check from start to finish. It's made of spans, one per step: the check as a whole, each call_claude() inside it, each tool call. Each span records how long it took and, for model calls, the tokens and cost. Laid out on a timeline, a trace shows where the time and money went.

p50 and p95 latency. Latency is how long a check takes. The average hides the slow ones, so you report percentiles: p50 (the median: half of checks are faster) and p95 (95% are faster; the slowest one in twenty is slower). Users remember the p95.

Cost per check. Add up every model call one check makes: the agent's steps, the reviewer, the explanation. That total, not the price per token, is the number that tells you what a thousand users would cost.

Prompt caching. Each agent step sends the same long system prompt again. With caching, the API keeps the processed start of the prompt for five minutes; the next request that starts with exactly the same text reads it from the cache instead, for a tenth of the normal input price, and a little faster. It works only on an identical prefix, and only above a minimum length (512 tokens on current models); below that, it silently does nothing.

Routing. Not every message needs the full agent. "See you at dinner tonight?" has no link to expand and no number to look up. Routing sends messages like that down a cheaper path: one model call instead of six. It's only a saving if your eval says recall didn't drop.

Timeout, fallback and graceful degradation. A timeout is how long you'll wait for one call. A fallback is what you try next: here, a second model when the first is overloaded. Graceful degradation is what the user gets when everything fails: not an error page, but "we couldn't check this; here's what to do with any suspicious message". A scam checker that crashes at the busiest moment teaches people not to rely on it.

Health checks. A health check is an address the platform calls every few seconds to ask "are you alive?". Liveness (/health) answers instantly and only says the process is up. Readiness (/health/ready) also checks the database. Neither calls Claude, because health checks run constantly and would cost money.

Structured logs. A log line like check 42 took 3100 ms is easy to read and hard to search. One JSON object per line, such as {"message": "check done", "check_id": 42, "duration_ms": 3100}, lets Cloud Logging and AWS CloudWatch filter and add up fields for you. Never log the message itself; log its id and the numbers.

The same job on another cloud. Every cloud has the same building blocks under different names: somewhere to run a container, somewhere to keep images, somewhere for secrets, a permissions system, logs. Once you've deployed on one, the second is mostly a matter of finding the names. Friday's map is that list.

Monday — See every check

Learn (75 min):

Build:

  1. Create a free account on Langfuse Cloud. The Hobby plan has no card requirement and enough monthly usage for this course. Langfuse offers several regions, including EU, US and Japan (https://jp.cloud.langfuse.com); Japan is the closest to Singapore. Create a project, then API keys, and add three lines to .env:

    LANGFUSE_PUBLIC_KEY=pk-lf-...
LANGFUSE_SECRET_KEY=sk-lf-...
LANGFUSE_BASE_URL=https://jp.cloud.langfuse.com

    Traces leave Singapore and sit on Langfuse's servers. That's acceptable only because of what you'll send: stripped text and numbers, never the raw message.

  2. Add the SDK:

    uv add langfuse

  3. Replace src/scam_checker/llm.py. The name, arguments and return type of call_claude() still don't change; ClaudeResult gains three fields with defaults, so old code that builds or reads it still works. The body is now split: call_claude() decides what to try, _call_once() makes one call.

    """Call Claude through the Anthropic API, using your API key.

Weeks 1-8 called the Claude Code CLI here. Week 9 swapped the body of
call_claude() for the Anthropic SDK; its name, arguments and ClaudeResult
are unchanged, so nothing that calls it needs to change.
"""

from __future__ import annotations

import json
import logging
import time
from collections.abc import Iterator
from contextlib import contextmanager
from contextvars import ContextVar
from dataclasses import dataclass
from functools import cache
from typing import Any

import anthropic
from langfuse import get_client

logger = logging.getLogger(__name__)

# The short names used since week 1, mapped to API model IDs.
# Check the models overview page when you start: IDs change as models are released.
MODELS = {
    "haiku": "claude-haiku-5-5",
    "sonnet": "claude-sonnet-5-5",
    "opus": "claude-opus-5-5",
}

# US dollars per million tokens: (input, output, cache write, cache read).
# From the pricing page; cache writes are the 5-minute kind.
# Haiku 5.5's price is for prompts up to 100,000 tokens; yours are far shorter.
PRICES = {
    "claude-haiku-5-5": (0.10, 0.50, 0.125, 0.01),
    "claude-sonnet-5-5": (2.00, 10.00, 2.50, 0.10),
    "claude-opus-5-5": (4.00, 20.00, 5.00, 0.20),
}

# If a model is overloaded or down, try this one once instead.
FALLBACK_MODEL = {"claude-haiku-5-5": "claude-sonnet-5-5"}

CACHE_SYSTEM_PROMPT = True  # set False to measure the "before" numbers

MAX_TOKENS = 4096  # room for the model's thinking plus a short JSON answer


class ClaudeError(RuntimeError):
    """The API call failed, timed out, or gave no usable answer."""

    def __init__(self, message: str, *, retryable: bool = False) -> None:
        super().__init__(message)
        self.retryable = retryable  # worth trying again, or another model


@dataclass
class ClaudeResult:
    text: str
    data: dict[str, Any] | None  # parsed JSON, only when you pass a schema
    input_tokens: int
    output_tokens: int
    cost_usd: float  # estimated from PRICES; your Console bill is the real number
    duration_ms: int
    model: str = ""
    cache_read_tokens: int = 0
    cache_write_tokens: int = 0


# A list that collects every call made while track_usage() is active.
_calls: ContextVar[list[ClaudeResult] | None] = ContextVar("calls", default=None)


@contextmanager
def track_usage() -> Iterator[list[ClaudeResult]]:
    """Collect every call_claude() result inside the with-block, e.g. one check."""
    calls: list[ClaudeResult] = []
    token = _calls.set(calls)
    try:
        yield calls
    finally:
        _calls.reset(token)


@cache
def _client() -> anthropic.Anthropic:
    # Reads ANTHROPIC_API_KEY from the environment. max_retries=0 because
    # check_message() already retries; two retry layers multiply the waiting.
    return anthropic.Anthropic(max_retries=0)


def estimate_cost(
    model_id: str,
    input_tokens: int,
    output_tokens: int,
    cache_write: int = 0,
    cache_read: int = 0,
) -> float:
    p_in, p_out, p_write, p_read = PRICES.get(model_id, (0.0, 0.0, 0.0, 0.0))
    dollars = (
        input_tokens * p_in
        + output_tokens * p_out
        + cache_write * p_write
        + cache_read * p_read
    )
    return dollars / 1_000_000


def call_claude(
    prompt: str,
    *,
    system: str | None = None,
    schema: dict[str, Any] | None = None,
    model: str = "haiku",
    timeout: float = 120,
) -> ClaudeResult:
    """Send one prompt to Claude and return the reply.

    If the model is overloaded or unreachable, try its fallback model once.
    """
    model_id = MODELS.get(model, model)  # a full model ID also works
    try:
        result = _call_once(model_id, prompt, system, schema, timeout)
    except ClaudeError as exc:
        backup = FALLBACK_MODEL.get(model_id)
        if not exc.retryable or backup is None:
            raise
        logger.warning("%s failed (%s); trying %s", model_id, exc, backup)
        result = _call_once(backup, prompt, system, schema, timeout)

    calls = _calls.get()
    if calls is not None:
        calls.append(result)
    return result


def _call_once(
    model_id: str,
    prompt: str,
    system: str | None,
    schema: dict[str, Any] | None,
    timeout: float,
) -> ClaudeResult:
    output_config: Any = anthropic.omit
    if schema:
        # transform_schema adds what the API insists on, such as
        # "additionalProperties": false, to a Pydantic-made schema.
        output_config = {
            "format": {
                "type": "json_schema",
                "schema": anthropic.transform_schema(schema),
            }
        }
    # Mark the system prompt for caching. It is identical on every call, so
    # later calls within 5 minutes read it from the cache at a tenth of the price.
    system_blocks: Any = system or anthropic.omit
    if system and CACHE_SYSTEM_PROMPT:
        system_blocks = [
            {"type": "text", "text": system, "cache_control": {"type": "ephemeral"}}
        ]

    langfuse = get_client()
    start = time.perf_counter()
    with langfuse.start_as_current_observation(
        as_type="generation", name="call_claude", model=model_id
    ) as generation:
        try:
            response = _client().messages.create(
                model=model_id,
                max_tokens=MAX_TOKENS,
                system=system_blocks,
                messages=[{"role": "user", "content": prompt}],
                output_config=output_config,
                timeout=timeout,
            )
        except anthropic.APITimeoutError as exc:  # check before APIConnectionError
            raise ClaudeError(
                f"no reply after {timeout} seconds", retryable=True
            ) from exc
        except anthropic.APIConnectionError as exc:
            raise ClaudeError(
                f"could not reach the API: {exc}", retryable=True
            ) from exc
        except anthropic.APIStatusError as exc:
            # 429 rate limit, 500s and 529 overloaded are worth another try;
            # 400 (bad request, or your own spend limit) and 401 (bad key) are not.
            retryable = exc.status_code == 429 or exc.status_code >= 500
            raise ClaudeError(
                f"API error {exc.status_code}: {exc.message}", retryable=retryable
            ) from exc
        duration_ms = int((time.perf_counter() - start) * 1000)

        usage = response.usage
        cache_read = usage.cache_read_input_tokens or 0
        cache_write = usage.cache_creation_input_tokens or 0
        cost = estimate_cost(
            model_id, usage.input_tokens, usage.output_tokens, cache_write, cache_read
        )
        # Token counts and cost only: no prompt or reply text, so no personal data.
        generation.update(
            usage_details={
                "input": usage.input_tokens,
                "output": usage.output_tokens,
                "cache_read_input_tokens": cache_read,
            },
            cost_details={"total": cost},
        )

    if response.stop_reason == "refusal":
        raise ClaudeError("the model declined to answer")
    if response.stop_reason == "max_tokens":
        raise ClaudeError(f"the reply was cut off at {MAX_TOKENS} tokens")

    # The reply can start with thinking blocks; the answer is in the text blocks.
    text = "".join(block.text for block in response.content if block.type == "text")
    data = None
    if schema:
        try:
            data = json.loads(text)
        except json.JSONDecodeError as exc:
            raise ClaudeError(f"reply was not valid JSON: {text[:200]}") from exc

    return ClaudeResult(
        text=text,
        data=data,
        input_tokens=usage.input_tokens,
        output_tokens=usage.output_tokens,
        cost_usd=cost,
        duration_ms=duration_ms,
        model=model_id,
        cache_read_tokens=cache_read,
        cache_write_tokens=cache_write,
    )


def stream_claude(
    prompt: str,
    *,
    system: str | None = None,
    model: str = "haiku",
    timeout: float = 60,
) -> Iterator[str]:
    """Yield the reply in small pieces of text, as the model writes it."""
    try:
        with _client().messages.stream(
            model=MODELS.get(model, model),
            max_tokens=MAX_TOKENS,
            system=system or anthropic.omit,
            messages=[{"role": "user", "content": prompt}],
            timeout=timeout,
        ) as stream:
            yield from stream.text_stream
    except anthropic.APIError as exc:  # covers timeouts, network and status errors
        raise ClaudeError(f"streaming failed: {exc}") from exc


if __name__ == "__main__":
    reply = call_claude("Say hello in five words.")
    print(reply.text)
    print(f"{reply.input_tokens} in, {reply.output_tokens} out, {reply.duration_ms} ms")
    print(f"about US${reply.cost_usd:.6f}")

    New today: track_usage() collects every call made during one check, wherever in your agent it happens, without changing the agent's code. A context variable (ContextVar) is a value that belongs to the current piece of work, so two checks running at once don't mix up their lists. PRICES now includes cache prices, and each call is recorded in Langfuse as a "generation" with token counts and cost but no text.

    Two parts are for later in the week. For today's "before" numbers, change these two lines:

    FALLBACK_MODEL: dict[str, str] = {}
CACHE_SYSTEM_PROMPT = False

    You'll set them back on Tuesday and Wednesday. (The dict[str, str] tells mypy what an empty dictionary will hold.)

  4. One check, timed, costed and traced, in a new src/scam_checker/service.py. Both endpoints and tomorrow's measuring script use it:

    """Run one check with a time limit, cost tracking and a safe fallback answer."""

import contextvars
import logging
import time
from collections.abc import Callable
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field

from langfuse import get_client, observe

from scam_checker.agent import AgentRun
from scam_checker.checker import FALLBACK
from scam_checker.llm import ClaudeError, ClaudeResult, track_usage
from scam_checker.pii import strip_pii

logger = logging.getLogger(__name__)

DEADLINE_SECONDS = 45.0  # after this, the user gets general advice instead of a wait
_pool = ThreadPoolExecutor(max_workers=8)

Agent = Callable[[str], AgentRun]


@dataclass
class Outcome:
    run: AgentRun
    calls: list[ClaudeResult] = field(default_factory=list)
    duration_ms: int = 0
    degraded: bool = False  # True when we gave the general-advice fallback

    @property
    def cost_usd(self) -> float:
        return sum(call.cost_usd for call in self.calls)

    @property
    def input_tokens(self) -> int:
        return sum(
            call.input_tokens + call.cache_read_tokens + call.cache_write_tokens
            for call in self.calls
        )

    @property
    def output_tokens(self) -> int:
        return sum(call.output_tokens for call in self.calls)


@observe(name="check", capture_input=False, capture_output=False)
def run_check(agent: Agent, text: str) -> Outcome:
    start = time.perf_counter()
    degraded = False
    with track_usage() as calls:
        # Run the agent in a worker thread so we can stop waiting at the deadline.
        # copy_context() carries track_usage() and the trace into that thread.
        future = _pool.submit(contextvars.copy_context().run, agent, text)
        try:
            run = future.result(timeout=DEADLINE_SECONDS)
        except (ClaudeError, TimeoutError) as exc:
            logger.warning("check degraded: %s", type(exc).__name__)
            run, degraded = AgentRun(result=FALLBACK, steps=[]), True
    duration_ms = int((time.perf_counter() - start) * 1000)
    outcome = Outcome(
        run=run, calls=list(calls), duration_ms=duration_ms, degraded=degraded
    )

    # What the trace shows: the message without personal data, and the numbers.
    get_client().update_current_span(
        input=strip_pii(text),
        output=run.result.verdict,
        metadata={
            "steps": len(run.steps),
            "cost_usd": outcome.cost_usd,
            "degraded": degraded,
        },
    )
    return outcome

    @observe makes the whole check one trace, with every call_claude() inside it nested underneath. capture_input=False stops Langfuse recording the raw message; update_current_span then sends the stripped one. The worker thread and DEADLINE_SECONDS are Wednesday's subject; today it's enough that run_check returns an Outcome with the calls and their cost.

  5. JSON logs, src/scam_checker/logs.py:

    """One JSON object per log line, so Cloud Logging and CloudWatch can search it."""

import json
import logging
import sys
from datetime import UTC, datetime


class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        entry = {
            "time": datetime.fromtimestamp(record.created, UTC).isoformat(),
            "severity": record.levelname,  # the field name Cloud Logging looks for
            "logger": record.name,
            "message": record.getMessage(),
        }
        entry.update(
            getattr(record, "fields", {})
        )  # logger.info(..., extra={"fields": {...}})
        if record.exc_info:
            entry["error"] = self.formatException(record.exc_info)
        return json.dumps(entry)


def configure_logging(level: int = logging.INFO) -> None:
    handler = logging.StreamHandler(sys.stdout)
    handler.setFormatter(JsonFormatter())
    logging.basicConfig(level=level, handlers=[handler], force=True)
    for name in (
        "uvicorn",
        "uvicorn.error",
        "uvicorn.access",
    ):  # the web server's own logs
        logging.getLogger(name).handlers = [handler]
        logging.getLogger(name).propagate = False
    for name in ("httpx2", "presidio-analyzer"):  # chatty libraries: warnings only
        logging.getLogger(name).setLevel(logging.WARNING)

    The severity field is the one Cloud Logging reads to colour a line as info, warning or error. The last loop quietens two chatty libraries: Presidio logs every recogniser it loads.

  6. Store cost and time with each check. Add Float and false to the sqlalchemy imports in src/scam_checker/api/db.py, and four columns at the end of CheckRecord:

        # Added in week 11. server_default fills in the rows that already exist.
    input_tokens: Mapped[int] = mapped_column(server_default="0")
    output_tokens: Mapped[int] = mapped_column(server_default="0")
    cost_usd: Mapped[float] = mapped_column(Float, server_default="0")
    degraded: Mapped[bool] = mapped_column(server_default=false())

    Then a second migration. server_default is what lets Alembic add a required column to a table that already has rows: the old rows get 0 and false.

    uv run --env-file .env alembic revision --autogenerate -m "add cost columns"
uv run ruff check --fix alembic && uv run ruff format alembic
uv run --env-file .env alembic upgrade head
DATABASE_URL='postgresql://...your Neon string...' uv run alembic upgrade head

    The last line updates Neon by hand this once. From now on, your deploy workflow applies new migrations itself, before each deploy.

  7. Add degraded to both response models in src/scam_checker/api/schemas.py. In CheckResponse:

        check_id: int
    degraded: bool = False  # True when this is the general-advice fallback

    And the same degraded: bool = False line at the end of CheckPanel.

  8. The new src/scam_checker/api/main.py. The endpoints now call run_check() instead of the agent directly, save_check() stores the numbers and writes one JSON log line, and two health endpoints appear (tomorrow and Wednesday explain the rest):

    """The checker as a web API.

Run locally with:  uv run fastapi dev src/scam_checker/api/main.py
"""

import logging
import os
from collections.abc import Callable, Iterator
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.sse import EventSourceResponse, ServerSentEvent
from sqlalchemy import text as sql
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import Session

from scam_checker.agent import run_agent
from scam_checker.api.db import CheckRecord, get_session
from scam_checker.api.schemas import CheckPanel, CheckRequest, CheckResponse
from scam_checker.llm import ClaudeError, stream_claude
from scam_checker.logs import configure_logging
from scam_checker.models import CheckResult
from scam_checker.pii import strip_pii
from scam_checker.retrieval import search_patterns
from scam_checker.service import Agent, Outcome, run_check

configure_logging()
logger = logging.getLogger(__name__)

app = FastAPI(title="Scam checker API", version="0.3.0")

# Which web pages may call this API from a browser. Set ALLOWED_ORIGINS on
# Cloud Run to your front end's address, e.g. https://<you>.github.io
app.add_middleware(
    CORSMiddleware,
    allow_origins=os.environ.get("ALLOWED_ORIGINS", "http://localhost:5173").split(","),
    allow_methods=["GET", "POST"],
    allow_headers=["Content-Type"],
)

Explainer = Callable[..., Iterator[str]]

EXPLAIN_SYSTEM = """You explain a scam checker's verdict to an ordinary person in Singapore.
Write two or three short sentences in plain words: why the message looks the way it does.
Do not repeat the advice. Do not include phone numbers, links or names from the message."""


def get_agent() -> Agent:
    """The agent to use. A function, so tests can swap in a fake one."""
    return run_agent


def get_explainer() -> Explainer:
    """The streaming explanation. Also swappable in tests."""
    return stream_claude


def save_check(session: Session, text: str, outcome: Outcome) -> CheckRecord:
    result = outcome.run.result
    record = CheckRecord(
        message=strip_pii(text),
        verdict=result.verdict,
        red_flags=[strip_pii(flag) for flag in result.red_flags],
        advice=strip_pii(result.advice),
        steps=len(outcome.run.steps),
        duration_ms=outcome.duration_ms,
        input_tokens=outcome.input_tokens,
        output_tokens=outcome.output_tokens,
        cost_usd=outcome.cost_usd,
        degraded=outcome.degraded,
    )
    session.add(record)
    session.commit()
    # One JSON line per check: numbers only, never the message itself.
    fields = {
        "check_id": record.id,
        "verdict": record.verdict,
        "steps": record.steps,
        "duration_ms": record.duration_ms,
        "cost_usd": round(record.cost_usd, 6),
        "degraded": record.degraded,
    }
    logger.info("check done", extra={"fields": fields})
    return record


def plain_label(result: CheckResult) -> str:
    """A confidence label a non-technical reader understands."""
    if result.verdict == "scam":
        return "Very likely a scam" if len(result.red_flags) >= 2 else "Likely a scam"
    if result.verdict == "not_scam":
        return "Probably genuine"
    return "Hard to tell"


def top_pattern(text: str) -> str | None:
    """The closest known scam pattern from week 4, if any."""
    matches = search_patterns(text, k=1)
    return matches[0].name if matches else None  # Match.name from week 4


@app.get("/health")
def health() -> dict[str, str]:
    """Liveness: the process is up and answering. Cheap, so it can be called often."""
    return {"status": "ok"}


@app.get("/health/ready")
def ready(session: Annotated[Session, Depends(get_session)]) -> dict[str, str]:
    """Readiness: the database answers too. Never calls the model, which costs money."""
    try:
        session.execute(sql("SELECT 1"))
    except SQLAlchemyError as exc:
        logger.error("database not ready: %s", type(exc).__name__)
        raise HTTPException(503, "database unavailable")
    return {"status": "ready"}


@app.post("/check", response_model=CheckResponse)
def check(
    request: CheckRequest,
    session: Annotated[Session, Depends(get_session)],
    agent: Annotated[Agent, Depends(get_agent)],
) -> CheckResponse:
    outcome = run_check(agent, request.text)
    record = save_check(session, request.text, outcome)
    return CheckResponse(
        check_id=record.id, degraded=outcome.degraded, **outcome.run.result.model_dump()
    )


@app.post("/check/stream", response_class=EventSourceResponse)
def check_stream(
    request: CheckRequest,
    session: Annotated[Session, Depends(get_session)],
    agent: Annotated[Agent, Depends(get_agent)],
    explain: Annotated[Explainer, Depends(get_explainer)],
) -> Iterator[ServerSentEvent]:
    """Send the answer in stages: status, result, explanation tokens, done."""
    yield ServerSentEvent(event="status", data="Checking the message...")
    outcome = run_check(agent, request.text)
    result = outcome.run.result

    pattern = top_pattern(request.text) if result.verdict != "not_scam" else None
    panel = CheckPanel(
        **result.model_dump(),
        label=plain_label(result),
        pattern=pattern,
        degraded=outcome.degraded,
    )
    yield ServerSentEvent(event="result", data=panel)

    if not outcome.degraded:  # if the model is down, don't wait on it a second time
        prompt = (
            f"Message:\n{request.text}\n\nVerdict: {result.verdict}\n"
            f"Warning signs: {'; '.join(result.red_flags) or 'none'}"
        )
        try:
            for piece in explain(prompt, system=EXPLAIN_SYSTEM):
                yield ServerSentEvent(event="token", data=piece)
        except ClaudeError as exc:  # the verdict has already arrived; carry on
            logger.warning("explanation failed: %s", exc)

    record = save_check(session, request.text, outcome)
    yield ServerSentEvent(event="done", data={"check_id": record.id})

    The 503 from week 9 is gone: when Claude fails, run_check() now returns the general-advice answer with degraded set, and the user always gets something useful.

  9. Update the two tests that expected the old behaviour. In tests/test_api.py, replace test_llm_failure_gives_503 with:

    def test_llm_failure_gives_general_advice(client):
    app.dependency_overrides[get_agent] = lambda: broken_agent
    response = client.post("/check", json={"text": "hello"})
    assert response.status_code == 200
    body = response.json()
    assert body["degraded"] is True
    assert body["verdict"] == "unsure"

    In tests/test_stream.py, replace test_stream_reports_a_busy_checker with:

    def test_stream_gives_general_advice_when_the_model_is_down(client):
    app.dependency_overrides[get_agent] = lambda: broken_agent
    events = read_events(client.post("/check/stream", json={"text": "hi"}).text)
    assert dict(events)["result"]["degraded"] is True
    assert "token" not in [name for name, _ in events]
    assert events[-1][0] == "done"

    And add two tests for the new pieces to tests/test_llm.py:

    def test_track_usage_collects_every_call(fake_api):
    fake_api(lambda request: httpx2.Response(200, json=api_message("hello")))
    with llm.track_usage() as calls:
        llm.call_claude("one")
        llm.call_claude("two")
    assert len(calls) == 2
    assert sum(c.input_tokens for c in calls) == 200


def test_cache_reads_are_priced_lower():
    full = llm.estimate_cost("claude-haiku-5-5", 1000, 0)
    cached = llm.estimate_cost("claude-haiku-5-5", 0, 0, cache_read=1000)
    assert cached == pytest.approx(full / 10)

  10. Run the API with your keys and check three messages through /docs:

    uv run --env-file .env fastapi dev src/scam_checker/api/main.py

    In Langfuse, open Tracing: each check is a trace called check, with the stripped message as input and a call_claude generation for each model call, carrying tokens and cost. In the terminal, each check ends with a JSON line like {"severity": "INFO", "message": "check done", "check_id": 12, "cost_usd": 0.000412, ...}.

  11. Your own numbers, straight from Postgres. Save this as scripts/report.sql; you'll run it against Neon on Friday:

    SELECT
  count(*) AS checks,
  round(avg(cost_usd)::numeric, 6) AS avg_cost_usd,
  round(sum(cost_usd)::numeric, 4) AS total_cost_usd,
  percentile_cont(0.5) WITHIN GROUP (ORDER BY duration_ms) AS p50_ms,
  percentile_cont(0.95) WITHIN GROUP (ORDER BY duration_ms) AS p95_ms,
  sum(CASE WHEN degraded THEN 1 ELSE 0 END) AS degraded,
  sum(CASE WHEN steps = 0 THEN 1 ELSE 0 END) AS quick_path
FROM checks
WHERE created_at > now() - interval '7 days';

    docker compose exec -T db psql -U scam -d scam < scripts/report.sql

Done when: a check shows up in Langfuse as one trace with its model calls nested inside, with no raw message anywhere in it; the checks table has cost and token columns filled in for new rows; and pytest passes.

Stuck?
  • Authentication error: Langfuse client initialized without public_key: the Langfuse lines aren't in the environment. Run with --env-file .env. In tests this warning is expected and harmless: tracing is simply off.
  • Context error: No active span in current context: the same cause. With no keys, @observe creates no span, so there's nothing to update.
  • No traces appear: Langfuse sends them in the background a few seconds later. Wait, refresh, and check LANGFUSE_BASE_URL matches the region your project is in.
  • Every call costs 0: the model ID isn't in PRICES. Print reply.model and add it.
  • The migration fails with column "input_tokens" contains null values: server_default is missing from the new column.
  • test_structured_reply_is_parsed fails: you set CACHE_SYSTEM_PROMPT = True too early. Leave it False until Tuesday.

Tuesday — Cheaper and faster, with numbers

Learn (60 min):

Build:

  1. A script that measures cost, time and quality on your labelled set, scripts/cost_latency.py. It assumes data/labelled100.csv has text and is_scam columns; adjust the two names if your week 5 file differs.

    """Cost, speed and quality of the checker on the labelled set, for the cost report.

Usage: uv run --env-file .env python scripts/cost_latency.py MODE [N]
MODE is one of: nocache (the "before"), cached, routed (cached plus routing).
"""

import statistics
import sys

import pandas as pd

from scam_checker import llm
from scam_checker.agent import run_agent
from scam_checker.metrics import precision_recall
from scam_checker.routing import routed_agent
from scam_checker.service import run_check

mode = sys.argv[1]
limit = int(sys.argv[2]) if len(sys.argv) > 2 else 100
llm.CACHE_SYSTEM_PROMPT = mode != "nocache"
agent = routed_agent if mode == "routed" else run_agent

df = pd.read_csv("data/labelled100.csv").head(limit)
outcomes = [run_check(agent, text) for text in df["text"]]

flagged = [o.run.result.verdict == "scam" for o in outcomes]
precision, recall = precision_recall(df["is_scam"].tolist(), flagged)
costs = [o.cost_usd for o in outcomes]
seconds = sorted(o.duration_ms / 1000 for o in outcomes)
cache_reads = sum(c.cache_read_tokens for o in outcomes for c in o.calls)
all_input = sum(o.input_tokens for o in outcomes)

print(f"{mode}: {len(df)} messages")
print(f"precision {precision:.2f}  recall {recall:.2f}")
print(
    f"cost per check: mean US${statistics.mean(costs):.5f}, total US${sum(costs):.4f}"
)
print(
    f"latency: p50 {seconds[len(seconds) // 2]:.1f}s, p95 {seconds[int(len(seconds) * 0.95)]:.1f}s"
)
print(f"input tokens read from cache: {cache_reads / max(all_input, 1):.0%}")
print(f"quick path used: {sum(len(o.run.steps) == 0 for o in outcomes)} of {len(df)}")
print(f"degraded answers: {sum(o.degraded for o in outcomes)}")

  2. The "before" numbers. With CACHE_SYSTEM_PROMPT = False, run the full agent on all 100 messages. It takes a while and costs a few US cents; note the total from the output.

    uv run --env-file .env python scripts/cost_latency.py nocache

    Copy all the output into a new file, docs/cost-report.md, under a heading "Before". Then look at one trace in Langfuse and write down which step takes longest.

  3. Caching. Set CACHE_SYSTEM_PROMPT = True in llm.py. Look at what it switches on in _call_once:

        # Mark the system prompt for caching. It is identical on every call, so
    # later calls within 5 minutes read it from the cache at a tenth of the price.
    system_blocks: Any = system or anthropic.omit
    if system and CACHE_SYSTEM_PROMPT:
        system_blocks = [
            {"type": "text", "text": system, "cache_control": {"type": "ephemeral"}}
        ]

    Update the system-prompt assertion in test_structured_reply_is_parsed, since the system prompt is now sent as a block:

        assert sent["system"][0]["text"] == "Be careful"
    assert sent["system"][0]["cache_control"] == {"type": "ephemeral"}

    Then measure again:

    uv run --env-file .env python scripts/cost_latency.py cached

    The "input tokens read from cache" line tells you whether it worked. If it says 0%, your system prompt is under 512 tokens, or something in it changes between calls, such as a timestamp. Caching helps most when the agent's instructions and tool descriptions are long and identical on every step.

  4. Routing. src/scam_checker/routing.py:

    """Send easy messages down a cheaper path: one model call instead of the agent."""

import re

from scam_checker.agent import AgentRun, run_agent
from scam_checker.checker import check_message

LINK = re.compile(
    r"https?://|www\.|\b[\w-]+\.(?:com|sg|net|org|ly|xyz|top)\b", re.IGNORECASE
)
PHONE = re.compile(r"(?:\+65[\s-]?)?\b[3689]\d{3}[\s-]?\d{4}\b")
MONEY = re.compile(r"[$£]\s?\d|\b\d+\s?(?:sgd|usd|dollars)\b", re.IGNORECASE)


def is_easy(text: str) -> bool:
    """Nothing for the agent's tools to look up: no links, phone numbers or money."""
    return not (LINK.search(text) or PHONE.search(text) or MONEY.search(text))


def routed_agent(text: str) -> AgentRun:
    if is_easy(text):
        return AgentRun(result=check_message(text), steps=[])  # 0 steps = quick path
    return run_agent(text)

    In main.py, swap the agent the API uses: replace from scam_checker.agent import run_agent with from scam_checker.routing import routed_agent, and make get_agent() return routed_agent. Then:

    uv run --env-file .env python scripts/cost_latency.py routed

  5. Your turn: write three tests for is_easy in tests/test_routing.py. A friendly message with no link, number or money is easy; a message with a link isn't; a message with a Singapore mobile number isn't. Then add one test for a case you think the rules get wrong.

    Stuck? Three tests 
    from scam_checker.routing import is_easy


def test_friendly_message_is_easy():
    assert is_easy("See you at dinner tonight?")


def test_link_is_not_easy():
    assert not is_easy("Your parcel is held, pay at singpost-redelivery.com")


def test_singapore_mobile_number_is_not_easy():
    assert not is_easy("Hi mum, this is my new number 8123 4567")

  6. Decide, with the numbers in front of you. Add a table to docs/cost-report.md: one row each for nocache, cached and routed, with precision, recall, mean cost per check, p50 and p95. Keep routing only if recall didn't drop. If it fell, look at which scams took the quick path and tighten is_easy; a cheaper checker that misses scams is a worse product.

Done when: docs/cost-report.md has the three rows, measured on the same messages, and a sentence on what you kept and why.

Stuck?
  • The script stops with KeyError: 'is_scam': your labelled file uses another column name. Change df["is_scam"] and df["text"] to match.
  • The cache rate is 0% in every run: check CACHE_SYSTEM_PROMPT is True and that your system prompt is long enough. A rough rule: 512 tokens is about 350 English words.
  • Costs went up with caching: the first call writes the cache, at 1.25 times the input price. You save only when later calls read it within five minutes. On a single check with one call, caching can't help.
  • p95 is much worse than p50: look at the slowest trace in Langfuse. It's usually one tool (a slow website lookup) or a retry.

Wednesday — When Claude is down

Learn (45 min):

Why a fallback model, not a fallback provider

The plan suggests LiteLLM with a second provider. That protects against a whole provider going down, but needs a second account, a second key, a second bill, and structured-output behaviour that differs between providers, all of which your eval would have to cover. This week uses a simpler layer that covers the common failures: if Haiku is overloaded or times out, try Sonnet once, and if that fails too, give general advice. Mention in interviews that a second provider is the next step, and why you didn't take it yet.

Build:

  1. Switch on the fallback model: set FALLBACK_MODEL back to {"claude-haiku-5-5": "claude-sonnet-5-5"}. Read call_claude() and the except blocks in _call_once() again. Errors now carry retryable: a rate limit (429), an overload (529), a server error or a timeout is worth one try on another model; a bad request or bad key isn't, because it would fail the same way.

  2. Tests for both cases, in tests/test_llm.py:

    def test_overloaded_model_falls_back_to_sonnet(fake_api):
    models = []

    def handler(request):
        model = json.loads(request.content)["model"]
        models.append(model)
        if model == "claude-haiku-5-5":
            error = {
                "type": "error",
                "error": {"type": "overloaded_error", "message": "busy"},
            }
            return httpx2.Response(529, json=error)
        return httpx2.Response(200, json=api_message("hello"))

    fake_api(handler)
    reply = llm.call_claude("hi")
    assert models == ["claude-haiku-5-5", "claude-sonnet-5-5"]
    assert reply.model == "claude-sonnet-5-5"


def test_bad_request_does_not_fall_back(fake_api):
    error = {
        "type": "error",
        "error": {"type": "invalid_request_error", "message": "no"},
    }
    fake_api(lambda request: httpx2.Response(400, json=error))
    with pytest.raises(llm.ClaudeError) as caught:
        llm.call_claude("hi")
    assert caught.value.retryable is False

  3. The time limit. Read run_check() in service.py again. The agent runs in a worker thread, and future.result(timeout=DEADLINE_SECONDS) waits at most 45 seconds. After that the user gets the general-advice answer, marked degraded. Without a limit, a slow website lookup plus retries could keep someone waiting for minutes. copy_context().run matters: it carries track_usage() and the Langfuse trace into the worker thread. (The worker keeps running in the background until its own call timeouts end; that's the price of this simple approach.)

  4. Your turn: test the time limit without waiting 45 seconds. In tests/test_api.py (add import time and from scam_checker import service at the top), shrink DEADLINE_SECONDS to 0.1 with monkeypatch, swap in an agent that sleeps for half a second, and check the response says degraded.

    Stuck? One version 
    def test_slow_agent_hits_the_deadline(client, monkeypatch):
    monkeypatch.setattr(service, "DEADLINE_SECONDS", 0.1)

    def slow_agent(text):
        time.sleep(0.5)
        return fake_agent(text)

    app.dependency_overrides[get_agent] = lambda: slow_agent
    assert client.post("/check", json={"text": "hello"}).json()["degraded"] is True

    monkeypatch puts the old value back after the test, so other tests still get 45 seconds.

  5. Health checks. main.py already has /health and /health/ready. Test them, including a database that isn't there:

    def test_health(client):
    assert client.get("/health").json() == {"status": "ok"}
    assert client.get("/health/ready").json() == {"status": "ready"}


def test_ready_fails_without_a_database(client):
    broken = create_engine("sqlite:////no/such/folder/checks.db")

    def broken_session():
        with Session(broken) as session:
            yield session

    app.dependency_overrides[get_session] = broken_session
    assert client.get("/health/ready").status_code == 503

  6. Show users when they're getting the fallback. In frontend/src/WhyPanel.jsx, under the heading:

          {result.degraded && (
        <p>We couldn&apos;t fully check this message just now, so here is general advice.</p>
      )}

  7. Break it on purpose, locally. Put a wrong value in .env (ANTHROPIC_API_KEY=sk-ant-wrong), run the API, and check a message. You should get the general advice, degraded: true, warning lines in the JSON log, and no crash. Put the real key back.

  8. Ship it to Cloud Run. Langfuse's secret key goes in Secret Manager like the others; the public key and address are ordinary settings:

    read -rs KEY && printf '%s' "$KEY" | gcloud secrets create langfuse-secret-key --data-file=- && unset KEY
gcloud secrets add-iam-policy-binding langfuse-secret-key \
  --member="serviceAccount:${PROJECT_NUMBER}-compute@developer.gserviceaccount.com" \
  --role=roles/secretmanager.secretAccessor
gcloud run services update scam-checker-api --region asia-southeast1 \
  --update-env-vars LANGFUSE_PUBLIC_KEY=pk-lf-...,LANGFUSE_BASE_URL=https://jp.cloud.langfuse.com \
  --update-secrets LANGFUSE_SECRET_KEY=langfuse-secret-key:latest \
  --startup-probe httpGet.path=/health,httpGet.port=8080

    The startup probe makes Cloud Run send traffic only once /health answers. Then add the new secret to the secrets: block in .github/workflows/deploy.yml. The action replaces the whole list on every deploy rather than adding to it, so a secret missing from the file disappears from the service:

              secrets: |-
            ANTHROPIC_API_KEY=anthropic-api-key:latest
            DATABASE_URL=database-url:latest
            LANGFUSE_SECRET_KEY=langfuse-secret-key:latest

    If deploy on merge slipped from week 9, set it up now, following week 9's Friday.

Done when: the fallback and time-limit tests pass, a wrong key gives general advice instead of an error, and the live service has the startup probe and Langfuse keys.

Stuck?
  • Every request falls back to Sonnet: Haiku is failing for a reason that isn't overload. Read the warning line; a 400 or 401 shouldn't fall back at all, so check retryable.
  • The deadline test passes, but slowly: the worker thread finishes its 0.5-second sleep in the background. That's expected.
  • /health/ready returns 503 on Cloud Run: the service can't reach Neon. Check the database-url secret, and remember Neon's free compute sleeps when idle and takes a moment to wake.
  • Cloud Run logs show text, not JSON: some lines come from Cloud Run itself, not your app. Your check done lines should show as structured entries with fields you can click.
  • Traces are missing on Cloud Run but work locally: Cloud Run slows a container's CPU between requests, which can delay the background upload. Traces usually arrive with the next request; if they never do, check the three Langfuse settings on the service.

Thursday — The same API on AWS

Learn (75 min):

This one costs money while it runs

Cloud Run scales to zero. An ECS Express Mode service doesn't: it keeps at least one task running, plus an Application Load Balancer billed by the hour, even when nobody visits. New-account credits should cover a few days, but check the Fargate and Elastic Load Balancing pricing pages, set the budget in step 1 first, and delete the service once you have your screenshots and numbers (step 9). Keep Cloud Run as your live version.

Build:

  1. Account basics, if you haven't done them on the cloud side track: create an AWS account, turn on multi-factor authentication for the root user, create an everyday admin user for yourself (AWS recommends IAM Identity Center; an IAM user with MFA also works for a solo project), and create a monthly cost budget with email alerts, say US$10. From here on, sign in as the admin user, never as root.

  2. The AWS command line, signed in through your browser:

    brew install awscli
aws --version            # aws login needs 2.32.0 or later
aws login
aws configure set region ap-southeast-1
ACCOUNT=$(aws sts get-caller-identity --query Account --output text)
REGION=ap-southeast-1

    ap-southeast-1 is Singapore, the same AWS region your Neon database is in.

  3. Push your image to ECR (Elastic Container Registry), AWS's version of Artifact Registry:

    aws ecr create-repository --repository-name scam-checker-api
aws ecr get-login-password | docker login --username AWS --password-stdin "${ACCOUNT}.dkr.ecr.${REGION}.amazonaws.com"
IMAGE="${ACCOUNT}.dkr.ecr.${REGION}.amazonaws.com/scam-checker-api:v1"
docker build -t "$IMAGE" .
docker push "$IMAGE"

    On an Apple-silicon Mac this builds an ARM image, which is why step 6 asks for ARM (Graviton) machines. Cloud Build made an x86 image for Cloud Run; both run the same code.

  4. Two secrets in Secrets Manager. Create a fourth Anthropic key in the Console, aws, so you can cancel it on its own:

    read -rs KEY && KEY_ARN=$(aws secretsmanager create-secret --name scam-checker/anthropic-api-key --secret-string "$KEY" --query ARN --output text) && unset KEY
read -rs KEY && DB_ARN=$(aws secretsmanager create-secret --name scam-checker/database-url --secret-string "$KEY" --query ARN --output text) && unset KEY
echo "$KEY_ARN" "$DB_ARN"

    An ARN (Amazon Resource Name) is AWS's full address for a thing, like the long resource names in Google Cloud.

  5. The two roles Express Mode needs, from AWS's tutorial. A role is an identity that AWS services can take on; the "trust policy" says which service may.

    aws iam create-role --role-name ecsTaskExecutionRole --assume-role-policy-document \
  '{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":{"Service":"ecs-tasks.amazonaws.com"},"Action":"sts:AssumeRole"}]}'
aws iam create-role --role-name ecsInfrastructureRoleForExpressServices --assume-role-policy-document \
  '{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":{"Service":"ecs.amazonaws.com"},"Action":"sts:AssumeRole"}]}'
aws iam attach-role-policy --role-name ecsTaskExecutionRole \
  --policy-arn arn:aws:iam::aws:policy/service-role/AmazonECSTaskExecutionRolePolicy
aws iam attach-role-policy --role-name ecsInfrastructureRoleForExpressServices \
  --policy-arn arn:aws:iam::aws:policy/service-role/AmazonECSInfrastructureRoleforExpressGatewayServices

    Then let the task execution role read your two secrets, and only those two. This is the AWS version of week 9's secretAccessor binding:

    aws iam put-role-policy --role-name ecsTaskExecutionRole --policy-name read-scam-checker-secrets \
  --policy-document "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Action\":\"secretsmanager:GetSecretValue\",\"Resource\":[\"${KEY_ARN}\",\"${DB_ARN}\"]}]}"

  6. Describe the container in a file, outside git (add container.json to .gitignore):

    cat > container.json <<EOF
{
  "image": "${IMAGE}",
  "containerPort": 8080,
  "secrets": [
    {"name": "ANTHROPIC_API_KEY", "valueFrom": "${KEY_ARN}"},
    {"name": "DATABASE_URL", "valueFrom": "${DB_ARN}"}
  ],
  "environment": [{"name": "ALLOWED_ORIGINS", "value": "https://<you>.github.io"}]
}
EOF

    Then create the service. It takes a few minutes; --monitor-resources shows progress:

    aws ecs create-express-gateway-service \
  --service-name scam-checker-api \
  --execution-role-arn "arn:aws:iam::${ACCOUNT}:role/ecsTaskExecutionRole" \
  --infrastructure-role-arn "arn:aws:iam::${ACCOUNT}:role/ecsInfrastructureRoleForExpressServices" \
  --primary-container file://container.json \
  --cpu-architecture ARM64 \
  --health-check-path /health \
  --scaling-target '{"minTaskCount":1,"maxTaskCount":1}' \
  --monitor-resources

    maxTaskCount of 1 is your AWS --max-instances: whatever happens, one copy runs.

  7. Test it at the address it prints, which looks like https://scam-checker-api.ecs.ap-southeast-1.on.aws/:

    curl -s https://<your-address>/health
curl -s -X POST https://<your-address>/check -H 'content-type: application/json' -d '{"text":"Your parcel is held. Pay $1.99 at singpost-redelivery.com"}'

    The check is stored in the same Neon database as Cloud Run's checks.

  8. Find your JSON logs in CloudWatch. Open CloudWatch in the AWS console, then Logs Insights, pick the service's log group, and run:

    fields @timestamp, check_id, verdict, cost_usd, duration_ms
| filter message = "check done"
| stats count(*) as checks, avg(cost_usd) as avg_cost, pct(duration_ms, 95) as p95_ms

    Logs Insights reads the JSON fields for you, which is the payoff for Monday's logs.py. Take a screenshot for your report.

  9. Clean up when you've got what you need:

    aws ecs delete-express-gateway-service --service-arn <the serviceArn it printed> --monitor-resources

    Keep the image, roles and secrets if you'll redeploy for an interview; delete the secrets too if you're done, since each one has a small monthly charge.

Done when: your API answered /check from an AWS address, the logs query ran, and the service is deleted (or you know exactly when you'll delete it).

Stuck?
  • aws: error: argument command: Invalid choice ... login: your AWS CLI is older than 2.32. Run brew upgrade awscli.
  • Unable to assume the service linked role right after creating the roles: IAM changes take a minute to spread. Wait and run the create command again.
  • The service never becomes healthy: the health check path must be /health and the container port 8080. Look at the task's logs for a Python error.
  • The task stops with an error about secrets: the inline policy's ARNs must match the secrets exactly. echo "$KEY_ARN" and compare.
  • exec format error in the logs: the image's architecture doesn't match. On an Intel Mac, leave out --cpu-architecture ARM64.
  • no basic auth credentials on docker push: the ECR login expired. Run the get-login-password line again.

Friday — The map, the report, and demo day

Learn (30 min): re-read your docs/cost-report.md and the Langfuse dashboard for the week, and pick the three numbers you'd quote in an interview.

Build (before 9:00 or after demo day):

  1. The one-page Google Cloud to AWS map, docs/cloud-map.md. Fill in the last column from what you did this week:

    Job Google Cloud (week 9) AWS (week 11) What surprised you
    Run the container Cloud Run ECS Express Mode (ECS on Fargate, with a load balancer)
    Store images Artifact Registry ECR
    Keep secrets Secret Manager Secrets Manager
    Who may do what IAM, service accounts IAM, roles and policies
    Logs Cloud Logging CloudWatch Logs
    Build an image Cloud Build (--source .) Your laptop, or CodeBuild or GitHub Actions
    Deploy from GitHub without keys Workload Identity Federation IAM OpenID Connect provider and a role
    Spending alerts Cloud Billing budgets AWS Budgets
    Managed Postgres Cloud SQL RDS (you used Neon, which runs on AWS)
    Health checks Startup and liveness probes Load balancer target health check
    Idle cost Zero: scales to nothing Not zero: a task and a load balancer

  2. Finish docs/cost-report.md. Keep it to one page:

    # Cost and performance report

## Headline
One check costs about US$__ and takes __ s (p50) / __ s (p95),
at precision __ and recall __ on 100 labelled messages.

## How it was measured
Model, date, data set, script (scripts/cost_latency.py), and what counts as one check.

## Before and after
| Setup | Precision | Recall | Cost per check (US$) | p50 (s) | p95 (s) |
| --- | --- | --- | --- | --- | --- |
| Full agent, no caching | | | | | |
| Plus prompt caching | | | | | |
| Plus routing | | | | | |

## Where the time and money go
From one typical trace: which steps, how long, how many tokens.

## Reliability
What happens on a timeout, an overload, a bad key; what the user sees; how often
answers were degraded this week (from scripts/report.sql).

## At scale
1,000 checks a day would cost about US$__ a month in model calls, plus hosting.

## Next
The next three things you'd change, and the number each should move.

    Run scripts/report.sql against Neon for the live numbers, using its SQL editor or psql. The "at scale" line is your cost per check × 1,000 × 30: simple arithmetic, stated as an estimate.

Demo day (9:00–12:00):

  1. Record a 3-minute video: one trace in Langfuse, the before-and-after table, the wrong-key test giving general advice, and the API answering from AWS.
  2. Write five lines on what you learned this week.
  3. Send both to Claude with the Friday check-in, with the cost report attached.

Friday afternoon: the cloud side track. This week's concept, logs, follows straight on from Thursday.

Deliverable checklist

  • Every check traced in Langfuse, with no raw message in the traces
  • Cost, tokens and time stored per check, through a second Alembic migration
  • Before-and-after numbers for caching and routing on the same 100 messages
  • A fallback model, a 45-second time limit and a general-advice answer, all tested
  • /health and /health/ready, and a startup probe on Cloud Run
  • One-line JSON logs, searchable in Cloud Logging and CloudWatch
  • The API answering from AWS, then cleaned up
  • docs/cloud-map.md and docs/cost-report.md
  • Demo video, five lines, and the Friday check-in sent

If you have time left: try the effort setting, which controls how much the model thinks before answering. Add "effort": "low" next to "format" in output_config, run cost_latency.py cached again, and see whether output tokens and p95 drop without recall dropping.