Frontier Agent Capabilities · Capstone

Voice agent with memory

The Frontier Agent track, assembled. Build a voice agent that remembers you across turns, escalates its hardest turns to a reasoning model, retrieves with a hybrid + rerank pipeline, and refuses to ship unless an eval gate passes — every piece runnable offline, then wired to real SDKs at the edges.

⏱️ ~3 hours🧪 6 labs🎯 Capstone

Learning objectives

Assemble the whole Frontier Agent track into one voice agent: STT → memory + retrieval → agent → TTS.
Model long-term memory (write/retrieve) so the agent remembers across turns.
Route hard turns to a slower reasoning path and keep easy turns on a fast path.
Feed the agent an advanced retrieval context (hybrid + rerank).
Gate the whole thing behind an eval over golden turns — ship only if it passes.

This is the capstone for the Frontier Agent Capabilities track. You've built the pieces one lesson at a time — voice/realtime (FA1), reasoning models (FA3), long-term memory (FA4), RAG evaluation (FA5), and advanced retrieval (FA6). Here you wire them together into a single voice agent with memory that escalates its hardest turns to a reasoning model, retrieves with a real pipeline, and refuses to ship unless an eval passes.

⚙️ Everything here runs offlineEvery lab is stdlib-only Python — no network, no SDK, no API key. We model each frontier capability with a small deterministic stand-in so the full system runs on your laptop with plain python file.py. Blocks that would need a real service are labelled needs the SDK — read them, run the models.

Architecture — the whole system on one line essential

A spoken turn flows left to right: the mic is transcribed (STT), we gather context from long-term memory and advanced retrieval, the agent answers on a fast model but escalates hard turns to a reasoning model, the reply is spoken (TTS), and an eval gate sits across the whole thing deciding whether this build is allowed to ship.

🗺️ How to read this diagram

This is the whole capstone on one line. A spoken turn enters at the left and flows right through six stages; the last box sits across the whole pipeline as a quality gate. Follow the arrows.

Mic → STT turns speech into text. In our labs STT is an identity function (audio is text) so we can build the brain on plain strings; the real one is a streaming speech provider.
Memory + Retrieval gathers context: the long-term memory of what this user told us before, plus hybrid + rerank retrieval over a knowledge base. Both hand back the passages the agent should answer from.
Agent (fast) answers easy turns on a fast, cheap model. When a turn is hard, it hands off to Reasoning (hard) — a slower reasoning model that thinks step by step. A router decides which.
TTS → Speaker turns the reply back into speech. Like STT, it's an identity function in the labs and a low-latency voice provider in production.
Eval gate is not in the live path — it stands guard over the build: a set of golden turns the agent must get right, or the build is blocked from shipping.

In short: audio in → text → remembered + retrieved context → an agent that escalates the hard turns → audio out, with an eval standing guard. The rest of the page builds exactly this, one stage per step.

Read it as one sentence: audio in → text → remembered + retrieved context → an agent that escalates the hard turns → audio out, with an eval standing guard. The rest of this page builds that sentence one step at a time; by Step 5 the pieces snap together into a running turn(), and Step 6 puts an eval gate in front of it.

Step 1 · The turn loop skeleton essential

Start with the shape of a conversation and nothing else. Voice is just text with a microphone on the front and a speaker on the back, so we stand in fake STT/TTS as identity functions (they pass text straight through). Real speech recognition and synthesis need the SDK; the loop is identical either way — that's the point.

Example code for learning — review, test, and adapt it before running against real or production systems. Commands can create, change, or delete resources. See the Terms & Disclaimer.

Python · the turn loop (runs)

step1_loop.pydef stt(audio):            # speech-to-text stand-in: audio IS text here
    return audio

def tts(text):             # text-to-speech stand-in: identity, we "speak" text
    return text

def respond(user_text):    # placeholder brain — replaced in later steps
    return f"You said: {user_text}"

def turn(audio_in):
    """One full spoken turn: STT -> brain -> TTS."""
    user_text = stt(audio_in)
    reply_text = respond(user_text)
    return tts(reply_text)

def converse(script):
    for utterance in script:
        print("USER :", utterance)
        print("AGENT:", turn(utterance))

converse(["hello there", "what can you do?"])

USER : hello there
AGENT: You said: hello there
USER : what can you do?
AGENT: You said: what can you do?

▶ How this works

This is the skeleton of a conversation and nothing more. A voice turn is just text with a microphone on the front and a speaker on the back, so we fake both ends and get the loop right first.

stt(audio) and tts(text) are identity functions — they return their input unchanged. That lets us treat spoken audio as plain text everywhere else, so the agent's logic is testable without any speech service.
respond(user_text) is a placeholder brain that just echoes; Steps 2–5 replace it with real memory, routing, and retrieval.
turn(audio_in) is the one function that matters: it runs STT → brain → TTS in order. Every later step plugs into this exact shape.
converse(script) feeds a list of utterances through turn() and prints each exchange, so we can watch a whole conversation at once.

