LLM Agents from First Principles

Agents
LLM
Machine Learning
Author

Ravi Kalia

Published

August 2, 2026

LLM Agents from First Principles

Strip the frameworks away and an LLM agent is a while loop with exactly one non-deterministic line in it. Everything else — the message history, the tool dispatch table, the regex that reads the model’s output, the counters that stop it running forever — is ordinary Python with no learned parameters and no surprises. This post builds that loop in one standard-library file, agent.py, and runs it live against a real model. The code below is imported, not retyped, and every transcript in this post was produced by the render that built the page.

An agent is a policy, a loop, and nothing else

Three words carry the whole design. An action is something the agent can do: call a tool, or stop and answer. An observation is what comes back from the world when it acts. A policy maps the history so far to the next action — and that is the job we hand to the language model. The model is not “the agent”; it is one function inside it, sampled once per iteration, whose output is a string that we then have to interpret as an action.

That framing tells you exactly where to look for the interesting failures. A policy that emits an action we cannot parse, a tool that throws, a loop that never terminates: all of these are scaffold problems, and the scaffold is code we write. Start with the one piece that isn’t.

The only stochastic line in the file

Here is the entire connection to the model — an HTTP POST and a dictionary lookup:

@dataclass
class GroqBackend:
    """OpenAI-compatible chat completions against Groq's free tier.

    The API key is read from the environment, never passed in and never
    written down. This is the only object in the module that touches the
    network, and the only source of randomness in the whole agent.
    """

    model: str = DEFAULT_MODEL
    calls: int = 0  # how many times the policy was sampled

    def create(
        self, messages: list[dict], max_tokens: int = 512, temperature: float = 0.0
    ) -> str:
        key = os.environ.get("GROQ_API_KEY")
        if not key:
            raise RuntimeError(
                "GROQ_API_KEY is not set. Get a free key at "
                "https://console.groq.com/keys and export it before running."
            )
        payload = json.dumps(
            {
                "model": self.model,
                "messages": messages,
                "max_tokens": max_tokens,
                "temperature": temperature,
            }
        ).encode()
        request = urllib.request.Request(
            GROQ_URL,
            data=payload,
            headers={
                "Authorization": f"Bearer {key}",
                "Content-Type": "application/json",
                # Without this the edge rejects the request with a bare 403
                # (Cloudflare 1010): the stdlib's default `Python-urllib/3.x`
                # agent string is on a blocklist. `requests` gets away with it
                # only because it sends a UA the blocklist doesn't cover.
                "User-Agent": "llm-agents-from-first-principles/1.0",
            },
        )
        try:
            with urllib.request.urlopen(request, timeout=90) as response:
                body = json.load(response)
        except urllib.error.HTTPError as exc:  # surface the API's own message
            raise RuntimeError(f"Groq {exc.code}: {exc.read().decode()[:400]}") from exc
        self.calls += 1
        return body["choices"][0]["message"]["content"].strip()

The policy behind it is Llama 3.3 70B Instruct, trained and released by Meta and served here through Groq’s free tier, which is why this post costs nothing to reproduce. It is worth being as careful about where a policy comes from as about where a dataset comes from. This one was pretrained on a large undisclosed web corpus and then instruction-tuned to follow directions in a chat format — that tuning is the only reason the format contract below works at all, and it is a statistical tendency, not a guarantee. What we are asking of it is narrow: given a question and a list of tools, emit the next action. The consequence of a wrong action here is a wrong arithmetic answer in a blog post; the same loop wired to a tool that spends money or writes to a database inherits every one of the failure modes shown further down, with real costs attached.

Note what the signature does not contain: no key, no retry policy, no state. create takes messages and returns a string. Swap it for any other function with that shape — a local model, a different provider, a stub that returns canned text — and nothing else in the file changes. That is the seam the rest of the design hangs from, so the next question is what we put on the message list.

State the model sees, and state it doesn’t

Agents have two kinds of state and conflating them is the usual source of confusion. Transcript is the dialogue: literally the list of messages the model conditions on. AgentState is the scaffold’s own bookkeeping — how many turns have elapsed, how much budget is left — which the model never sees except as one sentence of prompt.

@dataclass
class Transcript:
    """Dialogue state -- the exact message list handed to the model."""

    messages: list[dict] = field(default_factory=list)

    def user(self, content: str) -> None:
        self.messages.append({"role": "user", "content": content})

    def assistant(self, content: str) -> None:
        self.messages.append({"role": "assistant", "content": content})

    def observation(self, content: str) -> None:
        # Tool results re-enter as user turns: the chat schema has no separate
        # "environment" role, so the observation is framed as one.
        self.messages.append({"role": "user", "content": f"OBSERVATION: {content}"})

    def as_messages(self) -> list[dict]:
        return list(self.messages)

