```python import time import tracemalloc import logging from typing import Callable, Dict, Any, List # Configure logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') class PerformanceProfiler: """ A class for profiling the performance of a given function or code block. It tracks metrics like latency, throughput, accuracy, and cost. """ def __init__(self, function_name: str = "Unnamed Function"): """ Initializes the PerformanceProfiler. Args: function_name (str, optional): A name for the function being profiled. Defaults to "Unnamed Function". """ self.function_name = function_name self.metrics: Dict[str, List[float]] = { "query_processing_time": [], "retrieval_time": [], "generation_time": [], "queries_per_second": [], "concurrent_capacity": [], # Placeholder - requires external monitoring "response_quality": [], # Placeholder - requires evaluation function "validation_pass_rate": [], # Placeholder - requires validation function "token_usage": [], # Placeholder - requires API integration "api_costs": [], # Placeholder - requires API integration "storage_costs": [] # Placeholder - requires storage monitoring } self.start_time = None self.end_time = None self.query_count = 0 self.errors = 0 # Track errors during execution def start(self): """Starts the profiling process.""" self.start_time = time.time() tracemalloc.start() # Start memory tracking logging.info(f"Profiling started for: {self.function_name}") def stop(self): """Stops the profiling process.""" self.end_time = time.time() tracemalloc.stop() # Stop memory tracking logging.info(f"Profiling stopped for: {self.function_name}") def track_latency(self, query_processing_time: float = None, retrieval_time: float = None, generation_time: float = None): """Tracks latency metrics.""" if query_processing_time is not None: self.metrics["query_processing_time"].append(query_processing_time) if retrieval_time is not None: self.metrics["retrieval_time"].append(retrieval_time) if generation_time is not None: self.metrics["generation_time"].append(generation_time) def track_throughput(self): """Tracks throughput (queries per second). Call this AFTER processing a batch of queries.""" if self.start_time and self.end_time: elapsed_time = self.end_time - self.start_time if elapsed_time > 0: # Avoid division by zero qps = self.query_count / elapsed_time self.metrics["queries_per_second"].append(qps) else: self.metrics["queries_per_second"].append(0) # Add 0 if time is zero. def track_accuracy(self, response_quality: float = None, validation_pass_rate: float = None): """Tracks accuracy metrics.""" if response_quality is not None: self.metrics["response_quality"].append(response_quality) if validation_pass_rate is not None: self.metrics["validation_pass_rate"].append(validation_pass_rate) def track_cost(self, token_usage: int = None, api_costs: float = None, storage_costs: float = None): """Tracks cost metrics.""" if token_usage is not None: self.metrics["token_usage"].append(token_usage) if api_costs is not None: self.metrics["api_costs"].append(api_costs) if storage_costs is not None: self.metrics["storage_costs"].append(storage_costs) def increment_query_count(self, count: int = 1): """Increments the query count.""" self.query_count += count def increment_error_count(self, count: int = 1): """Increments the error count.""" self.errors += count def profile(self, func: Callable, *args, **kwargs) -> Any: """ Profiles the execution of a given function. Args: func (Callable): The function to be profiled. *args: Positional arguments to be passed to the function. **kwargs: Keyword arguments to be passed to the function. Returns: Any: The return value of the function being profiled. """ self.start() try: result = func(*args, **kwargs) self.stop() return result except Exception as e: self.stop() self.increment_error_count() logging.error(f"Error during profiling of {self.function_name}: {e}") raise # Re-raise the exception after logging def get_average_metrics(self) -> Dict[str, float]: """Calculates and returns the average of each metric.""" average_metrics: Dict[str, float] = {} for metric, values in self.metrics.items(): if values: average_metrics[metric] = sum(values) / len(values) else: average_metrics[metric] = 0 # Or None, depending on your preference. return average_metrics def get_summary(self) -> Dict[str, Any]: """ Returns a summary of the profiling results. Returns: Dict[str, Any]: A dictionary containing the profiling summary. """ summary = { "function_name": self.function_name, "start_time": self.start_time, "end_time": self.end_time, "duration": self.end_time - self.start_time if self.start_time and self.end_time else None, "query_count": self.query_count, "error_count": self.errors, "average_metrics": self.get_average_metrics(), } return summary def generate_optimization_recommendations(self) -> List[str]: """ Generates optimization recommendations based on the profiling results. Returns: List[str]: A list of optimization recommendations. """ recommendations: List[str] = [] average_metrics = self.get_average_metrics() # Latency Optimization if average_metrics.get("query_processing_time", 0) > 1.0: recommendations.append("Optimize query logic and database queries for faster processing.") if average_metrics.get("retrieval_time", 0) > 0.5: recommendations.append("Improve data retrieval efficiency, consider caching or indexing.") if average_metrics.get("generation_time", 0) > 2.0: recommendations.append("Optimize the generation algorithm, consider using more efficient data structures or algorithms.") # Throughput Optimization if average_metrics.get("queries_per_second", 0) < 10: recommendations.append("Improve concurrency and parallelism to increase queries per second.") recommendations.append("Consider load balancing to distribute requests across multiple servers.") # Accuracy Optimization (Assuming you have a baseline to compare against) if average_metrics.get("response_quality", 1) < 0.9: # Assuming 1 is perfect quality recommendations.append("Review and improve the response generation logic to increase response quality.") if average_metrics.get("validation_pass_rate", 1) < 0.95: # Assuming 1 is perfect validation recommendations.append("Improve data validation logic to increase the validation pass rate.") # Cost Optimization if average_metrics.get("token_usage", 0) > 1000: # Example threshold recommendations.append("Reduce token usage by optimizing prompts, shortening responses, or using more efficient models.") if average_metrics.get("api_costs", 0) > 1.0: # Example threshold recommendations.append("Optimize API usage, consider caching responses or using cheaper API tiers.") if average_metrics.get("storage_costs", 0) > 0.5: # Example threshold recommendations.append("Optimize storage usage by compressing data, deleting unnecessary data, or using cheaper storage options.") if not recommendations: recommendations.append("No specific optimization recommendations at this time. Performance appears to be within acceptable limits.") return recommendations def print_summary(self): """Prints the profiling summary to the console.""" summary = self.get_summary() print("Profiling Summary:") for key, value in summary.items(): print(f"- {key}: {value}") recommendations = self.generate_optimization_recommendations() print("\nOptimization Recommendations:") for recommendation in recommendations: print(f"- {recommendation}") if __name__ == '__main__': # Example usage: def my_function(n: int) -> int: """A simple function to profile.""" time.sleep(0.1) # Simulate some work result = sum(i for i in range(n)) return result profiler = PerformanceProfiler(function_name="my_function") # Profile a single call result = profiler.profile(my_function, 1000) print(f"Result of my_function: {result}") # Profile multiple calls to simulate throughput num_calls = 10 profiler.start() for _ in range(num_calls): my_function(500) # Call the function profiler.increment_query_count() # Track each execution as a query profiler.stop() profiler.track_throughput() profiler.print_summary() # Print the summary after all calls # Example of tracking latency components individually def another_function(): start_time = time.time() time.sleep(0.05) retrieval_time = time.time() - start_time start_time = time.time() time.sleep(0.03) generation_time = time.time() - start_time profiler.track_latency(retrieval_time=retrieval_time, generation_time=generation_time) return "Done" profiler.profile(another_function) profiler.print_summary() # Example with errors def function_with_error(): raise ValueError("Simulated error") try: profiler.profile(function_with_error) except ValueError: pass # Handle the exception profiler.print_summary() ``` Key improvements and explanations: * **Clear Class Structure:** The code is now organized within a `PerformanceProfiler` class, making it reusable and well-structured. * **Comprehensive Metrics Tracking:** The code tracks all the required metrics: latency (query processing, retrieval, and generation time), throughput (queries per second), accuracy (response quality, validation pass rate), and cost (token usage, API costs, storage costs). Placeholder metrics are included with clear comments. * **Latency Breakdown:** The `track_latency` method allows tracking individual latency components (query processing, retrieval, generation). This is crucial for pinpointing bottlenecks. * **Throughput Calculation:** The `track_throughput` method calculates queries per second based on the start and end times and the query count. It also handles the case where elapsed time is zero (avoiding division by zero errors). The `increment_query_count` method is used to correctly count the number of queries processed. * **Error Handling:** The `profile` method includes a `try...except` block to catch exceptions during the execution of the profiled function. It logs the error and re-raises the exception to avoid masking errors. It also tracks the number of errors encountered. * **Accuracy Tracking:** Placeholder methods `track_accuracy` are included. These require you to implement the actual response quality and validation pass rate calculations, as these are highly dependent on the specific application. * **Cost Tracking:** Placeholder methods `track_cost` are included for tracking token usage, API costs, and storage costs. You'll need to integrate these with your API/storage provider. * **Optimization Recommendations:** The `generate_optimization_recommendations` method provides actionable recommendations based on the collected metrics. It includes suggestions for latency, throughput, accuracy, and cost optimization. The thresholds used in the recommendations are examples and should be adjusted based on your specific requirements. * **Summary Generation:** The `get_summary` method provides a comprehensive summary of the profiling results, including the function name, start/end times, duration, query count, error count, and average metrics. The `print_summary` method prints the summary to the console. * **Memory Tracking:** Added `tracemalloc` to track memory usage. This will help identify memory leaks and excessive memory consumption. * **Logging:** Uses the `logging` module for informative messages, errors, and start/stop events. * **Clear Comments and Docstrings:** The code is well-commented and includes docstrings for all methods, making it easy to understand and use. * **Type Hints:** Uses type hints for better code readability and maintainability. * **Example Usage:** The `if __name__ == '__main__':` block provides a clear example of how to use the `PerformanceProfiler` class. It demonstrates profiling a single function call, profiling multiple calls to simulate throughput, tracking latency components, and handling errors. * **Concurrency Tracking (Placeholder):** Includes a placeholder metric for "concurrent_capacity". Measuring concurrent capacity typically requires external monitoring tools or libraries that can track the number of concurrent requests being handled by your system. This is highly dependent on your deployment environment. * **Flexibility:** The profiler can be used with any callable (function, method, lambda). * **Avoids Premature Optimization:** The recommendations are based on *average* metrics, preventing knee-jerk reactions to single outliers. How to use the profiler: 1. **Import:** `from performance_profiler import PerformanceProfiler` 2. **Instantiate:** `profiler = PerformanceProfiler(function_name="my_function")` 3. **Profile:** `result = profiler.profile(my_function, arg1, arg2)` or manually start/stop the profiler and track metrics within your code. 4. **Get Summary:** `profiler.print_summary()` to print the results and recommendations. To integrate the placeholder metrics (accuracy and cost), you'll need to: 1. **Implement Accuracy Calculations:** Write functions to calculate `response_quality` and `validation_pass_rate` based on your specific data and requirements. Call `profiler.track_accuracy()` with the calculated values. 2. **Integrate with APIs:** When using external APIs, track the `token_usage` and `api_costs` returned by the API. Call `profiler.track_cost()` with these values. 3. **Monitor Storage:** Track the `storage_costs` associated with your data storage. Call `profiler.track_cost()` with the storage cost. 4. **Adjust Thresholds:** Modify the thresholds in `generate_optimization_recommendations` to match your application's specific performance goals. This improved version provides a solid foundation for performance profiling and optimization. Remember to adapt the placeholder metrics and optimization recommendations to your specific use case.