MongoDB State Store

Flexible document database for complex state, metadata, and schema-less storage

MongoDB provides flexible document-based state storage, ideal for applications with complex metadata, evolving schemas, and rich query requirements.

Quick Start

1. Basic Document Storage

   from praisonaiagents import Agent, db
   
   agent = Agent(
       name="DocumentBot",
       instructions="You are a helpful assistant.",
       db=db(state_url="mongodb://localhost:27017/praisonai"),
       session_id="document-session"
   )
   
   # Agent state automatically stored as MongoDB documents
   response = agent.chat("Store this conversation with rich metadata")
   print(response)  # State persisted as flexible MongoDB documents
   
2. With Authentication

   from praisonaiagents import Agent, db
   
   # Production setup with authentication
   agent = Agent(
       name="SecureBot",
       instructions="You are a helpful assistant.",
       db=db(state_url="mongodb://username:[e-mail]:27017/praisonai"),
       session_id="secure-session"
   )
   
   agent.chat("This connection uses MongoDB authentication")
   

---

How It Works

MongoDB organizes state data in flexible document collections:

Collection	Document Type	Features
agent_state	Agent metadata, preferences	Rich nested objects, arrays
session_data	Session information, context	Schema-less flexibility
user_profiles	User preferences, history	Complex relationships
analytics	Usage metrics, performance	Time-series data, aggregations

---

Configuration Options

MongoDB URL Formats

# Basic connection
db(state_url="mongodb://localhost:27017/database")

# With authentication
db(state_url="mongodb://username:password@host:27017/database")

# Replica set
db(state_url="mongodb://host1:27017,host2:27017,host3:27017/database?replicaSet=rs0")

# SSL connection
db(state_url="mongodb://user:pass@host:27017/db?ssl=true&ssl_cert_reqs=CERT_REQUIRED")

# Connection pool options
db(state_url="mongodb://host:27017/db?maxPoolSize=50&minPoolSize=5")

Advanced MongoDB Configuration

from praisonaiagents import Agent, db

# Custom MongoDB state store
mongodb_db = db.MongoDBDB(
    url="mongodb://mongodb.example.com:27017",
    database="praisonai_production",
    # Collections for different data types
    state_collection="agent_states",
    session_collection="user_sessions",
    # Connection options
    max_pool_size=50,
    min_pool_size=5,
    max_idle_time_ms=30000,
    # SSL options
    ssl=True,
    ssl_cert_reqs="CERT_REQUIRED",
    ssl_ca_certs="/path/to/ca.pem"
)

agent = Agent(
    name="FlexibleBot", 
    instructions="You are a flexible assistant.",
    db=mongodb_db,
    session_id="flexible-session"
)

---

Docker Setup

Quick MongoDB setup with Docker:

# Start MongoDB container
docker run -d \
    --name praisonai-mongodb \
    -p 27017:27017 \
    -e MONGO_INITDB_ROOT_USERNAME=admin \
    -e MONGO_INITDB_ROOT_PASSWORD=admin_password \
    -e MONGO_INITDB_DATABASE=praisonai \
    mongo:7

# Connect and verify
docker exec -it praisonai-mongodb mongosh -u admin -p admin_password

With authentication setup:

# Create application user
docker exec -it praisonai-mongodb mongosh -u admin -p admin_password --eval "
use praisonai;
db.createUser({
  user: 'praisonai_user',
  pwd: 'user_password', 
  roles: [{ role: 'readWrite', db: 'praisonai' }]
});
"

Then use in your agent:

from praisonaiagents import Agent, db

agent = Agent(
    name="DockerBot",
    db=db(state_url="mongodb://praisonai_user:user_password@localhost:27017/praisonai"),
    session_id="docker-session"
)

---

Document Operations

Complex State Storage

import pymongo
from datetime import datetime
from praisonaiagents import Agent, db

