Human oversight and memory

Slide 1 of 31Open this slide

Human-in-the-Loop & Memory

Safe Agents with Persistent State

Week 11 — Applied AI Course

"An agent without guardrails is a liability. An agent without memory is a goldfish."

Spring 2026

Slide 2 of 31Open this slide

The Autonomy Spectrum

Where does your agent sit?

Step 1: Full Manual

Human does everything. AI provides information only.

• Traditional search, dashboards
• Zero autonomy, maximum control
• Example: ChatGPT in Q&A mode

Step 2: Copilot

AI suggests, human decides and executes.

• Code completion, writing assistants
• Human approves every action
• Example: GitHub Copilot suggesting code

Step 3: Supervised Autonomy

AI acts, human oversees and intervenes when needed.

• Agent executes with approval gates
• Human reviews critical decisions
• Example: AI scheduling assistant that confirms before sending invites

Step 4: Full Autonomy

AI acts independently within defined boundaries.

• Fully autonomous pipelines
• Human sets policy, agent executes
• Example: Automated trading within risk limits

Key insight: Most production agents should live in Steps 2-3. Full autonomy is rarely appropriate for high-stakes domains.

Sources

• Building Effective Agents (Anthropic, 2024)

Slide 3 of 31Open this slide

Why Human-in-the-Loop?

Autonomous agents are powerful — but power without oversight is dangerous.

Safety

• LLMs hallucinate. Agents act on hallucinations.
• A wrong API call can delete production data, send incorrect emails, or charge a credit card
• HITL creates a safety net before irreversible actions

Trust

• Users need to understand what agents are doing and why
• Transparency builds adoption; black-box agents get turned off
• Gradual autonomy: earn trust through demonstrated reliability

Compliance

• Regulated industries (finance, healthcare, legal) require human oversight
• Audit trails need human decision points
• GDPR, SOX, HIPAA all have human-review requirements for automated decisions

Error Correction

• Humans catch edge cases LLMs miss
• Correction data improves future agent performance
• Early intervention prevents cascading failures

The goal is not to eliminate humans — it is to put humans where they add the most value.

Slide 4 of 31Open this slide

HITL vs HOTL

Two oversight models with very different latency and throughput profiles.

Human-in-the-Loop (HITL)

Synchronous approval before each action. The agent pauses and waits for a human to approve every consequential step before it proceeds.

• Maximum control — nothing irreversible happens without sign-off
• Adds latency to every gated action; throughput is bounded by human availability
• Right choice for high-stakes, irreversible actions (wire transfers, production deletes, medical orders)

Human-on-the-Loop (HOTL)

Asynchronous monitoring; intervene only on anomalies. The agent acts autonomously while a human watches dashboards and alerts, stepping in only when something looks wrong.

• High throughput — the human is not in the critical path
• Relies on strong observability and anomaly detection to surface what needs attention
• Right choice for high-volume, lower-stakes or easily-reversible workflows

Choosing between them

Most production systems prefer HOTL for throughput and reserve HITL for high-stakes, irreversible actions. The two are not exclusive — a single system can run HOTL by default and escalate to HITL when confidence is low or an action crosses a risk threshold.

Sources

• Human-in-the-Loop vs Human-on-the-Loop AI Agents (Waxell)

Slide 5 of 31Open this slide

Why Memory?

Agents Without Memory Are Goldfish

Every conversation starts from zero. Every context is lost. Every preference forgotten.

The problem without memory:

• Session 1: "I prefer Python and use pytest for testing"
• Session 2: "What testing framework do you use?" — Agent has no idea
• Session 3: "Remember that bug we fixed last week?" — Agent draws a blank

What memory enables:

• Personalization — Remember user preferences, past decisions, coding style
• Continuity — Resume multi-session tasks without re-explaining context
• Learning — Improve over time from past successes and failures
• Efficiency — Avoid re-doing research, re-asking questions, re-establishing context

The context window is not memory

• Context windows are large (128K-1M+ tokens) but temporary
• They vanish when the session ends
• They are expensive to fill with redundant history
• Real memory requires persistence, retrieval, and curation

Memory transforms a stateless text predictor into a persistent collaborator.

Slide 6 of 31Open this slide

Approval Gates

Pause before irreversible actions. The most fundamental HITL pattern.

Implementation pattern:

def approval_gate(action: AgentAction) -> AgentAction:
    if action.is_irreversible or action.risk_level == "high":
        approval = request_human_approval(
            action=action.name,
            params=action.params,
            impact=action.estimated_impact,
            timeout=300  # 5 min timeout
        )
        if not approval.approved:
            return action.replan(reason=approval.feedback)
    return action.execute()

Rule of thumb: If you cannot undo it, do not auto-execute it.

Slide 7 of 31Open this slide

Confidence-Based Routing

Auto-execute when confident. Escalate when uncertain.

How to get confidence scores:

from pydantic import BaseModel, Field

class AgentDecision(BaseModel):
    action: str
    reasoning: str
    confidence: float = Field(
        description="0.0-1.0 confidence in this action being correct"
    )
    uncertainty_reason: str | None = Field(
        description="Why confidence is below 1.0, if applicable"
    )

