Spaces:
Sleeping
Sleeping
File size: 5,964 Bytes
dc32db8 |
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 |
"""
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
|