bruno-memory 0.1.5

Memory storage and retrieval system for Bruno AI Platform

Bruno Memory

Memory storage and retrieval system for Bruno AI Platform

Dependencies

pip install git+https://github.com/meggy-ai/bruno-core.git
pip install git+https://github.com/meggy-ai/bruno-llm.git

Bruno Memory provides multiple backend storage options for conversation history, user context, and semantic memory while maintaining a consistent API for AI assistants to interact with. It implements the MemoryInterface from bruno-core with full embedding support via bruno-llm.

📚 Documentation

View Full Documentation | Quick Start | API Reference | Examples

✨ Features

• Multiple Storage Backends: SQLite, PostgreSQL, Redis, ChromaDB, Qdrant
• Semantic Search: Vector-based similarity search using embeddings
• Session Management: Track conversations across multiple sessions
• Memory Compression: Automatic summarization of old conversations
• Context Building: Intelligent context window management
• Async/Await Support: Full async support for high performance
• Type Safety: Complete type hints and Pydantic models
• Easy Configuration: Environment-based configuration

🚀 Quick Start

Installation

# Basic installation
pip install bruno-memory

# With SQLite support (default)
pip install bruno-memory[sqlite]

# With PostgreSQL support  
pip install bruno-memory[postgresql]

# With Redis caching
pip install bruno-memory[redis]

# With vector database support
pip install bruno-memory[vector]

# With all backends
pip install bruno-memory[all]

Basic Usage

import asyncio
from bruno_memory import MemoryFactory
from bruno_core.models.message import Message, MessageRole

async def main():
    # Create memory backend
    memory = MemoryFactory.create("sqlite", database_path="./conversations.db")
    
    # Store a message
    message = Message(
        role=MessageRole.USER,
        content="Hello, how are you today?"
    )
    await memory.store_message(message, conversation_id="conv_1")
    
    # Retrieve messages
    messages = await memory.retrieve_messages("conv_1", limit=10)
    print(f"Found {len(messages)} messages")
    
    # Search messages semantically (requires embedding provider)
    results = await memory.search_messages("greeting", limit=5)
    
    # Get conversation context
    context = await memory.get_context(user_id="user_123")
    print(f"Context has {len(context.messages)} messages")

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

🏗️ Architecture

Bruno Memory is built with a modular architecture:

bruno-memory/
├── backends/          # Storage backend implementations
│   ├── sqlite/        # SQLite backend (file-based)
│   ├── postgresql/    # PostgreSQL backend (production)
│   ├── redis/         # Redis backend (caching)
│   └── vector/        # Vector DB backends (ChromaDB, Qdrant)
├── managers/          # Memory management components
│   ├── conversation.py    # Session lifecycle management
│   ├── context_builder.py # Context window construction
│   ├── retriever.py       # Search and retrieval
│   ├── compressor.py      # Memory compression
│   └── embedding.py       # Embedding management
└── utils/             # Utility components
    ├── cache.py           # Caching layer
    ├── migration_manager.py # Schema migrations
    └── analytics.py       # Usage analytics

📊 Backend Comparison

Backend	Use Case	Performance	Features	Setup Complexity
SQLite	Development, Single-user	Good	Basic search, FTS	⭐ Easy
PostgreSQL	Production, Multi-user	Excellent	Advanced search, JSON	⭐⭐ Medium
Redis	Caching, Sessions	Excellent	TTL, Pub/Sub	⭐⭐ Medium
ChromaDB	Semantic search	Good	Vector similarity	⭐⭐⭐ Advanced
Qdrant	Production vectors	Excellent	Hybrid search	⭐⭐⭐ Advanced

🔧 Configuration

Environment Variables

# Backend selection
BRUNO_MEMORY_BACKEND=postgresql
BRUNO_MEMORY_DATABASE_URL=postgresql://user:pass@localhost:5432/bruno

# Embedding provider (from bruno-llm)
BRUNO_LLM_EMBEDDING_PROVIDER=ollama
BRUNO_LLM_EMBEDDING_BASE_URL=http://localhost:11434

# Memory settings
BRUNO_MEMORY_MAX_MESSAGES=1000
BRUNO_MEMORY_COMPRESS_AFTER_DAYS=30

Programmatic Configuration

from bruno_memory import MemoryFactory

# SQLite backend
memory = MemoryFactory.create("sqlite", {
    "database_path": "./conversations.db",
    "enable_fts": True,
    "max_connections": 10
})

# PostgreSQL backend
memory = MemoryFactory.create("postgresql", {
    "host": "localhost",
    "port": 5432,
    "database": "bruno",
    "username": "postgres",
    "password": "secret",
    "pool_size": 20
})

# Redis backend
memory = MemoryFactory.create("redis", {
    "host": "localhost", 
    "port": 6379,
    "db": 0,
    "ttl_seconds": 3600
})

# ChromaDB backend
memory = MemoryFactory.create("chromadb", {
    "persist_directory": "./chroma_db",
    "collection_name": "conversations"
})

🔍 Semantic Search

Bruno Memory integrates with bruno-llm for embedding-based semantic search:

from bruno_memory import MemoryFactory
from bruno_llm.embedding_factory import EmbeddingFactory

