The Agent Loop

Strip away LangChain, CrewAI, AutoGen — every agent you'll ever build is one primitive: a model in a loop with tools. Think, act, observe, repeat. Understanding it is the difference between brittle scripts that break on edge cases and systems that survive production.

Naresh Madiraju

schedule 15 Min Read

event June 2026

bolt TL;DR

At its core, an agent loop is just an application intercepting an LLM's JSON requests, running the tools, and feeding the results back. It evolves from a single ReAct chain into hierarchical setups that separate planning, doing, and double-checking. Wrapping the model in this strict, hardcoded structure is what actually saves you from common production nightmares like runaway infinite loops, goal drift, hallucinated success, and massive token-bloat.

1 · The core primitive: Think → Act → Observe

At baseline, an agent is an elegant while loop. Instead of predicting text once and stopping, the model sits inside a cycle of three phases.

Think — the model assesses the goal and history and decides the next move: "I need User X's order status, so I'll look it up."

Act — the model outputs a structured request to call a tool. The point new engineers miss: the model does not run the code. It emits a JSON request; your application intercepts it, runs the real query, and captures the result.

Observe — your application filters, prunes, and appends that result back into the model's context. Then the loop ticks. The model is the single decision-maker; tools are stateful services it consults but does not control.

2 · ReAct: the base loop

ReAct (Reasoning + Acting, Yao et al. 2022) is the foundation; the other patterns are variations. It alternates a Thought, an Action, and an Observation, looping until it emits a final answer instead of another action. Its defining trait: it plans one step at a time — adaptive, but short-sighted on long interdependent tasks.

Think of it like driving in fog. You see ten feet ahead and steer by whatever appears next — perfect when the road is unpredictable, inefficient when you already know the route.

Here's that loop in code, two ways. First, the raw mechanics under the hood — a plain Python while loop against the Anthropic SDK, so you can see think → act → observe with nothing hidden. Then the same logic mapped into a framework (LangGraph), where the loop becomes nodes and edges. Same agent, two levels of abstraction:

import anthropic, json
client = anthropic.Anthropic()          # reads ANTHROPIC_API_KEY

def get_item_price(item_name: str) -> float:   # the "Act" boundary — real code
    prices = {"laptop": 1200.0, "mouse": 25.0, "monitor": 300.0}
    return prices.get(item_name.lower(), 0.0)

TOOL_MAP = {"get_item_price": get_item_price}
tools = [{"name": "get_item_price", "description": "Get an item's price.",
          "input_schema": {"type": "object",
            "properties": {"item_name": {"type": "string"}},
            "required": ["item_name"]}}]

def run_agent(goal: str, max_iters: int = 5):
    messages = [{"role": "user", "content": goal}]   # the loop's memory
    call_history = []                                # for stuttering detection

    for i in range(1, max_iters + 1):               # CIRCUIT BREAKER 1: max iters
        resp = client.messages.create(model="claude-sonnet-4-5",
                                      max_tokens=1024, tools=tools, messages=messages)

        if resp.stop_reason != "tool_use":          # THINK: no tool -> done
            return "".join(b.text for b in resp.content if b.type == "text")

        messages.append({"role": "assistant", "content": resp.content})
        results = []
        for block in resp.content:                  # ACT
            if block.type == "tool_use":
                sig = f"{block.name}:{json.dumps(block.input, sort_keys=True)}"
                if call_history and call_history[-1] == sig:   # BREAKER 2: stuttering
                    out = "System error: repeated identical call. Change strategy."
                else:
                    call_history.append(sig)
                    try:
                        out = str(TOOL_MAP[block.name](**block.input))[:2000]  # prune
                    except Exception as e:
                        out = f"error: {e}"
                results.append({"type": "tool_result",
                                "tool_use_id": block.id, "content": out})
        messages.append({"role": "user", "content": results})   # OBSERVE

    return "halted: max iterations reached"

run_agent("I bought a laptop and two mice. How much in total?")

from typing import Literal
from typing_extensions import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool

@tool
def get_item_price(item_name: str) -> float:
    """Get the price of an inventory item."""
    prices = {"laptop": 1200.0, "mouse": 25.0, "monitor": 300.0}
    return prices.get(item_name.lower(), 0.0)

tools = [get_item_price]
by_name = {t.name: t for t in tools}
llm = init_chat_model("anthropic:claude-sonnet-4-5").bind_tools(tools)

class State(TypedDict):
    messages: Annotated[list, add_messages]   # APPENDS — this is the memory

