Designing Four-Tier Agent Memory with LangGraph and Amazon Bedrock

Sooner or later, a production agent starts doing the same work twice. Nothing is broken and the prompt is fine; the agent simply has no recollection of solving an almost identical problem last week, so it reasons through the same steps, repeats the same intermediate mistakes, arrives at the same answer, and you pay for every token again. This guide shows how to fix that with real memory infrastructure rather than a bigger context window: four distinct memory tiers built with LangGraph on Amazon Bedrock, plus the production safeguards that stop memory from becoming a liability.

Picture a contract-analysis agent with a solid harness and a working multi-agent setup that still spends its first several tool calls on every similar contract rediscovering structural patterns it had already worked out before. What it lacks is memory.

A bigger context window is not memory

When the context fills up, the usual reflex is a larger window or more aggressive summarization. Both help within a run; neither is memory. A useful analogy: the context window is a whiteboard. It is handy during a meeting and wiped when the meeting ends. Memory is the notebook you carry from one meeting to the next.

Context management falls short in three specific ways:

• It does not bridge runs. When the agent finishes and its process exits, the context disappears. Anything learned is gone unless you explicitly persist it.
• It does not cross agent boundaries. In a multi-agent system each worker has its own context. What a security-review worker discovers is invisible to a code-analysis worker unless you deliberately route it there.
• It degrades as it grows. Models tend to lose track of information buried in the middle of a very long prompt. This "lost in the middle" effect is well documented in the research literature.

Memory addresses a different need: a persistent, queryable knowledge store that outlives individual runs and can be shared across whatever agent topology you use.

The four tiers of agent memory

Each tier implies different infrastructure, retrieval patterns and failure modes:

+---------------------+--------------------------------------------------+----------------------------+
| Memory Tier         | What It Stores                                   | Survives Run End?          |
+---------------------+--------------------------------------------------+----------------------------+
| Working Memory      | Active context during current run                | No                         |
|                     | Tool outputs, reasoning steps, partial results  |                            |
+---------------------+--------------------------------------------------+----------------------------+
| Episodic Memory     | Summaries of past runs                           | Yes                        |
|                     | What happened, what worked, what failed          |                            |
+---------------------+--------------------------------------------------+----------------------------+
| Semantic Memory     | Structured knowledge about the domain            | Yes                        |
|                     | Facts, relationships, learned patterns           |                            |
+---------------------+--------------------------------------------------+----------------------------+
| Procedural Memory   | Successful approaches to recurring task types    | Yes                        |
|                     | How to do things, not just what was done         |                            |
+---------------------+--------------------------------------------------+----------------------------+

The tiers interact in a cycle. Working memory holds the live run. When the run ends, a distillation step writes to the three persistent tiers. When the next run begins, a retrieval step pulls relevant material back out:

Run starts
      |
      v
+------------------+
|  Working Memory  |  <-- active context, tool outputs, LangGraph state
|  (current run)   |
+------------------+
      |
      | Run ends: distillation step
      v
+------------------+     +------------------+     +------------------+
|  Episodic Memory |     | Semantic Memory  |     | Procedural Memory|
|  "what happened" |     | "what I know"    |     | "how I do this"  |
|  DynamoDB        |     | Bedrock KB       |     | DynamoDB + KB    |
+------------------+     +------------------+     +------------------+
      |                        |                        |
      +------------------------+------------------------+
                               |
                    Next run starts: retrieval step
                               |
                               v
                    Agent bootstrapped with
                    relevant prior knowledge

Episodic memory tells the next run what happened, semantic memory what is known about the domain, and procedural memory how to approach the task. All three are retrieved at startup and placed into working memory before the agent takes its first reasoning step.

Tier 1: managing working memory inside a run

Working memory is everything that exists during a single run, and you are already managing it whether or not you call it that. In LangGraph, the graph state is the working memory: every message, tool output and intermediate result lives there.

The problem is that it keeps growing. Context compression between pipeline stages, a common multi-agent technique, is really working-memory management under another name. The class below turns that idea into a reusable component. maybe_compress runs before each agent step; once the message list passes a threshold (30 by default), it keeps system messages and the 15 most recent non-system messages verbatim and replaces everything older with a single summary produced by a small, cheap model. A second method, extract_run_artifacts, runs at the end of the run and asks the model for a structured JSON record of the task, outcome, findings, approach, tools and reusable patterns, which becomes the input to episodic memory. The snippet is Python, not JavaScript:

# harness/memory/working_memory.py
from typing import TypedDict, Annotated, List, Optional
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage, SystemMessage, HumanMessage, AIMessage
from langchain_aws import ChatBedrock
import boto3
class WorkingMemoryManager:
    """
    Manages the active context within a single agent run.
    Applies a sliding window with summarization to prevent context overflow.
    """
    def __init__(
        self,
        max_messages: int = 40,
        summarize_after: int = 30,
        region: str = "us-east-1",
    ):
        self.max_messages = max_messages
        self.summarize_after = summarize_after
        bedrock = boto3.client("bedrock-runtime", region_name=region)
        # Haiku for summarization: cheap and fast for mechanical compression
        self.summarizer = ChatBedrock(
            client=bedrock,
            model_id="anthropic.claude-haiku-4-5",
            model_kwargs={"temperature": 0, "max_tokens": 2000},
        )
    def maybe_compress(self, messages: List[BaseMessage]) -> List[BaseMessage]:
        """
        Called before each agent step. If the message list is getting long,
        compress the oldest messages into a summary and keep the recent ones.
        """
        if len(messages) < self.summarize_after:
            return messages
        # Keep the system message and the most recent messages
        system_messages = [m for m in messages if isinstance(m, SystemMessage)]
        non_system = [m for m in messages if not isinstance(m, SystemMessage)]
        # Split: old messages get summarized, recent ones stay verbatim
        cutoff = len(non_system) - 15  # always keep last 15 exchanges verbatim
        to_summarize = non_system[:cutoff]
        to_keep = non_system[cutoff:]
        if not to_summarize:
            return messages
        summary_text = self._summarize(to_summarize)
        summary_message = SystemMessage(
            content=f"[WORKING MEMORY SUMMARY - earlier in this run]\n{summary_text}"
        )
        return system_messages + [summary_message] + to_keep
    def _summarize(self, messages: List[BaseMessage]) -> str:
        messages_text = "\n".join([
            f"{type(m).__name__}: {m.content[:500]}"
            for m in messages
        ])
        response = self.summarizer.invoke([
            SystemMessage(content="""
Summarize these agent messages into a compact working memory record.
Include: decisions made, tools called and their results, key findings,
and anything the agent will need to continue its work.
Discard: failed attempts, repetitive reasoning, tool errors that were resolved.
"""),
            HumanMessage(content=messages_text)
        ])
        return response.content
    def extract_run_artifacts(self, messages: List[BaseMessage]) -> dict:
        """
        At end of run, extracts key artifacts for episodic memory storage.
        Separates what happened from how it happened.
        """
        response = self.summarizer.invoke([
            SystemMessage(content="""
Extract the following from these agent messages and return as JSON:
{
  "task_summary": "one sentence description of what was asked",
  "outcome": "COMPLETED | PARTIAL | FAILED",
  "key_findings": ["list of important discoveries"],
  "approach_used": "brief description of the approach that worked",
  "tools_used": ["list of tool names"],
  "time_taken_steps": 0,
  "reusable_patterns": ["any patterns that could help on similar future tasks"]
}
Return only valid JSON.
"""),
            HumanMessage(content="\n".join([
                f"{type(m).__name__}: {m.content[:300]}"
                for m in messages[-30:]  # last 30 messages are enough
            ]))
        ])
        import json
        try:
            return json.loads(response.content)
        except json.JSONDecodeError:
            return {"outcome": "UNKNOWN", "key_findings": [], "reusable_patterns": []}

The summarization prompt discards failed attempts and resolved tool errors, and the JSON parse falls back to an UNKNOWN outcome with empty lists, so a malformed response never crashes the pipeline. Model identifiers such as the Haiku ID shown here change over time and may require a region-specific inference profile, so check the current Bedrock model list before copying them.

For local development, you can swap the Bedrock summarizer for a model served by Ollama:

# Local dev alternative
# ollama pull llama3.1:8b
from langchain_ollama import ChatOllama
summarizer = ChatOllama(model="llama3.1:8b", temperature=0, format="json")

Tier 2: episodic memory of past runs

Episodic memory answers "what happened before?". At the start of a run, the agent retrieves summaries of earlier runs that resemble the current task and uses them to skip redundant work and avoid known mistakes.

The design stores distilled summaries rather than raw run logs. Full logs are expensive to keep and slow to search; compact summaries are cheap and fast. Storage is DynamoDB, and retrieval uses Bedrock embeddings for semantic similarity.

The table, defined in Terraform, uses agent_id as the partition key and run_id as the sort key. A global secondary index on agent_id and created_at supports "newest episodes for this agent" queries, and DynamoDB TTL on expires_at removes old records automatically:

# infrastructure/memory_tables.tf
resource "aws_dynamodb_table" "episodic_memory" {
  name         = "agent-episodic-memory"
  billing_mode = "PAY_PER_REQUEST"
  hash_key     = "agent_id"
  range_key    = "run_id"
attribute {
    name = "agent_id"
    type = "S"
  }
  attribute {
    name = "run_id"
    type = "S"
  }
  attribute {
    name = "created_at"
    type = "N"
  }
  global_secondary_index {
    name            = "agent-time-index"
    hash_key        = "agent_id"
    range_key       = "created_at"
    projection_type = "ALL"
  }
  ttl {
    attribute_name = "expires_at"
    enabled        = true
  }
  tags = {
    Component = "agent-memory"
  }
}

The store itself embeds the task summary together with the key findings, writes the record with a 90-day expiry, and on retrieval compares the current task's embedding with recent episodes using cosine similarity. Only episodes above a similarity threshold of 0.75 are kept; ties are broken in favor of COMPLETED outcomes; and format_for_injection renders the winners as a labeled block for the system prompt:

# harness/memory/episodic.py
import boto3
import time
import json
import uuid
from typing import List, Optional
from dataclasses import dataclass
from langchain_aws import BedrockEmbeddings
import numpy as np

@dataclass
class EpisodeRecord:
    run_id: str
    agent_id: str
    task_summary: str
    outcome: str
    key_findings: List[str]
    approach_used: str
    tools_used: List[str]
    reusable_patterns: List[str]
    embedding: Optional[List[float]]
    created_at: float
    expires_at: float

