dory-ai 0.1.0

A library for managing conversation history in AI-powered applications for reusability across projects.

Dory

AI Memory & Conversation Management Library

A library for managing conversation history and memory in AI-powered applications, designed for reusability across projects.

Overview

Dory provides two core services for AI applications:

Messages Service

Simple, reliable conversation and message management with:

• Automatic Conversation Management: Reuses conversations within a 2-week window
• Message Persistence: Stores user messages and AI responses
• LangChain/LangGraph Integration: Returns chat history in the required format
• MongoDB Support: Production-ready persistence

Embeddings Service

Advanced memory and vector search capabilities with:

• Semantic Memory Storage: Store and retrieve contextual memories
• Vector Search: Find relevant information using similarity search
• Raw Embeddings: Store and search unprocessed content for retrieval
• Multiple Backends: Support for Chroma (local) and MongoDB Atlas
• Powered by Mem0: Built on top of the robust Mem0 library

Installation

Using uv (Recommended)

# Add to an existing project
uv add dory

# Or add to pyproject.toml dependencies
# Then run:
uv sync

Using pip

pip install dory

Add to pyproject.toml

[project]
dependencies = [
    "dory>=0.1.0",
    # ... other dependencies
]

Quick Start

Messages Service

import asyncio
from dory.messages import Messages
from dory.messages.adapters.mongo import MongoDBAdapter
from dory.common import MessageType, ChatRole


async def messages_example():
    # Initialize with MongoDB
    adapter = MongoDBAdapter(
        connection_string="mongodb://localhost:27017/myapp",
        database="myapp",
    )

    # Create Messages service
    messages = Messages(adapter=adapter)

    # Get or create a conversation (reuses if within 2 weeks)
    conversation = await messages.get_or_create_conversation(user_id="user_123")

    # Add a user message
    await messages.add_message(
        conversation_id=conversation.id,
        user_id="user_123",
        chat_role=ChatRole.USER,
        content="What's the weather like?",
        message_type=MessageType.USER_MESSAGE
    )

    # Add an AI response
    await messages.add_message(
        conversation_id=conversation.id,
        user_id="user_123",
        chat_role=ChatRole.AI,
        content="It's sunny today!",
        message_type=MessageType.REQUEST_RESPONSE
    )

    # Get chat history for LangChain/LangGraph
    chat_history = await messages.get_chat_history(
        conversation_id=conversation.id,
        limit=30
    )
    # Returns: [{"user": "What's the weather like?"}, {"ai": "It's sunny today!"}]


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

Embeddings Service

import asyncio
from dory.embeddings import build_embeddings


async def embeddings_example():
    # Initialize with Chroma (local vector store)
    embeddings = build_embeddings(
        api_key="your-openai-api-key",  # Required for OpenAI embeddings
        store="chroma",
        store_path="./chroma_db",
        collection="my_memories"
    )

    # Store contextual memories
    memory_id = await embeddings.remember(
        content="User prefers Python over Java",
        user_id="user_123",
        conversation_id="conv_abc",
        metadata={"topic": "preferences"}
    )

    # Search for relevant memories
    memories = await embeddings.recall(
        query="What programming languages does the user like?",
        user_id="user_123",
        limit=5
    )
    # Returns memories with relevance scores

    # Store raw embeddings for retrieval
    embedding_id = await embeddings.store_embedding(
        content="Python is a high-level programming language",
        user_id="user_123",
        metadata={"source": "documentation"}
    )

    # Search embeddings by similarity
    results = await embeddings.search_embeddings(
        query="Tell me about Python",
        user_id="user_123",
        limit=3
    )


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

API Reference

Messages Service API

# Initialize Messages with adapter
adapter = MongoDBAdapter(connection_string="...")
messages = Messages(adapter=adapter)

# Messages methods (all require keyword arguments)
async def get_or_create_conversation(self, *, user_id: str) -> Conversation:
    """Get recent conversation or create new one (2-week reuse window)."""

