AI Agent Memory Architecture: Long-Term, Persistent Cognition for Production Agents

import asyncio
import json
from datetime import datetime, timedelta
from typing import Optional
import redis.asyncio as redis
import asyncpg
from openai import AsyncOpenAI
from langmem import AsyncMemoryClient  # LangMem 0.1+ (LangChain, 2025)
from qdrant_client import AsyncQdrantClient
from qdrant_client.models import PointStruct, Distance, VectorParams
import uuid


class ThreeLayerAgentMemory:
    """
    Unified three-layer memory system for persistent enterprise agents.
    - Episodic: Redis Streams + Qdrant (interaction history with semantic search)
    - Semantic: Postgres JSONB (facts, preferences, entity knowledge)
    - Procedural: Redis Hash (few-shot example cache)
    """

    def __init__(
        self,
        redis_url: str,
        postgres_dsn: str,
        qdrant_host: str,
        langmem_api_key: str,
    ):
        self.redis_url = redis_url
        self.postgres_dsn = postgres_dsn
        self.qdrant = AsyncQdrantClient(host=qdrant_host, port=6333)
        self.openai = AsyncOpenAI()
        self.langmem = AsyncMemoryClient(api_key=langmem_api_key)
        self._redis: Optional[redis.Redis] = None
        self._pg: Optional[asyncpg.Connection] = None

    async def connect(self):
        self._redis = await redis.from_url(self.redis_url, decode_responses=True)
        self._pg = await asyncpg.connect(self.postgres_dsn)
        await self._pg.execute("""
            CREATE TABLE IF NOT EXISTS agent_semantic_memory (
                entity_id TEXT NOT NULL,
                key TEXT NOT NULL,
                value JSONB NOT NULL,
                confidence FLOAT DEFAULT 1.0,
                updated_at TIMESTAMPTZ DEFAULT NOW(),
                PRIMARY KEY (entity_id, key)
            )
        """)
        await self._ensure_qdrant_collection()

    async def _ensure_qdrant_collection(self):
        collections = await self.qdrant.get_collections()
        names = [c.name for c in collections.collections]
        if "episodic_memory" not in names:
            await self.qdrant.create_collection(
                "episodic_memory",
                vectors_config=VectorParams(size=768, distance=Distance.COSINE)
            )

    async def _embed(self, text: str) -> list[float]:
        resp = await self.openai.embeddings.create(
            input=text, model="text-embedding-3-small", dimensions=768
        )
        return resp.data[0].embedding

    # --- EPISODIC MEMORY ---

    async def store_episode(self, entity_id: str, session_id: str,
                            summary: str, metadata: dict) -> None:
        """Store a session summary as an episodic memory."""
        vector = await self._embed(summary)
        point_id = str(uuid.uuid4())
        await self.qdrant.upsert(
            "episodic_memory",
            points=[PointStruct(
                id=point_id,
                vector=vector,
                payload={
                    "entity_id": entity_id,
                    "session_id": session_id,
                    "summary": summary,
                    "timestamp": datetime.utcnow().isoformat(),
                    **metadata
                }
            )]
        )
        # TTL via Redis (90-day expiry marker)
        await self._redis.setex(
            f"ep_ttl:{point_id}", int(timedelta(days=90).total_seconds()), "1"
        )

    async def recall_episodes(
        self, entity_id: str, query: str, top_k: int = 5
    ) -> list[dict]:
        """Semantically retrieve relevant past episodes."""
        from qdrant_client.models import Filter, FieldCondition, MatchValue
        vector = await self._embed(query)
        results = await self.qdrant.search(
            "episodic_memory",
            query_vector=vector,
            query_filter=Filter(must=[
                FieldCondition(key="entity_id", match=MatchValue(value=entity_id))
            ]),
            limit=top_k,
            with_payload=True,
        )
        return [{"summary": r.payload["summary"],
                 "session_id": r.payload["session_id"],
                 "timestamp": r.payload["timestamp"],
                 "score": r.score} for r in results]

    # --- SEMANTIC MEMORY ---

    async def store_fact(self, entity_id: str, key: str,
                         value: dict, confidence: float = 1.0) -> None:
        """Store or update a structured fact about an entity."""
        await self._pg.execute("""
            INSERT INTO agent_semantic_memory (entity_id, key, value, confidence, updated_at)
            VALUES ($1, $2, $3::jsonb, $4, NOW())
            ON CONFLICT (entity_id, key)
            DO UPDATE SET value=$3::jsonb, confidence=$4, updated_at=NOW()
        """, entity_id, key, json.dumps(value), confidence)

    async def get_facts(self, entity_id: str) -> dict:
        """Retrieve all semantic facts for an entity."""
        rows = await self._pg.fetch(
            "SELECT key, value, confidence FROM agent_semantic_memory WHERE entity_id = $1",
            entity_id
        )
        return {row["key"]: {"value": json.loads(row["value"]),
                              "confidence": row["confidence"]} for row in rows}

    # --- PROCEDURAL MEMORY ---

    async def store_successful_pattern(
        self, pattern_key: str, example: dict, score: float
    ) -> None:
        """Cache a successful interaction pattern for few-shot retrieval."""
        await self._redis.zadd(
            f"procedural:{pattern_key}",
            {json.dumps(example): score}
        )
        await self._redis.zremrangebyrank(f"procedural:{pattern_key}", 0, -11)  # Keep top 10

    async def retrieve_patterns(self, pattern_key: str, top_k: int = 3) -> list[dict]:
        """Retrieve top-scoring procedural patterns for few-shot prompting."""
        raw = await self._redis.zrevrange(
            f"procedural:{pattern_key}", 0, top_k - 1
        )
        return [json.loads(item) for item in raw]

    # --- MEMORY CONSOLIDATION ---

    async def consolidate_memory(self, entity_id: str) -> None:
        """Extract persistent facts from recent episodes and update semantic memory."""
        recent = await self.recall_episodes(entity_id, "recent interactions", top_k=10)
        if len(recent) < 5:
            return  # Not enough episodes to consolidate
        episodes_text = "\n\n".join(
            f"[{ep['timestamp']}]\n{ep['summary']}" for ep in recent
        )
        resp = await self.openai.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{
                "role": "user",
                "content": f"""Extract persistent facts from these interaction summaries.
Return JSON: {{"facts": [{{"key": "str", "value": {{"summary": "str"}}, "confidence": 0.0-1.0}}]}}
\n{episodes_text}"""
            }],
            temperature=0, response_format={"type": "json_object"}
        )
        extracted = json.loads(resp.choices[0].message.content)
        for fact in extracted.get("facts", []):
            await self.store_fact(entity_id, fact["key"],
                                  fact["value"], fact["confidence"])


# Usage example
async def main():
    memory = ThreeLayerAgentMemory(
        redis_url="redis://localhost:6379",
        postgres_dsn="postgresql://user:pass@localhost/agentdb",
        qdrant_host="localhost",
        langmem_api_key="lm-..."
    )
    await memory.connect()
    await memory.store_episode(
        entity_id="customer_acme",
        session_id="sess_2025101401",
        summary="Customer raised concern about Q4 pricing. Agreed to review in January.",
        metadata={"sentiment": "neutral", "topics": ["pricing", "renewal"]}
    )
    episodes = await memory.recall_episodes(
        "customer_acme", "pricing concerns", top_k=3
    )
    print(episodes)


if __name__ == "__main__":
    asyncio.run(main())