# Set up embedding provider
embedding_provider = EmbeddingFactory.create("ollama", {
    "base_url": "http://localhost:11434",
    "model": "nomic-embed-text"
})

# Create memory with embedding support
memory = MemoryFactory.create("chromadb", {
    "persist_directory": "./vectors",
    "embedding_provider": embedding_provider
})

# Search semantically
results = await memory.search_messages(
    query="What did we discuss about machine learning?",
    limit=5
)

📈 Performance

Bruno Memory is designed for high performance:

• Async/Await: All operations are async for maximum concurrency
• Connection Pooling: Efficient database connection management
• Caching: Multi-level caching with Redis support
• Batch Operations: Bulk operations for large datasets
• Streaming: Stream large result sets to avoid memory issues

Benchmarks

Operation	SQLite	PostgreSQL	Redis	ChromaDB
Store Message	1ms	2ms	0.5ms	5ms
Retrieve 100 Messages	10ms	8ms	2ms	15ms
Search (semantic)	N/A	50ms	N/A	20ms
Context Building	15ms	12ms	5ms	25ms

🧪 Testing

Quick Testing

# Install with dev dependencies
pip install bruno-memory[dev]

# Run tests
pytest

# Run with coverage
pytest --cov=bruno_memory --cov-report=html

Docker-Based Testing

Bruno Memory includes comprehensive Docker-based testing infrastructure for all backends:

# Quick start - run all tests with Docker services
make test-docker

# Test specific backends
make test-postgresql  # PostgreSQL backend only
make test-redis       # Redis backend only
make test-chromadb    # ChromaDB vector backend
make test-qdrant      # Qdrant vector backend
make test-vector      # All vector backends

# Use minimal profile (PostgreSQL + Redis only)
make test-docker-minimal

# Keep environment running for iterative development
./scripts/run-tests-docker.ps1 -KeepEnv

Manual Service Management

# Start all services
make docker-up

# Check service status
make docker-status

# View logs
make docker-logs

# Stop services (preserve data)
make docker-down

# Complete cleanup (remove data)
make docker-teardown

Cross-Platform Scripts

All scripts are available in both PowerShell (Windows) and Bash (Linux/Mac):

# PowerShell (Windows)
.\scripts\setup-test-env.ps1 -Profile full
.\scripts\run-tests-docker.ps1 -Verbose
.\scripts\teardown-test-env.ps1 -Volumes

# Bash (Linux/Mac)
bash scripts/setup-test-env.sh --profile full
bash scripts/run-tests-docker.sh --verbose
bash scripts/teardown-test-env.sh --volumes

📖 Full Docker testing documentation: DOCKER_TESTING_QUICKSTART.md

Test Profiles

• minimal: PostgreSQL + Redis (fastest)
• full: All services including ChromaDB and Qdrant
• ci: CI-optimized with tmpfs and no volumes

Service Ports

Service	Port	Use Case
PostgreSQL	5432	Relational storage with pgvector
Redis	6379	Caching and sessions
ChromaDB	8000	Vector similarity search
Qdrant	6333 (HTTP), 6334 (gRPC)	Production vector search

🔗 Integration

With Bruno Core

from bruno_core import BaseAssistant
from bruno_memory import MemoryFactory

class MyAssistant(BaseAssistant):
    def __init__(self):
        # Initialize memory backend
        memory = MemoryFactory.create_from_env()
        super().__init__(memory=memory)
    
    async def process_message(self, message):
        # Memory automatically handles storage
        response = await self.generate_response(message)
        return response

With Bruno LLM

from bruno_llm import LLMFactory
from bruno_llm.embedding_factory import EmbeddingFactory
from bruno_memory import MemoryFactory

# Create providers
llm = LLMFactory.create("ollama")
embeddings = EmbeddingFactory.create("ollama")

# Create memory with embedding support
memory = MemoryFactory.create("chromadb", {
    "embedding_provider": embeddings
})

# Use together
async def rag_search(query: str):
    # Find relevant memories
    memories = await memory.search_messages(query, limit=5)
    
    # Use in LLM prompt
    context = "\\n".join([m.content for m in memories])
    response = await llm.generate(
        f"Context: {context}\\n\\nQuestion: {query}"
    )
    return response

📚 Documentation

• Quick Start Guide
• Backend Configuration
• Context Management
• Semantic Search
• API Reference

🤝 Contributing

We welcome contributions! Please see CONTRIBUTING.md for guidelines.

Development Setup

# Clone the repository
git clone https://github.com/meggy-ai/bruno-memory.git
cd bruno-memory

# Create virtual environment
python -m venv .venv
source .venv/bin/activate  # On Windows: .venv\\Scripts\\activate

# Install in development mode
pip install -e .[dev,all]

# Install pre-commit hooks
pre-commit install

# Run tests
pytest

📄 License

This project is licensed under the MIT License - see the LICENSE file for details.

🔗 Related Projects

• bruno-core - Core interfaces and models
• bruno-llm - LLM providers and embeddings
• bruno-abilities - Function calling and tools
• bruno-pa - Personal assistant implementation

📞 Support

• GitHub Issues
• Discussions

---

Made with ❤️ by the meggy-ai team.