NovaBank · Solutions

Answers Q1–Q15

A1 · Agents

A2 · Prebuilt middleware core

A3 · Advanced

Working solutions

One complete solution per exercise. Environment cell (once): load_dotenv(), assert OPENAI_API_KEY, init_chat_model("openai:gpt-5-mini").

B1.1 — Agent with real memory

Proof is two separate .invoke() calls on the same thread_id with a checkpointer. Without InMemorySaver, call 2 cannot recall the name.

from langchain_core.tools import tool
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver

@tool
def check_balance(account_id: str) -> str:
    """Return the fake available balance for an account."""
    balances = {"NB-10000001": 84250.00, "NB-10000002": 1200.50}
    bal = balances.get(account_id, 0.0)
    return f"Available balance for {account_id}: ₹{bal:,.2f}"

@tool
def get_interest_rate() -> str:
    """Return NovaBank's current savings interest rate."""
    return "Current savings rate: 6.5% p.a."

memory_agent = create_agent(
    model="openai:gpt-5-mini",
    tools=[check_balance, get_interest_rate],
    checkpointer=InMemorySaver(),
    system_prompt="You are NovaBank's personal assistant.",
    name="novabank-memory",
)
config = {"configurable": {"thread_id": "nb-meera-01"}}

memory_agent.invoke({"messages": [("user", "Hi, my name is Meera.")]}, config=config)
result = memory_agent.invoke(
    {"messages": [("user", "What's my name? Also check NB-10000001.")]},
    config=config,
)
print(result["messages"][-1].content)

B1.2 — Per-run context

runtime.context is injected because of the ToolRuntime annotation — it never appears in the tool schema the model sees. Pass a fresh dataclass on every invoke.

from dataclasses import dataclass
from langchain.tools import ToolRuntime, tool as tool_rt

@dataclass
class BankContext:
    account_tier: str  # "standard" | "premium"

@tool_rt
def greet_customer(runtime: ToolRuntime) -> str:
    """Return a greeting that depends on the caller's account tier."""
    if runtime.context.account_tier == "premium":
        return "Welcome back, NovaBank Premium. A relationship manager is on standby."
    return "Welcome to NovaBank. How can I help today?"

ctx_agent = create_agent(
    model="openai:gpt-5-mini",
    tools=[greet_customer],
    context_schema=BankContext,
    system_prompt="Always call greet_customer before answering.",
)

r_std = ctx_agent.invoke({"messages": [("user", "Hello")]}, context=BankContext(account_tier="standard"))
r_prem = ctx_agent.invoke({"messages": [("user", "Hello")]}, context=BankContext(account_tier="premium"))
print("standard:", r_std["messages"][-1].content)
print("premium: ", r_prem["messages"][-1].content)

B2.1 — HITL with an edit resume

Interrupt, then resume on the same thread with Command. Confirm the ToolMessage used ₹5,000, not ₹50,000. Simulator on the Practice tab walks the four decisions without an API key.

from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.types import Command

@tool
def transfer_funds(to_account: str, amount: float) -> str:
    """Transfer funds to another NovaBank account. Consequential — requires HITL."""
    return f"Transferred ₹{amount:,.2f} to {to_account}."

hitl_agent = create_agent(
    model="openai:gpt-5-mini",
    tools=[transfer_funds],
    middleware=[HumanInTheLoopMiddleware(
        interrupt_on={"transfer_funds": {
            "allowed_decisions": ["approve", "edit", "reject", "respond"]
        }}
    )],
    checkpointer=InMemorySaver(),
)
cfg = {"configurable": {"thread_id": "nb-hitl-edit"}}

paused = hitl_agent.invoke(
    {"messages": [("user", "Transfer 50000 to NB-20000002")]},
    config=cfg,
)
print("interrupt payload:", paused)

final = hitl_agent.invoke(
    Command(resume={"decisions": [{
        "type": "edit",
        "edited_action": {
            "name": "transfer_funds",
            "args": {"to_account": "NB-20000002", "amount": 5000.0},
        },
    }]}),
    config=cfg,
)
for m in final["messages"]:
    if getattr(m, "type", None) == "tool":
        print("tool result:", m.content)

