DEV Community: varun pratap Bhardwaj

Your Multi-Agent System Probably Uses 1 Pattern. Here Are 12.

varun pratap Bhardwaj — Sun, 12 Apr 2026 13:34:20 +0000

Last month I watched a team burn $400 in API credits in a single afternoon.

They had four agents working on a research task. Each agent could see every other agent's output. Every time one agent updated its findings, the other three would re-process, generate new outputs, and trigger another round of updates. Four agents, talking to each other in a full mesh, with no termination condition. The token counter looked like a slot machine.

The fix took ten minutes. They didn't need a mesh. They needed a hierarchy — one coordinator agent that dispatched three specialists and merged the results. Same four agents, same task, same quality. One-eighth the cost.

The topology you choose for a multi-agent system — how agents communicate, who talks to whom, who decides — determines whether your system is fast or slow, reliable or brittle, cheap or expensive. Most teams default to chaining agents sequentially and never question it. But agent orchestration is a graph problem, and different tasks demand fundamentally different graph shapes.

I've spent the past year building and studying multi-agent architectures. Here are the 12 topology patterns that cover every coordination scenario I've encountered, grouped by complexity.

Simple Topologies

These are the building blocks. If you're new to multi-agent orchestration, start here.

1. Sequential

A ──▶ B ──▶ C ──▶ D

Each agent completes its work before passing output to the next. Classic pipeline.

When to use: Step-by-step workflows where order matters — document extraction, then classification, then summarization. Code generation, then review, then testing.

Example: An invoice processing pipeline where Agent A extracts line items from a PDF, Agent B validates amounts against a purchase order, Agent C flags discrepancies, and Agent D generates an approval recommendation. Each step needs the full output of the previous one.

Trade-off: Simple to reason about and debug. But slow — your total latency is the sum of every step. One slow agent blocks everything downstream.

2. Parallel

     ┌── A ──┐
     │       │
IN ──┼── B ──┼──▶ MERGE
     │       │
     └── C ──┘

All agents receive the same input and run simultaneously. Results merge at the end.

When to use: Independent analysis from multiple perspectives — sentiment analysis across languages, security scanning with different rule sets, researching a topic from multiple sources.

Example: Running a security audit, a performance profiler, and a code style linter on the same codebase simultaneously. Each analyzer is independent. Results merge into a single report. What took 90 seconds sequentially finishes in 30.

Trade-off: Dramatically faster than sequential when tasks are independent. But you need a merge strategy, and if agents produce conflicting outputs, the merge logic becomes the hard part.

Structured Topologies

These patterns impose hierarchy or dependency ordering. They handle complex, real-world workflows where "everything runs at once" isn't realistic.

3. Hierarchical

          BOSS
         / | \
        /  |  \
      W1  W2  W3

A supervisor agent decomposes a task and delegates sub-tasks to worker agents. Workers report back. The boss synthesizes.

When to use: Task decomposition — "write a research report" becomes "gather data" + "analyze trends" + "draft sections." Project management workflows where a lead agent coordinates specialists.

Example: A research agent receives "analyze the competitive landscape for AI agent frameworks." It spawns three workers: one searches academic papers, one crawls GitHub repos for stars and commit activity, one analyzes pricing pages. The boss merges findings into a structured competitive matrix.

Trade-off: Clean separation of concerns. The boss agent is a single point of failure, though — if it decomposes poorly, every worker goes in the wrong direction. Boss prompt quality is everything.

4. DAG (Directed Acyclic Graph)

A ──▶ B ──▶ D
 \         ▲
  └──▶ C ──┘

Agents form a dependency graph. An agent runs only when all its dependencies have completed. No cycles.

When to use: Complex workflows with mixed dependencies — data fetching (parallel) feeds into analysis (sequential) which feeds into report generation, but only after a separate validation step also completes.

Example: A CI/CD pipeline where linting and type-checking run in parallel after checkout, unit tests run after linting passes, integration tests wait for both unit tests and a database migration to complete, and deployment triggers only after every test is green.

Trade-off: Maximum flexibility and parallelism where the dependency structure allows it. The cost is complexity — you need a scheduler that understands the graph and handles failures mid-execution. This is the topology that build systems (Make, Bazel) have used for decades.

5. Star

      S1
      |
S4 ── HUB ── S2
      |
      S3

A central hub agent coordinates all communication. Spoke agents never talk to each other directly — everything routes through the hub.

When to use: Centralized coordination where you need one agent to maintain global state — a customer service router dispatching to specialist agents, or a planning agent that tracks progress across multiple workstreams.

Trade-off: Easy to monitor and control since all traffic flows through one point. But the hub is a bottleneck. If the coordinator agent is slow or hits rate limits, the entire system stalls.

6. Grid

A1 ── A2 ── A3
|     |     |
B1 ── B2 ── B3
|     |     |
C1 ── C2 ── C3

Agents are arranged in a matrix with structured communication along rows and columns. Each agent can communicate with its direct neighbors.

When to use: Structured team simulations — rows represent departments (engineering, design, QA) and columns represent features. Each agent has a specific role at a specific intersection.

Trade-off: Excellent for problems that naturally map to two dimensions. Rigid structure means it's overkill for simpler problems, and the communication overhead grows with grid size.

Collaborative Topologies

These patterns prioritize interaction quality over structural simplicity. Agents actively engage with each other's outputs.

7. Debate

PRO ◀──────▶ CON
      \   /
       ▼ ▼
      JUDGE

Two or more agents argue opposing positions. A judge agent evaluates the arguments and renders a decision.

When to use: High-stakes decisions where you need to stress-test reasoning — architecture decisions, risk assessments, legal analysis. Any situation where a single agent's confidence might mask a blind spot.

Example: An architecture decision record where one agent argues for microservices, another argues for a monolith, and a judge evaluates both against concrete constraints — team size, deployment frequency, latency budget. The judge doesn't just pick a winner; it synthesizes a recommendation with trade-offs acknowledged.

Trade-off: Produces higher-quality decisions by exposing weak arguments. Expensive in tokens — you're running 3+ agents on the same problem. Not worth it for routine tasks.

8. Mesh

A ◀──▶ B
|\ /|
| X  |
|/ \|
C ◀──▶ D

Every agent can communicate with every other agent. No hierarchy, no fixed paths.

When to use: Collaborative problem-solving where agents need to negotiate, share partial results, and build on each other's work — brainstorming sessions, collaborative writing, complex debugging where each agent sees a different part of the system.

Trade-off: Maximum flexibility and emergent collaboration. But message volume grows quadratically with agent count. Without careful protocol design, mesh systems devolve into noise. Works well with 3-5 agents; breaks down beyond that. This is the topology from my opening story — powerful, but easy to misuse.

9. Circular (Round-Robin)

A ──▶ B
▲     |
|     ▼
D ◀── C

Agents pass work around a ring, each refining or extending the previous agent's output. Multiple rounds.

When to use: Iterative refinement — draft, review, revise, polish. Translation chains where each pass improves quality. Red team / blue team cycles where attack and defense alternate.

Example: A writing pipeline where Agent A drafts, Agent B critiques for logical gaps, Agent C rewrites based on the critique, and Agent D fact-checks. The cycle repeats. Each round sharpens the output. After three rounds, the draft reads like it had a human editorial team.

Trade-off: Natural fit for refinement workflows. Quality improves with each round — up to a point. Diminishing returns kick in, and without a termination condition, the system loops forever. Always set a max-rounds limit.

10. Mixture-of-Agents

     ┌── A ──┐
     │       │
IN ──┼── B ──┼──▶ SYNTHESIZER ──▶ OUT
     │       │
     └── C ──┘

Multiple models (or the same model with different temperatures/prompts) process the same input. A dedicated synthesizer agent combines them into a single response that's better than any individual output.

When to use: Best-of-N generation — multiple agents draft answers, and a synthesizer picks the best parts of each. Content moderation where GPT-4, Claude, and Gemini each classify a piece of content, and majority voting catches edge cases any single model misses.

Trade-off: Consistently outperforms single-agent and simple-parallel approaches on quality benchmarks. The synthesizer is doing the hard intellectual work, though, so its prompt and capability matter enormously. Also 3-4x the token cost of a single agent.

Specialized Topologies

These patterns are purpose-built for specific domains.

11. Forest

   ROOT1         ROOT2
   / | \         / \
  A  B  C       D   E
    / \             |
   F   G            H

Multiple independent trees running in parallel. Each tree has its own hierarchy. No cross-tree communication until results merge at the end.

When to use: Parallel hierarchies — multiple teams working on separate features simultaneously, multi-repository analysis where each repo gets its own agent tree, running the same analysis against multiple datasets.

Trade-off: Scales linearly with the number of trees. Each tree is isolated, which is great for independence but means you can't share discoveries across trees mid-execution. Use this when the sub-problems are truly independent.

12. Maker (Build-Test-Ship)

ARCHITECT ──▶ BUILDER ──▶ TESTER ──▶ DEPLOYER
                 ▲            |
                 └── fix ─────┘

A specialized engineering pipeline with feedback loops. The builder creates, the tester validates, and failures loop back to the builder for fixes. Only passing work advances.

When to use: Software engineering workflows — code generation with automated testing, infrastructure-as-code with validation, any creative process where output quality must be verified before moving forward.

Example: A code generation workflow where the Maker agent writes a function and the Tester agent runs the test suite. Tests fail. The Tester sends the failure output and stack trace back to the Maker with instructions to fix. Two iterations later, all tests pass and the code advances to deployment.

Trade-off: Built-in quality gates mean output is more reliable than a straight pipeline. The feedback loop adds latency and cost, but catches errors before they propagate. The key design decision is how many retry cycles to allow before escalating.

Choosing the Right Topology

There's no universal best topology. The right choice depends on your constraints:

Constraint	Best Topologies
Minimize latency	Parallel, Forest, DAG
Maximize quality	Debate, Mixture-of-Agents, Circular
Minimize cost	Sequential, Star
Handle complex dependencies	DAG, Grid
Need iterative refinement	Circular, Maker
Collaboration-heavy	Mesh, Debate

Start with the simplest topology that fits your problem. Sequential until you need speed. Parallel until you need dependencies. DAG when Parallel gets tangled. Debate only when the decision is worth 3x the tokens. Mesh almost never — and when you do use it, set a message budget.

In practice, production systems compose topologies. A DAG might contain a Debate sub-graph at a critical decision node, with Parallel branches for independent analysis. The twelve patterns are composable primitives, not rigid choices.

One More Thing

All 12 topologies ship as composable primitives in Qualixar OS, with full execution semantics, retry policies, and observability built in. The formal definitions, message-passing protocols, and benchmark results are in the paper (arXiv:2604.06392).

DEMO

PORTAL WALKTHROUGH

I Tracked Why AI Agent Projects Fail. 80% of the Time, It's Not the Agents.

varun pratap Bhardwaj — Sun, 12 Apr 2026 11:54:59 +0000

Last quarter, a team I advise at a Fortune 100 company built a multi-agent pipeline that could analyze SEC filings, cross-reference market data, and generate investment summaries. In the demo, it was stunning. GPT-4o handled reasoning. Claude did the writing. A custom agent orchestrated the flow.

They spent 3 weeks building the agents. They spent the next 14 weeks building everything around them.

Routing logic. Retry policies. Cost tracking. Quality checks. Memory that persisted between sessions. Logging that was actually searchable. A dashboard so the ops team could see what was happening without reading Python.

The agents were 18% of the codebase. The infrastructure was the other 82%.

This is not an isolated story. This is the story.

The numbers nobody talks about

Let's start with what's public:

Gartner (March 2025): 40% of agentic AI projects will be scaled back or cancelled by 2028. Not because agents are dumb — because teams can't operationalize them.
Gartner (2026): 1,445% surge in enterprise inquiries about multi-agent systems. Everyone wants to build them. Few know how to run them.
GitHub data: 4% of all GitHub commits now come from Claude Code alone — roughly 135,000 commits per day. Agents aren't experimental. They're writing production code right now.
Flowise AI (March 2026): A CVSS 10.0 remote code execution vulnerability hit 12,000+ deployed instances. When agent infrastructure is an afterthought, security is too.

The pattern is consistent: building agents is a solved problem. Operating agents is where projects die.

The five infrastructure problems every agent team solves from scratch

After 15 years in enterprise IT — and after building agent systems that actually shipped into production — I've watched the same five problems appear on every project. Different company, different use case, same headaches.

1. The routing problem

You have a pipeline with four agents. One needs speed (classification). One needs depth (analysis). One needs to be cheap (summarization). One needs multimodal understanding (document processing).

That's four different models, potentially four different providers, with different rate limits, latency profiles, and pricing.

Who decides which model serves which agent? Most teams hardcode it. Which works until your provider changes pricing, deprecates a model, or has an outage. Then someone rewrites routing logic at 2 AM.

What good looks like: Declarative routing constraints. "This agent needs latency under 2 seconds, quality above 0.8, cost under $0.01 per call." The system figures out the rest. When a provider goes down, traffic shifts automatically.

# What teams WANT to write
routing:
  analysis_agent:
    constraints:
      max_latency_ms: 2000
      min_quality: 0.8
      max_cost_per_call: 0.01
    fallback: [claude-3-haiku, gpt-4o-mini]

# What teams ACTUALLY write: 200 lines of this
if task_type == "analysis":
    try:
        response = openai_client.chat(model="gpt-4o", ...)
    except RateLimitError:
        try:
            response = anthropic_client.messages(model="claude-3-5-sonnet", ...)
        except Exception:
            response = openai_client.chat(model="gpt-4o-mini", ...)
            log.warning("Fell back to mini, quality may be degraded")

Every team writes that second version. Nobody wants to.

2. The quality problem

Agent output is non-deterministic. Same prompt, same model, Monday vs. Friday — different quality. This is fine in a chatbot. It's not fine when your agent is generating financial reports, writing customer communications, or making decisions that affect revenue.

Most teams discover this the hard way: a customer complains, someone traces it back to a hallucinated data point, and the response is "we should probably add eval."

What good looks like: A judge pipeline that runs automatically. Every agent output gets evaluated against configurable criteria before it reaches the user. Multiple judges can form consensus. Quality scores feed back into routing — agents that produce lower quality get routed less traffic.

Here's the uncomfortable truth: the teams that skip quality enforcement are the teams that end up in the 40% that get cancelled. Leadership loses trust when agent output is unpredictable.

3. The memory problem

Your agent solved a problem yesterday. Today, the same user asks a related question. Your agent starts from zero.

This isn't a vector database problem. Bolting RAG onto an agent gives it "search" — it doesn't give it "memory." Real cognitive memory has structure:

Working memory: What's relevant right now, in this conversation
Episodic memory: What happened in past interactions (the story)
Semantic memory: What things mean (the knowledge)
Procedural memory: How to do things (the skills)

Humans don't remember everything. We consolidate — important things get reinforced, irrelevant things fade. Agent memory should work the same way. Most agent memory implementations are append-only vector stores that grow until they're too slow to query and too noisy to be useful.

What good looks like: Local-first memory with automatic consolidation. The agent remembers what matters, forgets what doesn't, and retrieves what's relevant — without a round-trip to a cloud vector database.

4. The cost problem

Three agents running in parallel, each hitting a different model API. One retries four times because of a transient error. Another loops because its termination condition is slightly wrong.

Your daily budget just became your weekly budget. And nobody noticed until the invoice arrived.

What good looks like: Per-agent cost tracking, circuit breakers that kill runaway agents, budget caps that actually enforce. This isn't exotic — it's what every cloud service does for compute. Agent compute just doesn't have the tooling yet.

5. The observability problem

Something went wrong. An agent produced bad output three steps deep in a pipeline. 217 events fired. 47 tool calls. 12 LLM invocations across 3 providers.

Where do you start looking?

Most agent systems log everything or nothing. Either you have a 50MB log file per request with no structure, or you have print("agent finished") and a prayer.

What good looks like: Structured traces with causality. "This output was produced by Agent C, which received input from Agent B, which was routed to GPT-4o because Agent A's Claude request exceeded the latency budget." Every decision point, every tool call, every retry — traceable and searchable.

Why this isn't a framework problem

