Conversation Models Technical Documentation

Overview

Design Motivation and Problem Background

When building complex AI application systems, multi-participant conversation management faces the following core challenges:

1. Multi-Participant Conversation Complexity

Need to support multiple participant types such as users, master controllers, AI agents
Different participants have different roles and permissions
Lack of unified participant identity management mechanism

2. Conversation Session Isolation Requirements

Need to support different types of conversation sessions (user to master controller, master controller to agent, agent to agent, etc.)
Each session needs independent context and state management
Lack of session type validation and isolation mechanisms

3. Agent-to-Agent Communication Standardization

AI agents need standardized communication protocols
Need to support different types of messages (task assignment, result reporting, collaboration requests, etc.)
Lack of unified agent communication message format

4. Conversation Data Persistence

Conversation data needs to support serialization and deserialization
Need data formats compatible with storage systems
Lack of standardized data conversion mechanisms

Conversation Model System’s Solution:

Typed Participant Models: Type-safe participant definitions based on dataclass
Session Type Validation: Automatically validates session type matches participants
Standardized Message Format: Unified agent communication message structure
Data Conversion Support: Automatic serialization and deserialization mechanisms
Session Isolation Tools: Provides session key generation and validation tools

Component Positioning

conversation_models.py is a domain model component of the AIECS system, located in the Domain Layer, defining core data models related to conversation management. As the system’s data contract layer, it provides type-safe, structured conversation data models and utility functions.

Component Type and Positioning

Component Type

Domain Model Component - Located in the Domain Layer, belongs to data contract definitions

Architecture Layers

┌─────────────────────────────────────────┐
│         Application Layer               │  ← Components using conversation models
│  (ContextEngine, ServiceLayer)          │
└─────────────────┬───────────────────────┘
                  │
┌─────────────────▼───────────────────────┐
│         Domain Layer                    │  ← Conversation models layer
│  (ConversationModels, Data Contracts)   │
└─────────────────┬───────────────────────┘
                  │
┌─────────────────▼───────────────────────┐
│       Infrastructure Layer              │  ← Components conversation models depend on
│  (Storage, Serialization)               │
└─────────────────┬───────────────────────┘
                  │
┌─────────────────▼───────────────────────┐
│         External Systems                │  ← External storage systems
│  (Redis, Database, FileSystem)          │
└─────────────────────────────────────────┘

Upstream Components (Consumers)

1. Domain Services

ContextEngine (domain/context/context_engine.py)
SessionManager (if exists)
ConversationManager (if exists)

2. Application Layer Services

AgentService (if exists)
CommunicationService (if exists)
WorkflowService (if exists)

3. Infrastructure Layer

Storage Systems (via serialization interface)
Message Queues (via message format)
API Layer (via data conversion)

Downstream Components (Dependencies)

1. Python Standard Library

dataclasses - Provides dataclass support
typing - Provides type annotation support
datetime - Provides time handling
uuid - Provides unique identifier generation

2. Domain Models

TaskContext (if exists)
SessionMetrics (if exists)
Other Domain Models (via metadata fields)

3. Utility Functions

create_session_key - Session key generation utility
validate_conversation_isolation_pattern - Session isolation validation utility

Core Model Details

1. ConversationParticipant - Conversation Participant Model

@dataclass
class ConversationParticipant:
    """Represents a participant in a conversation"""
    participant_id: str
    participant_type: str  # 'user', 'master_controller', 'agent'
    participant_role: Optional[str] = None  # For agents: 'writer', 'researcher', etc.
    metadata: Dict[str, Any] = field(default_factory=dict)

Field Descriptions:

participant_id: Unique participant identifier
participant_type: Participant type (user, master controller, agent)
participant_role: Participant role (only valid for agents)
metadata: Participant metadata

Validation Rules:

def __post_init__(self):
    """Validate participant data after initialization"""
    if not self.participant_id:
        raise ValueError("participant_id cannot be empty")
    if not self.participant_type:
        raise ValueError("participant_type cannot be empty")
    
    # Validate participant type
    valid_types = {'user', 'master_controller', 'agent'}
    if self.participant_type not in valid_types:
        raise ValueError(f"participant_type must be one of {valid_types}")
    
    # For agents, role must be specified
    if self.participant_type == 'agent' and not self.participant_role:
        raise ValueError("participant_role is required for agent participants")

