File size: 6,685 Bytes
5c244a3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 | """
Task Agent - Handles task creation, updates, and queries
"""
from typing import Dict, Any, Tuple, List
import re
from datetime import datetime, timedelta
from .base import BaseSkillAgent, RegexExtractor
class TaskAgent(BaseSkillAgent):
"""AI agent for task management operations"""
# Intent patterns
INTENT_PATTERNS = {
"create": [
r"create\s+(?:a\s+)?task",
r"add\s+(?:a\s+)?task",
r"new\s+task",
r"make\s+(?:a\s+)?task",
r"remind\s+me\s+to", # Overlaps with reminders
],
"update": [
r"update\s+(?:the\s+)?task",
r"change\s+(?:the\s+)?task",
r"modify\s+(?:the\s+)?task",
r"edit\s+(?:the\s+)?task",
],
"complete": [
r"complete\s+(?:the\s+)?task",
r"finish\s+(?:the\s+)?task",
r"done\s+with\s+(?:the\s+)?task",
r"mark\s+(?:the\s+)?task\s+(?:as\s+)?complete",
],
"delete": [
r"delete\s+(?:the\s+)?task",
r"remove\s+(?:the\s+)?task",
],
"list": [
r"list\s+(?:all\s+)?task",
r"show\s+(?:all\s+)?task",
r"what\s+task",
r"my\s+task",
],
}
def __init__(self):
super().__init__("TaskAgent")
def extract(self, user_input: str) -> Dict[str, Any]:
"""Extract task data from user input"""
# Detect intent
intent = self._detect_intent(user_input)
# Extract based on intent
if intent == "create":
return self._extract_task_creation(user_input)
elif intent == "update":
return self._extract_task_update(user_input)
elif intent == "complete":
return self._extract_task_completion(user_input)
elif intent == "delete":
return self._extract_task_deletion(user_input)
elif intent == "list":
return self._extract_task_list(user_input)
return {"intent": "unknown", "raw_input": user_input}
def validate(self, data: Dict[str, Any]) -> Tuple[bool, List[str]]:
"""Validate task data"""
errors = []
# Title is required for creation
if data.get("intent") == "create":
if not data.get("title"):
errors.append("Title is required")
# Task ID is required for update/delete/complete
if data.get("intent") in ["update", "delete", "complete"]:
if not data.get("task_id"):
errors.append("Task ID is required")
# Validate priority
if "priority" in data:
valid_priorities = ["low", "medium", "high", "urgent"]
if data["priority"] not in valid_priorities:
errors.append(f"Priority must be one of: {', '.join(valid_priorities)}")
# Validate status
if "status" in data:
valid_statuses = ["active", "completed", "deleted"]
if data["status"] not in valid_statuses:
errors.append(f"Status must be one of: {', '.join(valid_statuses)}")
return len(errors) == 0, errors
def execute(self, data: Dict[str, Any], context: Dict[str, Any]) -> Dict[str, Any]:
"""Execute task operation (called by orchestrator)"""
# This is a placeholder - actual execution happens in orchestrator
# via database operations and Dapr events
return {
"agent": self.name,
"operation": data.get("intent"),
"data": data,
}
def _detect_intent(self, user_input: str) -> str:
"""Detect user intent from input"""
for intent, patterns in self.INTENT_PATTERNS.items():
for pattern in patterns:
if re.search(pattern, user_input, re.IGNORECASE):
return intent
return "unknown"
def _extract_task_creation(self, user_input: str) -> Dict[str, Any]:
"""Extract task creation data"""
# Remove common phrases to isolate task title
title = user_input
for pattern in [
r"create\s+(?:a\s+)?task\s+(?:to\s+)?(?:called\s+)?",
r"add\s+(?:a\s+)?task\s+(?:to\s+)?",
r"new\s+task\s+(?:called\s+)?",
r"remind\s+me\s+to\s+",
]:
title = re.sub(pattern, "", title, flags=re.IGNORECASE)
# Clean up title
title = title.strip().capitalize()
return {
"intent": "create",
"title": title,
"priority": RegexExtractor.extract_priority(user_input),
"tags": RegexExtractor.extract_tags(user_input),
"raw_input": user_input,
}
def _extract_task_update(self, user_input: str) -> Dict[str, Any]:
"""Extract task update data"""
task_id = RegexExtractor.extract_task_id(user_input)
# Extract what to update
new_title = None
new_priority = None
# Check for priority change
if "priority" in user_input.lower():
new_priority = RegexExtractor.extract_priority(user_input)
return {
"intent": "update",
"task_id": task_id,
"priority": new_priority,
"raw_input": user_input,
}
def _extract_task_completion(self, user_input: str) -> Dict[str, Any]:
"""Extract task completion data"""
task_id = RegexExtractor.extract_task_id(user_input)
return {
"intent": "complete",
"task_id": task_id,
"status": "completed",
"raw_input": user_input,
}
def _extract_task_deletion(self, user_input: str) -> Dict[str, Any]:
"""Extract task deletion data"""
task_id = RegexExtractor.extract_task_id(user_input)
return {
"intent": "delete",
"task_id": task_id,
"raw_input": user_input,
}
def _extract_task_list(self, user_input: str) -> Dict[str, Any]:
"""Extract task list query"""
# Extract filters
filters = {}
if "active" in user_input.lower():
filters["status"] = "active"
elif "completed" in user_input.lower():
filters["status"] = "completed"
if "high" in user_input.lower() or "urgent" in user_input.lower():
filters["priority"] = "high"
return {
"intent": "list",
"filters": filters,
"raw_input": user_input,
}
|