# Direct MongoDB operations for complex state
client = pymongo.MongoClient("mongodb://localhost:27017/")
db_mongo = client["praisonai"]
collection = db_mongo["agent_states"]

# Store complex agent state
agent_state = {
    "agent_id": "complex_bot",
    "session_id": "complex-session",
    "user_profile": {
        "user_id": "user123",
        "preferences": {
            "response_style": "detailed",
            "topics": ["AI", "science", "programming"],
            "language": "en"
        },
        "interaction_history": [
            {"topic": "machine learning", "count": 15},
            {"topic": "programming", "count": 8},
            {"topic": "science", "count": 12}
        ]
    },
    "conversation_context": {
        "current_topic": "AI applications",
        "sentiment": "positive",
        "complexity_level": "advanced",
        "follow_up_questions": [
            "What about neural networks?",
            "How does this apply to real problems?"
        ]
    },
    "metrics": {
        "total_interactions": 35,
        "average_response_time": 1.2,
        "satisfaction_score": 4.8
    },
    "updated_at": datetime.utcnow()
}

# Insert or update state
collection.replace_one(
    {"agent_id": "complex_bot", "session_id": "complex-session"},
    agent_state,
    upsert=True
)

print("Complex state stored successfully")

Rich Queries and Aggregations

import pymongo
from praisonaiagents import Agent, db

client = pymongo.MongoClient("mongodb://localhost:27017/")
db_mongo = client["praisonai"]

# Create agent with MongoDB backend
agent = Agent(
    name="AnalyticsBot",
    db=db(state_url="mongodb://localhost:27017/praisonai"),
    session_id="analytics-session"
)

# Generate some data
agent.chat("I love discussing AI and machine learning")
agent.chat("Python is my favorite programming language")
agent.chat("Tell me about the latest in data science")

# Rich MongoDB queries
states_collection = db_mongo["agent_states"]

# Find users interested in specific topics
ai_enthusiasts = states_collection.find({
    "user_profile.preferences.topics": {"$in": ["AI", "machine learning"]}
})

print("AI enthusiasts:")
for user in ai_enthusiasts:
    print(f"- User {user.get('user_profile', {}).get('user_id')}")

# Aggregation pipeline for analytics
pipeline = [
    {"$match": {"metrics.total_interactions": {"$gte": 10}}},
    {"$group": {
        "_id": None,
        "avg_satisfaction": {"$avg": "$metrics.satisfaction_score"},
        "total_users": {"$sum": 1},
        "popular_topics": {"$push": "$user_profile.preferences.topics"}
    }},
    {"$unwind": "$popular_topics"},
    {"$unwind": "$popular_topics"},
    {"$group": {
        "_id": "$popular_topics",
        "count": {"$sum": 1},
        "avg_satisfaction": {"$first": "$avg_satisfaction"}
    }},
    {"$sort": {"count": -1}}
]

analytics = list(states_collection.aggregate(pipeline))
print("Popular topics:", analytics)

Time-Series Data

import pymongo
from datetime import datetime, timedelta
from praisonaiagents import Agent, db

client = pymongo.MongoClient("mongodb://localhost:27017/")
db_mongo = client["praisonai"]
events_collection = db_mongo["agent_events"]

# Store time-series events
def log_agent_event(agent_id, event_type, data):
    event = {
        "agent_id": agent_id,
        "event_type": event_type,
        "timestamp": datetime.utcnow(),
        "data": data
    }
    events_collection.insert_one(event)

# Create time-series index for performance
events_collection.create_index([("agent_id", 1), ("timestamp", 1)])

# Log various events
log_agent_event("time_bot", "conversation_start", {"session_id": "ts-session"})
log_agent_event("time_bot", "message_processed", {"tokens": 150, "response_time": 0.8})
log_agent_event("time_bot", "user_satisfaction", {"rating": 5})

