142 lines
5.9 KiB
Python
142 lines
5.9 KiB
Python
# progressTracker.py
|
|
# Progress tracking for adaptive React mode
|
|
|
|
import logging
|
|
from typing import Dict, Any, List
|
|
from datetime import datetime, timezone
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class ProgressTracker:
|
|
"""Tracks what has been accomplished and what's still needed"""
|
|
|
|
def __init__(self):
|
|
self.completedObjectives = []
|
|
self.partialAchievements = []
|
|
self.failedAttempts = []
|
|
self.learningInsights = []
|
|
self.currentPhase = "planning"
|
|
|
|
def updateProgress(self, result: Any, validation: Dict[str, Any], intent: Dict[str, Any]):
|
|
"""Updates progress tracking based on action result"""
|
|
try:
|
|
overallSuccess = validation.get('overallSuccess', False)
|
|
qualityScore = validation.get('qualityScore', 0)
|
|
improvementSuggestions = validation.get('improvementSuggestions', [])
|
|
|
|
if overallSuccess and qualityScore > 0.7:
|
|
# Successful completion
|
|
self.completedObjectives.append({
|
|
"objective": intent.get('primaryGoal', 'Unknown'),
|
|
"achievement": f"Quality score: {qualityScore:.2f}",
|
|
"qualityScore": qualityScore,
|
|
"timestamp": datetime.now(timezone.utc).timestamp()
|
|
})
|
|
self.currentPhase = "completed"
|
|
logger.info(f"Objective completed: {intent.get('primaryGoal', 'Unknown')}")
|
|
|
|
elif qualityScore > 0.3:
|
|
# Partial achievement
|
|
self.partialAchievements.append({
|
|
"objective": intent.get('primaryGoal', 'Unknown'),
|
|
"partialAchievement": f"Quality score: {qualityScore:.2f}",
|
|
"missingParts": improvementSuggestions,
|
|
"timestamp": datetime.now(timezone.utc).timestamp()
|
|
})
|
|
self.currentPhase = "partial"
|
|
logger.info(f"Partial achievement: {intent.get('primaryGoal', 'Unknown')}")
|
|
|
|
else:
|
|
# Failed attempt
|
|
self.failedAttempts.append({
|
|
"objective": intent.get('primaryGoal', 'Unknown'),
|
|
"failureReason": f"Quality score: {qualityScore:.2f}",
|
|
"learningOpportunity": improvementSuggestions,
|
|
"timestamp": datetime.now(timezone.utc).timestamp()
|
|
})
|
|
self.currentPhase = "failed"
|
|
logger.info(f"Failed attempt: {intent.get('primaryGoal', 'Unknown')}")
|
|
|
|
# Extract learning insights
|
|
if improvementSuggestions:
|
|
for suggestion in improvementSuggestions:
|
|
if suggestion not in self.learningInsights:
|
|
self.learningInsights.append(suggestion)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error updating progress: {str(e)}")
|
|
|
|
def getCurrentProgress(self) -> Dict[str, Any]:
|
|
"""Gets current progress state"""
|
|
return {
|
|
"completedObjectives": self.completedObjectives,
|
|
"partialAchievements": self.partialAchievements,
|
|
"failedAttempts": self.failedAttempts,
|
|
"learningInsights": self.learningInsights,
|
|
"currentPhase": self.currentPhase,
|
|
"nextActionsSuggested": self._getNextActionSuggestions()
|
|
}
|
|
|
|
def shouldContinue(self, progress: Dict[str, Any], validation: Dict[str, Any]) -> bool:
|
|
"""Determines if the task should continue"""
|
|
try:
|
|
# If we have completed objectives, don't continue
|
|
if progress.get('completedObjectives'):
|
|
return False
|
|
|
|
# If we have too many failed attempts, don't continue
|
|
if len(progress.get('failedAttempts', [])) >= 3:
|
|
return False
|
|
|
|
# If validation shows success, don't continue
|
|
if validation.get('overallSuccess', False):
|
|
return False
|
|
|
|
# Otherwise, continue
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error checking if should continue: {str(e)}")
|
|
return True # Default to continue on error
|
|
|
|
def _getNextActionSuggestions(self) -> List[str]:
|
|
"""Suggests next actions based on progress"""
|
|
suggestions = []
|
|
|
|
# If we have failed attempts, suggest avoiding those actions
|
|
if self.failedAttempts:
|
|
suggestions.append("Avoid actions that have failed before")
|
|
|
|
# If we have partial achievements, suggest building on them
|
|
if self.partialAchievements:
|
|
suggestions.append("Build on partial achievements")
|
|
|
|
# If we have learning insights, suggest applying them
|
|
if self.learningInsights:
|
|
suggestions.extend(self.learningInsights[:3]) # Top 3 insights
|
|
|
|
# Default suggestions
|
|
if not suggestions:
|
|
suggestions.append("Try a different approach")
|
|
suggestions.append("Focus on user intent")
|
|
|
|
return suggestions
|
|
|
|
def getProgressSummary(self) -> Dict[str, Any]:
|
|
"""Gets a summary of progress"""
|
|
return {
|
|
"totalCompleted": len(self.completedObjectives),
|
|
"totalPartial": len(self.partialAchievements),
|
|
"totalFailed": len(self.failedAttempts),
|
|
"totalInsights": len(self.learningInsights),
|
|
"currentPhase": self.currentPhase,
|
|
"successRate": len(self.completedObjectives) / max(len(self.completedObjectives) + len(self.failedAttempts), 1)
|
|
}
|
|
|
|
def reset(self):
|
|
"""Resets progress tracking"""
|
|
self.completedObjectives = []
|
|
self.partialAchievements = []
|
|
self.failedAttempts = []
|
|
self.learningInsights = []
|
|
self.currentPhase = "planning"
|