What the output means: Two exchanges print, each echoing the user (You said: …) — proof the STT → brain → TTS loop runs end to end with no network.

Try this: Change respond to upper-case the reply and re-run — the loop is unchanged, only the brain differs. That separation is why we can swap in memory and routing later without touching turn().

Why fake the voice?The hard, expensive parts of voice (streaming STT, barge-in, low-latency TTS from FA1) don't change the agent's logic at all. By making STT/TTS identity funcs we can build and test the brain on plain text, then swap in the real audio pipeline at the edges. Everything below plugs into this turn().

Step 2 · Plug in long-term memory essential

A voice agent that forgets you between turns feels broken. We add a MemoryStore (the FA4 idea): it writes facts and retrieves the most relevant ones by word overlap. Same interface a real vector memory would expose — swap the scorer for embeddings later.

Python · a MemoryStore (runs)

step2_memory.pydef _words(text):
    return {w.strip(".,!?").lower() for w in text.split() if w.strip(".,!?")}

class MemoryStore:
    """Long-term memory: write facts, retrieve by word overlap (FA4 idea)."""
    def __init__(self):
        self.facts = []

    def write(self, fact):
        self.facts.append(fact)

    def retrieve(self, query, k=2):
        q = _words(query)
        scored = []
        for f in self.facts:
            overlap = len(q & _words(f))
            if overlap:
                scored.append((overlap, f))
        scored.sort(key=lambda s: (-s[0], self.facts.index(s[1])))
        return [f for _, f in scored[:k]]

mem = MemoryStore()
mem.write("The user's name is Dana.")
mem.write("Dana prefers metric units.")
mem.write("Dana is allergic to peanuts.")

print("recall(name)  :", mem.retrieve("what is my name"))
print("recall(units) :", mem.retrieve("use which units for Dana"))
print("recall(none)  :", mem.retrieve("stock market forecast"))

recall(name)  : ["The user's name is Dana.", 'Dana is allergic to peanuts.']
recall(units) : ['Dana prefers metric units.', "The user's name is Dana."]
recall(none)  : []

▶ How this works

A voice agent that forgets you between turns feels broken. The MemoryStore fixes that: it writes facts as the conversation goes and retrieves the most relevant ones on demand — the FA4 long-term-memory idea in miniature.

_words(text) lowercases a string and splits it into a set of words (stripping punctuation). Sets make overlap a fast intersection.
write(fact) just appends to a list — that list is the long-term memory, and it survives across turns because the store outlives any single turn.
retrieve(query, k=2) scores every stored fact by how many words it shares with the query (len(q & _words(f))), drops the zero-overlap ones, sorts best-first, and returns the top k.
When nothing overlaps, it returns an empty list — the agent should notice it has no relevant memory rather than invent one.

What the output means: Asking for the name recalls the stored name fact; asking about units recalls the units fact first; an unrelated query returns [].

Try this: Add mem.write("Dana lives in Berlin.") then mem.retrieve("where does Dana live") — the new fact surfaces because it shares "dana" and "live/lives" prefixes. Swap the overlap scorer for embeddings and the interface is unchanged.

Retrieve returns [] when nothing matchesJust like the RAG grounding rule (FA5): an empty result is a signal, not a bug. The agent should notice it has no relevant memory rather than invent one. We reuse this exact word-overlap scorer for retrieval in Step 4.

Step 3 · A difficulty router (escalate hard turns) intermediate

Most turns are easy — a fast, cheap model handles them. The hard ones (multi-step, ambiguous, high-stakes) deserve a slower reasoning model (FA3). A difficulty router decides which path a turn takes. We model both paths as deterministic functions so the routing logic is testable without a model; the real version swaps reason_path for a reasoning-model call.

Python · difficulty router (runs)

step3_router.pyHARD_SIGNALS = ("why", "compare", "explain", "step by step",
                "trade-off", "tradeoff", "prove", "debug")

def difficulty(user_text):
    """Return 'hard' if the turn needs the reasoning path, else 'easy'."""
    t = user_text.lower()
    if any(sig in t for sig in HARD_SIGNALS):
        return "hard"
    if len(user_text.split()) > 12:      # long, multi-clause asks are harder
        return "hard"
    return "easy"

def fast_path(user_text, context):
    return f"[fast] {user_text.rstrip('?.')} -> {context or 'no context'}"

def reason_path(user_text, context):
    # stand-in for a reasoning model: shows explicit deliberation, deterministic
    steps_ = [s for s in ("recall", "retrieve", "weigh", "answer")]
    return f"[reason:{'>'.join(steps_)}] {user_text.rstrip('?.')} -> {context or 'no context'}"

def route(user_text, context):
    return (reason_path if difficulty(user_text) == "hard" else fast_path)(user_text, context)

print(difficulty("what time is it"))
print(difficulty("explain why the two plans differ"))
print(route("what time is it", "clock=10:00"))
print(route("compare plan A and plan B", "A cheaper; B faster"))

