The ReAct Agent — Production Loop Safety

Notebook Summary#

Hand-copy this. Everything here is derived from the detailed notes below.

Detailed Study Notes#

A customer writes in: "My account C1002 shows past due, but I paid last week." A classifier can tag this ticket billing and promise a reply in two hours. It cannot actually look at the account. To answer, something has to pause, go fetch the real record, read it, maybe check a second source, and only then speak. The moment you give a language model that ability, you stop building a labeler and start building an agent, and you inherit a new problem: knowing when to stop.

2.1From a classifier to an agent that acts#

Yesterday's build could only sort a ticket into one of four buckets (technical, billing, fraud, general). That is a one-shot decision. The upgrade in this lesson is a real tool-execution layer: the model can now call functions that read a mock CRM database and a troubleshooting knowledge base, then ground its answer in whatever comes back. The plain-English version: instead of a receptionist who files your complaint, you now have a support rep who can actually open your file while you wait.

Two things make that work. First, the functions are exposed as tools the model is allowed to call. Second, the graph is wired so the model can call a tool, look at the result, and decide whether to call another one, which is the loop the rest of this note is about.

2.2Anatomy of the CRM support agent#

The agent is a small graph of nodes that pass a shared state between them. The state is a typed dictionary; the key field is the message list, and it uses a reducer so new messages are appended rather than overwriting the history.

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages

class SupportState(TypedDict):
    raw_input: str
    category: str                                # technical | billing | fraud | general
    messages: Annotated[list, add_messages]      # add_messages APPENDS, never overwrites
    customer_facts: dict                         # for selective context preservation (2.10)
    iteration: int                               # loop guard (2.10)

A tool is just a decorated function with a docstring. The docstring is not decoration, it is the contract the model reads to decide when to call the tool, what arguments to pass, and what shape comes back. Notice the three layers inside: validate the argument, look it up, then filter out the noisy fields so only the relevant data returns.

from langchain_core.tools import tool

@tool
def get_customer_details(customer_id: str) -> dict:
    """Fetch billing status, tier, and recent transactions for a customer.
    Call before answering ANY billing question. IDs look like 'C1002'.
    Never guess an ID, ask the user if it is missing."""
    cid = customer_id.strip().upper()                  # layer 1: validate + normalize
    if not cid.startswith("C"):
        return {"error": "invalid customer id format"}
    raw = MOCK_CRM.get(cid)                             # layer 2: database lookup
    if raw is None:
        return {"error": "customer not found"}
    keep = {"name", "billing_status", "tier", "outstanding_balance", "recent_transactions"}
    return {k: v for k, v in raw.items() if k in keep} # layer 3: drop noisy fields

A second tool, search_knowledge_base, does keyword lookups over troubleshooting articles (no semantic/vector search, just exact keyword matches on a dict). Both tools are bound to the model with llm.bind_tools([...]), which is how the model knows they exist. The full graph then flows like this:

2.3ReAct = Reason + Act, and the scratchpad#

ReAct is a way of running an LLM where it writes its reasoning as text, then takes an action, then reads the result, and repeats. Formalized in a 2022 paper by Yao and colleagues at Princeton and Google, the finding was simple: a model that interleaves reasoning traces with tool calls beats both a pure-reasoning model and a pure-tool-using model, on tasks that genuinely need several steps.

Why write the reasoning down at all? Because a transformer is stateless. It processes a sequence and predicts the next token; each token is conditioned on the tokens before it in the context window. There is no separate working-memory register, no private notepad. So the only mechanism the model has to "think" is to emit its thoughts as tokens into the context before it acts. Those tokens become the scratchpad.

2.4The core loop: Thought → Action → Observation#

Every ReAct agent cycles through three phases until it decides it's done.

The observation phase matters more than it looks. By forcing the raw result back through a fresh Thought before answering, the loop stops the agent from short-circuiting, jumping to a conclusion off a single data point. Termination is decided by the model itself, not a fixed counter: when the last message carries no tool calls, the agent has produced a final answer.

2.5When to reach for React (and when not to)#

React is not automatically better. For a task the model can answer from its own knowledge, or with a single lookup, the loop just adds latency, context, and cost.

2.6From a DAG to a cycle: the conditional edge#

The classifier graph you started with was a DAG, a directed acyclic graph: a fixed sequence of nodes joined by deterministic edges, with a path decided at design time and no way to revisit a node. A React graph is fundamentally different: it contains a cycle. The agent node can route to the tool node, which routes back to the agent node, as many times as needed. In graph terms you've turned an acyclic graph into a cyclic state machine.

LangGraph is built to make that conversion safe: it offers recursion limits, state accumulation, and cycle detection precisely because cycles can misbehave (2.10). After every agent execution the graph asks one question, did the model ask for a tool, or give a final answer?, and the conditional edge routes on the answer.

2.7The router stays structurally pure#

The routing function is a plain Python function: it takes the current state, inspects the shape of the last message, and returns the name of the next node as a string. It reads whether tool calls exist; it does not read what they say.

def route_after_agent(state: SupportState) -> str:
    """Pure router: inspect the LAST message, return the next node.
    The model already decided by emitting (or not emitting) tool calls."""
    last = state["messages"][-1]
    if getattr(last, "tool_calls", None):   # non-empty -> model wants a tool
        return "tools"
    return "respond"                        # no tool calls -> it's the final answer