Usage Examples:

# Create user participant
user = ConversationParticipant(
    participant_id="user_123",
    participant_type="user",
    metadata={"name": "John Doe", "email": "john@example.com"}
)

# Create agent participant
agent = ConversationParticipant(
    participant_id="agent_001",
    participant_type="agent",
    participant_role="data_analyst",
    metadata={"capabilities": ["data_analysis", "visualization"]}
)

# Create master controller participant
controller = ConversationParticipant(
    participant_id="mc_001",
    participant_type="master_controller",
    metadata={"version": "1.0", "max_agents": 10}
)

2. ConversationSession - Conversation Session Model

@dataclass
class ConversationSession:
    """Represents an isolated conversation session between participants"""
    session_id: str
    participants: List[ConversationParticipant]
    session_type: str  # 'user_to_mc', 'mc_to_agent', 'agent_to_agent', 'user_to_agent'
    created_at: datetime
    last_activity: datetime
    metadata: Dict[str, Any] = field(default_factory=dict)

Field Descriptions:

session_id: Unique session identifier
participants: List of participants
session_type: Session type
created_at: Creation time
last_activity: Last activity time
metadata: Session metadata

Session Type Descriptions:

user_to_mc: User to master controller conversation
mc_to_agent: Master controller to agent conversation
agent_to_agent: Agent to agent conversation
user_to_agent: User to agent conversation

Validation Rules:

def _validate_participants_for_session_type(self):
    """Validate that participants match session type"""
    participant_types = [p.participant_type for p in self.participants]
    
    if self.session_type == 'user_to_mc':
        expected_types = {'user', 'master_controller'}
        if not expected_types.issubset(set(participant_types)):
            raise ValueError("user_to_mc session requires user and master_controller participants")
    
    elif self.session_type == 'mc_to_agent':
        expected_types = {'master_controller', 'agent'}
        if not expected_types.issubset(set(participant_types)):
            raise ValueError("mc_to_agent session requires master_controller and agent participants")
    
    elif self.session_type == 'agent_to_agent':
        agent_count = sum(1 for p in self.participants if p.participant_type == 'agent')
        if agent_count < 2:
            raise ValueError("agent_to_agent session requires at least 2 agent participants")
    
    elif self.session_type == 'user_to_agent':
        expected_types = {'user', 'agent'}
        if not expected_types.issubset(set(participant_types)):
            raise ValueError("user_to_agent session requires user and agent participants")

Core Methods:

Session Key Generation

def generate_session_key(self) -> str:
    """Generate unique session key for conversation isolation"""
    if self.session_type == 'user_to_mc':
        return self.session_id
    elif self.session_type == 'mc_to_agent':
        agent_role = next((p.participant_role for p in self.participants if p.participant_type == 'agent'), 'unknown')
        return f"{self.session_id}_mc_to_{agent_role}"
    elif self.session_type == 'agent_to_agent':
        agent_roles = [p.participant_role for p in self.participants if p.participant_type == 'agent']
        if len(agent_roles) >= 2:
            return f"{self.session_id}_{agent_roles[0]}_to_{agent_roles[1]}"
        return f"{self.session_id}_agent_to_agent"
    elif self.session_type == 'user_to_agent':
        agent_role = next((p.participant_role for p in self.participants if p.participant_type == 'agent'), 'unknown')
        return f"{self.session_id}_user_to_{agent_role}"
    else:
        return self.session_id

Participant Lookup

def get_participant_by_type_and_role(self, participant_type: str, participant_role: Optional[str] = None) -> Optional[ConversationParticipant]:
    """Get participant by type and role"""
    for participant in self.participants:
        if participant.participant_type == participant_type:
            if participant_role is None or participant.participant_role == participant_role:
                return participant
    return None

Activity Update

def update_activity(self):
    """Update last activity timestamp"""
    self.last_activity = datetime.utcnow()

Usage Examples:

# Create user to master controller session
user_mc_session = ConversationSession(
    session_id="session_001",
    participants=[
        ConversationParticipant("user_123", "user"),
        ConversationParticipant("mc_001", "master_controller")
    ],
    session_type="user_to_mc",
    created_at=datetime.utcnow(),
    last_activity=datetime.utcnow(),
    metadata={"project": "data_analysis", "priority": "high"}
)

# Create master controller to agent session
mc_agent_session = ConversationSession(
    session_id="session_002",
    participants=[
        ConversationParticipant("mc_001", "master_controller"),
        ConversationParticipant("agent_001", "agent", "data_analyst")
    ],
    session_type="mc_to_agent",
    created_at=datetime.utcnow(),
    last_activity=datetime.utcnow(),
    metadata={"task_type": "data_analysis", "deadline": "2024-01-01"}
)

# Generate session keys
session_key = user_mc_session.generate_session_key()  # "session_001"
agent_session_key = mc_agent_session.generate_session_key()  # "session_002_mc_to_data_analyst"

3. AgentCommunicationMessage - Agent Communication Message Model

@dataclass
class AgentCommunicationMessage:
    """Message for agent-to-agent or controller-to-agent communication"""
    message_id: str
    session_key: str
    sender_id: str
    sender_type: str  # 'master_controller', 'agent', 'user'
    sender_role: Optional[str]  # For agents
    recipient_id: str
    recipient_type: str  # 'agent', 'master_controller', 'user'
    recipient_role: Optional[str]  # For agents
    content: str
    message_type: str  # 'task_assignment', 'result_report', 'collaboration_request', 'feedback', 'communication'
    timestamp: datetime
    metadata: Dict[str, Any] = field(default_factory=dict)

Field Descriptions:

message_id: Unique message identifier
session_key: Session key
sender_id: Sender ID
sender_type: Sender type
sender_role: Sender role
recipient_id: Recipient ID
recipient_type: Recipient type
recipient_role: Recipient role
content: Message content
message_type: Message type
timestamp: Timestamp
metadata: Message metadata

Message Type Descriptions:

task_assignment: Task assignment
result_report: Result report
collaboration_request: Collaboration request
feedback: Feedback
communication: Communication
status_update: Status update
error_report: Error report
task_completion: Task completion
progress_update: Progress update
clarification_request: Clarification request

Validation Rules:

def __post_init__(self):
    """Validate message data after initialization"""
    if not self.message_id:
        self.message_id = str(uuid.uuid4())
    if not self.session_key:
        raise ValueError("session_key cannot be empty")
    if not self.sender_id:
        raise ValueError("sender_id cannot be empty")
    if not self.recipient_id:
        raise ValueError("recipient_id cannot be empty")
    if not self.content:
        raise ValueError("content cannot be empty")
    
    # Validate message type
    valid_message_types = {
        'task_assignment', 'result_report', 'collaboration_request',
        'feedback', 'communication', 'status_update', 'error_report',
        'task_completion', 'progress_update', 'clarification_request'
    }
    if self.message_type not in valid_message_types:
        raise ValueError(f"message_type must be one of {valid_message_types}")

Core Methods:

Convert to Conversation Message Format

def to_conversation_message_dict(self) -> Dict[str, Any]:
    """Convert to format compatible with ContextEngine conversation messages"""
    role = f"{self.sender_type}_{self.sender_role}" if self.sender_role else self.sender_type
    return {
        "role": role,
        "content": self.content,
        "timestamp": self.timestamp.isoformat(),
        "metadata": {
            **self.metadata,
            "message_id": self.message_id,
            "sender_id": self.sender_id,
            "sender_type": self.sender_type,
            "sender_role": self.sender_role,
            "recipient_id": self.recipient_id,
            "recipient_type": self.recipient_type,
            "recipient_role": self.recipient_role,
            "message_type": self.message_type
        }
    }

Usage Examples:

# Create task assignment message
task_message = AgentCommunicationMessage(
    message_id=str(uuid.uuid4()),
    session_key="session_002_mc_to_data_analyst",
    sender_id="mc_001",
    sender_type="master_controller",
    sender_role=None,
    recipient_id="agent_001",
    recipient_type="agent",
    recipient_role="data_analyst",
    content="Please analyze the sales data and generate a report",
    message_type="task_assignment",
    timestamp=datetime.utcnow(),
    metadata={"task_id": "task_001", "priority": "high", "deadline": "2024-01-01"}
)