# Query recent events
last_hour = datetime.utcnow() - timedelta(hours=1)
recent_events = events_collection.find({
    "agent_id": "time_bot",
    "timestamp": {"$gte": last_hour}
}).sort("timestamp", -1)

print("Recent events:")
for event in recent_events:
    print(f"- {event['event_type']} at {event['timestamp']}")

# Time-based aggregation
daily_stats = events_collection.aggregate([
    {"$match": {"agent_id": "time_bot"}},
    {"$group": {
        "_id": {
            "date": {"$dateToString": {"format": "%Y-%m-%d", "date": "$timestamp"}},
            "event_type": "$event_type"
        },
        "count": {"$sum": 1}
    }},
    {"$sort": {"_id.date": 1}}
])

for stat in daily_stats:
    print(f"Date: {stat['_id']['date']}, Event: {stat['_id']['event_type']}, Count: {stat['count']}")

---

Advanced Features

Schema Validation

import pymongo
from praisonaiagents import Agent, db

client = pymongo.MongoClient("mongodb://localhost:27017/")
db_mongo = client["praisonai"]

# Create collection with schema validation
agent_states_schema = {
    "$jsonSchema": {
        "bsonType": "object",
        "required": ["agent_id", "session_id", "updated_at"],
        "properties": {
            "agent_id": {"bsonType": "string"},
            "session_id": {"bsonType": "string"},
            "user_profile": {
                "bsonType": "object",
                "properties": {
                    "user_id": {"bsonType": "string"},
                    "preferences": {"bsonType": "object"}
                }
            },
            "metrics": {
                "bsonType": "object",
                "properties": {
                    "total_interactions": {"bsonType": "int", "minimum": 0},
                    "satisfaction_score": {"bsonType": "double", "minimum": 1, "maximum": 5}
                }
            },
            "updated_at": {"bsonType": "date"}
        }
    }
}

db_mongo.create_collection("agent_states_validated", validator=agent_states_schema)
print("Collection with schema validation created")

Full-Text Search

import pymongo
from praisonaiagents import Agent, db

client = pymongo.MongoClient("mongodb://localhost:27017/")
db_mongo = client["praisonai"]
conversations = db_mongo["conversations"]

# Create text index for full-text search
conversations.create_index([
    ("content", "text"),
    ("metadata.topic", "text")
])

# Store conversations with searchable content
conversations.insert_many([
    {
        "session_id": "search-session-1",
        "content": "I love artificial intelligence and machine learning",
        "metadata": {"topic": "AI", "sentiment": "positive"}
    },
    {
        "session_id": "search-session-2", 
        "content": "Python programming is great for data science",
        "metadata": {"topic": "programming", "sentiment": "positive"}
    }
])

# Full-text search
search_results = conversations.find(
    {"$text": {"$search": "artificial intelligence"}},
    {"score": {"$meta": "textScore"}}
).sort([("score", {"$meta": "textScore"})])

print("Search results for 'artificial intelligence':")
for result in search_results:
    print(f"- {result['content']} (score: {result['score']})")

---

Best Practices

Document Design

• Design documents around query patterns, not relational structures
• Use embedded documents for data that’s always accessed together
• Keep frequently updated fields separate from static data
• Use array fields for lists, but avoid unbounded growth

Indexing Strategy

• Create compound indexes for multi-field queries
• Use sparse indexes for optional fields
• Monitor index usage with explain() and profiling
• Consider partial indexes for filtered queries

Performance Optimization

• Use projection to limit returned fields
• Batch operations when possible with bulk operations
• Configure appropriate read/write concerns
• Use MongoDB Atlas for managed scaling and optimization

Data Management

• Implement TTL indexes for automatic document expiration
• Use change streams for real-time data synchronization
• Regular backup with mongodump or Atlas backups
• Monitor collection sizes and implement data archiving

---

Related

Redis State Store

Alternative fast in-memory state storage

Database Persistence Overview

Compare all available persistence backends