xbe.at

Miscellany / Time series and sequences

Structuring Chat Training Data

10 min read, 15 sections

Contents
  1. Preparing Chat Training Data Structure
  2. Token Count Analysis
  3. Data Validation Framework
  4. Cost Estimation and Optimization
  5. Dataset Augmentation Techniques
  6. Data Quality Assessment
  7. Formatting Validation Rules
  8. Token Distribution Analysis
  9. Content Quality Metrics
  10. Data Preprocessing Pipeline
  11. Real-world Implementation: Customer Service Bot
  12. Real-world Implementation: Technical Documentation Assistant
  13. Dataset Export and Validation Pipeline
  14. Training Data Analytics Dashboard
  15. Additional Resources

Preparing Chat Training Data Structure

The foundation of chat model fine-tuning lies in properly structuring conversation data. This involves organizing messages into clear roles and maintaining consistent formatting throughout the dataset. The data structure must follow OpenAI's JSONL format specifications.

import json
import tiktoken
from typing import List, Dict

def create_chat_format(system_prompt: str, conversations: List[Dict]) -> List[Dict]:
    formatted_data = []
    
    for conv in conversations:
        messages = [{"role": "system", "content": system_prompt}]
        
        for turn in conv:
            messages.append({
                "role": "user",
                "content": turn["user_input"]
            })
            messages.append({
                "role": "assistant",
                "content": turn["assistant_response"]
            })
            
        formatted_data.append({"messages": messages})
    
    return formatted_data

# Example usage
system_prompt = "You are a helpful AI assistant."
sample_conversations = [
    [
        {"user_input": "How do I analyze data?", 
         "assistant_response": "There are several steps to data analysis..."}
    ]
]

formatted = create_chat_format(system_prompt, sample_conversations)
print(json.dumps(formatted[0], indent=2))

Token Count Analysis

Understanding token usage is crucial for managing model constraints and costs. We implement a token counter using OpenAI's tiktoken library to analyze conversation lengths and ensure compliance with model limitations.

def count_tokens(messages: List[Dict]) -> int:
    encoding = tiktoken.get_encoding("cl100k_base")
    num_tokens = 0
    
    for message in messages:
        num_tokens += 4  # Every message follows {role/name}\n{content}\n format
        for key, value in message.items():
            num_tokens += len(encoding.encode(value))
            if key == "name":  # Role is handled by the 4 tokens above
                num_tokens -= 1  # Role is handled by the 4 tokens above
    
    num_tokens += 2  # Every reply is primed with <im_start>assistant
    return num_tokens

# Example usage
example_messages = [
    {"role": "system", "content": "You are a helpful assistant."},
    {"role": "user", "content": "What is machine learning?"},
    {"role": "assistant", "content": "Machine learning is a field of AI..."}
]

token_count = count_tokens(example_messages)
print(f"Total tokens: {token_count}")

Data Validation Framework

Data validation ensures dataset integrity and prevents common formatting issues. This comprehensive validator checks for required fields, proper role assignments, and content validity across the entire training dataset.

def validate_chat_data(data: List[Dict]) -> Dict:
    issues = {"missing_fields": [], "invalid_roles": [], "empty_content": []}
    valid_roles = {"system", "user", "assistant"}
    
    for idx, entry in enumerate(data):
        if "messages" not in entry:
            issues["missing_fields"].append(idx)
            continue
            
        for msg_idx, msg in enumerate(entry["messages"]):
            if not all(key in msg for key in ["role", "content"]):
                issues["missing_fields"].append(f"{idx}-{msg_idx}")
            
            if msg["role"] not in valid_roles:
                issues["invalid_roles"].append(f"{idx}-{msg_idx}")
                
            if not msg["content"].strip():
                issues["empty_content"].append(f"{idx}-{msg_idx}")
    
    return issues

# Example validation
sample_data = [
    {"messages": [
        {"role": "system", "content": "You are a helpful assistant"},
        {"role": "user", "content": "Hello"},
        {"role": "invalid_role", "content": "Hi there!"}
    ]}
]

validation_results = validate_chat_data(sample_data)
print("Validation issues:", json.dumps(validation_results, indent=2))

Cost Estimation and Optimization

An essential aspect of fine-tuning is understanding and optimizing costs. This implementation calculates estimated fine-tuning costs based on token usage and provides optimization recommendations.

