170 lines
6.2 KiB
Python
170 lines
6.2 KiB
Python
"""
|
|
Response formatter enforcing Rules 53-60:
|
|
- Generate response after state processing.
|
|
- One Question Only (Rule 54).
|
|
- Acknowledge + Ask pattern (Rule 55).
|
|
- No unnecessary confirmation / No robotic language (Rules 56 & 57).
|
|
- No medical explanations (Rule 58).
|
|
- Natural conversation with short answers (Rule 59).
|
|
- Response length enforcement: 5-15 words target, max 20 words (Rule 60).
|
|
"""
|
|
|
|
import re
|
|
from typing import Optional
|
|
|
|
|
|
ROBOTIC_PHRASES = [
|
|
r"i\s+have\s+(?:recorded|noted|documented)(?:\s+that|\s+this)?\.?",
|
|
r"thank\s+you\s+for\s+(?:sharing|providing|the)\s+(?:that\s+|this\s+)?(?:information|details|response)?\.?",
|
|
r"i\s+understand\s+your\s+response\.?",
|
|
r"please\s+provide\s+(?:the\s+)?(?:duration|severity|location|additional|details)",
|
|
r"kindly\s+specify\s+(?:the\s+)?(?:duration|severity|location)",
|
|
r"your\s+response\s+has\s+been\s+recorded\.?",
|
|
r"your\s+response\s+indicates\s+that[^.!?]+[.!?]",
|
|
]
|
|
|
|
DIAGNOSTIC_PATTERNS = [
|
|
r"this\s+could\s+(?:indicate|suggest|be)\s+[^.!?]+[.!?]",
|
|
r"possible\s+causes\s+include\s+[^.!?]+[.!?]",
|
|
]
|
|
|
|
FORM_READBACK_PATTERNS = [
|
|
r"i\s+understand\s+(?:that\s+)?you\s+(?:have|are|were|take|have\s+been|suffered)[^.!?]+[.!?]",
|
|
r"you\s+mentioned\s+(?:that\s+)?you\s+(?:have|are|were|take|have\s+been)[^.!?]+[.!?]",
|
|
r"i\s+see\s+that\s+you\s+(?:have|are|were|take|have\s+been)[^.!?]+[.!?]",
|
|
]
|
|
|
|
DISCLAIMER_PATTERNS = [
|
|
r"as\s+an\s+ai\s+(?:language\s+model|assistant|clinical\s+intake)[^.!?]*[.!?]?",
|
|
r"please\s+note\s+that\s+i\s+cannot\s+provide\s+medical\s+advice[^.!?]*[.!?]?",
|
|
r"i\s+am\s+not\s+a\s+doctor[^.!?]*[.!?]?",
|
|
]
|
|
|
|
|
|
def deduplicate_sentences(text: str) -> str:
|
|
"""Removes duplicate consecutive or identical sentences."""
|
|
if not text:
|
|
return text
|
|
sentences = [s.strip() for s in re.split(r"(?<=[.!?])\s+", text) if s.strip()]
|
|
seen = set()
|
|
unique = []
|
|
for s in sentences:
|
|
s_norm = s.lower()
|
|
if s_norm not in seen:
|
|
seen.add(s_norm)
|
|
unique.append(s)
|
|
return " ".join(unique)
|
|
|
|
|
|
def sanitize_final_response(raw_reply: str, lang: str = "en", scenario: str = "normal") -> str:
|
|
"""
|
|
Final response sanitization pipeline (Rule 70):
|
|
Generated Response
|
|
↓ Remove duplicate sentences
|
|
↓ Remove form readbacks (Rule 63) & diagnostic/robotic text (Rule 61)
|
|
↓ Remove multiple questions (Rule 54)
|
|
↓ Enforce dynamic response length (Rule 62)
|
|
↓ Return short natural response
|
|
"""
|
|
if not raw_reply or not raw_reply.strip():
|
|
if scenario in ["emergency", "crisis"]:
|
|
return "Please seek emergency medical evaluation immediately or call local emergency services."
|
|
if scenario == "completion":
|
|
return "Thank you. Your intake information is registered for the doctor."
|
|
return "Got it. What symptoms are you having?"
|
|
|
|
# Emergency / Crisis scenario (Rule 62): Return concise safety instruction without trimming
|
|
if scenario in ["emergency", "crisis"]:
|
|
text = raw_reply.strip()
|
|
text = re.sub(r"\s+", " ", text)
|
|
return text
|
|
|
|
text = raw_reply.strip()
|
|
|
|
# 1. Strip disclaimers (Rule 61)
|
|
for disc in DISCLAIMER_PATTERNS:
|
|
text = re.sub(disc, "", text, flags=re.IGNORECASE)
|
|
|
|
# 2. Strip diagnostic explanations (Rule 58 & 61)
|
|
for diag in DIAGNOSTIC_PATTERNS:
|
|
text = re.sub(diag, "", text, flags=re.IGNORECASE)
|
|
|
|
# 3. Strip form readbacks (Rule 63)
|
|
for readback in FORM_READBACK_PATTERNS:
|
|
text = re.sub(readback, "", text, flags=re.IGNORECASE)
|
|
|
|
# 4. Strip robotic confirmation phrases (Rule 56, 57, 64)
|
|
for rob in ROBOTIC_PHRASES:
|
|
text = re.sub(rob, "", text, flags=re.IGNORECASE)
|
|
|
|
text = re.sub(r"\s+", " ", text).strip()
|
|
if not text:
|
|
if scenario == "completion":
|
|
return "Thank you. Your intake information is registered for the doctor."
|
|
return "Got it. When did it start?"
|
|
|
|
# 5. Deduplicate sentences (Rule 70)
|
|
text = deduplicate_sentences(text)
|
|
|
|
# 6. Enforce ONE question only (Rule 54 & 70)
|
|
if text.count("?") > 1:
|
|
q_parts = text.split("?")
|
|
text = q_parts[0].strip() + "?"
|
|
|
|
# Completion scenario (Rule 62)
|
|
if scenario == "completion":
|
|
words = text.split()
|
|
if len(words) > 15:
|
|
text = " ".join(words[:12]) + "."
|
|
return text
|
|
|
|
# 7. Enforce max 1-2 short sentences (Rule 55, 60, 62)
|
|
sentences = [s.strip() for s in re.split(r"(?<=[.!?])\s+", text) if s.strip()]
|
|
if len(sentences) > 2:
|
|
q_idx = next((i for i, s in enumerate(sentences) if "?" in s), -1)
|
|
if q_idx != -1:
|
|
if q_idx > 0:
|
|
sentences = [sentences[0], sentences[q_idx]]
|
|
else:
|
|
sentences = [sentences[0]]
|
|
else:
|
|
sentences = sentences[:2]
|
|
|
|
text = " ".join(sentences)
|
|
|
|
# 8. Dynamic length enforcement: 5-15 words (max 20 words)
|
|
words = text.split()
|
|
if len(words) > 20:
|
|
if "?" in text:
|
|
q_sent = next((s for s in sentences if "?" in s), None)
|
|
ack_sent = next((s for s in sentences if "?" not in s), None)
|
|
|
|
if q_sent:
|
|
q_words = q_sent.split()
|
|
if len(q_words) > 20:
|
|
clean_words = [re.sub(r"[.!?]+$", "", w) for w in q_words[:18]]
|
|
text = " ".join(clean_words) + "...?"
|
|
elif ack_sent:
|
|
ack_words = ack_sent.split()
|
|
if len(ack_words) + len(q_words) <= 20:
|
|
text = f"{ack_sent} {q_sent}"
|
|
else:
|
|
candidate = f"Got it. {q_sent}"
|
|
if len(candidate.split()) <= 20:
|
|
text = candidate
|
|
else:
|
|
text = q_sent
|
|
else:
|
|
text = q_sent
|
|
else:
|
|
clean_words = [re.sub(r"[.!?]+$", "", w) for w in words[:18]]
|
|
text = " ".join(clean_words) + "..."
|
|
|
|
return text
|
|
|
|
|
|
def format_conversational_response(raw_reply: str, lang: str = "en", scenario: str = "normal") -> str:
|
|
"""Wrapper for final response sanitization pipeline (Rule 70)."""
|
|
return sanitize_final_response(raw_reply, lang=lang, scenario=scenario)
|
|
|