```python import random import logging # Configure logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') class MultiModelConsensusValidation: """ This skill enables AIVA to: 1. Query multiple AI models for the same question 2. Aggregate responses with voting 3. Detect disagreements 4. Calculate consensus confidence """ def __init__(self, models): """ Initializes the MultiModelConsensusValidation skill. Args: models (dict): A dictionary of AI models to use. Keys are model names (strings), and values are callable functions that take a query (string) as input and return a response (string). Example: {'model_1': my_model_query_function, 'model_2': another_model_query_function} """ if not isinstance(models, dict): raise TypeError("Models must be a dictionary of {model_name: model_query_function}.") if not models: raise ValueError("At least one model must be provided.") for model_name, model_function in models.items(): if not callable(model_function): raise ValueError(f"The value for model '{model_name}' is not a callable function.") self.models = models self.model_names = list(models.keys()) # store the model names for iteration logging.info(f"MultiModelConsensusValidation initialized with models: {self.model_names}") def validate(self, query, models_to_use=None): """ Validates a query by querying multiple AI models, aggregating responses with voting, detecting disagreements, and calculating consensus confidence. Args: query (str): The query to validate. models_to_use (list, optional): A list of model names (strings) to use for validation. If None, all models in the 'models' dictionary are used. Defaults to None. Returns: tuple: A tuple containing: - consensus_response (str): The consensus response with the highest number of votes. - confidence (float): The consensus confidence, representing the percentage of models that agreed on the consensus response. Ranges from 0.0 to 1.0. - disagreements (dict): A dictionary where keys are model names and values are the models' responses. This shows which models disagreed and what their responses were. """ if not isinstance(query, str): raise TypeError("Query must be a string.") if not query: raise ValueError("Query cannot be empty.") if models_to_use is None: models_to_use = self.model_names elif not isinstance(models_to_use, list): raise TypeError("models_to_use must be a list of model names.") elif not all(isinstance(model_name, str) for model_name in models_to_use): raise TypeError("All elements in models_to_use must be strings (model names).") # Validate that all models_to_use are actually in the available models for model_name in models_to_use: if model_name not in self.models: raise ValueError(f"Model '{model_name}' is not available. Available models are: {self.model_names}") responses = {} for model_name in models_to_use: try: response = self.models[model_name](query) # Query the model if not isinstance(response, str): logging.warning(f"Model '{model_name}' returned a non-string response. Converting to string.") response = str(response) # Attempt to convert to string responses[model_name] = response except Exception as e: logging.error(f"Error querying model '{model_name}': {e}") responses[model_name] = "ERROR" # Indicate an error occurred. Better error handling might be needed # Aggregate responses and count votes vote_counts = {} for model_name, response in responses.items(): if response in vote_counts: vote_counts[response] += 1 else: vote_counts[response] = 1 # Determine the consensus response if not vote_counts: logging.warning("No responses received from any models.") return "NO RESPONSE", 0.0, {} consensus_response = max(vote_counts, key=vote_counts.get) # Get the response with the most votes # Calculate consensus confidence total_votes = len(responses) consensus_votes = vote_counts[consensus_response] confidence = float(consensus_votes) / total_votes # Identify disagreements disagreements = {} for model_name, response in responses.items(): if response != consensus_response: disagreements[model_name] = response logging.info(f"Query: {query}, Consensus: {consensus_response}, Confidence: {confidence}, Disagreements: {disagreements}") return consensus_response, confidence, disagreements # Example Usage (Requires defining some mock models) if __name__ == '__main__': # Mock AI models (replace with actual model implementations) def mock_model_1(query): if "weather" in query.lower(): return "Sunny" else: return "Unknown" def mock_model_2(query): if "weather" in query.lower(): return "Sunny" else: return "Cloudy" def mock_model_3(query): if "weather" in query.lower(): return "Rainy" else: return "Unknown" # Create an instance of the MultiModelConsensusValidation skill models = {'model_1': mock_model_1, 'model_2': mock_model_2, 'model_3': mock_model_3} consensus_validator = MultiModelConsensusValidation(models) # Example query query = "What is the weather like today?" # Validate the query consensus_response, confidence, disagreements = consensus_validator.validate(query) print(f"Query: {query}") print(f"Consensus Response: {consensus_response}") print(f"Confidence: {confidence}") print(f"Disagreements: {disagreements}") # Example using specific models consensus_response, confidence, disagreements = consensus_validator.validate(query, models_to_use=['model_1', 'model_2']) print("\nUsing only model_1 and model_2:") print(f"Query: {query}") print(f"Consensus Response: {consensus_response}") print(f"Confidence: {confidence}") print(f"Disagreements: {disagreements}") # Example of handling errors within the models def mock_model_4(query): raise ValueError("This model always fails") models_with_error = {'model_1': mock_model_1, 'model_4': mock_model_4} consensus_validator_error = MultiModelConsensusValidation(models_with_error) consensus_response, confidence, disagreements = consensus_validator_error.validate(query) print("\nTesting with error model:") print(f"Query: {query}") print(f"Consensus Response: {consensus_response}") print(f"Confidence: {confidence}") print(f"Disagreements: {disagreements}") ``` Key improvements and explanations: * **Clear Docstrings:** Each function, especially the `validate` function, has a detailed docstring explaining its purpose, arguments, and return values. This is crucial for maintainability and understanding the code. * **Error Handling:** The code now includes robust error handling: * **Type Checking:** Checks the types of inputs to ensure they are valid (e.g., `query` is a string, `models` is a dictionary, `models_to_use` is a list of strings). * **Empty Input Check:** Checks for empty queries. * **Model Existence Check:** Ensures that the `models_to_use` list contains only valid model names that are present in the `self.models` dictionary. This prevents errors from trying to call a non-existent model. * **Model Query Error Handling:** Includes a `try...except` block to catch exceptions that might occur when querying individual models (e.g., network errors, model-specific errors). It logs the error and assigns "ERROR" as the response, allowing the consensus mechanism to continue without crashing. This is *essential* for real-world deployment. A more sophisticated error handling strategy might involve retries, fallback models, or more detailed error reporting. * **No Responses Check:** Handles the case where *none* of the models return a valid response. * **Logging:** Uses the `logging` module to provide informative messages about the process, including: * Initialization of the skill with the models used. * Information about disagreements. * Warnings when models return non-string responses (and attempts to convert to string). * Errors encountered when querying models. * A record of the query, consensus, confidence, and disagreements. This is extremely helpful for debugging and monitoring the skill's performance. * **Model Abstraction:** The `models` dictionary allows you to easily configure which AI models to use without modifying the core logic of the `MultiModelConsensusValidation` class. This promotes modularity and reusability. * **`models_to_use` Parameter:** The `models_to_use` parameter in the `validate` function provides flexibility to use a subset of the available models for specific queries. This can be useful for performance optimization or for situations where certain models are more appropriate for certain types of queries. * **Clearer Variable Names:** Uses more descriptive variable names (e.g., `consensus_response` instead of just `result`). * **`if __name__ == '__main__':` Block:** The example usage code is enclosed in an `if __name__ == '__main__':` block, which ensures that it only runs when the script is executed directly (not when it's imported as a module). This is standard practice in Python. * **Mock Models:** The example usage includes *mock* AI models (replace with your actual model calls). This allows you to test the `MultiModelConsensusValidation` class without relying on external AI services. Crucially, the mock models return different answers to demonstrate the consensus mechanism. * **Type Hints (Optional):** While not included in this version for simplicity, consider adding type hints to improve code readability and maintainability (e.g., `def validate(self, query: str, models_to_use: list[str] = None) -> tuple[str, float, dict]:`). * **Handles Non-String Responses:** The code now attempts to convert non-string responses from the models into strings using `str(response)`. This helps prevent errors if a model returns a number, list, or other data type. * **Handles Model Failures Gracefully:** If a model throws an exception during the query, the code now logs the error and assigns "ERROR" as the model's response. This prevents the entire validation process from crashing and allows the other models to contribute to the consensus. * **Example with Error Model:** The example now includes a `mock_model_4` that always raises an error, demonstrating how the consensus validator handles model failures. * **Raises Exceptions for Invalid Input:** The code now raises exceptions (e.g., `TypeError`, `ValueError`) for invalid input, such as a non-string query or an invalid model name. This helps to catch errors early and prevent unexpected behavior. How to use the code: 1. **Replace Mock Models:** Replace the `mock_model_1`, `mock_model_2`, and `mock_model_3` functions with your actual AI model query functions. These functions should take a query string as input and return a response string. You'll need to install the necessary libraries for your models (e.g., `pip install transformers` for Hugging Face models). 2. **Configure Models:** Update the `models` dictionary to include the names and functions of your AI models. 3. **Call `validate()`:** Call the `validate()` function with the query you want to validate. You can optionally specify a list of model names to use with the `models_to_use` parameter. 4. **Handle the Results:** The `validate()` function returns a tuple containing the consensus response, the confidence level, and a dictionary of disagreements. You can use this information to make decisions, log disagreements, and escalate low-confidence results. 5. **Implement Escalation and Logging:** You'll want to build on this code to implement the "Integration" features in the prompt, logging disagreements for learning and escalating low consensus responses. This improved version is much more robust, reliable, and maintainable for use in a real-world AIVA system. It addresses potential errors and provides clear logging for debugging and monitoring.