def estimate_training_costs(data: List[Dict], 
                          cost_per_1k_tokens: float = 0.008,
                          num_epochs: int = 4) -> Dict:
    total_tokens = sum(count_tokens(entry["messages"]) for entry in data)
    
    # Calculate costs
    training_cost = (total_tokens * num_epochs * cost_per_1k_tokens) / 1000
    
    # Calculate statistics
    avg_tokens_per_example = total_tokens / len(data)
    total_cost = training_cost
    
    recommendations = []
    if avg_tokens_per_example > 4096:
        recommendations.append("Consider truncating long conversations")
    if len(data) < 100:
        recommendations.append("Dataset might be too small for effective fine-tuning")
    
    return {
        "total_tokens": total_tokens,
        "avg_tokens_per_example": avg_tokens_per_example,
        "estimated_cost": total_cost,
        "recommendations": recommendations
    }

# Example usage
cost_analysis = estimate_training_costs(formatted)
print(json.dumps(cost_analysis, indent=2))

Dataset Augmentation Techniques

Dataset augmentation enhances model generalization by creating diverse training examples through systematic variations. The implementation uses techniques like paraphrasing, synonym replacement, and context modification to expand the training dataset effectively.

import random
from typing import List, Dict

def augment_conversation_data(conversations: List[Dict]) -> List[Dict]:
    # Dictionary of synonym mappings for common words
    synonyms = {
        'help': ['assist', 'aid', 'support'],
        'how': ['what way', 'in what manner', 'by what means'],
        'explain': ['describe', 'clarify', 'elaborate on']
    }
    
    augmented_data = []
    for conv in conversations:
        # Add original conversation
        augmented_data.append(conv)
        
        # Create augmented version
        new_conv = {"messages": []}
        for msg in conv["messages"]:
            new_content = msg["content"]
            
            # Only augment user messages
            if msg["role"] == "user":
                for word, replacements in synonyms.items():
                    if word in new_content.lower():
                        new_content = new_content.replace(
                            word, 
                            random.choice(replacements)
                        )
                        
            new_conv["messages"].append({
                "role": msg["role"],
                "content": new_content
            })
        
        augmented_data.append(new_conv)
    
    return augmented_data

# Example usage
sample_conv = [{
    "messages": [
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Help me explain this concept."},
        {"role": "assistant", "content": "I'll explain it step by step."}
    ]
}]

augmented = augment_conversation_data(sample_conv)
print(f"Original conversations: {len(sample_conv)}")
print(f"Augmented conversations: {len(augmented)}")

Data Quality Assessment

Measuring dataset quality ensures effective fine-tuning outcomes. This implementation analyzes conversation coherence, response diversity, and content relevance through automated metrics and statistical analysis.

from collections import Counter
import numpy as np

def assess_data_quality(conversations: List[Dict]) -> Dict:
    metrics = {
        'avg_turn_length': [],
        'vocabulary_size': set(),
        'response_diversity': [],
        'role_balance': Counter()
    }
    
    for conv in conversations:
        for msg in conv["messages"]:
            # Track message lengths
            tokens = msg["content"].split()
            metrics['avg_turn_length'].append(len(tokens))
            
            # Build vocabulary
            metrics['vocabulary_size'].update(tokens)
            
            # Track role distribution
            metrics['role_balance'][msg["role"]] += 1
            
            # Calculate response diversity
            if msg["role"] == "assistant":
                unique_words = len(set(tokens))
                total_words = len(tokens)
                if total_words > 0:
                    metrics['response_diversity'].append(
                        unique_words / total_words
                    )
    
    return {
        'avg_turn_length': np.mean(metrics['avg_turn_length']),
        'vocabulary_size': len(metrics['vocabulary_size']),
        'response_diversity': np.mean(metrics['response_diversity']),
        'role_distribution': dict(metrics['role_balance'])
    }

# Example usage
quality_metrics = assess_data_quality(sample_conv)
print(json.dumps(quality_metrics, indent=2))

Formatting Validation Rules

Consistent formatting across the training dataset ensures optimal fine-tuning results. This implementation validates conversation structures, checks message sequences, and enforces content guidelines according to OpenAI's specifications.

def validate_chat_format(conversations: List[Dict]) -> Dict:
    results = {'valid': 0, 'issues': []}
    
    for idx, conv in enumerate(conversations):
        messages = conv.get('messages', [])
        
        # Basic structure validation
        if not messages:
            results['issues'].append(f'Empty conversation at index {idx}')
            continue
            
        # Check system message
        if not (messages[0]['role'] == 'system' and messages[0]['content']):
            results['issues'].append(f'Missing/invalid system message at {idx}')
            continue
            
        # Validate conversation flow
        for i in range(1, len(messages), 2):
            if i >= len(messages) or messages[i]['role'] != 'user':
                results['issues'].append(f'Invalid user message at {idx}')
                continue
                
            if i+1 >= len(messages) or messages[i+1]['role'] != 'assistant':
                results['issues'].append(f'Invalid assistant message at {idx}')
                continue
        
        if not results['issues']:
            results['valid'] += 1
            
    return results