def route_by_confidence(decision: AgentDecision):
    if decision.confidence >= 0.9:
        return execute_automatically(decision)
    elif decision.confidence >= 0.5:
        return suggest_and_confirm(decision)
    else:
        return escalate_to_human(decision)

Calibration matters

• LLMs are often overconfident — calibrate thresholds empirically
• Use structured output to force explicit confidence reasoning
• Track accuracy at each confidence level to tune thresholds over time

Slide 8 of 31Open this slide

Review Queues

Batch human review for non-urgent decisions. Not everything needs real-time approval.

The Pattern

• Agent processes items and flags those needing review
• Items queue up in a review dashboard
• Humans review in batches at convenient times
• Agent continues with non-blocked work

When to use review queues:

• Content moderation — Flag borderline posts for human review
• Data processing — Queue ambiguous records for manual classification
• Email drafts — Agent writes, human approves batch before sending
• Report generation — Agent compiles, human validates before distribution

Implementation considerations:

class ReviewQueue:
    def submit(self, item: ReviewItem, priority: str = "normal"):
        """Add item to queue. Agent continues other work."""
        self.queue.push(item, priority=priority)

    def process_batch(self, reviewer_id: str, batch_size: int = 20):
        """Human reviews a batch of items."""
        items = self.queue.pop(batch_size)
        return ReviewBatch(items=items, reviewer=reviewer_id)

    def apply_decisions(self, batch: ReviewBatch):
        """Apply human decisions back to the agent workflow."""
        for item, decision in batch.decisions:
            if decision.approved:
                self.agent.execute(item.pending_action)
            else:
                self.agent.handle_rejection(item, decision.feedback)

Queue design tips:

• Priority levels — Urgent items surface first
• SLA tracking — Alert if items sit too long
• Auto-escalation — If queue grows too large, increase automation thresholds
• Feedback capture — Record why items were approved/rejected for training

Slide 9 of 31Open this slide

Feedback Loops

Humans correct agent outputs. Agent improves from corrections.

Types of feedback:

1. Immediate correction

# Human edits agent's draft email
correction = HumanCorrection(
    original=agent_draft,
    corrected=human_edited_version,
    explanation="Tone was too formal for this client"
)
agent.memory.store_correction(correction)

2. Rating-based feedback

# Thumbs up/down on agent responses
feedback = UserFeedback(response_id="abc", rating="negative",
                        reason="Missed the key requirement")

3. Preference learning

# Track which suggestions users accept vs reject
agent.preference_model.update(
    accepted_actions=accepted,
    rejected_actions=rejected
)

Virtuous cycle: Better outputs lead to more trust, which leads to more autonomy, which generates more feedback data.

Slide 10 of 31Open this slide

LangGraph Breakpoints & Interrupt Patterns

Programmatic pause points in agent execution graphs.

Breakpoints in LangGraph

LangGraph (v0.4) provides first-class support for HITL via interrupt(). Calling interrupt() pauses graph execution and now surfaces Interrupt objects automatically in the .invoke() return value, so you no longer have to inspect graph state separately to detect a pause:

from langgraph.graph import StateGraph
from langgraph.types import interrupt, Command

def sensitive_action_node(state):
    """Node that pauses for human approval before executing."""
    action = plan_action(state)

    # This pauses execution and sends data to the human
    human_response = interrupt({
        "action": action.name,
        "params": action.params,
        "question": "Should I proceed with this action?"
    })

    if human_response["approved"]:
        return {"result": execute_action(action)}
    else:
        return {"result": "Action cancelled", "feedback": human_response.get("reason")}

Compile-time breakpoints

# Add breakpoints when compiling the graph
graph = builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["sensitive_action"],  # Pause BEFORE this node
    interrupt_after=["draft_response"],     # Pause AFTER this node
)

Resuming after interrupt

# In v0.4, the interrupt surfaces directly in the invoke() return value
result = graph.invoke({"messages": [...]}, config=thread_config)
if result.get("__interrupt__"):
    interrupts = result["__interrupt__"]   # list of Interrupt objects

# You can also reach in via state if you need full context
current_state = graph.get_state(thread_config)
print(current_state.next)       # Which node is paused
print(current_state.tasks)      # Pending interrupt data

# Resume with human input
graph.invoke(
    Command(resume={"approved": True}),
    config=thread_config
)

Key patterns:

• interrupt_before — Review inputs before a node runs
• interrupt_after — Review outputs before they propagate
• Dynamic interrupts — Use interrupt() inside node logic for conditional pausing
• Checkpointer required — State must be persisted to survive the pause

Sources

• LangGraph Interrupts Documentation (LangChain)
• LangGraph v0.4: HITL, Checkpoints & State Persistence (AITechConnect)
• LangGraph Persistence (LangChain Docs)
• LangGraph Library (LangChain)

Slide 11 of 31Open this slide

Tool Permission Levels

Not all tools carry equal risk. Permission levels should match the blast radius.

Three-tier permission model:

Tier 1: Read (Auto-execute)

No confirmation needed. Low risk, no side effects.

• search_documents — Query knowledge base
• get_weather — Fetch weather data
• list_files — Browse directory contents
• read_database — SELECT queries only