B2.2 — Custom PII detector

from langchain.agents.middleware import PIIMiddleware

pii_agent = create_agent(
    model="openai:gpt-5-mini",
    tools=[],
    middleware=[PIIMiddleware(
        "novabank_account",
        detector=r"NB-\d{8}",
        strategy="mask",
        apply_to_input=True,
    )],
)
# "Please look up NB-12345678" is masked before the model sees it.
result = pii_agent.invoke({"messages": [("user", "Please look up NB-12345678")]})
print(result["messages"][-1].content)

A callable detector is also valid if you need span offsets; a regex string is enough for this pattern. Use the Practice tab to see mask vs redact vs block on the same input.

B2.3 — Retry a flaky balance API

Print inside the tool so you can see attempt 1 fail and attempt 2 succeed. Run the cell several times.

import random
from langchain.agents.middleware import ToolRetryMiddleware

@tool
def check_live_balance(account_id: str) -> str:
    """Hit a flaky core-banking API (~50% ConnectionError)."""
    print(f"attempt for {account_id}")
    if random.random() < 0.5:
        raise ConnectionError("core-banking timeout")
    return f"{account_id} balance: ₹12,400.00"

retry_agent = create_agent(
    model="openai:gpt-5-mini",
    tools=[check_live_balance],
    middleware=[ToolRetryMiddleware(
        max_retries=4,
        retry_on=(ConnectionError,),
        initial_delay=0.2,
        backoff_factor=2.0,
        jitter=False,
        on_failure="continue",
    )],
)
print(retry_agent.invoke({"messages": [("user", "Live balance for NB-10000001")]})["messages"][-1].content)

B3.1 — Three limit/context middleware together

Each solves a different failure mode at the same time: history growth, runaway transfers, runaway model spend.

from langchain.agents.middleware import (
    SummarizationMiddleware,
    ToolCallLimitMiddleware,
    ModelCallLimitMiddleware,
)

limited = create_agent(
    model="openai:gpt-5-mini",
    tools=[check_balance, transfer_funds, get_interest_rate],
    checkpointer=InMemorySaver(),
    middleware=[
        SummarizationMiddleware(model="openai:gpt-5-mini"),
        ToolCallLimitMiddleware(tool_name="transfer_funds", run_limit=1, thread_limit=3),
        ModelCallLimitMiddleware(run_limit=8, thread_limit=40),
    ],
    name="novabank-limited",
)
cfg = {"configurable": {"thread_id": "nb-limits"}}
print(limited.invoke({"messages": [("user", "What is the savings rate?")]}, config=cfg)["messages"][-1].content)

B3.2 — Tool selection at scale

Assignment forbids custom wraps, so we do not ship the class show_tools debug hook as the official answer. Prove the subset from which tools were actually called (and LangSmith if you have it). always_include=["check_balance"] must survive every filter.

from langchain.agents.middleware import LLMToolSelectorMiddleware

@tool
def list_branches(city: str) -> str:
    """Locate NovaBank branches in a city."""
    return f"Branches in {city}: Bandra West, Andheri East."

@tool
def loan_info(loan_type: str) -> str:
    """Summarize a retail loan product."""
    return f"{loan_type} loan: 9.25% p.a. floating, subject to KYC."

@tool
def open_support_ticket(topic: str) -> str:
    """Create a support ticket."""
    return f"Ticket opened for {topic}."

selector_agent = create_agent(
    model="openai:gpt-5-mini",
    tools=[
        check_balance, get_interest_rate, transfer_funds,
        list_branches, loan_info, open_support_ticket,
    ],
    middleware=[LLMToolSelectorMiddleware(
        model="openai:gpt-5-mini",   # explicit — don't inherit a surprise default
        max_tools=3,
        always_include=["check_balance"],
    )],
)
result = selector_agent.invoke({"messages": [("user", "What's my balance on NB-10000001?")]})
print([getattr(m, "name", m.type) for m in result["messages"]])

B3.3 — ToolError outer, ToolRetry inner

