Conversation Model

Represents a complete AI conversation with messages and metadata.

Overview

The Conversation model is the primary data structure for representing AI conversations. It includes metadata (title, timestamps) and a collection of messages organized in a tree structure.

API Reference

Conversation

Bases: BaseModel

Immutable conversation structure with tree navigation (per FR-222, FR-227, FR-278).

This model represents a complete AI conversation with metadata and all messages. Messages form a tree structure via parent_id references, enabling branching dialogue paths and multi-turn interactions.

Immutability

This model is FROZEN - attempting to modify fields will raise ValidationError. Use .model_copy(update={...}) to create modified instances.

Tree Navigation

Conversations support tree navigation via helper methods: - get_root_messages(): Entry points for conversation traversal - get_message_by_id(): Fast lookup for specific messages - get_children(): Get direct replies to a message - get_thread(): Get message and all ancestors up to root - get_all_threads(): Get all root-to-leaf paths

Example

from datetime import datetime, UTC

messages = [
    Message(id="1", content="Hello", role="user", timestamp=datetime.now(UTC), parent_id=None),
    Message(id="2", content="Hi!", role="assistant", timestamp=datetime.now(UTC), parent_id="1"),
    Message(id="3", content="Alt response", role="assistant", timestamp=datetime.now(UTC), parent_id="1"),
]

conversation = Conversation(
    id="conv-001",
    title="Greeting",
    created_at=datetime.now(UTC),
    updated_at=datetime.now(UTC),
    messages=messages
)

# Tree navigation
roots = conversation.get_root_messages()  # [msg-1]
children = conversation.get_children("1")  # [msg-2, msg-3]
threads = conversation.get_all_threads()  # [[msg-1, msg-2], [msg-1, msg-3]]

Attributes:

Name	Type	Description
id	str	Unique conversation identifier (non-empty string)
title	str	Conversation title (non-empty, any UTF-8)
created_at	datetime	Conversation creation timestamp (timezone-aware UTC, REQUIRED)
updated_at	datetime | None	Last modification timestamp (timezone-aware UTC, None if never updated)
messages	list[Message]	All messages in conversation (non-empty list)
metadata	dict[str, Any]	Provider-specific fields (e.g., moderation_results, plugin_ids)

Computed Properties

updated_at_or_created: Returns updated_at if set, else created_at (never None) message_count: Number of messages in conversation

updated_at_or_created property

updated_at_or_created: datetime

Get the last update timestamp, falling back to created_at if not set.

This property ensures downstream code always has a valid "last modified" timestamp without needing to handle Optional[datetime]. If the conversation has never been updated (updated_at is None), returns created_at.

Returns:

Type	Description
datetime	Last update timestamp (updated_at if set, else created_at)

Example

# Conversation never updated
conv = Conversation(..., created_at=ts1, updated_at=None)
assert conv.updated_at_or_created == ts1

# Conversation updated
conv2 = Conversation(..., created_at=ts1, updated_at=ts2)
assert conv2.updated_at_or_created == ts2

Usage Notes

• Prefer this property over direct updated_at access for display/sorting
• Use direct updated_at field when you need to distinguish null vs. set
• Guaranteed non-null return value (mypy --strict compliant)

message_count property

message_count: int

Get the total number of messages in the conversation.

Returns:

Type	Description
int	Number of messages in the conversation

Example

conversation = Conversation(
    id="conv-001",
    title="Test",
    created_at=datetime.now(UTC),
    updated_at=datetime.now(UTC),
    messages=[msg1, msg2, msg3]
)
assert conversation.message_count == 3

Requirements

• FR-018: Metadata includes message count

validate_created_at_timezone_aware classmethod

validate_created_at_timezone_aware(v: datetime) -> datetime

Ensure created_at is timezone-aware and normalized to UTC.

Parameters:

Name	Type	Description	Default
v	datetime	Timestamp value to validate	required

Returns:

Type	Description
datetime	Timezone-aware datetime normalized to UTC

Raises:

Type	Description
ValueError	If timestamp is timezone-naive

Requirements

• FR-244: Timestamps must be timezone-aware
• FR-245: Timestamps normalized to UTC
• FR-246: Validation enforced at parse time

Source code in src/echomine/models/conversation.py

@field_validator("created_at")
@classmethod
def validate_created_at_timezone_aware(cls, v: datetime) -> datetime:
    """Ensure created_at is timezone-aware and normalized to UTC.

    Args:
        v: Timestamp value to validate

    Returns:
        Timezone-aware datetime normalized to UTC

    Raises:
        ValueError: If timestamp is timezone-naive

    Requirements:
        - FR-244: Timestamps must be timezone-aware
        - FR-245: Timestamps normalized to UTC
        - FR-246: Validation enforced at parse time
    """
    if v.tzinfo is None or v.tzinfo.utcoffset(v) is None:
        msg = f"created_at must be timezone-aware: {v}"
        raise ValueError(msg)
    return v.astimezone(UTC)  # Normalize to UTC

