LLM Agents from First Principles

Agents
LLM
Machine Learning
Author

Ravi Kalia

Published

August 2, 2026

LLM Agents from First Principles

An LLM agent is a while loop with one stochastic step per iteration. Everything else — message history, tool dispatch, output parsing, turn budgets — is deterministic Python with no learned parameters.

This post builds that loop in agent.py, imports it below (not retyped), and runs it live against a real model. Transcripts were captured at render time.

1 Agent components

Three terms define the design:

  • Action — call a tool, or stop and answer.
  • Observation — what the environment returns after an action.
  • Policy — maps history to the next action; here, the language model.

The model is not the agent. It is one function sampled once per iteration; its output is a string that must be parsed into an action.

Failure modes to expect: unparseable policy output, tool exceptions, non-terminating loops. All are scaffold problems — code you write, not model weights.

2 Model backend

The only stochastic connection is 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()

Policy: Llama 3.3 70B Instruct (Meta), served via Groq free tier.

Provenance: Pretrained on a large undisclosed web corpus; instruction-tuned for chat-format following.

Task in this post: Given a question and tool list, emit the next action.

Downstream impact here: Wrong action → wrong arithmetic in a blog post. The same loop wired to spending or database tools inherits these failure modes with real cost.

Why this policy: Instruction tuning makes the format contract below work as a statistical tendency, not a guarantee.

create(messages) -> str has no key, retry policy, or state. Swap in any function with that signature — local model, other provider, stub — and the rest of the file is unchanged.

3 State

Two state kinds must stay separate:

  • Transcript — dialogue; the message list the model conditions on.
  • AgentState — scaffold bookkeeping (turn count, budget); the model sees only what the prompt includes.
@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 environment role. Tool results re-enter as user turns prefixed OBSERVATION:. The memory dict is unused in this version.

4 Tool registry

A tool is a Python function plus a description string. The registry holds both; execute is the main design choice:

@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 raises. Unknown tool names, bad arguments, and in-tool exceptions all return observation strings. The policy can recover on the next turn; raising kills the run.

Example tools:

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})"
  • calculator — AST walk with a whitelist, not eval; model-written expressions are untrusted input.
  • Cost bound — whitelist blocks arbitrary code but not expensive legal expressions (e.g. 9**9**9).
  • Both tools return strings for transcript insertion.

5 Output parsing

The prompt requests one of two output shapes. Compliance is not guaranteed; parse enforces the contract:

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."
    )

_NULL_ACTIONS handles a observed bug: Llama 3.3 often emits ACTION: None alongside a valid FINAL: line. Treating null tokens as “no action” fixed wasted turns.

Genuine violations return a ParseError value, not an exception — malformed output is a normal event on the happy path.

6 Agent loop

render assembles protocol, tool specs, budget, and transcript. It is a pure function of state:

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()]

The loop:

@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

Per iteration: increment turn, sample policy, parse, then return, feed back a parse error, or run a tool and append its observation. Turn and tool-call budgets make termination a code property, not a model hope.

7 Example: direct answer, no tools

Arithmetic the model can solve without tools; tests bare loop termination:

from agent import Agent, GroqBackend, default_registry

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

Captured transcript from the original render (one sample; re-running the cell needs GROQ_API_KEY and may differ):

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. Answer: 139 / 5 = 27.8. Raw output may include ACTION: None, absorbed by the parser.

8 Example: two tool round trips

Same loop with both tools; question requires computation the model cannot know without tools:

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()

Captured transcript from the original render:

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 tool round trips:

  1. ACTION: calculator → 16387 / 7 = 2341
  2. ACTION: is_prime → 2341 is prime
  3. FINAL: — stop

Checkable facts: 128² + 3 = 16387, 16387 / 7 = 2341, 2341 is prime.

Between model calls: regex match, dict.get, function call.

9 Constraints

Both runs used temperature=0.0 (greedy decoding). Batching and floating-point non-associativity on the server can still yield different text for identical requests.

Transcripts here are one sample from a trajectory distribution, not a fixed point. Re-rendering may change route or tool order. Turn and tool-call budgets exist in code for this reason.

10 Local execution

agent.py uses only the standard library.

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

Key source: console.groq.com/keys. If unset, GroqBackend.create raises with that instruction.

11 Summary

  • Stochastic: create calls only.
  • Deterministic: prompt assembly, parsing, dispatch, budgets, termination.
  • Tool outputs: deterministic in these examples; real environments (search, files) add a second randomness source.

Debugging: When an agent misbehaves, ask whether the policy picked a bad action or the scaffold mishandled a good one. The ACTION: None case was a strict parser, not a model failure.

Frameworks add memory across runs, multi-step planning, multi-agent coordination — all scaffold changes; the loop shape stays the same.

12 References