Spaces:
Sleeping
Sleeping
File size: 8,273 Bytes
fd22f8e 7196ae9 a72fec7 7196ae9 8117b70 7196ae9 8117b70 7196ae9 8117b70 7196ae9 49f77e8 7196ae9 8117b70 49f77e8 7a1ebee 4386026 8117b70 7196ae9 |
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 |
# ────────────────────────────── memo/plan/intent.py ──────────────────────────────
"""
Intent Detection
Handles user intent detection for memory planning.
"""
import re, os
from typing import List, Dict, Any, Tuple, Optional
from enum import Enum
from utils.logger import get_logger
logger = get_logger("INTENT_DETECTOR", __name__)
class QueryIntent(Enum):
"""Types of user query intents"""
ENHANCEMENT = "enhancement" # User wants more details/elaboration
CLARIFICATION = "clarification" # User wants clarification
CONTINUATION = "continuation" # User is continuing previous topic
NEW_TOPIC = "new_topic" # User is starting a new topic
COMPARISON = "comparison" # User wants to compare with previous content
REFERENCE = "reference" # User is referencing specific past content
class IntentDetector:
"""Handles user intent detection for memory planning"""
def __init__(self):
# Enhancement request patterns
self.enhancement_patterns = [
r'\b(enhance|elaborate|expand|detail|elaborate on|be more detailed|more details|more information)\b',
r'\b(explain more|tell me more|go deeper|dive deeper|more context)\b',
r'\b(what else|anything else|additional|further|supplement)\b',
r'\b(comprehensive|thorough|complete|full)\b',
r'\b(based on|from our|as we discussed|following up|regarding)\b'
]
# Clarification patterns
self.clarification_patterns = [
r'\b(what do you mean|clarify|explain|what is|define)\b',
r'\b(how does|why does|when does|where does)\b',
r'\b(can you explain|help me understand)\b'
]
# Comparison patterns
self.comparison_patterns = [
r'\b(compare|versus|vs|difference|similar|different)\b',
r'\b(like|unlike|similar to|different from)\b',
r'\b(contrast|opposite|better|worse)\b'
]
# Reference patterns
self.reference_patterns = [
r'\b(you said|we discussed|earlier|before|previously)\b',
r'\b(that|this|it|the above|mentioned)\b',
r'\b(according to|based on|from|in)\b'
]
async def detect_intent(self, question: str, nvidia_rotator=None, user_id: str = "") -> QueryIntent:
"""Detect user intent from the question"""
try:
question_lower = question.lower()
# Check for enhancement patterns
if any(re.search(pattern, question_lower) for pattern in self.enhancement_patterns):
return QueryIntent.ENHANCEMENT
# Check for clarification patterns
if any(re.search(pattern, question_lower) for pattern in self.clarification_patterns):
return QueryIntent.CLARIFICATION
# Check for comparison patterns
if any(re.search(pattern, question_lower) for pattern in self.comparison_patterns):
return QueryIntent.COMPARISON
# Check for reference patterns
if any(re.search(pattern, question_lower) for pattern in self.reference_patterns):
return QueryIntent.REFERENCE
# Use AI for more sophisticated intent detection
if nvidia_rotator:
try:
return await self._ai_intent_detection(question, nvidia_rotator, user_id)
except Exception as e:
logger.warning(f"[INTENT_DETECTOR] AI intent detection failed: {e}")
# Default to continuation if no clear patterns
return QueryIntent.CONTINUATION
except Exception as e:
logger.warning(f"[INTENT_DETECTOR] Intent detection failed: {e}")
return QueryIntent.CONTINUATION
async def _ai_intent_detection(self, question: str, nvidia_rotator, user_id: str = "") -> QueryIntent:
"""Use AI to detect user intent more accurately"""
try:
from utils.api.router import generate_answer_with_model
from utils.analytics import get_analytics_tracker
# Track memory agent usage
tracker = get_analytics_tracker()
if tracker:
await tracker.track_agent_usage(
user_id=user_id,
agent_name="memory",
action="detect",
context="intent_detection",
metadata={"question": question[:100]}
)
sys_prompt = """You are an expert at analyzing user intent in questions.
Classify the user's question into one of these intents:
- ENHANCEMENT: User wants more details, elaboration, or comprehensive information
- CLARIFICATION: User wants explanation or clarification of something
- CONTINUATION: User is continuing a previous topic or conversation
- NEW_TOPIC: User is starting a completely new topic
- COMPARISON: User wants to compare or contrast things
- REFERENCE: User is referencing specific past content or discussions
Respond with only the intent name (e.g., "ENHANCEMENT")."""
user_prompt = f"Question: {question}\n\nWhat is the user's intent?"
# Track memory agent usage
try:
from utils.analytics import get_analytics_tracker
tracker = get_analytics_tracker()
if tracker and user_id:
await tracker.track_agent_usage(
user_id=user_id,
agent_name="memory",
action="intent",
context="intent_detection",
metadata={"question": question[:100]}
)
except Exception:
pass
# Track memory agent usage
tracker = get_analytics_tracker()
if tracker:
await tracker.track_agent_usage(
user_id=user_id,
agent_name="memory",
action="detect",
context="intent_detection",
metadata={"question": question[:100]}
)
# Track memo agent usage
try:
from utils.analytics import get_analytics_tracker
tracker = get_analytics_tracker()
if tracker:
await tracker.track_agent_usage(
user_id=user_id,
agent_name="memo",
action="intent",
context="intent_detection",
metadata={"query": query}
)
except Exception:
pass
# Use Qwen for better intent detection reasoning
from utils.api.router import qwen_chat_completion
response = await qwen_chat_completion(sys_prompt, user_prompt, nvidia_rotator, user_id, "intent_detection")
# Parse response
response_upper = response.strip().upper()
for intent in QueryIntent:
if intent.name in response_upper:
return intent
return QueryIntent.CONTINUATION
except Exception as e:
logger.warning(f"[INTENT_DETECTOR] AI intent detection failed: {e}")
return QueryIntent.CONTINUATION
# ────────────────────────────── Global Instance ──────────────────────────────
_intent_detector: Optional[IntentDetector] = None
def get_intent_detector() -> IntentDetector:
"""Get the global intent detector instance"""
global _intent_detector
if _intent_detector is None:
_intent_detector = IntentDetector()
logger.info("[INTENT_DETECTOR] Global intent detector initialized")
return _intent_detector
|