def think(state):                              # THINK node
    return {"messages": [llm.invoke(state["messages"])]}

def act(state):                                # ACT node
    from langchain_core.messages import ToolMessage
    out = []
    for call in state["messages"][-1].tool_calls:
        result = by_name[call["name"]].invoke(call["args"])
        out.append(ToolMessage(content=str(result)[:2000],   # trim at the boundary
                               tool_call_id=call["id"]))
    return {"messages": out}

def route(state) -> Literal["act", "__end__"]:  # ROUTER: loop or stop
    last = state["messages"][-1]                  # may be a plain text turn
    return "act" if getattr(last, "tool_calls", None) else "__end__"

g = StateGraph(State)
g.add_node("think", think); g.add_node("act", act)
g.add_edge(START, "think")
g.add_conditional_edges("think", route, {"act": "act", "__end__": END})
g.add_edge("act", "think")                      # the loop
app = g.compile()
app.invoke({"messages": [("user", "A laptop and two mice. Total?")]})

Notice three things: the for range(max_iters) is the circuit breaker; the message list is the agent's memory (without appending, it forgets last turn); and the model never runs your code — it returns a request, your loop executes it. Where is route used? You never call it — you register it with add_conditional_edges, and LangGraph invokes it every time execution leaves think, routing on the string it returns. Keep routers pure: read state, return a string, no side effects.

3 · The two variants: Plan-and-Execute & Reflection

ReAct decides one move at a time — adaptive, but short-sighted on long tasks. Two variants reshape that same base loop, and in production you don't choose between them:

checklistPlan-and-Execute

Decide the whole sequence of steps up front, then execute them in order. More efficient and predictable on long, known tasks — but less adaptive when the path changes mid-run.

rate_reviewReflection

Add a self-critique pass: the agent reviews its own output and can send the work back for another attempt. Spends extra tokens to buy quality on trial-and-error work like code generation.

4 · The production ideal: layering all three patterns

The pattern-by-pattern view hides the real point: production agents are layered hybrids. A Plan-and-Execute outer loop sets the sequence; each step is a ReAct sub-agent with its own tools; a Reflection gate validates the result.

Why it works. Blast-radius isolation: a failing low-level ReAct node is trapped in local state; the outer loop catches it and re-plans without a system crash. Token economy: expensive Reflection passes run only at critical boundaries, not on every data fetch.

Pattern	Focus	Best for
ReAct	Step-by-step decisions	Exploration, unpredictable APIs
Plan-Execute	Long-horizon structure	Pipelines, batch jobs, known sequences
Reflection	Quality gate	Code generation, validation, content

5 · From theory to code: building a deterministic multi-agent topology

Deploy a flat, unconstrained ReAct loop into production and you are letting an LLM write its own control flow at runtime — which is exactly how you get the infinite loops, goal drift, and hallucinated task completion this article keeps warning about. The fix is to stop treating the agent as an operator and treat it as a deterministic state machine: decouple the work into isolated, single-responsibility nodes governed by a rigid execution graph the model cannot rewrite. For a high-stakes job like a zero-downtime database migration, that means nesting all three loops:

Outer loop — Plan-and-Execute. Owns the roadmap. It breaks the migration into isolated phases (verify prerequisites, apply the schema delta, validate row counts) and hands them out one at a time.
Inner loop — local ReAct. A worker that executes exactly one phase. It calls tools freely but cannot see or edit the global plan.
Guard loop — Reflection. Before any phase is marked complete, an independent node checks the result against strict constraints — and on failure sends the worker back to retry that same phase.

Before the code, here is how state moves through the system. The roadmap lives in an immutable array the model can't touch; each phase runs in an isolated scope; and a hard loop counter trips a circuit breaker if any phase refuses to converge:

The guard runs after every phase. A failed reflection loops back to the executor, so the plan pointer in advance never moves past a phase that hasn't passed validation.

Here is the topology in LangGraph. MigrationState tracks what is pending (plan), what is running right now (current_step), what has already passed validation (completed), and a loop_counter that arms the circuit breaker. The four tabs below are state, nodes, routing, and the graph wiring — in build order.

# The global state machine: one source of truth for the whole topology.
from typing import List, Dict, Any, Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class MigrationState(TypedDict):
    input_request: str               # the migration goal
    plan: List[str]                  # pending high-level phases
    current_step: str                # the phase running right now
    step_history: List[str]          # raw execution logs for the active phase
    completed: List[Dict[str, Any]]  # verified audit trail of finished phases
    critique: str                    # feedback from the reflection guard
    loop_counter: int                # hard counter — the circuit breaker vs runaway loops