# Example usage
conversations = [
    {
        'messages': [
            {'role': 'system', 'content': 'You are a helpful assistant'},
            {'role': 'user', 'content': 'Hello'},
            {'role': 'assistant', 'content': 'Hi there!'}
        ]
    }
]

validation_results = validate_chat_format(conversations)
print(f"Valid conversations: {validation_results['valid']}")
print(f"Issues found: {len(validation_results['issues'])}")

Token Distribution Analysis

Understanding token distribution patterns helps optimize training data efficiency. This implementation analyzes token counts across different message roles and identifies potential optimization opportunities.

def analyze_token_distribution(conversations: List[Dict]) -> Dict:
    import tiktoken
    encoding = tiktoken.get_encoding("cl100k_base")
    
    stats = {
        'role_tokens': {'system': [], 'user': [], 'assistant': []},
        'total_tokens': 0,
        'conversations_above_limit': 0
    }
    
    for conv in conversations:
        conv_tokens = 0
        for msg in conv['messages']:
            tokens = len(encoding.encode(msg['content']))
            stats['role_tokens'][msg['role']].append(tokens)
            conv_tokens += tokens
            
        stats['total_tokens'] += conv_tokens
        if conv_tokens > 4096:  # OpenAI's typical limit
            stats['conversations_above_limit'] += 1
    
    # Calculate averages
    for role in stats['role_tokens']:
        tokens = stats['role_tokens'][role]
        stats[f'avg_{role}_tokens'] = sum(tokens) / len(tokens) if tokens else 0
    
    return stats

# Example usage
token_stats = analyze_token_distribution(conversations)
print(f"Total tokens: {token_stats['total_tokens']}")
print(f"Average system tokens: {token_stats['avg_system_tokens']:.2f}")
print(f"Long conversations: {token_stats['conversations_above_limit']}")

Content Quality Metrics

Content quality assessment focuses on measuring the effectiveness of training examples through quantitative metrics. This implementation evaluates message coherence, vocabulary richness, and semantic relevance using statistical analysis.

def assess_content_quality(conversations: List[Dict]) -> Dict:
    total_messages = 0
    total_words = 0
    unique_words = set()
    response_lengths = []
    
    quality_scores = {
        'lexical_density': 0.0,
        'avg_response_length': 0.0,
        'content_variety': 0.0
    }
    
    for conv in conversations:
        for msg in conv['messages']:
            if msg['role'] == 'assistant':
                words = msg['content'].lower().split()
                total_messages += 1
                total_words += len(words)
                unique_words.update(words)
                response_lengths.append(len(words))
    
    if total_messages > 0 and total_words > 0:
        quality_scores['lexical_density'] = len(unique_words) / total_words
        quality_scores['avg_response_length'] = total_words / total_messages
        quality_scores['content_variety'] = len(unique_words) / total_messages
    
    return quality_scores

# Example usage
sample_conversations = [
    {
        'messages': [
            {'role': 'system', 'content': 'You are a helpful assistant'},
            {'role': 'user', 'content': 'Explain neural networks'},
            {'role': 'assistant', 'content': 'Neural networks are computational models inspired by biological neural networks'}
        ]
    }
]

quality_metrics = assess_content_quality(sample_conversations)
print(f"Lexical Density: {quality_metrics['lexical_density']:.3f}")
print(f"Average Response Length: {quality_metrics['avg_response_length']:.1f}")

Data Preprocessing Pipeline

A robust preprocessing pipeline ensures consistent data quality across the training set. This implementation handles text normalization, special character handling, and maintains conversation context integrity.

import re
from typing import List, Dict

def preprocess_conversations(conversations: List[Dict]) -> List[Dict]:
    def clean_text(text: str) -> str:
        # Remove extra whitespace
        text = ' '.join(text.split())
        # Normalize quotes and dashes
        text = re.sub(r'[""']', '"', text)
        text = re.sub(r'[–—]', '-', text)
        return text.strip()
    
    processed_data = []
    
    for conv in conversations:
        processed_conv = {'messages': []}
        
        for msg in conv['messages']:
            processed_msg = {
                'role': msg['role'],
                'content': clean_text(msg['content'])
            }
            
            # Skip empty messages
            if processed_msg['content']:
                processed_conv['messages'].append(processed_msg)
        
        # Only add conversations with valid exchanges
        if len(processed_conv['messages']) >= 3:  # system + user + assistant
            processed_data.append(processed_conv)
    
    return processed_data