easy
hard
[fast] what time is it -> clock=10:00
[reason:recall>retrieve>weigh>answer] compare plan A and plan B -> A cheaper; B faster

▶ How this works

Most turns are easy and belong on a fast, cheap model; the hard ones deserve a slower reasoning model. The difficulty router decides which path each turn takes — the FA3 test-time-compute discipline as plain code.

difficulty(user_text) returns "hard" if the text contains a reasoning signal word (why, compare, explain, …) or is long (>12 words, i.e. multi-clause); otherwise "easy".
fast_path and reason_path stand in for two real models. They're deterministic so the routing is testable; reason_path shows explicit recall>retrieve>weigh>answer deliberation to signal the slower thinking.
route is the whole idea in one line: pick reason_path when the turn is hard, else fast_path, and call it with the same arguments either way.

What the output means: "what time is it" routes easy; "explain why…" routes hard; the two route() calls show a fast answer and a deliberate [reason:…] answer.

Try this: Add "prove" is already a signal — try "prove the total is right" and watch it route hard. In production you'd also escalate on low confidence or a failed self-check, not just keywords.

Routing is a cost/quality leverSending every turn to a reasoning model is slow and expensive; sending none loses the hard cases. The router captures the FA3 discipline: spend test-time compute only where it pays. In production you'd also escalate on low confidence or a failed self-check, not just keywords.

Step 4 · Advanced retrieval (hybrid + rerank) advanced

The agent answers from a knowledge base, and retrieval quality is answer quality (FA6). We model an advanced pipeline: hybrid scoring (keyword overlap + a lexical-prefix bonus standing in for a dense signal), then a rerank stub that boosts passages whose terms cluster near the query. Same retrieve() shape as the memory store, so it drops straight into the agent.

Python · hybrid retrieve + rerank (runs)

step4_retrieval.pydef _words(text):
    return {w.strip(".,!?").lower() for w in text.split() if w.strip(".,!?")}

KB = {
    "kb-1": "Reset your password from Settings > Security.",
    "kb-2": "Refunds are issued to the original payment method within 5 days.",
    "kb-3": "Export your data as CSV or JSON from the account page.",
    "kb-4": "Passwords must be at least 12 characters and rotated yearly.",
}

def _lexical(query_words, text_words):        # keyword overlap
    return len(query_words & text_words)

def _dense_stub(query_words, text_words):     # stands in for an embedding signal
    # reward shared word-prefixes (e.g. 'reset'/'resetting') as a cheap semantic proxy
    prefixes = {w[:4] for w in query_words}
    return sum(1 for w in text_words if w[:4] in prefixes) * 0.5

def hybrid_search(query, k=3):
    q = _words(query)
    scored = []
    for cid, text in KB.items():
        tw = _words(text)
        score = _lexical(q, tw) + _dense_stub(q, tw)
        if score:
            scored.append((score, cid, text))
    scored.sort(key=lambda s: (-s[0], s[1]))
    return scored[:k]

def rerank(query, hits):
    """Rerank stub: boost hits whose matched terms are dense (short passages)."""
    q = _words(query)
    reranked = []
    for score, cid, text in hits:
        density = len(q & _words(text)) / max(len(_words(text)), 1)
        reranked.append((round(score + density, 3), cid, text))
    reranked.sort(key=lambda s: (-s[0], s[1]))
    return reranked

def retrieve(query, k=2):
    hits = hybrid_search(query, k=3)
    best = rerank(query, hits)[:k]
    return [f"[{cid}] {text}" for _, cid, text in best]

print("password ->", retrieve("how do I reset my password"))
print("refund   ->", retrieve("can I get a refund"))

password -> ['[kb-1] Reset your password from Settings > Security.', '[kb-4] Passwords must be at least 12 characters and rotated yearly.']
refund   -> ['[kb-2] Refunds are issued to the original payment method within 5 days.']

▶ How this works

The agent answers from a knowledge base, and retrieval quality is answer quality. This models the FA6 advanced pipeline: a hybrid score (two signals) followed by a rerank pass — the shape a real system uses, with the heavy parts stubbed.

_lexical is keyword overlap (the sparse/BM25-style signal); _dense_stub rewards shared word-prefixes as a cheap stand-in for an embedding (dense) signal.
hybrid_search sums both signals per KB entry, keeps the non-zero ones, sorts best first, and returns the top few candidates — this is the hybrid retrieve.
rerank is a second pass: it boosts hits whose matched terms are dense in a short passage (matched-words ÷ passage-length), then re-sorts. Real systems use a cross-encoder here; the shape is the same.
retrieve(query, k) ties them together and returns cited [kb-N] text strings — the exact same interface the memory store exposes, so it drops into the agent unchanged.

What the output means: "reset my password" returns the password article (with the password-policy article second); "refund" returns the refunds article.