validate_updated_at_timezone_aware classmethod

validate_updated_at_timezone_aware(
    v: datetime | None, info: Any
) -> datetime | None

Ensure updated_at is timezone-aware and >= created_at (if provided).

Parameters:

Name	Type	Description	Default
v	datetime | None	updated_at value to validate (may be None)	required
info	Any	Field validation info containing created_at	required

Returns:

Type	Description
datetime | None	Timezone-aware datetime normalized to UTC, or None if not provided

Raises:

Type	Description
ValueError	If timestamp is timezone-naive or < created_at

Requirements

• FR-244: Timestamps must be timezone-aware (when provided)
• FR-245: Timestamps normalized to UTC
• FR-246: Validation enforced at parse time
• FR-273: updated_at must be >= created_at (when provided)

Source code in src/echomine/models/conversation.py

@field_validator("updated_at")
@classmethod
def validate_updated_at_timezone_aware(cls, v: datetime | None, info: Any) -> datetime | None:
    """Ensure updated_at is timezone-aware and >= created_at (if provided).

    Args:
        v: updated_at value to validate (may be None)
        info: Field validation info containing created_at

    Returns:
        Timezone-aware datetime normalized to UTC, or None if not provided

    Raises:
        ValueError: If timestamp is timezone-naive or < created_at

    Requirements:
        - FR-244: Timestamps must be timezone-aware (when provided)
        - FR-245: Timestamps normalized to UTC
        - FR-246: Validation enforced at parse time
        - FR-273: updated_at must be >= created_at (when provided)
    """
    # Handle None (optional field)
    if v is None:
        return None

    # Validate timezone-aware
    if v.tzinfo is None or v.tzinfo.utcoffset(v) is None:
        msg = f"updated_at must be timezone-aware: {v}"
        raise ValueError(msg)

    # Normalize to UTC
    v_utc = v.astimezone(UTC)

    # Validate updated_at >= created_at
    created_at = info.data.get("created_at")
    if created_at and v_utc < created_at:
        msg = f"updated_at ({v_utc}) must be >= created_at ({created_at})"
        raise ValueError(msg)

    return v_utc

get_message_by_id

get_message_by_id(message_id: str) -> Message | None

Find message by ID.

Parameters:

Name	Type	Description	Default
message_id	str	Message identifier to find	required

Returns:

Type	Description
Message | None	Message if found, None otherwise

Example

msg = conversation.get_message_by_id("msg-001")
if msg:
    print(msg.content)

Requirements

• FR-278: Support message lookup by ID

Source code in src/echomine/models/conversation.py

def get_message_by_id(self, message_id: str) -> Message | None:
    """Find message by ID.

    Args:
        message_id: Message identifier to find

    Returns:
        Message if found, None otherwise

    Example:
        ```python
        msg = conversation.get_message_by_id("msg-001")
        if msg:
            print(msg.content)
        ```

    Requirements:
        - FR-278: Support message lookup by ID
    """
    return next((m for m in self.messages if m.id == message_id), None)

get_root_messages

get_root_messages() -> list[Message]

Get all root messages (parent_id is None).

Root messages are entry points for conversation tree traversal.

Returns:

Type	Description
list[Message]	List of root messages (may be empty if malformed data)

Example

roots = conversation.get_root_messages()
for root in roots:
    print(f"Thread starting with: {root.content}")

Requirements

• FR-278: Support identifying conversation entry points

Source code in src/echomine/models/conversation.py

def get_root_messages(self) -> list[Message]:
    """Get all root messages (parent_id is None).

    Root messages are entry points for conversation tree traversal.

    Returns:
        List of root messages (may be empty if malformed data)

    Example:
        ```python
        roots = conversation.get_root_messages()
        for root in roots:
            print(f"Thread starting with: {root.content}")
        ```

    Requirements:
        - FR-278: Support identifying conversation entry points
    """
    return [m for m in self.messages if m.parent_id is None]

get_children

get_children(message_id: str) -> list[Message]

Get all direct children of a message.

Parameters:

Name	Type	Description	Default
message_id	str	Parent message identifier	required

Returns:

Type	Description
list[Message]	List of messages with parent_id == message_id (may be empty)

Example

children = conversation.get_children("msg-001")
if children:
    print(f"Found {len(children)} replies")