# Example usage
preprocessed = preprocess_conversations(sample_conversations)
print(f"Preprocessed conversations: {len(preprocessed)}")

Real-world Implementation: Customer Service Bot

A practical implementation of a customer service chatbot training pipeline demonstrates the complete workflow from raw conversation logs to fine-tuning ready dataset. This includes data cleaning, formatting, and validation steps.

def prepare_customer_service_dataset(raw_logs: List[Dict]) -> Dict:
    # Configuration
    MAX_TOKENS = 2048
    MIN_RESPONSE_LENGTH = 10
    
    # Initialize metrics
    stats = {
        'processed': 0,
        'skipped': 0,
        'total_tokens': 0
    }
    
    training_data = []
    system_prompt = "You are a helpful customer service assistant."
    
    for log in raw_logs:
        # Basic validation
        if not all(k in log for k in ['query', 'response', 'category']):
            stats['skipped'] += 1
            continue
            
        # Format conversation
        conversation = {
            'messages': [
                {'role': 'system', 'content': system_prompt},
                {'role': 'user', 'content': log['query'].strip()},
                {'role': 'assistant', 'content': log['response'].strip()}
            ]
        }
        
        # Apply quality filters
        if len(conversation['messages'][-1]['content'].split()) < MIN_RESPONSE_LENGTH:
            stats['skipped'] += 1
            continue
            
        # Add metadata
        conversation['category'] = log['category']
        training_data.append(conversation)
        stats['processed'] += 1
        
    return {'data': training_data, 'metrics': stats}

# Example usage
customer_logs = [
    {
        'query': 'How do I reset my password?',
        'response': 'To reset your password, go to the login page and click Forgot Password. Follow the email instructions.',
        'category': 'account_management'
    }
]

result = prepare_customer_service_dataset(customer_logs)
print(f"Processed: {result['metrics']['processed']}")
print(f"Skipped: {result['metrics']['skipped']}")

Real-world Implementation: Technical Documentation Assistant

This implementation showcases the preparation of technical documentation for training a specialized documentation assistant, including code snippet handling and technical term validation.

def prepare_technical_docs_dataset(documentation: List[Dict]) -> Dict:
    # Initialize processing metrics
    metrics = {'processed': 0, 'code_blocks': 0, 'technical_terms': 0}
    
    def extract_code_blocks(text: str) -> List[str]:
        # Simple code block extraction
        code_pattern = r'```[\s\S]*?```'
        return re.findall(code_pattern, text)
    
    training_data = []
    
    for doc in documentation:
        # Validate document structure
        if 'question' not in doc or 'explanation' not in doc:
            continue
            
        # Process code blocks
        code_blocks = extract_code_blocks(doc['explanation'])
        metrics['code_blocks'] += len(code_blocks)
        
        # Format conversation
        conversation = {
            'messages': [
                {
                    'role': 'system',
                    'content': 'You are a technical documentation assistant.'
                },
                {'role': 'user', 'content': doc['question']},
                {'role': 'assistant', 'content': doc['explanation']}
            ]
        }
        
        training_data.append(conversation)
        metrics['processed'] += 1
    
    return {
        'data': training_data,
        'metrics': metrics
    }

# Example usage
docs_data = [
    {
        'question': 'How do I implement a binary search?',
        'explanation': 'Here is a binary search implementation:\n```python\ndef binary_search(arr, target):\n    left, right = 0, len(arr)-1\n    while left <= right:\n        mid = (left + right) // 2\n        if arr[mid] == target:\n            return mid\n        elif arr[mid] < target:\n            left = mid + 1\n        else:\n            right = mid - 1\n    return -1\n```'
    }
]

result = prepare_technical_docs_dataset(docs_data)
print(f"Processed documents: {result['metrics']['processed']}")
print(f"Code blocks found: {result['metrics']['code_blocks']}")

[Continuing with remaining slides...]

Dataset Export and Validation Pipeline

The final stage of data preparation involves exporting the processed dataset in JSONL format while performing final validations. This implementation ensures the output meets OpenAI's fine-tuning requirements.