@dataclass
class AgentState:
    """Scaffold state. The model never sees this except as a budget line."""

    turn: int = 0
    max_turns: int = 6
    tool_calls: int = 0
    max_tool_calls: int = 4
    memory: dict[str, Any] = field(default_factory=dict)

    def budget_line(self) -> str:
        left = self.max_turns - self.turn
        return f"You have {left} turn(s) left. Finish before they run out."

The chat schema has no role for “the environment”, so a tool result has to re-enter as a user turn labelled OBSERVATION:. That is a workaround, not a principle, and it is worth knowing it’s there. The memory dict is the seed of everything an agent framework would later grow: right now nothing reads it back.

Tool errors are observations, not exceptions

A tool is a Python function plus a sentence describing it, because the model can only choose a tool it has been told about. The registry holds both, and its one real design decision is in execute:

@dataclass
class Tool:
    """A callable the model may invoke, plus the text describing it."""

    name: str
    description: str
    args: dict[str, str]  # arg name -> type, purely for the prompt
    fn: Callable[..., str]

    def spec(self) -> str:
        signature = ", ".join(f"{k}: {v}" for k, v in self.args.items())
        return f"  {self.name}({signature}) -- {self.description}"

@dataclass
class ToolRegistry:
    """Holds the tools and runs them. Never raises into the agent loop.

    Every failure mode -- unknown tool name, wrong argument names, an
    exception inside the tool itself -- comes back as an observation string.
    A crashed tool is information the policy can act on, not a bug in the
    scaffold, so the loop keeps running and the model gets told what broke.
    """

    tools: dict[str, Tool] = field(default_factory=dict)

    def add(self, tool: Tool) -> "ToolRegistry":
        self.tools[tool.name] = tool
        return self

    def specs(self) -> str:
        if not self.tools:
            return "  (none)"
        return "\n".join(t.spec() for t in self.tools.values())

    def execute(self, name: str, args: dict) -> str:
        tool = self.tools.get(name)
        if tool is None:
            known = ", ".join(self.tools) or "none"
            return f"ERROR: no tool named {name!r}. Available tools: {known}."
        try:
            return str(tool.fn(**args))
        except TypeError as exc:
            expected = ", ".join(tool.args) or "no arguments"
            return f"ERROR: bad arguments for {name} (expects {expected}): {exc}"
        except Exception as exc:
            return f"ERROR: {name} raised {type(exc).__name__}: {exc}"

Nothing in there raises. An unknown tool name, wrong argument names, an exception thrown inside the tool itself — all three are caught and returned as an observation string. This is the single most important line of judgement in the file. A crashed tool is not a bug in the scaffold; it is information the policy can act on, and handing it back as text gives the model a chance to correct itself on the next turn. Raise instead, and one malformed argument kills the whole run.

The two tools the examples use are unremarkable, except that calculator parses to an AST and walks a whitelist rather than calling eval — the model writes the expression, so the expression is untrusted input. Note that the whitelist alone is not enough: it blocks arbitrary code, but 9**9**9 is four perfectly legal nodes that would hang the process, so cost needs bounding separately.

def _eval_node(node: ast.AST) -> float:
    """Evaluate a whitelisted arithmetic AST node."""
    if isinstance(node, ast.Expression):
        return _eval_node(node.body)
    if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)):
        return node.value
    if isinstance(node, ast.BinOp) and type(node.op) in _BINOPS:
        left, right = _eval_node(node.left), _eval_node(node.right)
        if isinstance(node.op, ast.Pow) and abs(right) > _MAX_EXPONENT:
            raise ValueError(f"exponent {right} exceeds the limit of {_MAX_EXPONENT}")
        return _BINOPS[type(node.op)](left, right)
    if isinstance(node, ast.UnaryOp) and type(node.op) in _UNARYOPS:
        return _UNARYOPS[type(node.op)](_eval_node(node.operand))
    raise ValueError(f"unsupported syntax: {type(node).__name__}")

def calculator(expr: str) -> str:
    """Evaluate an arithmetic expression exactly, without `eval`."""
    # `^` means xor in Python but exponentiation in maths notation, and models
    # write the maths one. Translate it rather than failing on it.
    cleaned = expr.replace("^", "**")
    value = _eval_node(ast.parse(cleaned, mode="eval"))
    # `/` always yields a float, so an exact result reads as "2341.0" and the
    # model then feeds that string to a tool wanting an integer. Narrow it.
    if isinstance(value, float) and value.is_integer():
        value = int(value)
    return f"{expr} = {value}"

def is_prime(n: int) -> str:
    """Trial-divide to decide primality."""
    n = int(n)
    if n < 2:
        return f"{n} is not prime"
    factor = next((d for d in range(2, int(n**0.5) + 1) if n % d == 0), None)
    if factor is None:
        return f"{n} is prime"
    return f"{n} is not prime (divisible by {factor})"

