From Reddit to Papr: Feature Mapping

1. Event Log (SQLite/Postgres) - Transparent, debuggable storage
2. Vector Search (Pinecone/Weaviate) - Semantic similarity
3. Keyword Search (BM25/FTS5) - Exact token matching
4. Knowledge Graph (Neo4j/custom) - Real relationships
5. Fusion Layer (RRF) - Combine results from 1-4
6. Scoring Layer - Recency, importance, feedback signals
7. Consolidation Jobs (Airflow/cron) - Episodes → Facts
8. ACL System - Multi-tenant isolation

# Store everything in SQLite for debugging
db.execute("""
    CREATE TABLE events (
        id INTEGER PRIMARY KEY,
        content TEXT,
        user_id TEXT,
        timestamp DATETIME,
        metadata JSON
    )
""")

# Direct Memory API - explicit storage with full control
client.memory.add(
    content="User prefers email notifications",
    metadata={
        "external_user_id": "user_123",
        "timestamp": "2024-03-21T10:00:00Z",
        "source": "preferences"
    }
)

# Add Pinecone/Weaviate for semantic search
embedding = model.encode(query)
results = pinecone.query(embedding, top_k=10)

# Built-in semantic search
results = client.memory.search(
    query="notification preferences",  # Automatically embedded
    max_memories=10
)

# Add FTS5 for keyword matching
db.execute("""
    CREATE VIRTUAL TABLE events_fts 
    USING fts5(content, user_id)
""")
keyword_results = db.execute(
    "SELECT * FROM events_fts WHERE content MATCH ?",
    query
)

# Hybrid retrieval includes keyword matching
results = client.memory.search(
    query="email notification",  # Matches exact tokens + semantics
    enable_agentic_graph=True
)

# Add Neo4j for relationships
neo4j.execute("""
    CREATE (u:User {id: 'user_123'})
    CREATE (p:Preference {type: 'notification', value: 'email'})
    CREATE (u)-[:HAS_PREFERENCE]->(p)
""")

# Automatic entity extraction and linking
client.memory.add(
    content="User prefers email notifications",
    memory_policy={"mode": "auto"}  # Auto-extracts: User -[HAS_PREFERENCE]-> Notification
)

# Query with GraphQL
insights = client.graphql.query("""
    query {
        user(id: "user_123") {
            preferences { type, value }
        }
    }
""")

# Manually combine results from vector + keyword + graph
def reciprocal_rank_fusion(results_lists, k=60):
    scores = {}
    for results in results_lists:
        for rank, item in enumerate(results, 1):
            scores[item] = scores.get(item, 0) + 1 / (k + rank)
    return sorted(scores.items(), key=lambda x: x[1], reverse=True)

# Combine 3 result sets
keyword_results = bm25_search(query)
vector_results = pinecone_search(query)
graph_results = neo4j_traverse(entities)
final = reciprocal_rank_fusion([keyword_results, vector_results, graph_results])

# Automatic fusion of all retrieval methods
results = client.memory.search(
    query="notification preferences",
    enable_agentic_graph=True  # Fuses keyword + vector + graph automatically
)

# Add custom scoring
def score_results(results):
    for r in results:
        # Recency decay
        days_old = (now - r.timestamp).days
        r.score *= 0.95 ** days_old
        
        # Importance boost
        if r.feedback_type == "thumbs_up":
            r.score *= 1.5
        
        # Hotness (access frequency)
        r.score *= math.log(1 + r.access_count)
    
    return sorted(results, key=lambda x: x.score, reverse=True)

# Built-in predictive scoring
results = client.memory.search(
    query="notification preferences",
    enable_agentic_graph=True
)
# Scoring formula (automatic):
# predicted_importance = 0.6 × vector_similarity 
#                      + 0.3 × transition_probability 
#                      + 0.2 × normalized_hotness

# Provide feedback to improve scoring
client.feedback.submit(
    search_id=results.search_id,
    feedback_data={"feedback_type": "thumbs_up"}
)

# Cron job: Consolidate repeated patterns
@cron.schedule("0 2 * * *")  # 2am daily
def consolidate_memories():
    # Find repeated interactions
    for user in users:
        events = db.get_events(user, last_7_days)
        
        # Custom summarization logic
        if "notification" in events.frequent_topics():
            summary = summarize_preferences(events)
            db.store_consolidated_fact(user, summary)