Tier 2: Write (Confirm)

Show the user what will change. Require confirmation.

• send_email — Show draft, confirm recipients
• create_file — Show content, confirm path
• update_record — Show diff of changes
• schedule_meeting — Show details, confirm attendees

Tier 3: Delete / Irreversible (Approve + Review)

Require explicit approval with full context and impact assessment.

• delete_records — Show what will be deleted, count affected rows
• deploy_code — Show changes, require sign-off
• transfer_funds — Show amount, source, destination
• modify_permissions — Show before/after access levels

Implementation:

TOOL_PERMISSIONS = {
    "search":         {"level": "auto",    "tier": 1},
    "read_file":      {"level": "auto",    "tier": 1},
    "send_email":     {"level": "confirm", "tier": 2},
    "write_file":     {"level": "confirm", "tier": 2},
    "delete_records": {"level": "approve", "tier": 3},
    "deploy":         {"level": "approve", "tier": 3},
}

async def execute_tool(tool_name: str, params: dict):
    perm = TOOL_PERMISSIONS[tool_name]
    if perm["level"] == "auto":
        return await tool.run(params)
    elif perm["level"] == "confirm":
        await show_preview(tool_name, params)
        if await get_confirmation():
            return await tool.run(params)
    elif perm["level"] == "approve":
        await show_impact_assessment(tool_name, params)
        if await get_explicit_approval(require_reason=True):
            return await tool.run(params)

Slide 12 of 31Open this slide

Escalation Patterns

When and how agents hand off to humans.

Step 1: Detect Escalation Trigger

Reasons to escalate:

• Confidence below threshold
• User explicitly requests human help
• Error count exceeds limit (3 retries)
• Domain outside agent's scope
• Sensitive/emotional content detected

Step 2: Preserve Context

Before handing off, package everything the human needs:

• Full conversation history
• Agent's attempted actions and reasoning
• Relevant retrieved documents
• Current state of any in-progress tasks

escalation_package = EscalationContext(
    conversation=state.messages,
    agent_reasoning=state.scratchpad,
    attempted_actions=state.action_log,
    user_sentiment=analyze_sentiment(state.messages),
    priority=determine_priority(state),
)

Step 3: Route to Right Human

Match escalation to expertise:

• Technical issues -> Engineering support
• Billing questions -> Finance team
• Complaints -> Senior customer success
• Legal/compliance -> Legal team

Step 4: Handoff with Warm Transfer

async def escalate(state, reason: str):
    context = build_escalation_context(state)
    human_agent = route_to_specialist(context)

    # Notify human with full context
    await notify_human(human_agent, context)

    # Tell user what's happening
    return "I'm connecting you with a specialist who can help. " \
           "I've shared our conversation context so you won't need to repeat yourself."

Step 5: Learn from Resolution

After human resolves: capture what the agent missed and feed back into training/prompts.

Slide 13 of 31Open this slide

UX Patterns for Human-in-the-Loop

How humans interact with agent pause points in practice.

Inline Approval

The simplest pattern: agent proposes, user confirms in chat.

• "I'd like to send this email to Alice. Proceed? [Yes / No / Edit]"
• Works well for single actions
• Low friction, stays in conversation flow
• Risk: approval fatigue if too frequent

Diff Review

Show before/after for modifications. Borrowed from code review.

• Side-by-side or unified diff of proposed changes
• Highlight additions, deletions, modifications
• Best for: file edits, database updates, config changes
• Example: Agent shows PR-style diff before applying code changes

Suggested Actions

Present multiple options with reasoning. User picks.

• "I found 3 approaches. Option A is fastest, Option B is most thorough..."
• Preserves human agency and decision-making
• Agent provides analysis, human provides judgment
• Works well when there's no single "right" answer

Batch Dashboard

For high-volume review, provide a dedicated UI.

• Table view of pending items with accept/reject buttons
• Bulk approve/reject with filters
• Priority sorting and SLA indicators
• Best for: content moderation, data processing, email triage

Slide 14 of 31Open this slide

Memory Types

Inspired by human cognition: different memory systems for different purposes.

Working Memory (Context Window)

The agent's "RAM" — what it can see right now.

• Current conversation context
• Active tool results and intermediate state
• Limited by token window (128K-1M tokens)
• Volatile: gone when the session ends
• Analogy: what's on your desk right now

Short-Term Memory (Conversation)

Tracks the current interaction across turns.

• Full message history for this session
• Maintains coherence within a conversation
• Strategies: full history, sliding window, summary
• Session-scoped: persists within one conversation
• Analogy: what you discussed in a meeting

Long-Term Memory (Persistent)

Survives across sessions. The agent's durable knowledge.

• User preferences, past decisions, learned facts
• Stored externally: databases, vector stores, files
• Requires explicit storage and retrieval mechanisms
• Permanent: available across all future sessions
• Analogy: your notes, journals, and personal knowledge base

Slide 15 of 31Open this slide

Conversation Memory Strategies

How to manage message history within a conversation.

Strategy Comparison

Full History

Keep every message. Simple but expensive.

• Pro: No information loss
• Con: Hits token limits on long conversations
• Use when: Short interactions, high-stakes accuracy