Requirements

• FR-278: Support tree navigation via parent-child relationships

Source code in src/echomine/models/conversation.py

def get_children(self, message_id: str) -> list[Message]:
    """Get all direct children of a message.

    Args:
        message_id: Parent message identifier

    Returns:
        List of messages with parent_id == message_id (may be empty)

    Example:
        ```python
        children = conversation.get_children("msg-001")
        if children:
            print(f"Found {len(children)} replies")
        ```

    Requirements:
        - FR-278: Support tree navigation via parent-child relationships
    """
    return [m for m in self.messages if m.parent_id == message_id]

get_thread

get_thread(message_id: str) -> list[Message]

Get message and all ancestors up to root.

Returns messages in chronological order (oldest first), representing the conversation path from root to the specified message.

Parameters:

Name	Type	Description	Default
message_id	str	Target message identifier	required

Returns:

Type	Description
list[Message]	List of messages from root to target (empty if message_id not found)

Example

thread = conversation.get_thread("msg-005")
for msg in thread:
    print(f"{msg.role}: {msg.content}")

Requirements

• FR-278: Support retrieving conversation context for a message

Source code in src/echomine/models/conversation.py

def get_thread(self, message_id: str) -> list[Message]:
    """Get message and all ancestors up to root.

    Returns messages in chronological order (oldest first), representing
    the conversation path from root to the specified message.

    Args:
        message_id: Target message identifier

    Returns:
        List of messages from root to target (empty if message_id not found)

    Example:
        ```python
        thread = conversation.get_thread("msg-005")
        for msg in thread:
            print(f"{msg.role}: {msg.content}")
        ```

    Requirements:
        - FR-278: Support retrieving conversation context for a message
    """
    thread: list[Message] = []
    current = self.get_message_by_id(message_id)

    while current:
        thread.insert(0, current)  # Prepend (oldest first)
        current = self.get_message_by_id(current.parent_id) if current.parent_id else None

    return thread

get_all_threads

get_all_threads() -> list[list[Message]]

Get all conversation threads (root-to-leaf paths).

Returns list of threads, where each thread is a list of messages from root to leaf in chronological order. Useful for understanding all conversation branches.

Returns:

Type	Description
list[list[Message]]	List of message threads (each thread is a list of messages)

Example

# Conversation structure:
#   msg-1 (user: "Hello")
#   ├── msg-2 (assistant: "Hi! How can I help?")
#   └── msg-3 (assistant: "Alternative response")
#
# Result: [[msg-1, msg-2], [msg-1, msg-3]]

threads = conversation.get_all_threads()
for i, thread in enumerate(threads):
    print(f"Thread {i+1}: {len(thread)} messages")

Requirements

• FR-278: Support comprehensive tree traversal
• FR-280: Enable analysis of all conversation branches

Source code in src/echomine/models/conversation.py

def get_all_threads(self) -> list[list[Message]]:
    """Get all conversation threads (root-to-leaf paths).

    Returns list of threads, where each thread is a list of messages
    from root to leaf in chronological order. Useful for understanding
    all conversation branches.

    Returns:
        List of message threads (each thread is a list of messages)

    Example:
        ```python
        # Conversation structure:
        #   msg-1 (user: "Hello")
        #   ├── msg-2 (assistant: "Hi! How can I help?")
        #   └── msg-3 (assistant: "Alternative response")
        #
        # Result: [[msg-1, msg-2], [msg-1, msg-3]]

        threads = conversation.get_all_threads()
        for i, thread in enumerate(threads):
            print(f"Thread {i+1}: {len(thread)} messages")
        ```

    Requirements:
        - FR-278: Support comprehensive tree traversal
        - FR-280: Enable analysis of all conversation branches
    """
    threads: list[list[Message]] = []

    def build_threads(msg: Message, path: list[Message]) -> None:
        """Recursive helper to build all threads from a message.

        Args:
            msg: Current message in traversal
            path: Accumulated messages from root to current
        """
        path = [*path, msg]  # Extend path with current message
        children = self.get_children(msg.id)

        if not children:
            # Leaf node: complete thread
            threads.append(path)
        else:
            # Branch node: recurse into children
            for child in children:
                build_threads(child, path)

    # Start traversal from each root
    for root in self.get_root_messages():
        build_threads(root, [])

    return threads

flatten_messages

flatten_messages() -> str

Flatten all message content to single string for search.

Concatenates all message content with space separation, useful for full-text search operations.

Returns:

Type	Description
str	Single string containing all message content

Example

text = conversation.flatten_messages()
if "algorithm" in text.lower():
    print("Conversation mentions algorithms")