Both tools return strings, because a string is what has to go back into the transcript. That leaves the harder direction: turning the model’s string into a typed action.

Parsing is where the contract is actually enforced

The prompt asks for one of two shapes. Nothing makes the model comply, so parse is where we find out:

def parse(text: str) -> Action:
    """Extract a tool call or a final answer from raw model output.

    This is where the contract with the model is actually enforced. The model
    was *asked* for a format; nothing guarantees it complied. A violation is
    an expected event, so it returns a ParseError value rather than raising --
    the loop feeds it back and lets the policy correct itself.
    """
    thought_match = _THOUGHT_RE.search(text)
    thought = thought_match.group(1).strip() if thought_match else ""

    action_match = _ACTION_RE.search(text)
    if action_match and action_match.group(1).strip().lower() in _NULL_ACTIONS:
        action_match = None

    if action_match:
        input_match = _INPUT_RE.search(text)
        if not input_match:
            return ParseError(
                "ACTION was given without a valid ACTION_INPUT line. "
                "ACTION_INPUT must be a single-line JSON object."
            )
        try:
            args = json.loads(input_match.group(1))
        except json.JSONDecodeError as exc:
            return ParseError(f"ACTION_INPUT was not valid JSON: {exc}")
        if not isinstance(args, dict):
            return ParseError("ACTION_INPUT must be a JSON object, not a scalar.")
        return ToolCall(name=action_match.group(1), args=args, thought=thought)

    final_match = _FINAL_RE.search(text)
    if final_match:
        return Final(answer=final_match.group(1).strip(), thought=thought)

    return ParseError(
        "Output matched neither form. Emit either ACTION plus ACTION_INPUT, "
        "or FINAL followed by your answer."
    )

The _NULL_ACTIONS set is not defensive programming for its own sake — it is a bug this post actually hit. Told to reply in “exactly one of two forms”, Llama 3.3 routinely emits both, filling the unused action slot with ACTION: None beside a perfectly good FINAL: line. The first version of this loop read that as a malformed tool call and burned four of its six turns arguing with the model before it gave in. Treating the null word as “no action” fixed it, and the prompt now says explicitly not to write the line. Both halves were needed.

Note that a genuine violation returns a ParseError value rather than raising, for the same reason tool failures do: in a stochastic system, malformed output is a normal event on the happy path, not an exception.

The loop

render assembles the prompt — protocol, tool specs, remaining budget, then the transcript. It is a pure function of state, which means the prompt the model sees is fully reconstructible from what the scaffold holds:

def render(
    transcript: Transcript, state: AgentState, registry: ToolRegistry
) -> list[dict]:
    """Assemble the full message list from transcript + state + tool specs."""
    system = PROTOCOL.format(tools=registry.specs(), budget=state.budget_line())
    return [{"role": "system", "content": system}, *transcript.as_messages()]

And the loop itself. Read it as a while with two guards, one sampled line, and a three-way branch:

@dataclass
class Agent:
    """Deterministic control flow around one stochastic call."""

    backend: Backend
    registry: ToolRegistry = field(default_factory=ToolRegistry)
    max_turns: int = 6
    max_tool_calls: int = 4
    temperature: float = 0.0

    def run(self, question: str) -> Trace:
        state = AgentState(max_turns=self.max_turns, max_tool_calls=self.max_tool_calls)
        transcript = Transcript()
        transcript.user(question)
        trace = Trace(question=question)

        while state.turn < state.max_turns:
            state.turn += 1

            # The one stochastic line in this function.
            raw = self.backend.create(
                render(transcript, state, self.registry),
                temperature=self.temperature,
            )
            transcript.assistant(raw)
            action = parse(raw)
            step = Step(turn=state.turn, raw=raw, action=action)
            trace.steps.append(step)

            if isinstance(action, Final):
                trace.answer = action.answer
                trace.stop_reason = "answered"
                return trace

            if isinstance(action, ParseError):
                step.observation = f"ERROR: {action.reason}"
                transcript.observation(step.observation)
                continue

            if state.tool_calls >= state.max_tool_calls:
                step.observation = (
                    "ERROR: tool call budget exhausted. Answer with FINAL now."
                )
                transcript.observation(step.observation)
                continue

            state.tool_calls += 1
            step.observation = self.registry.execute(action.name, action.args)
            state.memory[f"turn{state.turn}:{action.name}"] = step.observation
            transcript.observation(step.observation)

        trace.stop_reason = "turn budget exhausted"
        return trace

That is the whole agent. Increment the turn, sample the policy, parse, and either return, feed back an error, or run a tool and feed back its output. The turn and tool-call budgets are what make termination a property of the code rather than a hope about the model’s behaviour.

Example 1: the loop terminates without touching a tool

