| """ |
| Enhanced Agent Learning with Feedback |
| |
| Integrates user feedback into agent confidence scoring and learning. |
| Provides feedback-weighted confidence adjustments and learning signals. |
| |
| Usage: |
| from core.agent_learning_enhanced import AgentLearningEnhanced |
| |
| learning = AgentLearningEnhanced(db) |
| |
| # Adjust confidence based on feedback |
| new_confidence = learning.adjust_confidence_with_feedback( |
| agent_id="agent-1", |
| feedback=feedback_obj |
| ) |
| |
| # Get learning signals from feedback |
| signals = learning.get_learning_signals("agent-1", days=30) |
| """ |
|
|
| from datetime import datetime, timedelta, timezone |
| import logging |
| import json |
| from typing import Any, Dict, List, Optional |
| import uuid |
| from sqlalchemy.orm import Session |
|
|
| from core.agent_world_model import AgentExperience, WorldModelService |
| from core.models import AgentExecution, AgentFeedback, AgentRegistry, CognitiveExperience, AgentLearning |
| from core.continuous_learning_service import ContinuousLearningService |
|
|
| logger = logging.getLogger(__name__) |
|
|
|
|
| class AgentLearningEnhanced: |
| """ |
| Enhanced learning service with feedback integration. |
| |
| Incorporates user feedback (thumbs up/down, ratings, corrections) |
| into agent confidence scoring and world model learning. |
| """ |
|
|
| def __init__(self, db: Session): |
| """ |
| Initialize enhanced learning service. |
| |
| Args: |
| db: Database session |
| """ |
| self.db = db |
| self.world_model = WorldModelService() |
| self.continuous_learning = ContinuousLearningService(db) |
|
|
| def adjust_confidence_with_feedback( |
| self, |
| agent_id: str, |
| feedback: AgentFeedback, |
| current_confidence: float |
| ) -> float: |
| """ |
| Adjust agent confidence based on user feedback. |
| |
| Feedback weights: |
| - Thumbs up: +0.05 |
| - Thumbs down: -0.05 |
| - 5-star rating: +0.10 |
| - 4-star rating: +0.05 |
| - 3-star rating: 0.00 |
| - 2-star rating: -0.05 |
| - 1-star rating: -0.10 |
| - Correction: -0.03 (indicates mistake) |
| |
| Args: |
| agent_id: ID of the agent |
| feedback: Feedback object |
| current_confidence: Current confidence score |
| |
| Returns: |
| Adjusted confidence score (0.0 to 1.0) |
| """ |
| adjustment = 0.0 |
|
|
| |
| if feedback.thumbs_up_down is True: |
| adjustment += 0.05 |
| elif feedback.thumbs_up_down is False: |
| adjustment -= 0.05 |
|
|
| |
| if feedback.rating is not None: |
| rating_weights = { |
| 1: -0.10, |
| 2: -0.05, |
| 3: 0.00, |
| 4: 0.05, |
| 5: 0.10 |
| } |
| adjustment += rating_weights.get(feedback.rating, 0.0) |
|
|
| |
| if feedback.feedback_type == "correction": |
| adjustment -= 0.03 |
|
|
| |
| new_confidence = max(0.0, min(1.0, current_confidence + adjustment)) |
|
|
| logger.info( |
| f"Adjusted confidence for agent {agent_id}: " |
| f"{current_confidence:.3f} -> {new_confidence:.3f} " |
| f"(adjustment: {adjustment:+.3f})" |
| ) |
|
|
| return new_confidence |
|
|
| def get_learning_signals( |
| self, |
| agent_id: str, |
| days: int = 30 |
| ) -> Dict[str, Any]: |
| """ |
| Get learning signals from recent feedback. |
| |
| Analyzes feedback patterns to provide insights for agent improvement. |
| |
| Args: |
| agent_id: ID of the agent |
| days: Number of days to analyze |
| |
| Returns: |
| Dictionary with learning signals and insights |
| """ |
| cutoff_date = datetime.now() - timedelta(days=days) |
|
|
| |
| feedback = self.db.query(AgentFeedback).filter( |
| AgentFeedback.agent_id == agent_id, |
| AgentFeedback.created_at >= cutoff_date |
| ).all() |
|
|
| if not feedback: |
| |
| learning_record = self.db.query(AgentLearning).filter( |
| AgentLearning.agent_id == agent_id |
| ).first() |
| |
| if not learning_record: |
| return { |
| "agent_id": agent_id, |
| "total_feedback": 0, |
| "learning_signals": [], |
| "improvement_suggestions": [] |
| } |
| |
| |
| return { |
| "agent_id": agent_id, |
| "total_feedback": learning_record.total_feedback or 0, |
| "positive_ratio": learning_record.success_rate or 0, |
| "parameters": learning_record.parameters_json or {}, |
| "learning_signals": [{ |
| "type": "info", |
| "message": "Aggregate learning data available, but no feedback in specified period.", |
| "confidence_impact": "neutral" |
| }], |
| "improvement_suggestions": [] |
| } |
|
|
| |
| total = len(feedback) |
|
|
| |
| positive = sum( |
| 1 for f in feedback |
| if f.thumbs_up_down is True or (f.rating is not None and f.rating >= 4) |
| ) |
|
|
| negative = sum( |
| 1 for f in feedback |
| if f.thumbs_up_down is False or (f.rating is not None and f.rating <= 2) |
| ) |
|
|
| positive_ratio = positive / total if total > 0 else 0 |
|
|
| |
| corrections = [f for f in feedback if f.feedback_type == "correction"] |
|
|
| |
| signals: List[Dict[str, Any]] = [] |
|
|
| if positive_ratio >= 0.8: |
| signals.append({ |
| "type": "strength", |
| "message": "Agent is performing well with high positive feedback", |
| "confidence_impact": "positive" |
| }) |
| elif positive_ratio <= 0.4: |
| signals.append({ |
| "type": "weakness", |
| "message": "Agent is struggling with low positive feedback", |
| "confidence_impact": "negative" |
| }) |
|
|
| if len(corrections) >= 5: |
| signals.append({ |
| "type": "pattern", |
| "message": f"Agent received {len(corrections)} corrections - may need retraining", |
| "confidence_impact": "negative", |
| "correction_count": len(corrections) |
| }) |
|
|
| |
| ratings = [f.rating for f in feedback if f.rating is not None] |
| if ratings: |
| avg_rating = sum(ratings) / len(ratings) |
| if avg_rating >= 4.5: |
| signals.append({ |
| "type": "strength", |
| "message": f"Excellent average rating: {avg_rating:.1f}/5.0", |
| "confidence_impact": "positive" |
| }) |
| elif avg_rating <= 2.5: |
| signals.append({ |
| "type": "weakness", |
| "message": f"Poor average rating: {avg_rating:.1f}/5.0", |
| "confidence_impact": "negative" |
| }) |
|
|
| |
| suggestions = [] |
|
|
| if len(corrections) > 0: |
| suggestions.append({ |
| "type": "training", |
| "message": "Review common correction patterns to identify knowledge gaps", |
| "priority": "high" |
| }) |
|
|
| if positive_ratio < 0.6: |
| suggestions.append({ |
| "type": "supervision", |
| "message": "Increase human supervision until performance improves", |
| "priority": "medium" |
| }) |
|
|
| |
| learning_record = self.db.query(AgentLearning).filter( |
| AgentLearning.agent_id == agent_id |
| ).first() |
| |
| aggregate_data = {} |
| if learning_record: |
| |
| success_rate = 0.0 |
| if learning_record.total_feedback > 0: |
| success_rate = learning_record.positive_feedback / learning_record.total_feedback |
| |
| aggregate_data = { |
| "aggregate_total": learning_record.total_feedback, |
| "aggregate_success_rate": success_rate, |
| "current_parameters": learning_record.parameters_json |
| } |
| |
| if success_rate < 0.5: |
| signals.append({ |
| "type": "warning", |
| "message": f"Long-term success rate for agent is low: {success_rate:.1%}", |
| "confidence_impact": "negative" |
| }) |
|
|
| return { |
| "agent_id": agent_id, |
| "total_feedback_in_period": total, |
| "positive_ratio_in_period": positive_ratio, |
| "correction_count_in_period": len(corrections), |
| "aggregate_data": aggregate_data, |
| "learning_signals": signals, |
| "improvement_suggestions": suggestions |
| } |
|
|
| async def record_feedback_in_world_model( |
| self, |
| feedback: AgentFeedback |
| ) -> bool: |
| """ |
| Record feedback as a learning experience in the world model. |
| |
| This enables agents to learn from past feedback and avoid repeating mistakes. |
| |
| Args: |
| feedback: Feedback object to record |
| |
| Returns: |
| True if successfully recorded, False otherwise |
| """ |
| try: |
| |
| execution = None |
| if feedback.agent_execution_id: |
| execution = self.db.query(AgentExecution).filter( |
| AgentExecution.id == feedback.agent_execution_id |
| ).first() |
|
|
| |
| if feedback.thumbs_up_down is True or (feedback.rating and feedback.rating >= 4): |
| outcome = "Success" |
| elif feedback.thumbs_up_down is False or (feedback.rating and feedback.rating <= 2): |
| outcome = "Failure" |
| else: |
| outcome = "Mixed" |
|
|
| |
| feedback_score = 0.0 |
|
|
| if feedback.thumbs_up_down is not None: |
| feedback_score += 0.5 if feedback.thumbs_up_down else -0.5 |
|
|
| if feedback.rating is not None: |
| |
| feedback_score += (feedback.rating - 3) / 2.0 |
|
|
| |
| feedback_score = max(-1.0, min(1.0, feedback_score)) |
|
|
| |
| experience = AgentExperience( |
| id=str(uuid.uuid4()), |
| agent_id=feedback.agent_id, |
| task_type=feedback.feedback_type or "general", |
| input_summary=feedback.input_context or "User feedback", |
| outcome=outcome, |
| learnings=feedback.user_correction or feedback.ai_reasoning or "", |
| confidence_score=0.5, |
| feedback_score=feedback_score, |
| artifacts=[feedback.agent_execution_id] if feedback.agent_execution_id else [], |
| agent_role="Agent", |
| specialty=None, |
| timestamp=datetime.now() |
| ) |
|
|
| |
| success = await self.world_model.record_experience(experience) |
|
|
| if success: |
| logger.info( |
| f"Recorded feedback in world model: agent={feedback.agent_id}, " |
| f"feedback_score={feedback_score:.2f}" |
| ) |
|
|
| return success |
|
|
| except Exception as e: |
| logger.error(f"Failed to record feedback in world model: {e}") |
| return False |
|
|
| def batch_update_confidence_from_feedback( |
| self, |
| agent_id: str, |
| days: int = 30 |
| ) -> Optional[float]: |
| """ |
| Batch update agent confidence based on recent feedback. |
| |
| Aggregates all feedback from the last N days and adjusts confidence. |
| |
| Args: |
| agent_id: ID of the agent |
| days: Number of days to analyze |
| |
| Returns: |
| New confidence score, or None if agent not found |
| """ |
| agent = self.db.query(AgentRegistry).filter( |
| AgentRegistry.id == agent_id |
| ).first() |
|
|
| if not agent: |
| return None |
|
|
| cutoff_date = datetime.now() - timedelta(days=days) |
|
|
| |
| feedback = self.db.query(AgentFeedback).filter( |
| AgentFeedback.agent_id == agent_id, |
| AgentFeedback.created_at >= cutoff_date |
| ).all() |
|
|
| if not feedback: |
| return agent.confidence_score |
|
|
| |
| total_adjustment = 0.0 |
|
|
| for f in feedback: |
| |
| |
| days_old = (datetime.now() - f.created_at).days |
| recency_weight = max(0.1, 1.0 - (days_old / days)) |
|
|
| adjustment = 0.0 |
|
|
| if f.thumbs_up_down is True: |
| adjustment += 0.05 |
| elif f.thumbs_up_down is False: |
| adjustment -= 0.05 |
|
|
| if f.rating is not None: |
| rating_weights = {1: -0.10, 2: -0.05, 3: 0.00, 4: 0.05, 5: 0.10} |
| adjustment += rating_weights.get(f.rating, 0.0) |
|
|
| if f.feedback_type == "correction": |
| adjustment -= 0.03 |
|
|
| total_adjustment += adjustment * recency_weight |
|
|
| |
| new_confidence = max(0.0, min(1.0, agent.confidence_score + total_adjustment)) |
|
|
| logger.info( |
| f"Batch confidence update for agent {agent_id}: " |
| f"{agent.confidence_score:.3f} -> {new_confidence:.3f} " |
| f"(total adjustment: {total_adjustment:+.3f} from {len(feedback)} feedback)" |
| ) |
|
|
| return new_confidence |
|
|
| async def record_user_correction( |
| self, |
| agent_id: str, |
| tenant_id: str, |
| original_action: Dict[str, Any], |
| corrected_action: Dict[str, Any], |
| context: Optional[str] = None |
| ) -> str: |
| """ |
| Record a user correction for agent learning. |
| Ported from SaaS LearningService. |
| """ |
| experience_id = str(uuid.uuid4()) |
| try: |
| |
| correction_type = self._classify_correction(original_action, corrected_action) |
|
|
| experience = CognitiveExperience( |
| id=experience_id, |
| tenant_id=tenant_id, |
| agent_id=agent_id, |
| experience_type="user_correction", |
| task_type=corrected_action.get("action_type", "unknown"), |
| input_summary=context or "User correction in GuidancePanel", |
| output_summary=json.dumps({ |
| "original": original_action, |
| "corrected": corrected_action |
| }), |
| outcome="correction", |
| learnings={ |
| "original_action": original_action, |
| "corrected_action": corrected_action, |
| "correction_type": correction_type, |
| "timestamp": datetime.now(timezone.utc).isoformat() |
| }, |
| effectiveness_score=0.0 |
| ) |
|
|
| self.db.add(experience) |
| |
| |
| agent = self.db.query(AgentRegistry).filter(AgentRegistry.id == agent_id).first() |
| if agent: |
| |
| agent.confidence_score = max(0.0, (agent.confidence_score or 0.5) - 0.05) |
| logger.info(f"Penalty for correction: Agent {agent_id} confidence -> {agent.confidence_score:.2f}") |
|
|
| self.db.commit() |
| |
| |
| try: |
| self.continuous_learning.update_from_feedback(AgentFeedback( |
| tenant_id=tenant_id, |
| agent_id=agent_id, |
| feedback_type="correction", |
| user_correction=json.dumps(corrected_action), |
| created_at=datetime.now(timezone.utc) |
| )) |
| except Exception as le: |
| logger.warning(f"Continuous learning update failed: {le}") |
|
|
| logger.info(f"Recorded user correction for agent {agent_id}: {correction_type}") |
| return experience_id |
|
|
| except Exception as e: |
| logger.error(f"Failed to record user correction: {e}") |
| self.db.rollback() |
| raise |
|
|
| def _classify_correction(self, original: Dict, corrected: Dict) -> str: |
| """Classify the type of correction made.""" |
| if not isinstance(original, dict) or not isinstance(corrected, dict): |
| return "other_correction" |
| if original.get("action_type") != corrected.get("action_type"): |
| return "action_type_change" |
| if original.get("parameters") != corrected.get("parameters"): |
| return "parameter_adjustment" |
| return "other_correction" |
|
|
| async def record_rejection( |
| self, |
| agent_id: str, |
| tenant_id: str, |
| action_type: str, |
| action_data: Dict[str, Any], |
| reason: Optional[str] = None, |
| context: Optional[str] = None |
| ) -> str: |
| """Record a user rejection for agent learning.""" |
| experience_id = str(uuid.uuid4()) |
| try: |
| experience = CognitiveExperience( |
| id=experience_id, |
| tenant_id=tenant_id, |
| agent_id=agent_id, |
| experience_type="user_rejection", |
| task_type=action_type, |
| input_summary=context or "User rejection in GuidancePanel", |
| output_summary=json.dumps({ |
| "proposed_action": action_data, |
| "rejection_reason": reason |
| }), |
| outcome="rejection", |
| learnings={ |
| "proposed_action": action_data, |
| "rejection_reason": reason, |
| "rejection_type": "explicit_rejection" |
| }, |
| effectiveness_score=-0.5 |
| ) |
|
|
| self.db.add(experience) |
| |
| |
| agent = self.db.query(AgentRegistry).filter(AgentRegistry.id == agent_id).first() |
| if agent: |
| agent.confidence_score = max(0.0, (agent.confidence_score or 0.5) - 0.1) |
| logger.info(f"Penalty for rejection: Agent {agent_id} confidence -> {agent.confidence_score:.2f}") |
|
|
| self.db.commit() |
| |
| |
| try: |
| self.continuous_learning.update_from_feedback(AgentFeedback( |
| tenant_id=tenant_id, |
| agent_id=agent_id, |
| feedback_type="rejection", |
| ai_reasoning=reason, |
| created_at=datetime.now(timezone.utc) |
| )) |
| except Exception as le: |
| logger.warning(f"Continuous learning update failed: {le}") |
|
|
| return experience_id |
| except Exception as e: |
| logger.error(f"Failed to record rejection: {e}") |
| self.db.rollback() |
| raise |
|
|
| async def analyze_failure_patterns( |
| self, |
| agent_id: str, |
| tenant_id: str, |
| min_occurrences: int = 3 |
| ) -> List[Dict[str, Any]]: |
| """Identify recurring failure patterns from CognitiveExperience records.""" |
| try: |
| failures = self.db.query(CognitiveExperience).filter( |
| CognitiveExperience.agent_id == agent_id, |
| CognitiveExperience.tenant_id == tenant_id, |
| CognitiveExperience.outcome.in_(["failure", "correction", "rejection"]) |
| ).order_by(CognitiveExperience.created_at.desc()).limit(100).all() |
|
|
| patterns: Dict[str, Dict[str, Any]] = {} |
| for exp in failures: |
| l = exp.learnings or {} |
| c_type = l.get("correction_type") or l.get("rejection_type") or "unknown" |
| if c_type not in patterns: |
| patterns[c_type] = {"type": c_type, "count": 0, "examples": []} |
| patterns[c_type]["count"] += 1 |
| if len(patterns[c_type]["examples"]) < 3: |
| patterns[c_type]["examples"].append(exp.task_type) |
|
|
| return [p for p in patterns.values() if p["count"] >= min_occurrences] |
| except Exception as e: |
| logger.error(f"Failed to analyze failure patterns: {e}") |
| return [] |
|
|