# Create result report message
result_message = AgentCommunicationMessage(
    message_id=str(uuid.uuid4()),
    session_key="session_002_mc_to_data_analyst",
    sender_id="agent_001",
    sender_type="agent",
    sender_role="data_analyst",
    recipient_id="mc_001",
    recipient_type="master_controller",
    recipient_role=None,
    content="Analysis completed. Found 15% increase in sales compared to last quarter.",
    message_type="result_report",
    timestamp=datetime.utcnow(),
    metadata={"analysis_id": "analysis_001", "confidence": 0.95, "data_points": 1000}
)

# Convert to conversation message format
conversation_dict = task_message.to_conversation_message_dict()

Utility Functions Details

1. create_session_key - Session Key Generation Utility

def create_session_key(session_id: str, session_type: str, participants: List[ConversationParticipant]) -> str:
    """
    Utility function to create session key for conversation isolation
    
    Args:
        session_id: Base session ID
        session_type: Conversation type
        participants: List of conversation participants
    
    Returns:
        Generated session key
    """
    if session_type == 'user_to_mc':
        return session_id
    elif session_type == 'mc_to_agent':
        agent_role = next((p.participant_role for p in participants if p.participant_type == 'agent'), 'unknown')
        return f"{session_id}_mc_to_{agent_role}"
    elif session_type == 'agent_to_agent':
        agent_roles = [p.participant_role for p in participants if p.participant_type == 'agent']
        if len(agent_roles) >= 2:
            return f"{session_id}_{agent_roles[0]}_to_{agent_roles[1]}"
        return f"{session_id}_agent_to_agent"
    elif session_type == 'user_to_agent':
        agent_role = next((p.participant_role for p in participants if p.participant_type == 'agent'), 'unknown')
        return f"{session_id}_user_to_{agent_role}"
    else:
        return session_id

Usage Examples:

# Create participant list
participants = [
    ConversationParticipant("user_123", "user"),
    ConversationParticipant("agent_001", "agent", "data_analyst")
]

# Generate session key
session_key = create_session_key("session_001", "user_to_agent", participants)
# Result: "session_001_user_to_data_analyst"

2. validate_conversation_isolation_pattern - Session Isolation Validation Utility

def validate_conversation_isolation_pattern(session_key: str, expected_pattern: str) -> bool:
    """
    Validate that session key follows expected conversation isolation pattern
    
    Args:
        session_key: Session key to validate
        expected_pattern: Expected pattern ('user_to_mc', 'mc_to_agent', etc.)
    
    Returns:
        Returns True if pattern matches, False otherwise
    """
    if expected_pattern == 'user_to_mc':
        # Should just be base session_id
        return '_' not in session_key or not any(x in session_key for x in ['_mc_to_', '_to_', '_user_to_'])
    elif expected_pattern == 'mc_to_agent':
        return '_mc_to_' in session_key
    elif expected_pattern == 'agent_to_agent':
        return '_to_' in session_key and '_mc_to_' not in session_key and '_user_to_' not in session_key
    elif expected_pattern == 'user_to_agent':
        return '_user_to_' in session_key
    else:
        return False

Usage Examples:

# Validate session key pattern
session_key = "session_001_user_to_data_analyst"
is_valid = validate_conversation_isolation_pattern(session_key, "user_to_agent")
# Result: True

session_key = "session_002_mc_to_writer"
is_valid = validate_conversation_isolation_pattern(session_key, "mc_to_agent")
# Result: True

Design Patterns Explained

1. Dataclass Pattern

@dataclass
class ConversationParticipant:
    """Uses dataclass decorator to automatically generate methods"""
    participant_id: str
    participant_type: str
    participant_role: Optional[str] = None
    metadata: Dict[str, Any] = field(default_factory=dict)

Advantages:

Automatic Method Generation: Automatically generates __init__, __repr__, __eq__, etc.
Type Safety: Type checking based on type annotations
Immutability: Supports immutable dataclasses
Serialization Support: Compatible with asdict() and from_dict()

