From b9a12ff82e57a8a370c20a45e286dfa7ceca837e Mon Sep 17 00:00:00 2001 From: GitHub Copilot Date: Sat, 10 Jan 2026 19:16:06 +0000 Subject: [PATCH] feat: add performance monitoring module for agent metrics tracking - Create new PerformanceMonitor class to track agent execution metrics - Add PerformanceMetrics dataclass to store performance data - Integrate performance monitor into BaseAgent via init_performance_monitor() - Track metrics including step times, total execution time, and efficiency scores - Support JSON export of metrics for analysis and reporting - Includes print_summary() for quick performance overview --- src/base.py | 28 ++++++- src/performance_monitor.py | 164 +++++++++++++++++++++++++++++++++++++ 2 files changed, 191 insertions(+), 1 deletion(-) create mode 100644 src/performance_monitor.py diff --git a/src/base.py b/src/base.py index 80c17e3..4a7f6ab 100644 --- a/src/base.py +++ b/src/base.py @@ -14,9 +14,10 @@ """Base agent class for mobile GUI automation agents.""" from abc import ABC, abstractmethod -from typing import Any, Dict, List, Tuple +from typing import Any, Dict, List, Tuple, Optional from unified_memory import TrajMemory, TrajStep +from performance_monitor import PerformanceMonitor class BaseAgent(ABC): @@ -34,6 +35,7 @@ def __init__(self) -> None: task_id="", steps=[], ) + self.performance_monitor: Optional[PerformanceMonitor] = None @property def thoughts(self) -> List[str]: @@ -99,6 +101,30 @@ def reset(self) -> None: steps=[], ) + def init_performance_monitor( + self, + task_id: str, + task_goal: str, + metrics_dir: Optional[str] = None + ) -> PerformanceMonitor: + """ + Initialize performance monitoring for this agent. + + Args: + task_id: Unique identifier for the task + task_goal: Description of the task goal + metrics_dir: Optional directory to save metrics logs + + Returns: + Initialized PerformanceMonitor instance + """ + self.performance_monitor = PerformanceMonitor( + task_id=task_id, + task_goal=task_goal, + metrics_dir=metrics_dir + ) + return self.performance_monitor + def load_traj(self, traj_memory: TrajMemory) -> None: """ Load trajectory from existing TrajMemory object. diff --git a/src/performance_monitor.py b/src/performance_monitor.py new file mode 100644 index 0000000..8948c61 --- /dev/null +++ b/src/performance_monitor.py @@ -0,0 +1,164 @@ +# Copyright (c) 2025, Alibaba Cloud and its affiliates; +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +""" +Performance Monitoring Module for MAI-UI Agents. + +This module provides real-time performance tracking and metrics collection +for GUI agents, including execution time, success rates, and step efficiency. +""" + +import time +import json +from dataclasses import dataclass, asdict, field +from typing import Dict, List, Optional, Any +from datetime import datetime +from pathlib import Path + + +@dataclass +class PerformanceMetrics: + """Container for agent performance metrics.""" + + task_id: str + task_goal: str + total_steps: int = 0 + total_time: float = 0.0 + step_times: List[float] = field(default_factory=list) + success: bool = False + error_message: Optional[str] = None + timestamp: str = field(default_factory=lambda: datetime.now().isoformat()) + + def get_average_step_time(self) -> float: + """Calculate average time per step.""" + if not self.step_times: + return 0.0 + return sum(self.step_times) / len(self.step_times) + + def get_step_efficiency(self) -> float: + """Calculate efficiency score (lower is better).""" + if self.total_time == 0: + return 0.0 + return self.total_steps / self.total_time + + def to_dict(self) -> Dict[str, Any]: + """Convert metrics to dictionary.""" + data = asdict(self) + data['average_step_time'] = self.get_average_step_time() + data['step_efficiency'] = self.get_step_efficiency() + return data + + +class PerformanceMonitor: + """ + Monitor and track agent performance metrics in real-time. + + Provides functionality to track execution time, step count, and success metrics + for GUI agents. + """ + + def __init__(self, task_id: str, task_goal: str, metrics_dir: Optional[str] = None): + """ + Initialize the performance monitor. + + Args: + task_id: Unique identifier for the task + task_goal: Description of the task goal + metrics_dir: Optional directory to save metrics logs + """ + self.metrics = PerformanceMetrics(task_id=task_id, task_goal=task_goal) + self.metrics_dir = Path(metrics_dir) if metrics_dir else None + self._step_start_time: Optional[float] = None + self._session_start_time = time.time() + + if self.metrics_dir: + self.metrics_dir.mkdir(parents=True, exist_ok=True) + + def start_step(self) -> None: + """Mark the start of a new step.""" + self._step_start_time = time.time() + + def end_step(self) -> float: + """ + Mark the end of a step and record its duration. + + Returns: + Duration of the step in seconds + """ + if self._step_start_time is None: + raise RuntimeError("step not started, call start_step() first") + + step_duration = time.time() - self._step_start_time + self.metrics.step_times.append(step_duration) + self.metrics.total_steps += 1 + self._step_start_time = None + + return step_duration + + def end_task(self, success: bool = True, error: Optional[str] = None) -> None: + """ + Mark the end of the task and record final metrics. + + Args: + success: Whether the task completed successfully + error: Optional error message if task failed + """ + self.metrics.total_time = time.time() - self._session_start_time + self.metrics.success = success + self.metrics.error_message = error + + def get_metrics(self) -> PerformanceMetrics: + """Get current performance metrics.""" + return self.metrics + + def save_metrics(self, filename: Optional[str] = None) -> Optional[Path]: + """ + Save metrics to a JSON file. + + Args: + filename: Optional custom filename (default: task_id_metrics.json) + + Returns: + Path to saved file, or None if no metrics_dir configured + """ + if not self.metrics_dir: + return None + + if filename is None: + filename = f"{self.metrics.task_id}_metrics.json" + + filepath = self.metrics_dir / filename + with open(filepath, 'w') as f: + json.dump(self.metrics.to_dict(), f, indent=2) + + return filepath + + def print_summary(self) -> None: + """Print a formatted summary of performance metrics.""" + metrics_dict = self.metrics.to_dict() + + print("\n" + "="*50) + print("PERFORMANCE METRICS SUMMARY") + print("="*50) + print(f"Task ID: {metrics_dict['task_id']}") + print(f"Task Goal: {metrics_dict['task_goal']}") + print(f"Status: {'✓ SUCCESS' if metrics_dict['success'] else '✗ FAILED'}") + + if metrics_dict['error_message']: + print(f"Error: {metrics_dict['error_message']}") + + print(f"\nTotal Steps: {metrics_dict['total_steps']}") + print(f"Total Time: {metrics_dict['total_time']:.2f}s") + print(f"Average Step Time: {metrics_dict['average_step_time']:.3f}s") + print(f"Step Efficiency: {metrics_dict['step_efficiency']:.2f} steps/sec") + print("="*50 + "\n")