annator-atom / backend /core /agent_learning_enhanced.py
techprotrade's picture
Full stack ATOM backend + AIMONEYFLOW clients (port 7860) (part 2)
ff0e46c verified
Raw
History Blame Contribute Delete
21.2 kB
"""
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
# Thumbs up/down
if feedback.thumbs_up_down is True:
adjustment += 0.05
elif feedback.thumbs_up_down is False:
adjustment -= 0.05
# Star rating
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)
# Correction feedback
if feedback.feedback_type == "correction":
adjustment -= 0.03
# Apply adjustment and clamp to [0.0, 1.0]
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)
# Get recent feedback
feedback = self.db.query(AgentFeedback).filter(
AgentFeedback.agent_id == agent_id,
AgentFeedback.created_at >= cutoff_date
).all()
if not feedback:
# Check if we have aggregate learning data even if no recent 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": []
}
# Use aggregate data if no recent feedback
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": []
}
# Analyze patterns
total = len(feedback)
# Positive vs negative
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
# Correction analysis
corrections = [f for f in feedback if f.feedback_type == "correction"]
# Generate learning signals
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)
})
# Rating analysis
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"
})
# Improvement suggestions
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"
})
# Add aggregate signals if available
learning_record = self.db.query(AgentLearning).filter(
AgentLearning.agent_id == agent_id
).first()
aggregate_data = {}
if learning_record:
# Calculate success rate from aggregate stats
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:
# Get execution context if available
execution = None
if feedback.agent_execution_id:
execution = self.db.query(AgentExecution).filter(
AgentExecution.id == feedback.agent_execution_id
).first()
# Determine outcome based on feedback
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"
# Calculate feedback score (-1.0 to 1.0)
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:
# Map 1-5 to -1.0 to 1.0
feedback_score += (feedback.rating - 3) / 2.0
# Clamp to [-1.0, 1.0]
feedback_score = max(-1.0, min(1.0, feedback_score))
# Create experience
experience = AgentExperience(
id=str(uuid.uuid4()), # Will need to import uuid
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", # Could be enhanced
specialty=None,
timestamp=datetime.now()
)
# Record in world model
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)
# Get recent feedback
feedback = self.db.query(AgentFeedback).filter(
AgentFeedback.agent_id == agent_id,
AgentFeedback.created_at >= cutoff_date
).all()
if not feedback:
return agent.confidence_score
# Calculate aggregate adjustment
total_adjustment = 0.0
for f in feedback:
# Use individual feedback adjustments
# Weight by recency (more recent = higher weight)
days_old = (datetime.now() - f.created_at).days
recency_weight = max(0.1, 1.0 - (days_old / days)) # Decay to 0.1
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
# Apply adjustment
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:
# Classify correction type
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)
# Also adjust confidence (penalty for needing correction)
agent = self.db.query(AgentRegistry).filter(AgentRegistry.id == agent_id).first()
if agent:
# Penalty: -0.05
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()
# Continuous learning update (adaptive parameters)
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)
# Confidence penalty: -0.1 (stronger than correction)
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()
# Continuous learning update (adaptive parameters)
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 []