Prem Cortex: Human-Like Memory for Smarter Agents

# Agent creates specialized searchers
recent_searcher = AgenticMemorySystem(
    enable_smart_collections=False,  # Skip for speed
    stm_capacity=200  # Larger recent window
)

deep_searcher = AgenticMemorySystem(
    enable_smart_collections=True,  # Full intelligence
    enable_background_processing=True
)

# Agent runs parallel searches with different strategies
results = await asyncio.gather(
    recent_searcher.search(query, temporal_weight=0.9),  # Recent-focused
    deep_searcher.search(query, memory_source="ltm"),    # Deep historical
    deep_searcher.search(query, date_range="2024-01")    # Time-boxed
)

import os
from typing import Optional, List, Dict, Any
from datetime import datetime, timedelta
from pydantic import BaseModel, Field
from langchain.tools import StructuredTool
from cortex.memory_system import AgenticMemorySystem

# Initialize Cortex (enables smart collections if OPENAI_API_KEY is set)
api_key = os.getenv("OPENAI_API_KEY")
memory = AgenticMemorySystem(
    api_key=api_key,
    enable_smart_collections=bool(api_key),
    model_name="all-MiniLM-L6-v2",  # local/fast embeddings
)

# Tool input schemas
class CortexAddNoteInput(BaseModel):
    content: str
    user_id: Optional[str] = None
    session_id: Optional[str] = None
    time: Optional[str] = Field(
        default=None, description="RFC3339 timestamp (e.g., 2025-08-07T10:00:00+00:00)"
    )
    context: Optional[str] = None
    tags: Optional[List[str]] = None
    metadata: Optional[Dict[str, Any]] = None

class CortexSearchInput(BaseModel):
    query: str
    limit: int = Field(default=10, ge=1, le=50)
    memory_source: str = Field(
        default="all", description='"stm" | "ltm" | "all"'
    )
    temporal_weight: float = Field(
        default=0.0, ge=0.0, le=1.0, description="0.0 = pure semantic, 1.0 = pure recency"
    )
    date_range: Optional[str] = Field(
        default=None, description='e.g. "last week", "2024-01", RFC3339 range'
    )
    where_filter: Optional[Dict[str, Any]] = None
    user_id: Optional[str] = None
    session_id: Optional[str] = None

# Tool functions

def cortex_add_note_fn(
    content: str,
    user_id: Optional[str] = None,
    session_id: Optional[str] = None,
    time: Optional[str] = None,
    context: Optional[str] = None,
    tags: Optional[List[str]] = None,
    metadata: Optional[Dict[str, Any]] = None,
):
    return memory.add_note(
        content=content,
        user_id=user_id,
        session_id=session_id,
        time=time,
        context=context,
        tags=tags,
        **(metadata or {}),
    )

def cortex_search_fn(
    query: str,
    limit: int = 10,
    memory_source: str = "all",
    temporal_weight: float = 0.0,
    date_range: Optional[str] = None,
    where_filter: Optional[Dict[str, Any]] = None,
    user_id: Optional[str] = None,
    session_id: Optional[str] = None,
):
    return memory.search(
        query=query,
        limit=limit,
        memory_source=memory_source,
        temporal_weight=temporal_weight,
        date_range=date_range,
        where_filter=where_filter,
        user_id=user_id,
        session_id=session_id,
    )

# Structured tools with schemas
cortex_add_note = StructuredTool.from_function(
    name="cortex_add_note",
    description="Store a memory note (supports time/user/session/context/tags/metadata)",
    func=cortex_add_note_fn,
    args_schema=CortexAddNoteInput,
)

cortex_search = StructuredTool.from_function(
    name="cortex_search",
    description="Search memories with optional temporal/date filtering and metadata filters",
    func=cortex_search_fn,
    args_schema=CortexSearchInput,
)

# Example usage (agents will call .invoke under the hood)
_ = cortex_add_note.invoke(
    {
        "content": "User prefers TypeScript over JavaScript",
        "user_id": "user_123", # optional
        "session_id": "onboarding", # optional
        "time": datetime.now().astimezone().isoformat(), # optional
        "context": "preferences.programming", # optional
        "tags": ["typescript", "language"], # optional
        "metadata": {"source": "chat"}, # optional
    }
)

results = cortex_search.invoke(
    {
        "query": "programming preferences",
        "limit": 5,
        "memory_source": "all",
        "temporal_weight": 0.2, # optional
        "date_range": "last week", # optional
        "where_filter": {"context": {"$eq": "preferences.programming"}}, # optional
        "user_id": "user_123", # optional
        "session_id": "onboarding", # include if you added with a session
    }
)
print(f"Top result: {results[0]['content'][:60]}..." if results else "No results")