Frameworks are doing their job. CrewAI gives you role-based teams. LangGraph gives you stateful graphs. AutoGen gives you conversations. These are real, useful tools.

But they're solving the what — what agents do, how they reason, which tools they call.

The five problems above are the how of production operations. And they're framework-agnostic. Whether your agent is built with CrewAI or LangGraph or raw API calls, it still needs routing, quality enforcement, memory, cost control, and observability.

This is the Docker-to-Kubernetes gap. In 2013, Docker let you run a container. But running containers in production needed a layer above the runtime — scheduling, networking, scaling, health checks, recovery. That was Kubernetes. Container runtime was the capability. Kubernetes was the operations layer.

Agent frameworks are the capability. The operations layer is what's missing.

The topology dimension most teams miss

Beyond the five operational problems, there's a design problem that compounds everything: how agents communicate determines whether your system works.

Most teams default to sequential pipelines (A passes to B passes to C) or simple parallel execution (A, B, C run simultaneously, merge results). These cover maybe 20% of real-world multi-agent needs.

There are at least 12 distinct coordination patterns, and choosing the wrong one silently kills performance:

Pattern	When it wins	When it fails
Sequential	Strict ordering matters	Latency-sensitive tasks
Parallel	Independent analysis	Conflicting outputs need reconciliation
Hierarchical	Clear task decomposition	Boss agent decomposes poorly
DAG	Mixed dependencies	Complex failure handling
Debate	High-stakes decisions	Routine tasks (waste of tokens)
Mesh	3-5 agents collaborating	>5 agents (quadratic message growth)
Mixture-of-Agents	Quality-critical output	Cost-sensitive workloads
Circular	Iterative refinement	No termination condition = infinite loop

The architectural decision isn't just "which framework." It's "which communication pattern, for which sub-task, with which failure mode." Most teams pick one pattern for the whole system because switching patterns means rewriting the orchestration layer.

A checklist before you build

If you're about to build (or rebuild) a multi-agent system, here's what I'd verify before writing agent code:

[ ] Routing strategy defined. Do you know which model serves which agent, and what happens when that model is unavailable?
[ ] Quality gates in place. Is there a judge or eval step before agent output reaches users?
[ ] Memory architecture chosen. Are you using structured memory or just appending to a vector store?
[ ] Cost controls configured. Per-agent budgets, circuit breakers, retry limits?
[ ] Observability instrumented. Can you trace a bad output back to its root cause in under 5 minutes?
[ ] Topology selected intentionally. Did you pick your communication pattern, or did you default to sequential?
[ ] Framework lock-in assessed. Can you swap or add a new framework without rewriting your operations layer?

If four or more of these are "no" or "we'll figure it out later" — you're in the 40%.

What I'm building about it

I've spent the last several months working on this exact problem. Not a framework. An operations layer that sits above frameworks and handles routing, quality, cost, memory, and observability for any agent, from any framework.

It's called Qualixar OS. Here's the honest version:

What it does: Imports agents from CrewAI, LangGraph, AutoGen, and others through a bridge protocol. Routes tasks to models based on cost-quality-latency constraints. Runs a judge pipeline on agent output. Provides local-first cognitive memory (4-layer, with consolidation). Ships a 24-tab dashboard for operations teams. Supports 12 execution topologies with formal semantics.

Where it is: The core is solid — 2,831 tests, 49 database tables, 25 MCP tools. The paper is published and peer-reviewable. But this is an independent research project, not a VC-backed startup. I'm one researcher with 15 years of enterprise experience who got tired of watching teams rebuild the same infrastructure.

What it isn't: It's not a hosted service. It runs on your machine. It doesn't replace your agents or your framework. It's the layer underneath that you'd otherwise build yourself.

I wrote a 20-page paper formalizing the architecture, the topology algebra, and the execution semantics: arxiv.org/abs/2604.06392. Because claims without math are just marketing.

Try it:

npx qualixar-os

Or just read the paper first. If you've felt the pain described in this post, the architecture section will feel familiar — it's the infrastructure you already wished existed.

Paper: arxiv.org/abs/2604.06392
Project: qualixar.com

I'm Varun Pratap Bhardwaj — independent researcher, 15 years in enterprise IT. I build open tools for AI agent reliability. If you're dealing with the same infrastructure pain, I'd genuinely love to hear what's broken in your setup. The comments are open.

Demo

I built a peer-to-peer communication layer for AI coding agents — here's how it works

varun pratap Bhardwaj — Wed, 08 Apr 2026 14:30:12 +0000

I run 3-4 AI coding sessions in parallel. Claude Code in VS Code for the frontend, another Claude session in the terminal for backend, sometimes Cursor or Antigravity for a third workstream.

The biggest pain point? They're completely isolated.

Session A refactors the authentication module. Session B starts editing the same file because it doesn't know Session A is working on it. I become the message bus — copy-pasting context between terminals like it's 2005.

This isn't unique to Claude Code or Cursor. Every AI coding agent has this problem. The Model Context Protocol (MCP) gives agents tools, but no way to coordinate with other agents on the same machine.

The Solution: SLM Mesh

SLM Mesh is an open-source MCP server that gives AI coding agents 8 tools for peer-to-peer communication:

Peer Discovery — agents auto-detect each other (scope by machine, directory, or git repo)
Direct Messaging — send structured messages between specific sessions
Broadcast — one-to-all message delivery
Shared State — key-value scratchpad accessible by all peers
File Locking — advisory locks with auto-expire to prevent edit conflicts
Event Bus — subscribe to peer_joined, state_changed, file_locked events
Summary — each agent announces what it's working on
Status — broker health and mesh statistics

Quick Start

# Install
npm install -g slm-mesh

# Add to Claude Code
claude mcp add --scope user slm-mesh -- npx slm-mesh

# Add to Cursor / VS Code / Windsurf
# mcp.json:
{
  "mcpServers": {
    "slm-mesh": {
      "command": "npx",
      "args": ["slm-mesh"]
    }
  }
}

That's it. Open two sessions and ask one of them to "check mesh_peers" — it will see the other.

Architecture

┌─────────────────────────────────────────────────┐
│                  Your Machine                    │
│                                                  │
│  ┌──────────┐   ┌──────────┐   ┌──────────┐    │
│  │ Claude    │   │ Cursor   │   │ Aider    │    │
│  │ Code      │   │          │   │          │    │
│  └─────┬────┘   └─────┬────┘   └─────┬────┘    │
│        │              │              │           │
│  ┌─────┴────┐   ┌─────┴────┐   ┌─────┴────┐    │
│  │ MCP      │   │ MCP      │   │ MCP      │    │
│  │ Server   │   │ Server   │   │ Server   │    │
│  └─────┬────┘   └─────┬────┘   └─────┬────┘    │
│        │              │              │           │
│        └──────────────┼──────────────┘           │
│                       │                          │
│              ┌────────┴────────┐                 │
│              │ SLM Mesh Broker │                 │
│              │ localhost:7899  │                 │
│              │ SQLite + UDS    │                 │
│              └─────────────────┘                 │
└─────────────────────────────────────────────────┘

Key design decisions:

Auto-lifecycle: First MCP server auto-starts the broker. Last peer leaves, broker shuts down after 60s. No daemon to manage.
SQLite + WAL: Concurrent reads, single writer, crash-safe. Messages and events auto-pruned after 24/48 hours.
Unix Domain Sockets: Real-time push delivery in <100ms. No polling.
Bearer token auth: Random 32-byte token per broker session. No dangerous flags needed.
Agent-agnostic: Works with any MCP client. Auto-detects Claude Code, Cursor, Aider, Codex, Windsurf, VS Code.

Real-World Workflow

Here's what I actually do with it daily:

Morning (3 sessions):

Session 1 (VS Code): "I'm refactoring the auth module"
→ Sets summary via mesh_summary
→ Locks auth.ts via mesh_lock

Session 2 (Terminal): "What are the other sessions doing?"
→ Calls mesh_peers — sees Session 1 is on auth
→ Calls mesh_lock query auth.ts — sees it's locked
→ Works on database instead

Session 3 (Antigravity): Starts a migration
→ Broadcasts "database schema changing to v2.1" via mesh_send
→ Sets db_version = 2.1 in shared state via mesh_state
→ Sessions 1 and 2 see the update

No copy-pasting. No context switching. The agents coordinate themselves.

The Numbers

Metric	Value
Tests	480 passing
Coverage	100% lines
MCP tools	8
CLI commands	12
Dependencies	4 (MCP SDK, better-sqlite3, commander, zod)
Python client	Zero deps (stdlib only)
Install size	~80 KB (packed)

Comparison with claude-peers

claude-peers proved the demand for this — 1,600 stars in 2 weeks. SLM Mesh is the production-grade answer:

Feature	SLM Mesh	claude-peers
MCP tools	8	4
File locking	Yes	No
Shared state	Yes	No
Event bus	Yes	No
Agent-agnostic	Any MCP agent	Claude only
Dangerous flags	Not needed	Required
Tests	480 (100% cov)	0
Auth	Bearer token	None
Python client	Yes	No

Try It

npm install -g slm-mesh

GitHub: github.com/qualixar/slm-mesh
PyPI: pip install slm-mesh

MIT licensed. Part of the Qualixar research initiative by Varun Pratap Bhardwaj.

Feedback welcome — especially interested in what multi-session workflows you'd use this for.

SuperLocalMemory V3: Mathematical Foundations for Production-Grade Agent Memory

varun pratap Bhardwaj — Wed, 18 Mar 2026 04:00:44 +0000