# Automatic analysis (background)
client.messages.store(
    content="I prefer email notifications",
    role="user",
    session_id="session_123",
    process_messages=True  # Triggers background analysis
)

# Later retrieval returns consolidated facts, not raw events
results = client.memory.search(
    query="What are the user's notification preferences?",
    enable_agentic_graph=True
)
# Returns: Consolidated preference, not individual messages

# Manual tenant filtering (error-prone)
def get_memories(query, user_id, org_id):
    results = db.execute("""
        SELECT * FROM memories 
        WHERE org_id = ? AND user_id = ?  # Easy to forget!
        AND content MATCH ?
    """, org_id, user_id, query)
    return results

# Built-in namespace isolation
client.memory.search(
    query="preferences",
    organization_id="org_123",
    namespace_id="customer_456"
    # Can't query across namespaces - enforced at API level
)

# 1. Store in SQLite
db.execute("INSERT INTO events ...")

# 2. Generate embedding
embedding = model.encode(content)

# 3. Index in Pinecone
pinecone.upsert(embedding)

# 4. Update FTS5 index
db.execute("INSERT INTO events_fts ...")

# 5. Extract entities
entities = ner_model(content)

# 6. Create graph relationships
for entity in entities:
    neo4j.create(entity)

# 7. At query time: Retrieve from 3 sources
keyword = db.execute("SELECT * FROM events_fts WHERE content MATCH ?")
vector = pinecone.query(query_embedding)
graph = neo4j.traverse(entities)

# 8. Manually fuse results
final = reciprocal_rank_fusion([keyword, vector, graph])

# 9. Apply custom scoring
scored = apply_scoring(final, recency_weight, importance_weight)

# 10. Filter by tenant
filtered = [r for r in scored if r.org_id == org_id]

return filtered

# Store (automatic indexing, entity extraction, graph creation)
client.messages.store(
    content="I prefer email notifications",
    session_id="session_123",
    external_user_id="user_123",
    process_messages=True
)

# Retrieve (automatic fusion, scoring, filtering)
results = client.memory.search(
    query="notification preferences",
    external_user_id="user_123",
    enable_agentic_graph=True,
    organization_id="org_123"
)

# Store interaction
db.execute("INSERT INTO interactions (user_id, content, timestamp) VALUES (?, ?, ?)")
embedding = model.encode(content)
pinecone.upsert(embedding)
entities = ner.extract(content)
for e in entities:
    neo4j.create(e)

# Retrieve for next interaction (3 weeks later)
query_embedding = model.encode("What did we discuss about billing?")
vector_results = pinecone.query(query_embedding, top_k=10)
keyword_results = db.execute("SELECT * FROM interactions_fts WHERE content MATCH 'billing'")
graph_results = neo4j.query("""
    MATCH (u:User {id: 'user_123'})-[*1..2]-(n)
    WHERE n.content CONTAINS 'billing'
    RETURN n
""")

# Fuse and rank
combined = reciprocal_rank_fusion([vector_results, keyword_results, graph_results])
scored = apply_recency_boost(combined)
top_10 = scored[:10]

# Consolidate (manual job)
if len(user_interactions) > 100:
    summary = llm.summarize(user_interactions)
    db.execute("INSERT INTO consolidated_facts ...")

# Store interaction
client.messages.store(
    content="I want to update my billing information",
    session_id="support_123",
    external_user_id="user_123",
    process_messages=True  # Auto-extracts: Billing, Update Request
)

# Retrieve for next interaction (3 weeks later)
results = client.memory.search(
    query="What did we discuss about billing?",
    external_user_id="user_123",
    enable_agentic_graph=True  # Auto-fuses keyword + vector + graph
)
# Returns: Consolidated context including:
# - Original billing discussion
# - Related account information
# - Previous billing changes
# - Connected support tickets
# All ranked by predicted relevance

# Keep existing system running
existing_store(content)

# Start populating Papr
client.memory.add(content, metadata=existing_metadata)

# Compare retrieval quality
existing_results = your_retrieval_pipeline(query)
papr_results = client.memory.search(query, enable_agentic_graph=True)

# Measure accuracy, latency, relevance

# Route queries based on confidence
if query_type == "simple":
    results = existing_system(query)
else:
    results = papr_search(query)

# Replace DIY stack with Papr
results = client.memory.search(query, enable_agentic_graph=True)