Bad input always raises ValueError. Retry it (to show retries are wasted on non-transient errors), then re-raise into ToolError so the model never sees the raw Python text.

from langchain.agents.middleware import ToolErrorMiddleware, ToolRetryMiddleware

@tool
def close_account(account_id: str, confirm: str) -> str:
    """Close an account. Raises ValueError on a bad confirmation phrase."""
    if confirm != "CLOSE":
        raise ValueError(f"Refusing close: got {confirm!r} — internals must not leak")
    return f"Account {account_id} closed."

def safe_on_error(exc: Exception, request) -> str:
    return "That request could not be completed. Re-check the confirmation phrase."

safe_agent = create_agent(
    model="openai:gpt-5-mini",
    tools=[close_account],
    middleware=[
        ToolErrorMiddleware(on_error=safe_on_error),   # OUTER
        ToolRetryMiddleware(                           # INNER
            retry_on=(ValueError,),
            max_retries=2,
            on_failure="error",
            initial_delay=0.05,
            jitter=False,
        ),
    ],
)
result = safe_agent.invoke({"messages": [("user", "Close NB-10000001, confirm=yes")]})
text = " ".join(str(getattr(m, "content", "")) for m in result["messages"])
assert "Refusing close" not in text
assert "internals must not leak" not in text
print(result["messages"][-1].content)

One coherent NovaBank agent

Design (markdown cell, before code)

Six required pieces, all built-in:

1. response_format=TransferRequest with amount constrained gt=0, le=100000.
2. Three tools: check_balance, transfer_funds, get_interest_rate.
3. HumanInTheLoopMiddleware on transfer_funds.
4. PIIMiddleware masking NB-######## on input, output, and tool results.
5. ToolRetryMiddleware on the balance tool path (transient ConnectionError).
6. ToolCallLimitMiddleware tight on transfers + checkpointer / thread_id + context_schema for tier.

HITL already forces a checkpointer, so short-term memory is not an extra object — it is the same InMemorySaver.

from dataclasses import dataclass
from pydantic import BaseModel, Field
from langchain.agents.middleware import (
    HumanInTheLoopMiddleware, PIIMiddleware,
    ToolRetryMiddleware, ToolCallLimitMiddleware,
)

class TransferRequest(BaseModel):
    customer_name: str
    to_account: str
    amount: float = Field(gt=0, le=100000)

@dataclass
class BankContext:
    account_tier: str

capstone = create_agent(
    model="openai:gpt-5-mini",
    tools=[check_balance, transfer_funds, get_interest_rate],
    system_prompt="You are NovaBank's assistant. Mask nothing you weren't given; use tools.",
    context_schema=BankContext,
    response_format=TransferRequest,
    checkpointer=InMemorySaver(),
    name="novabank-capstone",
    middleware=[
        PIIMiddleware(
            "novabank_account", detector=r"NB-\d{8}", strategy="mask",
            apply_to_input=True, apply_to_output=True, apply_to_tool_results=True,
        ),
        ToolCallLimitMiddleware(tool_name="transfer_funds", run_limit=1, thread_limit=3),
        HumanInTheLoopMiddleware(interrupt_on={"transfer_funds": True}),
        ToolRetryMiddleware(retry_on=(ConnectionError,), max_retries=3, on_failure="continue"),
    ],
)
cfg = {"configurable": {"thread_id": "nb-capstone"}}
ctx = BankContext(account_tier="premium")

# Call 1 — short-term memory + structured extract (no transfer yet)
r1 = capstone.invoke(
    {"messages": [("user", "I'm Meera. What's the savings rate?")]},
    config=cfg, context=ctx,
)
print("call 1:", r1["messages"][-1].content)

# Call 2 — same thread should still know Meera; transfer will HITL-pause
r2 = capstone.invoke(
    {"messages": [("user", "Send 1500 to NB-20000002 from my account.")]},
    config=cfg, context=ctx,
)
print("call 2 (expect interrupt):", r2)

Reflection (3–5 sentences)

Simulators & quizzes

HITL — four decisions

PII — NB-########

Retry / error onion

Quick checks

Capstone stack checklist

Toggle at least six. This is a design aid — Q15 has no single correct list.