laserbrain 0.3.4

Attach the smart recursion harness to any agent loop — a provably-correct, external check for when an AI agent (or team) has drifted from its goal.

laserbrain

Attach the smart recursion harness to any agent loop — a provably-correct, external check for when an AI agent (or a team of agents) has drifted from its goal.

An agent watching only itself provably can't catch its own drift: each step looks fine next to the last while it wanders far from where it began. laserbrain is a fixed reference it checks against instead. There's a proof. The theorem and the studies (nulls included). · Watch it work.

The check is a pure function, so this SDK runs it locally and free — no key, no latency. Add a key and it also mirrors to the API for retained drift history, alerts and the fleet view: you pay to see your agents drift, not for the check.

pip install laserbrain

See it work before writing any code:

laserbrain demo          # watch an agent drift off-goal and get returned
laserbrain check --goal "write a poem" --against "build a parser"   # a one-shot drift check

The check (local, free)

from laserbrain import Harness

hz = Harness()                      # add key="lb_live_…" to also retain history
v = hz.check(goal="build the JSON parser", progress="advancing", distance=6)
if v.drifting:
    print(v.reason, "—", v.advice)  # e.g. "goal-drift — your goal no longer matches…"

progress is one of advancing | stuck | circling; distance is 0–10 to done. Reasons: advancing, grounded, goal-drift, stalled, self-report:stuck/circling, ungrammatical. Want a bounded reading instead of a raw distance? v.ground_score maps Φ to [0,1] — 1.0 fully grounded, falling as it drifts (1/(1+4·Φ)).

The grammar — bring your own

The theorem blesses a fixed reference, never a particular vocabulary. The default grammar is word overlap: frozen, zero-dependency, and honestly crude — an agent that restates the same goal in different words ("build the JSON parser" → "construct the JSON decoder") trips it as drift. Pass similarity= and the same theorem runs on a better vocabulary:

def semantic(a: str, b: str) -> float:      # 1.0 = same goal, 0.0 = unrelated
    return cosine(embed(a), embed(b))       # your embedding model, or any metric

hz = Harness(similarity=semantic)

Only the goal term changes — thresholds, the stall rule and the return logic are untouched, and children from sub() inherit it. Omit it and you get the published instrument, byte for byte. A metric that misbehaves degrades safely rather than crashing the check.

The act layer — close the loop

Give laserbrain your step function and it detects drift and injects the return, so the agent recovers instead of spinning. Your step reads ctx["return"] and steers back.

def step(ctx):
    if ctx.get("return"):            # laserbrain told us to return to ground
        ...                          # steer the agent back toward its goal
    ...
    return dict(goal="build the JSON parser", progress="advancing", distance=d, done=d == 0)

ctx = Harness().run(step, on_return=lambda v, ctx: print("↩", v.advice))

Async agents get the same loop, awaited — your step and callbacks may be async, and the API mirror runs off the event loop so it never blocks. After any run, report() prints the shape of it:

hz = Harness()
await hz.arun(async_step)                 # the act layer for asyncio agents
print(hz.report())
# laserbrain · 11 steps · 1 drift(s)
#   goal: 'build the JSON parser'
#   Φ  ▁▂▃▃▃▃▃▃▅▇█  peak 0.21
#   drifts: stalled×1

Nested recursion — a recursion as a set of recursions

Agents decompose. "Build the parser" becomes "write the tokenizer", which becomes "handle string escapes" — and each subtask is its own recursion with its own ground. sub() opens one. Every node runs the same proven check against its own goal, so a subtask is judged on its own terms.

What no node can see is the tree. Every agent can report advancing on its own sub-goal while the whole decomposition never brings the root any closer — the same blindness, one level up. So the set gets its own fixed reference: the root's ground.

root = Harness()
root.check("build the JSON parser", "advancing", 8)
tok  = root.sub("write the tokenizer", distance=4)   # a child recursion, own ground
tok.check("write the tokenizer", "advancing", 2)     # → "advancing" — locally fine
esc  = tok.sub("handle string escapes", distance=3)  # nest as deep as you like