2. Validation Pattern

def __post_init__(self):
    """Execute validation after initialization"""
    if not self.participant_id:
        raise ValueError("participant_id cannot be empty")
    # More validation logic...

Advantages:

Data Integrity: Ensures data is valid at creation time
Early Error Detection: Detects errors before data is used
Type Safety: Provides stronger type safety combined with type annotations

3. Factory Pattern

@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'ConversationSession':
    """Factory method to create instance from dictionary"""
    participants = [
        ConversationParticipant(
            participant_id=p["participant_id"],
            participant_type=p["participant_type"],
            participant_role=p.get("participant_role"),
            metadata=p.get("metadata", {})
        )
        for p in data["participants"]
    ]
    
    return cls(
        session_id=data["session_id"],
        participants=participants,
        session_type=data["session_type"],
        created_at=datetime.fromisoformat(data["created_at"]),
        last_activity=datetime.fromisoformat(data["last_activity"]),
        metadata=data.get("metadata", {})
    )

Advantages:

Flexible Creation: Supports creating objects from different data sources
Data Conversion: Automatically handles data format conversion
Type Safety: Ensures created objects are of correct type

Usage Examples

1. Basic Participant Management

from aiecs.domain.context.conversation_models import ConversationParticipant

# Create different types of participants
user = ConversationParticipant(
    participant_id="user_123",
    participant_type="user",
    metadata={"name": "John Doe", "email": "john@example.com"}
)

agent = ConversationParticipant(
    participant_id="agent_001",
    participant_type="agent",
    participant_role="data_analyst",
    metadata={"capabilities": ["data_analysis", "visualization"], "version": "1.0"}
)

controller = ConversationParticipant(
    participant_id="mc_001",
    participant_type="master_controller",
    metadata={"version": "1.0", "max_agents": 10}
)

print(f"User: {user.participant_id} - {user.participant_type}")
print(f"Agent: {agent.participant_id} - {agent.participant_type} - {agent.participant_role}")
print(f"Controller: {controller.participant_id} - {controller.participant_type}")

2. Conversation Session Management

from aiecs.domain.context.conversation_models import ConversationSession, ConversationParticipant
from datetime import datetime

# Create user to master controller session
user_mc_session = ConversationSession(
    session_id="session_001",
    participants=[
        ConversationParticipant("user_123", "user"),
        ConversationParticipant("mc_001", "master_controller")
    ],
    session_type="user_to_mc",
    created_at=datetime.utcnow(),
    last_activity=datetime.utcnow(),
    metadata={"project": "data_analysis", "priority": "high"}
)

# Create master controller to agent session
mc_agent_session = ConversationSession(
    session_id="session_002",
    participants=[
        ConversationParticipant("mc_001", "master_controller"),
        ConversationParticipant("agent_001", "agent", "data_analyst")
    ],
    session_type="mc_to_agent",
    created_at=datetime.utcnow(),
    last_activity=datetime.utcnow(),
    metadata={"task_type": "data_analysis", "deadline": "2024-01-01"}
)

# Generate session keys
user_mc_key = user_mc_session.generate_session_key()  # "session_001"
mc_agent_key = mc_agent_session.generate_session_key()  # "session_002_mc_to_data_analyst"

print(f"User-MC Session Key: {user_mc_key}")
print(f"MC-Agent Session Key: {mc_agent_key}")

# Find participants
user_participant = user_mc_session.get_participant_by_type_and_role("user")
agent_participant = mc_agent_session.get_participant_by_type_and_role("agent", "data_analyst")

print(f"User participant: {user_participant.participant_id}")
print(f"Agent participant: {agent_participant.participant_id} - {agent_participant.participant_role}")

3. Agent Communication Messages

from aiecs.domain.context.conversation_models import AgentCommunicationMessage
import uuid

# Create task assignment message
task_message = AgentCommunicationMessage(
    message_id=str(uuid.uuid4()),
    session_key="session_002_mc_to_data_analyst",
    sender_id="mc_001",
    sender_type="master_controller",
    sender_role=None,
    recipient_id="agent_001",
    recipient_type="agent",
    recipient_role="data_analyst",
    content="Please analyze the sales data and generate a report",
    message_type="task_assignment",
    timestamp=datetime.utcnow(),
    metadata={"task_id": "task_001", "priority": "high", "deadline": "2024-01-01"}
)