def plan_node(state: MigrationState) -> Dict[str, Any]:
    """Outer loop: turn the request into a rigid, linear sequence of phases."""
    # In production, call the planner LLM here to emit a clean JSON list.
    plan = [
        "Phase 1: verify source connection and replication latency",
        "Phase 2: apply schema-delta scripts to the target cluster",
        "Phase 3: run the parallel data-synchronisation job",
    ]
    return {"plan": plan}

def executor_node(state: MigrationState) -> Dict[str, Any]:
    """Inner loop: run the active phase inside an isolated local ReAct loop."""
    active = state["current_step"] or state["plan"][0]
    count = state.get("loop_counter", 0) + 1
    # The local loop calls database tools freely in a bounded scope and logs
    # what it did — it never touches the global plan.
    log = f"Executed: {active}. Row counts match the target."
    return {
        "current_step": active,
        "step_history": state.get("step_history", []) + [log],
        "loop_counter": count,
    }

def reflection_guard_node(state: MigrationState) -> Dict[str, Any]:
    """Guard loop: validate the phase output against strict rules."""
    last_log = state["step_history"][-1]
    # In production, hand the log to an evaluator LLM or a deterministic checker.
    if "error" in last_log.lower() or "fail" in last_log.lower():
        return {"critique": "Validation failed: anomalous execution logs."}
    return {"critique": "PASSED"}

def advance_step_node(state: MigrationState) -> Dict[str, Any]:
    """Commit the verified phase and move the pointer to the next one."""
    completed = state.get("completed", []) + [
        {"step": state["current_step"], "log": state["step_history"][-1]}
    ]
    remaining = state["plan"][1:]
    return {
        "plan": remaining,
        "current_step": remaining[0] if remaining else "",
        "completed": completed,
        "step_history": [],   # purge per-phase logs — bounds context growth
        "critique": "",       # reset for the next cycle
        "loop_counter": 0,    # re-arm the breaker for the next phase
    }

# Routers decide control flow without mutating state.
def route_after_guard(state: MigrationState) -> Literal["advance", "executor", "abort"]:
    """Circuit breaker first, then commit-or-retry on the reflection verdict."""
    if state["loop_counter"] > 10:                  # kill a phase that won't converge
        return "abort"
    return "advance" if state["critique"] == "PASSED" else "executor"

def route_next_phase(state: MigrationState) -> Literal["executor", "__end__"]:
    """Phases left in the plan? Keep going. Otherwise terminate."""
    return "executor" if state["plan"] else "__end__"

workflow = StateGraph(MigrationState)

workflow.add_node("planner", plan_node)
workflow.add_node("executor", executor_node)
workflow.add_node("guard", reflection_guard_node)
workflow.add_node("advance", advance_step_node)

workflow.add_edge(START, "planner")
workflow.add_edge("planner", "executor")
workflow.add_edge("executor", "guard")

# Did the phase pass validation? Loop back locally if not — unless the
# circuit breaker has tripped, which aborts the whole run.
workflow.add_conditional_edges("guard", route_after_guard,
    {"advance": "advance", "executor": "executor", "abort": END})

# Any phases left in the global plan?
workflow.add_conditional_edges("advance", route_next_phase,
    {"executor": "executor", "__end__": END})

migration_agent = workflow.compile()

Why this resists production drift. If the executor fails to apply the schema scripts in phase 2, the guard flags it, writes a specific reason into critique, and routes state straight back to the executor. Because advance was never reached, the plan pointer stays locked on phase 2 — and if that phase still won't converge, the loop_counter circuit breaker aborts the run rather than spinning forever. The agent can never silently skip to phase 3 and dump raw data into an unconfigured target cluster. That is the line between flaky prompt chains and deterministic, industrial-grade agent design.

Conclusion: engineering the autonomy

The move from brittle prompt engineering to predictable software engineering means treating the agent loop not as a sentient operator, but as a standard, state-driven runtime environment.

The lesson from the early Auto-GPT and BabyAGI era is blunt: give an LLM unchecked control over its own execution graph and it will optimise for token consumption over goal completion. Wrap it in a deterministic state machine instead — strip it of state-mutation privileges, bound every loop with a hard counter, isolate tool execution, and validate each output through an independent guard node — and a fragile looping experiment becomes a reliable, enterprise-ready infrastructure component.

A useful lens: most production agent failures between 2024 and 2026 weren't model-quality failures — the model worked fine. They were architectural failures. The loop design was wrong. The model is rarely the problem; the loop is.