[(https://varunpratap.com/blog/superlocalmemory-v3-mathematical-foundations)] We applied information geometry, algebraic topology, and stochastic dynamics to AI agent memory. 74.8% on LoCoMo with data staying local — the highest score reported without cloud dependency. 87.7% in full-power mode. 60.4% with no LLM at any stage. Open source under MIT.

The Problem Is Scale, Not Storage

Every AI coding assistant — Claude, Cursor, Copilot, ChatGPT — starts every session from scratch. The memory problem has been solved at development scale: Mem0, Zep, Letta, and others provide memory layers that work well for individual developers and small teams.

The unsolved problem is what happens at production scale.

At 10,000 memories, cosine similarity stops discriminating between relevant and irrelevant results. At 100,000 memories, contradictions accumulate silently — "Alice moved to London" and "Alice lives in Paris" coexist without detection. At enterprise scale, hardcoded lifecycle thresholds ("archive after 30 days") break because usage patterns vary across teams, projects, and domains.

And there is a regulatory dimension. The EU AI Act takes full effect August 2, 2026. Every memory system that sends data to cloud LLMs for core operations faces a compliance question that engineering alone cannot resolve — it requires an architectural answer.

We spent the last year applying mathematics to these problems.

Three Mathematical Techniques — Each a First in Agent Memory

1. Fisher-Rao Geodesic Distance (Retrieval)

Standard memory systems use cosine similarity. Cosine treats every embedding as equally confident — a memory accessed once scores identically to one accessed a thousand times, if their directions match.

We model each memory embedding as a diagonal Gaussian distribution with learned mean and variance. The Fisher-Rao geodesic distance — the natural metric on statistical manifolds — measures similarity along the curved surface of the probability space, not through flat Euclidean space.

In practice: memories that have been accessed more become more precise. Variance shrinks with repeated access via Bayesian conjugate updates. The system provably improves at finding things the more you use it.

At scale, this matters. When 10,000 memories compete for relevance, confidence-weighted distance provides discriminative power that flat cosine cannot.

Ablation: Removing Fisher-Rao drops multi-hop accuracy by 12 percentage points.

2. Sheaf Cohomology (Consistency)

Pairwise contradiction checking is O(n²) and misses transitive contradictions. At enterprise scale, it is both too slow and too weak.

We model the knowledge graph as a cellular sheaf — an algebraic structure from topology that assigns vector spaces to nodes and edges. Computing the first cohomology group H¹(G,F) reveals global inconsistencies from local data:

H¹ = 0 → All memories are globally consistent
H¹ ≠ 0 → Contradictions exist, even if every local pair looks fine

This scales algebraically, not quadratically. And it catches contradictions that no pairwise method can detect.

3. Riemannian Langevin Dynamics (Lifecycle)

Memory lifecycle management in current systems means hardcoded thresholds: "archive after 30 days," "promote after 10 accesses." These thresholds are tuned for average workloads and fail on everything else.

We replace thresholds with stochastic gradient flow on the Poincaré ball. The potential function encodes access frequency, trust score, and recency. The dynamics provably converge to a stationary distribution — the mathematically optimal allocation of memories across lifecycle states (Active → Warm → Cold → Archived).

No manual tuning. The system self-organizes based on actual usage patterns.

Results

Evaluated on the LoCoMo benchmark (Long Conversation Memory):

Configuration	Score	What It Means
Mode A Retrieval	74.8%	Data stays on your machine. Highest local-first score.
Mode C (Full Power)	87.7%	Cloud LLM at every layer. Comparable to industry systems.
Mode A Raw	60.4%	No LLM at any stage. First in the field.

For context — the competitive landscape:

System	Score	Cloud LLM Required
EverMemOS	92.3%	Yes
MemMachine	91.7%	Yes
Hindsight	89.6%	Yes
SLM V3 Mode C	87.7%	Yes (every layer)
Zep	~85%	Yes
SLM V3 Mode A	74.8%	No
Mem0	~58-66%	Yes
SLM V3 Mode A Raw	60.4%	No (zero-LLM)

The gap between Mode A Raw (60.4%) and Mode A Retrieval (74.8%) demonstrates that the four-channel mathematical retrieval pipeline captures the vast majority of benchmark requirements without any cloud dependency. The remaining gap between 74.8% and 87.7% is answer synthesis quality — not knowledge retrieval.

Why This Matters for Production

Current memory systems work at development scale. The mathematical foundations in V3 address three production-scale problems:

Retrieval quality at scale. Fisher-Rao provides discriminative power that cosine similarity loses when thousands of memories compete for relevance.
Consistency at scale. Sheaf cohomology detects contradictions algebraically, not quadratically. As knowledge graphs grow, this becomes the difference between reliable and unreliable memory.
Lifecycle at scale. Langevin dynamics self-organize memory allocation based on actual usage — no manual threshold tuning that breaks when workloads change.

These are not theoretical advantages. They are measurable on the benchmark, and they become more pronounced as memory count grows.

Three Operating Modes

V3 offers a privacy-accuracy spectrum:

Mode A: Local Guardian — All processing local. No cloud calls. EU AI Act compliant by architecture. 74.8% on LoCoMo.

Mode B: Smart Local — Mode A + local LLM via Ollama. Still fully private. No data leaves your machine.

Mode C: Full Power — Cloud LLM at every layer. 87.7% on LoCoMo. This is the configuration comparable to other memory systems. Data leaves the machine for processing.

The choice is yours. Switch anytime. Your memories stay consistent across all modes.

Getting Started

npm install -g superlocalmemory
slm setup
slm warmup    # Optional: pre-download embedding model
slm dashboard # 17-tab web dashboard at localhost:8765

Works with 17+ AI tools: Claude Code, Cursor, VS Code Copilot, Windsurf, ChatGPT Desktop, Gemini CLI, JetBrains, Zed, Continue, Cody, and more.

What We Believe

Current memory systems are impressive engineering. Every system in the competitive table represents meaningful work solving real problems for real users.

Our contribution is mathematical. We believe the future of agent memory is not more heuristics, but principled mathematics — techniques that provide guarantees, scale predictably, and can be adopted by any system.

The three techniques in V3 (Fisher-Rao, sheaf cohomology, Langevin dynamics) are not specific to our product. They are mathematical tools. We open-sourced everything under MIT because we believe the entire field benefits from mathematical foundations.

If these techniques make other memory systems better, we have succeeded.

Paper: Zenodo DOI: 10.5281/zenodo.19038659
Code: github.com/qualixar/superlocalmemory
Website: superlocalmemory.com

Varun Pratap Bhardwaj — Independent Researcher

Part of Qualixar

Part of Qualixar | Author: Varun Pratap Bhardwaj

5 AI Agent Memory Systems Compared: Mem0, Zep, Letta, Supermemory, SuperLocalMemory (2026 Benchmark Data)

varun pratap Bhardwaj — Wed, 18 Mar 2026 03:38:44 +0000

A factual comparison of the five most-referenced AI agent memory systems on architecture, LoCoMo benchmark scores, and EU AI Act compliance.

Why This Post Exists

Every comparison post I've read either reads like marketing for one system, or compares them on features without benchmark data. This is different: I'm the author of one of the systems (SuperLocalMemory), which means I have strong incentive to be honest — I can't be credibly wrong about the others without undermining my own credibility.

All scores are from published papers or official documentation. I've noted where scores vary across sources.

The Five Systems

System	Architecture	Creator	License	Status
Mem0	Cloud-hosted	Mem0 AI ($24M funded)	Open core	Production
Zep	Cloud-hosted + self-host	Getzep	Apache 2.0 + Commercial	Production
Letta (MemGPT)	Agent framework + LLM memory	Letta AI	Apache 2.0	Production
Supermemory	Cloud-hosted	Open source project	MIT	Production
SuperLocalMemory	Local-first mathematical	Independent research	MIT	Production

LoCoMo Benchmark Results

The LoCoMo benchmark (Long Conversation Memory) is the most widely cited evaluation for this space — 81 question-answer pairs across long multi-session conversations.

System	Score	Cloud LLM Required	Open Source
EverMemOS	92.3%	Yes	No
MemMachine	91.7%	Yes	No
Hindsight	89.6%	Yes	No
SLM V3 Mode C	87.7%	Yes (synthesis)	Yes (MIT)
Zep	~85%	Yes	Partial
Letta / MemGPT	~83.2%	Yes	Yes (Apache)
SLM V3 Mode A	74.8%	No	Yes (MIT)
Supermemory	~70%*	Yes	Yes (MIT)
Mem0 (self-reported)	~66%	Yes	Partial
SLM V3 Zero-LLM	60.4%	No LLM at all	Yes (MIT)
Mem0 (independent)	~58%	Yes	Partial

*Supermemory score estimated from limited published data.

Key takeaway: Every system requiring cloud LLMs clusters between 83-92%. SuperLocalMemory Mode A achieves 74.8% with zero cloud dependency — demonstrating that mathematical retrieval captures most of the benchmark value without cloud compute. Mode C reaches 87.7%, competitive with the top tier.

Architecture Comparison

Mem0

Model: Cloud-first, API-based. Memories stored on Mem0's servers.
Retrieval: Vector similarity over cloud embeddings (typically OpenAI).
Best for: Teams needing shared memory, managed infrastructure, cross-device access.
Limitation: Data sovereignty, offline use, EU AI Act compliance require additional work.

Zep

Model: Temporal knowledge graph hosted in cloud (or self-hosted Community Edition).
Retrieval: Graph-based temporal reasoning + semantic similarity.
Best for: Complex agent workflows requiring temporal entity relationships.
Limitation: Self-hosting requires infrastructure management; cloud version has same data locality issues as Mem0.

Letta (MemGPT)

Model: OS-inspired agent framework. LLM manages memory tiers (core context, recall, archival).
Retrieval: LLM-driven — the model decides what to retrieve and when.
Best for: Building agents where memory management logic needs to be customizable by the LLM.
Limitation: Requires LLM for all memory operations. Memory decisions inherit LLM opacity.

Supermemory

Model: Cloud-hosted with importable sources (tweets, web pages, documents).
Retrieval: Vector similarity + semantic search.
Best for: Personal knowledge management with multi-source ingestion.
Limitation: Cloud dependency; primarily designed for personal knowledge, not agent memory.

SuperLocalMemory V3

Model: Local-first with three mathematical retrieval layers.
Retrieval: 4-channel RRF fusion: Fisher-Rao geometric + BM25 lexical + entity graph + temporal.
Best for: Privacy-required workloads, EU AI Act compliance, individual developer memory, zero-cloud operation.
Limitation: Single-device by default; no native team sharing.

EU AI Act Compliance (Takes Effect August 2, 2026)

This dimension is increasingly important for enterprise deployment in the EU.

System	Mode A Compliance	Notes
SuperLocalMemory Mode A	✅ By architecture	Data never leaves device. Zero cloud calls.
All others	❌ Requires work	DPA required. Data sent to cloud providers.

SuperLocalMemory is the only system in this table that claims compliance-by-architecture. All others can achieve compliance through additional legal and technical measures, but require active work.

The Right Tool for the Job

None of these systems is "best." The right choice depends on your requirements:

Need team memory? → Mem0 or Zep. Both are designed for shared memory.

Need LLM to manage memory logic? → Letta. It's designed for LLM-driven memory management.

Need data sovereignty or EU AI Act compliance? → SuperLocalMemory Mode A. Only local-first provides this by architecture.

Need the highest benchmark score? → None of the open systems. EverMemOS/MemMachine/Hindsight score higher, but aren't open source.

Need open source + high score? → SuperLocalMemory Mode C (87.7%) or Letta (~83.2%).

Need zero cloud costs forever? → SuperLocalMemory Mode A. No API costs, no subscription.

My System (Full Disclosure)

I'm the author of SuperLocalMemory V3. I've tried to be factually accurate about all five systems. If I've gotten something wrong, open an issue on the repo or comment below.

Paper: arXiv:2603.14588
Code: github.com/qualixar/superlocalmemory
Website: superlocalmemory.com

Varun Pratap Bhardwaj — Independent Researcher
A Qualixar Research Initiative

SuperLocalMemory vs Mem0: When Zero-Cloud Beats Managed Memory (Benchmark Analysis)

varun pratap Bhardwaj — Wed, 18 Mar 2026 03:36:28 +0000

How a local-first system with mathematical foundations scores 74.8% on LoCoMo — higher than Mem0 — without a single cloud API call.

The Setup

I've spent the last year building SuperLocalMemory V3 — an open-source AI agent memory system that uses information geometry instead of cloud LLMs for core operations. Before I talk about how we compare, I want to be clear: this is not a hit piece on Mem0. They've built something genuinely useful with a great team. This is a factual benchmark analysis for developers choosing an architecture.

The key question: does local-first memory with mathematical foundations perform better than cloud-hosted managed memory?

On LoCoMo (Long Conversation Memory benchmark), the answer is yes.

The Numbers

System	LoCoMo Score	Cloud LLM Required	Cost
SLM V3 Mode C	87.7%	Yes (synthesis only)	$0 base (your API key)
SLM V3 Mode A	74.8%	No	$0 forever
Mem0 (self-reported)	~66%	Yes	Subscription
Mem0 (independent)	~58%	Yes	Subscription
SLM V3 Zero-LLM	60.4%	No LLM at all	$0

The headline: SLM Mode A (local-only) scores 74.8% — higher than Mem0's best-reported 66% — with data never leaving your machine.

A note on Mem0 scores: they vary across reports. Their self-reported number is ~66%, but independent measurements are closer to 58%. We cite both. Our scores are from our paper: arXiv:2603.14588.

Why Does Local Beat Cloud Here?

Mem0's architecture is straightforward: you send memories to their cloud, they store them, you query via API. It works well for team collaboration and managed infrastructure.

The weakness is retrieval quality. Standard cloud memory systems use cosine similarity over vector embeddings. At thousands of memories, cosine similarity stops discriminating well — everything starts looking similar.

We replaced cosine similarity with three mathematical techniques:

1. Fisher-Rao Geodesic Distance
Instead of treating each memory embedding as a point, we model it as a Gaussian distribution with a mean and variance. The Fisher-Rao distance measures similarity on the curved probability manifold — not through flat Euclidean space.

The key insight: memories accessed more often become more precise (variance shrinks via Bayesian updates). So frequently-relevant memories get geometrically closer to matching queries. Cosine can't do this.

Ablation: removing Fisher-Rao drops multi-hop accuracy by 12 percentage points.

2. Sheaf Cohomology for Consistency
Pairwise contradiction checking is O(n²) and misses transitive contradictions. Sheaf cohomology computes H¹(G, F) — a global consistency measure from local checks. Algebraic, not quadratic. Catches contradictions no pairwise method can.

3. Riemannian Langevin Dynamics for Lifecycle
Instead of "archive after 30 days," we use stochastic gradient flow on the Poincaré ball. The system self-organizes lifecycle states based on actual usage patterns. No manual threshold tuning.

Architecture Differences

Dimension	SLM Mode A	Mem0
Data location	On your device	Mem0's cloud
Embedding	Local model (nomic-embed-text)	OpenAI API
Retrieval	4-channel mathematical	Vector similarity
Offline	Full	None
API key needed	No	Yes
EU AI Act	Compliant by architecture	Requires DPA
Team memory	Single-device default	Native

When Mem0 Is Better

Honesty matters here. Mem0 wins on:

Team collaboration: Native multi-user shared memory. SLM is single-device by default.
Managed infrastructure: No local model to run, no disk usage, no maintenance.
Cross-device access: Memory follows you across devices natively.

If you're building a team-facing product where multiple users share memory, Mem0 is probably a better fit today.

When SLM Is Better

SLM wins when:

Data sovereignty is required: EU AI Act compliance, HIPAA-adjacent data, air-gapped environments.
Individual developer workflow: Personal coding assistant memory — Claude Code, Cursor, etc.
Zero ongoing cost: Mode A is free forever. No API costs, no subscription.
Offline operation: Mode A/B work with no internet connection.
Explainable retrieval: Every retrieval decision shows 4-channel scores. No black-box LLM.

Installing Both (30 seconds each)

# SuperLocalMemory
npm install -g superlocalmemory
slm setup
slm remember "Factoring auth into shared middleware on the payments service"
slm recall "payments auth"

# Mem0 (Python)
pip install mem0ai
# Then: set OPENAI_API_KEY and call mem0.add(), mem0.search()

Conclusion

The benchmark data is clear: local-first memory with information-geometric foundations outperforms cloud-hosted vector similarity on LoCoMo — by 8+ percentage points in the direct comparison, while requiring zero cloud infrastructure.

This matters because the AI memory field is mostly converging on "use a cloud LLM for everything." We're demonstrating a different path: mathematics as the intelligence layer, cloud as an optional enhancement.

Everything is MIT licensed and open source: github.com/qualixar/superlocalmemory

Paper: arXiv:2603.14588

Varun Pratap Bhardwaj — Independent Researcher
A Qualixar Research Initiative

I Added Persistent Memory to Claude Code in 60 Seconds (and It Actually Works)

varun pratap Bhardwaj — Wed, 18 Mar 2026 03:34:25 +0000

Claude Code forgets everything between sessions. Here's how I fixed it with one command and a local database that never leaves my machine.

The Problem

If you use Claude Code daily, you've felt this: every session starts from zero. You re-explain your codebase architecture. You remind it which patterns you prefer. You tell it again that you use uv not pip. Again. Every. Single. Session.

Claude Code is exceptional at reasoning within a session. Across sessions, it has no memory at all.

I built SuperLocalMemory to fix this. Here's the 60-second setup.

Setup (One Command)

npm install -g superlocalmemory
slm setup

That's it. slm setup downloads the embedding model (~275MB, one-time), initializes the local database, and registers the MCP server. Everything runs on your machine — no API keys, no cloud, no accounts.

Connect to Claude Code

Add to your Claude Code MCP config (~/.claude/settings.json or via the UI):

{
  "mcpServers": {
    "superlocalmemory": {
      "command": "slm",
      "args": ["mcp"]
    }
  }
}

Restart Claude Code. You'll see SuperLocalMemory tools appear in the toolbar.

Start Using It

The workflow is simple: tell Claude what to remember, ask it to recall later.

You: slm remember "This project uses uv not pip. Always use uv run python."
Claude: ✓ Stored memory about package manager preference.

[Next session, next week]

You: What package manager does this project use?
Claude: [calls slm recall] Based on your stored preferences: this project uses uv, not pip.

Or use the CLI directly:

# Remember something
slm remember "Auth is handled in middleware/auth.py — JWT, 24h expiry"

# Recall anything
slm recall "where is auth handled"

# See all memories
slm list

# Open the 17-tab dashboard
slm dashboard

What Actually Gets Stored

Claude Code can call the memory tools autonomously as it works. Common patterns that work well:

Project context:

slm remember "Monorepo: packages/api (FastAPI), packages/web (Next.js 15), packages/worker (Celery)"
slm remember "Production: Railway (API + worker), Vercel (web), Upstash Redis"
slm remember "Database: Supabase Postgres. Migrations with Alembic."

Preferences:

slm remember "Always use type hints. Prefer dataclasses over plain dicts."
slm remember "Test with pytest. Use pytest-asyncio for async tests. 80% coverage minimum."
slm remember "No print statements in production code — use structlog."

Decisions:

slm remember "Decided against GraphQL — REST is simpler for this use case (March 2026)"
slm remember "Payment processing uses Stripe not Paddle — existing contracts"

Bugs and fixes:

slm remember "Fixed: celery worker crashes if Redis connection drops during task — added retry with exponential backoff"

The Dashboard

slm dashboard

Opens a 17-tab web UI at localhost:8765. The tabs I use most:

Memories — browse all stored facts, edit or delete inline
Recall Lab — test queries before using in code sessions
Knowledge Graph — visual map of entities and relationships
Trust Dashboard — Bayesian trust scores per memory source

How It Actually Works (For the Curious)

Standard memory systems use cosine similarity over embeddings — it works but degrades at scale. SuperLocalMemory uses three mathematical techniques from our research paper:

Fisher-Rao geodesic distance — models each memory as a Gaussian distribution, not a point. Frequently accessed memories become more precise (variance shrinks via Bayesian updates). The result: the system gets better at finding things the more you use it.

Sheaf cohomology — detects contradictory memories globally, not just pairwise. If you've stored conflicting facts ("Auth uses JWT" and "Auth uses sessions"), the system surfaces the conflict.

Langevin dynamics — self-organizes memory lifecycle based on actual usage. Frequently accessed memories stay active; stale ones archive automatically.

On the LoCoMo benchmark, this approach achieves 74.8% accuracy with data staying fully local — higher than Mem0's 58-66% while requiring zero cloud dependency.

Three Modes

By default you get Mode A (fully local, no cloud):

slm mode a   # Default — zero cloud, 74.8% on LoCoMo
slm mode b   # + local Ollama LLM for synthesis (still private)
slm mode c   # + cloud LLM for synthesis (87.7% on LoCoMo)

I use Mode A. It's fast (sub-millisecond retrieval), private, and works offline. I switch to Mode B when I want the Ollama-powered cluster summaries in the dashboard.

Works With More Than Claude Code

The same memory layer works with every AI tool that supports MCP:

Cursor — add the same MCP config to Cursor settings
VS Code Copilot — via Continue.dev extension
Windsurf, Zed, JetBrains AI — same config
ChatGPT Desktop — via MCP bridge

One memory store, all your tools.

MIT License, Fully Open Source

npm install -g superlocalmemory   # npm
pip install superlocalmemory      # Python

GitHub: github.com/qualixar/superlocalmemory
Website: superlocalmemory.com
Paper: arXiv:2603.14588

1400+ tests. No telemetry. No accounts. Data never leaves your machine in Mode A.

Varun Pratap Bhardwaj — Independent Researcher
A Qualixar Research Initiative

Agentic Engineering Patterns: Architectural Building Blocks for AI Agent Systems

varun pratap Bhardwaj — Fri, 06 Mar 2026 06:55:25 +0000

Building an AI agent is not the same as calling an LLM in a loop. The moment you need an agent to use tools, remember past interactions, revise its own plans, or collaborate with other agents, you enter the domain of systems architecture. The patterns you choose — how the agent reasons, when it retrieves context, how it delegates — determine whether your system is reliable or a stochastic mess. This post breaks down the core architectural patterns that have emerged in agentic AI engineering, explains when each one applies, and shows you how memory layers tie them all together.

What You Will Learn

The four foundational agentic patterns: ReAct, Plan-and-Execute, Reflection, and Delegation

How each pattern structures the loop between reasoning, action, and observation

Where memory (short-term, long-term, episodic) fits into each pattern

Concrete Python code implementing each pattern with persistent memory retrieval

Trade-offs and failure modes so you know when not to use a given pattern

Conceptual Foundation: What Makes a System "Agentic"

An LLM call is stateless. You send a prompt, you get a completion. An agent, by contrast, operates in a loop: it perceives its environment, decides on an action, executes that action, observes the result, and feeds that observation back into its next decision. This loop is the defining characteristic.

Three capabilities distinguish an agent from a simple chain:

Tool use — the agent can invoke external functions (search, databases, APIs, code execution).
State management — the agent maintains context across multiple steps, including across sessions.
Autonomous decision-making — the agent decides what to do next without human intervention at each step.

The patterns we will examine are different ways of structuring this loop. None of them is universally superior. Each makes a trade-off between autonomy, reliability, latency, and cost.

graph TD
    subgraph "Agentic Loop"
        A[User Query] --> B[Reasoning / Planning]
        B --> C{Select Action}
        C -->|Tool Call| D[Execute Tool]
        C -->|Respond| H[Final Answer]
        D --> E[Observation / Result]
        E --> F[Memory Write]
        F --> B
    end

    subgraph "Memory Layer"
        G[Short-Term Memory<br/>Current conversation] -.-> B
        I[Long-Term Memory<br/>Past sessions, facts] -.-> B
        J[Episodic Memory<br/>Past task outcomes] -.-> B
        F --> G
        F --> I
        F --> J
    end

    style B fill:#4a90d9,color:#fff
    style F fill:#d9a34a,color:#fff

This diagram shows the general shape. Every pattern we discuss is a specific instantiation of this loop with different control flow decisions at the "Reasoning / Planning" and "Select Action" nodes.

Pattern 1: ReAct (Reason + Act)

ReAct, introduced by Yao et al. (2023), interleaves reasoning traces with actions. At each step, the agent produces a Thought (natural language reasoning), then an Action (tool invocation), then receives an Observation (tool output). This cycle repeats until the agent has enough information to produce a final answer.

ReAct is the simplest agentic pattern and the one you should reach for first.

1. The agent receives a query and generates a Thought

The thought is an explicit reasoning trace: "I need to find the current stock price of AAPL. I should use the stock_price tool."

2. The agent selects and invokes a tool (Action)

Based on the thought, the agent emits a structured action: stock_price(symbol="AAPL").

3. The tool returns a result (Observation)

The tool returns {"price": 187.42, "currency": "USD"}. This is appended to the agent's context.

4. The loop repeats or the agent responds

The agent decides whether it has enough information. If yes, it produces a final answer. If not, it generates another Thought and continues.

Here is a minimal ReAct implementation:

import openai
import json

# Define available tools
TOOLS = {
    "search": lambda query: f"Results for '{query}': [Wikipedia article about {query}]",
    "calculate": lambda expr: str(eval(expr)),  # simplified; use a sandbox in production
}

TOOL_DESCRIPTIONS = """
Available tools:
- search(query: str) -> str: Search the web for information
- calculate(expr: str) -> str: Evaluate a math expression
"""

def react_agent(query: str, max_steps: int = 5) -> str:
    """A minimal ReAct agent loop."""
    messages = [
        {"role": "system", "content": f"""You are a ReAct agent. For each step:
1. Thought: reason about what to do next
2. Action: call a tool using JSON: {{"tool": "name", "args": {{"key": "value"}}}}
3. When ready, respond with: {{"answer": "your final answer"}}
{TOOL_DESCRIPTIONS}"""},
        {"role": "user", "content": query},
    ]

    for step in range(max_steps):
        response = openai.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            temperature=0,
        )
        content = response.choices[0].message.content
        messages.append({"role": "assistant", "content": content})

        # Try to parse the agent's output as JSON
        try:
            parsed = json.loads(content)
        except json.JSONDecodeError:
            # If it's not JSON, treat it as the final answer
            return content

        if "answer" in parsed:
            return parsed["answer"]

        if "tool" in parsed:
            tool_name = parsed["tool"]
            tool_args = parsed.get("args", {})
            # Execute the tool
            observation = TOOLS[tool_name](**tool_args)
            # Feed observation back into the loop
            messages.append({
                "role": "user",
                "content": f"Observation: {observation}"
            })

    return "Max steps reached without a final answer."

# Usage
result = react_agent("What is the population of France divided by 3?")
print(result)

When to use ReAct: When your task requires interleaving information gathering with reasoning. It works well for question answering, research tasks, and data lookups where the agent needs 2-5 tool calls.

When not to use it: When the task requires a long-horizon plan with 10+ steps. ReAct is greedy — it decides one step at a time, which can lead to wandering.

Pattern 2: Plan-and-Execute

Plan-and-Execute separates planning from execution. First, the agent creates an explicit multi-step plan. Then a separate execution loop carries out each step. After execution, the agent can optionally revise the plan based on what it learned.

This pattern is better suited for complex tasks because the planning step forces the agent to commit to a strategy before spending tokens and tool calls on execution.

import openai
import json

def plan_and_execute(query: str) -> str:
    """Plan-and-Execute pattern: create a plan, then execute each step."""

    # Phase 1: Generate a plan
    plan_response = openai.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": """Create a step-by-step plan to answer the user's query.
Return a JSON array of step strings. Example: ["Step 1: ...", "Step 2: ..."]
Each step should be a concrete, actionable instruction."""},
            {"role": "user", "content": query},
        ],
        temperature=0,
    )
    plan = json.loads(plan_response.choices[0].message.content)
    print(f"Plan: {plan}")

    # Phase 2: Execute each step
    context = ""  # Accumulates results from previous steps
    for i, step in enumerate(plan):
        exec_response = openai.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": f"""You are executing step {i+1} of a plan.
Previous context: {context}
Execute this step and return the result."""},
                {"role": "user", "content": step},
            ],
            temperature=0,
        )
        step_result = exec_response.choices[0].message.content
        context += f"\nStep {i+1} result: {step_result}"

    # Phase 3: Synthesize final answer
    final_response = openai.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "Synthesize a final answer from the execution results."},
            {"role": "user", "content": f"Original query: {query}\n\nExecution results:{context}"},
        ],
        temperature=0,
    )
    return final_response.choices[0].message.content

The key insight: by separating planning from execution, you can use different models for each phase (a stronger model for planning, a cheaper one for execution), and you can checkpoint and resume the plan.

Pattern 3: Reflection

Reflection adds a self-critique step. After the agent produces an output, a second pass evaluates that output for correctness, completeness, and adherence to instructions. If the evaluation fails, the agent revises its output.

This is not a standalone pattern — it layers on top of ReAct or Plan-and-Execute.

def reflect_and_revise(query: str, max_revisions: int = 2) -> str:
    """Generate an answer, then reflect on it and revise if needed."""

    # Initial generation
    draft = openai.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "Answer the user's question thoroughly."},
            {"role": "user", "content": query},
        ],
    ).choices[0].message.content

    for revision in range(max_revisions):
        # Reflection step: critique the draft
        critique = openai.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": """Critique the following answer.
Identify factual errors, missing information, or logical gaps.
If the answer is satisfactory, respond with exactly: APPROVED
Otherwise, list the specific issues."""},
                {"role": "user", "content": f"Query: {query}\n\nAnswer: {draft}"},
            ],
        ).choices[0].message.content

        if "APPROVED" in critique:
            return draft

        # Revision step: fix the issues
        draft = openai.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": "Revise the answer based on the critique."},
                {"role": "user", "content": f"Original query: {query}\n\nDraft: {draft}\n\nCritique: {critique}"},
            ],
        ).choices[0].message.content

    return draft

Reflection Can Be Wasteful

Reflection doubles (or triples) your LLM calls. Do not apply it to every agent response. Reserve it for high-stakes outputs: generated code that will be executed, answers to complex multi-step questions, or content that will be published. For simple lookups, reflection adds cost without meaningful quality improvement.

Pattern 4: Delegation (Multi-Agent Coordination)

Delegation splits a complex task across specialized agents. A supervisor agent breaks the task into subtasks and routes each to a specialist agent — a coder, a researcher, a data analyst. Each specialist has its own tools, system prompt, and potentially its own memory context.

def supervisor_agent(query: str) -> str:
    """A supervisor that delegates to specialist agents."""

    # Decide which specialists to invoke
    routing = openai.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": """You are a supervisor agent.
Available specialists: ["researcher", "coder", "analyst"]
Given a query, return a JSON plan:
[{"agent": "researcher", "task": "..."}, {"agent": "coder", "task": "..."}]"""},
            {"role": "user", "content": query},
        ],
    ).choices[0].message.content

    subtasks = json.loads(routing)
    results = {}

    for subtask in subtasks:
        agent_name = subtask["agent"]
        task = subtask["task"]
        # Each specialist has a different system prompt and tool set
        specialist_result = run_specialist(agent_name, task, context=results)
        results[agent_name] = specialist_result

    # Synthesize results
    synthesis = openai.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "Combine the specialist results into a final answer."},
            {"role": "user", "content": f"Query: {query}\nResults: {json.dumps(results)}"},
        ],
    ).choices[0].message.content
    return synthesis

def run_specialist(name: str, task: str, context: dict) -> str:
    """Run a specialist agent with its own system prompt."""
    prompts = {
        "researcher": "You are a research agent. Find factual information.",
        "coder": "You are a coding agent. Write correct, tested code.",
        "analyst": "You are a data analyst. Interpret data and produce insights.",
    }
    response = openai.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": prompts[name]},
            {"role": "user", "content": f"Task: {task}\nContext from other agents: {json.dumps(context)}"},
        ],
    )
    return response.choices[0].message.content

The hard part of delegation is not the routing — it is shared state. When the coder agent needs context from the researcher agent, how does it get it? Passing everything in the prompt works at small scale but breaks down quickly. This is where memory becomes critical.

How Memory Ties These Patterns Together

Every pattern above has a shared weakness: context management. ReAct accumulates observations in its message history. Plan-and-Execute passes results between steps as text. Delegation passes context between agents as JSON blobs. None of these approaches scale beyond a single session.

Memory solves this by providing a persistent, queryable store that any agent (or any step within an agent) can read from and write to.

There are three memory layers:

Memory Type	Scope	Lifetime	Example
Short-term	Current task/session	Minutes to hours	Conversation history, intermediate results
Long-term	Cross-session	Days to permanent	User preferences, learned facts, past decisions
Episodic	Per-task	Permanent	"Last time I tried approach X, it failed because Y"

Here is how memory integrates into a ReAct agent:

import numpy as np

class AgentMemory:
    """A simple vector-based memory store for agent state."""

    def __init__(self):
        self.entries = []  # List of {"text": str, "embedding": list, "metadata": dict}

    def store(self, text: str, metadata: dict = None):
        """Store a memory entry with its embedding."""
        embedding = get_embedding(text)  # Your embedding function
        self.entries.append({
            "text": text,
            "embedding": embedding,
            "metadata": metadata or {},
        })

    def retrieve(self, query: str, top_k: int = 3) -> list[str]:
        """Retrieve the most relevant memories for a query."""
        query_embedding = get_embedding(query)
        scored = []
        for entry in self.entries:
            # Cosine similarity
            sim = np.dot(query_embedding, entry["embedding"]) / (
                np.linalg.norm(query_embedding) * np.linalg.norm(entry["embedding"])
            )
            scored.append((sim, entry["text"]))
        scored.sort(reverse=True, key=lambda x: x[0])
        return [text for _, text in scored[:top_k]]

def react_agent_with_memory(query: str, memory: AgentMemory, max_steps: int = 5) -> str:
    """ReAct agent augmented with persistent memory retrieval."""

    # Retrieve relevant past memories before starting
    relevant_memories = memory.retrieve(query, top_k=3)
    memory_context = "\n".join(f"- {m}" for m in relevant_memories) if relevant_memories else "None"

    messages = [
        {"role": "system", "content": f"""You are a ReAct agent with access to memory.
Relevant memories from past sessions:
{memory_context}

Use these memories to avoid repeating past mistakes and to build on prior knowledge.
Follow the Thought -> Action -> Observation loop."""},
        {"role": "user", "content": query},
    ]

    # ... standard ReAct loop from earlier ...
    final_answer = run_react_loop(messages, max_steps)

    # Store the outcome as episodic memory
    memory.store(
        f"Task: {query} | Outcome: {final_answer}",
        metadata={"type": "episodic", "task": query}
    )

    return final_answer

The critical detail: memory retrieval happens before the agent starts reasoning, and memory storage happens after the agent finishes. This creates a learning loop where each task execution improves future performance.

Seeing This in Practice

Multi-agent delegation introduces a harder memory problem: trust scoring. When Agent B retrieves a memory that Agent A wrote, how much should it trust that memory? If Agent A's task failed, its stored observations might be misleading.

SuperLocalMemory implements a local agent memory layer with hybrid search (combining vector similarity and keyword matching) that addresses this. It exposes a straightforward API for storing memories with metadata — including agent identity and task outcomes — and retrieving them with configurable scoring:

from superlocalmemory import MemoryStore

store = MemoryStore(path="./agent_memories")

# Agent A stores a research finding
store.add(
    text="The API rate limit for service X is 100 requests/minute as of March 2026.",
    metadata={
        "agent": "researcher",
        "task_id": "task-42",
        "task_outcome": "success",
        "confidence": 0.95,
    }
)

# Agent B (coder) retrieves relevant context, filtered by trust
results = store.search(
    query="rate limits for service X",
    top_k=5,
    filters={"task_outcome": "success"},  # Only trust successful task memories
)

for result in results:
    print(f"[{result.metadata['agent']}] {result.text} (score: {result.score:.3f})")

The hybrid search combines dense vector retrieval with sparse keyword matching, which matters in agentic contexts where queries often contain specific identifiers (API names, error codes) that pure semantic search can miss. You can inspect the full implementation in the GitHub repository to see how the scoring and filtering work under the hood.

Real-World Considerations

The Abstraction Trap

A recurring concern in the developer community — highlighted in discussions like "The Abstraction Trap: Why Layers Are Lobotomizing Your Model" — is that adding too many layers between the LLM and the task degrades performance. Every abstraction layer (planner, reflector, memory retrieval, routing) adds latency and potential error. Start with the simplest pattern (ReAct) and add complexity only when you have evidence that it helps.

Cost. A single ReAct loop with 4 steps costs 4 LLM calls. Add reflection and that doubles to 8. Add a planner and you are at 9+. Delegation multiplies this by the number of agents. Profile your token usage early.

Debugging. Agentic systems are hard to debug because the LLM's reasoning is non-deterministic. Log every step: the full prompt, the model's response, the tool inputs and outputs, and the memory retrievals. Without these logs, you are flying blind.

Failure modes by pattern:

ReAct: gets stuck in loops, calls the same tool repeatedly with slightly different arguments
Plan-and-Execute: creates plans that are too rigid or too vague; early step failures cascade
Reflection: the critic always finds something to complain about, causing infinite revision loops (always cap revision count)
Delegation: specialists produce incompatible outputs; the supervisor cannot reconcile them

Tool Execution Safety

If your agent can execute code or write to databases, sandbox it. A ReAct agent calling eval() on untrusted expressions — as in our simplified example above — is a remote code execution vulnerability. Use containers, restricted interpreters (like asteval), or separate execution environments with strict timeouts and resource limits.

Choosing the Right Pattern

Pattern	Best For	Steps	LLM Calls	Complexity
ReAct	Simple tool-use, Q&A	2-5	Low	Low
Plan-and-Execute	Multi-step tasks, research	5-15	Medium	Medium
Reflection	High-stakes outputs	+1-2 per cycle	Medium-High	Medium
Delegation	Complex tasks needing specialization	Varies	High	High

A practical heuristic: start with ReAct. If you find the agent wandering or failing on tasks that require more than 5 steps, move to Plan-and-Execute. If output quality matters more than speed, add Reflection. If the task genuinely requires different expertise domains, use Delegation.

These patterns also compose. A delegation supervisor can use Plan-and-Execute for routing, while each specialist uses ReAct internally, and the final synthesis uses Reflection. The architecture is modular by design.

Universal Memory Layer Architecture for AI Agents

varun pratap Bhardwaj — Thu, 05 Mar 2026 06:57:50 +0000

Most AI agents today are stateless. They receive a prompt, generate a response, and forget everything. If you have built anything with an LLM, you have felt this limitation firsthand: your agent cannot recall what happened two conversations ago, cannot learn from its mistakes, and cannot share context with other agents in your system. The context window is not memory. It is a scratchpad that gets wiped clean. To build agents that genuinely improve over time, you need a dedicated memory layer — one that persists knowledge, organizes it for fast retrieval, and works across multiple agents without coupling them together.

This post walks you through the architecture of such a memory layer from first principles. No handwaving, no black boxes. By the end, you will understand the design decisions behind memory persistence for AI agents and have runnable code you can adapt for your own systems.

What You Will Learn

The difference between episodic and semantic memory in agent systems, and when to use each.

How to design an interoperable memory schema that works across multi-agent architectures.

How hybrid search (vector + keyword + graph) enables efficient memory recall at scale.

Concrete implementation patterns in Python with working code examples.

Trade-offs and failure modes you will encounter in production.

Why Agents Need a Memory Layer

An LLM's context window is finite. GPT-4 Turbo offers 128K tokens; Claude gives you 200K. That sounds like a lot until your agent has been running for days, processing hundreds of tasks, accumulating observations, and collaborating with other agents. You cannot stuff all of that into a single prompt.

More importantly, a context window is ephemeral. When the session ends, the context is gone. Your agent starts from zero next time. This is the equivalent of a developer who loses all their notes every time they close their laptop.

A memory layer solves this by acting as persistent, queryable storage that sits outside the LLM. The agent writes to it during execution and reads from it when it needs context. This is not a new idea — cognitive architectures like SOAR and ACT-R modeled human memory as distinct subsystems decades ago. What is new is applying these patterns to LLM-based agents at scale.

As Krishnan (2025) notes in AI Agents: Evolution, Architecture, and Real-World Applications, modern agent architectures integrate large language models with dedicated modules for perception, planning, and tool use. Memory is the connective tissue between those modules.

Conceptual Foundation: Types of Agent Memory

Human cognitive science distinguishes between several types of memory. Two are particularly useful for agent design: episodic memory and semantic memory.

Episodic memory stores specific events and experiences. For an agent, this means: "At 2:30 PM, I called the weather API and got a timeout error." Episodic memories are timestamped, ordered, and tied to a specific context.

Semantic memory stores general knowledge and facts distilled from experience. For an agent, this means: "The weather API tends to time out during peak hours; use the cached endpoint instead." Semantic memories are abstracted, deduplicated, and context-independent.

There is a third type worth mentioning: procedural memory, which stores how to do things. In agent systems, this maps to learned tool-use patterns, prompt templates that worked well, or refined chain-of-thought strategies. We will not cover procedural memory in depth here, but know that it exists in the design space.

Property	Episodic Memory	Semantic Memory
Content	Specific events, observations	Distilled facts, generalizations
Structure	Timestamped, sequential	Key-value, graph-linked
Retention	Can be pruned over time	Long-lived, updated in place
Retrieval	By time range, similarity to current context	By concept, relationship, keyword
Example	"User asked about refund policy on March 3"	"Refund policy requires order ID and proof of purchase"

A well-designed memory layer supports both types. Episodic memory gives your agent a detailed history to reason over. Semantic memory gives it compressed, reliable knowledge to act on quickly.

Architecture Overview

Here is the core architecture for a universal memory layer. Study the flow from agent action through to retrieval.

graph TD
    A["Agent Action / Observation"] --> B["Episodic Buffer"]
    B --> C["Memory Processor"]
    C --> D["Long-Term Memory Store"]
    D --> E["Vector Index"]
    D --> F["Keyword Index"]
    D --> G["Graph Index"]
    H["Agent Context Window"] -->|"retrieval query"| I["Hybrid Retrieval Engine"]
    E --> I
    F --> I
    G --> I
    I -->|"ranked memories"| H
    C -->|"semantic extraction"| J["Semantic Memory Store"]
    J --> G
    J --> E

Let us walk through each component.

1. Agent Action Produces an Observation

Every time your agent does something — calls a tool, receives user input, generates a response — it produces an observation. This is the raw material for memory. The observation includes the content itself, a timestamp, the agent's identifier, and any metadata (tool name, user ID, task ID).

2. Episodic Buffer Stages the Memory

Before writing to long-term storage, observations land in an episodic buffer. This is a short-lived queue that allows batching, deduplication, and importance scoring. Not every observation deserves to be a long-term memory. The buffer gives you a place to apply filters.

3. Memory Processor Writes to Long-Term Storage

The processor takes buffered observations, generates embeddings (for vector search), extracts entities and relationships (for graph search), and indexes keywords (for BM25/keyword search). It also runs a semantic extraction step: distilling episodic memories into semantic memories when patterns emerge.

4. Hybrid Retrieval Fetches Relevant Memories

When the agent needs context, it sends a retrieval query to the hybrid retrieval engine. This engine queries all three indexes — vector, keyword, and graph — then merges and re-ranks the results. The top-ranked memories are injected into the agent's context window as part of the system prompt or as retrieved context.

Designing an Interoperable Memory Schema

If you are building a multi-agent system, your memory schema needs to work across agents that may have different roles, different tools, and different LLM backends. This means the schema must be self-describing and loosely coupled.

Here is a practical schema in JSON that covers both episodic and semantic memories:

from dataclasses import dataclass, field
from typing import Optional
from datetime import datetime
import uuid
import json

@dataclass
class MemoryRecord:
    """Universal memory record usable across any agent in the system."""
    content: str                          # The actual memory content
    memory_type: str                      # "episodic" or "semantic"
    agent_id: str                         # Which agent created this memory
    timestamp: str = field(
        default_factory=lambda: datetime.utcnow().isoformat()
    )
    memory_id: str = field(
        default_factory=lambda: str(uuid.uuid4())
    )
    embedding: Optional[list[float]] = None  # Vector embedding, set during processing
    entities: list[str] = field(default_factory=list)  # Extracted entities for graph index
    relationships: list[dict] = field(default_factory=list)  # Entity-to-entity links
    metadata: dict = field(default_factory=dict)  # Flexible metadata (tool, task_id, etc.)
    trust_score: float = 1.0              # How much other agents should trust this memory
    access_count: int = 0                 # How often this memory has been retrieved
    ttl: Optional[int] = None             # Time-to-live in seconds; None = permanent

    def to_dict(self) -> dict:
        return {k: v for k, v in self.__dict__.items() if v is not None}

# Example: creating an episodic memory
episode = MemoryRecord(
    content="Called /api/weather for zip 94103. Received 504 Gateway Timeout after 30s.",
    memory_type="episodic",
    agent_id="weather-agent-01",
    entities=["weather_api", "zip_94103"],
    relationships=[{"from": "weather_api", "relation": "timed_out_for", "to": "zip_94103"}],
    metadata={"tool": "http_get", "status_code": 504, "task_id": "task-abc-123"}
)

print(json.dumps(episode.to_dict(), indent=2))

Expected output:

{
  "content": "Called /api/weather for zip 94103. Received 504 Gateway Timeout after 30s.",
  "memory_type": "episodic",
  "agent_id": "weather-agent-01",
  "timestamp": "2026-03-05T14:22:01.123456",
  "memory_id": "a1b2c3d4-...",
  "entities": ["weather_api", "zip_94103"],
  "relationships": [{"from": "weather_api", "relation": "timed_out_for", "to": "zip_94103"}],
  "metadata": {"tool": "http_get", "status_code": 504, "task_id": "task-abc-123"},
  "trust_score": 1.0,
  "access_count": 0
}

The trust_score field is critical in multi-agent systems. When Agent B reads a memory written by Agent A, it needs to assess reliability. Trust scores can be updated based on whether memories led to successful outcomes. The access_count field enables least-recently-used pruning for storage management.

Schema Versioning Matters

Once multiple agents share a memory store, changing the schema becomes a coordination problem. Include a schema_version field in your metadata from day one. Migrations in multi-agent systems are significantly harder than in single-service architectures because you cannot take the memory layer offline for a migration while agents are running.

Hybrid Search: Vector + Keyword + Graph

No single search method handles all memory retrieval scenarios well. Here is why you need all three.

Vector search (using embeddings) excels at semantic similarity. "The API returned an error" will match "HTTP request failed" even though they share no keywords. But vector search struggles with precise lookups: searching for "task-abc-123" by embedding similarity is unreliable.

Keyword search (BM25 or similar) excels at exact and partial matches. Searching for a specific task ID, agent name, or error code is fast and deterministic. But keyword search misses semantic relationships entirely.

Graph search excels at relationship traversal. "What do we know about all entities connected to the weather API?" is a graph query. Neither vector nor keyword search can answer this efficiently.

Here is a practical implementation of hybrid retrieval using Python:

from dataclasses import dataclass
import numpy as np
from typing import Optional

@dataclass
class SearchResult:
    memory_id: str
    content: str
    score: float
    source: str  # "vector", "keyword", or "graph"

def cosine_similarity(a: list[float], b: list[float]) -> float:
    """Compute cosine similarity between two vectors."""
    a_arr, b_arr = np.array(a), np.array(b)
    return float(np.dot(a_arr, b_arr) / (np.linalg.norm(a_arr) * np.linalg.norm(b_arr) + 1e-10))

def vector_search(
    query_embedding: list[float],
    memory_store: list[dict],
    top_k: int = 5
) -> list[SearchResult]:
    """Search memories by embedding similarity."""
    scored = []
    for mem in memory_store:
        if mem.get("embedding"):
            sim = cosine_similarity(query_embedding, mem["embedding"])
            scored.append(SearchResult(
                memory_id=mem["memory_id"],
                content=mem["content"],
                score=sim,
                source="vector"
            ))
    scored.sort(key=lambda r: r.score, reverse=True)
    return scored[:top_k]

def keyword_search(
    query_terms: list[str],
    memory_store: list[dict],
    top_k: int = 5
) -> list[SearchResult]:
    """Simple BM25-style keyword matching."""
    scored = []
    for mem in memory_store:
        content_lower = mem["content"].lower()
        # Count how many query terms appear in the content
        hits = sum(1 for term in query_terms if term.lower() in content_lower)
        if hits > 0:
            # Normalize by number of query terms
            score = hits / len(query_terms)
            scored.append(SearchResult(
                memory_id=mem["memory_id"],
                content=mem["content"],
                score=score,
                source="keyword"
            ))
    scored.sort(key=lambda r: r.score, reverse=True)
    return scored[:top_k]

def graph_search(
    entity: str,
    memory_store: list[dict],
    max_hops: int = 2
) -> list[SearchResult]:
    """Find memories connected to a given entity through relationships."""
    results = []
    visited_entities = {entity}
    frontier = [entity]

    for hop in range(max_hops):
        next_frontier = []
        for mem in memory_store:
            for rel in mem.get("relationships", []):
                if rel["from"] in frontier or rel["to"] in frontier:
                    results.append(SearchResult(
                        memory_id=mem["memory_id"],
                        content=mem["content"],
                        score=1.0 / (hop + 1),  # Closer hops score higher
                        source="graph"
                    ))
                    # Add connected entities to next hop
                    for key in ["from", "to"]:
                        if rel[key] not in visited_entities:
                            next_frontier.append(rel[key])
                            visited_entities.add(rel[key])
        frontier = next_frontier
    return results

def hybrid_search(
    query_embedding: list[float],
    query_terms: list[str],
    entity: Optional[str],
    memory_store: list[dict],
    weights: dict = None,
    top_k: int = 5
) -> list[SearchResult]:
    """
    Merge results from all three search methods using weighted reciprocal rank fusion.
    """
    if weights is None:
        weights = {"vector": 0.5, "keyword": 0.3, "graph": 0.2}

    all_results: dict[str, float] = {}  # memory_id -> fused score
    result_content: dict[str, str] = {}

    for search_fn, args, source_key in [
        (vector_search, (query_embedding, memory_store), "vector"),
        (keyword_search, (query_terms, memory_store), "keyword"),
    ]:
        results = search_fn(*args)
        for rank, result in enumerate(results):
            # Reciprocal rank fusion: 1/(rank+1) * weight
            rrf_score = (1.0 / (rank + 1)) * weights[source_key]
            all_results[result.memory_id] = all_results.get(result.memory_id, 0) + rrf_score
            result_content[result.memory_id] = result.content

    if entity:
        graph_results = graph_search(entity, memory_store)
        for rank, result in enumerate(graph_results):
            rrf_score = (1.0 / (rank + 1)) * weights["graph"]
            all_results[result.memory_id] = all_results.get(result.memory_id, 0) + rrf_score
            result_content[result.memory_id] = result.content

    # Sort by fused score
    sorted_ids = sorted(all_results.keys(), key=lambda mid: all_results[mid], reverse=True)
    return [
        SearchResult(
            memory_id=mid,
            content=result_content[mid],
            score=all_results[mid],
            source="hybrid"
        )
        for mid in sorted_ids[:top_k]
    ]

The hybrid_search function uses reciprocal rank fusion (RRF) to combine results. RRF is simple and effective: it assigns each result a score of 1/(rank+1), weighted by the search method's importance. Memories that appear in multiple search results get boosted naturally. This approach outperforms naive score averaging because scores from different search methods are not on the same scale.

Tuning Hybrid Search Weights

The default weights (0.5 vector, 0.3 keyword, 0.2 graph) are a reasonable starting point. In practice, tune these based on your retrieval evaluation set. If your agent frequently needs exact ID lookups, increase the keyword weight. If your domain is heavily relational (e.g., knowledge graphs, org charts), increase the graph weight.

Semantic Extraction: Turning Episodes into Knowledge

Episodic memories accumulate fast. If your agent runs 1,000 tasks per day, you will have 1,000+ episodic records within hours. Retrieval degrades as the store grows. Semantic extraction addresses this by periodically distilling episodic memories into compact semantic memories.

Here is a simplified extraction pipeline:

def extract_semantic_memories(
    episodic_memories: list[dict],
    similarity_threshold: float = 0.85
) -> list[dict]:
    """
    Group similar episodic memories and distill them into semantic memories.
    In production, you would use an LLM to generate the summary.
    """
    clusters: list[list[dict]] = []

    for mem in episodic_memories:
        placed = False
        for cluster in clusters:
            # Compare against the first memory in each cluster (simplified)
            sim = cosine_similarity(mem["embedding"], cluster[0]["embedding"])
            if sim >= similarity_threshold:
                cluster.append(mem)
                placed = True
                break
        if not placed:
            clusters.append([mem])

    semantic_memories = []
    for cluster in clusters:
        if len(cluster) >= 3:  # Only distill if we have enough evidence
            # In production, pass cluster contents to an LLM for summarization
            combined_content = " | ".join(m["content"] for m in cluster)
            semantic_memories.append({
                "content": f"[Distilled from {len(cluster)} episodes] {combined_content[:200]}...",
                "memory_type": "semantic",
                "agent_id": "memory-processor",
                "source_count": len(cluster),
                "source_ids": [m["memory_id"] for m in cluster],
                "trust_score": min(m.get("trust_score", 1.0) for m in cluster),
            })

    return semantic_memories

In a production system, you would replace the simple concatenation with an LLM call that generates a proper summary. The key design choice is the similarity_threshold: too low and you merge unrelated memories; too high and nothing gets distilled.

Do Not Delete Episodic Memories After Extraction

It is tempting to delete episodic memories once they have been distilled into semantic memories. Do not do this in your first iteration. Semantic extraction is lossy — the summary may miss critical details. Keep episodic memories with a TTL (e.g., 30 days) so you can fall back to them during retrieval. Archive rather than delete.

Real-World Considerations

Storage Costs and Pruning

Embeddings are not small. A 1536-dimensional float32 embedding consumes about 6 KB. At 10,000 memories, that is 60 MB of embeddings alone. At 10 million, it is 60 GB. Plan your pruning strategy early: TTL-based expiry for episodic memories, access-count-based eviction for infrequently-used semantic memories, and compression for archived records.

Latency Budgets

Your agent is waiting for memories before it can generate a response. If hybrid retrieval takes 500ms, that 500ms is added to every agent turn. For real-time applications, consider a two-tier cache: a fast in-memory cache of recently accessed memories (hits in under 5ms) backed by the full indexed store (hits in 50-200ms).

Multi-Agent Trust and Conflict Resolution

When Agent A writes "the refund policy requires a receipt" and Agent B writes "the refund policy does not require a receipt," your memory layer has a conflict. Trust scores help here — you can weight memories by the trust score of the authoring agent — but they do not eliminate the problem. Consider adding a conflict detection step during retrieval that flags contradictory memories for the consuming agent to resolve.

When Not to Use a Memory Layer

Not every agent needs persistent memory. If your agent handles isolated, stateless tasks (e.g., a code formatter, a one-shot classifier), the overhead of a memory layer adds complexity without benefit. The memory layer pays off when agents run over extended periods, collaborate with other agents, or need to improve based on past experience.

Seeing This in Practice

The patterns described above — interoperable memory schemas, trust scoring across agents, and hybrid retrieval from a shared memory store — are implemented in SuperLocalMemory, an open-source memory layer that runs entirely on your local machine with no cloud dependency.

Its multi-agent shared memory architecture assigns trust scores to memories based on the authoring agent's track record and uses a schema compatible with 16+ tools including Claude and Cursor. You can inspect how the memory write/read flow works by cloning the repository:

git clone https://github.com/varun369/SuperLocalMemoryV2.git
cd SuperLocalMemoryV2
# Examine the memory schema and retrieval logic
cat src/memory/schema.py
cat src/memory/retrieval.py

This gives you a concrete reference implementation to compare against the architectural patterns discussed here. Reviewing working code is often more instructive than reading about patterns in the abstract.

Building a Universal Memory Layer for AI Agents: Architecture and Patterns

varun pratap Bhardwaj — Wed, 04 Mar 2026 06:53:58 +0000

AI agents without memory are stateless functions. They process a prompt, return a response, and forget everything. That works for one-shot tasks, but it fails the moment an agent needs to recall a previous conversation, learn from a mistake, or coordinate with another agent. The moment you need continuity — across sessions, across tools, across a team of agents — you need a memory layer.

This post teaches you how to build one. Not a toy demo, but an architecture you can reason about and extend. We will cover how memory gets stored, how it gets retrieved (and why naive vector search is not enough), and how multiple agents can share and trust each other's memories. Every pattern comes with runnable code.

What You Will Learn

The three types of agent memory (episodic, semantic, procedural) and when to use each

How to implement hybrid retrieval combining semantic search and BM25 keyword search

How Reciprocal Rank Fusion merges results from different retrieval strategies

State management patterns for multi-agent systems with shared memory

Trust scoring: how agents decide which memories to rely on

Practical Python code for each component

Conceptual Foundation: What Is Agent Memory?

Human memory is not a single system. You have short-term working memory (what you are thinking about right now), long-term episodic memory (what happened at lunch yesterday), and procedural memory (how to ride a bike). Agent memory works the same way — different types serve different purposes.

Three Types of Agent Memory

Episodic memory stores specific interactions. "The user asked me to summarize Q3 revenue on Tuesday. I retrieved data from the finance API and the user corrected my interpretation of 'net revenue'." This is the agent's autobiography — a timestamped log of what happened, what it did, and what feedback it received.

Semantic memory stores facts and knowledge extracted from interactions. "Net revenue for this company means revenue after returns and discounts, not after all expenses." These are distilled truths the agent can reference without replaying entire episodes.

Procedural memory stores learned behaviors and strategies. "When the user asks about revenue, always clarify whether they mean gross or net before querying." These are the agent's acquired skills.

Most memory layer implementations today conflate these three types into a single vector store. That works at small scale, but it creates retrieval noise as the memory grows. A well-designed memory layer distinguishes between them.

graph TD
    A[Agent Interaction] --> B{Memory Classification}
    B -->|Raw interaction log| C[Episodic Memory Store]
    B -->|Extracted facts| D[Semantic Memory Store]
    B -->|Learned strategies| E[Procedural Memory Store]

    C --> F[Embedding + BM25 Index]
    D --> F
    E --> F

    F --> G[Hybrid Retrieval Engine]
    G --> H[Reciprocal Rank Fusion]
    H --> I[Ranked Memory Results]
    I --> J[Agent Context Window]

How It Works: The Memory Write/Read Pipeline

Every memory layer has two fundamental operations: writing memories and reading them back. The architecture of each determines everything about your system's quality.

Writing Memories: Ingestion Pipeline

When an agent completes an interaction, the memory layer needs to process and store that interaction. This is not as simple as appending text to a database.

1. Capture the Raw Interaction

The agent sends the full interaction context — the user's input, the agent's reasoning trace, tool calls made, the final response, and any feedback received. Store this as structured data, not a flat string.

from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional

@dataclass
class MemoryRecord:
    content: str
    memory_type: str  # "episodic", "semantic", "procedural"
    agent_id: str
    session_id: str
    timestamp: datetime = field(default_factory=datetime.utcnow)
    metadata: dict = field(default_factory=dict)
    trust_score: float = 1.0  # 0.0 to 1.0
    source_agent: Optional[str] = None
    embedding: Optional[list[float]] = None

2. Generate Embeddings

Convert the text content into a dense vector representation. This enables semantic search — finding memories that are conceptually similar even when they use different words.

import numpy as np
from openai import OpenAI

client = OpenAI()

def generate_embedding(text: str, model: str = "text-embedding-3-small") -> list[float]:
    """Generate a dense vector embedding for the given text."""
    response = client.embeddings.create(input=text, model=model)
    return response.data[0].embedding

# Example usage
record = MemoryRecord(
    content="User prefers revenue figures in EUR, not USD.",
    memory_type="semantic",
    agent_id="finance-agent-01",
    session_id="sess-abc-123",
    metadata={"domain": "finance", "confidence": 0.95}
)
record.embedding = generate_embedding(record.content)
# Embedding is a 1536-dimensional float vector for text-embedding-3-small
print(f"Embedding dimensions: {len(record.embedding)}")
# Output: Embedding dimensions: 1536

3. Build the BM25 Index

Embeddings capture meaning, but they can miss exact keyword matches. BM25 (Best Matching 25) is a term-frequency ranking function that excels at finding documents containing specific terms. You need both.

import math
from collections import Counter

class BM25Index:
    """Simple BM25 index for keyword-based retrieval."""

    def __init__(self, k1: float = 1.5, b: float = 0.75):
        self.k1 = k1
        self.b = b
        self.docs: list[dict] = []      # [{id, tokens}]
        self.avg_dl: float = 0.0        # average document length
        self.doc_freqs: dict = {}        # term -> number of docs containing term
        self.n_docs: int = 0

    def add_document(self, doc_id: str, text: str):
        tokens = text.lower().split()
        self.docs.append({"id": doc_id, "tokens": tokens})
        self.n_docs += 1

        # Update document frequencies
        unique_terms = set(tokens)
        for term in unique_terms:
            self.doc_freqs[term] = self.doc_freqs.get(term, 0) + 1

        # Recalculate average document length
        self.avg_dl = sum(len(d["tokens"]) for d in self.docs) / self.n_docs

    def score(self, query: str) -> list[tuple[str, float]]:
        """Return (doc_id, score) pairs sorted by BM25 relevance."""
        query_tokens = query.lower().split()
        scores = []

        for doc in self.docs:
            doc_score = 0.0
            doc_len = len(doc["tokens"])
            term_counts = Counter(doc["tokens"])

            for term in query_tokens:
                if term not in self.doc_freqs:
                    continue

                df = self.doc_freqs[term]
                # Inverse document frequency
                idf = math.log((self.n_docs - df + 0.5) / (df + 0.5) + 1.0)

                tf = term_counts.get(term, 0)
                # BM25 term frequency normalization
                numerator = tf * (self.k1 + 1)
                denominator = tf + self.k1 * (1 - self.b + self.b * doc_len / self.avg_dl)

                doc_score += idf * (numerator / denominator)

            scores.append((doc["id"], doc_score))

        return sorted(scores, key=lambda x: x[1], reverse=True)

4. Persist to Storage

Store the record with its embedding in a vector database and index the text in BM25. In production, you would use something like PostgreSQL with pgvector, Qdrant, Weaviate, or ChromaDB for vector storage.

import json

class MemoryStore:
    """Unified memory store with both vector and BM25 indexing."""

    def __init__(self):
        self.records: dict[str, MemoryRecord] = {}
        self.bm25_index = BM25Index()
        self.embeddings: dict[str, list[float]] = {}  # doc_id -> embedding

    def write(self, record: MemoryRecord) -> str:
        doc_id = f"{record.agent_id}:{record.timestamp.isoformat()}"

        # Store the record
        self.records[doc_id] = record

        # Index for BM25 keyword search
        self.bm25_index.add_document(doc_id, record.content)

        # Store embedding for vector search
        if record.embedding is None:
            record.embedding = generate_embedding(record.content)
        self.embeddings[doc_id] = record.embedding

        return doc_id

Reading Memories: Hybrid Retrieval

Here is where most implementations get it wrong. They use only vector search. Pure vector search retrieves memories that are semantically similar, but it can miss results that contain exact terms the query specifies. Pure BM25 finds keyword matches but misses conceptually related memories.

The solution is hybrid retrieval: run both searches, then merge the results.

Why Pure Vector Search Is Not Enough

Suppose your agent stores the memory: "Customer account ID is CX-7742-B." A later query for "CX-7742-B" will likely fail with pure semantic search because the embedding of an alphanumeric ID carries almost no semantic meaning. BM25 handles this trivially because it matches the exact token. Always combine both retrieval strategies.

Reciprocal Rank Fusion (RRF)

Reciprocal Rank Fusion is a simple, effective algorithm for merging ranked lists from different retrieval methods. For each document, its RRF score is calculated as:

RRF(d) = sum over all rankers of 1 / (k + rank(d))

Where k is a constant (typically 60) that dampens the influence of high-ranking outliers.

def reciprocal_rank_fusion(
    ranked_lists: list[list[tuple[str, float]]],
    k: int = 60,
    top_n: int = 10
) -> list[tuple[str, float]]:
    """
    Merge multiple ranked result lists using Reciprocal Rank Fusion.

    Args:
        ranked_lists: List of [(doc_id, score)] lists, each sorted by relevance
        k: Damping constant (default 60, from the original Cormack et al. paper)
        top_n: Number of results to return

    Returns:
        Merged [(doc_id, rrf_score)] list sorted by fused score
    """
    rrf_scores: dict[str, float] = {}

    for ranked_list in ranked_lists:
        for rank, (doc_id, _original_score) in enumerate(ranked_list, start=1):
            if doc_id not in rrf_scores:
                rrf_scores[doc_id] = 0.0
            rrf_scores[doc_id] += 1.0 / (k + rank)

    # Sort by fused score descending
    fused = sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True)
    return fused[:top_n]

Now let's wire it all together in the MemoryStore:

def cosine_similarity(a: list[float], b: list[float]) -> float:
    a_arr, b_arr = np.array(a), np.array(b)
    return float(np.dot(a_arr, b_arr) / (np.linalg.norm(a_arr) * np.linalg.norm(b_arr)))

class MemoryStore:
    # ... (previous methods remain)

    def search_vector(self, query_embedding: list[float], top_n: int = 20) -> list[tuple[str, float]]:
        """Brute-force cosine similarity search. Replace with ANN in production."""
        scores = []
        for doc_id, emb in self.embeddings.items():
            sim = cosine_similarity(query_embedding, emb)
            scores.append((doc_id, sim))
        return sorted(scores, key=lambda x: x[1], reverse=True)[:top_n]

    def search_hybrid(self, query: str, top_n: int = 10) -> list[tuple[str, float]]:
        """Run both BM25 and vector search, fuse with RRF."""
        query_embedding = generate_embedding(query)

        bm25_results = self.bm25_index.score(query)
        vector_results = self.search_vector(query_embedding, top_n=20)

        fused = reciprocal_rank_fusion(
            [bm25_results, vector_results],
            k=60,
            top_n=top_n
        )
        return fused

Feature	BM25 (Keyword)	Vector Search (Semantic)	Hybrid (RRF)
Exact term matching	Excellent	Poor	Excellent
Semantic similarity	None	Excellent	Excellent
Handles typos	Poor	Moderate	Moderate
Alphanumeric IDs	Excellent	Very poor	Excellent
Latency (10K docs)	~1ms	~5ms (brute force) / ~1ms (ANN)	~6ms combined
Index storage	Inverted index (~small)	Vectors (~6KB per doc at 1536d)	Both

Multi-Agent State Management

When multiple agents share a memory layer, two new problems emerge: state consistency and trust.

State Consistency

If Agent A writes a memory and Agent B reads it simultaneously, you need to decide on a consistency model. For most agent workloads, eventual consistency is fine — agents are not database transactions. But you need a clear ownership model.

@dataclass
class AgentMemoryNamespace:
    """Each agent gets its own namespace. Shared memories are explicitly published."""
    agent_id: str
    private_store: MemoryStore
    shared_store: MemoryStore  # reference to the shared memory pool

    def remember(self, content: str, memory_type: str, share: bool = False):
        record = MemoryRecord(
            content=content,
            memory_type=memory_type,
            agent_id=self.agent_id,
            session_id="current",
            source_agent=self.agent_id,
        )
        # Always write to private store
        self.private_store.write(record)

        # Optionally publish to shared store
        if share:
            self.shared_store.write(record)

    def recall(self, query: str, include_shared: bool = True, top_n: int = 5):
        """Search private memories, optionally include shared pool."""
        private_results = self.private_store.search_hybrid(query, top_n=top_n)

        if not include_shared:
            return private_results

        shared_results = self.shared_store.search_hybrid(query, top_n=top_n)

        # Fuse private and shared results, giving private a slight boost
        # by prepending them in the ranked list
        return reciprocal_rank_fusion(
            [private_results, shared_results],
            k=60,
            top_n=top_n
        )

Trust Scoring

Not all memories deserve equal weight. A memory written by a well-tested agent with human-confirmed feedback is more trustworthy than one written by an experimental agent's first run. Trust scoring assigns a confidence weight to each memory.

def compute_trust_score(record: MemoryRecord, agent_registry: dict) -> float:
    """
    Compute a trust score for a memory record based on:
    - Source agent's historical accuracy
    - Recency of the memory
    - Whether a human confirmed it
    - Number of times other agents corroborated it
    """
    base_score = agent_registry.get(record.agent_id, {}).get("accuracy", 0.5)

    # Recency decay: memories older than 30 days lose trust
    age_days = (datetime.utcnow() - record.timestamp).days
    recency_factor = max(0.5, 1.0 - (age_days / 365))  # floor at 0.5

    # Human confirmation boost
    human_confirmed = record.metadata.get("human_confirmed", False)
    confirmation_boost = 1.3 if human_confirmed else 1.0

    # Corroboration: how many other agents wrote similar memories
    corroboration_count = record.metadata.get("corroboration_count", 0)
    corroboration_factor = min(1.5, 1.0 + corroboration_count * 0.1)

    score = base_score * recency_factor * confirmation_boost * corroboration_factor
    return min(1.0, score)  # Cap at 1.0

Why Trust Scoring Matters for Multi-Agent Systems

In a system with 10+ agents, one misconfigured agent can pollute the shared memory pool with incorrect facts. Trust scoring acts as an immune system — it lets the memory layer deprioritize memories from unreliable sources without deleting them outright. This is especially important in domains like finance or healthcare where incorrect context leads to real harm.

Seeing This in Practice

The patterns described above — hybrid search combining vector and BM25 retrieval, Reciprocal Rank Fusion for result merging, and trust-scored multi-agent memory — are the building blocks of production agent memory systems. Memonto's agent memory system implements these patterns as a working reference. Their architecture exposes memory namespaces per agent, provides hybrid retrieval out of the box, and includes trust scoring for cross-agent memory sharing.

If you want to see how the ingestion-to-retrieval pipeline looks in a deployed system rather than in isolated code snippets, the Memonto GitHub repository is worth studying as a reference implementation of these concepts.

# Conceptual usage pattern (based on the architecture described above)
# This illustrates how a production memory layer surfaces relevant context

agent_memory = AgentMemoryNamespace(
    agent_id="research-agent",
    private_store=MemoryStore(),
    shared_store=shared_memory_pool  # Shared across all agents
)

# Agent stores a learned fact
agent_memory.remember(
    content="The client's fiscal year ends in March, not December.",
    memory_type="semantic",
    share=True  # Make available to other agents
)

# Later, any agent can retrieve it
results = agent_memory.recall(
    query="When does the client's fiscal year end?",
    include_shared=True
)
# Returns the stored fact ranked by hybrid retrieval score

Real-World Considerations

When NOT to Build a Memory Layer

Not every agent system needs persistent memory. If your agent handles fully self-contained tasks (e.g., "convert this CSV to JSON"), adding a memory layer adds complexity without benefit. Memory layers pay off when:

Agents interact with the same users or datasets repeatedly
Multiple agents need to coordinate and share context
The cost of re-deriving information is high (expensive API calls, slow computations)

Failure Modes to Watch For

Memory bloat. Without a pruning strategy, your memory store grows indefinitely. Implement TTLs (time-to-live) for episodic memories and periodic deduplication for semantic memories.

Stale memories. Facts change. The client's preferred currency might switch from EUR to GBP. Your memory layer needs an update-or-supersede mechanism, not just append.

Context window overflow. Retrieving 50 relevant memories and stuffing them all into the agent's prompt defeats the purpose. Rank aggressively and inject only the top 3-5 most relevant memories.

Security: Memory Injection Attacks

If an agent's memory can be written to by external input (e.g., user messages get stored as semantic memories), an adversary can inject false facts: "The admin password is 'letmein'." Always sanitize memories before storage, and never store user input as trusted procedural memory without explicit human review.

Choosing Your Storage Backend

Backend	Vector Search	BM25/Keyword	Multi-tenancy	Operational Complexity
PostgreSQL + pgvector	Good (IVFFlat, HNSW)	via `tsvector`	Native (schemas/RLS)	Low (if you already run Postgres)
Qdrant	Excellent (HNSW)	Built-in sparse vectors	Collection-level	Medium (separate service)
Weaviate	Excellent	Built-in BM25	Native	Medium
ChromaDB	Good (HNSW)	Limited	Limited	Low (embedded mode)
SQLite + sqlite-vec	Adequate for <100K docs	via FTS5	Manual	Very low

For prototyping, start with ChromaDB or SQLite. For production multi-agent systems, PostgreSQL with pgvector gives you the best balance of features and operational simplicity — you get vector search, full-text search via tsvector, row-level security for multi-agent namespaces, and ACID transactions, all in a database you probably already run.

Building a Universal Memory Layer for AI Agents: Architecture Patterns and Implementation

varun pratap Bhardwaj — Tue, 03 Mar 2026 06:57:44 +0000

AI agents are stateless by default. Every time you invoke an LLM, it has no recollection of what it did five minutes ago unless you explicitly provide that context. This is fine for single-turn interactions, but it falls apart the moment you need agents that learn from past tasks, coordinate with other agents, or accumulate knowledge over time. The problem compounds in multi-agent systems where Agent A's discoveries need to be accessible to Agent B without piping everything through a shared prompt.

A universal memory layer solves this by abstracting persistent storage, retrieval, and state management behind a single interface that any agent — regardless of the underlying LLM provider — can read from and write to. This post teaches you how to build one.

What You Will Learn

Why AI agents need a dedicated memory layer separate from the LLM context window

How to design a memory storage schema that supports episodic, semantic, and procedural memory types

Three retrieval strategies (semantic search, keyword search, hybrid ranking) and when to use each

How to manage shared state across multi-agent systems with trust scoring and conflict resolution

Runnable Python code for a minimal but functional universal memory layer

Why Agents Need a Memory Layer

An LLM's context window is not memory. It is a scratchpad. When a context window fills up, older content gets truncated or summarized, and information is permanently lost. Even with 128k or 200k token windows, you cannot fit weeks of accumulated agent interactions into a single prompt.

Consider an agent that triages customer support tickets. On day one, it learns that "Project Atlas" is an internal code name for a database migration. On day thirty, a new ticket references "Atlas" without explanation. Without persistent memory, the agent has to re-learn this every session or rely on an engineer to hardcode it into the system prompt.

A memory layer provides three capabilities that context windows cannot:

Persistence — information survives beyond a single session or context window.
Selective retrieval — the agent fetches only the memories relevant to the current task instead of stuffing everything into the prompt.
Shared access — multiple agents can read and write to the same memory store, enabling coordination without direct message passing.

As Nowaczyk argues in Architectures for Building Agentic AI, reliability in agentic systems is "chiefly an architectural property" that emerges from principled componentization. Memory is one of those core components — alongside planners, tool routers, and safety monitors — that must be explicitly designed rather than bolted on.

Conceptual Foundation: Types of Agent Memory

Before writing any code, you need a mental model for what "memory" means in the context of AI agents. Cognitive science gives us a useful taxonomy that maps surprisingly well to software architecture.

Memory Type	Cognitive Analogy	Agent Example	Storage Pattern
Episodic	Personal experiences	"Last time I called the weather API, it returned a 429 error"	Timestamped event log
Semantic	General knowledge	"The capital of France is Paris"	Key-value or document store
Procedural	Skills and how-to	"To deploy to staging, run `make deploy-staging`"	Structured instructions
Working	Short-term scratch	Current task context and intermediate results	In-memory / context window

A universal memory layer must handle the first three types. Working memory is typically managed within the agent's execution loop and the LLM context window itself — it does not need to persist.

The critical insight is that these memory types have different access patterns. Episodic memory is usually queried by time range or similarity to a current situation. Semantic memory is queried by topic or concept. Procedural memory is queried by task type. Your storage schema and retrieval strategy must account for these differences.

Architecture: How the Memory Layer Works

Here is the high-level architecture. An agent sends a query to the memory layer, which fans out across multiple storage backends, merges the results using hybrid ranking, and returns a unified set of relevant memories.

graph TD
    A["Agent Request<br/>(query + metadata)"] --> B["Memory Layer API"]
    B --> C["Embedding Service"]
    B --> D["Query Parser"]
    C --> E["Vector Store<br/>(Semantic Search)"]
    D --> F["Keyword Index<br/>(BM25 / Full-Text)"]
    D --> G["Knowledge Graph<br/>(Entity Relationships)"]
    E --> H["Hybrid Ranker<br/>(RRF / Weighted Fusion)"]
    F --> H
    G --> H
    H --> I["Ranked Memory Results"]
    I --> A

    style B fill:#4a90d9,stroke:#2c5f8a,color:#fff
    style H fill:#d4944a,stroke:#8a5f2c,color:#fff

Let us walk through each component.

1. Memory Layer API

This is the single interface every agent interacts with. It exposes two primary operations: store(memory) and retrieve(query, filters). By standardizing this interface, you decouple agents from storage implementation details. An agent using OpenAI's GPT-4 and an agent using Claude both call the same API.

The API also enforces a common memory schema — every memory object carries metadata like agent_id, timestamp, memory_type, and trust_score alongside the content itself.

2. Parallel Retrieval Backends

When a retrieval request comes in, the memory layer queries multiple backends in parallel:

Vector Store — converts the query into an embedding and performs approximate nearest neighbor (ANN) search. This excels at finding semantically similar memories even when the exact words differ.
Keyword Index — performs BM25 or full-text search. This catches exact matches that semantic search might rank lower, such as specific error codes, proper nouns, or configuration values.
Knowledge Graph (optional) — traverses entity relationships. If the query mentions "Project Atlas," the graph can surface all memories connected to Atlas entities — team members, related services, deployment dates.

3. Hybrid Ranking and Fusion

Each backend returns a ranked list of candidates. The hybrid ranker merges these lists into a single ordering. The most common approach is Reciprocal Rank Fusion (RRF), which we will implement below. RRF does not require score normalization across backends — it only needs the rank positions, making it robust when combining search systems that produce incomparable score scales.

Practical Implementation

Let us build a minimal but functional memory layer in Python. We will use SQLite for metadata storage, numpy for vector operations, and a simple BM25 implementation for keyword search.

Memory Schema

First, define the data model:

import uuid
import time
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class Memory:
    content: str
    memory_type: str  # "episodic", "semantic", "procedural"
    agent_id: str
    metadata: dict = field(default_factory=dict)
    memory_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    timestamp: float = field(default_factory=time.time)
    trust_score: float = 1.0  # 0.0 to 1.0, explained below
    embedding: Optional[list] = None

The trust_score field deserves explanation. In multi-agent systems, not all memories are equally reliable. An agent that hallucinates or operates on stale data should have its memories weighted lower during retrieval. Trust scores provide a mechanism for this — they can be updated based on verification outcomes, agent reputation, or human feedback.

Storage Layer

import sqlite3
import json
import numpy as np

class MemoryStore:
    def __init__(self, db_path: str = ":memory:"):
        self.conn = sqlite3.connect(db_path)
        self._create_tables()
        self._memories_vectors = {}  # memory_id -> np.array

    def _create_tables(self):
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS memories (
                memory_id TEXT PRIMARY KEY,
                content TEXT NOT NULL,
                memory_type TEXT NOT NULL,
                agent_id TEXT NOT NULL,
                timestamp REAL NOT NULL,
                trust_score REAL DEFAULT 1.0,
                metadata TEXT DEFAULT '{}'
            )
        """)
        # Full-text search index for keyword retrieval
        self.conn.execute("""
            CREATE VIRTUAL TABLE IF NOT EXISTS memories_fts
            USING fts5(memory_id, content)
        """)
        self.conn.commit()

    def store(self, memory: Memory):
        self.conn.execute(
            "INSERT INTO memories VALUES (?, ?, ?, ?, ?, ?, ?)",
            (memory.memory_id, memory.content, memory.memory_type,
             memory.agent_id, memory.timestamp, memory.trust_score,
             json.dumps(memory.metadata))
        )
        self.conn.execute(
            "INSERT INTO memories_fts VALUES (?, ?)",
            (memory.memory_id, memory.content)
        )
        self.conn.commit()

        if memory.embedding is not None:
            self._memories_vectors[memory.memory_id] = np.array(
                memory.embedding, dtype=np.float32
            )