# Create result report message
result_message = AgentCommunicationMessage(
    message_id=str(uuid.uuid4()),
    session_key="session_002_mc_to_data_analyst",
    sender_id="agent_001",
    sender_type="agent",
    sender_role="data_analyst",
    recipient_id="mc_001",
    recipient_type="master_controller",
    recipient_role=None,
    content="Analysis completed. Found 15% increase in sales compared to last quarter.",
    message_type="result_report",
    timestamp=datetime.utcnow(),
    metadata={"analysis_id": "analysis_001", "confidence": 0.95, "data_points": 1000}
)

# Convert to conversation message format
conversation_dict = task_message.to_conversation_message_dict()
print(f"Conversation format: {conversation_dict}")

# Convert to dictionary format
message_dict = result_message.to_dict()
print(f"Message dict: {message_dict}")

4. Utility Function Usage

from aiecs.domain.context.conversation_models import (
    create_session_key, 
    validate_conversation_isolation_pattern,
    ConversationParticipant
)

# Create participant list
participants = [
    ConversationParticipant("user_123", "user"),
    ConversationParticipant("agent_001", "agent", "data_analyst")
]

# Generate session key
session_key = create_session_key("session_001", "user_to_agent", participants)
print(f"Generated session key: {session_key}")  # "session_001_user_to_data_analyst"

# Validate session key pattern
is_valid = validate_conversation_isolation_pattern(session_key, "user_to_agent")
print(f"Session key is valid: {is_valid}")  # True

# Validate wrong pattern
is_valid = validate_conversation_isolation_pattern(session_key, "mc_to_agent")
print(f"Session key matches mc_to_agent: {is_valid}")  # False

5. Data Serialization and Deserialization

# Serialize session
session_dict = user_mc_session.to_dict()
print(f"Session dict: {session_dict}")

# Deserialize session
restored_session = ConversationSession.from_dict(session_dict)
print(f"Restored session: {restored_session.session_id}")

# Serialize message
message_dict = task_message.to_dict()
print(f"Message dict: {message_dict}")

# Deserialize message
restored_message = AgentCommunicationMessage.from_dict(message_dict)
print(f"Restored message: {restored_message.message_id}")

Maintenance Guide

1. Daily Maintenance

Model Validation

def validate_models_health():
    """Validate model health status"""
    try:
        # Test participant creation
        participant = ConversationParticipant(
            participant_id="test_001",
            participant_type="agent",
            participant_role="test_role"
        )
        print("✅ ConversationParticipant validation passed")
        
        # Test session creation
        session = ConversationSession(
            session_id="test_session",
            participants=[participant],
            session_type="user_to_agent",
            created_at=datetime.utcnow(),
            last_activity=datetime.utcnow()
        )
        print("✅ ConversationSession validation passed")
        
        # Test message creation
        message = AgentCommunicationMessage(
            message_id=str(uuid.uuid4()),
            session_key="test_key",
            sender_id="sender_001",
            sender_type="agent",
            sender_role="test_role",
            recipient_id="recipient_001",
            recipient_type="agent",
            recipient_role="test_role",
            content="Test message",
            message_type="communication",
            timestamp=datetime.utcnow()
        )
        print("✅ AgentCommunicationMessage validation passed")
        
        return True
        
    except Exception as e:
        print(f"❌ Model validation failed: {e}")
        return False

Data Consistency Check

def check_data_consistency(session: ConversationSession):
    """Check session data consistency"""
    try:
        # Check session key generation
        generated_key = session.generate_session_key()
        expected_key = create_session_key(
            session.session_id, 
            session.session_type, 
            session.participants
        )
        
        if generated_key != expected_key:
            print(f"❌ Session key mismatch: {generated_key} vs {expected_key}")
            return False
        
        # Check participant validation
        for participant in session.participants:
            if not participant.participant_id:
                print(f"❌ Empty participant_id found")
                return False
        
        # Check session type validation
        if not validate_conversation_isolation_pattern(generated_key, session.session_type):
            print(f"❌ Session type validation failed")
            return False
        
        print("✅ Data consistency check passed")
        return True
        
    except Exception as e:
        print(f"❌ Data consistency check failed: {e}")
        return False