Requirements

• FR-322: Enable full-text search across conversation content

Source code in src/echomine/models/conversation.py

def flatten_messages(self) -> str:
    """Flatten all message content to single string for search.

    Concatenates all message content with space separation, useful for
    full-text search operations.

    Returns:
        Single string containing all message content

    Example:
        ```python
        text = conversation.flatten_messages()
        if "algorithm" in text.lower():
            print("Conversation mentions algorithms")
        ```

    Requirements:
        - FR-322: Enable full-text search across conversation content
    """
    return " ".join(msg.content for msg in self.messages)

Usage Examples

Basic Access

from echomine import OpenAIAdapter
from pathlib import Path

adapter = OpenAIAdapter()
conversation = adapter.get_conversation_by_id(Path("export.json"), "conv-abc123")

# Access metadata
print(f"Title: {conversation.title}")
print(f"Created: {conversation.created_at}")
print(f"Messages: {len(conversation.messages)}")

# Access messages
for message in conversation.messages:
    print(f"{message.role}: {message.content[:50]}...")

Message Tree Navigation

Conversations can have branching message trees (e.g., regenerated AI responses):

# Get all threads (root-to-leaf paths)
threads = conversation.get_all_threads()
print(f"Conversation has {len(threads)} branches")

for i, thread in enumerate(threads, 1):
    print(f"\nThread {i} ({len(thread)} messages):")
    for msg in thread:
        print(f"  {msg.role}: {msg.content[:50]}...")

# Get specific thread by leaf message ID
thread = conversation.get_thread("msg-xyz-789")

# Get root messages (conversation starters)
roots = conversation.get_root_messages()

# Get children of a specific message
children = conversation.get_children("msg-abc-123")

# Check if message has children (branches)
has_branches = conversation.has_children("msg-abc-123")

Immutability

Models are frozen (immutable) to prevent accidental data corruption:

from pydantic import ValidationError

# ❌ This raises ValidationError (frozen model)
try:
    conversation.title = "New Title"
except ValidationError:
    print("Error: Cannot modify frozen model")

# ✅ Create modified copy instead
updated = conversation.model_copy(update={"title": "New Title"})
print(f"Original: {conversation.title}")
print(f"Updated: {updated.title}")

Validation

All fields are strictly validated:

from echomine.models import Conversation
from datetime import datetime, timezone
from pydantic import ValidationError

# ❌ Invalid: missing required fields
try:
    invalid = Conversation(
        id="conv-123",
        title="Test"
        # Missing: created_at, messages
    )
except ValidationError as e:
    print(f"Validation error: {e}")

# ✅ Valid: all required fields provided
valid = Conversation(
    id="conv-123",
    title="Test Conversation",
    created_at=datetime.now(timezone.utc),
    messages=[],
    metadata={}
)

Model Fields

Required Fields

• id (str): Unique conversation identifier
• title (str): Conversation title (1-2000 characters)
• created_at (datetime): Creation timestamp (UTC, timezone-aware)
• messages (list[Message]): List of conversation messages

Optional Fields

• updated_at (datetime | None): Last update timestamp (UTC, timezone-aware), None if never updated
• metadata (dict[str, Any]): Provider-specific metadata (default: empty dict)

Computed Properties

• message_count (int): Number of messages in conversation

Message Tree Structure

Conversations are organized as trees, not linear sequences:

• Each message has an optional parent_id pointing to its predecessor
• Messages with parent_id=None are root messages (conversation starters)
• Messages can have multiple children (branches from regenerated responses)

Example Tree:

Root (parent_id=None)
├── Child 1 (parent_id=Root)
│   └── Child 1.1 (parent_id=Child 1)
└── Child 2 (parent_id=Root)  # Branch from regeneration
    └── Child 2.1 (parent_id=Child 2)

Thread Extraction

Thread: A root-to-leaf path through the message tree.

# Get all threads (all possible conversation paths)
threads = conversation.get_all_threads()

# Each thread is a list of messages in chronological order
for thread in threads:
    for msg in thread:
        print(f"{msg.role}: {msg.content}")

Metadata

Provider-specific data is stored in the metadata dictionary:

# OpenAI-specific metadata
conversation.metadata.get("openai_model", "unknown")
conversation.metadata.get("openai_conversation_template_id")
conversation.metadata.get("openai_plugin_ids", [])

# Check if metadata exists
if "openai_model" in conversation.metadata:
    print(f"Model: {conversation.metadata['openai_model']}")

Related Models

• Message: Individual message model
• SearchResult: Search result containing a conversation

See Also

• Library Usage Guide
• OpenAI Adapter