File size: 3,972 Bytes
d1f3f31 | 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 | from typing import Any, Dict, List
FLOW_STAGE_KEYWORDS = {
"GREETING": ["hello", "hi", "good morning", "good afternoon", "good evening", "hey"],
"INTRODUCTION": ["my name is", "this is", "calling from", "work as", "am a student", "am a teacher"],
"DISCOVERY": ["looking for", "what features", "what requirement", "your budget", "need a", "want to buy"],
"PRODUCT_DISCUSSION": ["suggest model", "we have", "comes with", "specifications", "ram", "processor", "screen", "warranty", "available"],
"OBJECTION_HANDLING": ["too expensive", "any discount", "not sure", "thinking", "installment", "emi", "high price", "costly"],
"FOLLOW_UP": ["get back to you", "will follow up", "share the details", "contact you later", "call you back"],
"CLOSING": ["thank you", "thanks for calling", "have a nice day", "goodbye", "bye"]
}
def classify_flow_stage(text: str) -> str:
"""Classify the conversation flow stage for a given turn."""
text_lower = text.lower()
best_stage = "START"
max_matches = 0
for stage, keywords in FLOW_STAGE_KEYWORDS.items():
matches = sum(1 for kw in keywords if kw in text_lower)
if matches > max_matches:
max_matches = matches
best_stage = stage
return best_stage
def validate_and_correct_roles(turns: List[Dict[str, Any]], classifications: Dict[str, Dict[str, Any]], threshold: float = 0.85) -> Dict[str, Dict[str, Any]]:
"""
Validate and correct speaker roles dynamically based on flow stages.
Returns the corrected classifications dictionary.
"""
for i, turn in enumerate(turns):
speaker = turn.get("speaker")
text = turn.get("text", "")
text_lower = text.lower()
cls = classifications.get(speaker)
if not cls:
continue
# Only apply correction rules if initial confidence is below threshold
if cls.get("confidence", 1.0) < threshold:
stage = classify_flow_stage(text)
# Correction Rule 1: Agent asking discovery questions
if stage == "DISCOVERY" and any(q in text_lower for q in ["what", "how", "budget", "need", "preference"]):
if "?" in text or any(kw in text_lower for kw in ["what features", "what is your", "brand preference"]):
cls["role"] = "Agent"
cls["confidence"] = 0.90
cls["method"] = "flow_validator_correction"
# Correction Rule 2: Customer budget/needs statements
elif stage == "DISCOVERY" and any(kw in text_lower for kw in ["my budget", "i want", "i need", "looking for"]):
cls["role"] = "Customer"
cls["confidence"] = 0.90
cls["method"] = "flow_validator_correction"
# Correction Rule 3: Customer presenting objections
elif stage == "OBJECTION_HANDLING" and any(kw in text_lower for kw in ["too expensive", "discount", "not sure", "get back to you"]):
cls["role"] = "Customer"
cls["confidence"] = 0.92
cls["method"] = "flow_validator_correction"
# Correction Rule 4: Agent proposing follow-up
elif stage == "FOLLOW_UP" and any(kw in text_lower for kw in ["i will share", "i'll share", "follow up", "call you back"]):
cls["role"] = "Agent"
cls["confidence"] = 0.90
cls["method"] = "flow_validator_correction"
# Correction Rule 5: Agent introducing company/greetings at start of call
elif i < 2 and stage in ["GREETING", "INTRODUCTION"] and any(kw in text_lower for kw in ["this is", "calling from", "how can i"]):
cls["role"] = "Agent"
cls["confidence"] = 0.95
cls["method"] = "flow_validator_correction"
return classifications
|