Spaces:
Sleeping
Sleeping
File size: 6,205 Bytes
ddcc450 |
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 |
"""
Dialogue management for multi-turn conversations.
"""
from typing import Dict, Any, Optional, List, Tuple
from enum import Enum
class DialogueState(Enum):
"""Dialogue states."""
INITIAL = "initial"
COLLECTING_INFO = "collecting_info"
CLARIFYING = "clarifying"
PROVIDING_ANSWER = "providing_answer"
FOLLOW_UP = "follow_up"
COMPLETED = "completed"
class DialogueManager:
"""Manages dialogue state and multi-turn conversations."""
def __init__(self):
self.state = DialogueState.INITIAL
self.slots = {} # Slot filling for missing information
self.context_switch_detected = False
def update_state(
self,
query: str,
intent: str,
results_count: int,
confidence: float,
recent_messages: Optional[List[Dict[str, Any]]] = None
) -> DialogueState:
"""
Update dialogue state based on current query and context.
Args:
query: Current user query.
intent: Detected intent.
results_count: Number of results found.
confidence: Confidence score.
recent_messages: Recent conversation messages.
Returns:
Updated dialogue state.
"""
# Detect context switching
if recent_messages and len(recent_messages) > 0:
last_intent = recent_messages[-1].get("intent")
if last_intent and last_intent != intent and intent != "greeting":
self.context_switch_detected = True
self.state = DialogueState.INITIAL
self.slots = {}
return self.state
# State transitions
if results_count == 0 and confidence < 0.5:
# No results and low confidence - need clarification
self.state = DialogueState.CLARIFYING
elif results_count > 0 and confidence >= 0.7:
# Good results - providing answer
self.state = DialogueState.PROVIDING_ANSWER
elif results_count > 0 and confidence < 0.7:
# Some results but uncertain - might need follow-up
self.state = DialogueState.FOLLOW_UP
else:
self.state = DialogueState.PROVIDING_ANSWER
return self.state
def needs_clarification(
self,
query: str,
intent: str,
results_count: int
) -> Tuple[bool, Optional[str]]:
"""
Check if clarification is needed.
Args:
query: User query.
intent: Detected intent.
results_count: Number of results.
Returns:
Tuple of (needs_clarification, clarification_message).
"""
if results_count == 0:
# No results - ask for clarification
clarification_messages = {
"search_fine": "Bạn có thể cho biết cụ thể hơn về loại vi phạm không? Ví dụ: vượt đèn đỏ, không đội mũ bảo hiểm...",
"search_procedure": "Bạn muốn tìm thủ tục nào? Ví dụ: đăng ký cư trú, thủ tục ANTT...",
"search_office": "Bạn muốn tìm đơn vị nào? Ví dụ: công an phường, điểm tiếp dân...",
"search_advisory": "Bạn muốn tìm cảnh báo về chủ đề gì?",
}
message = clarification_messages.get(intent, "Bạn có thể cung cấp thêm thông tin không?")
return (True, message)
return (False, None)
def detect_missing_slots(
self,
intent: str,
query: str,
results_count: int
) -> Dict[str, Any]:
"""
Detect missing information slots.
Args:
intent: Detected intent.
query: User query.
results_count: Number of results.
Returns:
Dictionary of missing slots.
"""
missing_slots = {}
if intent == "search_fine":
# Check for fine code or fine name
if "v001" not in query.lower() and "v002" not in query.lower():
if not any(kw in query.lower() for kw in ["vượt đèn đỏ", "mũ bảo hiểm", "nồng độ cồn"]):
missing_slots["fine_specification"] = True
elif intent == "search_procedure":
# Check for procedure name or domain
if not any(kw in query.lower() for kw in ["cư trú", "antt", "pccc", "đăng ký"]):
missing_slots["procedure_specification"] = True
elif intent == "search_office":
# Check for office name or location
if not any(kw in query.lower() for kw in ["phường", "huyện", "tỉnh", "điểm tiếp dân"]):
missing_slots["office_specification"] = True
return missing_slots
def handle_follow_up(
self,
query: str,
recent_messages: List[Dict[str, Any]]
) -> Optional[str]:
"""
Generate follow-up question if needed.
Args:
query: Current query.
recent_messages: Recent conversation messages.
Returns:
Follow-up question or None.
"""
if not recent_messages:
return None
# Check if query is very short (likely a follow-up)
if len(query.split()) <= 3:
last_message = recent_messages[-1]
last_intent = last_message.get("intent")
if last_intent == "search_fine":
return "Bạn muốn biết thêm thông tin gì về mức phạt này? (ví dụ: điều luật, biện pháp khắc phục)"
elif last_intent == "search_procedure":
return "Bạn muốn biết thêm thông tin gì về thủ tục này? (ví dụ: hồ sơ, lệ phí, thời hạn)"
return None
def reset(self):
"""Reset dialogue manager state."""
self.state = DialogueState.INITIAL
self.slots = {}
self.context_switch_detected = False
|