Skip to content

Building an Agent Loop

An agent is a program that keeps thinking until it is confident enough — or until it runs out of iterations, tokens, or money. Every harness re-implements that shape by hand: a while loop, a step counter, a spend tracker, a break-on-repeat guard. Sema makes each of those a language construct, so the loop you write is the loop that runs — nothing hidden in a framework.

This guide builds a small bounded research agent, one piece at a time, from the verified research-agent example.

Concern Construct Where
Bounded do-until loop loop until <cond> max_iters N: Control Flow
Value-returning loop std.agent_loop.loop_until std.agent_loop
Spend ceiling with budget(...) as b: Budgets & Metering
Ambient usage with meter as u: Budgets & Metering
Drift alarms monitor … on <fn>: Monitors & Drift
Calling functions as tools tools.run(...) Tools, Skills & MCP

Alongside while and for, Sema has a declarative surface for the bounded agentic loop. The body runs, then the condition is checked — it is a do-until, so it always runs at least once — and max_iters N caps the iteration count:

loop until decision.confidence >= 0.9 max_iters 8:
analysis = breakdown(query, state)
state.facts += fact_extract(search(query_gen(analysis)))
decision = decide(query, state)

break and continue work inside. Omit max_iters and the loop runs until the condition holds, under the same runaway guard as while. This replaces the hand-rolled while i < max: … if stop: break shape.

The research-agent example drives its loop from a belief — a Beta-Bernoulli posterior that updates as evidence arrives — and stops once confidence crosses a threshold, bounded by the evidence available:

from std.belief import Belief
decisions = [0.7, 0.85, 0.95]
mut belief = Belief(alpha=1.0, beta=1.0, history=[0.5])
mut iters = 0
loop until belief.confidence() >= 0.7 max_iters len(decisions):
belief.update(decisions[iters])
iters = iters + 1

When you want the loop to return a value and keep the state immutable, use the combinator std.agent_loop.loop_until. It runs a pure step: state -> state until done(state) or max_iters:

from std.agent_loop import loop_until
final = loop_until(init_state, 8, s => advance(s), s => s.confidence >= 0.9)

step/done are pure lambdas over an immutable state (Sema forbids a lambda mutating captured state), which keeps the loop parallel-safe.

An unbounded loop is dangerous even with an iteration cap, because a single iteration can call a model many times. Two ambient scopes bound spend without threading (result, usage) tuples through every call.

with meter as u: accumulates every model call’s usage inside the block into uu.total_calls, u.prompt_tokens, u.completion_tokens, u.total_tokens, and u.cost:

mut calls = 0
mut cost = 0.0
with meter as u:
_synthesis = generate("Summarize renewable energy findings", 64)
calls = u.total_calls
cost = u.cost

with budget(tokens=N, calls=M) as b: is a meter with a hard cap — a model call that would push spend past the cap raises BudgetExceeded instead of silently overspending:

with budget(calls=200, tokens=1_000_000) as b:
research = deep_search(query) # BudgetExceeded if it overspends

Meters and budgets nest; each call attributes to all enclosing frames. Wrap the whole agent loop in a budget and each iteration in a meter and you get a hard ceiling plus per-step accounting for free.

A bounded agent that quietly starts producing worse output is worse than one that crashes. A monitor attaches an anytime-valid statistical test to a function’s output stream and fires on drifted: when the distribution shifts:

monitor answer_drift on synthesize:
capture result.embedding, takeaways
baseline from assure
test conformal_martingale(alpha=0.01)
on drifted: alert("agent answers drifting from the assured profile")
on undecided: log.debug("answer monitor undecided")

The baseline comes from your verification runs (baseline from assure), the test is a conformal test martingale that bounds false-alarm probability over an unbounded horizon, and a stable stream does not raise a false alarm. See Monitors & Drift for the full model.

In Sema a function is a tool. Pass functions to tools.run and the runtime introspects each one — name, typed parameters, and a leading sem "…" as the description — into a schema, drives the loop, executes the real functions, and returns the answer plus a trace:

import tools
def get_weather(city: str) -> str !{net.connect}:
sem "Get the current weather for a city"
return fetch_weather(city)
result = tools.run("what's the weather in Berlin?", [get_weather], max_steps=6)
# result.answer, result.steps, result.status, result.trace

Because the tool is a governed Sema function, its effect row (!{net.connect}) still applies when the agent calls it — tool calling inherits the language’s governance, rather than being an ungoverned side channel. The loop is bounded by max_steps and detects same-tool-same-args spinning. The full surface — MCP servers, Markdown skills — is covered in Tools, Skills & MCP.

The research-agent example composes all of this into one small pipeline: de-duplicate candidate facts semantically, run a belief-driven loop, draft a synthesis under an ambient meter, and render a typed report.

from std.provenance import Cit, Doc, build_url_to_id, rewrite
from std.belief import Belief
from std.document import Report, render
from std.collections import join_str
def main() -> None !{model.invoke, model.embed, observe.record}:
# 1. Sources with local citations → stable global ids + rewritten text.
docs = [
Doc(text="Solar capacity grew [1]. Costs fell [2].",
citations=[Cit(url="iea.org", start=20, end=23), Cit(url="irena.org", start=36, end=39)]),
Doc(text="Costs fell sharply [1].",
citations=[Cit(url="irena.org", start=19, end=22)]),
]
ids = build_url_to_id(docs)
mut sections: list[str] = []
for d in docs:
sections.append(rewrite(d.text, d.citations, ids))
# 2. Candidate facts, de-duplicated semantically.
unique_facts = semantic.dedup(["costs fell", "costs fell", "capacity grew"], 0.99)
# 3. Belief-driven loop: iterate until confidence crosses the threshold.
decisions = [0.7, 0.85, 0.95]
mut belief = Belief(alpha=1.0, beta=1.0, history=[0.5])
mut iters = 0
loop until belief.confidence() >= 0.7 max_iters len(decisions):
belief.update(decisions[iters])
iters = iters + 1
# 4. Draft a synthesis with ambient usage metering — no usage tuples.
mut calls = 0
mut cost = 0.0
with meter as u:
_synthesis = generate("Summarize renewable energy findings", 64)
calls = u.total_calls
cost = u.cost
# 5. Render a typed report to markdown.
r = Report(
title="Renewable Energy Findings",
context="Auto-synthesized from " + str(len(docs)) + " sources.",
confidence=belief.confidence(),
rationale="Confidence is a Beta-Bernoulli posterior over iteration evidence.",
takeaways=unique_facts,
section_titles=["Findings"],
sections_text=join_str(sections, "\n\n"),
conclusion="Costs continue to decline as capacity scales.",
)
print(render(r, "\n"))
log.info("run", stopped_iter=iters, confidence=belief.confidence(), model_calls=calls, cost=cost)

From the sema/ directory:

Terminal window
sema check examples/research-agent
SEMA_STRICT=1 sema run examples/research-agent
sema assure examples/research-agent --grade silver

sema check catches an unrecognized directive (a misplaced on drifted: or a typo does nothing silently otherwise); SEMA_STRICT=1 turns any runtime degradation into a hard error while you verify; sema assure … --grade silver runs the test blocks and requires explicit effect rows.

  • Return the best partial answer. Wrap the loop body in expect … except BudgetExceeded: and return the highest-confidence draft so far.
  • Swap the stop condition. Replace the belief threshold with a semantic guard (semantics("the answer fully addresses the question", answer)) — see Semantic Operations.
  • Give the agent tools. Feed a toolset to tools.run inside the loop; the effect rows on those tools bound what each iteration is allowed to touch.