Try this: Add a KB entry about "password recovery email" and re-query — watch hybrid scoring surface it and rerank reorder by density. Swap the two stubs for real BM25 + embedding retrievers and nothing downstream changes.

Hybrid + rerank, in miniatureReal systems combine a sparse (BM25) and a dense (embedding) retriever, then rerank with a cross-encoder. We keep the shape — two signals summed, then a rerank pass — so the agent's retrieve() contract matches what you'd deploy. Swap the two stubs for the real retrievers and nothing downstream changes.

Step 5 · Assemble the full agent turn (end-to-end) professional

Now compose everything. This lab imports nothing external — it inlines the four pieces from Steps 1–4 (memory, router, hybrid retrieval, STT/TTS) and wires them into one VoiceAgent.turn(): transcribe → gather memory + retrieval context → route (escalating hard turns to the reasoning path) → write anything learned back to memory → speak. Then we run a scripted conversation end to end and watch memory and routing actually take effect.

Python · the composed voice agent (runs end-to-end)

step5_agent.py# ---- pieces from Steps 1-4, composed into one runnable agent ----

def _words(text):
    return {w.strip(".,!?").lower() for w in text.split() if w.strip(".,!?")}

def stt(audio):  return audio          # Step 1
def tts(text):   return text           # Step 1

class MemoryStore:                      # Step 2
    def __init__(self): self.facts = []
    def write(self, fact): self.facts.append(fact)
    def retrieve(self, query, k=2):
        q = _words(query)
        scored = [(len(q & _words(f)), f) for f in self.facts]
        scored = [(o, f) for o, f in scored if o]
        scored.sort(key=lambda s: (-s[0], self.facts.index(s[1])))
        return [f for _, f in scored[:k]]

HARD_SIGNALS = ("why", "compare", "explain", "trade-off", "tradeoff", "prove", "debug")
def difficulty(t):                      # Step 3
    if any(s in t.lower() for s in HARD_SIGNALS): return "hard"
    return "hard" if len(t.split()) > 12 else "easy"

KB = {                                   # Step 4
    "kb-1": "Reset your password from Settings > Security.",
    "kb-2": "Refunds go to the original payment method within 5 days.",
    "kb-3": "Export your data as CSV or JSON from the account page.",
}
def retrieve(query, k=1):
    q = _words(query)
    scored = [(len(q & _words(t)), cid, t) for cid, t in KB.items()]
    scored = [(s, c, t) for s, c, t in scored if s]
    scored.sort(key=lambda x: (-x[0], x[1]))
    return [f"[{c}] {t}" for _, c, t in scored[:k]]

class VoiceAgent:
    def __init__(self):
        self.mem = MemoryStore()

    def _brain(self, user_text, context):
        path = difficulty(user_text)
        prefix = "[reason]" if path == "hard" else "[fast]"
        ctx = context or "no context"
        return f"{prefix} {ctx}"

    def turn(self, audio_in):
        user_text = stt(audio_in)                       # Step 1: STT
        # remember simple "my name is / I am" facts
        low = user_text.lower()
        if "my name is" in low:
            self.mem.write("The user's name is " + user_text.split()[-1].strip(".!?") + ".")
        # gather context: memory first, then KB retrieval (Steps 2 + 4)
        ctx_parts = self.mem.retrieve(user_text) + retrieve(user_text)
        context = " | ".join(ctx_parts)
        reply = self._brain(user_text, context)         # Step 3: route
        return tts(reply)                                # Step 1: TTS

agent = VoiceAgent()
script = [
    "my name is Dana",
    "what is my name",
    "how do I reset my password",
    "explain why my refund is delayed compared to a chargeback",
]
for utterance in script:
    print("USER :", utterance)
    print("AGENT:", agent.turn(utterance))

USER : my name is Dana
AGENT: [fast] The user's name is Dana.
USER : what is my name
AGENT: [fast] The user's name is Dana.
USER : how do I reset my password
AGENT: [fast] [kb-1] Reset your password from Settings > Security.
USER : explain why my refund is delayed compared to a chargeback
AGENT: [reason] The user's name is Dana. | [kb-2] Refunds go to the original payment method within 5 days.

▶ How this works

This is the capstone: the four pieces from Steps 1–4 inlined and wired into one VoiceAgent.turn(), then run over a scripted conversation so you can watch memory, retrieval, and routing all fire together — offline.

The top of the file is Steps 1–4 in compact form: stt/tts identities, a MemoryStore, a difficulty router, and a retrieve over a tiny KB. Nothing is imported — the composition is self-contained and runnable.
VoiceAgent.turn() is the whole architecture in order: STT the audio, write any "my name is …" fact to memory, gather context from memory + retrieval, route the turn through _brain (fast vs reason), and TTS the reply.
_brain tags the reply [fast] or [reason] based on difficulty, and folds in whatever context was gathered — so the trace shows exactly which path and which context each turn used.
The scripted loop drives four turns that each exercise a different capability: store, recall, retrieve, and escalate.

