A ReAct agent memory component is what separates a demo that solves toy puzzles from an agent that survives real multi-turn tasks. The base ReAct loop interleaves thought, action, and observation, but if you discard those traces after each step the agent repeats mistakes and loses user context. This guide shows how to add ReAct agent memory to a reasoning-action loop with a concrete Python implementation you can run against any OpenAI-compatible endpoint.
Step 1: Define the memory interface
Before writing storage code, decide what the loop needs from memory. You require two capabilities: episodic recall of the current task trajectory (working memory) and semantic retrieval of prior facts (long-term memory). A clean abstraction prevents the rest of the agent from caring whether records live in SQLite, Redis, or a vector index.
from abc import ABC, abstractmethod
from dataclasses import dataclass
@dataclass
class MemoryRecord:
text: str
meta: dict
embedding: list[float] | None = None
class MemoryStore(ABC):
@abstractmethod
def add(self, rec: MemoryRecord) -> None: ...
@abstractmethod
def search(self, query_embedding: list[float], k: int) -> list[MemoryRecord]: ...
@abstractmethod
def get_recent(self, n: int) -> list[MemoryRecord]: ...The meta field stores the record type (thought, action, observation) and any tool metadata. Keeping the interface narrow means you can swap the backend later without touching the ReAct loop.
Step 2: Implement a persistent store with embeddings
For long-term recall you need similarity search. The simplest runnable approach is SQLite plus embeddings from an OpenAI-compatible API. Store the vector as JSON for clarity; in production you would use a real vector extension or a dedicated store.
import sqlite3, json, numpy as np, openai
class SqliteMemory(MemoryStore):
def __init__(self, db_path="memory.db", client=None):
self.conn = sqlite3.connect(db_path)
self.conn.execute("CREATE TABLE IF NOT EXISTS mem (id INTEGER PRIMARY KEY, text TEXT, meta TEXT, emb TEXT)")
self.client = client or openai.OpenAI()
def _embed(self, text: str) -> list[float]:
resp = self.client.embeddings.create(model="text-embedding-3-small", input=text)
return resp.data[0].embedding
def add(self, rec: MemoryRecord) -> None:
emb = rec.embedding or self._embed(rec.text)
self.conn.execute(
"INSERT INTO mem (text, meta, emb) VALUES (?,?,?)",
(rec.text, json.dumps(rec.meta), json.dumps(emb)),
)
self.conn.commit()
def search(self, query_embedding: list[float], k: int = 5) -> list[MemoryRecord]:
rows = self.conn.execute("SELECT text, meta, emb FROM mem").fetchall()
scored = []
q = np.array(query_embedding)
for text, meta, emb in rows:
vec = np.array(json.loads(emb))
score = float(np.dot(q, vec) / (np.linalg.norm(q) * np.linalg.norm(vec)))
scored.append((score, text, meta))
scored.sort(reverse=True)
return [MemoryRecord(text=t, meta=json.loads(m)) for _, t, m in scored[:k]]
def get_recent(self, n: int = 10) -> list[MemoryRecord]:
rows = self.conn.execute(
"SELECT text, meta FROM mem ORDER BY id DESC LIMIT ?", (n,)
).fetchall()
return [MemoryRecord(text=r[0], meta=json.loads(r[1])) for r in rows]This implementation gives you durable ReAct agent memory across process restarts. The cosine similarity in search is adequate for a few thousand records; beyond that, move to a proper ANN index.
Step 3: Retrieve and inject memories into the ReAct prompt
The ReAct prompt must include retrieved context or the memory is useless. Build a function that embeds the current task, pulls the top-k long-term records, and prepends recent trajectory lines.
def build_prompt(task: str, memory: MemoryStore, client) -> str:
q_emb = client.embeddings.create(model="text-embedding-3-small", input=task).data[0].embedding
long_term = memory.search(q_emb, k=3)
recent = memory.get_recent(5)
sys = "You are a ReAct agent. Emit 'Thought:', 'Action:', then 'Observation:' lines."
mem_block = "\n".join(f"[Memory] {r.text}" for r in long_term)
traj_block = "\n".join(f"[Trajectory] {r.text}" for r in recent)
return f"{sys}\n{mem_block}\n{traj_block}\nTask: {task}\n"Injecting both blocks lets the model reference a fact from last week and the action it took two steps ago. Without explicit retrieval, the ReAct agent memory stays latent.
Step 4: Persist thoughts, actions, and observations
After each LLM turn, parse the output and write every meaningful step to the store. Do not wait until the task ends; a crash mid-loop should not lose state.
def persist_step(thought: str, action: str, obs: str, memory: MemoryStore) -> None:
memory.add(MemoryRecord(text=f"Thought: {thought}", meta={"type": "thought"}))
memory.add(MemoryRecord(text=f"Action: {action}", meta={"type": "action"}))
memory.add(MemoryRecord(text=f"Observation: {obs}", meta={"type": "observation"}))Call this inside the loop immediately after you execute the action and capture the observation. This discipline is what makes ReAct agent memory reliable.
Step 5: Bound the working set
Unbounded trajectory growth blows the context window. Keep a short working list in memory and archive older records to long-term storage once it exceeds a threshold.
class WorkingMemory:
def __init__(self, long_term: MemoryStore, limit: int = 12):
self.buffer: list[MemoryRecord] = []
self.long_term = long_term
self.limit = limit
def add(self, rec: MemoryRecord) -> None:
self.buffer.append(rec)
if len(self.buffer) > self.limit:
old = self.buffer.pop(0)
self.long_term.add(old)
def recent(self, n: int) -> list[MemoryRecord]:
return self.buffer[-n:]The ReAct agent memory thus keeps the last dozen steps hot and pushes the rest into searchable storage. You avoid re-embedding on every turn by embedding only when archiving.
Step 6: Run the full loop
Wire the pieces together. Use a chat completion call, parse the ReAct format, execute a stub tool, and persist. For resilience across providers, point the OpenAI client at a gateway that fronts many models; for example, n4n.ai exposes one OpenAI-compatible endpoint for 240+ models with automatic fallback when a provider is degraded.
from openai import OpenAI
client = OpenAI(base_url="https://api.openai.com/v1", api_key="sk-...")
memory = SqliteMemory(client=client)
working = WorkingMemory(memory)
task = "Find the cheapest flight to NYC under $300"
prompt = build_prompt(task, memory, client)
for _ in range(6):
resp = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "system", "content": prompt}],
)
text = resp.choices[0].message.content
# Minimal parser: split on labels. Replace with robust regex in prod.
thought = text.split("Thought:")[1].split("Action:")[0].strip()
action = text.split("Action:")[1].split("Observation:")[0].strip()
obs = "tool result stub"
persist_step(thought, action, obs, working)
prompt = build_prompt(task, memory, client)The loop now reads from and writes to memory every iteration. The build_prompt call after each step refreshes both retrieved context and recent trajectory.
Step 7: Verify success
Verification must be concrete, not vibes.
- Unit test retrieval. Insert two records, embed a query matching one, assert
searchreturns it first. - Integration run. Execute the loop on a task requiring a fact from step 1 at step 4. Open
memory.dband confirm row counts increase each iteration. - Prompt inspection. Log the prompt string. Confirm it contains
[Memory]lines referencing earlier steps and[Trajectory]lines from the current episode. - Failure drill. Kill the process after step 3, restart, rebuild
SqliteMemory, and confirmget_recentreturns prior observations.
If the agent stops repeating early mistakes and completes a task that spans more than ten tool calls without context overflow, your ReAct agent memory is working. Treat the store as a first-class component: back it up, index it, and version your schema.