def export_training_dataset(conversations: List[Dict], output_file: str) -> Dict:
    import json
    
    stats = {
        'exported_count': 0,
        'validation_errors': [],
        'file_size_mb': 0
    }
    
    def validate_conversation(conv: Dict) -> bool:
        if not conv.get('messages'):
            return False
        if len(conv['messages']) < 2:
            return False
        if any(not msg.get('content') for msg in conv['messages']):
            return False
        return True
    
    # Export valid conversations to JSONL
    with open(output_file, 'w', encoding='utf-8') as f:
        for conv in conversations:
            if validate_conversation(conv):
                f.write(json.dumps(conv) + '\n')
                stats['exported_count'] += 1
            else:
                stats['validation_errors'].append(
                    f"Invalid conversation format at index {stats['exported_count']}"
                )
    
    # Calculate file size
    import os
    stats['file_size_mb'] = os.path.getsize(output_file) / (1024 * 1024)
    
    return stats

# Example usage
conversations = [
    {
        'messages': [
            {'role': 'system', 'content': 'You are a helpful assistant.'},
            {'role': 'user', 'content': 'How do I use Python generators?'},
            {'role': 'assistant', 'content': 'Generators are functions that can pause and resume their state...'}
        ]
    }
]

export_stats = export_training_dataset(conversations, 'training_data.jsonl')
print(f"Exported conversations: {export_stats['exported_count']}")
print(f"File size: {export_stats['file_size_mb']:.2f} MB")

Training Data Analytics Dashboard

Implementing data analytics capabilities helps monitor dataset quality and identify potential improvements. This code generates key metrics for dataset assessment and optimization.

def generate_dataset_analytics(conversations: List[Dict]) -> Dict:
    analytics = {
        'total_conversations': len(conversations),
        'conversation_lengths': [],
        'role_distribution': {'system': 0, 'user': 0, 'assistant': 0},
        'content_stats': {
            'avg_tokens_per_message': 0,
            'max_message_length': 0,
            'min_message_length': float('inf')
        }
    }
    
    total_tokens = 0
    total_messages = 0
    
    for conv in conversations:
        # Track conversation length
        conv_length = len(conv['messages'])
        analytics['conversation_lengths'].append(conv_length)
        
        for msg in conv['messages']:
            # Update role distribution
            analytics['role_distribution'][msg['role']] += 1
            
            # Track message lengths
            msg_length = len(msg['content'].split())
            total_tokens += msg_length
            total_messages += 1
            
            analytics['content_stats']['max_message_length'] = max(
                analytics['content_stats']['max_message_length'], 
                msg_length
            )
            analytics['content_stats']['min_message_length'] = min(
                analytics['content_stats']['min_message_length'], 
                msg_length
            )
    
    # Calculate averages
    analytics['content_stats']['avg_tokens_per_message'] = (
        total_tokens / total_messages if total_messages > 0 else 0
    )
    
    return analytics

# Example usage
analytics_results = generate_dataset_analytics(conversations)
print(f"Total conversations: {analytics_results['total_conversations']}")
print(f"Average tokens per message: {analytics_results['content_stats']['avg_tokens_per_message']:.2f}")

Additional Resources

  1. https://arxiv.org/abs/2307.09288 - "Large Language Models for Automated Data Preparation and Quality Assessment"
  2. https://arxiv.org/abs/2305.14688 - "Quality Metrics for Chat-based Language Model Training Data"
  3. https://arxiv.org/abs/2308.12035 - "Best Practices for Fine-tuning Large Language Models with Conversation Data"
  4. https://arxiv.org/abs/2304.11117 - "Data Quality Assessment for Large Language Model Training Sets"

More in Time series and sequences

  1. Time Series Analysis with Neural-Prophet14 min

    Time series analysis is a crucial technique for understanding and predicting patterns in sequential data. Neural-Prophet, an advanced forecasting model...

  2. Scaling ARIMA Models Across Industries8 min

    Time series forecasting is a crucial technique for predicting future values based on historical data. This presentation will compare two significant...

  3. Time Series Forecasting Project Walkthrough10 min

    Time series analysis requires specific Python libraries for statistical computations, data manipulation, and visualization. We'll import essential packages...

  4. Dynamic Mode Decomposition17 min

    Dynamic Mode Decomposition (DMD) is a powerful data-driven method for analyzing complex dynamical systems. It extracts spatiotemporal coherent structures...

  5. Time Series Modeling Challenges and Solutions13 min

    Time series modeling is a critical component in various applications, from weather forecasting to stock market analysis. This presentation introduces a novel...

  6. Challenges and Applications of Dynamic Mode Decomposition10 min

    Dynamic Mode Decomposition (DMD) is a powerful data-driven method for analyzing complex dynamical systems. It extracts spatiotemporal coherent structures...