後端 take-home 作業測試模型 token 中途耗盡時的降級處理

A candidate's repo arrived last month with a tidy README and a demo that worked. I cloned it, ran the tests, then pointed the summarizer at a model that dies after six tokens. The endpoint answered 200 OK with {"summary": ""}. Every test was green. Nothing worked.

That is the take-home I now hand out. Not "call a model", not "wrap it in FastAPI" — that part takes an afternoon and a cheatsheet. The interesting question is what your code does in the half-second after the budget runs out, because on a free tier that half-second is not hypothetical. It is Tuesday.

So this post is the full task: the prompt, the rubric, a harness that fails on purpose, one reference solution, and the failure modes I keep grading. Run it yourself before you ship it to anyone.

The prompt, handed over verbatim

Build POST /summarize. It takes {"text": "..."} and returns {"summary": "...", "source": "model" | "fallback"}.

When the model is unavailable — quota exhausted, timeout, empty completion — the endpoint must still return a usable summary, and it must never return a half-written stream. Ship the test that proves it.

Notice what the prompt does not say. It does not say which provider, which model, or how many tokens you get. That ambiguity is deliberate. A candidate who starts by hard-coding a model name has already told me how they think about capacity.

The rubric

Five rows, and I have never given full marks on more than three. The last row is where most submissions quietly give up: they catch the error, log it, and hand the user an empty string. That is not degradation. That is a different bug with better manners.

A harness that dies on purpose

Do not test this against a live provider and hope. Build the failure into a fake you control, so the test is deterministic and free.

# harness.py
import asyncio

class QuotaExhausted(RuntimeError):
    pass

class FlakyModel:
    """Yields tokens until the budget is gone, then raises."""

    def __init__(self, token_budget=12, delay=0.01):
        self.remaining = token_budget
        self.delay = delay
        self.calls = 0

    async def stream(self, prompt):
        self.calls += 1
        for token in prompt.split():
            if self.remaining <= 0:
                raise QuotaExhausted("free tier budget spent")
            self.remaining -= 1
            await asyncio.sleep(self.delay)
            yield token

Two tests matter more than the rest. The first kills the model mid-stream and asserts you still get text. The second sets the budget to zero and asserts you did not retry your way into a bill.

# test_degrade.py
import pytest
from harness import FlakyModel
from summarize import summarize

@pytest.mark.asyncio
async def test_falls_back_mid_stream():
    model = FlakyModel(token_budget=6)
    out = await summarize("one two three four five six seven eight", model=model)
    assert out.source == "fallback"
    assert out.text.strip()
    assert out.tokens_spent == 6

@pytest.mark.asyncio
async def test_deadline_beats_retries():
    model = FlakyModel(token_budget=0)
    out = await summarize("anything at all", model=model, deadline_s=0.05)
    assert out.source == "fallback"
    assert model.calls == 1

Run it with python -m pytest -q. On a fresh checkout the second test is the one that fails, and it fails loudly, which is exactly what you want in a reviewer's terminal.

One solution that survives

The shape I look for is boring on purpose: a deadline that covers the whole request, a counter that survives the exception, and a fallback that touches no network at all.

# summarize.py
import asyncio, dataclasses, re, time

@dataclasses.dataclass
class Summary:
    text: str
    source: str
    tokens_spent: int
    elapsed_ms: int

def extractive(text: str, n: int = 2) -> str:
    sentences = re.split(r"(?<=[.!?])\s+", text.strip())
    return " ".join(s.strip() for s in sentences[:n] if s.strip()) or text[:200]

async def summarize(text, model, deadline_s=2.0, max_tokens=8):
    start = time.monotonic()
    spent, buf = 0, []
    try:
        async with asyncio.timeout(deadline_s):  # Python 3.11+
            async for tok in model.stream(text):
                buf.append(tok)
                spent += 1
                if spent >= max_tokens:
                    break
        if not buf:
            raise RuntimeError("empty completion")
        return Summary(" ".join(buf), "model", spent, _ms(start))
    except (QuotaExhausted, TimeoutError, RuntimeError):
        return Summary(extractive(text), "fallback", spent, _ms(start))

def _ms(t0):
    return int((time.monotonic() - t0) * 1000)

On 3.10, wrap the loop in asyncio.wait_for instead; the point is that the timeout is total, not per attempt. And note what the code refuses to do: it does not stream to the client as tokens arrive. Buffering looks slower in a demo and behaves better in production, because a partial stream that gets truncated is worse than a short summary that is complete.

Failure modes I keep grading

The retry ladder. Three attempts with backoff inside a 2-second budget means the user waits 2 seconds and you spend three times as much. Retries belong behind a deadline, and the deadline belongs above the retry.

Catching Exception. It hides the QuotaExhausted you were supposed to handle and the KeyError you were not. Enumerate what degrades; let the rest crash.

Fallback by apology. "Sorry, I couldn't summarize that." is a string, not a fallback. If a reviewer can delete the model call and see identical output quality, your fallback is the product.

Headers before tokens. If you flush a 200 and then fail, the client is committed to a body you cannot finish. Decide the status after the first successful token, or buffer.

Trusting the free tier. Free tiers change shape without notice. Any design that assumes today's allowance is tomorrow's allowance will be found out by the first morning it is not.

Where the free tier is actually useful

The reason to rehearse this at all is that the 429 is real money to someone. MonkeyCode's free model access makes the exhaustion path something you can trigger on purpose instead of reading about, and the free server option gives the harness somewhere to live so a reviewer can hit /summarize without provisioning anything. Disclosure: This article was prepared as part of MonkeyCode's product outreach. Limits and terms move, so check the current ones rather than planning around a number someone quoted in a blog post — including this one.

The honest framing: a free tier is a rehearsal space for failure, not a capacity plan. Use it to watch your own code hit the wall, then write the test that keeps it graceful.

Who should skip this

If your endpoint has a hard latency SLO under 200ms, an extractive fallback is still the right shape but the buffer has to shrink and the deadline has to shrink with it. If your users would be actively misled by a degraded answer — medical, legal, financial summaries — do not degrade at all; return an error and let the UI say so. And if your team has already pinned a specific model version for compliance reasons, the swap-in-a-fake trick still works but the fallback needs a human review path, not a regex.

Everyone else: write the failing test first. The moment you watch test_deadline_beats_retries go red, the design gets obvious.

If you want to run this against a live free-tier endpoint instead of a fake, seek out MonkeyCode when you have the failing test in hand — the drill is worth more than the demo.