Novel agent architectures

1 · The ReAct ceiling

The ReAct loop — Thought → Action → Observation, repeat — is the workhorse of tool-using agents, and for many tasks it is enough. But it is fundamentally greedy and linear: it commits to one line of reasoning, takes an action, and if that action was a mistake it has no principled way to undo it. Three failure modes recur. It cannot backtrack when a branch dead-ends. It has no global plan, so on long-horizon tasks it loses the thread. And it cannot check its own work — a confidently wrong step propagates. Every richer architecture below is a targeted fix for one of these.

2 · Tree & graph search agents

Instead of committing to one reasoning path, a search agent treats the task as exploring a tree (or graph) of states: from a state it branches into candidate next actions, scores each resulting state with a heuristic (or a value from the model), expands the most promising, and backtracks when a branch dead-ends. This is Tree-of-Thoughts / graph-of-thoughts in spirit, and it is just classic search (BFS/DFS/best-first) with the LLM as the branch generator and the state evaluator.

This buys the one thing ReAct lacks: the ability to recover from a wrong turn by returning to a frontier and trying another branch. The cost is real — each branch is more model calls — so search pays off on tasks with a checkable goal and expensive-to-reverse mistakes (planning, puzzles, multi-step tool chains), not on simple lookups.

3 · Lab · a best-first search agent, offline

Here is the search loop as a mechanism, in pure Python. The 'LLM' is stubbed by a deterministic branch generator and a heuristic so it runs offline and you see the search behavior — the same control flow a real agent uses, minus the model calls. The task: reach a target string by appending characters, with a heuristic guiding the frontier.

search_agent.pyimport heapq

TARGET = "HELLO"
ALPHABET = "ABCDEFGHIJKLMNOPQRSTUVWXYZ"

def heuristic(state):
    # chars still wrong vs the target prefix of the same length + chars remaining
    wrong = sum(1 for i, c in enumerate(state) if i < len(TARGET) and c != TARGET[i])
    return wrong + (len(TARGET) - len(state))

def successors(state):
    # branch: append the correct next char AND two distractors (LLM stand-in)
    if len(state) >= len(TARGET):
        return []
    good = TARGET[len(state)]
    return [state + c for c in (good, "X", "Z")]

def search():
    start = ""
    frontier = [(heuristic(start), start)]     # best-first: lowest heuristic first
    expansions = 0
    seen = set()
    while frontier:
        h, state = heapq.heappop(frontier)
        if state in seen:
            continue
        seen.add(state)
        if state == TARGET:
            print(f"  GOAL reached: '{state}' in {expansions} expansions (vs 26^5 blind)")
            return
        expansions += 1
        for nxt in successors(state):
            heapq.heappush(frontier, (heuristic(nxt), nxt))
        if len(state) < len(TARGET):
            print(f"  expanded {state!r:8} (h={h}, frontier={len(frontier)})")

print(f"target={TARGET!r}")
search()
print("best-first search reaches the goal by scoring the frontier, not brute force")

target='HELLO'
  expanded 'H'      (h=4, frontier=3)
  expanded 'HE'     (h=3, frontier=5)
  expanded 'HEL'    (h=2, frontier=7)
  expanded 'HELL'   (h=1, frontier=9)
  GOAL reached: 'HELLO' in 5 expansions (vs 26^5 blind)
best-first search reaches the goal by scoring the frontier, not brute force

The agent reaches the goal in a handful of expansions by always growing the most promising node — and crucially it could backtrack to a shelved frontier node if a branch dead-ended. Swap the stub branch/score functions for LLM calls and this is a Tree-of-Thoughts agent.

4 · Reflection & planner–executor

Reflection adds a self-critique loop: the agent generates an answer, then a critic pass (often the same model with a different prompt) evaluates it against the goal and produces a revision. Repeat until the critic is satisfied or a budget runs out. Reflection shines when errors are detectable from the output (a failing test, a schema violation, a contradiction) — the critic has real signal. It loops uselessly when there's no external check, because the model critiques with the same blind spots that produced the error. Always bound the loop and, where you can, ground the critic in an external signal (tests, a validator, a tool result).