What the output means: Turn 1 stores and recalls the name; turn 2 recalls it again; turn 3 pulls the KB password article; turn 4 hits a why/compare signal and answers on the [reason] path while still attaching memory + a KB passage.

Try this: Add a turn like "compare CSV and JSON export" — it should route [reason] and retrieve kb-3. This is the whole FA track running as one function.

This is the capstone workingWatch the trace: turn 1 stores the name (and recalls it the same turn), turn 2 recalls it from memory again, turn 3 retrieves from the KB, and turn 4 hits a why/compare signal so it escalates to the reasoning path ([reason]) while still pulling both memory and a KB passage into context. Memory, retrieval, and routing all fired in one composed turn() — the whole FA track, running offline.

Step 6 · An eval gate over golden turns tech-lead

A capstone isn't done because it runs — it's done because it passes an eval (FA5). We define a handful of golden turns with the behaviour we require (the right routing, a grounded recall) and gate the build on them. If any golden turn fails, the eval hard-fails — you don't ship. This is the same eval-gate discipline that a tech lead puts in front of every agent.

Python · the eval gate (runs, hard-fails on regression)

step6_evalgate.py# reuse the routing + a tiny agent for a self-contained, checkable eval
HARD_SIGNALS = ("why", "compare", "explain", "trade-off", "tradeoff", "prove", "debug")
def difficulty(t):
    if any(s in t.lower() for s in HARD_SIGNALS): return "hard"
    return "hard" if len(t.split()) > 12 else "easy"

# golden turns: (utterance, expected_route)
GOLDEN = [
    ("what time is it", "easy"),
    ("turn on the light", "easy"),
    ("explain why plan A costs more", "hard"),
    ("compare the two refund options step by step for me now", "hard"),
    ("debug this failing deploy", "hard"),
]

def run_eval(golden):
    passed = 0
    for utterance, expected in golden:
        got = difficulty(utterance)
        ok = got == expected
        passed += ok
        print(f"[{'PASS' if ok else 'FAIL'}] route={got:<4} expect={expected:<4} :: {utterance}")
    rate = passed / len(golden)
    print(f"routing accuracy: {passed}/{len(golden)}")
    gate = rate == 1.0
    print("GATE:", "PASS - safe to ship" if gate else "FAIL - do not ship")
    return gate

ok = run_eval(GOLDEN)
raise SystemExit(0 if ok else 1)

[PASS] route=easy expect=easy :: what time is it
[PASS] route=easy expect=easy :: turn on the light
[PASS] route=hard expect=hard :: explain why plan A costs more
[PASS] route=hard expect=hard :: compare the two refund options step by step for me now
[PASS] route=hard expect=hard :: debug this failing deploy
routing accuracy: 5/5
GATE: PASS - safe to ship

▶ How this works

A capstone isn't done because it runs — it's done because it passes an eval. This gate defines golden turns with the routing we require and refuses to ship on any regression (FA5 discipline).

GOLDEN is a list of (utterance, expected_route) pairs — the behaviour we have decided is correct. Two easy turns, three hard ones.
run_eval runs each utterance through the same difficulty router the agent uses, compares to the expected route, and prints PASS/FAIL per turn.
gate = rate == 1.0 means every golden turn must pass — a single miss fails the gate. It prints "safe to ship" only when all pass.
raise SystemExit(0 if ok else 1) makes the script exit non-zero on failure, so it can sit in CI and physically block a regressing build.

What the output means: All five golden turns route correctly (5/5) and the gate prints PASS - safe to ship; the process exits 0.

Try this: Change one expected route to the wrong value and re-run — that line flips to FAIL, the gate says "do not ship", and the exit code becomes 1. That non-zero exit is the guardrail doing its job.

The gate is the pointThe eval exits non-zero on any failure (raise SystemExit(1)), so it can sit in CI and physically block a regressing build from shipping. Add golden turns for memory recall and grounding too — a capstone that runs but doesn't pass the gate is not done.

Going real — the same shape with SDKs tech-lead

The offline simulation and a production build share one architecture; only the edges change. Here is what the real wiring looks like — needs the SDK, shown for orientation, not to run:

Python · real wiring (needs the SDK)

real_agent.py# needs the SDK: pip install anthropic + a speech provider; set ANTHROPIC_API_KEY
import anthropic
client = anthropic.Anthropic()

FAST_MODEL   = "claude-haiku-4-5"     # easy turns: fast + cheap
REASON_MODEL = "claude-opus-4-1"      # hard turns: reasoning / extended thinking

def brain(user_text, context, model):
    msg = client.messages.create(
        model=model,
        max_tokens=400,
        system="You are a voice assistant. Answer only from the provided context; "
               "if it is empty, say you don't know.",
        messages=[{"role": "user",
                   "content": f"Context:\n{context}\n\nUser: {user_text}"}],
    )
    return msg.content[0].text

