Spaces:
Sleeping
Sleeping
File size: 7,544 Bytes
519b145 |
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 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 |
"""
Context manager for conversation sessions and messages.
"""
from typing import List, Dict, Any, Optional
from uuid import UUID
from hue_portal.core.models import ConversationSession, ConversationMessage
class ConversationContext:
"""Manages conversation sessions and context."""
@staticmethod
def get_session(session_id: Optional[str] = None, user_id: Optional[str] = None) -> ConversationSession:
"""
Get or create a conversation session.
Args:
session_id: Optional session ID (UUID string). If None, creates new session.
user_id: Optional user ID for tracking.
Returns:
ConversationSession instance.
"""
if session_id:
try:
# Try to get existing session
session = ConversationSession.objects.get(session_id=session_id)
# Update updated_at timestamp
session.save(update_fields=["updated_at"])
return session
except ConversationSession.DoesNotExist:
# Create new session with provided session_id
return ConversationSession.objects.create(
session_id=session_id,
user_id=user_id
)
else:
# Create new session
return ConversationSession.objects.create(user_id=user_id)
@staticmethod
def add_message(
session_id: str,
role: str,
content: str,
intent: Optional[str] = None,
entities: Optional[Dict[str, Any]] = None,
metadata: Optional[Dict[str, Any]] = None
) -> ConversationMessage:
"""
Add a message to a conversation session.
Args:
session_id: Session ID (UUID string).
role: Message role ('user' or 'bot').
content: Message content.
intent: Detected intent (optional).
entities: Extracted entities (optional).
metadata: Additional metadata (optional).
Returns:
ConversationMessage instance.
"""
session = ConversationContext.get_session(session_id=session_id)
return ConversationMessage.objects.create(
session=session,
role=role,
content=content,
intent=intent or "",
entities=entities or {},
metadata=metadata or {}
)
@staticmethod
def get_recent_messages(session_id: str, limit: int = 10) -> List[ConversationMessage]:
"""
Get recent messages from a session.
Args:
session_id: Session ID (UUID string).
limit: Maximum number of messages to return.
Returns:
List of ConversationMessage instances, ordered by timestamp (oldest first).
"""
try:
session = ConversationSession.objects.get(session_id=session_id)
return list(session.messages.all()[:limit])
except ConversationSession.DoesNotExist:
return []
@staticmethod
def get_context_summary(session_id: str, max_messages: int = 5) -> Dict[str, Any]:
"""
Create a summary of conversation context.
Args:
session_id: Session ID (UUID string).
max_messages: Maximum number of messages to include in summary.
Returns:
Dictionary with context summary including:
- recent_messages: List of recent messages
- entities: Aggregated entities from conversation
- intents: List of intents mentioned
- message_count: Total number of messages
"""
messages = ConversationContext.get_recent_messages(session_id, limit=max_messages)
# Aggregate entities
all_entities = {}
intents = []
for msg in messages:
if msg.entities:
for key, value in msg.entities.items():
if key not in all_entities:
all_entities[key] = []
if value not in all_entities[key]:
all_entities[key].append(value)
if msg.intent:
if msg.intent not in intents:
intents.append(msg.intent)
return {
"recent_messages": [
{
"role": msg.role,
"content": msg.content,
"intent": msg.intent,
"timestamp": msg.timestamp.isoformat()
}
for msg in messages
],
"entities": all_entities,
"intents": intents,
"message_count": len(messages)
}
@staticmethod
def extract_entities(query: str) -> Dict[str, Any]:
"""
Extract entities from a query (basic implementation).
This is a placeholder - will be enhanced by entity_extraction.py
Args:
query: User query string.
Returns:
Dictionary with extracted entities.
"""
entities = {}
query_lower = query.lower()
# Basic fine code extraction (V001, V002, etc.)
import re
fine_codes = re.findall(r'\bV\d{3}\b', query, re.IGNORECASE)
if fine_codes:
entities["fine_codes"] = fine_codes
# Basic procedure keywords
procedure_keywords = ["thủ tục", "hồ sơ", "giấy tờ"]
if any(kw in query_lower for kw in procedure_keywords):
entities["has_procedure"] = True
# Basic fine keywords
fine_keywords = ["phạt", "mức phạt", "vi phạm"]
if any(kw in query_lower for kw in fine_keywords):
entities["has_fine"] = True
return entities
@staticmethod
def get_session_metadata(session_id: str) -> Dict[str, Any]:
"""
Return metadata stored with the conversation session.
"""
if not session_id:
return {}
try:
session = ConversationSession.objects.get(session_id=session_id)
return session.metadata or {}
except ConversationSession.DoesNotExist:
return {}
@staticmethod
def update_session_metadata(session_id: str, data: Dict[str, Any]) -> Dict[str, Any]:
"""
Merge provided data into session metadata and persist.
"""
if not session_id:
return {}
session = ConversationContext.get_session(session_id=session_id)
metadata = session.metadata or {}
metadata.update(data)
session.metadata = metadata
session.save(update_fields=["metadata", "updated_at"])
return metadata
@staticmethod
def clear_session_metadata_keys(session_id: str, keys: List[str]) -> Dict[str, Any]:
"""
Remove specific keys from session metadata.
"""
if not session_id:
return {}
session = ConversationContext.get_session(session_id=session_id)
metadata = session.metadata or {}
changed = False
for key in keys:
if key in metadata:
metadata.pop(key)
changed = True
if changed:
session.metadata = metadata
session.save(update_fields=["metadata", "updated_at"])
return metadata
|