class EpisodicMemoryStore:
    """
    Stores and retrieves summaries of past agent runs.
    Uses embeddings for semantic retrieval so the agent can find
    relevant episodes even when the wording is different.
    """
    RETENTION_DAYS = 90
    def __init__(
        self,
        agent_id: str,
        table_name: str = "agent-episodic-memory",
        region: str = "us-east-1",
    ):
        self.agent_id = agent_id
        self.region = region
        dynamodb = boto3.resource("dynamodb", region_name=region)
        self.table = dynamodb.Table(table_name)
        bedrock = boto3.client("bedrock-runtime", region_name=region)
        self.embeddings = BedrockEmbeddings(
            client=bedrock,
            model_id="amazon.titan-embed-text-v2:0",
        )
    def store_episode(self, artifacts: dict) -> str:
        """
        Called at the end of each run to store what happened.
        Returns the run_id of the stored episode.
        """
        run_id = str(uuid.uuid4())
        now = time.time()
        # Embed the task summary for semantic retrieval later
        text_to_embed = artifacts.get("task_summary", "") + " " + \
                        " ".join(artifacts.get("key_findings", []))
        embedding = self.embeddings.embed_query(text_to_embed)
        record = {
            "run_id": run_id,
            "agent_id": self.agent_id,
            "task_summary": artifacts.get("task_summary", ""),
            "outcome": artifacts.get("outcome", "UNKNOWN"),
            "key_findings": artifacts.get("key_findings", []),
            "approach_used": artifacts.get("approach_used", ""),
            "tools_used": artifacts.get("tools_used", []),
            "reusable_patterns": artifacts.get("reusable_patterns", []),
            "embedding": embedding,
            "created_at": int(now),
            "expires_at": int(now + self.RETENTION_DAYS * 86400),
        }
        self.table.put_item(Item=record)
        return run_id
    def retrieve_relevant_episodes(
        self,
        current_task: str,
        max_results: int = 5,
        min_similarity: float = 0.75,
    ) -> List[EpisodeRecord]:
        """
        Finds past episodes semantically similar to the current task.
        Returns them ranked by relevance, filtered by outcome quality.
        """
        query_embedding = self.embeddings.embed_query(current_task)
        # Fetch recent episodes for this agent
        # In production, you'd paginate this and use a vector DB for scale
        response = self.table.query(
            IndexName="agent-time-index",
            KeyConditionExpression="agent_id = :aid",
            ExpressionAttributeValues={":aid": self.agent_id},
            ScanIndexForward=False,  # newest first
            Limit=100,
        )
        episodes = []
        for item in response.get("Items", []):
            stored_embedding = item.get("embedding")
            if not stored_embedding:
                continue
            similarity = self._cosine_similarity(query_embedding, stored_embedding)
            if similarity >= min_similarity:
                episodes.append((similarity, item))
        # Sort by similarity, take top results, prefer COMPLETED outcomes
        episodes.sort(key=lambda x: (x[0], x[1].get("outcome") == "COMPLETED"), reverse=True)
        return [
            EpisodeRecord(**{k: v for k, v in item.items() if k != "embedding"}, embedding=None)
            for _, item in episodes[:max_results]
        ]
    def format_for_injection(self, episodes: List[EpisodeRecord]) -> str:
        """
        Formats retrieved episodes into a system prompt block
        for injection at the start of a new run.
        """
        if not episodes:
            return ""
        lines = ["[EPISODIC MEMORY: relevant past runs]\n"]
        for ep in episodes:
            lines.append(f"Task: {ep.task_summary}")
            lines.append(f"Outcome: {ep.outcome}")
            if ep.key_findings:
                lines.append("Key findings:")
                for f in ep.key_findings:
                    lines.append(f"  - {f}")
            if ep.reusable_patterns:
                lines.append("Patterns that worked:")
                for p in ep.reusable_patterns:
                    lines.append(f"  - {p}")
            lines.append("")
        return "\n".join(lines)
    def _cosine_similarity(self, a: List[float], b: List[float]) -> float:
        a, b = np.array(a), np.array(b)
        return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b) + 1e-9))

The query fetches at most 100 recent episodes and computes similarity in application code; as the comment notes, larger deployments should paginate and use a vector database. Also note: the boto3 DynamoDB resource rejects Python float values, so the embedding list needs to be converted to Decimal (or serialized, for example as JSON or binary) before put_item will accept it. The embedding model used here is described in the Amazon Titan Embeddings documentation.

Tier 3: semantic memory with Bedrock Knowledge Bases

Episodic memory records events; semantic memory records knowledge. The difference is subtle but important. An episode might read: "a vendor contract analyzed last Tuesday had a non-standard clause 7.3". The corresponding semantic fact reads: "clause 7.3 in vendor contracts usually covers liability caps; the normal range is twice the contract value, and anything under one times is a red flag".

Semantic facts are longer-lived than episodes. They do not expire after 90 days; they are refined over time, and they help across many task types rather than only tasks that resemble a past run.

Bedrock Knowledge Bases is a natural home for this tier because it handles chunking, embedding, storage and retrieval. What most teams have not built is the write path, since Knowledge Bases are usually treated as read-only. The store below covers both directions:

• retrieve runs a semantic vector search, optionally filtered by a domain metadata key, and drops results below a score of 0.6.
• store_fact writes the fact as a JSON document to S3 and starts an ingestion job so the Knowledge Base picks it up.
• extract_facts_from_episode asks a model to pull generalizable facts from a completed run, skipping task-specific details that belong in episodic memory and keeping only facts with confidence of at least 0.7.

# harness/memory/semantic.py
import boto3
import json
import uuid
import time
from typing import List, Optional
from dataclasses import dataclass
@dataclass
class SemanticFact:
    fact_id: str
    content: str
    domain: str          # "contract_law", "security_patterns", etc.
    confidence: float    # 0.0 to 1.0
    source_run_id: str
    created_at: float
    last_confirmed_at: float