def turn(audio_bytes, mem):
    user_text = real_stt(audio_bytes)             # streaming STT provider
    context = " | ".join(mem.retrieve(user_text) + retrieve(user_text))
    model = REASON_MODEL if difficulty(user_text) == "hard" else FAST_MODEL
    reply = brain(user_text, context, model)
    return real_tts(reply)                         # low-latency TTS provider

Same diagram, real partsstt/tts become a streaming speech provider; _brain becomes a real model call that picks the model by difficulty; retrieve becomes your hybrid + rerank retriever. The memory store, router, and eval gate you built above are unchanged. Check current model IDs against C1 before deploying.

Capstone exercise — extend the agent

Context: A capstone extension proves you can add a capability across the whole stack and defend it with an eval — the same discipline that keeps a shipped voice agent from silently regressing.

Your task: Take the Step-5 VoiceAgent and add one capability end-to-end, then defend it with a golden turn in the Step-6 gate; ship only when the gate is green.

Requirements:

Pick one: confidence-based escalation (re-run low-confidence fast answers on the reasoning path)
or memory recall as a golden eval (store a fact, assert recall two turns later)
or a grounding check that refuses when both memory and retrieval return []
Wire the chosen capability through the existing turn pipeline
Add a golden turn to the Step-6 gate that would fail if the capability broke
Ship only when the whole gate passes

💡 Hint: Whichever you choose, the golden turn is the point — a capability with no eval defending it is a capability that will quietly regress.

🪜 Practice ladder beginner → industry

Six graded exercises, easy to real-world. Try each before opening its solution.

Exercise 1 · Turn-loop skeleton with STT/TTS stubsBeginner

Context: A voice agent is a loop: speech-to-text → respond → text-to-speech. Stubbing the ends as identity functions makes the whole turn pipeline runnable in CI — you swap in real STT/TTS at the boundary without changing the agent logic.

Your task: Implement stt and tts as identity stubs and a turn(audio_in) that echoes a canned reply.

Requirements:

stt and tts are identity functions offline
turn runs the full stt → respond → tts pipeline
It returns a canned reply derived from the input
The whole turn runs with no mic or speech provider
This is the harness every later step plugs into

💡 Hint: Let audio 'be' the text offline so the middle is testable; the real providers attach at the stt/tts boundary later.

Show solution

Stub the ends so the middle is testable without a mic or speech provider:

def stt(audio):            # real: streaming speech-to-text provider
    return audio           # offline: audio IS the text

def tts(text):             # real: low-latency text-to-speech provider
    return text            # offline: return the string we'd speak

def turn(audio_in):
    user_text = stt(audio_in)
    reply = f"You said: {user_text}"
    return tts(reply)

print(turn("hello"))       # You said: hello

Identity stubs mean the whole turn pipeline runs in CI. You swap in real STT/TTS at the boundary without changing the agent logic in the middle.

Exercise 2 · MemoryStore that recalls facts by word overlapIntermediate

Context: Long-term memory that's embeddings-free and deterministic is ideal for offline tests. The empty-result path matters as much as the hit — returning nothing on no overlap is what stops the agent 'recalling' an unrelated fact.

Your task: Build MemoryStore with write(fact) and retrieve(query, k=2) scoring by word overlap.

Requirements:

write stores a fact
retrieve scores stored facts by word overlap with the query
It returns the top-k overlapping facts
It returns [] when nothing overlaps
Show it recalls a name but returns nothing for an unrelated query

💡 Hint: Score by the size of the word-set intersection and skip facts with zero overlap so no false memory surfaces.

Show solution

Word-overlap memory is embeddings-free and deterministic — ideal for offline tests:

class MemoryStore:
    def __init__(self): self.facts = []
    def write(self, fact): self.facts.append(fact)
    def retrieve(self, query, k=2):
        q = set(query.lower().split())
        scored = []
        for f in self.facts:
            overlap = len(q & set(f.lower().split()))
            if overlap:
                scored.append((overlap, f))
        scored.sort(reverse=True)
        return [f for _, f in scored[:k]]

m = MemoryStore()
m.write("my name is Sam")
m.write("I prefer window seats")
print(m.retrieve("what is my name"))     # ['my name is Sam']
print(m.retrieve("weather tomorrow"))    # []  -> no false memory

Returning [] on no overlap is what prevents the agent from "recalling" an unrelated fact and confusing the user. The empty-result path is as important as the hit.

Exercise 3 · Difficulty router: escalate hard turns to a reasoning modelAdvanced

Context: Cheap turns should hit a fast model and hard ones a reasoning model. The router is the cost lever: sending every turn to the reasoning model is slow and expensive; sending none loses quality on the turns that need it.

Your task: Build difficulty(user_text) and a _brain that tags its reply with the path taken.

Requirements:

difficulty returns 'hard' on reasoning signals (why/compare/explain) or long input (>12 words)
Otherwise it returns 'easy'
_brain routes hard turns to the reasoning path, easy to the fast path
The reply is tagged with which path handled it
Tune the hard-signal list from real transcripts