2.8State accumulates: the message ledger#

Each pass through the loop appends to the message list. Because the field uses the add_messages reducer, nothing is overwritten, the list grows into a full record of the conversation and the reasoning history.

When the agent node runs a second time, it receives the entire prior context, the earlier thought and the tool's observation included. That's the whole mechanism by which observations feed back into reasoning: there is no special API, just message accumulation in the context window. It's elegant, and it's also the seed of the next two problems.

2.9Context bloat, and three ways to tame it#

The most dangerous failure of a React agent isn't a crash, it's a slow degradation. Each iteration adds tokens (a thought is ~50–150, a tool call ~30–80, an observation ~50–500). Five iterations might only reach ~3,000 tokens, which is nothing. But a realistic enterprise ticket can need 15–20 tool calls, and a gnarly billing dispute can push context past 50,000 tokens. At that scale three problems emerge.

Once context is already huge, you often can't recover without starting over, so the real move is to manage growth as it happens. Three strategies, from crudest to most surgical:

Strategy 1 transforms the view the model sees without mutating stored state, the full history is retained, but the model receives a trimmed slice:

from langchain_core.messages import trim_messages

def prepare_messages(state: SupportState) -> list:
    return trim_messages(
        state["messages"],
        max_tokens=2000,
        strategy="last",        # rolling window: keep the most recent
        token_counter=llm,
        include_system=True,    # never drop the system prompt
        start_on="human",       # keep the original user turn intact
    )

Strategy 3 is the one worth the extra code. Instead of trusting a trim or a summary to keep what matters, you explicitly pull the critical fields out of each observation and store them in a queryable state field. The message list can then be cut hard, the facts survive independently:

def capture_facts(state: SupportState, observation: dict) -> dict:
    facts = dict(state.get("customer_facts", {}))
    for key in ("billing_status", "outstanding_balance", "tier"):
        if key in observation:
            facts[key] = observation[key]          # exact values, not prose
    return {"customer_facts": facts}               # merged back into state

2.10Keeping the loop bounded#

A cyclic graph can loop forever. It happens for three reasons: a tool-failure cascade (a tool keeps erroring and the model keeps retrying it), oscillating reasoning (the model alternates between two hypotheses and never commits), or context corruption (accumulated observations now contradict each other). A stuck agent is expensive, thousands of wasted tokens per run, so production agents layer several guards.

The guards work at two levels. The recursion limit is an infra-level backstop set at compile/invoke time; keep it low so a runaway graph crashes fast against it rather than burning minutes. The iteration count is application-level and smarter, when it trips, it produces a graceful, user-friendly message instead of a 500 error.

graph = builder.compile()
result = graph.invoke(initial_state, config={"recursion_limit": 15})  # infra backstop

import hashlib, json

def call_fingerprint(name: str, args: dict) -> str:
    payload = name + json.dumps(args, sort_keys=True)   # sort so arg order can't fool it
    return hashlib.sha256(payload.encode()).hexdigest()

def is_repeat_call(state: SupportState, name: str, args: dict) -> bool:
    return call_fingerprint(name, args) in state["seen_calls"]  # seen_calls: set[str]

2.11Graceful escalation to a human#

When a breaker fires, the agent must not just throw an error. The production requirement is a graceful degradation: a message that tells the user what was tried and hands the next human a briefing. The template has four parts and is hardcoded, not generated.

ESCALATION = (
    "I've logged your request about {topic}.\n"          # 1. acknowledge the request
    "Here's what I checked: {findings}.\n"               # 2. what was gathered
    "I couldn't fully resolve this automatically ({reason}), "   # 3. couldn't resolve + why
    "so I've routed it to a specialist. "
    "Someone will follow up within {sla}."               # 4. human follow-up + timeframe
)

2.12Interview & practice Q&A#

2.13Key takeaways#

React earns its complexity only on multi-step, multi-tool, self-correcting tasks; for a lookup, skip it. The loop is Thought → Action → Observation, powered entirely by message accumulation, which is also why the two hard engineering problems are context growth and termination. Tame context with trimming, summarization, or, best, selective fact extraction. Bound the loop with a low recursion limit, an application iteration cap, and call-fingerprinting, then fail gracefully into a hardcoded human hand-off. And keep the router pure: it inspects message shape, never business logic.

Microexercises#

Reading proves you recognise it. These prove you can use it. Attempt each before opening the walkthrough.

3.1Diagnose it: the impure router#

def route_after_agent(state):
    category = state["category"]
    if category == "billing":
        return "tools"          # billing always needs the CRM tool... right?
    return "respond"

def route_after_agent(state):
    last = state["messages"][-1]
    if getattr(last, "tool_calls", None):   # any tool, any category
        return "tools"
    return "respond"

3.2Build it: the repeated-call guard#

import hashlib, json

def fingerprint(name, args):
    payload = name + json.dumps(args, sort_keys=True)
    return hashlib.sha256(payload.encode()).hexdigest()

def guard(seen: set, name, args) -> str:
    fp = fingerprint(name, args)
    if fp in seen:
        return "DUPLICATE"        # short-circuit: same call, same result
    seen.add(fp)
    return "OK"

seen = set()
guard(seen, "get_customer_details", {"id": "C1", "x": 1})   # -> "OK"
guard(seen, "get_customer_details", {"x": 1, "id": "C1"})   # -> "DUPLICATE"