class SemanticMemoryStore:
    """
    Manages domain knowledge that persists across runs.
    Reads from Bedrock Knowledge Base.
    Writes new facts to S3 and triggers KB sync.
    """
    def __init__(
        self,
        knowledge_base_id: str,
        s3_bucket: str,
        s3_prefix: str = "semantic-memory/",
        region: str = "us-east-1",
    ):
        self.kb_id = knowledge_base_id
        self.s3_bucket = s3_bucket
        self.s3_prefix = s3_prefix
        self.region = region
        self.bedrock_agent = boto3.client("bedrock-agent-runtime", region_name=region)
        self.bedrock_agent_mgmt = boto3.client("bedrock-agent", region_name=region)
        self.s3 = boto3.client("s3", region_name=region)
    def retrieve(
        self,
        query: str,
        domain: Optional[str] = None,
        max_results: int = 5,
        min_score: float = 0.6,
    ) -> List[str]:
        """
        Retrieves relevant semantic facts for the current task.
        Called at agent startup to bootstrap domain knowledge.
        """
        retrieval_config = {
            "vectorSearchConfiguration": {
                "numberOfResults": max_results,
                "overrideSearchType": "SEMANTIC",
            }
        }
        if domain:
            retrieval_config["vectorSearchConfiguration"]["filter"] = {
                "equals": {"key": "domain", "value": domain}
            }
        response = self.bedrock_agent.retrieve(
            knowledgeBaseId=self.kb_id,
            retrievalQuery={"text": query},
            retrievalConfiguration=retrieval_config,
        )
        return [
            r["content"]["text"]
            for r in response.get("retrievalResults", [])
            if r.get("score", 0) >= min_score
        ]
    def store_fact(self, fact: SemanticFact) -> str:
        """
        Writes a new fact to S3 and triggers a KB ingestion job.
        The fact will be available for retrieval after the sync completes
        (typically 1-5 minutes).
        For time-sensitive use cases, consider a separate fast-path
        retrieval from DynamoDB alongside the KB retrieval.
        """
        document = {
            "fact_id": fact.fact_id,
            "content": fact.content,
            "domain": fact.domain,
            "confidence": fact.confidence,
            "source_run_id": fact.source_run_id,
            "created_at": fact.created_at,
            "metadata": {
                "domain": fact.domain,
                "confidence": str(fact.confidence),
            }
        }
        key = f"{self.s3_prefix}{fact.domain}/{fact.fact_id}.json"
        self.s3.put_object(
            Bucket=self.s3_bucket,
            Key=key,
            Body=json.dumps(document),
            ContentType="application/json",
        )
        # Trigger sync so new fact becomes retrievable
        self.bedrock_agent_mgmt.start_ingestion_job(
            knowledgeBaseId=self.kb_id,
            dataSourceId=self._get_data_source_id(),
        )
        return fact.fact_id
    def extract_facts_from_episode(
        self,
        episode_artifacts: dict,
        extractor_model,
        domain: str,
    ) -> List[SemanticFact]:
        """
        At end of a successful run, extracts generalizable facts
        from the episode and prepares them for semantic memory storage.
        Only extracts from COMPLETED runs with confidence.
        """
        if episode_artifacts.get("outcome") != "COMPLETED":
            return []
        from langchain_core.messages import SystemMessage, HumanMessage
        response = extractor_model.invoke([
            SystemMessage(content=f"""
Extract generalizable domain facts from this completed agent run.
Domain: {domain}
Rules:
- Only extract facts that would be useful across many different tasks
- Do not extract task-specific details (those belong in episodic memory)
- Rate confidence 0.0 to 1.0 based on how certain the agent was
- Skip anything that might be specific to this one instance
Return JSON array:
[
  {{"content": "the fact", "confidence": 0.9}},
  ...
]
Return empty array if no generalizable facts found.
"""),
            HumanMessage(content=json.dumps(episode_artifacts))
        ])
        try:
            facts_data = json.loads(response.content)
        except json.JSONDecodeError:
            return []
        now = time.time()
        return [
            SemanticFact(
                fact_id=str(uuid.uuid4()),
                content=f["content"],
                domain=domain,
                confidence=f.get("confidence", 0.7),
                source_run_id=episode_artifacts.get("run_id", "unknown"),
                created_at=now,
                last_confirmed_at=now,
            )
            for f in facts_data
            if f.get("confidence", 0) >= 0.7  # only store high-confidence facts
        ]
    def _get_data_source_id(self) -> str:
        response = self.bedrock_agent_mgmt.list_data_sources(
            knowledgeBaseId=self.kb_id,
            maxResults=1,
        )
        return response["dataSourceSummaries"][0]["dataSourceId"]

New facts are not retrievable immediately. Ingestion takes a while, typically a few minutes according to the code comment, so time-sensitive workloads need a fast-path lookup (for example in DynamoDB) alongside the Knowledge Base. Two further points deserve verification against the Knowledge Base ingestion documentation before you rely on this code. First, metadata filtering in Knowledge Bases works on metadata attributes supplied in a separate metadata file next to each S3 document, not on fields embedded inside the document body, so the domain filter may not match anything as written. Second, starting an ingestion job for every single fact can collide with a job already running on the same data source; batching writes and syncing on a schedule is usually more robust.

Tier 4: procedural memory, with a validation gate

Procedural memory is both the most powerful tier and the most dangerous. It stores how to do something, not just what was done: a successful approach to a recurring task type is distilled into a reusable list of steps and injected into future system prompts.

The danger is that a flawed procedure, once stored, is followed without question on every subsequent task. That is why this tier needs a validation gate on the write path. In the store below, propose_procedure only runs for completed runs and saves new procedures with validated set to false, meaning pending. Another agent or a human must approve a procedure before get_procedure will return it, and get_procedure additionally requires at least three prior applications. record_application_result tracks how often a procedure has been applied and how often it succeeded, and format_for_injection presents the steps together with the success rate and a reminder that they are a starting point, not a script:

# harness/memory/procedural.py
import boto3
import json
import time
from typing import List, Optional
from dataclasses import dataclass
@dataclass
class Procedure:
    procedure_id: str
    task_type: str          # "contract_review", "security_audit", etc.
    title: str
    steps: List[str]
    success_rate: float     # updated as procedure is applied to new runs
    times_applied: int
    times_succeeded: int
    created_at: float
    last_updated_at: float
    validated: bool         # human-validated procedures get priority

class ProceduralMemoryStore:
    """
    Stores and retrieves learned procedures for recurring task types.
    Write path has a validation gate: procedures are stored as PENDING
    and must be validated (manually or by a second agent) before use.
    This prevents a single bad run from poisoning future behavior.
    """
    def __init__(
        self,
        table_name: str = "agent-procedural-memory",
        region: str = "us-east-1",
    ):
        dynamodb = boto3.resource("dynamodb", region_name=region)
        self.table = dynamodb.Table(table_name)
    def get_procedure(self, task_type: str) -> Optional[Procedure]:
        """
        Returns the best validated procedure for a task type, if one exists.
        Prefers human-validated procedures, then sorts by success rate.
        """
        response = self.table.query(
            KeyConditionExpression="task_type = :tt",
            FilterExpression="validated = :v AND times_applied >= :min",
            ExpressionAttributeValues={
                ":tt": task_type,
                ":v": True,
                ":min": 3,          # must have been used at least 3 times
            }
        )
        items = response.get("Items", [])
        if not items:
            return None
        # Sort by success rate, prefer human-validated
        best = sorted(
            items,
            key=lambda x: (x.get("validated", False), x.get("success_rate", 0)),
            reverse=True
        )[0]
        return Procedure(**best)
    def propose_procedure(
        self,
        task_type: str,
        episode_artifacts: dict,
        extractor_model,
    ) -> Optional[str]:
        """
        After a successful run, proposes a new procedure based on
        what worked. Stored as PENDING until validated.
        Returns procedure_id if a procedure was proposed, None otherwise.
        """
        if episode_artifacts.get("outcome") != "COMPLETED":
            return None
        from langchain_core.messages import SystemMessage, HumanMessage
        response = extractor_model.invoke([
            SystemMessage(content=f"""
Based on this successful agent run, extract a reusable procedure for
task type: {task_type}
The procedure should be:
- A numbered list of clear, actionable steps
- Generalizable to similar future tasks, not specific to this one
- Based only on what actually worked, not what was attempted and failed
Return JSON:
{{
  "title": "short title for this procedure",
  "steps": ["step 1", "step 2", ...]
}}
If the run does not contain a clearly reusable procedure, return null.
"""),
            HumanMessage(content=json.dumps(episode_artifacts))
        ])
        try:
            procedure_data = json.loads(response.content)
        except (json.JSONDecodeError, TypeError):
            return None
        if not procedure_data or not procedure_data.get("steps"):
            return None
        import uuid
        now = time.time()
        procedure_id = str(uuid.uuid4())
        self.table.put_item(Item={
            "procedure_id": procedure_id,
            "task_type": task_type,
            "title": procedure_data["title"],
            "steps": procedure_data["steps"],
            "success_rate": 1.0,
            "times_applied": 0,
            "times_succeeded": 0,
            "created_at": now,
            "last_updated_at": now,
            "validated": False,     # starts as PENDING
            "source_run_id": episode_artifacts.get("run_id", "unknown"),
        })
        return procedure_id
    def record_application_result(
        self,
        procedure_id: str,
        succeeded: bool,
    ):
        """
        Updates the success rate after a procedure is applied.
        If success rate drops below threshold, flag for review.
        """
        self.table.update_item(
            Key={"procedure_id": procedure_id},
            UpdateExpression="""
                SET times_applied = times_applied + :one,
                    times_succeeded = times_succeeded + :inc,
                    last_updated_at = :now,
                    success_rate = (times_succeeded + :inc) / (times_applied + :one)
            """,
            ExpressionAttributeValues={
                ":one": 1,
                ":inc": 1 if succeeded else 0,
                ":now": time.time(),
            }
        )
    def format_for_injection(self, procedure: Procedure) -> str:
        """Formats a procedure for injection into agent system prompt."""
        lines = [
            f"[PROCEDURAL MEMORY: recommended approach for {procedure.task_type}]",
            f"{procedure.title}",
            f"(Success rate: {procedure.success_rate:.0%} over {procedure.times_applied} uses)",
            "",
        ]
        for i, step in enumerate(procedure.steps, 1):
            lines.append(f"{i}. {step}")
        lines.append(
            "\nThis procedure is a starting point. Adapt it if the current task requires it."
        )
        return "\n".join(lines)

Treat this snippet as a design sketch rather than drop-in code, because a few details need adjusting. DynamoDB update expressions only support addition and subtraction, so the success_rate division has to be computed in application code (or at read time) instead of inside UpdateExpression. The query uses task_type as the key while updates use procedure_id, so the table needs a key schema, or a secondary index, that supports both access patterns. The float timestamps and success_rate need the same Decimal conversion mentioned earlier. Finally, combining "validated" with "applied at least three times" can create a chicken-and-egg situation: decide how a freshly validated procedure earns its first applications, for example through a trial mode or a human-approved exception.

Forgetting on purpose: memory decay

Keeping every memory forever causes two problems. Storage for hundreds of thousands of run summaries in DynamoDB and S3 has a real cost. More subtly, stale memories actively mislead: a pattern that worked six months ago may be wrong now because the domain moved on. Well-designed forgetting is a feature, not a defect.

The decay manager applies a different policy per tier. For episodes older than a minimum retention period (seven days by default), it deletes records whose outcome was FAILED. For procedures, it flags rather than deletes: anything not updated within 60 days, or anything used at least three times with a success rate below 0.6, gets a needs_review marker for a human to examine. This code is Python, despite being labeled as SQL:

# harness/memory/decay.py
import boto3
import time
from decimal import Decimal
from typing import List
class MemoryDecayManager:
    """
    Manages the lifecycle of stored memories.
    Applies different decay strategies per memory tier.
    """
    def __init__(self, region: str = "us-east-1"):
        self.dynamodb = boto3.resource("dynamodb", region_name=region)
        self.region = region
    def apply_episodic_decay(
        self,
        agent_id: str,
        table_name: str = "agent-episodic-memory",
        max_episodes: int = 500,
        min_retention_days: int = 7,
    ):
        """
        Prunes old episodic records when the store gets large.
        Keeps: recent episodes, COMPLETED outcomes, high-similarity clusters.
        Removes: old FAILED runs, duplicates of similar episodes.
        """
        table = self.dynamodb.Table(table_name)
        now = time.time()
        min_age = now - (min_retention_days * 86400)
        response = table.query(
            IndexName="agent-time-index",
            KeyConditionExpression="agent_id = :aid AND created_at < :cutoff",
            ExpressionAttributeValues={
                ":aid": agent_id,
                ":cutoff": Decimal(str(min_age)),
            },
            ScanIndexForward=True,  # oldest first
            Limit=max_episodes,
        )
        candidates_for_removal = [
            item for item in response.get("Items", [])
            if item.get("outcome") == "FAILED"
        ]
        with table.batch_writer() as batch:
            for item in candidates_for_removal:
                batch.delete_item(Key={
                    "agent_id": item["agent_id"],
                    "run_id": item["run_id"],
                })
    def flag_stale_procedures(
        self,
        table_name: str = "agent-procedural-memory",
        stale_after_days: int = 60,
        min_success_rate: float = 0.6,
    ) -> List[str]:
        """
        Identifies procedures that should be reviewed:
        - Not used in a long time (might be outdated)
        - Success rate has dropped below acceptable threshold
        Does NOT delete automatically. Flags for human review.
        Returns list of procedure_ids that need attention.
        """
        table = self.dynamodb.Table(table_name)
        stale_cutoff = time.time() - (stale_after_days * 86400)
        response = table.scan(
            FilterExpression=(
                "last_updated_at < :cutoff OR "
                "(times_applied >= :min AND success_rate < :threshold)"
            ),
            ExpressionAttributeValues={
                ":cutoff": Decimal(str(stale_cutoff)),
                ":min": 3,
                ":threshold": Decimal(str(min_success_rate)),
            }
        )
        flagged_ids = []
        for item in response.get("Items", []):
            table.update_item(
                Key={"procedure_id": item["procedure_id"]},
                UpdateExpression="SET needs_review = :v",
                ExpressionAttributeValues={":v": True},
            )
            flagged_ids.append(item["procedure_id"])
        return flagged_ids

Losing an episode costs a little context, whereas silently deleting a procedure could discard a hard-won, human-validated approach, hence the asymmetry. flag_stale_procedures uses a table scan, which suits a scheduled batch job (such as a Lambda) but not a hot path.

Bringing it together at run start and run end

All tiers meet at the beginning of each run. The bootstrap class queries the episodic store for up to three similar episodes, the Knowledge Base for up to five domain facts, and the procedural store for a validated procedure matching the task type. It then wraps whatever it finds in clearly delimited markers so the model can tell retrieved memory apart from the task itself:

# harness/memory/bootstrap.py
from typing import Optional
from langchain_core.messages import SystemMessage
from harness.memory.episodic import EpisodicMemoryStore
from harness.memory.semantic import SemanticMemoryStore
from harness.memory.procedural import ProceduralMemoryStore
class MemoryBootstrap:
    """
    Assembles the memory context for a new agent run.
    Called once at startup, before the agent sees the task.
    """
    def __init__(
        self,
        agent_id: str,
        knowledge_base_id: str,
        s3_bucket: str,
        region: str = "us-east-1",
    ):
        self.episodic = EpisodicMemoryStore(agent_id, region=region)
        self.semantic = SemanticMemoryStore(knowledge_base_id, s3_bucket, region=region)
        self.procedural = ProceduralMemoryStore(region=region)
    def build_memory_context(
        self,
        task: str,
        task_type: Optional[str] = None,
        domain: Optional[str] = None,
    ) -> str:
        """
        Retrieves relevant memories from all tiers and assembles
        them into a single context block for system prompt injection.
        """
        sections = []
        # Episodic: what happened on similar tasks before
        episodes = self.episodic.retrieve_relevant_episodes(task, max_results=3)
        if episodes:
            sections.append(self.episodic.format_for_injection(episodes))
        # Semantic: what is known about this domain
        if domain:
            facts = self.semantic.retrieve(task, domain=domain, max_results=5)
            if facts:
                sections.append(
                    "[SEMANTIC MEMORY: domain knowledge]\n" +
                    "\n".join(f"- {f}" for f in facts)
                )
        # Procedural: how to approach this task type
        if task_type:
            procedure = self.procedural.get_procedure(task_type)
            if procedure:
                sections.append(self.procedural.format_for_injection(procedure))
        if not sections:
            return ""
        return (
            "\n\n[MEMORY CONTEXT - retrieved from prior runs and domain knowledge]\n\n" +
            "\n\n".join(sections) +
            "\n\n[END MEMORY CONTEXT]\n"
        )

The write path is deliberately separate from the read path. At the end of a run, the consolidator extracts artifacts from working memory, stores the episode, extracts and stores semantic facts only when the run completed, and proposes a procedure that will wait for validation. It uses a larger model than in-run compression, since this step shapes future behavior; check that the model ID is still available in your region. The snippet also uses Optional without importing it from typing, so add that import if you reuse it:

# harness/memory/consolidator.py
from harness.memory.working_memory import WorkingMemoryManager
from harness.memory.episodic import EpisodicMemoryStore
from harness.memory.semantic import SemanticMemoryStore
from harness.memory.procedural import ProceduralMemoryStore
from langchain_aws import ChatBedrock
import boto3
class MemoryConsolidator:
    """
    Runs at the end of each agent run to distill and store memories.
    The write path is separate from the read path by design.
    """
    def __init__(
        self,
        agent_id: str,
        knowledge_base_id: str,
        s3_bucket: str,
        region: str = "us-east-1",
    ):
        bedrock = boto3.client("bedrock-runtime", region_name=region)
        self.extractor = ChatBedrock(
            client=bedrock,
            model_id="anthropic.claude-3-5-sonnet-20241022-v2:0",
            model_kwargs={"temperature": 0, "max_tokens": 4000},
        )
        self.working = WorkingMemoryManager(region=region)
        self.episodic = EpisodicMemoryStore(agent_id, region=region)
        self.semantic = SemanticMemoryStore(knowledge_base_id, s3_bucket, region=region)
        self.procedural = ProceduralMemoryStore(region=region)
    def consolidate(
        self,
        final_state: dict,
        task_type: Optional[str] = None,
        domain: Optional[str] = None,
    ):
        """
        End-of-run consolidation. Extracts and stores memories from all tiers.
        """
        messages = final_state.get("messages", [])
        # Step 1: Extract run artifacts from working memory
        artifacts = self.working.extract_run_artifacts(messages)
        # Step 2: Store as episodic memory
        run_id = self.episodic.store_episode(artifacts)
        artifacts["run_id"] = run_id
        # Step 3: Extract and store semantic facts (only from successful runs)
        if domain and artifacts.get("outcome") == "COMPLETED":
            facts = self.semantic.extract_facts_from_episode(
                artifacts, self.extractor, domain
            )
            for fact in facts:
                self.semantic.store_fact(fact)
        # Step 4: Propose procedural memory update
        if task_type:
            self.procedural.propose_procedure(task_type, artifacts, self.extractor)

Production hardening

Guarding against memory poisoning

One bad run can write a wrong fact into semantic memory or propose a broken procedure, and at scale the error is amplified, because every future run that retrieves the fact or follows the procedure inherits it. The defenses are layered:

• Write only from COMPLETED runs whose artifacts carry high confidence.
• Demand repeated, independent sightings before a fact is promoted: a single run only suggests something, while agreement across three runs is real evidence.
• Put human review in front of procedural memory before any procedure reaches validated status.

A simple way to implement the second layer is a confirmation counter keyed by a hash of the fact's content. Each observation atomically increments the counter with if_not_exists, and the fact is promoted only once the count reaches the threshold:

# Simple confirmation counter before promoting a fact
def promote_fact_if_confirmed(
    fact_content: str,
    confirmation_threshold: int = 3,
    table_name: str = "fact-confirmation-log",
    region: str = "us-east-1",
):
    """
    Tracks how many independent runs have observed the same fact.
    Only writes to semantic memory after threshold is reached.
    """
    table = boto3.resource("dynamodb", region_name=region).Table(table_name)
    import hashlib
    fact_hash = hashlib.sha256(fact_content.encode()).hexdigest()[:16]
    response = table.update_item(
        Key={"fact_hash": fact_hash},
        UpdateExpression="SET confirmation_count = if_not_exists(confirmation_count, :z) + :inc, content = :c",
        ExpressionAttributeValues={":z": 0, ":inc": 1, ":c": fact_content},
        ReturnValues="UPDATED_NEW",
    )
    return int(response["Attributes"]["confirmation_count"]) >= confirmation_threshold

Because exact text is hashed, differently worded versions of one fact are counted separately; normalize or cluster facts first if promotions turn out too rare.

Keeping PII out of memory

Agent memory will collect personally identifiable information unless you actively prevent it. A contract-analysis agent stores facts about contracts, and contracts mention people, salaries and other personal details. The cleanest approach is to scrub text before anything is written to the episodic or semantic tiers. Amazon Comprehend offers a PII detection API that works well on English text. The function below replaces each detected entity with its type label, iterating from the end of the string so earlier offsets stay valid:

import boto3

def scrub_pii_before_storage(text: str, region: str = "us-east-1") -> str:
    comprehend = boto3.client("comprehend", region_name=region)
    response = comprehend.detect_pii_entities(Text=text[:5000], LanguageCode="en")
    scrubbed = text
    # Replace detected PII with type labels, working backwards to preserve offsets
    for entity in sorted(response["Entities"], key=lambda x: x["BeginOffset"], reverse=True):
        scrubbed = (
            scrubbed[:entity["BeginOffset"]] +
            f"[{entity['Type']}]" +
            scrubbed[entity["EndOffset"]:]
        )
    return scrubbed

Be aware that the call only examines the first 5,000 characters, so anything beyond that is stored unscrubbed. Longer texts need to be split into chunks and each chunk scrubbed separately. Supported entity types and limits are listed in the Comprehend PII documentation.

Isolating tenants

If agents serve several customers on shared infrastructure, their memories must be fully isolated: a run for one customer must never retrieve another customer's memories. The simplest pattern is to namespace every memory key with a tenant identifier and to enforce the boundary in the storage layer, for example with DynamoDB condition expressions, rather than relying on application-level filtering alone:

# Always include tenant_id in all memory operations
def store_episode_isolated(tenant_id: str, agent_id: str, artifacts: dict):
    namespaced_agent_id = f"{tenant_id}#{agent_id}"
    # All subsequent operations use namespaced_agent_id, never bare agent_id