The first question needs no tools at all, which tests that the bare loop can stop. It is deliberately arithmetic the model can do in its head:

from agent import Agent, GroqBackend, default_registry

Agent(backend=GroqBackend()).run("What is (37 * 4 - 9) / 5?").show()
QUESTION: What is (37 * 4 - 9) / 5?

--- turn 1 | model output ---
THOUGHT: To find the result of the given expression, I need to follow the order of operations: first multiplication, then subtraction, and finally division.
ACTION: None
FINAL: (37 * 4 - 9) / 5 = (148 - 9) / 5 = 139 / 5 = 27.8

ANSWER (answered): (37 * 4 - 9) / 5 = (148 - 9) / 5 = 139 / 5 = 27.8

One turn, no tools registered, no tool called. The answer is right — 139 / 5 really is 27.8 — and notice the ACTION: None line sitting in the raw output exactly as described above, silently absorbed by the parser instead of costing a round trip. This is the trajectory that shows how little an “agent” is doing when the policy already knows the answer: one sample, one parse, return. Nothing tests the tool path until the question is one the model cannot answer alone.

Example 2: the round trip, twice

Now the same loop with both tools registered, and a question whose answer the model has no way to know: whether a number it has not yet computed is prime.

Agent(backend=GroqBackend(), registry=default_registry()).run(
    "Use the calculator tool to compute (128^2 + 3) / 7, "
    "then tell me whether the result is prime."
).show()
QUESTION: Use the calculator tool to compute (128^2 + 3) / 7, then tell me whether the result is prime.

--- turn 1 | model output ---
THOUGHT: First, we need to compute the value of (128^2 + 3) / 7 using the calculator tool.
ACTION: calculator
ACTION_INPUT: {"expr": "(128^2 + 3) / 7"}
--- turn 1 | observation ---
(128^2 + 3) / 7 = 2341

--- turn 2 | model output ---
THOUGHT: Now that we have the result of the calculation, we need to check whether 2341 is a prime number using the is_prime tool.
ACTION: is_prime
ACTION_INPUT: {"n": 2341}
--- turn 2 | observation ---
2341 is prime

--- turn 3 | model output ---
THOUGHT: We have now determined that the result of the calculation, 2341, is indeed a prime number.
FINAL: 2341 is prime

ANSWER (answered): 2341 is prime

Three turns, two complete round trips. Turn one: the model emits ACTION: calculator with a JSON argument, parse turns those two lines into a ToolCall, the registry dispatches to a Python function, and 16387 / 7 = 2341 goes back into the transcript as an observation. Turn two: conditioned on that observation, the policy picks a different tool — it was never told to use is_prime, it chose it — and gets 2341 is prime back. Turn three: nothing left to look up, so it stops. Both results are independently checkable, which is the point of using arithmetic: 128² + 3 = 16387, 16387 / 7 = 2341 exactly, and 2341 has no divisor below its square root.

The interesting thing is how boring the mechanism is. Between the two model calls, the only things that happened were a regex match, a dict.get, and a function call.

Caveat: temperature 0 is not determinism

Both runs above used temperature=0.0, which selects the most likely token at each step rather than sampling — but batching and floating-point non-associativity on the serving side mean identical requests can still return different text. The transcripts above are one trajectory, captured when this page was rendered, not a fixed point. Re-run them and the model may reach the same answers by a different route, call is_prime first, or skip a tool it used here. An agent’s behaviour is a distribution over trajectories; a single printed run is a sample from it, and that is precisely why the turn and tool-call budgets exist in code rather than in the prompt.

Running it yourself

agent.py imports nothing outside the standard library. Get a free key at console.groq.com/keys, then:

export GROQ_API_KEY=...          # never hardcode it; the module reads the env
python agent.py                  # runs both examples above

If the variable is unset, GroqBackend.create raises with that instruction rather than failing somewhere less obvious.

What you just ran, and what it isn’t

The claim at the top was that an agent is deterministic control flow plus one stochastic call, looped, and the two traces make the split concrete. Stochastic: four calls to create, and nothing else. Deterministic: prompt assembly, parsing, dispatch, budget accounting, termination — every one of them plain code you can step through, test without a network, and reason about the way you reason about any other program. The tool outputs are deterministic here too, though in general the environment is a second source of randomness, and a search or read_file tool would make it one.

What that buys you in practice is a debugging strategy. When an agent misbehaves, the question is always which half broke: did the policy pick a bad action, or did the scaffold mishandle a good one? The ACTION: None bug looked like a stupid model and was really a strict parser. Frameworks are worth reaching for once you need what this toy has no answer for — memory that outlives a single run, planning over horizons longer than one greedy step, several agents sharing work — but each of those is a change to the scaffold, the part with no learned parameters, and none of them changes the shape of the loop you just read.