💡 Hint: A keyword set plus a length threshold is enough; the fast/reason paths stand in for a cheap vs a reasoning model.

Show solution

Routing by signal keeps latency and cost low on the 80% of easy turns:

HARD_SIGNALS = ("why", "compare", "explain", "difference", "trade-off", "analyze")

def difficulty(user_text):
    t = user_text.lower()
    if any(sig in t for sig in HARD_SIGNALS) or len(t.split()) > 12:
        return "hard"
    return "easy"

def fast_path(text, ctx):    # real: claude-haiku-4-5
    return f"[fast] {text}"
def reason_path(text, ctx):  # real: claude-opus-4-1 with extended thinking
    return f"[reason] {text}"

def _brain(user_text, context=""):
    if difficulty(user_text) == "hard":
        return reason_path(user_text, context)
    return fast_path(user_text, context)

print(_brain("hi"))                                  # [fast] hi
print(_brain("why is the sky blue exactly?"))        # [reason] ...

The router is the cost lever: sending every turn to the reasoning model is slow and expensive; sending none loses quality on the turns that need it. Tune HARD_SIGNALS from real transcripts.

Exercise 4 · Hybrid retrieval + density rerankExpert

Context: Pure keyword retrieval misses paraphrases; pure dense misses exact terms. The same two-stage shape as text RAG — hybrid recall then a precision rerank — sized for a voice turn keeps the spoken answer tight.

Your task: Build hybrid_search combining lexical overlap with a dense stand-in, then rerank the top hits by term density.

Requirements:

A lexical-overlap score catches exact term matches
A dense stand-in (e.g. prefix overlap) catches near-misses
The two are combined into a hybrid score for recall
A rerank by term density promotes the most on-topic chunk
Return the reranked top-k

💡 Hint: Sum the lexical and (weighted) dense scores for stage one, then sort the survivors by matched-terms-per-token for the rerank.

Show solution

Hybrid recall then a precision rerank — the same two-stage shape as text RAG, sized for a voice turn:

KB = {1: "reset your password from the settings page",
      2: "billing invoices are emailed monthly",
      3: "password resets expire after one hour"}

def lexical(q, text):
    qs = set(q.lower().split()); ts = set(text.lower().split())
    return len(qs & ts)

def dense_stub(q, text):                 # prefix overlap stands in for embeddings
    return sum(1 for w in q.lower().split() if any(t.startswith(w[:4])
               for t in text.lower().split()))

def hybrid_search(q, k=3):
    scored = [(lexical(q, t) + 0.5*dense_stub(q, t), cid, t)
              for cid, t in KB.items()]
    scored.sort(reverse=True)
    return [(cid, t) for _, cid, t in scored[:k]]

def rerank(q, hits):
    def density(text):
        toks = text.split()
        return lexical(q, text) / (len(toks) or 1)
    return sorted(hits, key=lambda h: density(h[1]), reverse=True)

def retrieve(query, k=2):
    hits = rerank(query, hybrid_search(query))
    return [f"[kb-{cid}] {t}" for cid, t in hits[:k]]

print(retrieve("how do I reset my password"))
# ['[kb-1] reset your password ...', '[kb-3] password resets expire ...']

The lexical term catches exact matches ("password"); the dense stub catches near-misses; the rerank promotes the densest, most on-topic chunk so the spoken answer is tight.

Exercise 5 · Eval gate on golden turns — all-or-nothing in CIProfessional

Context: A voice agent that regresses is invisible until a user hits it. An all-or-nothing gate is strict on purpose — for a shipped agent, one regressed critical turn is a bug, so CI must block on it.

Your task: Build a golden-turn eval that checks routing AND memory recall and hard-fails unless every golden turn passes.

Requirements:

Golden turns assert the expected routing decision
Memory recall is checked (a stored fact is recalled; no cross-session leak)
Print the failing turns for diagnosis
Hard-fail with a non-zero exit unless the pass rate is 100%
CI blocks the regression

💡 Hint: Require a perfect pass rate and SystemExit(1) otherwise — the eval is the contract between 'works on my laptop' and 'safe to ship'.

Show solution

An all-or-nothing gate is strict on purpose — for a shipped agent, one regressed critical turn is a bug:

GOLDEN = [
    {"text": "hi there",                    "route": "easy"},
    {"text": "why did my payment fail",     "route": "hard"},
    {"text": "compare the two plans please", "route": "hard"},
]

def run_eval(golden):
    passed = 0
    for g in golden:
        got = difficulty(g["text"])
        ok = got == g["route"]
        passed += ok
        if not ok:
            print(f"FAIL {g['text']!r}: got {got}, want {g['route']}")
    return passed / len(golden)

rate = run_eval(GOLDEN)
print(f"pass rate {rate:.0%}")
gate = rate == 1.0
raise SystemExit(0 if gate else 1)     # CI: block unless 100%

Add memory-leak checks to the same gate (a fact from session A must not surface in session B). The eval is the contract between "works on my laptop" and "safe to ship".