SQLite FTS5 Is Not Production-Grade Vector Search

This implementation uses SQLite's FTS5 for keyword search and an in-memory dictionary for vectors. This works for learning and prototyping. For production systems, replace the vector store with a dedicated ANN index like FAISS, pgvector, Qdrant, or Weaviate. The keyword index can stay as FTS5 for moderate-scale workloads or move to Elasticsearch/Tantivy for larger datasets.

Retrieval: Semantic Search

Semantic search computes cosine similarity between the query embedding and all stored memory embeddings. For small datasets (under 100k memories), brute-force search is fast enough. Beyond that, use an ANN index.

def retrieve_semantic(
    self, query_embedding: list, top_k: int = 10
) -> list[tuple[str, float]]:
    """Returns list of (memory_id, similarity_score) tuples."""
    query_vec = np.array(query_embedding, dtype=np.float32)
    # Normalize for cosine similarity
    query_norm = query_vec / (np.linalg.norm(query_vec) + 1e-10)

    scores = []
    for mid, vec in self._memories_vectors.items():
        vec_norm = vec / (np.linalg.norm(vec) + 1e-10)
        similarity = float(np.dot(query_norm, vec_norm))
        scores.append((mid, similarity))

    scores.sort(key=lambda x: x[1], reverse=True)
    return scores[:top_k]