Sliding Window

Keep only the last N messages.

• Pro: Bounded cost, always fits in context
• Con: Loses early context
• Use when: Casual chat, FAQ-style interactions

Summary + Recent

Summarize old messages, keep recent ones verbatim.

• Pro: Preserves key context, bounded cost
• Con: Summary may lose nuance
• Use when: Long multi-turn workflows

Implementation

from langchain_core.messages import trim_messages

# Sliding window: keep last 10 messages
trimmed = trim_messages(
    messages,
    max_tokens=4000,
    strategy="last",
    token_counter=ChatOpenAI(model="gpt-4o"),
    include_system=True,  # Always keep system prompt
)

# Summary + recent hybrid
def summarize_and_trim(messages, threshold=20):
    if len(messages) > threshold:
        old = messages[:-10]
        recent = messages[-10:]
        summary = llm.invoke(
            f"Summarize this conversation:\n{old}"
        )
        return [SystemMessage(f"Previous context: {summary}")] + recent
    return messages

Tip: Always preserve the system message and the most recent user message regardless of strategy.

Sources

• LangChain trim_messages API Reference (LangChain)
• LangGraph Memory Documentation (LangChain)

Slide 16 of 31Open this slide

Summary Memory

Compress long conversations into running summaries. Keep the signal, discard the noise.

The problem

• A 100-turn conversation might use 50K tokens
• Most of those tokens are routine back-and-forth
• Only a fraction contains critical decisions and facts

Running summary pattern

from langchain.memory import ConversationSummaryBufferMemory

# Keeps recent messages verbatim + summary of older ones
memory = ConversationSummaryBufferMemory(
    llm=ChatOpenAI(model="gpt-4o-mini"),  # Cheap model for summaries
    max_token_limit=2000,  # Summarize when buffer exceeds this
    return_messages=True,
)

Custom summary with structured extraction

SUMMARY_PROMPT = """Summarize this conversation segment. Extract:
1. Key decisions made
2. User preferences expressed
3. Tasks completed and pending
4. Important facts or constraints mentioned

Previous summary: {existing_summary}
New messages: {new_messages}

Updated summary:"""

async def update_running_summary(existing_summary: str, new_messages: list):
    response = await llm.ainvoke(
        SUMMARY_PROMPT.format(
            existing_summary=existing_summary,
            new_messages=format_messages(new_messages)
        )
    )
    return response.content

Progressive summarization

• Level 1: Raw messages (keep last 10)
• Level 2: Turn-by-turn summary (keep last session)
• Level 3: Session summary (keep last 10 sessions)
• Level 4: Relationship summary (permanent, updated weekly)

Each level compresses further, trading detail for longevity.

Sources

• LangChain ConversationSummaryBufferMemory API Reference (LangChain)

Slide 17 of 31Open this slide

Entity Memory

Track structured facts about people, projects, and concepts across conversations.

What is entity memory?

A knowledge base of entities the agent has learned about, updated as new information emerges.

Entity extraction pattern

from pydantic import BaseModel

class EntityFact(BaseModel):
    entity_name: str          # "Alice Chen"
    entity_type: str          # "person", "project", "concept"
    attribute: str            # "role"
    value: str                # "Engineering Manager"
    source_conversation: str  # Thread ID where this was learned
    confidence: float         # How confident we are
    last_updated: datetime

# Extract entities from conversation
ENTITY_PROMPT = """Analyze this message and extract any facts about
people, projects, or concepts mentioned.

Message: {message}
Known entities: {existing_entities}

Return new or updated facts as structured data."""

async def extract_entities(message: str, existing: list[EntityFact]):
    response = await llm.ainvoke(
        ENTITY_PROMPT.format(message=message, existing_entities=existing)
    )
    return parse_entity_facts(response)

Example entity store

Storage options

• Key-value store — Simple, fast lookup by entity name
• Graph database — Rich relationships between entities (Neo4j)
• Structured JSON — Entity per document, easy to version
• SQL — Entity table with attribute columns, queryable

Key challenge: Entity resolution — "Alice", "Alice Chen", and "the engineering manager" may all be the same entity.

Slide 18 of 31Open this slide

Episodic Memory

Remember past interactions and outcomes. Learn from experience.

What is episodic memory?

Records of specific past events and their outcomes — not just facts, but stories the agent can learn from.

Episodic memory structure

class Episode(BaseModel):
    episode_id: str
    timestamp: datetime
    task_description: str       # What was the agent trying to do?
    actions_taken: list[str]    # What steps did it take?
    outcome: str                # What happened?
    success: bool               # Did it work?
    user_feedback: str | None   # What did the human say?
    lessons_learned: str        # What should be remembered for next time?

# Example episode
Episode(
    task_description="Deploy hotfix to production",
    actions_taken=[
        "Ran tests locally — passed",
        "Created PR and requested review",
        "Merged without waiting for CI — MISTAKE",
        "CI failed on staging after merge"
    ],
    outcome="Rollback required. 15 minutes of downtime.",
    success=False,
    user_feedback="Always wait for CI before merging hotfixes",
    lessons_learned="Even for urgent hotfixes, wait for CI pipeline. "
                    "The 10 min wait is cheaper than a rollback."
)

