Create generate_dataset.py

generate_dataset.py

@@ -0,0 +1,1603 @@
+import os
+import json
+import pandas as pd
+import numpy as np
+from typing import List, Dict, Tuple, Optional, Any
+import logging
+import random
+import time
+from tqdm import tqdm
+from openai import AzureOpenAI
+from datetime import datetime
+import concurrent.futures
+import threading
+from dataclasses import dataclass
+import queue
+import math
+import re
+
+# Configure logging
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
+    handlers=[
+        logging.FileHandler("sales_ai_system.log"),
+        logging.StreamHandler()
+    ]
+)
+
+logger = logging.getLogger(__name__)
+
+# Hard-coded Azure OpenAI credentials
+AZURE_OPENAI_API_KEY =""
+AZURE_OPENAI_DEPLOYMENT_NAME = ""
+AZURE_OPENAI_ENDPOINT = "https://resource_name.openai.azure.com/"
+AZURE_EMBEDDING_DEPLOYMENT_NAME = "text-embedding-3-large"
+
+# Rate limiting parameters
+RATE_LIMIT_RPM = 2500  # Requests per minute limit
+# Use only 60% of the limit to account for retry overhead and be conservative
+SAFE_RATE_LIMIT_RPM = int(RATE_LIMIT_RPM * 0.6)  
+RATE_LIMIT_RPS = SAFE_RATE_LIMIT_RPM / 60  # Requests per second
+MIN_REQUEST_INTERVAL = 1.0 / RATE_LIMIT_RPS  # Minimum interval between requests
+
+# Thread-local storage for clients
+local = threading.local()
+
+# More granular rate limiters for different endpoints
+class FixedWindowRateLimiter:
+    """
+    Fixed window rate limiter that tracks requests in a sliding window.
+    More conservative than token bucket for API rate limits.
+    """
+    def __init__(self, max_requests, window_size=60):
+        self.max_requests = max_requests
+        self.window_size = window_size  # in seconds
+        self.request_timestamps = []
+        self.lock = threading.Lock()
+    
+    def wait_if_needed(self):
+        """Wait if the rate limit would be exceeded."""
+        with self.lock:
+            now = time.time()
+            
+            # Remove timestamps outside the window
+            self.request_timestamps = [ts for ts in self.request_timestamps 
+                                     if now - ts < self.window_size]
+            
+            # Check if we're at the limit
+            if len(self.request_timestamps) >= self.max_requests:
+                # Calculate wait time - oldest timestamp will be removed after window_size
+                wait_time = self.window_size - (now - self.request_timestamps[0])
+                
+                # Add a small jitter to avoid all threads waking up at the same time
+                wait_time = max(0.1, wait_time + random.uniform(0, 0.5))
+                
+                return wait_time
+            
+            # No need to wait
+            self.request_timestamps.append(now)
+            return 0
+    
+    def record_request(self):
+        """Record a request without waiting."""
+        with self.lock:
+            now = time.time()
+            self.request_timestamps.append(now)
+
+# Create rate limiters
+chat_limiter = FixedWindowRateLimiter(SAFE_RATE_LIMIT_RPM)
+embedding_limiter = FixedWindowRateLimiter(SAFE_RATE_LIMIT_RPM)
+
+@dataclass
+class SaaSProfile:
+    """Profile of a SaaS product."""
+    company_id: str
+    company_name: str
+    product_name: str
+    product_description: str
+    target_audience: List[str]
+    key_features: List[str]
+    value_propositions: List[str]
+    pricing_structure: Dict[str, Any]
+    common_objections: List[str]
+    industry: str
+    company_size: str  # small, medium, large, enterprise
+
+class ResultWriter:
+    """Thread-safe writer for results to CSV"""
+    
+    def __init__(self, output_path: str, columns: List[str]):
+        self.output_path = output_path
+        self.columns = columns
+        self.lock = threading.Lock()
+        
+        # Create the file with headers if it doesn't exist
+        if not os.path.exists(output_path):
+            with open(output_path, 'w') as f:
+                f.write(','.join(columns) + '\n')
+        
+        logger.info(f"Initialized ResultWriter for {output_path}")
+    
+    def write_rows(self, rows: List[Dict]):
+        """Write rows directly to the CSV file"""
+        if not rows:
+            return
+            
+        try:
+            with self.lock:
+                with open(self.output_path, 'a') as f:
+                    for row in rows:
+                        # Convert values to strings and escape commas
+                        csv_values = []
+                        for col in self.columns:
+                            val = str(row.get(col, ""))
+                            # Escape double quotes by doubling them
+                            val = val.replace('"', '""')
+                            # Wrap in quotes
+                            csv_values.append(f'"{val}"')
+                        csv_line = ','.join(csv_values)
+                        f.write(csv_line + '\n')
+                        
+            logger.debug(f"Wrote {len(rows)} rows to {self.output_path}")
+        except Exception as e:
+            logger.error(f"Error writing to CSV: {str(e)}")
+
+class EnhancedSaaSDatasetGenerator:
+    """
+    Generates diverse and realistic synthetic sales conversation datasets for SaaS companies
+    using Azure OpenAI with multithreading support.
+    """
+    
+    def __init__(self):
+        """Initialize the dataset generator."""
+        self.profile_lock = threading.Lock()
+        self.profiles = None
+        self.writer = None
+        self.total_generated = 0
+        self.total_counter_lock = threading.Lock()
+        
+        # Track API call statistics
+        self.api_calls = 0
+        self.api_call_lock = threading.Lock()
+        self.retry_count = 0
+        self.retry_lock = threading.Lock()
+        
+        # Track last retry timestamp to prevent thundering herd
+        self.last_retry_time = 0
+        self.last_retry_lock = threading.Lock()
+        
+        # Conversation style templates - define various styles to create diverse conversations
+        self.conversation_styles = [
+            "casual_friendly",
+            "direct_professional",
+            "technical_detailed",
+            "consultative_advisory",
+            "empathetic_supportive",
+            "skeptical_challenging",
+            "urgent_time_pressed",
+            "confused_overwhelmed",
+            "knowledgeable_assertive",
+            "storytelling_narrative"
+        ]
+        
+        # Communication channel templates with their unique characteristics
+        self.communication_channels = {
+            "email": {
+                "formality": "high",
+                "response_time": "delayed",
+                "message_length": "medium to long",
+                "format_elements": ["subject lines", "signatures", "quoted replies"]
+            },
+            "live_chat": {
+                "formality": "low to medium",
+                "response_time": "immediate",
+                "message_length": "short to medium",
+                "format_elements": ["quick responses", "emojis", "typing indicators"]
+            },
+            "phone_call": {
+                "formality": "medium",
+                "response_time": "immediate",
+                "message_length": "conversational",
+                "format_elements": ["verbal pauses", "interruptions", "voice tone indicators"]
+            },
+            "video_call": {
+                "formality": "medium",
+                "response_time": "immediate",
+                "message_length": "medium",
+                "format_elements": ["screen sharing references", "visual cues", "environment mentions"]
+            },
+            "in_person": {
+                "formality": "varies",
+                "response_time": "immediate",
+                "message_length": "varies",
+                "format_elements": ["environment references", "body language cues", "material handouts"]
+            },
+            "sms": {
+                "formality": "low",
+                "response_time": "varies",
+                "message_length": "very short",
+                "format_elements": ["abbreviations", "emojis", "brief statements"]
+            },
+            "social_media": {
+                "formality": "low to medium",
+                "response_time": "varies",
+                "message_length": "short",
+                "format_elements": ["hashtags", "mentions", "public/private context references"]
+            }
+        }
+        
+        # Customer personas templates with more diverse traits
+        self.persona_templates = [
+            {
+                "name": "Time-Pressed Executive",
+                "traits": ["direct", "value-focused", "impatient", "decisive"],
+                "communication_style": "brief and to-the-point, may use truncated sentences and check messages between meetings",
+                "typical_objections": ["too time-consuming", "prove ROI quickly", "competitor comparisons"]
+            },
+            {
+                "name": "Technical Evaluator",
+                "traits": ["detail-oriented", "skeptical", "analytical", "research-driven"],
+                "communication_style": "asks specific technical questions, uses industry jargon, references research",
+                "typical_objections": ["technical limitations", "integration concerns", "security issues"]
+            },
+            {
+                "name": "Budget-Conscious Manager",
+                "traits": ["price-sensitive", "cautious", "ROI-focused", "deliberate"],
+                "communication_style": "frequently mentions costs, compares alternatives, asks about discounts",
+                "typical_objections": ["too expensive", "budget constraints", "not worth the investment"]
+            },
+            {
+                "name": "Relationship Builder",
+                "traits": ["conversational", "personable", "story-driven", "trust-focused"],
+                "communication_style": "shares personal anecdotes, asks about the sales rep, builds rapport before business",
+                "typical_objections": ["need to build trust", "want references", "need team buy-in"]
+            },
+            {
+                "name": "Innovation Seeker",
+                "traits": ["trend-aware", "competitive", "growth-focused", "risk-tolerant"],
+                "communication_style": "references industry trends, talks about growth goals, explores cutting-edge features",
+                "typical_objections": ["not innovative enough", "will soon be outdated", "competitive advantage concerns"]
+            },
+            {
+                "name": "Overwhelmed User",
+                "traits": ["stressed", "confused", "seeking guidance", "time-constrained"],
+                "communication_style": "asks many basic questions, may seem scattered, expresses feeling overwhelmed",
+                "typical_objections": ["too complicated", "training requirements", "implementation time"]
+            },
+            {
+                "name": "Delegated Researcher",
+                "traits": ["information-gathering", "non-decision-maker", "thorough", "process-oriented"],
+                "communication_style": "mentions reporting back to others, asks for materials, follows structured evaluation",
+                "typical_objections": ["need to consult others", "gathering information only", "complex approval process"]
+            },
+            {
+                "name": "Competitor User",
+                "traits": ["comparative", "experienced", "specific needs", "solution-aware"],
+                "communication_style": "frequently mentions current solution, asks about specific differences, uses competitor terminology",
+                "typical_objections": ["switching costs", "feature parity", "disruption concerns"]
+            },
+            {
+                "name": "Enthusiastic Champion",
+                "traits": ["excited", "vision-aligned", "quick to connect", "internal seller"],
+                "communication_style": "expresses excitement, talks about company vision, discusses internal advocacy",
+                "typical_objections": ["need help convincing others", "implementation support", "proving value to team"]
+            },
+            {
+                "name": "Resistant Stakeholder",
+                "traits": ["change-averse", "skeptical", "security-focused", "process-oriented"],
+                "communication_style": "questions necessity, raises potential problems, defensive about current solutions",
+                "typical_objections": ["disruption to workflow", "employee resistance", "security concerns"]
+            }
+        ]
+        
+        # Conversation flow patterns to create non-linear, realistic conversations
+        self.conversation_flows = [
+            "standard_linear",  # Traditional linear sales conversation
+            "multiple_objection_loops",  # Customer raises several objections that must be addressed
+            "subject_switching",  # Conversation jumps between different topics
+            "interrupted_followup",  # Conversation gets interrupted and resumes later
+            "technical_deep_dive",  # Detailed exploration of technical aspects
+            "competitive_comparison",  # Heavy focus on comparing with competitors
+            "gradual_discovery",  # Slow revelation of needs throughout conversation
+            "immediate_interest",  # Customer shows high interest from the beginning
+            "initial_rejection",  # Starts negative but potentially turns around
+            "stakeholder_expansion",  # Involves bringing in additional decision makers
+            "pricing_negotiation",  # Extended discussion about pricing and terms
+            "implementation_concerns",  # Focused on implementation challenges
+            "value_justification",  # Customer needs convincing on ROI
+            "relationship_building",  # Heavy on personal connection before business
+            "multi_session",  # Simulates a conversation occurring across multiple contacts
+            "demo_walkthrough"  # Simulates a product demonstration conversation
+        ]
+        
+        # Speech patterns and quirks to make conversations more human-like
+        self.speech_patterns = [
+            # Hesitations and fillers
+            {"pattern": "filler_words", "examples": ["um", "uh", "like", "you know", "actually", "basically", "I mean"]},
+            
+            # Grammatical quirks
+            {"pattern": "run_on_sentences", "examples": ["and then we also", "plus we need to", "which also means"]},
+            {"pattern": "self_corrections", "examples": ["I mean", "what I meant was", "actually, let me rephrase", "sorry, what I'm trying to say"]},
+            
+            # Typing variations (for written channels)
+            {"pattern": "typos", "examples": ["teh", "adn", "waht", "thigns", "compnay", "prodcut", "featuers"]},
+            {"pattern": "autocorrections", "examples": ["Our team is looking for skeletons (solutions)", "We need better coffin (conferencing)"]},
+            
+            # Regional expressions
+            {"pattern": "regionalisms", "examples": ["y'all", "folks", "brilliant", "cheers", "no worries", "wicked", "proper"]},
+            
+            # Punctuation habits
+            {"pattern": "over_punctuation", "examples": ["!!!", "???", "..."]},
+            {"pattern": "under_punctuation", "examples": ["no periods", "run on thoughts", "missing question marks"]},
+            
+            # Message structure
+            {"pattern": "fragmented_thoughts", "examples": ["Need to check on...", "Not sure if...", "Let me think...", "One more thing."]},
+            {"pattern": "tangents", "examples": ["By the way", "Oh that reminds me", "Not related, but", "Random thought"]},
+            
+            # Emphasis patterns
+            {"pattern": "emphasis_capital", "examples": ["REALLY", "VERY", "NEVER", "ALWAYS", "MUST"]},
+            {"pattern": "emphasis_repetition", "examples": ["very very", "really really", "many many"]}
+        ]
+        
+        # Customer needs categories with specific language patterns
+        self.customer_needs = [
+            {"type": "efficiency", "keywords": ["faster", "streamline", "automate", "time-consuming", "manual", "process", "workflow"]},
+            {"type": "cost_reduction", "keywords": ["expenses", "budget", "save money", "affordable", "cost-effective", "ROI", "investment"]},
+            {"type": "growth", "keywords": ["scale", "expand", "increase revenue", "market share", "competitive", "opportunity", "growth"]},
+            {"type": "compliance", "keywords": ["regulations", "requirements", "standards", "audit", "legal", "compliance", "risk"]},
+            {"type": "integration", "keywords": ["connect", "compatible", "ecosystem", "work with", "existing systems", "API", "integration"]},
+            {"type": "usability", "keywords": ["easy to use", "intuitive", "learning curve", "training", "user-friendly", "simple", "interface"]},
+            {"type": "reliability", "keywords": ["uptime", "stable", "dependable", "trust", "consistent", "failover", "backup"]},
+            {"type": "security", "keywords": ["protect", "data security", "encryption", "sensitive information", "breach", "privacy", "secure"]},
+            {"type": "support", "keywords": ["help", "customer service", "response time", "training", "documentation", "support team", "assistance"]},
+            {"type": "analytics", "keywords": ["insights", "reporting", "dashboard", "metrics", "data", "visibility", "analytics"]}
+        ]
+        
+        logger.info("Initialized EnhancedSaaSDatasetGenerator with diverse conversation templates")
+    
+    def get_openai_client(self):
+        """Get thread-local Azure OpenAI client"""
+        if not hasattr(local, 'client'):
+            local.client = AzureOpenAI(
+                api_key=AZURE_OPENAI_API_KEY,
+                api_version="2023-05-15",
+                azure_endpoint=AZURE_OPENAI_ENDPOINT
+            )
+        return local.client
+    
+    def _wait_with_jitter(self, base_time):
+        """Wait with jitter to avoid thundering herd problem"""
+        jitter = random.uniform(0, 1)
+        wait_time = base_time + jitter
+        time.sleep(wait_time)
+    
+    def _handle_rate_limit(self, retry_after=None):
+        """Handle rate limit exceptions with proper backoff"""
+        with self.retry_lock:
+            self.retry_count += 1
+        
+        # If retry_after is provided in header, use it; otherwise use default
+        wait_time = retry_after if retry_after else 30
+        
+        # Add jitter to avoid thundering herd problem
+        jitter = random.uniform(0, 5)
+        wait_time += jitter
+        
+        # Update last retry time - all threads can see when the last retry happened
+        with self.last_retry_lock:
+            self.last_retry_time = time.time()
+        
+        logger.warning(f"Rate limit exceeded. Waiting for {wait_time:.2f} seconds before retry.")
+        time.sleep(wait_time)
+    
+    def _get_embedding(self, text: str) -> List[float]:
+        """Get embeddings using Azure OpenAI."""
+        max_retries = 5
+        
+        for attempt in range(max_retries):
+            try:
+                # Check if we need to wait based on embedding rate limiter
+                wait_time = embedding_limiter.wait_if_needed()
+                if wait_time > 0:
+                    time.sleep(wait_time)
+                
+                # Check if any thread recently hit a rate limit
+                with self.last_retry_lock:
+                    time_since_last_retry = time.time() - self.last_retry_time
+                
+                # If a retry happened recently, stagger our requests
+                if time_since_last_retry < 10:
+                    time.sleep(random.uniform(0.5, 3.0))
+                
+                client = self.get_openai_client()
+                response = client.embeddings.create(
+                    model=AZURE_EMBEDDING_DEPLOYMENT_NAME,
+                    input=text
+                )
+                
+                # Track successful API call
+                with self.api_call_lock:
+                    self.api_calls += 1
+                
+                return response.data[0].embedding
+                
+            except Exception as e:
+                error_msg = str(e).lower()
+                
+                # Handle rate limit errors
+                if "429" in error_msg or "too many requests" in error_msg:
+                    self._handle_rate_limit()
+                    continue
+                
+                # Log other errors
+                logger.error(f"Error attempt {attempt+1}/{max_retries} getting embedding: {str(e)}")
+                
+                if attempt < max_retries - 1:
+                    # Exponential backoff with jitter
+                    backoff_time = (2 ** attempt) + random.uniform(0, 1)
+                    time.sleep(backoff_time)
+                else:
+                    logger.error(f"Failed to get embedding after {max_retries} attempts")
+                    return [0.0] * 3072  # Default dimension for embeddings
+    
+    def _generate_completion(self, system_prompt: str, user_prompt: str, temperature: float = 0.7, retries: int = 5) -> str:
+        """Generate text completion using Azure OpenAI with improved error handling."""
+        client = self.get_openai_client()
+        
+        for attempt in range(retries):
+            try:
+                # Check if we need to wait based on chat rate limiter
+                wait_time = chat_limiter.wait_if_needed()
+                if wait_time > 0:
+                    time.sleep(wait_time)
+                
+                # Check if any thread recently hit a rate limit
+                with self.last_retry_lock:
+                    time_since_last_retry = time.time() - self.last_retry_time
+                
+                # If a retry happened recently, stagger our requests
+                if time_since_last_retry < 10:
+                    time.sleep(random.uniform(0.5, 3.0))
+                
+                # Add explicit JSON formatting request to the system prompt
+                enhanced_system_prompt = f"{system_prompt}\nImportant: Your response must be valid JSON only, with no explanations or additional text."
+                
+                # Add explicit JSON formatting instructions to the user prompt
+                enhanced_user_prompt = f"{user_prompt}\n\nYour response should be formatted as valid JSON only. Do not include any text before or after the JSON."
+                
+                response = client.chat.completions.create(
+                    model=AZURE_OPENAI_DEPLOYMENT_NAME,
+                    messages=[
+                        {"role": "system", "content": enhanced_system_prompt},
+                        {"role": "user", "content": enhanced_user_prompt}
+                    ],
+                    temperature=temperature
+                )
+                
+                # Track successful API call
+                with self.api_call_lock:
+                    self.api_calls += 1
+                
+                content = response.choices[0].message.content.strip()
+                
+                # Remove any potential non-JSON content before and after the actual JSON
+                if content.startswith("```json"):
+                    content = content.split("```json", 1)[1]
+                if content.endswith("```"):
+                    content = content.rsplit("```", 1)[0]
+                
+                # Further cleanup to ensure we have valid JSON
+                content = content.strip()
+                
+                # Check if the first character is not '{' but contains '{' somewhere
+                if not content.startswith('{') and '{' in content:
+                    content = content[content.find('{'):]
+                
+                # Check if the last character is not '}' but contains '}' somewhere
+                if not content.endswith('}') and '}' in content:
+                    content = content[:content.rfind('}')+1]
+                
+                try:
+                    # Parse and re-serialize to ensure valid JSON
+                    parsed_json = json.loads(content)
+                    return json.dumps(parsed_json)
+                except json.JSONDecodeError:
+                    # If we can't parse the JSON, try to fix common issues
+                    # Replace single quotes with double quotes
+                    content = content.replace("'", '"')
+                    # Fix unquoted keys
+                    content = re.sub(r'(\s*?)(\w+)(\s*?):', r'\1"\2"\3:', content)
+                    try:
+                        parsed_json = json.loads(content)
+                        return json.dumps(parsed_json)
+                    except json.JSONDecodeError:
+                        if attempt < retries - 1:
+                            continue
+                        else:
+                            # Create minimal valid JSON as fallback
+                            logger.warning("Returning fallback JSON due to parsing error")
+                            return '{"error": "Could not generate valid JSON", "partial_content": "Content generation failed"}'
+                
+            except Exception as e:
+                error_msg = str(e).lower()
+                
+                # Handle rate limit errors
+                if "429" in error_msg or "too many requests" in error_msg:
+                    self._handle_rate_limit()
+                    continue
+                
+                # Log other errors
+                logger.error(f"Error attempt {attempt+1}/{retries} generating completion: {str(e)}")
+                
+                if attempt < retries - 1:
+                    # Exponential backoff with jitter
+                    backoff_time = (2 ** attempt) + random.uniform(0, 1)
+                    time.sleep(backoff_time)
+                else:
+                    logger.error(f"Failed to generate completion after {retries} attempts")
+                    # Return a minimal valid JSON as fallback
+                    return '{"error": "Failed to generate completion", "message": "Maximum retries exceeded"}'
+    
+    def generate_saas_profiles(self, num_profiles: int = 10) -> List[SaaSProfile]:
+        """Generate a diverse set of SaaS company profiles with improved error handling."""
+        
+        profiles = []
+        
+        # Expanded SaaS categories with modern AI/ML and other specialized categories
+        saas_categories = [
+            # Core SaaS Categories
+            "Project Management", "CRM", "Marketing Automation", "HR Software",
+            "Customer Support", "Accounting", "Business Intelligence", "Collaboration",
+            "DevOps", "Security", "E-commerce", "ERP", "Content Management",
+            
+            # AI/ML/LLM Specific
+            "LLM Development Platform", "AI Orchestration", "Prompt Engineering Tools",
+            "AI Agent Framework", "Machine Learning Operations", "Vector Database",
+            "AI Content Generation", "Computer Vision Platform", "NLP Solutions",
+            "AI Model Marketplace", "Semantic Search", "AI Development Environment",
+            "Multimodal AI Platform", "AI Workflow Automation", "GenAI Enterprise Solutions",
+            
+            # Emerging Tech SaaS
+            "Blockchain Services", "IoT Platform", "Augmented Reality", "Virtual Reality",
+            "Edge Computing", "Quantum Computing Services", "Digital Twin Platform",
+            
+            # Industry-Specific SaaS
+            "HealthTech", "FinTech", "EdTech", "LegalTech", "PropTech", "AgriTech",
+            "InsurTech", "RegTech", "CleanTech", "BioTech", "FoodTech", "RetailTech",
+            
+            # Data-Focused SaaS
+            "Data Integration", "ETL Platform", "Data Visualization", "Data Governance",
+            "Big Data Analytics", "Predictive Analytics", "Data Labeling", "Data Quality",
+            "Real-time Analytics", "Data Pipeline", "Time Series Database",
+            
+            # Specialized SaaS
+            "API Management", "Workflow Automation", "Knowledge Management", "Network Monitoring",
+            "Identity Management", "Email Marketing", "Video Conferencing", "Product Analytics",
+            "Customer Data Platform", "Event Management", "Subscription Management",
+            "Conversational AI", "Pricing Optimization", "Sales Enablement", "Revenue Operations",
+            
+            # Developer Tools
+            "Code Repository", "CI/CD Pipeline", "Testing Automation", "Microservices Platform",
+            "Serverless Computing", "API Development", "Low-Code Platform", "No-Code Platform",
+            "Database Management", "Container Orchestration", "Application Monitoring",
+            
+            # Security SaaS
+            "Endpoint Protection", "Cloud Security", "Identity Access Management",
+            "Vulnerability Management", "Threat Intelligence", "Data Loss Prevention",
+            "Security Information Management", "Privileged Access Management", "Zero Trust Security",
+            
+            # Remote Work SaaS
+            "Virtual Desktop", "Remote Team Collaboration", "Digital Workplace", "Employee Monitoring",
+            "Virtual Onboarding", "Distributed Team Management", "Workforce Analytics"
+        ]
+        
+        # Expanded industries list to match the diverse SaaS categories
+        industries = [
+            # Traditional Industries
+            "Technology", "Healthcare", "Finance", "Education", "Retail",
+            "Manufacturing", "Media", "Real Estate", "Legal", "Non-profit",
+            
+            # Expanded Technology Sectors
+            "Software Development", "Cloud Services", "Data Science", "Artificial Intelligence",
+            "Cybersecurity", "Telecommunications", "Gaming", "Digital Marketing",
+            
+            # Specific Verticals
+            "E-commerce", "Banking", "Insurance", "Pharmaceuticals", "Entertainment",
+            "Hospitality", "Transportation", "Logistics", "Construction", "Energy",
+            "Agriculture", "Automotive", "Aerospace", "Public Sector", "Professional Services",
+            
+            # Emerging Industries
+            "Renewable Energy", "Biotechnology", "Nanotechnology", "Space Technology",
+            "Smart Cities", "Sustainable Development", "Circular Economy",
+            
+            # Service Sectors
+            "Consulting", "Staffing", "Training & Development", "Research", "Marketing Services"
+        ]
+        
+        company_sizes = ["small", "medium", "large", "enterprise"]
+        
+        # Calculate effective thread count - be conservative for profile generation
+        effective_threads = min(num_profiles, 15)
+        
+        # Use ThreadPoolExecutor for parallel profile generation
+        with concurrent.futures.ThreadPoolExecutor(max_workers=effective_threads) as executor:
+            futures = []
+            
+            # Submit each profile generation task
+            for i in range(num_profiles):
+                category = random.choice(saas_categories)
+                industry = random.choice(industries)
+                size = random.choice(company_sizes)
+                
+                futures.append(executor.submit(
+                    self._generate_single_profile, i, category, industry, size
+                ))
+            
+            # Collect results as they complete
+            for future in tqdm(concurrent.futures.as_completed(futures), total=len(futures), desc="Generating SaaS profiles"):
+                try:
+                    profile = future.result()
+                    if profile:
+                        profiles.append(profile)
+                except Exception as e:
+                    logger.error(f"Error in profile generation thread: {str(e)}")
+        
+        logger.info(f"Generated {len(profiles)} SaaS profiles")
+        return profiles
+    
+    def _generate_single_profile(self, i: int, category: str, industry: str, size: str) -> Optional[SaaSProfile]:
+        """Generate a single SaaS profile."""
+        max_retries = 3
+        for attempt in range(max_retries):
+            try:
+                # Use a simpler prompt focused on just creating valid JSON
+                system_prompt = "You are a SaaS industry expert. Generate a JSON object according to the specification."
+                
+                user_prompt = f"""
+                Create a SaaS company profile for a {category} software company targeting {size} businesses in the {industry} industry.
+                
+                Return ONLY a JSON object with this exact structure:
+                {{
+                    "company_id": "saas-{i}",
+                    "company_name": "Company name",
+                    "product_name": "Product name",
+                    "product_description": "A 2-3 sentence description",
+                    "target_audience": ["Target 1", "Target 2", "Target 3"],
+                    "key_features": ["Feature 1", "Feature 2", "Feature 3", "Feature 4"],
+                    "value_propositions": ["Value prop 1", "Value prop 2", "Value prop 3"],
+                    "pricing_structure": {{
+                        "model": "Pricing model type",
+                        "tiers": [
+                            {{
+                                "name": "Tier name",
+                                "price": "Price",
+                                "features": ["Feature 1", "Feature 2"]
+                            }}
+                        ]
+                    }},
+                    "common_objections": ["Objection 1", "Objection 2", "Objection 3", "Objection 4"],
+                    "industry": "{industry}",
+                    "company_size": "{size}"
+                }}
+                
+                Ensure your response is ONLY valid JSON with no additional text, markdown code blocks, or explanations.
+                """
+                
+                content = self._generate_completion(system_prompt, user_prompt)
+                profile_json = json.loads(content)
+                
+                profile = SaaSProfile(
+                    company_id=profile_json.get("company_id", f"saas-{i}"),
+                    company_name=profile_json["company_name"],
+                    product_name=profile_json["product_name"],
+                    product_description=profile_json["product_description"],
+                    target_audience=profile_json["target_audience"],
+                    key_features=profile_json["key_features"],
+                    value_propositions=profile_json["value_propositions"],
+                    pricing_structure=profile_json["pricing_structure"],
+                    common_objections=profile_json["common_objections"],
+                    industry=profile_json["industry"],
+                    company_size=profile_json["company_size"]
+                )
+                
+                logger.info(f"Successfully generated profile {i}: {profile.company_name}")
+                return profile
+                
+            except Exception as e:
+                logger.error(f"Error attempt {attempt+1}/{max_retries} generating profile {i}: {str(e)}")
+                if attempt < max_retries - 1:
+                    # Exponential backoff
+                    backoff_time = (2 ** attempt) + random.uniform(0, 1)
+                    time.sleep(backoff_time)
+        
+        # Create a fallback profile on final attempt
+        logger.warning(f"Using fallback profile for {i}")
+        return SaaSProfile(
+            company_id=f"saas-{i}",
+            company_name=f"{category} Solutions Inc",
+            product_name=f"{category} Pro",
+            product_description=f"A {category} solution for {size} businesses in the {industry} industry.",
+            target_audience=[f"{industry} professionals", f"{size} businesses", "Department managers"],
+            key_features=["Easy setup", "Intuitive interface", "Advanced reporting", "Team collaboration"],
+            value_propositions=["Increase productivity", "Reduce costs", "Improve visibility"],
+            pricing_structure={
+                "model": "tiered",
+                "tiers": [
+                    {
+                        "name": "Basic",
+                        "price": "$10/user/month",
+                        "features": ["Core features", "Email support"]
+                    }
+                ]
+            },
+            common_objections=["Too expensive", "Complex implementation", "Training required", "Integration concerns"],
+            industry=industry,
+            company_size=size
+        )
+    
+    def _generate_customer_persona(self, profile: SaaSProfile) -> Dict:
+        """Generate a realistic customer persona based on the SaaS profile with improved diversity."""
+        max_retries = 3
+        for attempt in range(max_retries):
+            try:
+                # Select a random persona template for more diversity
+                persona_template = random.choice(self.persona_templates)
+                
+                system_prompt = """You are an expert in creating realistic customer personas. 
+                Generate a JSON object according to the specification. Make the persona feel real, with specific challenges and needs."""
+                
+                user_prompt = f"""
+                Generate a realistic customer persona for a potential {profile.product_name} customer based on:
+                
+                Company: {profile.company_name}
+                Product: {profile.product_name}
+                Description: {profile.product_description}
+                Target Audience: {", ".join(profile.target_audience)}
+                Industry: {profile.industry}
+                
+                Please base this persona loosely on the following template:
+                Template name: {persona_template["name"]}
+                Traits: {", ".join(persona_template["traits"])}
+                Communication style: {persona_template["communication_style"]}
+                Typical objections: {", ".join(persona_template["typical_objections"])}
+                
+                Return ONLY a JSON object with this exact structure:
+                {{
+                    "name": "Full customer name (realistic)",
+                    "company": "Customer's company name (specific, not generic)",
+                    "role": "Customer's specific job title",
+                    "company_size": "{profile.company_size}",
+                    "industry": "{profile.industry}",
+                    "pain_points": ["Specific pain point 1", "Specific pain point 2", "Specific pain point 3"],
+                    "needs": ["Specific need 1", "Specific need 2", "Specific need 3"],
+                    "communication_preferences": ["Email", "Phone", "Chat", etc.],
+                    "technical_expertise": "low/medium/high",
+                    "budget_sensitivity": "low/medium/high",
+                    "decision_making_authority": "none/influence/decide",
+                    "personality_traits": ["Trait 1", "Trait 2", "Trait 3"],
+                    "background": "Brief background of the customer (2-3 sentences)",
+                    "objection_style": "How they typically raise objections (direct, passive, etc.)"
+                }}
+                
+                Ensure your response is ONLY valid JSON with no additional text or markdown.
+                """
+                
+                content = self._generate_completion(system_prompt, user_prompt, temperature=0.8)
+                persona_json = json.loads(content)
+                
+                # Adding more details to each persona to enhance realism
+                persona_json["preferred_speech_patterns"] = random.sample([p["pattern"] for p in self.speech_patterns], 2)
+                persona_json["primary_need_type"] = random.choice([n["type"] for n in self.customer_needs])
+                
+                return persona_json
+                
+            except Exception as e:
+                logger.error(f"Error attempt {attempt+1}/{max_retries} generating persona: {str(e)}")
+                if attempt < max_retries - 1:
+                    # Exponential backoff
+                    backoff_time = (2 ** attempt) + random.uniform(0, 1)
+                    time.sleep(backoff_time)
+                else:
+                    # Return a default persona as fallback but with more variation
+                    logger.warning(f"Using fallback persona for {profile.company_name}")
+                    return {
+                        "name": f"{random.choice(['Alex', 'Jordan', 'Taylor', 'Sam', 'Morgan', 'Jamie'])} {random.choice(['Smith', 'Johnson', 'Wong', 'Garcia', 'Patel', 'Kim'])}",
+                        "company": f"{random.choice(['Innovative', 'Global', 'Modern', 'Premier', 'Advanced'])} {random.choice(['Systems', 'Solutions', 'Technologies', 'Enterprises', 'Group'])}",
+                        "role": random.choice(["Department Manager", "IT Director", "Operations Lead", "VP of Technology", "CFO", "CEO", "CTO"]),
+                        "company_size": profile.company_size,
+                        "industry": profile.industry,
+                        "pain_points": ["Efficiency issues", "Cost management", "Integration challenges"],
+                        "needs": ["Better reporting", "Team collaboration", "Process automation"],
+                        "communication_preferences": random.sample(["Email", "Phone", "Chat", "In-person", "Video call"], 2),
+                        "technical_expertise": random.choice(["low", "medium", "high"]),
+                        "budget_sensitivity": random.choice(["low", "medium", "high"]),
+                        "decision_making_authority": random.choice(["none", "influence", "decide"]),
+                        "personality_traits": random.sample(["analytical", "direct", "cautious", "friendly", "detail-oriented", "big-picture", "skeptical"], 3),
+                        "background": "Has been with the company for 5 years. Previously worked at a competitor.",
+                        "objection_style": random.choice(["direct", "passive-aggressive", "analytical", "price-focused"]),
+                        "preferred_speech_patterns": random.sample([p["pattern"] for p in self.speech_patterns], 2),
+                        "primary_need_type": random.choice([n["type"] for n in self.customer_needs])
+                    }
+    
+    def _generate_conversation_scenario(self, profile: SaaSProfile, customer_persona: Dict) -> Dict:
+        """Generate a diverse conversation scenario with improved complexity."""
+        max_retries = 3
+        for attempt in range(max_retries):
+            try:
+                # Select random conversation style and flow pattern for diversity
+                conversation_style = random.choice(self.conversation_styles)
+                conversation_flow = random.choice(self.conversation_flows)
+                communication_channel = random.choice(list(self.communication_channels.keys()))
+                
+                # Randomize expected outcome to ensure dataset balance
+                expected_outcome = random.choice([True, False])
+                
+                # Generate a conversion probability that matches the expected outcome
+                if expected_outcome:
+                    # For positive outcomes, higher probability
+                    expected_probability = random.uniform(0.6, 0.95)
+                else:
+                    # For negative outcomes, lower probability
+                    expected_probability = random.uniform(0.05, 0.4)
+                
+                system_prompt = """You are an expert in creating realistic sales scenarios. 
+                Generate a JSON object according to the specification. Focus on making the scenario detailed and context-rich."""
+                
+                user_prompt = f"""
+                Create a detailed sales scenario between a {profile.company_name} representative and this customer:
+                
+                {json.dumps(customer_persona, indent=2)}
+                
+                Company & Product Details:
+                - Product: {profile.product_name}
+                - Description: {profile.product_description}
+                - Key Features: {", ".join(profile.key_features)}
+                - Common Objections: {", ".join(profile.common_objections)}
+                
+                Use the following parameters to shape the scenario:
+                - Conversation style: {conversation_style}
+                - Conversation flow pattern: {conversation_flow}
+                - Communication channel: {communication_channel} ({self.communication_channels[communication_channel]["formality"]} formality, {self.communication_channels[communication_channel]["response_time"]} response time)
+                - Expected outcome: {"successful conversion" if expected_outcome else "no conversion"}
+                
+                Return ONLY a JSON object with this exact structure:
+                {{
+                    "customer_persona": {json.dumps(customer_persona)},
+                    "sales_channel": "{communication_channel}",
+                    "conversation_context": "Detailed background context for the conversation",
+                    "customer_intent": "Specific reason why the customer initiated contact",
+                    "customer_knowledge_level": "How much the customer already knows about the product or solution space",
+                    "objection_focus": ["Specific objection 1", "Specific objection 2"],
+                    "complexity": "simple/moderate/complex",
+                    "expected_outcome": {"true" if expected_outcome else "false"},
+                    "expected_conversion_probability": {expected_probability},
+                    "conversation_style": "{conversation_style}",
+                    "conversation_flow": "{conversation_flow}",
+                    "critical_moment": "The turning point in the conversation where the outcome might be decided",
+                    "time_pressure": "Whether there's urgency for a decision (none/some/high)",
+                    "external_factors": ["Factor 1", "Factor 2"] 
+                }}
+                
+                Ensure your response is ONLY valid JSON with no additional text or markdown.
+                """
+                
+                content = self._generate_completion(system_prompt, user_prompt, temperature=0.8)
+                scenario_json = json.loads(content)
+                
+                # Ensure expected_outcome is a boolean
+                if isinstance(scenario_json["expected_outcome"], str):
+                    scenario_json["expected_outcome"] = scenario_json["expected_outcome"].lower() == "true"
+                
+                # Ensure expected_conversion_probability is a float
+                if isinstance(scenario_json["expected_conversion_probability"], str):
+                    scenario_json["expected_conversion_probability"] = float(scenario_json["expected_conversion_probability"])
+                
+                return scenario_json
+                
+            except Exception as e:
+                logger.error(f"Error attempt {attempt+1}/{max_retries} generating scenario: {str(e)}")
+                if attempt < max_retries - 1:
+                    # Exponential backoff
+                    backoff_time = (2 ** attempt) + random.uniform(0, 1)
+                    time.sleep(backoff_time)
+                else:
+                    # Return a default scenario as fallback but with more variation
+                    logger.warning(f"Using fallback scenario for {profile.company_name}")
+                    
+                    # Randomize expected outcome
+                    expected_outcome = random.choice([True, False])
+                    expected_probability = 0.7 if expected_outcome else 0.3
+                    
+                    return {
+                        "customer_persona": customer_persona,
+                        "sales_channel": random.choice(list(self.communication_channels.keys())),
+                        "conversation_context": random.choice(["Initial inquiry", "Follow-up call", "Demo meeting", "Pricing discussion", "Implementation planning"]),
+                        "customer_intent": random.choice(["Exploring options", "Comparing vendors", "Addressing urgent need", "Planning future implementation", "Evaluating potential ROI"]),
+                        "customer_knowledge_level": random.choice(["minimal", "moderate", "extensive"]),
+                        "objection_focus": profile.common_objections[:2] if len(profile.common_objections) >= 2 else ["Price", "Implementation"],
+                        "complexity": random.choice(["simple", "moderate", "complex"]),
+                        "expected_outcome": expected_outcome,
+                        "expected_conversion_probability": expected_probability,
+                        "conversation_style": random.choice(self.conversation_styles),
+                        "conversation_flow": random.choice(self.conversation_flows),
+                        "critical_moment": "Discussion of pricing and ROI",
+                        "time_pressure": random.choice(["none", "some", "high"]),
+                        "external_factors": ["Budget cycle", "Competitor evaluation"]
+                    }
+    
+    def _apply_speech_patterns(self, text: str, patterns: List[str], probability: float = 0.3) -> str:
+        """Apply realistic speech patterns to make text more human."""
+        if not text or not patterns or random.random() > probability:
+            return text
+        
+        modified_text = text
+        
+        for pattern_name in patterns:
+            # Find the pattern details
+            pattern_details = next((p for p in self.speech_patterns if p["pattern"] == pattern_name), None)
+            if not pattern_details or random.random() > 0.4:  # Only apply some patterns
+                continue
+                
+            examples = pattern_details["examples"]
+            
+            if pattern_name == "filler_words":
+                # Insert filler words
+                words = modified_text.split()
+                for i in range(len(words) - 1):
+                    if random.random() < 0.1:  # 10% chance per position
+                        filler = random.choice(examples)
+                        words.insert(i + 1, filler)
+                modified_text = " ".join(words)
+                
+            elif pattern_name == "typos":
+                # Replace some words with typos
+                words = modified_text.split()
+                for i, word in enumerate(words):
+                    if len(word) > 3 and random.random() < 0.05:  # 5% chance per word
+                        # Simple typo simulation - swap two adjacent characters
+                        char_list = list(word)
+                        j = random.randint(0, len(char_list) - 2)
+                        char_list[j], char_list[j + 1] = char_list[j + 1], char_list[j]
+                        words[i] = "".join(char_list)
+                modified_text = " ".join(words)
+                
+            elif pattern_name == "over_punctuation":
+                # Add excessive punctuation
+                for punct in ["!", "?", "..."]:
+                    modified_text = modified_text.replace(f"{punct}", random.choice([f"{punct}", f"{punct}{punct}", f"{punct}{punct}{punct}"]))
+                    
+            elif pattern_name == "self_corrections":
+                # Add self-corrections
+                sentences = modified_text.split('. ')
+                if len(sentences) > 1:
+                    i = random.randint(0, len(sentences) - 1)
+                    correction = random.choice(examples)
+                    sentences[i] = f"{sentences[i]}... {correction}, {sentences[i]}"
+                modified_text = '. '.join(sentences)
+                
+            elif pattern_name == "emphasis_capital":
+                # Capitalize some words for emphasis
+                words = modified_text.split()
+                for i, word in enumerate(words):
+                    if len(word) > 3 and word.isalpha() and random.random() < 0.05:
+                        words[i] = word.upper()
+                modified_text = " ".join(words)
+        
+        return modified_text
+    
+    def _apply_channel_formatting(self, message: str, channel: str) -> str:
+        """Apply formatting specific to different communication channels."""
+        if not channel in self.communication_channels:
+            return message
+            
+        channel_format = self.communication_channels[channel]
+        modified_message = message
+        
+        # Apply channel-specific formatting
+        if channel == "email":
+            # Add email elements like signatures or formal greetings randomly
+            if random.random() < 0.3 and "sales_rep" in message:
+                signature = f"\n\nBest regards,\n[Name]\n[Company]\n[Contact Info]"
+                modified_message += signature
+                
+        elif channel == "live_chat":
+            # Add chat elements like quick responses or emojis
+            if random.random() < 0.4:
+                emojis = ["👍", "😊", "👋", "👏", "🙌", "💯", "🤔", "📊", "📈"]
+                if random.random() < 0.5:
+                    modified_message += f" {random.choice(emojis)}"
+                else:
+                    modified_message = f"{random.choice(emojis)} {modified_message}"
+                    
+        elif channel == "phone_call" or channel == "video_call":
+            # Add verbal pause indicators
+            verbal_pauses = ["*pauses*", "*brief silence*", "*thinking*"]
+            if random.random() < 0.2:
+                sentences = modified_message.split('. ')
+                if len(sentences) > 1:
+                    i = random.randint(0, len(sentences) - 1)
+                    sentences[i] = f"{sentences[i]}. {random.choice(verbal_pauses)} "
+                    modified_message = '. '.join(sentences)
+                    
+        elif channel == "sms":
+            # Shorten message and add abbreviations
+            if len(modified_message) > 100 and random.random() < 0.5:
+                # Replace some common words with abbreviations
+                abbr = {"with": "w/", "without": "w/o", "thanks": "thx", "please": "pls", 
+                        "about": "abt", "meeting": "mtg", "tomorrow": "tmrw"}
+                for word, abbr_word in abbr.items():
+                    if random.random() < 0.5:
+                        modified_message = re.sub(r'\b' + word + r'\b', abbr_word, modified_message, flags=re.IGNORECASE)
+        
+        return modified_message
+    
+    def _generate_conversation(self, profile: SaaSProfile, scenario: Dict) -> Dict:
+        """Generate a complete human-like sales conversation with non-linear flows and realistic patterns."""
+        max_retries = 3
+        for attempt in range(max_retries):
+            try:
+                # Extract key scenario parameters for controlling the conversation
+                channel = scenario.get("sales_channel", "email")
+                conversation_style = scenario.get("conversation_style", "direct_professional")
+                conversation_flow = scenario.get("conversation_flow", "standard_linear")
+                expected_outcome = scenario.get("expected_outcome", True)
+                
+                # Prepare persona information
+                persona = scenario.get("customer_persona", {})
+                persona_name = persona.get("name", "Customer")
+                speech_patterns = persona.get("preferred_speech_patterns", 
+                                           random.sample([p["pattern"] for p in self.speech_patterns], 2))
+                
+                # Prepare template parameters based on scenario
+                min_messages = 8  # Minimum messages for a meaningful conversation
+                max_messages = 18  # Maximum to keep conversations reasonably sized
+                
+                # Adjust message count based on complexity
+                if scenario.get("complexity") == "simple":
+                    target_messages = random.randint(8, 12)
+                elif scenario.get("complexity") == "complex":
+                    target_messages = random.randint(14, 18)
+                else:  # moderate
+                    target_messages = random.randint(10, 14)
+                
+                # Further customize conversation parameters based on flow type
+                flow_params = {}
+                if conversation_flow == "multiple_objection_loops":
+                    flow_params["objections_to_raise"] = min(len(profile.common_objections), 3)
+                    flow_params["objection_resolution_probability"] = 0.7 if expected_outcome else 0.4
+                elif conversation_flow == "subject_switching":
+                    flow_params["topic_switches"] = random.randint(2, 4)
+                    flow_params["switch_probability"] = 0.3
+                elif conversation_flow == "interrupted_followup":
+                    flow_params["interruption_point"] = random.randint(3, 5)
+                    flow_params["followup_delay"] = "a few hours" if channel in ["email", "chat"] else "a week"
+                
+                # Create a conversation prompt that allows for natural, human-like dialogue
+                system_prompt = f"""You are an expert in creating realistic sales conversations. 
+                Generate a JSON object with a natural, imperfect conversation between a sales rep and customer.
+                
+                Important guidelines:
+                - The conversation should feel HUMAN and NATURAL, not scripted or perfect
+                - Include human elements: hesitations, typos, interruptions, tangents, repetition, in a millennial style
+                - Do NOT open with cheesy greetings like "Hi X, this is Y from Z company" unless natural for the context
+                - The customer should sometimes be unclear, ambiguous, or send incomplete thoughts
+                - People sometimes send multiple messages in a row
+                - Sometimes include pauses or pacing in the conversation
+                - Not every objection needs to be perfectly addressed
+                - Use natural language that matches the {channel} communication channel"""
+                
+                user_prompt = f"""
+                Generate a realistic {channel} sales conversation between a {profile.company_name} representative and a customer with:
+                
+                CUSTOMER DETAILS:
+                Name: {persona_name}
+                Company: {persona.get('company', 'Company')}
+                Role: {persona.get('role', 'Role')}
+                Industry: {persona.get('industry', profile.industry)}
+                Current pain points: {', '.join(persona.get('pain_points', ['Unspecified']))}
+                Communication style: {persona.get('objection_style', 'direct')}
+                Tech expertise: {persona.get('technical_expertise', 'medium')}
+                
+                CONVERSATION CONTEXT:
+                Channel: {channel}
+                Context: {scenario.get('conversation_context', 'Initial inquiry')}
+                Customer intent: {scenario.get('customer_intent', 'Exploring options')}
+                Flow pattern: {conversation_flow}
+                Style: {conversation_style}
+                Expected outcome: {"conversion" if expected_outcome else "no conversion"}
+                Primary objections: {', '.join(scenario.get('objection_focus', profile.common_objections[:2]))}
+                
+                PRODUCT DETAILS:
+                Product: {profile.product_name}
+                Description: {profile.product_description}
+                Key features: {', '.join(profile.key_features)}
+                Value props: {', '.join(profile.value_propositions)}
+                Pricing: {profile.pricing_structure.get('model', 'tiered')} - {profile.pricing_structure.get('tiers', [{}])[0].get('price', '$X/month')}
+                
+                Return ONLY a JSON object with this exact structure:
+                {{
+                    "messages": [
+                        {{"speaker": "customer/sales_rep", "message": "message text"}}
+                    ],
+                    "outcome": {"true" if expected_outcome else "false"},
+                    "key_objections": ["Specific objection 1", "Specific objection 2"],
+                    "key_value_props_mentioned": ["Value prop 1", "Value prop 2"],
+                    "customer_engagement_level": 0.X,
+                    "sales_rep_effectiveness": 0.X,
+                    "conversation_length": X,
+                    "conversion_probability_at_turn": {{"0": 0.X, "1": 0.X}}
+                }}
+                
+                Include {target_messages} messages in the conversation. The conversion_probability_at_turn should show
+                how the probability changes throughout the conversation.
+                
+                IMPORTANT FORMATTING:
+                - Make messages sound like real humans typing/talking with millennial style - include occasional typos, hesitations, filler words
+                - Format the conversation appropriately for a {channel} conversation
+                - Messages should vary in length - some short, some longer
+                - Avoid perfect grammar and complete sentences when natural
+                - Use contractions, abbreviations, and casual language where appropriate
+                - For the customer, occasionally use multiple messages in a row
+                - Use natural greetings appropriate to the channel, not formulaic ones
+                """
+                
+                content = self._generate_completion(system_prompt, user_prompt, temperature=0.9)
+                conversation_json = json.loads(content)
+                
+                # Apply additional human-like formatting and speech patterns to each message
+                for i, message in enumerate(conversation_json["messages"]):
+                    speaker = message.get("speaker", "")
+                    message_text = message.get("message", "")
+                    
+                    # Apply appropriate speech patterns based on speaker
+                    if speaker == "customer":
+                        # Apply customer persona speech patterns
+                        message_text = self._apply_speech_patterns(message_text, speech_patterns, 0.5)
+                    else:
+                        # Apply general speech patterns for sales rep
+                        rep_patterns = random.sample([p["pattern"] for p in self.speech_patterns], 2)
+                        message_text = self._apply_speech_patterns(message_text, rep_patterns, 0.3)
+                    
+                    # Apply channel-specific formatting
+                    message_text = self._apply_channel_formatting(message_text, channel)
+                    
+                    # Apply non-linearity based on conversation flow
+                    if conversation_flow == "subject_switching" and random.random() < flow_params.get("switch_probability", 0):
+                        # Add topic switch indicators
+                        switches = ["By the way", "Actually, I also wanted to ask", "On another note", 
+                                   "While we're talking", "That reminds me", "Before I forget"]
+                        message_text += f" {random.choice(switches)}, {random.choice(profile.key_features)}?"
+                    
+                    # Update the message
+                    conversation_json["messages"][i]["message"] = message_text
+                
+                # Add markers for interrupted conversations if applicable
+                if conversation_flow == "interrupted_followup":
+                    interrupt_point = flow_params.get("interruption_point", 4)
+                    if len(conversation_json["messages"]) > interrupt_point + 2:
+                        # Add interruption marker
+                        interrupt_idx = min(interrupt_point, len(conversation_json["messages"]) - 3)
+                        
+                        # Add time passage marker
+                        time_marker = {"speaker": "system", "message": f"--- {flow_params.get('followup_delay', 'some time')} later ---"}
+                        conversation_json["messages"].insert(interrupt_idx + 1, time_marker)
+                        
+                        # Add follow-up message from sales rep
+                        followup = {"speaker": "sales_rep", "message": self._apply_speech_patterns(
+                            f"Hi {persona_name}, I wanted to follow up on our previous conversation about {profile.product_name}. Have you had a chance to think more about it?",
+                            random.sample([p["pattern"] for p in self.speech_patterns], 2)
+                        )}
+                        conversation_json["messages"].insert(interrupt_idx + 2, followup)
+                
+                # Ensure outcome is a boolean
+                if isinstance(conversation_json["outcome"], str):
+                    conversation_json["outcome"] = conversation_json["outcome"].lower() == "true"
+                
+                # Ensure numeric values are floats/ints
+                if isinstance(conversation_json["customer_engagement_level"], str):
+                    conversation_json["customer_engagement_level"] = float(conversation_json["customer_engagement_level"])
+                
+                if isinstance(conversation_json["sales_rep_effectiveness"], str):
+                    conversation_json["sales_rep_effectiveness"] = float(conversation_json["sales_rep_effectiveness"])
+                
+                if isinstance(conversation_json["conversation_length"], str):
+                    conversation_json["conversation_length"] = int(conversation_json["conversation_length"])
+                
+                # Ensure probability trajectory has numeric keys and values
+                probability_trajectory = {}
+                for k, v in conversation_json["conversion_probability_at_turn"].items():
+                    if isinstance(v, str):
+                        v = float(v)
+                    probability_trajectory[int(k)] = float(v)
+                conversation_json["conversion_probability_at_turn"] = probability_trajectory
+                
+                # Add scenario data to the conversation for reference
+                conversation_json["scenario"] = {
+                    "channel": channel,
+                    "conversation_style": conversation_style,
+                    "conversation_flow": conversation_flow
+                }
+                
+                return conversation_json
+                
+            except Exception as e:
+                logger.error(f"Error attempt {attempt+1}/{max_retries} generating conversation: {str(e)}")
+                if attempt < max_retries - 1:
+                    # Exponential backoff
+                    backoff_time = (2 ** attempt) + random.uniform(0, 1)
+                    time.sleep(backoff_time)
+                else:
+                    # Return a minimal conversation as fallback
+                    logger.warning(f"Using fallback conversation for {profile.company_name}")
+                    expected_outcome = scenario.get("expected_outcome", True)
+                    
+                    # Create more varied fallback conversations
+                    starters = [
+                        "Hey, I've been looking around for a solution like yours.",
+                        "I need some help with my current processes.",
+                        "Been hearing about your product, got some questions.",
+                        "Our team needs something better than what we're using.",
+                        "Quick question about your pricing."
+                    ]
+                    
+                    responses = [
+                        f"Thanks for reaching out! What specific challenges are you facing?",
+                        f"Happy to help! What's your current setup like?",
+                        f"I'd be glad to answer questions. What would you like to know?",
+                        f"Sure thing. What's not working with your current solution?",
+                        f"Of course, I can walk you through our pricing. What's your use case?"
+                    ]
+                    
+                    # Generate a basic conversation with some variation
+                    messages = [
+                        {"speaker": "customer", "message": random.choice(starters)},
+                        {"speaker": "sales_rep", "message": random.choice(responses)},
+                        {"speaker": "customer", "message": f"We're looking for a {profile.product_name} solution. What makes your product different?"},
+                        {"speaker": "sales_rep", "message": f"Our {profile.product_name} stands out because of {profile.value_propositions[0] if profile.value_propositions else 'its features'}. Many of our customers appreciate this."}
+                    ]
+                    
+                    # Add a bit more variation based on expected outcome
+                    if expected_outcome:
+                        messages.append({"speaker": "customer", "message": "That sounds interesting. Can you tell me more about your pricing?"})
+                        messages.append({"speaker": "sales_rep", "message": f"Our pricing starts at {profile.pricing_structure.get('tiers', [{}])[0].get('price', '$X/month')}. Would you like to schedule a demo?"})
+                        messages.append({"speaker": "customer", "message": "Yes, that would be helpful. Let's set something up for next week."})
+                    else:
+                        messages.append({"speaker": "customer", "message": "That's interesting, but I'm not sure it fits our needs right now. Let me think about it."})
+                        messages.append({"speaker": "sales_rep", "message": f"I understand. Would it be helpful if I sent you some more information about our {profile.product_name}?"})
+                        messages.append({"speaker": "customer", "message": "Maybe later. We're still evaluating other options at the moment."})
+                    
+                    # Create probability trajectory
+                    probability_trajectory = {}
+                    for i in range(len(messages)):
+                        if expected_outcome:
+                            prob = min(0.5 + i * 0.07, 0.9)
+                        else:
+                            prob = max(0.5 - i * 0.07, 0.1)
+                        probability_trajectory[i] = prob
+                    
+                    return {
+                        "messages": messages,
+                        "outcome": expected_outcome,
+                        "key_objections": scenario.get("objection_focus", ["Price", "Implementation"]),
+                        "key_value_props_mentioned": profile.value_propositions[:2] if len(profile.value_propositions) >= 2 else ["Value", "Efficiency"],
+                        "customer_engagement_level": 0.7 if expected_outcome else 0.4,
+                        "sales_rep_effectiveness": 0.6 if expected_outcome else 0.5,
+                        "conversation_length": len(messages),
+                        "conversion_probability_at_turn": probability_trajectory,
+                        "scenario": {
+                            "channel": scenario.get("sales_channel", "email"),
+                            "conversation_style": scenario.get("conversation_style", "direct_professional"),
+                            "conversation_flow": scenario.get("conversation_flow", "standard_linear")
+                        }
+                    }
+    
+    def _generate_full_conversation(self, profile_idx: int, conversation_idx: int) -> Dict:
+        """Generate a complete conversation from profile to final conversation data."""
+        with self.profile_lock:
+            profile = self.profiles[profile_idx]
+        
+        try:
+            # Generate customer persona with more richness and diversity
+            persona = self._generate_customer_persona(profile)
+            
+            # Generate conversation scenario with more variables
+            scenario = self._generate_conversation_scenario(profile, persona)
+            
+            # Generate complete conversation with more natural and varied flow
+            conversation_data = self._generate_conversation(profile, scenario)
+            
+            # Get conversation embeddings
+            full_text = " ".join([msg["message"] for msg in conversation_data["messages"] if msg.get("speaker") != "system"])
+            embeddings = self._get_embedding(full_text)
+            
+            # Create row data
+            row_data = {
+                'company_id': profile.company_id,
+                'company_name': profile.company_name,
+                'product_name': profile.product_name,
+                'product_type': profile.industry,
+                'conversation_id': f"{profile.company_id}-conv-{conversation_idx}",
+                'scenario': json.dumps(scenario),
+                'conversation': json.dumps(conversation_data["messages"]),
+                'full_text': full_text,
+                'outcome': 1 if conversation_data["outcome"] else 0,
+                'conversation_length': conversation_data["conversation_length"],
+                'customer_engagement': conversation_data["customer_engagement_level"],
+                'sales_effectiveness': conversation_data["sales_rep_effectiveness"],
+                'probability_trajectory': json.dumps(conversation_data["conversion_probability_at_turn"]),
+                'conversation_style': scenario.get("conversation_style", "direct_professional"),
+                'conversation_flow': scenario.get("conversation_flow", "standard_linear"),
+                'communication_channel': scenario.get("sales_channel", "email")
+            }
+            
+            # Add embeddings
+            for j, embed_value in enumerate(embeddings):
+                row_data[f'embedding_{j}'] = embed_value
+            
+            # Update counter
+            with self.total_counter_lock:
+                self.total_generated += 1
+                total = self.total_generated
+            
+            if total % 100 == 0:
+                logger.info(f"Generated {total} conversations so far")
+            
+            return row_data
+        
+        except Exception as e:
+            logger.error(f"Error generating conversation {conversation_idx} for profile {profile_idx}: {str(e)}")
+            return None
+    
+    def _worker(self, task_queue, result_batch_size=5):
+        """Worker function that processes tasks from the queue with immediate saving"""
+        batch = []
+        
+        while True:
+            try:
+                # Get task from queue with timeout
+                task = task_queue.get(timeout=5)
+                
+                # Check for termination signal
+                if task is None:
+                    # Submit any remaining items in batch
+                    if batch:
+                        self.writer.write_rows(batch)
+                    task_queue.task_done()
+                    break
+                
+                # Add a random delay to stagger requests
+                time.sleep(random.uniform(0.1, 0.5))
+                
+                # Process the task
+                profile_idx, conv_idx = task
+                row_data = self._generate_full_conversation(profile_idx, conv_idx)
+                
+                if row_data:
+                    batch.append(row_data)
+                    
+                    # Write batch when it reaches the threshold - use smaller batch size (5 instead of 10)
+                    if len(batch) >= result_batch_size:
+                        self.writer.write_rows(batch)
+                        batch = []  # Clear batch after writing
+                
+                # Mark task as done
+                task_queue.task_done()
+                
+            except queue.Empty:
+                # Check if there are any items in the batch to write
+                if batch:
+                    self.writer.write_rows(batch)
+                    batch = []
+                continue
+            except Exception as e:
+                logger.error(f"Error in worker thread: {str(e)}")
+                try:
+                    # Write any collected data before potentially crashing
+                    if batch:
+                        self.writer.write_rows(batch)
+                        batch = []
+                    task_queue.task_done()
+                except:
+                    pass
+    
+    def generate_dataset(
+        self, 
+        num_conversations: int = 100000, 
+        num_profiles: int = 20,
+        output_path: str = "enhanced_saas_sales_conversations.csv",
+        num_threads: int = 25,
+        result_batch_size: int = 5
+    ) -> str:
+        """
+        Generate a complete dataset of diverse sales conversations using multithreading.
+        
+        Args:
+            num_conversations: Total number of conversations to generate
+            num_profiles: Number of unique SaaS profiles to use
+            output_path: Path to save the CSV dataset
+            num_threads: Number of worker threads to use
+            result_batch_size: Number of items to batch before writing to CSV
+            
+        Returns:
+            Path to the generated dataset
+        """
+        logger.info(f"Starting enhanced dataset generation: {num_conversations} conversations using {num_profiles} profiles with {num_threads} threads")
+        
+        start_time = time.time()
+        
+        # Generate SaaS profiles first (this is done in parallel already)
+        self.profiles = self.generate_saas_profiles(num_profiles)
+        
+        # Initialize counters
+        self.total_generated = 0
+        self.api_calls = 0
+        self.retry_count = 0
+        self.last_retry_time = 0
+        
+        # Define columns for the dataset
+        columns = ['company_id', 'company_name', 'product_name', 'product_type', 
+                'conversation_id', 'scenario', 'conversation', 'full_text',
+                'outcome', 'conversation_length', 'customer_engagement', 
+                'sales_effectiveness', 'probability_trajectory', 
+                'conversation_style', 'conversation_flow', 'communication_channel']
+        
+        # Add embedding columns
+        for i in range(3072):
+            columns.append(f'embedding_{i}')
+        
+        # Initialize the writer
+        self.writer = ResultWriter(output_path, columns)
+        
+        # Create task queue and populate it
+        task_queue = queue.Queue()
+        
+        # Calculate conversations per profile
+        conversations_per_profile = num_conversations // len(self.profiles)
+        remaining = num_conversations % len(self.profiles)
+        
+        # Create tasks (profile_idx, conversation_idx) with more even distribution
+        for profile_idx in range(len(self.profiles)):
+            profile_conversations = conversations_per_profile
+            if profile_idx < remaining:
+                profile_conversations += 1
+                
+            for conv_idx in range(profile_conversations):
+                task_queue.put((profile_idx, conv_idx))
+        
+        # Calculate effective thread count based on rate limits
+        # Each conversation makes about 4 API calls (persona, scenario, conversation, embedding)
+        # We want to use only about 50% of the rate limit for better stability
+        requests_per_conversation = 4
+        requests_per_minute_per_thread = 10  # Conservative estimate
+        target_threads = (SAFE_RATE_LIMIT_RPM) / requests_per_minute_per_thread
+        
+        # Determine effective thread count
+        effective_threads = min(
+            num_threads,  # User requested threads
+            int(target_threads),  # Rate-limit-based threads
+            30,  # Hard limit for stability
+            task_queue.qsize()  # No more threads than tasks
+        )
+        
+        logger.info(f"Using {effective_threads} threads based on rate limit of {RATE_LIMIT_RPM} RPM")
+        
+        # Start worker threads
+        workers = []
+        for _ in range(effective_threads):
+            worker = threading.Thread(target=self._worker, args=(task_queue, result_batch_size))
+            worker.daemon = True
+            worker.start()
+            workers.append(worker)
+        
+        # Progress indicator and statistics tracker
+        last_count = 0
+        last_stats_time = time.time()
+        with tqdm(total=num_conversations, desc="Generating conversations") as pbar:
+            while self.total_generated < num_conversations:
+                time.sleep(1)  # Update progress every second
+                
+                with self.total_counter_lock:
+                    current_count = self.total_generated
+                    
+                # Update progress bar
+                delta = current_count - last_count
+                if delta > 0:
+                    pbar.update(delta)
+                    last_count = current_count
+                
+                # Print statistics every 60 seconds
+                now = time.time()
+                if now - last_stats_time > 60:
+                    # Calculate API call rate
+                    with self.api_call_lock:
+                        api_calls = self.api_calls
+                    
+                    with self.retry_lock:
+                        retries = self.retry_count
+                    
+                    elapsed = now - start_time
+                    calls_per_minute = (api_calls / elapsed) * 60
+                    
+                    logger.info(f"Statistics - Generated: {current_count}, API calls: {api_calls} " +
+                               f"({calls_per_minute:.1f}/min), Retries: {retries}")
+                    
+                    last_stats_time = now
+                
+                # Check if we've reached our target or if the queue is empty
+                if current_count >= num_conversations or task_queue.empty():
+                    break
+        
+        # Signal workers to terminate
+        for _ in range(effective_threads):
+            task_queue.put(None)
+        
+        # Wait for all workers to finish (with timeout)
+        for worker in workers:
+            worker.join(timeout=30)
+        
+        end_time = time.time()
+        duration = end_time - start_time
+        
+        # Calculate final statistics
+        with self.api_call_lock:
+            total_api_calls = self.api_calls
+        
+        with self.retry_lock:
+            total_retries = self.retry_count
+        
+        logger.info(f"Dataset generation complete. Statistics:")
+        logger.info(f"- Total conversations: {self.total_generated}")
+        logger.info(f"- Total API calls: {total_api_calls}")
+        logger.info(f"- Total retries: {total_retries}")
+        logger.info(f"- Generation took {duration:.2f} seconds ({self.total_generated / duration:.2f} conversations/second)")
+        logger.info(f"- API call rate: {(total_api_calls / duration) * 60:.1f} calls/minute")
+        
+        return output_path
+
+def main():
+    """Main function to run the enhanced dataset generator."""
+    import argparse
+    
+    parser = argparse.ArgumentParser(description="Enhanced SaaS Sales Dataset Generator")
+    parser.add_argument("--num_conversations", type=int, default=100000,
+                       help="Number of conversations to generate")
+    parser.add_argument("--num_profiles", type=int, default=20,
+                       help="Number of SaaS profiles to generate")
+    parser.add_argument("--output_path", type=str, default="enhanced_saas_sales_conversations.csv",
+                       help="Path to save the generated dataset")
+    parser.add_argument("--num_threads", type=int, default=25,
+                       help="Number of worker threads to use (will be rate-limited)")
+    parser.add_argument("--rate_limit", type=int, default=2500,
+                      help="API rate limit in requests per minute")
+    parser.add_argument("--batch_size", type=int, default=5,
+                       help="Number of conversations to batch before writing to CSV")
+    
+    args = parser.parse_args()
+    
+    # Update rate limit if provided
+    global RATE_LIMIT_RPM, SAFE_RATE_LIMIT_RPM, RATE_LIMIT_RPS, MIN_REQUEST_INTERVAL
+    if args.rate_limit:
+        RATE_LIMIT_RPM = args.rate_limit
+        SAFE_RATE_LIMIT_RPM = int(RATE_LIMIT_RPM * 0.6)
+        RATE_LIMIT_RPS = SAFE_RATE_LIMIT_RPM / 60
+        MIN_REQUEST_INTERVAL = 1.0 / RATE_LIMIT_RPS
+    
+    # Initialize generator
+    generator = EnhancedSaaSDatasetGenerator()
+    
+    # Generate dataset
+    output_path = generator.generate_dataset(
+        num_conversations=args.num_conversations,
+        num_profiles=args.num_profiles,
+        output_path=args.output_path,
+        num_threads=args.num_threads,
+        result_batch_size=args.batch_size
+    )
+    
+    print(f"\nEnhanced dataset generation complete!")
+    print(f"Dataset saved to: {output_path}")
+
+if __name__ == "__main__":
+    main()
+#python enhanced_saas_generator.py --num_conversations 100000 --num_profiles 20 --output_path custom_dataset.csv --num_threads 15 --rate_limit 2000 --batch_size 10