AI Agent Memory Systems Cut Costs 60% with Long-Term Context 2026

#!/usr/bin/env python3
"""
Production AI Agent Memory System
Implements episodic, semantic, and procedural memory with Redis + Pinecone
Author: Bhuvaneshwar A
"""

import asyncio
import json
from datetime import datetime, timedelta
from typing import List, Dict, Any, Optional
import redis.asyncio as redis
from pinecone import Pinecone, ServerlessSpec
from openai import AsyncOpenAI
import hashlib

class AgentMemorySystem:
    """
    Production-grade agent memory system with three memory types:
    - Episodic: Conversation history and events
    - Semantic: Learned facts and preferences
    - Procedural: Workflow patterns and successful actions
    """

    def __init__(
        self,
        redis_url: str,
        pinecone_api_key: str,
        openai_api_key: str,
        agent_id: str,
        user_id: str
    ):
        self.redis = redis.from_url(redis_url)
        self.pc = Pinecone(api_key=pinecone_api_key)
        self.openai = AsyncOpenAI(api_key=openai_api_key)
        self.agent_id = agent_id
        self.user_id = user_id

        # Initialize Pinecone index for semantic memory
        self.index_name = f"agent-memory-{agent_id}"
        if self.index_name not in self.pc.list_indexes().names():
            self.pc.create_index(
                name=self.index_name,
                dimension=1536,  # text-embedding-3-small
                metric='cosine',
                spec=ServerlessSpec(cloud='aws', region='us-east-1')
            )
        self.index = self.pc.Index(self.index_name)

        # Redis key patterns
        self.episodic_key = f"agent:{agent_id}:user:{user_id}:episodes"
        self.semantic_key = f"agent:{agent_id}:user:{user_id}:semantic"
        self.procedural_key = f"agent:{agent_id}:user:{user_id}:procedural"

    async def store_episodic_memory(
        self,
        conversation_id: str,
        messages: List[Dict[str, str]],
        metadata: Optional[Dict] = None
    ) -> str:
        """
        Store conversation episode with timestamp and metadata.

        Returns:
            episode_id: Unique identifier for this episode
        """
        episode_id = hashlib.sha256(
            f"{conversation_id}-{datetime.utcnow().isoformat()}".encode()
        ).hexdigest()[:16]

        episode = {
            'episode_id': episode_id,
            'conversation_id': conversation_id,
            'timestamp': datetime.utcnow().isoformat(),
            'messages': messages,
            'metadata': metadata or {},
            'message_count': len(messages)
        }

        # Store in Redis with 90-day TTL (configurable retention)
        await self.redis.zadd(
            self.episodic_key,
            {json.dumps(episode): datetime.utcnow().timestamp()}
        )
        await self.redis.expire(self.episodic_key, 90 * 24 * 3600)  # 90 days

        # Extract semantic knowledge asynchronously
        asyncio.create_task(self._extract_semantic_memory(messages, episode_id))

        return episode_id

    async def retrieve_episodic_memory(
        self,
        limit: int = 5,
        time_window_days: Optional[int] = None
    ) -> List[Dict]:
        """
        Retrieve recent conversation episodes.

        Args:
            limit: Maximum number of episodes to return
            time_window_days: Only return episodes from last N days
        """
        cutoff_time = None
        if time_window_days:
            cutoff_time = (
                datetime.utcnow() - timedelta(days=time_window_days)
            ).timestamp()

        # Get episodes from Redis sorted set (newest first)
        episodes = await self.redis.zrevrangebyscore(
            self.episodic_key,
            max='+inf',
            min=cutoff_time or '-inf',
            start=0,
            num=limit
        )

        return [json.loads(ep) for ep in episodes]

    async def _extract_semantic_memory(
        self,
        messages: List[Dict[str, str]],
        episode_id: str
    ):
        """
        Extract general knowledge and preferences from conversation.
        Uses GPT-5.2 to identify facts, preferences, and user characteristics.
        """
        conversation_text = "\n".join([
            f"{msg['role']}: {msg['content']}" for msg in messages
        ])

        extraction_prompt = f"""Analyze this conversation and extract:
1. User preferences (explicit statements like "I prefer X")
2. Facts about the user (e.g., job title, tech stack, interests)
3. Behavioral patterns (e.g., asks for examples, prefers concise answers)

Conversation:
{conversation_text}

Return JSON with:
{{
  "preferences": ["preference 1", "preference 2"],
  "facts": ["fact 1", "fact 2"],
  "patterns": ["pattern 1", "pattern 2"]
}}
"""

        try:
            response = await self.openai.chat.completions.create(
                model="gpt-4o-mini",  # Cheaper for extraction
                messages=[{"role": "user", "content": extraction_prompt}],
                response_format={"type": "json_object"},
                temperature=0.3
            )

            extracted = json.loads(response.choices[0].message.content)

            # Store each fact as semantic memory with embedding
            for fact_list in extracted.values():
                for fact in fact_list:
                    await self.store_semantic_memory(fact, episode_id)

        except Exception as e:
            print(f"Error extracting semantic memory: {e}")

    async def store_semantic_memory(
        self,
        fact: str,
        source_episode_id: str
    ) -> str:
        """
        Store a learned fact or preference as semantic memory.
        Uses embeddings for efficient retrieval.
        """
        # Generate embedding
        embedding_response = await self.openai.embeddings.create(
            model="text-embedding-3-small",
            input=fact
        )
        embedding = embedding_response.data[0].embedding

        # Create unique ID for this fact
        fact_id = hashlib.sha256(fact.encode()).hexdigest()[:16]

        # Store in Pinecone with metadata
        self.index.upsert(
            vectors=[{
                'id': f"{self.user_id}-{fact_id}",
                'values': embedding,
                'metadata': {
                    'fact': fact,
                    'source_episode': source_episode_id,
                    'timestamp': datetime.utcnow().isoformat(),
                    'user_id': self.user_id,
                    'agent_id': self.agent_id
                }
            }],
            namespace=f"semantic-{self.agent_id}"
        )

        return fact_id

    async def retrieve_semantic_memory(
        self,
        query: str,
        top_k: int = 5
    ) -> List[Dict]:
        """
        Retrieve relevant semantic memories using similarity search.
        """
        # Generate query embedding
        embedding_response = await self.openai.embeddings.create(
            model="text-embedding-3-small",
            input=query
        )
        query_embedding = embedding_response.data[0].embedding

        # Search Pinecone
        results = self.index.query(
            vector=query_embedding,
            top_k=top_k,
            include_metadata=True,
            namespace=f"semantic-{self.agent_id}",
            filter={'user_id': self.user_id}
        )

        return [
            {
                'fact': match.metadata['fact'],
                'relevance_score': match.score,
                'source': match.metadata['source_episode'],
                'timestamp': match.metadata['timestamp']
            }
            for match in results.matches
        ]

    async def store_procedural_memory(
        self,
        workflow_name: str,
        steps: List[str],
        success: bool,
        metadata: Optional[Dict] = None
    ):
        """
        Store a workflow pattern for future reuse.
        """
        workflow = {
            'name': workflow_name,
            'steps': steps,
            'success': success,
            'timestamp': datetime.utcnow().isoformat(),
            'metadata': metadata or {}
        }

        # Store in Redis hash with workflow name as key
        await self.redis.hset(
            self.procedural_key,
            workflow_name,
            json.dumps(workflow)
        )
        await self.redis.expire(self.procedural_key, 180 * 24 * 3600)  # 180 days

    async def retrieve_procedural_memory(
        self,
        workflow_name: Optional[str] = None
    ) -> Dict[str, Any]:
        """
        Retrieve learned workflow patterns.
        """
        if workflow_name:
            workflow_data = await self.redis.hget(self.procedural_key, workflow_name)
            return json.loads(workflow_data) if workflow_data else None
        else:
            # Return all workflows
            all_workflows = await self.redis.hgetall(self.procedural_key)
            return {
                k.decode(): json.loads(v.decode())
                for k, v in all_workflows.items()
            }

    async def get_contextual_memory(
        self,
        current_query: str,
        episodic_limit: int = 3,
        semantic_limit: int = 5
    ) -> Dict[str, Any]:
        """
        Retrieve all relevant memories for current context.
        This is what gets passed to the LLM for each request.
        """
        # Get recent episodes
        episodes = await self.retrieve_episodic_memory(limit=episodic_limit)

        # Get relevant semantic memories
        semantic_memories = await self.retrieve_semantic_memory(
            query=current_query,
            top_k=semantic_limit
        )

        # Get relevant workflows (if query suggests a known pattern)
        procedural_memories = await self.retrieve_procedural_memory()

        return {
            'episodic': episodes,
            'semantic': semantic_memories,
            'procedural': list(procedural_memories.values())[:3],  # Top 3 workflows
            'memory_summary': self._create_memory_summary(
                episodes, semantic_memories, procedural_memories
            )
        }

    def _create_memory_summary(
        self,
        episodes: List[Dict],
        semantic: List[Dict],
        procedural: Dict
    ) -> str:
        """
        Create human-readable memory summary for LLM context.
        """
        summary = "## User Context from Memory\n\n"

        if episodes:
            summary += "**Recent Interactions:**\n"
            for ep in episodes[:2]:  # Only summarize 2 most recent
                summary += f"- {ep['metadata'].get('summary', 'Previous conversation')}\n"

        if semantic:
            summary += "\n**Known Preferences & Facts:**\n"
            for mem in semantic[:5]:
                summary += f"- {mem['fact']}\n"

        if procedural:
            summary += "\n**Successful Workflow Patterns:**\n"
            for name, workflow in list(procedural.items())[:2]:
                if workflow.get('success'):
                    summary += f"- {name}: {' → '.join(workflow['steps'][:3])}\n"

        return summary

    async def cleanup_old_memories(self, days_to_keep: int = 90):
        """
        Remove old episodic memories beyond retention policy.
        Semantic and procedural memories are kept longer.
        """
        cutoff_time = (datetime.utcnow() - timedelta(days=days_to_keep)).timestamp()

        # Remove old episodes from Redis
        removed = await self.redis.zremrangebyscore(
            self.episodic_key,
            min='-inf',
            max=cutoff_time
        )

        return removed


# Example Usage
async def main():
    memory_system = AgentMemorySystem(
        redis_url="redis://localhost:6379",
        pinecone_api_key="your-pinecone-key",
        openai_api_key="your-openai-key",
        agent_id="customer-service-bot",
        user_id="user-12345"
    )

    # Store a conversation
    messages = [
        {"role": "user", "content": "I need help with my deployment"},
        {"role": "assistant", "content": "I can help. What cloud provider are you using?"},
        {"role": "user", "content": "We use AWS with EKS"},
        {"role": "assistant", "content": "Great, let me check your EKS cluster..."}
    ]

    episode_id = await memory_system.store_episodic_memory(
        conversation_id="conv-abc123",
        messages=messages,
        metadata={"topic": "deployment", "resolved": True}
    )

    print(f"Stored episode: {episode_id}")

    # Later, retrieve context for new query
    context = await memory_system.get_contextual_memory(
        current_query="How do I scale my deployment?"
    )

    print("\n## Retrieved Memory Context:")
    print(context['memory_summary'])


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