Using episodes for in-context learning

async def retrieve_relevant_episodes(current_task: str, k: int = 3):
    """Find past episodes similar to current task."""
    episodes = await vector_store.similarity_search(
        query=current_task, k=k, filter={"collection": "episodes"}
    )
    return format_episodes_for_context(episodes)

# Inject into system prompt
system_prompt = f"""You are a deployment assistant.
Here are relevant past experiences:
{retrieved_episodes}
Learn from past mistakes and successes."""

Episodic vs. semantic memory

• Episodic: "Last Tuesday, the deploy failed because we skipped CI"
• Semantic: "Always wait for CI before merging"
• Both are valuable — episodes provide context, semantic provides rules

Slide 19 of 31Open this slide

Vector-Based Long-Term Memory

Embed past experiences and retrieve by semantic similarity.

Implementation

from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings

# Initialize vector memory
memory_store = Chroma(
    collection_name="agent_memory",
    embedding_function=OpenAIEmbeddings(model="text-embedding-3-small"),
    persist_directory="./memory_db"
)

# Store a memory
def store_memory(content: str, metadata: dict):
    memory_store.add_texts(
        texts=[content],
        metadatas=[{
            **metadata,
            "timestamp": datetime.now().isoformat(),
            "type": metadata.get("type", "general"),  # episode, entity, preference
        }]
    )

# Retrieve relevant memories
def recall(query: str, k: int = 5, memory_type: str = None):
    filters = {"type": memory_type} if memory_type else None
    results = memory_store.similarity_search(
        query=query, k=k, filter=filters
    )
    return results

Tip: Combine vector similarity with metadata filters (recency, type, importance) for best results.

Sources

• LangChain Chroma Integration (LangChain)
• Chroma Documentation (Chroma)
• OpenAI Embeddings API (OpenAI)

Slide 20 of 31Open this slide

Persistent State

Checkpointing agent state for resume and replay.

Why persist state?

• Resume interrupted workflows (network failure, timeout, user away)
• Audit what the agent did and when
• Replay and debug past agent runs
• Enable long-running tasks that span hours or days

Checkpointing pattern

from langgraph.checkpoint.postgres import PostgresSaver

# Create a checkpointer backed by Postgres
checkpointer = PostgresSaver.from_conn_string(
    "postgresql://user:pass@localhost/agent_state"
)

# Compile graph with checkpointing
graph = workflow.compile(checkpointer=checkpointer)

# Every node execution is automatically checkpointed
# Resume from any point by using the same thread_id
config = {"configurable": {"thread_id": "task-abc-123"}}
result = graph.invoke(initial_state, config)

# Later: resume from where we left off
result = graph.invoke(None, config)  # Loads last checkpoint

What gets checkpointed:

• Full graph state at each node boundary
• Messages, tool results, intermediate values
• Which node to execute next
• Metadata: timestamps, version, parent checkpoint

Checkpoint storage options

Serverless alternative: Cloudflare Agents

You do not have to self-host the persistence layer. Cloudflare Agents is a serverless platform with built-in durable execution, human-approval hooks for HITL gating, and hibernation that lets long-running agents sleep between events without holding compute — an alternative to running your own LangGraph + Postgres stack.

Time-travel debugging

# List all checkpoints for a thread
history = list(graph.get_state_history(config))

# Inspect state at any point
past_state = history[5]  # 5th checkpoint
print(past_state.values)  # What was the state?
print(past_state.next)     # What was about to run?

# Fork from a past state and replay with different input
fork_config = {"configurable": {"thread_id": "task-abc-123-retry",
                                 "checkpoint_id": past_state.config["configurable"]["checkpoint_id"]}}
graph.invoke(Command(resume=new_input), fork_config)

Sources

• LangGraph Persistence Documentation (LangChain)
• Human-in-the-Loop with Cloudflare Agents (Cloudflare Docs)
• LangGraph Library (LangChain)

Slide 21 of 31Open this slide

Memory Architecture

Combining multiple memory types in one agent system.

Wiring it together

class AgentMemoryManager:
    def __init__(self):
        self.conversation = ConversationSummaryBuffer(llm=cheap_llm)
        self.entities = EntityStore(db=postgres)
        self.episodes = VectorStore(collection="episodes")
        self.preferences = KeyValueStore(db=redis)

    async def build_context(self, current_input: str) -> str:
        """Assemble working memory from all sources."""
        relevant_entities = await self.entities.search(current_input)
        similar_episodes = await self.episodes.similarity_search(current_input, k=3)
        user_prefs = await self.preferences.get_all()
        conv_summary = self.conversation.get_summary()

        return MEMORY_TEMPLATE.format(
            preferences=user_prefs,
            entities=relevant_entities,
            past_episodes=similar_episodes,
            conversation_summary=conv_summary,
        )

    async def store(self, interaction: Interaction):
        """After each turn, update all memory systems."""
        await self.conversation.add_messages(interaction.messages)
        await self.entities.extract_and_store(interaction)
        if interaction.is_notable:
            await self.episodes.store(interaction.to_episode())

Design principle: Each memory type answers a different question — "What did we just discuss?", "What do I know about X?", "What happened last time I tried this?"