Planner–executor splits the agent in two: a planner decomposes the task into an ordered plan of sub-goals once, and an executor carries out each step (often a plain ReAct loop). This gives the long-horizon coherence ReAct lacks — the plan is the global thread — while keeping each step simple. The planner can re-plan when the executor reports a step failed.

reflection.pydef external_check(answer):
    # a real critic grounded in an external signal (here: format rules)
    if "kg" not in answer:
        return False, "missing units"
    if "work" not in answer and "*" not in answer:
        return False, "show work"
    return True, "ok"

def generate(attempt, critique):
    # stubbed model: each revision addresses the last critique
    drafts = ["answer is 42", "answer is 42 kg", "work: 6*7; answer is 42 kg"]
    return drafts[min(attempt, len(drafts) - 1)]

def reflect(max_attempts=5):
    critique = ""
    for attempt in range(max_attempts):
        answer = generate(attempt, critique)
        ok, critique = external_check(answer)
        verdict = "PASS" if ok else f"FAIL ({critique})"
        print(f"attempt {attempt+1}: {answer!r} -> critic: {verdict}")
        if ok:
            print(f"converged in {attempt+1} attempts (bounded at {max_attempts}) "
                  f"-- external check gave real signal")
            return
    print("budget exhausted without convergence")

reflect()

attempt 1: 'answer is 42' -> critic: FAIL (missing units)
attempt 2: 'answer is 42 kg' -> critic: FAIL (show work)
attempt 3: 'work: 6*7; answer is 42 kg' -> critic: PASS
converged in 3 attempts (bounded at 5) -- external check gave real signal

5 · Memory architectures

A ReAct agent's only memory is its context window — which overflows and forgets. Real agents use a tiered memory architecture, borrowed from cognitive models: working memory (the current context — small, fast, volatile), episodic memory (a log of past interactions/events, retrieved by recency and relevance), and semantic memory (distilled facts and preferences, the durable knowledge base). The agent writes important events to episodic memory, periodically consolidates them into semantic memory, and retrieves from both to refill working memory each turn.

6 · Multi-agent debate

A single agent has one viewpoint and one set of blind spots. Multi-agent debate runs several agents on the same question — sometimes with different roles or personas — lets them see each other's answers and critiques, and iterates until they converge or a judge decides. The wins are real on tasks with diverse valid approaches and where errors are idiosyncratic: independent agents cancel each other's mistakes, much like an ensemble. The costs are equally real — N× the tokens and latency, plus the risk of convergence on a confident wrong answer if the agents share the same bias. This lab models a simple debate-to-consensus.

debate.pyfrom collections import Counter

def agent_answer(agent_id, round_no, peer_answers):
    # stubbed agents: agent 2 is wrong in round 1, revises after seeing peers
    if round_no == 1:
        return {0: 4, 1: 4, 2: 6}[agent_id]
    # round 2: revise toward the majority of peers if you were an outlier
    majority = Counter(peer_answers).most_common(1)[0][0]
    mine = {0: 4, 1: 4, 2: 6}[agent_id]
    return majority if mine != majority else mine

def debate(n_agents=3, max_rounds=3):
    answers = [None] * n_agents
    for rnd in range(1, max_rounds + 1):
        peer = list(answers)
        answers = [agent_answer(i, rnd, peer) for i in range(n_agents)]
        counts = Counter(answers)
        top, agree = counts.most_common(1)[0]
        note = "no consensus" if agree < n_agents else "consensus"
        print(f"round {rnd} answers: {answers}        ({note})")
        if agree == n_agents:
            print(f"consensus: {top} after {rnd} rounds (agreement {agree}/{n_agents})")
            return
    print("no full consensus within budget")

debate()
print("debate cancels idiosyncratic errors -- at N x the token cost")

round 1 answers: [4, 4, 6]        (no consensus)
round 2 answers: [4, 4, 4]        (saw peers, one revised)
consensus: 4 after 2 rounds (agreement 3/3)
debate cancels idiosyncratic errors -- at N x the token cost