Retrieval: Keyword Search (BM25 via FTS5)

SQLite's FTS5 provides a bm25() ranking function out of the box:

def retrieve_keyword(
    self, query: str, top_k: int = 10
) -> list[tuple[str, float]]:
    """Returns list of (memory_id, bm25_score) tuples."""
    cursor = self.conn.execute(
        """
        SELECT memory_id, bm25(memories_fts) as score
        FROM memories_fts
        WHERE memories_fts MATCH ?
        ORDER BY score
        LIMIT ?
        """,
        (query, top_k)
    )
    # FTS5 bm25() returns negative scores; lower = better match
    results = [(row[0], -row[1]) for row in cursor.fetchall()]
    return results

Hybrid Ranking with Reciprocal Rank Fusion

RRF merges ranked lists without needing to normalize scores across different retrieval systems. The formula for each document's fused score is:

RRF(d) = sum over all lists of 1 / (k + rank(d))

where k is a constant (typically 60) that dampens the impact of high-ranking positions.

def reciprocal_rank_fusion(
    *ranked_lists: list[tuple[str, float]],
    k: int = 60
) -> list[tuple[str, float]]:
    """
    Merges multiple ranked lists using RRF.

    Each ranked_list is [(memory_id, score), ...] sorted by relevance.
    Returns a unified ranked list of (memory_id, rrf_score).
    """
    rrf_scores = {}

    for ranked_list in ranked_lists:
        for rank, (memory_id, _score) in enumerate(ranked_list):
            if memory_id not in rrf_scores:
                rrf_scores[memory_id] = 0.0
            # rank is 0-indexed, so rank + 1 gives 1-indexed position
            rrf_scores[memory_id] += 1.0 / (k + rank + 1)

    fused = sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True)
    return fused