print(root.tree_report())
# laserbrain · recursion tree · depth 1 · 3 recursions · 7 steps
# 'build the JSON parser' · 1 steps
#   └ 'write the tokenizer' · 4 steps
#   └ 'write the AST nodes' · 2 steps
#   ⚑ the TREE is spinning — 6 steps across the set since the root got closer

Each node's check is the proven detector; the tree-level signal is a prototype extension, like the teams below — useful, not a theorem.

Recursion teams — styled multi-agent oversight

A recursion team styles each role's recursion: a deep explorer tolerates displacement, a tight checker returns fast. laserbrain runs the team, watches the shared goal (the fixed reference), and injects the return per role — catching the echo/agreement spiral a self-watching group can't see.

from laserbrain import Team

def agent(role, history, injected):
    # your LLM call for this role; `injected` is a return-to-ground note (or None)
    return position, distance

Team("adversarial-deliberation", goal="…").run(agent)
# presets: deep-search · iterative-refinement · adversarial-deliberation

Oversight, provenance, continuity

Human-in-the-loop. A self-correcting return usually takes. When it doesn't — the agent keeps drifting past escalate_after steps — laserbrain escalates that drift to a human. The human doesn't watch every step; they see only what the fixed reference caught. Their decision overrides the auto-return.

def on_escalate(v, ctx):
    return ask_a_human(v.reason, v.advice)     # Slack, a queue, a webhook — you wire it
                                               # returning a decision injects it as the return
Harness().run(step, escalate_after=3, on_escalate=on_escalate)

Provenance. Every check is written to a hash-chained ledger — tamper-evident and verifiable offline, by anyone, no key. Editing a past verdict to hide a drift breaks the chain at that link.

hz.export_audit("run.json")
from laserbrain import verify_audit
verify_audit(json.load(open("run.json")))      # (True, -1) intact · (False, i) broken at link i

Team continuity. Snapshot a running team and resume it in a later session — the shared goal (the fixed reference) and the dialogue carry over, so the group re-grounds instead of starting cold.

snap = team.snapshot()                         # JSON-safe; persist it anywhere
team = Team.restore(snap)                       # keeps watching the same ground

Framework adapters

Already on LangGraph, CrewAI, AutoGen, or the OpenAI Agents SDK? Attach laserbrain without changing your loop. Because it checks a fixed reference, it needs the agent to spell its state — so each adapter takes an extract that maps your framework's state to (goal, progress, distance). No adapter imports a framework: each returns a plain callable you hand to the framework's own hook, so install only the one you use.

from laserbrain.adapters import guard, langgraph_node, crewai_step_callback, middleware

# generic — wrap any step that returns dict(goal=, progress=, distance=)
@guard
def step(state): ...

# LangGraph — a node that writes the Verdict into graph state; branch on it
g.add_node("laserbrain", langgraph_node(extract=lambda s: (s["goal"], "advancing", s["dist"])))
g.add_edge("agent", "laserbrain")
g.add_conditional_edges("laserbrain",
    lambda s: "return" if s["laserbrain"].drifting else "agent")   # .advice steers the return

# CrewAI — a step_callback that fires each agent step
Agent(..., step_callback=crewai_step_callback(lambda o: (o.goal, o.status, o.dist)))

# anything else (AutoGen, OpenAI Agents, a custom loop) — one check per step
lb = middleware(extract=my_extract)
v = lb(step_output)
if v.drifting: reinject(v.advice)

Each adapter runs the check locally and free; pass key=/run_id= (or your own Harness) to also retain history. The LangGraph path is verified end to end against a real StateGraph — see example_langgraph.py (pip install laserbrain langgraph): the agent drifts to a different goal, langgraph_node catches it, and the conditional edge routes it back.

What's proven, and what isn't

The single-agent detector mirrors the frozen, published instrument (drift.ts @ 6b483de7) and rests on a theorem: detection is sound and complete, and no self-monitoring agent can be. The multi-agent dialogue and recursion teams are a prototype extension — useful, not (yet) a theorem. Whether returning an agent keeps the answer as good is an honest open question; this SDK gives you the detection and the return mechanism, and says plainly what each is.

MIT · phronesis.world/laserbrain