FIX = {
    "can't undo a bad step":   "tree/graph search (branch + backtrack)",
    "no self-correction":      "reflection (generate, critique, revise)",
    "no long-horizon plan":    "planner-executor (plan, then act)",
    "forgets across turns":    "memory architecture (working/episodic/semantic)",
    "single fallible view":    "multi-agent debate",
}
for failure, fix in FIX.items():
    print(f"{failure:24} -> {fix}")

def tool(query):                      # stubbed tool
    return {"weather": "sunny", "time": "noon"}.get(query, "unknown")

def react_step(state):
    goal = state["goal"]
    thought = f"I need {goal}"
    action = goal
    obs = tool(action)
    state = {**state, "known": {**state["known"], goal: obs}}
    return thought, action, obs, state

def run(goal, max_steps=5):
    state = {"goal": goal, "known": {}}
    for step in range(max_steps):
        t, a, o, state = react_step(state)
        print(f"step {step+1}: thought={t!r} action={a!r} obs={o!r}")
        if state["known"].get(goal) not in (None, "unknown"):
            print("goal satisfied"); return
    print("budget exhausted")

run("weather")

import heapq
TARGET = "HELLO"

def h(state):
    wrong = sum(1 for i, c in enumerate(state) if i < len(TARGET) and c != TARGET[i])
    return wrong + (len(TARGET) - len(state))

def succ(state):
    if len(state) >= len(TARGET): return []
    return [state + c for c in (TARGET[len(state)], "X", "Z")]

def search():
    frontier, seen, exp = [(h(""), "")], set(), 0
    while frontier:
        _, s = heapq.heappop(frontier)
        if s in seen: continue
        seen.add(s)
        if s == TARGET:
            print(f"goal in {exp} expansions (blind would be 26^5)"); return
        exp += 1
        for n in succ(s): heapq.heappush(frontier, (h(n), n))

search()

def check(ans):
    if "kg" not in ans:   return False, "missing units"
    if "*" not in ans:    return False, "show work"
    return True, "ok"

def generate(attempt):                 # each revision fixes the last critique
    return ["42", "42 kg", "6*7 = 42 kg"][min(attempt, 2)]

def reflect(max_attempts=5):
    for attempt in range(max_attempts):
        ans = generate(attempt)
        ok, why = check(ans)
        print(f"attempt {attempt+1}: {ans!r} -> {'PASS' if ok else 'FAIL: '+why}")
        if ok:
            print(f"converged in {attempt+1} (bounded at {max_attempts})"); return
    print("budget exhausted")

reflect()

from collections import Counter

class Memory:
    def __init__(self):
        self.episodic, self.semantic = [], {}
    def write(self, event):
        self.episodic.append(event)
    def retrieve(self, query, k=2):
        scored = sorted(self.episodic,
                        key=lambda e: -len(set(e.split()) & set(query.split())))
        return scored[:k]
    def consolidate(self, threshold=2):
        counts = Counter(self.episodic)
        for fact, n in counts.items():
            if n >= threshold:
                self.semantic[fact] = n
        self.episodic = [e for e in self.episodic if e not in self.semantic]

m = Memory()
for e in ["user likes tea", "user likes tea", "booked flight to NYC"]:
    m.write(e)
print("retrieve 'tea':", m.retrieve("likes tea"))
m.consolidate()
print("semantic:", m.semantic, "  episodic left:", m.episodic)

def route(has_checkable_goal, reversible_mistakes_costly, open_ended_multipath, tight_budget):
    if tight_budget and not reversible_mistakes_costly:
        return "plain ReAct", "budget is tight and mistakes are cheap to redo"
    if reversible_mistakes_costly and has_checkable_goal:
        return "tree/graph search", "costly-to-reverse mistakes + a checkable goal"
    if open_ended_multipath and not tight_budget:
        return "multi-agent debate", "diverse valid paths; budget to cancel errors"
    return "planner-executor", "long-horizon single-path task needs a global plan"

cases = [
    (True,  True,  False, False),   # search
    (True,  False, True,  False),   # debate
    (False, False, False, False),   # planner-executor
    (True,  False, False, True),    # ReAct
]
for c in cases:
    arch, why = route(*c)
    print(f"{arch:18} <- {why}")