Let us see the full retrieval pipeline:

def retrieve(
    self, query: str, query_embedding: list, top_k: int = 5
) -> list[Memory]:
    """
    Hybrid retrieval: combines semantic and keyword search via RRF.
    """
    semantic_results = self.retrieve_semantic(query_embedding, top_k=20)
    keyword_results = self.retrieve_keyword(query, top_k=20)

    fused = reciprocal_rank_fusion(semantic_results, keyword_results)

    # Fetch full memory objects for top-k results
    memories = []
    for memory_id, rrf_score in fused[:top_k]:
        cursor = self.conn.execute(
            "SELECT * FROM memories WHERE memory_id = ?", (memory_id,)
        )
        row = cursor.fetchone()
        if row:
            mem = Memory(
                memory_id=row[0], content=row[1], memory_type=row[2],
                agent_id=row[3], timestamp=row[4], trust_score=row[5],
                metadata=json.loads(row[6])
            )
            memories.append(mem)

    return memories

Putting It Together

# Example usage
store = MemoryStore(db_path="agent_memory.db")

# Simulate storing a memory with a pre-computed embedding
# In practice, you'd call an embedding API (OpenAI, Cohere, etc.)
fake_embedding = np.random.randn(384).tolist()  # 384-dim for all-MiniLM-L6-v2