Since agent_id is the episodic partition key, the prefix keeps normal queries inside one tenant. Apply the same namespacing to the Knowledge Base, the procedural table and the confirmation log.

What to expect once memory is live

Memory makes agents measurably better over time, but it also introduces failure modes that are harder to debug than anything a stateless system produces:

• Bad memories are worse than none. A confused stateless agent spoils one run. An agent with a poisoned fact spoils every run that retrieves it, possibly for months. Invest in validation gates before expanding memory coverage.
• Retrieval quality degrades silently. As a domain's vocabulary shifts, a fact stored months ago may stop matching queries that are semantically similar today. Track retrieval rates and spot-check what is actually returned; healthy infrastructure does not imply healthy retrieval.
• Startup latency grows. Querying three tiers before the first response adds real time. Profile it on your workload; prefetching asynchronously while a request waits in a queue can hide some of it, and in other cases you should decide per task type which tiers are worth querying.
• Writes lag behind reads. Knowledge Base ingestion is not instant, so a fact saved as one run finishes can stay invisible to searches for minutes. For agents that process many tasks back to back, add a fast-path store next to the Knowledge Base.
• Versioning remains unsolved. Upgrading the embedding model makes existing vectors incompatible with the new vector space, and old and new embeddings cannot be mixed in one retrieval. A model upgrade means re-embedding the entire store, so plan for it before you accumulate millions of records.

Reference architecture

The complete lifecycle of a run, from bootstrap through consolidation into the three persistent stores, looks like this:

New Run Starts
       |
       v
+-------------------------------+
|     Memory Bootstrap          |
|  1. Query Episodic Store      |
|  2. Query Bedrock KB          |
|  3. Query Procedural Store    |
|  4. Assemble context block    |
+-------------------------------+
       |
       v
+-------------------------------+
|   Agent Run (LangGraph)       |
|   Working Memory Manager      |
|   Compression at threshold    |
|   Tool calls with auth/retry  |
+-------------------------------+
       |
       v
+-------------------------------+
|     Memory Consolidator       |
|  1. Extract run artifacts     |
|  2. Scrub PII                 |
|  3. Write episodic record     |
|  4. Propose semantic facts    |
|     (if confirmed: store)     |
|  5. Propose procedure         |
|     (pending validation)      |
+-------------------------------+
       |
       v
+------------------+   +------------------+   +------------------+
|  DynamoDB        |   |  Bedrock KB      |   |  DynamoDB        |
|  Episodic Memory |   |  Semantic Memory |   |  Procedural Mem. |
|  TTL: 90 days    |   |  S3 backed       |   |  Validated flag  |
+------------------+   +------------------+   +------------------+

And this is the full infrastructure stack, including the local development alternative based on Ollama, ChromaDB and Docker Compose that lets you exercise the memory stack without AWS costs:

+-----------------------------+---------------------+------------------------------+
| Component                   | Technology          | Role                         |
+-----------------------------+---------------------+------------------------------+
| Working Memory              | LangGraph State +   | Active context compression,  |
|                             | MemorySaver         | sliding window management    |
+-----------------------------+---------------------+------------------------------+
| Episodic Memory Store       | DynamoDB + TTL      | Past run summaries, semantic |
|                             |                     | similarity retrieval         |
+-----------------------------+---------------------+------------------------------+
| Episodic Embeddings         | Titan Embed v2      | Semantic similarity search   |
|                             | (Bedrock)           | over episode store           |
+-----------------------------+---------------------+------------------------------+
| Semantic Memory             | Bedrock Knowledge   | Domain facts, structured     |
|                             | Bases + S3          | knowledge, write via ingest  |
+-----------------------------+---------------------+------------------------------+
| Procedural Memory           | DynamoDB            | Learned task procedures,     |
|                             |                     | validated flag, success rate |
+-----------------------------+---------------------+------------------------------+
| PII Scrubbing               | AWS Comprehend      | Before any memory write,     |
|                             |                     | English PII detection        |
+-----------------------------+---------------------+------------------------------+
| Fact Confirmation           | DynamoDB            | Multi-run confirmation gate  |
|                             |                     | before semantic write        |
+-----------------------------+---------------------+------------------------------+
| Memory Decay                | DynamoDB TTL +      | Episode pruning, stale       |
|                             | scheduled Lambda    | procedure flagging           |
+-----------------------------+---------------------+------------------------------+
| Context Compression         | Claude Haiku 4.5    | Working memory summarization |
+-----------------------------+---------------------+------------------------------+
| Artifact Extraction         | Claude 3.5 Sonnet   | End-of-run distillation      |
+-----------------------------+---------------------+------------------------------+
| Local Dev Alternative       | Ollama + ChromaDB   | Full memory stack testable   |
|                             | Docker Compose      | without AWS costs            |
+-----------------------------+---------------------+------------------------------+
| Infrastructure as Code      | Terraform           | DynamoDB tables, S3 bucket,  |
|                             |                     | KB data source, IAM roles    |
+-----------------------------+---------------------+------------------------------+

Key takeaways

• Treat the context window as a whiteboard and build a separate, persistent notebook for knowledge that must survive a run.
• Separate memory into working, episodic, semantic and procedural tiers, because each needs different storage, retrieval and failure handling.
• Distill runs into compact summaries and facts; never store or search raw logs as memory.
• Put the strongest gates on the tiers that change behavior most: confirmation counts for facts, human validation for procedures.
• Scrub PII and enforce tenant isolation in the storage layer before the first memory is ever written.
• Budget from day one for forgetting, retrieval monitoring, ingestion lag and eventual re-embedding.