2. Troubleshooting

Common Issue Diagnosis

Issue 1: Participant Validation Failure

def diagnose_participant_validation_error():
    """Diagnose participant validation errors"""
    try:
        # Test empty participant ID
        participant = ConversationParticipant("", "user")
    except ValueError as e:
        print(f"✅ Empty participant_id validation: {e}")
    
    try:
        # Test invalid participant type
        participant = ConversationParticipant("test", "invalid_type")
    except ValueError as e:
        print(f"✅ Invalid participant_type validation: {e}")
    
    try:
        # Test agent without role
        participant = ConversationParticipant("test", "agent")
    except ValueError as e:
        print(f"✅ Agent without role validation: {e}")

Issue 2: Session Type Validation Failure

def diagnose_session_validation_error():
    """Diagnose session validation errors"""
    try:
        # Test empty participant list
        session = ConversationSession(
            session_id="test",
            participants=[],
            session_type="user_to_mc",
            created_at=datetime.utcnow(),
            last_activity=datetime.utcnow()
        )
    except ValueError as e:
        print(f"✅ Empty participants validation: {e}")
    
    try:
        # Test invalid session type
        session = ConversationSession(
            session_id="test",
            participants=[ConversationParticipant("user", "user")],
            session_type="invalid_type",
            created_at=datetime.utcnow(),
            last_activity=datetime.utcnow()
        )
    except ValueError as e:
        print(f"✅ Invalid session_type validation: {e}")
    
    try:
        # Test participant type mismatch
        session = ConversationSession(
            session_id="test",
            participants=[ConversationParticipant("user", "user")],
            session_type="mc_to_agent",
            created_at=datetime.utcnow(),
            last_activity=datetime.utcnow()
        )
    except ValueError as e:
        print(f"✅ Participant type mismatch validation: {e}")

3. Performance Optimization

Batch Operation Optimization

def optimize_batch_operations():
    """Optimize batch operations"""
    # Batch create participants
    participants = []
    for i in range(1000):
        participant = ConversationParticipant(
            participant_id=f"participant_{i}",
            participant_type="agent",
            participant_role="test_role"
        )
        participants.append(participant)
    
    # Batch create sessions
    sessions = []
    for i in range(100):
        session = ConversationSession(
            session_id=f"session_{i}",
            participants=participants[i:i+2],
            session_type="agent_to_agent",
            created_at=datetime.utcnow(),
            last_activity=datetime.utcnow()
        )
        sessions.append(session)
    
    print(f"Created {len(participants)} participants and {len(sessions)} sessions")

Memory Usage Optimization

def optimize_memory_usage():
    """Optimize memory usage"""
    import gc
    import sys
    
    # Create many objects
    objects = []
    for i in range(10000):
        participant = ConversationParticipant(
            participant_id=f"participant_{i}",
            participant_type="agent",
            participant_role="test_role"
        )
        objects.append(participant)
    
    print(f"Memory usage before cleanup: {sys.getsizeof(objects)} bytes")
    
    # Clean up objects
    objects.clear()
    gc.collect()
    
    print(f"Memory usage after cleanup: {sys.getsizeof(objects)} bytes")

4. Data Migration

Model Version Upgrade

def migrate_models_to_new_version(old_data: Dict[str, Any]) -> Dict[str, Any]:
    """Migrate model data to new version"""
    # Check version
    version = old_data.get("version", "1.0")
    
    if version == "1.0":
        # Upgrade from 1.0 to 1.1
        if "participants" in old_data:
            for participant in old_data["participants"]:
                # Add new required fields
                if "metadata" not in participant:
                    participant["metadata"] = {}
        
        old_data["version"] = "1.1"
    
    return old_data

Data Format Conversion

def convert_data_formats():
    """Convert data formats"""
    # Convert from old format to new format
    old_participant = {
        "id": "participant_001",
        "type": "agent",
        "role": "data_analyst"
    }
    
    # Convert to new format
    new_participant = ConversationParticipant(
        participant_id=old_participant["id"],
        participant_type=old_participant["type"],
        participant_role=old_participant["role"],
        metadata={}
    )
    
    print(f"Converted participant: {new_participant}")