Exercise 6 · Barge-in and latency budget for a live callIndustry scenario

Context: On a real call the user interrupts (barge-in) and expects sub-second first audio. Latency and interruption are what separate a demo from a call people tolerate.

Your task: Add a turn latency budget (STT + brain + TTS) and barge-in cancellation.

Requirements:

A per-turn latency budget summed across STT, brain, and TTS
A turn that exceeds the budget reports over-budget and where
Barge-in mid-reply cancels the turn
A cancelled turn reports where it stopped
Demonstrate an in-budget turn, an over-budget turn, and a cancellation

💡 Hint: Accumulate per-stage cost and bail when it exceeds the budget or the turn is cancelled; real wiring streams partial TTS and aborts on interrupt.

Show solution

Latency and interruption are what separate a demo from a call people tolerate:

class Turn:
    def __init__(self, budget_ms=900):
        self.budget_ms = budget_ms; self.cancelled = False
    def barge_in(self):            # new user audio arrived mid-reply
        self.cancelled = True
    def run(self, audio_in, stt_ms, brain_ms, tts_ms):
        spent = 0
        for stage, cost in [("stt", stt_ms), ("brain", brain_ms), ("tts", tts_ms)]:
            if self.cancelled:
                return {"status": "cancelled", "at": stage, "spent_ms": spent}
            spent += cost
            if spent > self.budget_ms:
                return {"status": "over_budget", "at": stage, "spent_ms": spent}
        return {"status": "ok", "spent_ms": spent}

t = Turn(budget_ms=900)
print(t.run("q", 120, 500, 200))          # ok, 820ms
t2 = Turn(); t2.barge_in()
print(t2.run("q", 120, 500, 200))         # cancelled at stt

The budget lets you route long turns to the fast model to stay under target; barge-in cancellation stops the agent talking over the user. Real wiring streams partial TTS and aborts the model request on interrupt.

✓ Checkpoint — you can move on when you can…

Explain the capstone architecture: STT → memory + retrieval → agent → TTS, with an eval gate.
Run a MemoryStore that writes facts and retrieves them by overlap.
Route hard turns to a reasoning path and easy turns to a fast path.
Retrieve with a hybrid + rerank pipeline behind a stable retrieve().
Run the composed turn() end to end and see memory + routing fire.
Gate the build on golden turns and refuse to ship on a regression.

📋 📋 Grade your voice agent

Dimension	Meets the bar	Above the bar (staff)
Turn latency	End-to-end response under a defined budget (~800ms) on the happy path	p95 stays under budget with barge-in and graceful fallback on slow turns
Memory correctness	Recalls facts written earlier in the session; no cross-session leakage	Consolidates/forgets sensibly; retrieval stays relevant as history grows
Reasoning escalation	Hard turns route to the reasoning path; easy turns stay fast	Routing is measured (cost vs quality), not hard-coded guesswork
Retrieval quality	Relevant context is fetched and actually used in the reply	Advanced pipeline (hybrid/rerank) with a measured recall gain
Eval gate	A golden set of turns runs; regressions are caught before ship	Safety + quality gated in CI; failing the gate blocks release
Robustness	Handles STT errors / empty input without crashing	Degrades gracefully; observable per-turn (latency, cost, path taken)

Score each 0 (missing) / 1 (meets) / 2 (above). 6–8 = solid; 9–12 = staff-level. Blocking: no eval gate, or memory leaks across sessions.

Knowledge check check yourself

✓ Knowledge check

Why does the agent route hard turns to a slower reasoning model while keeping easy turns on a fast, cheap model?

Show answer

Most turns are easy and don't justify the latency and cost of step-by-step reasoning; a router keeps them on the fast path and escalates only the hard turns to the reasoning model. This spends expensive test-time compute where it changes the answer, not everywhere.

✓ Knowledge check

Why are STT and TTS modeled as identity functions in the offline labs, and why doesn't that weaken the design?

Show answer

Voice is just text with a microphone on the front and a speaker on the back, so treating audio as text lets the whole agent brain (memory, routing, retrieval, eval) be built and tested offline with no SDK. The turn() loop is identical when real streaming STT/TTS are swapped in at the edges.

Educational content, provided as-is and without warranty. Code samples are examples — review, test, and adapt them before using in production. See the Terms of Use & Disclaimer. Use at your own risk.

Learning objectives

Architecture — the whole system on one line essential

Step 1 · The turn loop skeleton essential

Step 2 · Plug in long-term memory essential

Step 3 · A difficulty router (escalate hard turns) intermediate

Step 4 · Advanced retrieval (hybrid + rerank) advanced

Step 5 · Assemble the full agent turn (end-to-end) professional

Step 6 · An eval gate over golden turns tech-lead

Going real — the same shape with SDKs tech-lead

Capstone exercise — extend the agent

🪜 Practice ladder beginner → industry

✓ Checkpoint — you can move on when you can…

Knowledge check check yourself

More in Projects