The dual-layer production standard

The 2026 production pattern splits memory across two paths, coordinated by a dedicated Memory Node in the agent graph:

• Hot Path (in-session) — Low-latency retrieval that blends vector search, BM25 keyword matching, and entity lookup to assemble working context on every turn.
• Cold Path (cross-session) — Asynchronous consolidation that merges facts, resolves conflicts, and prunes stale entries so long-term memory does not bloat or drift.

The Memory Node owns the contract between them: the Hot Path serves the current request while the Cold Path runs out of band to keep the store coherent over time.

Sources

• State of AI Agent Memory 2026 (Mem0)

Slide 22 of 31Open this slide

Memory in LangGraph

Checkpointers, state persistence, and thread management.

LangGraph's memory model

LangGraph treats memory as persistent graph state, managed through checkpointers and stores.

Thread-level persistence (short-term)

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, MessagesState

# MemorySaver is development-only — state is lost when the process dies.
# Use AsyncPostgresSaver as the recommended production checkpointer (see below).
checkpointer = MemorySaver()

graph = StateGraph(MessagesState)
# ... define nodes and edges ...
app = graph.compile(checkpointer=checkpointer)

# Each thread_id maintains its own conversation state
config = {"configurable": {"thread_id": "user-123-session-1"}}
response = app.invoke({"messages": [("user", "Hello")]}, config)
# Next call with same thread_id continues the conversation
response = app.invoke({"messages": [("user", "Follow up")]}, config)

Cross-thread memory (long-term) with Store

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()  # Use PostgresStore in production

# Store memories scoped to a user (namespace)
store.put(("user", "alice"), "preferences", {
    "language": "Python",
    "style": "concise",
    "timezone": "US/Pacific"
})

# Access store inside graph nodes
def agent_node(state, config, store):
    user_id = config["configurable"]["user_id"]
    # Retrieve cross-thread memories
    memories = store.search(("user", user_id))
    prefs = store.get(("user", user_id), "preferences")
    # Use in response generation
    ...

Thread management patterns:

• One thread per conversation — Standard chat pattern
• One thread per task — Long-running workflows, resume on failure
• Thread branching — Fork from a checkpoint to explore alternatives
• Thread cleanup — Archive old threads, delete stale checkpoints

Production setup:

AsyncPostgresSaver is the recommended production checkpointer — durable, async, and survives process restarts (unlike the development-only MemorySaver).

from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

async with AsyncPostgresSaver.from_conn_string(DATABASE_URL) as checkpointer:
    await checkpointer.setup()  # Creates tables if needed
    graph = workflow.compile(checkpointer=checkpointer)

Sources

• LangGraph Memory Documentation (LangChain)
• LangGraph Persistence Documentation (LangChain)
• LangGraph Persistence (LangChain Docs, OSS Python)
• LangGraph v0.4: HITL, Checkpoints & State Persistence (AITechConnect)
• LangGraph Library (LangChain)

Slide 23 of 31Open this slide

Mem0 and Dedicated Memory Layers

Managed memory-as-a-service for agent systems.

The case for a memory layer

• Building memory from scratch is hard: extraction, deduplication, retrieval, decay
• Dedicated memory services handle the complexity
• Focus on your agent logic, not memory infrastructure

Mem0: memory-as-a-service

from mem0 import Memory

# Initialize with configuration
config = {
    "llm": {"provider": "openai", "config": {"model": "gpt-4o-mini"}},
    "vector_store": {"provider": "chroma", "config": {"collection_name": "memories"}},
}
memory = Memory.from_config(config)

# Add memories from conversation
memory.add(
    "I prefer dark mode and use VS Code as my editor",
    user_id="alice",
    metadata={"source": "preferences"}
)

# Memories are automatically extracted, deduplicated, and stored
# "I prefer dark mode" and "I use VS Code" become separate memory entries

# Search relevant memories
results = memory.search("What editor does Alice use?", user_id="alice")
# Returns: [{"memory": "Uses VS Code as editor", "score": 0.95}]

# Get all memories for a user
all_memories = memory.get_all(user_id="alice")

# Update conflicting memories automatically
memory.add("I switched to Cursor from VS Code", user_id="alice")
# Old "uses VS Code" memory is updated to "uses Cursor"

What Mem0 handles automatically:

• Extraction — Identifies memorable facts from conversations
• Deduplication — Avoids storing the same fact twice
• Conflict resolution — Updates stale memories when new info contradicts
• Retrieval — Multi-signal search combining semantic similarity, BM25, and entity matching

Benchmarks

On LoCoMo, Mem0's multi-signal retrieval (semantic + BM25 + entity) outperforms OpenAI's native memory by roughly 26% accuracy, at lower latency and token cost. The standard memory benchmarks in 2026 are LoCoMo, LongMemEval, and BEAM — use them to compare memory layers rather than trusting vendor claims.

Alternatives and similar tools:

• Zep — Long-term memory with temporal awareness and entity graphs
• Letta (MemGPT) — Self-editing memory with tiered storage
• Custom solutions — Vector DB + extraction prompts + cron cleanup

When to use managed vs. custom:

• Managed (Mem0/Zep): Fast to integrate, good defaults, limited customization
• Custom: Full control, specific domain needs, willing to maintain infrastructure

Sources

• Mem0 (mem0ai)
• Agent Memory Infrastructure: Mem0 vs OpenAI (MindStudio)
• Zep (getzep)
• Letta (formerly MemGPT) (letta-ai)
• MemGPT: Towards LLMs as Operating Systems (Packer et al., 2023)

Slide 24 of 31Open this slide

State Machines for Agent Workflows

Explicit state transitions with persistence for complex, multi-step agent tasks.

Implementation with LangGraph

from langgraph.graph import StateGraph, END
from typing import Literal

class WorkflowState(TypedDict):
    status: str
    request: str
    research: list[str]
    draft: str
    feedback: str | None

def route_after_review(state) -> Literal["executing", "revising"]:
    if state.get("approved"):
        return "executing"
    return "revising"

# Build the state machine
builder = StateGraph(WorkflowState)
builder.add_node("intake", intake_node)
builder.add_node("researching", research_node)
builder.add_node("drafting", draft_node)
builder.add_node("human_review", review_node)  # Interrupt here
builder.add_node("executing", execute_node)
builder.add_node("revising", revise_node)

builder.set_entry_point("intake")
builder.add_edge("intake", "researching")
builder.add_edge("researching", "drafting")
builder.add_edge("drafting", "human_review")
builder.add_conditional_edges("human_review", route_after_review)
builder.add_edge("executing", END)
builder.add_edge("revising", "drafting")  # Loop back

graph = builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["human_review"]  # HITL pause point
)

Why state machines? They make agent behavior predictable, debuggable, and auditable. Every transition is explicit.

Sources

• LangGraph Library (LangChain)
• LangGraph Interrupts Documentation (LangChain)

Slide 25 of 31Open this slide

Designing Multi-Turn Conversations

Context management strategies for long-running agent interactions.

The context budget problem

You have a fixed token window. Every turn adds messages. How do you decide what stays?

Strategy 1: Token-aware message management

def manage_context(messages: list, max_tokens: int = 100_000):
    """Keep messages within budget using priority-based trimming."""
    total = count_tokens(messages)
    if total <= max_tokens:
        return messages

    # Priority order: system > recent user > recent assistant > older
    system_msgs = [m for m in messages if m.role == "system"]
    recent = messages[-6:]  # Always keep last 3 exchanges
    older = messages[len(system_msgs):-6]

    # Summarize older messages
    summary = summarize(older)
    return system_msgs + [SystemMessage(f"Earlier context: {summary}")] + recent

Strategy 2: Scoped context injection

# Only inject memory relevant to the current turn
async def build_turn_context(user_message: str, memory_manager):
    relevant_memories = await memory_manager.search(user_message, k=5)
    active_task = await memory_manager.get_active_task()

    context_parts = []
    if active_task:
        context_parts.append(f"Active task: {active_task.summary}")
    if relevant_memories:
        context_parts.append(f"Relevant context: {format_memories(relevant_memories)}")

    return "\n".join(context_parts)

Conversation design patterns:

Explicit state tracking

• Track what phase the conversation is in (gathering requirements, executing, reviewing)
• Adjust context injection based on phase
• Prune irrelevant earlier phases

User intent continuity

• Detect when user changes topic vs. continues current thread
• On topic change: archive current context, start fresh
• On continuation: keep full relevant context

Graceful degradation

• When context is full, inform the user
• Offer to start a new thread with a summary
• Never silently drop important context

Anti-patterns to avoid:

• Stuffing entire conversation history into every call
• Ignoring token costs of tool results in context
• Not accounting for response tokens in budget
• Treating all messages as equally important

Slide 26 of 31Open this slide

Memory Retrieval

When to remember what — relevance scoring and decay.

The retrieval challenge

Having memories is useless if you retrieve the wrong ones. Good retrieval is the difference between a helpful agent and a confused one.

Retrieval scoring: multi-factor ranking

def score_memory(memory, query, current_time):
    """Score a memory based on multiple factors."""

    # 1. Semantic relevance (cosine similarity)
    relevance = cosine_similarity(embed(query), memory.embedding)

    # 2. Recency decay (exponential)
    age_hours = (current_time - memory.timestamp).total_seconds() / 3600
    recency = math.exp(-0.01 * age_hours)  # Half-life ~70 hours

    # 3. Importance (set at storage time)
    importance = memory.importance  # 0.0 - 1.0

    # 4. Access frequency (memories used often are more valuable)
    frequency = min(memory.access_count / 10, 1.0)

    # Weighted combination
    score = (
        0.4 * relevance +
        0.2 * recency +
        0.25 * importance +
        0.15 * frequency
    )
    return score

Temporal decay models

Importance scoring at storage time

IMPORTANCE_PROMPT = """Rate the importance of this memory on a scale of 0-10.

Memory: {memory_text}
Context: {conversation_context}

Consider:
- Is this a user preference or personal fact? (high)
- Is this a one-time procedural detail? (low)
- Would forgetting this cause problems later? (high)
- Is this widely available information? (low)

Return just the number."""

Retrieval strategies:

• Top-k similarity — Simple, fast, good baseline
• MMR (Maximum Marginal Relevance) — Diverse results, avoids redundancy
• Hybrid search — Combine keyword (BM25) + semantic for better recall
• Filtered retrieval — Pre-filter by metadata before similarity search
• Contextual compression — Retrieve broadly, then re-rank with LLM

Sources

• Generative Agents: Interactive Simulacra of Human Behavior (Park et al., 2023)
• MemGPT: Towards LLMs as Operating Systems (Packer et al., 2023)

Slide 27 of 31Open this slide

Privacy Considerations

What to remember, what to forget, and giving users control.

The memory privacy paradox

• More memory = better personalization = better user experience
• More memory = more data stored = more privacy risk
• Users want both: "Remember my preferences" AND "Don't be creepy"

What to remember (opt-in by default):

• Explicit preferences stated by user ("I prefer Python")
• Workflow configurations and tool settings
• Task-relevant facts shared in context
• Past decisions and their outcomes

What NOT to remember:

• Sensitive personal data (SSN, health info, financial details)
• Temporary emotional states or venting
• Information shared about third parties without their consent
• Anything the user asks you to forget

User control patterns

class MemoryPrivacyManager:
    async def forget(self, user_id: str, query: str):
        """Delete specific memories matching query."""
        memories = await self.store.search(query, user_id=user_id)
        for mem in memories:
            await self.store.delete(mem.id)
        return f"Deleted {len(memories)} memories matching '{query}'"

    async def export_memories(self, user_id: str) -> dict:
        """GDPR-style: export all stored memories for a user."""
        return await self.store.get_all(user_id=user_id)

    async def clear_all(self, user_id: str):
        """Nuclear option: delete everything for a user."""
        await self.store.delete_all(user_id=user_id)

    async def set_retention_policy(self, user_id: str, policy: RetentionPolicy):
        """User controls how long memories persist."""
        # e.g., auto-delete after 30 days, keep preferences only
        await self.policies.set(user_id, policy)

Design principles for memory privacy:

1. Transparency — Tell users what you remember and why
2. Control — Users can view, edit, and delete their memories
3. Minimization — Only store what is needed for the task
4. Retention limits — Auto-expire memories that are no longer relevant
5. Audit trail — Log when memories are created, accessed, and deleted
6. Scope boundaries — Work memories stay in work context, personal stays personal

Regulatory requirements:

• GDPR (EU) — Right to erasure, data portability, consent requirements
• CCPA (California) — Right to know what is collected, right to delete
• HIPAA (Healthcare) — Special handling for health-related information
• Best practice: Build for the strictest regulation you might face

Slide 28 of 31Open this slide

HITL + Memory: Combined Patterns for Production Agents

Putting it all together — how HITL and memory work in concert.

Safe Autonomous Agent

• Approval gates on irreversible actions
• Confidence-based routing for everything else
• Episodic memory of past mistakes prevents repeat errors
• Escalation to humans when confidence is low

Personalized Assistant

• Entity memory tracks user preferences across sessions
• Summary memory compresses long interactions
• HITL on first encounter with new action types
• Gradually increases autonomy as trust builds

Multi-Day Workflow Agent

• State machine with explicit HITL review checkpoints
• Persistent checkpointing survives restarts and failures
• Long-term memory of project context and decisions
• Review queues for batched human oversight

Collaborative Team Agent

• Shared entity memory across team conversations
• Individual preference memory per team member
• Approval gates for actions affecting shared resources
• Feedback loops capture team corrections and norms

Slide 29 of 31Open this slide

Key Takeaways

Design for Oversight, Not Just Autonomy

The goal is not to remove humans from the loop. It is to put them where they add the most value. Approval gates, confidence routing, and escalation patterns keep humans in control without bottlenecking every action.

Memory Transforms Stateless into Stateful

Without memory, every session starts from zero. Conversation memory, entity memory, episodic memory, and vector retrieval give agents continuity, personalization, and the ability to learn from experience.

Persistence Enables Reliability

Checkpointing agent state means workflows survive crashes, timeouts, and interruptions. Time-travel debugging makes agent behavior auditable and reproducible.

Privacy Is a Feature, Not an Afterthought

Users must control what agents remember. Build in transparency, deletion, export, and retention policies from day one. The best memory system is one users trust.

Slide 30 of 31Open this slide

Key Papers & Research

Essential reading on agent memory and human-in-the-loop systems.

Foundational papers

Recommended tools and frameworks

Further reading:

• LangGraph HITL documentation: interrupt patterns and breakpoints
• Anthropic research on Constitutional AI and human feedback
• OpenAI research on RLHF and InstructGPT methodology
• Episodic and Semantic Memory (Tulving, 1972)

Slide 31 of 31Open this slide

Week 11 Recap

Human-in-the-Loop + Memory = Production-Ready Agents

HITL gives agents guardrails. Approval gates, confidence routing, escalation, and review queues.

Memory gives agents continuity. Conversation, entity, episodic, and vector-based long-term recall.

Together they enable agents you can actually deploy.

Next week:

Week 12: Small Models and Model Routing — measuring quality, cost, latency, and hardware tradeoffs.