async def add_message(
    self,
    *,
    conversation_id: str | None = None,
    message_id: str | None = None,
    user_id: str,
    chat_role: ChatRole,
    content: Any,
    message_type: MessageType,
) -> Message:
    """Add a message. If conversation_id is None, a new conversation is created.
    If message_id is None, an ID is auto-generated.
    """

async def get_chat_history(
    self,
    *,
    conversation_id: str,
    limit: int | None = None
) -> list[dict[str, Any]]:
    """Get chat history in LangChain/LangGraph format."""

Message Types

class MessageType(str, Enum):
    USER_MESSAGE = "user_message"              # User input
    REQUEST_RESPONSE = "request_response"        # Final AI response

Optional IDs

Both conversation_id and message_id can be provided. If omitted:

• conversation_id: a new conversation is created for the given user_id
• message_id: an ID is generated using the configured prefix

Chat Roles

class ChatRole(str, Enum):
    USER = "user"
    HUMAN = "human"
    AI = "ai"

Models

class Conversation:
    id: str                # Format: "CONV_<uuid>"
    user_id: str
    created_at: datetime
    updated_at: datetime

class Message:
    id: str                # Format: "MSG_<uuid>"
    conversation_id: str
    user_id: str
    chat_role: ChatRole
    content: Any           # String or dict
    message_type: MessageType
    created_at: datetime

Embeddings Service API

# Initialize with builder function
embeddings = build_embeddings(
    api_key="openai-api-key",           # Optional: for OpenAI embeddings
    store="chroma",                      # Options: "chroma", "mongodb", "memory"
    store_path="./chroma_db",           # For local stores
    connection_string="mongodb://...",  # For MongoDB Atlas
    collection="memories"                # Collection/index name
)

# Embeddings methods (all async, require keyword arguments)
async def remember(
    self,
    *,
    content: str,
    user_id: str,
    conversation_id: str | None = None,
    metadata: dict[str, Any] | None = None
) -> str:
    """Store a memory with LLM processing for context extraction."""

async def recall(
    self,
    *,
    query: str,
    user_id: str,
    conversation_id: str | None = None,
    limit: int = 10
) -> list[dict[str, Any]]:
    """Search memories using semantic similarity."""

async def forget(
    self,
    *,
    user_id: str,
    conversation_id: str | None = None,
    memory_ids: list[str] | None = None
) -> int:
    """Delete memories and return count deleted."""

async def store_embedding(
    self,
    *,
    content: str,
    user_id: str,
    conversation_id: str | None = None,
    message_id: str | None = None,
    metadata: dict[str, Any] | None = None
) -> str:
    """Store raw content without LLM processing."""

async def search_embeddings(
    self,
    *,
    query: str,
    user_id: str,
    conversation_id: str | None = None,
    limit: int = 10
) -> list[dict[str, Any]]:
    """Search raw embeddings using vector similarity."""

Configuration

MongoDB Setup

# For Messages
adapter = MongoDBAdapter(
    connection_string="mongodb://localhost:27017",
    database="myapp",
)

# For Embeddings (MongoDB Atlas with Vector Search)
embeddings = build_embeddings(
    api_key="openai-api-key",
    store="mongodb",
    connection_string="mongodb+srv://...",
    collection="memories"
)

Indexes Created:

• Conversations: user_id, updated_at
• Messages: conversation_id, created_at, compound index
• Embeddings: Requires MongoDB Atlas Vector Search index

Chroma Setup

embeddings = build_embeddings(
    api_key="openai-api-key",
    store="chroma",
    store_path="./chroma_db",  # Local directory
    collection="my_memories"
)

In-Memory Setup (Testing)

# For testing - no persistence
embeddings = build_embeddings(
    store="memory",
    collection="test_memories"
)

Testing

# Run all tests
uv run pytest

# Run with coverage
uv run pytest --cov

# Run specific service tests
uv run pytest tests/messages/
uv run pytest tests/embeddings/

License

MIT