mem = Memory(
    content="Project Atlas refers to the Q3 database migration from PostgreSQL to CockroachDB.",
    memory_type="semantic",
    agent_id="support-triage-agent",
    embedding=fake_embedding,
    metadata={"source": "engineering-slack", "verified": True}
)
store.store(mem)

# Retrieve later
query = "What is Project Atlas?"
query_emb = np.random.randn(384).tolist()  # Would be real embedding in practice
results = store.retrieve(query, query_emb, top_k=3)

for r in results:
    print(f"[{r.memory_type}] {r.content[:80]}... (trust: {r.trust_score})")

Expected output (content will match since we only stored one memory):

[semantic] Project Atlas refers to the Q3 database migration from PostgreSQL to C... (trust: 1.0)

Multi-Agent State Management

When multiple agents share a memory layer, you face two challenges: conflicts and trust.

Conflict Resolution

Two agents might store contradictory memories. Agent A writes "deployment window is 2-4 AM UTC" and Agent B writes "deployment window is 3-5 AM UTC." A naive last-write-wins policy is dangerous.

Instead, use a versioned approach:

@dataclass
class VersionedMemory(Memory):
    version: int = 1
    parent_id: Optional[str] = None  # Points to the memory this updates
    superseded: bool = False  # True if a newer version exists

When an agent updates existing knowledge, it creates a new VersionedMemory with parent_id pointing to the original. The retrieval layer can then present the latest version while maintaining a full audit trail.

Trust Scoring

Trust scores allow the system to weight memories by reliability. Here is a straightforward scoring model:

def compute_trust_score(memory: Memory, store: MemoryStore) -> float:
    score = 0.5  # Base score

    # Verified by a human or authoritative source
    if memory.metadata.get("verified"):
        score += 0.3

    # Recency boost: memories less than 7 days old get a bump
    age_days = (time.time() - memory.timestamp) / 86400
    if age_days < 7:
        score += 0.1

    # Corroboration: other agents stored similar information
    # (simplified — in practice, check semantic similarity)
    cursor = store.conn.execute(
        "SELECT COUNT(*) FROM memories WHERE agent_id != ? AND content LIKE ?",
        (memory.agent_id, f"%{memory.content[:50]}%")
    )
    corroboration_count = cursor.fetchone()[0]
    if corroboration_count > 0:
        score += min(0.1 * corroboration_count, 0.2)  # Cap at 0.2

    return min(score, 1.0)

