Case 5 — Cross-Session Long-Term Memory Agent

Executive Briefing: A digital wealth advisory agent must maintain persistent customer context across months of interactions (e.g. risk tolerance, children's college target years, tax filing bracket). The agent combines an active scratchpad for intermediate calculations, PostgreSQL checkpointing for in-session conversational state, and a durable user_memories relational table with semantic category indexing to recall user preferences across separate sessions.

1. Enterprise Scenario & Requirements

In-session chat history is insufficient for long-lived client relationships. In this project:

1. Database Schema: Create a PostgreSQL table user_memories with columns: id, user_id, fact, category (preference, financial_goal, constraint), importance (1-10), and last_referenced_at.
2. Memory Tools: Equip the agent with recall_user_memories(user_id, query) and store_user_memory(user_id, fact, category).
3. Tenant Privacy: Strictly enforce user_id parameter isolation so that User A cannot retrieve or overwrite User B's memories under any prompt injection attempt.
4. Session Rehydration: Demonstrate an agent waking up in a new thread, querying memory for the user's previously stated risk tolerance, and tailoring advice without asking the user to repeat themselves.

2. Explicit Deliverables

1. schema.sql: PostgreSQL DDL for user_memories with indices on (user_id, category) and user_id.
2. memory_service.py: CRUD operations with SQL query parameterization and user-scoping enforcement.
3. agent.py: LangGraph conversational agent equipped with memory retrieval and storage tools.
4. test_memory.py: Automated tests verifying: (a) persistent recall across differing thread_id sessions, (b) tenant isolation preventing cross-user data leakage, and (c) memory deduplication.

3. Starter Code & Scaffold

-- DDL for durable cross-session agent memory
CREATE TABLE IF NOT EXISTS user_memories (
    id SERIAL PRIMARY KEY,
    user_id TEXT NOT NULL,
    fact TEXT NOT NULL,
    category TEXT CHECK (category IN ('preference', 'financial_goal', 'constraint', 'general')),
    importance INTEGER DEFAULT 5 CHECK (importance BETWEEN 1 AND 10),
    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    last_referenced_at TIMESTAMP WITH TIME ZONE
);
CREATE INDEX IF NOT EXISTS idx_user_memories_user_cat ON user_memories(user_id, category);

from langchain_core.tools import tool
from pydantic import BaseModel, Field

class MemoryStoreInput(BaseModel):
    user_id: str = Field(description="Unique tenant or user identifier")
    fact: str = Field(description="Specific, timeless fact learned about the user")
    category: str = Field(description="One of: 'preference', 'financial_goal', 'constraint'")

@tool(args_schema=MemoryStoreInput)
def store_user_fact(user_id: str, fact: str, category: str) -> str:
    '''Persist an important personal or financial fact to long-term memory.'''
    # Insert into PostgreSQL user_memories table
    return f"Stored memory for user {user_id}: [{category}] {fact}"

4. 100-Point Grading Rubric