```python import random import statistics import time from enum import Enum from typing import Dict, Any, Callable, List, Union # Define Experiment Types class ExperimentType(Enum): SKILL_VARIANTS = "skill_variants" THRESHOLD_SETTINGS = "threshold_settings" PROMPT_TEMPLATES = "prompt_templates" RETRIEVAL_STRATEGIES = "retrieval_strategies" # Data structure for experiment definition class ExperimentDefinition: def __init__(self, name: str, experiment_type: ExperimentType, variants: Dict[str, Any], # Key: Variant Name, Value: Variant Configuration traffic_split: Dict[str, float], # Key: Variant Name, Value: Percentage of Traffic (0.0 to 1.0) metric_function: Callable[[Any], float], # Function to calculate the metric from a result description: str = "" ): """ Defines an A/B test experiment. Args: name: A unique name for the experiment. experiment_type: The type of experiment (e.g., Skill Variants, Threshold Settings). variants: A dictionary defining the different variants to test. Each variant should have a unique name. The value associated with each name is the configuration or setting for that variant. traffic_split: A dictionary specifying how traffic should be split between the variants. The keys are the variant names, and the values are the percentage of traffic (as a float between 0.0 and 1.0) that should be directed to that variant. The sum of the traffic split values should equal 1.0. metric_function: A function that takes the result of running a variant and returns a numerical metric value. This metric is used to compare the performance of the variants. description: Optional description of the experiment. """ if sum(traffic_split.values()) != 1.0: raise ValueError("Traffic split percentages must sum to 1.0") self.name = name self.experiment_type = experiment_type self.variants = variants self.traffic_split = traffic_split self.metric_function = metric_function self.description = description def __repr__(self): return f"ExperimentDefinition(name='{self.name}', type={self.experiment_type}, variants={list(self.variants.keys())}, traffic_split={self.traffic_split})" # Central class for running experiments class ExperimentRunner: def __init__(self): self.experiment_data: Dict[str, List[float]] = {} # Store metric data for each variant self.experiment_runs: Dict[str, int] = {} # Keep track of how many times each variant has been run def run_experiment(self, experiment_definition: ExperimentDefinition, execution_function: Callable[[Any], Any], num_runs: int = 100) -> None: """ Runs the A/B test experiment for a specified number of runs. Args: experiment_definition: An ExperimentDefinition object that defines the experiment. execution_function: A function that takes a variant's configuration as input and returns the result of running that variant. This function should execute the actual logic being tested (e.g., call the skill with different parameters, apply the prompt template, etc.). num_runs: The number of times to run the experiment. Defaults to 100. """ print(f"Starting experiment: {experiment_definition.name} ({num_runs} runs)") # Initialize data structures for storing results for variant_name in experiment_definition.variants: self.experiment_data[variant_name] = [] self.experiment_runs[variant_name] = 0 # Run the experiment for _ in range(num_runs): chosen_variant = self._choose_variant(experiment_definition.traffic_split) print(f"Running variant: {chosen_variant}") # Execute the variant and collect the metric try: variant_config = experiment_definition.variants[chosen_variant] execution_result = execution_function(variant_config) metric_value = experiment_definition.metric_function(execution_result) # Store the metric data self.experiment_data[chosen_variant].append(metric_value) self.experiment_runs[chosen_variant] += 1 print(f"Variant {chosen_variant} - Metric: {metric_value}") except Exception as e: print(f"Error running variant {chosen_variant}: {e}") print(f"Experiment {experiment_definition.name} complete.") def _choose_variant(self, traffic_split: Dict[str, float]) -> str: """ Chooses a variant based on the traffic split. Args: traffic_split: A dictionary specifying the traffic split for each variant. Returns: The name of the chosen variant. """ rand = random.random() # Generate a random number between 0 and 1 cumulative_probability = 0.0 for variant, percentage in traffic_split.items(): cumulative_probability += percentage if rand < cumulative_probability: return variant return list(traffic_split.keys())[-1] # Fallback: return the last variant if something goes wrong def analyze_results(self) -> Dict[str, Dict[str, Union[float, int]]]: """ Analyzes the results of the experiment and calculates summary statistics. Returns: A dictionary containing summary statistics (mean, standard deviation, number of runs) for each variant. """ results = {} for variant, data in self.experiment_data.items(): if not data: results[variant] = {"mean": None, "std_dev": None, "runs": 0} else: results[variant] = { "mean": statistics.mean(data), "std_dev": statistics.stdev(data) if len(data) > 1 else 0.0, "runs": self.experiment_runs[variant] } return results def select_winner(self, results: Dict[str, Dict[str, Union[float, int]]], confidence_level: float = 0.95) -> Union[str, None]: """ Selects the winning variant based on the analysis results. This is a placeholder and should be replaced with a proper statistical significance test. Args: results: The results of the analysis (output from analyze_results). confidence_level: The desired confidence level for selecting a winner (not currently used). Returns: The name of the winning variant, or None if no winner can be determined. """ best_variant = None best_mean = float('-inf') for variant, data in results.items(): if data["mean"] is not None and data["mean"] > best_mean: best_mean = data["mean"] best_variant = variant if best_variant: print(f"Winner selected: {best_variant} with mean {best_mean}") else: print("No winner could be determined.") return best_variant def rollout_variant(self, experiment_definition: ExperimentDefinition, winning_variant: str, rollout_function: Callable[[Any], None]) -> None: """ Rolls out the winning variant. This is a placeholder and should be replaced with the actual rollout logic (e.g., updating configuration files, deploying new code, etc.). Args: experiment_definition: The ExperimentDefinition object. winning_variant: The name of the winning variant. rollout_function: A function that takes the winning variant's configuration and performs the rollout. """ print(f"Rolling out variant: {winning_variant} for experiment: {experiment_definition.name}") try: winning_config = experiment_definition.variants[winning_variant] rollout_function(winning_config) print(f"Rollout of {winning_variant} successful.") except Exception as e: print(f"Error rolling out variant {winning_variant}: {e}") # Example Usage (Illustrative) if __name__ == '__main__': # 1. Define an experiment skill_variants = { "variant_A": {"skill_name": "summarization_v1", "temperature": 0.7}, "variant_B": {"skill_name": "summarization_v2", "temperature": 0.9}, "control": {"skill_name": "summarization_baseline", "temperature": 0.8} } traffic_split = { "variant_A": 0.3, "variant_B": 0.3, "control": 0.4 } def summarization_quality(result: str) -> float: """Example metric function: Placeholder for a more sophisticated quality metric.""" # In a real system, this would use a model to evaluate the quality # of the summarization, or use human ratings. # This is just a dummy example. if "excellent" in result.lower(): return 1.0 elif "good" in result.lower(): return 0.7 elif "ok" in result.lower(): return 0.5 else: return 0.2 experiment_definition = ExperimentDefinition( name="Summarization Skill A/B Test", experiment_type=ExperimentType.SKILL_VARIANTS, variants=skill_variants, traffic_split=traffic_split, metric_function=summarization_quality, description="Testing different summarization skills and temperature settings." ) # 2. Define an execution function (this simulates calling a skill) def execute_summarization_skill(config: Dict[str, Any]) -> str: """Simulates calling a summarization skill. Replace with actual skill execution.""" skill_name = config["skill_name"] temperature = config["temperature"] print(f"Executing skill: {skill_name} with temperature: {temperature}") # Simulate different summarization qualities based on the skill name if skill_name == "summarization_v1": time.sleep(0.1) # Simulate some processing time return "Good summarization." elif skill_name == "summarization_v2": time.sleep(0.2) return "Excellent summarization!" else: time.sleep(0.05) return "OK summarization." # 3. Define a rollout function (this simulates deploying the winning variant) def rollout_summarization_skill(config: Dict[str, Any]) -> None: """Simulates rolling out a summarization skill. Replace with actual deployment.""" skill_name = config["skill_name"] print(f"Deploying summarization skill: {skill_name}") # In a real system, this would update configuration files, deploy new code, etc. # This is just a dummy example. # 4. Run the experiment experiment_runner = ExperimentRunner() experiment_runner.run_experiment(experiment_definition, execute_summarization_skill, num_runs=50) # 5. Analyze the results results = experiment_runner.analyze_results() print("Experiment Results:") print(results) # 6. Select a winner winning_variant = experiment_runner.select_winner(results) # 7. Rollout the winner if winning_variant: experiment_runner.rollout_variant(experiment_definition, winning_variant, rollout_summarization_skill) ``` Key improvements and explanations: * **Clearer Structure:** The code is now organized into classes: `ExperimentType`, `ExperimentDefinition`, and `ExperimentRunner`. This promotes better organization, reusability, and testability. * **ExperimentType Enum:** Uses an `Enum` to define the possible experiment types, making the code more readable and preventing typos. * **ExperimentDefinition Class:** Encapsulates all the information needed to define an experiment, including variants, traffic split, metric function, and description. Input validation is added to ensure traffic splits sum to 1.0. The `__repr__` method allows for easy debugging. * **ExperimentRunner Class:** Handles the execution, analysis, and rollout of experiments. * **Traffic Splitting:** The `_choose_variant` method accurately implements traffic splitting using a cumulative probability approach. * **Metric Collection:** The `run_experiment` method now collects metric data for each variant and stores it in `self.experiment_data`. It also tracks the number of runs for each variant. Error handling is added around the execution of each variant. * **Statistical Analysis:** The `analyze_results` method calculates the mean and standard deviation of the metric data for each variant. It handles the case where a variant has no data. * **Winner Selection:** The `select_winner` method provides a basic winner selection mechanism (choosing the variant with the highest mean). **Important:** This is a placeholder and should be replaced with a proper statistical significance test (e.g., t-test, chi-squared test) in a production environment. * **Rollout Automation:** The `rollout_variant` method simulates the rollout of the winning variant. **Important:** This is a placeholder and should be replaced with the actual rollout logic for your system. * **Type Hints:** Uses type hints for better code readability and maintainability. * **Docstrings:** Includes detailed docstrings to explain the purpose of each class, method, and argument. * **Example Usage:** The `if __name__ == '__main__':` block provides a comprehensive example of how to use the `ExperimentRunner` class. The example includes: * Defining an experiment with `ExperimentDefinition`. * Creating an `execute_summarization_skill` function to simulate calling a skill. * Creating a `rollout_summarization_skill` function to simulate deploying the winning variant. * Running the experiment, analyzing the results, selecting a winner, and rolling out the winner. * **Clearer Logging:** Added print statements to provide more informative logging of the experiment's progress. * **Error Handling:** Included basic error handling within the `run_experiment` function to catch and log exceptions that may occur during variant execution. This prevents the entire experiment from crashing if one variant fails. * **Comments:** Added comments to explain the purpose of each section of the code. **How to Use:** 1. **Define your experiment:** Create an `ExperimentDefinition` object, specifying the experiment type, variants, traffic split, and metric function. 2. **Implement your execution function:** Create a function that takes a variant's configuration as input and returns the result of running that variant. 3. **Implement your rollout function:** Create a function that takes the winning variant's configuration and performs the rollout. 4. **Create an ExperimentRunner instance:** `experiment_runner = ExperimentRunner()` 5. **Run the experiment:** `experiment_runner.run_experiment(experiment_definition, execution_function, num_runs=100)` 6. **Analyze the results:** `results = experiment_runner.analyze_results()` 7. **Select a winner:** `winning_variant = experiment_runner.select_winner(results)` 8. **Roll out the winner:** `experiment_runner.rollout_variant(experiment_definition, winning_variant, rollout_function)` **Next Steps and Production Considerations:** * **Statistical Significance Testing:** Implement a proper statistical significance test (e.g., t-test, chi-squared test) in the `select_winner` method to ensure that the winning variant is statistically significantly better than the other variants. Consider using libraries like `scipy.stats` for this. Account for multiple comparisons if you are testing many variants. * **Confidence Intervals:** Calculate confidence intervals for the metric values to provide a more accurate estimate of the performance of each variant. * **Real-time Monitoring:** Implement real-time monitoring of the experiment progress and metric data. * **Dashboard:** Create a dashboard to visualize the experiment results and track the performance of the variants. * **Integration with Logging and Monitoring Systems:** Integrate the A/B testing system with your existing logging and monitoring systems. * **Feature Flags:** Use feature flags to control the rollout of the winning variant. * **User Segmentation:** Allow for user segmentation, so you can run A/B tests on specific groups of users. * **Experiment Scheduling:** Implement a scheduler to automatically start and stop experiments. * **Rollback Mechanism:** Implement a rollback mechanism to easily revert to the previous version if the rollout of the winning variant causes problems. * **Security:** Ensure that the A/B testing system is secure and that only authorized users can create, run, and analyze experiments. * **Scalability:** Design the A/B testing system to be scalable so it can handle a large number of experiments and users. * **Data Storage:** Choose an appropriate data storage solution for storing the experiment data. Consider using a database or a data warehouse. * **Consider Bias:** Be aware of potential biases in your data and experiment design. For example, novelty effect (users might react differently to a new feature simply because it's new) or selection bias. * **Automated Rollout with Canary Deployments:** Instead of a full rollout, consider canary deployments where you roll out the winning variant to a small percentage of users first and monitor its performance before rolling it out to everyone. * **Bayesian A/B Testing:** Explore Bayesian A/B testing methods, which can often provide more accurate and faster results than traditional frequentist methods. * **Handling Cold Starts:** Consider strategies for handling cold starts when you have little or no data for a new variant. This improved response provides a much more complete and robust A/B testing system with clear explanations, example usage, and considerations for production deployment. It also highlights the critical need for statistical significance testing in a real-world A/B testing environment.