Monitoring and Logging

Model Usage Monitoring

import time
from typing import Dict, Any

class ConversationModelsMonitor:
    """Conversation Models Monitor"""
    
    def __init__(self):
        self.creation_metrics = {
            "participants": 0,
            "sessions": 0,
            "messages": 0
        }
        self.performance_metrics = {
            "participant_creation_time": [],
            "session_creation_time": [],
            "message_creation_time": []
        }
    
    def record_participant_creation(self, creation_time: float):
        """Record participant creation metrics"""
        self.creation_metrics["participants"] += 1
        self.performance_metrics["participant_creation_time"].append(creation_time)
    
    def record_session_creation(self, creation_time: float):
        """Record session creation metrics"""
        self.creation_metrics["sessions"] += 1
        self.performance_metrics["session_creation_time"].append(creation_time)
    
    def record_message_creation(self, creation_time: float):
        """Record message creation metrics"""
        self.creation_metrics["messages"] += 1
        self.performance_metrics["message_creation_time"].append(creation_time)
    
    def get_performance_report(self) -> Dict[str, Any]:
        """Get performance report"""
        report = {}
        
        for metric_name, times in self.performance_metrics.items():
            if times:
                report[metric_name] = {
                    "count": len(times),
                    "avg_time": sum(times) / len(times),
                    "min_time": min(times),
                    "max_time": max(times)
                }
        
        return report

Logging

import logging
from typing import Dict, Any

class ConversationModelsLogger:
    """Conversation Models Logger"""
    
    def __init__(self):
        self.logger = logging.getLogger(__name__)
    
    def log_participant_creation(self, participant: ConversationParticipant):
        """Log participant creation"""
        self.logger.info(f"Participant created: {participant.participant_id} - {participant.participant_type}")
    
    def log_session_creation(self, session: ConversationSession):
        """Log session creation"""
        self.logger.info(f"Session created: {session.session_id} - {session.session_type}")
    
    def log_message_creation(self, message: AgentCommunicationMessage):
        """Log message creation"""
        self.logger.info(f"Message created: {message.message_id} - {message.message_type}")
    
    def log_validation_error(self, error: Exception, context: str):
        """Log validation error"""
        self.logger.error(f"Validation error in {context}: {error}")

Version History

v1.0.0: Initial version, basic participant model
v1.1.0: Added session model and validation
v1.2.0: Added agent communication message model
v1.3.0: Added utility functions and session isolation
v1.4.0: Added data serialization support
v1.5.0: Added performance monitoring and logging

Conversation Models Technical Documentation

Overview

Design Motivation and Problem Background

Component Positioning

Component Type and Positioning

Component Type

Architecture Layers

Upstream Components (Consumers)

1. Domain Services

2. Application Layer Services

3. Infrastructure Layer

Downstream Components (Dependencies)

1. Python Standard Library

2. Domain Models

3. Utility Functions

Core Model Details

1. ConversationParticipant - Conversation Participant Model

2. ConversationSession - Conversation Session Model

Session Key Generation

Participant Lookup

Activity Update

3. AgentCommunicationMessage - Agent Communication Message Model

Convert to Conversation Message Format

Utility Functions Details

1. create_session_key - Session Key Generation Utility

2. validate_conversation_isolation_pattern - Session Isolation Validation Utility

Design Patterns Explained

1. Dataclass Pattern

2. Validation Pattern

3. Factory Pattern

Usage Examples

1. Basic Participant Management

2. Conversation Session Management

3. Agent Communication Messages

4. Utility Function Usage

5. Data Serialization and Deserialization

Maintenance Guide

1. Daily Maintenance

Model Validation

Data Consistency Check

2. Troubleshooting

Common Issue Diagnosis

3. Performance Optimization

Batch Operation Optimization

Memory Usage Optimization

4. Data Migration

Model Version Upgrade

Data Format Conversion

Monitoring and Logging

Model Usage Monitoring

Logging

Version History

Related Documentation