Trust Is Not Binary

A memory with a trust score of 0.3 is not "untrustworthy" — it is unverified. The retrieval layer should still return low-trust memories but can flag them for the consuming agent. This mirrors how humans treat information: you might act on a rumor while still seeking confirmation.

Seeing This in Practice

The patterns described above — hybrid retrieval, trust scoring, and shared memory across different AI providers — are implemented in SuperLocalMemory, an open-source project that runs these operations entirely on your local machine with zero cloud dependency.

The architecture maps directly to what we covered. Memories are stored with per-agent trust scores. Retrieval uses a hybrid approach combining semantic embeddings with keyword matching. Multiple agents — whether backed by OpenAI, Claude, or Gemini — read from and write to the same local memory store through a unified API.

You can inspect the implementation to see how these patterns translate to a working system. The repository is a useful reference if you are building your own memory layer and want to see how trust propagation, memory versioning, and multi-provider support interact in practice.

Real-World Considerations

When Not to Use a Universal Memory Layer

Not every agent system needs one. If your agent is stateless by design (a single-turn classifier, a one-shot code generator), adding persistent memory introduces complexity without benefit. The maintenance cost of a memory layer — schema migrations, index tuning, garbage collection of stale memories — is only justified when agents genuinely need to learn and coordinate over time.

Storage Scaling

The in-memory vector approach shown above works for up to roughly 100k memories with 384-dimensional embeddings (about 150MB of RAM). Beyond that, you need a proper ANN index. FAISS with an IVF index reduces search from O(n) to approximately O(sqrt(n)) at the cost of some recall. For most agent workloads, 95% recall at 10x speed is an acceptable tradeoff.

Embedding Model Choice

Your embedding model determines the quality of semantic retrieval. Smaller models like all-MiniLM-L6-v2 (384 dimensions, 80MB) are fast and good enough for many use cases. Larger models like OpenAI's text-embedding-3-large (3072 dimensions) provide better recall but increase storage and latency proportionally.

Never Mix Embedding Models in the Same Vector Space

If you compute memory embeddings with all-MiniLM-L6-v2 and query embeddings with text-embedding-3-small, the similarity scores will be meaningless. Every vector in a given store must be produced by the same model. If you need to change models, you must re-embed all existing memories.

Memory Garbage Collection

Without cleanup, a memory store grows indefinitely. Implement time-based expiration for episodic memories (do you really need a record of every API call from six months ago?) and access-frequency-based pruning for semantic memories. A simple heuristic: if a memory has not been retrieved in 90 days and its trust score is below 0.5, archive or delete it.

Separation of Planning and Execution: The Key Pattern for Reliable AI Coding Agents

varun pratap Bhardwaj — Mon, 23 Feb 2026 07:09:04 +0000

If you have spent any time using an AI coding agent, you have probably experienced this: you ask it to refactor a module, and it immediately starts editing files, halfway through realizes it needs a different approach, backtracks, leaves orphaned imports, and produces something that half-works. The agent was not stupid. It was doing two fundamentally different cognitive tasks at the same time — figuring out what to do and doing it — and that interleaving is where things fall apart.

The fix is an architectural pattern borrowed from classical AI planning: separate the planning phase from the execution phase. Generate a complete plan first — a spec, a file map, test criteria, an ordering of changes — and only then execute against that plan. This single structural change transforms flaky, meandering agent sessions into predictable, reviewable workflows.

What You Will Learn

Why interleaving planning and execution causes most AI coding agent failures

The two-phase workflow pattern: what it looks like in practice

How to write prompts that produce structured, actionable plans

How to implement a feedback loop between execution and the original plan

Concrete code examples you can adapt for your own agent pipelines

When this pattern helps and when it adds unnecessary overhead

Why Agents Fail: The Interleaving Problem

When you ask an LLM to "add authentication to this Express app," the model faces two distinct problems. First, it needs to reason about the task: which files need to change, what dependencies are required, what the middleware chain should look like, how existing routes will be affected. Second, it needs to produce code: actual syntax, correct imports, proper error handling.

These are different kinds of work. Planning is divergent — it explores possibilities, considers constraints, and makes trade-offs. Execution is convergent — it commits to specific syntax and produces concrete artifacts. When an LLM tries to do both simultaneously, it makes execution decisions before planning is complete. It starts writing a JWT middleware before deciding whether sessions or tokens are the right approach. Then it has to backtrack, and backtracking inside a code generation context means lost coherence.

This is not just a theory. If you audit failed agent sessions, you will find a consistent pattern: the agent made a reasonable local decision early on that turned out to be globally wrong. It edited file A in a way that is incompatible with the changes file B needs, because it had not thought through the full dependency graph before starting.

Classical AI planning research formalized this problem decades ago. The HDDL 2.1 formalism for Hierarchical Task Networks, for example, explicitly separates task decomposition (planning) from action execution, because interleaving them in complex domains leads to invalid plans. The same principle applies to LLM-based coding agents.

The Two-Phase Pattern: Plan, Then Execute

The core idea is straightforward: split every agent task into two distinct phases with a clear boundary between them.

Phase 1 — Planning: The LLM analyzes the task, examines relevant code, and produces a structured plan. This plan includes what files will be created or modified, what the expected behavior change is, what tests should pass afterward, and in what order changes should be applied. No code is written yet.

Phase 2 — Execution: A separate agent call (or series of calls) takes the plan as input and implements each step. The execution agent can reference the plan to maintain consistency. After each step, validation checks confirm the execution matches the plan.

Here is the workflow visualized:

graph TD
    A["Task Description"] --> B["Phase 1: Planning Agent"]
    B --> C["Structured Plan"]
    C --> D{"Human Review<br/>(optional)"}
    D -->|Approved| E["Phase 2: Execution Agent"]
    D -->|Revise| B
    E --> F["Step 1: Implement Change"]
    F --> G{"Validate Against Plan"}
    G -->|Pass| H["Step 2: Next Change"]
    G -->|Fail| I["Re-attempt with<br/>Plan Context"]
    I --> G
    H --> J["Step N: Final Change"]
    J --> K{"Run Test Criteria<br/>from Plan"}
    K -->|Pass| L["Done"]
    K -->|Fail| M["Diagnose Against Plan"]
    M --> E

The boundary between phases is the structured plan — a document that both humans and the execution agent can read, review, and reference. This is where the pattern gets its reliability: you can inspect the plan before any code changes happen.

The Plan Is Not a Suggestion

A common mistake is generating a vague, prose-heavy plan and then letting the execution agent freestyle. The plan must be specific enough that you could hand it to a different developer (or a different LLM) and get the same result. If your plan says "update the authentication logic," it is too vague. If it says "add a verifyToken middleware in src/middleware/auth.ts that decodes a JWT from the Authorization header and attaches the payload to req.user," that is actionable.

Designing the Planning Phase

The planning phase prompt needs to guide the LLM toward producing structured, actionable output. Here is a concrete prompt template you can adapt:

PLANNING_PROMPT = """
You are a senior software architect. Your job is to produce a detailed
implementation plan. Do NOT write any code yet.

## Task
{task_description}

## Current Codebase Context
{relevant_file_contents}

## Instructions
Analyze the task and produce a plan with the following sections:

### 1. Summary
One paragraph describing what this change accomplishes.

### 2. Files to Modify
For each file:
- File path
- What changes are needed (be specific about functions, classes, exports)
- Dependencies on other file changes (ordering constraints)

### 3. New Files to Create
For each new file:
- File path
- Purpose
- Key exports or interfaces

### 4. Dependency Changes
- New packages to install (with versions if relevant)
- Configuration changes (tsconfig, eslint, etc.)

### 5. Execution Order
Numbered list of steps, ordered so that each step can be validated
independently. Earlier steps should not depend on later ones.

### 6. Test Criteria
Specific, verifiable conditions that confirm the change is correct:
- Unit tests to add or modify
- Manual verification steps
- Edge cases to handle

### 7. Risks and Assumptions
- What could go wrong
- What assumptions you are making about the codebase
"""

Let us walk through using this with a real task.

1. Gather Context

Before sending the planning prompt, collect the files the LLM will need to reason about. This means reading the relevant source files, the project's package.json or equivalent, and any existing tests. Do not dump the entire codebase — scope it to what is relevant.

import os

def gather_context(file_paths: list[str]) -> str:
    """Read and concatenate file contents with clear delimiters."""
    context_parts = []
    for path in file_paths:
        if os.path.exists(path):
            with open(path, 'r') as f:
                content = f.read()
            context_parts.append(f"### {path}\n```
{% endraw %}
\n{content}\n
{% raw %}
```")
    return "\n\n".join(context_parts)

# Example: gathering context for an auth feature
context = gather_context([
    "src/app.ts",
    "src/routes/users.ts",
    "src/middleware/index.ts",
    "package.json",
    "tsconfig.json"
])

2. Generate the Plan

Send the planning prompt to your LLM. Use a high-reasoning model for this phase — planning benefits more from careful thinking than from fast token generation.

import openai

def generate_plan(task: str, context: str) -> str:
    """Generate an implementation plan without writing any code."""
    client = openai.OpenAI()

    prompt = PLANNING_PROMPT.format(
        task_description=task,
        relevant_file_contents=context
    )

    response = client.chat.completions.create(
        model="o3",  # Use a reasoning-capable model for planning
        messages=[
            {"role": "system", "content": "You are a software architect. Produce plans, not code."},
            {"role": "user", "content": prompt}
        ],
        temperature=1  # For reasoning models, temperature is typically fixed
    )

    return response.choices[0].message.content

plan = generate_plan(
    task="Add JWT-based authentication middleware to the Express app. "
         "Protect all /api/* routes except /api/auth/login and /api/auth/register.",
    context=context
)

print(plan)

3. Review the Plan

This is the step most people skip, and it is the most valuable. Read the plan. Does the execution order make sense? Are there missing files? Does the LLM assume something about your codebase that is wrong? Catching a mistake here costs you 30 seconds. Catching it after execution costs you a debugging session.

def save_plan_for_review(plan: str, output_path: str = "plan.md"):
    """Save the plan to a file for human review before execution."""
    with open(output_path, 'w') as f:
        f.write(plan)
    print(f"Plan saved to {output_path}. Review before proceeding.")

save_plan_for_review(plan)

Designing the Execution Phase

Once the plan is approved, the execution agent takes over. The key design decision here is granularity: execute one plan step at a time, validate, then proceed. Do not send the entire plan and ask the LLM to implement everything at once — that reintroduces the interleaving problem.

import json
import subprocess

EXECUTION_PROMPT = """
You are a senior developer implementing a specific step from an approved plan.

## Full Plan (for context — do NOT implement other steps)
{full_plan}

## Current Step to Implement
Step {step_number}: {step_description}

## Current File Contents
{current_file_contents}

## Instructions
Implement ONLY this step. Return the complete updated file contents.
Do not skip to later steps. Do not modify files not mentioned in this step.
"""

def execute_step(
    plan: str,
    step_number: int,
    step_description: str,
    file_paths: list[str]
) -> str:
    """Execute a single step from the plan."""
    client = openai.OpenAI()

    context = gather_context(file_paths)

    prompt = EXECUTION_PROMPT.format(
        full_plan=plan,
        step_number=step_number,
        step_description=step_description,
        current_file_contents=context
    )

    response = client.chat.completions.create(
        model="gpt-4o",  # Fast model is fine for execution
        messages=[
            {"role": "system", "content": "You are a developer. Write clean, working code."},
            {"role": "user", "content": prompt}
        ],
        temperature=0  # Low temperature for deterministic code output
    )

    return response.choices[0].message.content

Notice that the execution prompt includes the full plan for context. This is critical — the execution agent needs to know the big picture to make locally correct decisions. But the instruction constrains it to only implement the current step.

The Validation Loop

After each step, run validation. This can be as simple as a syntax check or as thorough as running the test suite. The validation result feeds back into the execution agent if something failed.

def validate_step(step_number: int, validation_commands: list[str]) -> dict:
    """Run validation commands and return results."""
    results = {"step": step_number, "passed": True, "errors": []}

    for cmd in validation_commands:
        try:
            result = subprocess.run(
                cmd.split(),
                capture_output=True,
                text=True,
                timeout=30  # Prevent hanging
            )
            if result.returncode != 0:
                results["passed"] = False
                results["errors"].append({
                    "command": cmd,
                    "stderr": result.stderr[:500]  # Truncate long errors
                })
        except subprocess.TimeoutExpired:
            results["passed"] = False
            results["errors"].append({
                "command": cmd,
                "stderr": "Command timed out after 30 seconds"
            })

    return results

# After implementing step 1 (e.g., creating the auth middleware file):
validation = validate_step(1, [
    "npx tsc --noEmit",          # Type check
    "npx eslint src/middleware/auth.ts"  # Lint the new file
])

if not validation["passed"]:
    print(f"Step 1 failed validation: {validation['errors']}")
    # Feed errors back to the execution agent for a retry

sequenceDiagram
    participant H as Human
    participant P as Planning Agent
    participant E as Execution Agent
    participant V as Validator

    H->>P: Task + Codebase Context
    P->>H: Structured Plan
    H->>H: Review & Approve Plan
    H->>E: Plan + Step 1
    E->>V: Generated Code
    V->>V: Run checks (tsc, eslint, tests)
    alt Validation Passes
        V->>E: Proceed to Step 2
    else Validation Fails
        V->>E: Error details + retry
        E->>V: Revised code
    end
    E->>H: All steps complete

Comparing Approaches

Aspect	Single-Phase (Plan + Execute Together)	Two-Phase (Plan, Then Execute)
Reliability	Degrades on multi-file tasks	Consistent across task sizes
Reviewability	Must read generated code to understand intent	Plan is reviewable before any code changes
Error recovery	Must undo partial changes and restart	Retry individual steps against the plan
Token efficiency	Sometimes fewer tokens for simple tasks	More tokens overall but fewer wasted on backtracking
Latency	Faster for trivial changes	Adds planning overhead (seconds to minutes)
Best for	Single-file edits, quick fixes	Multi-file features, refactors, architectural changes

Real-World Considerations

When this pattern is overkill. If you are asking an agent to rename a variable or fix a typo, a planning phase adds latency with no benefit. Use the two-phase pattern for tasks that touch multiple files, introduce new abstractions, or change system behavior. A good heuristic: if you would write a design doc before doing it yourself, use the two-phase pattern.

Plan staleness. If your plan references file contents that change during execution (because earlier steps modified them), the plan can become stale. Mitigate this by re-reading file contents before each execution step rather than caching them from the planning phase.

Plan format matters. Structured formats (numbered steps, explicit file paths) work better than prose paragraphs. Consider having the planning agent output JSON or YAML if you want to parse the plan programmatically:

STRUCTURED_PLAN_PROMPT = """
...same as before, but add:

Output the plan as a JSON object with this schema:
{
  "summary": "string",
  "steps": [
    {
      "order": 1,
      "description": "string",
      "files_to_modify": ["path/to/file.ts"],
      "files_to_create": ["path/to/new_file.ts"],
      "validation": ["npx tsc --noEmit"]
    }
  ],
  "test_criteria": ["string"],
  "risks": ["string"]
}
"""

Context window pressure. Including the full plan in every execution prompt consumes tokens. For large plans, consider summarizing completed steps and only including the full detail for the current and adjacent steps.

Do Not Skip Validation Between Steps

It is tempting to batch multiple execution steps together for speed. Resist this. If step 3 depends on step 2 and step 2 introduced a bug, batching means step 3 builds on a broken foundation. The validation loop between steps is what makes this pattern reliable. Without it, you are back to hoping the agent gets everything right in one shot.

Persisting plans across sessions. One practical challenge is that planning context gets lost when a session ends. If you generate a plan today and want to resume execution tomorrow, you need the plan — and the reasoning behind it — to be available. This is where persistent memory layers become useful: they let execution agents reference prior planning decisions without re-deriving them from scratch.