435 lines
16 KiB
Python
435 lines
16 KiB
Python
import json
|
||
import re
|
||
from typing import Any
|
||
|
||
|
||
_LOW_SIGNAL_TEXTS = {
|
||
"",
|
||
"ага",
|
||
"алло",
|
||
"да",
|
||
"добрый",
|
||
"добрый день",
|
||
"здравствуйте",
|
||
"ладно",
|
||
"неа",
|
||
"нет",
|
||
"ой",
|
||
"ок",
|
||
"okay",
|
||
"понял",
|
||
"поняла",
|
||
"привет",
|
||
"слышу",
|
||
"слышно",
|
||
"угу",
|
||
"хм",
|
||
"хорошо",
|
||
"hi",
|
||
"hello",
|
||
"ясно",
|
||
}
|
||
|
||
_CITY_ALIASES = {
|
||
"алма ата": "Алма-Ата",
|
||
"алмат": "Алмата",
|
||
"астан": "Астана",
|
||
"актау": "Актау",
|
||
"актоб": "Актобе",
|
||
"атырау": "Атырау",
|
||
"караганд": "Караганда",
|
||
"кокшетау": "Кокшетау",
|
||
"костан": "Костанай",
|
||
"кызылорд": "Кызылорда",
|
||
"орал": "Орал",
|
||
"павлодар": "Павлодар",
|
||
"петропавл": "Петропавловск",
|
||
"семей": "Семей",
|
||
"талдыкорган": "Талдыкорган",
|
||
"тар": "Тараз",
|
||
"туркестан": "Туркестан",
|
||
"уральск": "Уральск",
|
||
"шымкент": "Шымкент",
|
||
}
|
||
|
||
_SERVICE_HINTS = (
|
||
(("график", "работает", "часы работы", "режим работы", "jumys uaqyty"), "график работы", "schedule"),
|
||
(("адрес", "филиал", "офис", "отделени", "мкр", "микрорайон", "район", "мекен"), "филиал или адрес", "address"),
|
||
(("тариф", "цена", "стоимость", "оплат", "баға", "сколько стоит"), "тариф или стоимость", "price"),
|
||
(("статус", "заявк", "заказ", "өтінім", "обращени"), "статус заявки", "status"),
|
||
(("не работает", "ошибка", "проблем", "сбой", "интернет", "связь"), "техническая проблема", "problem"),
|
||
)
|
||
|
||
_INTENT_SLOT_DEFAULTS = {
|
||
"address": ["city_or_branch"],
|
||
"price": ["service_or_tariff"],
|
||
"problem": ["problem_details"],
|
||
"schedule": ["city_or_branch"],
|
||
"status": ["request_number_or_phone"],
|
||
}
|
||
|
||
|
||
def _default_summary() -> dict[str, Any]:
|
||
return {
|
||
"version": 1,
|
||
"channel": None,
|
||
"language": None,
|
||
"customer_name": None,
|
||
"active_intent": None,
|
||
"active_request_text": None,
|
||
"scenario_state": None,
|
||
"confirmed_facts": {
|
||
"branch_hint": None,
|
||
"city": None,
|
||
"request_number": None,
|
||
"service_hint": None,
|
||
},
|
||
"requested_clarifications": [],
|
||
"open_slots": [],
|
||
"updated_at": None,
|
||
}
|
||
|
||
|
||
def _text_key(value: str | None) -> str:
|
||
compact = re.sub(r"[^\w\s-]+", " ", str(value or "").lower(), flags=re.UNICODE)
|
||
return re.sub(r"\s+", " ", compact).strip()
|
||
|
||
|
||
def load_context_summary(raw: str | dict[str, Any] | None) -> dict[str, Any]:
|
||
summary = _default_summary()
|
||
payload: dict[str, Any] = {}
|
||
if isinstance(raw, dict):
|
||
payload = raw
|
||
elif isinstance(raw, str) and raw.strip():
|
||
try:
|
||
parsed = json.loads(raw)
|
||
except json.JSONDecodeError:
|
||
parsed = {}
|
||
if isinstance(parsed, dict):
|
||
payload = parsed
|
||
for key in ("channel", "language", "customer_name", "active_intent", "active_request_text", "scenario_state", "updated_at"):
|
||
value = payload.get(key)
|
||
if value not in {None, ""}:
|
||
summary[key] = str(value).strip()
|
||
facts = payload.get("confirmed_facts")
|
||
if isinstance(facts, dict):
|
||
for key in summary["confirmed_facts"]:
|
||
value = facts.get(key)
|
||
if value not in {None, ""}:
|
||
summary["confirmed_facts"][key] = str(value).strip()
|
||
clarifications = payload.get("requested_clarifications")
|
||
if isinstance(clarifications, list):
|
||
summary["requested_clarifications"] = [str(item).strip() for item in clarifications if str(item).strip()]
|
||
open_slots = payload.get("open_slots")
|
||
if isinstance(open_slots, list):
|
||
summary["open_slots"] = [str(item).strip() for item in open_slots if str(item).strip()]
|
||
return summary
|
||
|
||
|
||
def dump_context_summary(summary: dict[str, Any]) -> str:
|
||
normalized = load_context_summary(summary)
|
||
return json.dumps(normalized, ensure_ascii=False, sort_keys=True)
|
||
|
||
|
||
def _is_low_signal_text(text: str | None) -> bool:
|
||
normalized = _text_key(text)
|
||
return normalized in _LOW_SIGNAL_TEXTS
|
||
|
||
|
||
def _canonical_city(candidate: str) -> str | None:
|
||
normalized = _text_key(candidate)
|
||
if not normalized:
|
||
return None
|
||
for alias, canonical in _CITY_ALIASES.items():
|
||
if alias in normalized:
|
||
return canonical
|
||
tokens = [token for token in re.split(r"[\s-]+", normalized) if token]
|
||
if not tokens:
|
||
return None
|
||
trimmed = " ".join(tokens[:3])
|
||
if len(trimmed) < 3:
|
||
return None
|
||
return " ".join(token.capitalize() for token in trimmed.split())
|
||
|
||
|
||
def _extract_city(text: str | None) -> str | None:
|
||
source = str(text or "").strip()
|
||
if not source:
|
||
return None
|
||
lowered = _text_key(source)
|
||
for alias, canonical in _CITY_ALIASES.items():
|
||
if alias in lowered:
|
||
return canonical
|
||
match = re.search(
|
||
r"(?:в городе|город(?:е)?|қала(?:да|сы)?|qala(?:da)?)\s+([A-Za-zА-Яа-яЁёІіҢңҒғҮүҰұҚқӨөҺһ\- ]{2,40})",
|
||
source,
|
||
flags=re.IGNORECASE,
|
||
)
|
||
if not match:
|
||
return None
|
||
candidate = re.split(r"[,.!?;:]", match.group(1), maxsplit=1)[0].strip()
|
||
return _canonical_city(candidate)
|
||
|
||
|
||
def _extract_request_number(text: str | None) -> str | None:
|
||
normalized = str(text or "").strip()
|
||
if not normalized:
|
||
return None
|
||
match = re.search(r"\b[А-ЯA-Z]{0,3}\d{5,}\b", normalized, flags=re.IGNORECASE)
|
||
if match:
|
||
return match.group(0).strip()
|
||
match = re.search(r"\b\d{6,}\b", normalized)
|
||
if match:
|
||
return match.group(0).strip()
|
||
return None
|
||
|
||
|
||
def _extract_branch_hint(text: str | None) -> str | None:
|
||
normalized = _text_key(text)
|
||
if not normalized:
|
||
return None
|
||
if not any(marker in normalized for marker in ("адрес", "филиал", "офис", "отделени", "район", "мкр", "микрорайон", "мекен")):
|
||
return None
|
||
compact = re.sub(r"\s+", " ", str(text or "").strip())
|
||
return compact[:160] if compact else None
|
||
|
||
|
||
def _extract_service_hint(text: str | None) -> tuple[str | None, str | None]:
|
||
normalized = _text_key(text)
|
||
if not normalized:
|
||
return None, None
|
||
for markers, label, intent in _SERVICE_HINTS:
|
||
if any(marker in normalized for marker in markers):
|
||
return label, intent
|
||
return None, None
|
||
|
||
|
||
def _intent_open_slots(intent: str | None, facts: dict[str, Any]) -> list[str]:
|
||
normalized_intent = str(intent or "").strip().lower()
|
||
city = str(facts.get("city") or "").strip()
|
||
branch_hint = str(facts.get("branch_hint") or "").strip()
|
||
service_hint = str(facts.get("service_hint") or "").strip()
|
||
request_number = str(facts.get("request_number") or "").strip()
|
||
if normalized_intent in {"schedule", "address"}:
|
||
if not city and not branch_hint:
|
||
return ["city_or_branch"]
|
||
if city and not branch_hint:
|
||
return ["branch_or_address"]
|
||
return []
|
||
if normalized_intent == "price":
|
||
return [] if service_hint else ["service_or_tariff"]
|
||
if normalized_intent == "status":
|
||
return [] if request_number else ["request_number_or_phone"]
|
||
if normalized_intent == "problem":
|
||
return [] if service_hint else ["problem_details"]
|
||
return list(_INTENT_SLOT_DEFAULTS.get(normalized_intent, []))
|
||
|
||
|
||
def _requested_clarifications_from_reply(text: str | None) -> list[str]:
|
||
normalized = _text_key(text)
|
||
if not normalized:
|
||
return []
|
||
if any(marker in normalized for marker in ("город", "в городе", "қала")):
|
||
if any(marker in normalized for marker in ("филиал", "адрес", "офис", "отделени")):
|
||
return ["branch_or_address"]
|
||
return ["city_or_branch"]
|
||
if any(marker in normalized for marker in ("филиал", "адрес", "офис", "отделени", "район")):
|
||
return ["branch_or_address"]
|
||
if any(marker in normalized for marker in ("тариф", "стоимость", "цена", "услуга", "баға")):
|
||
return ["service_or_tariff"]
|
||
if any(marker in normalized for marker in ("номер заявки", "номер заказа", "телефон", "өтінім")):
|
||
return ["request_number_or_phone"]
|
||
if any(marker in normalized for marker in ("что именно", "уточните", "подскажите точнее")):
|
||
return ["goal"]
|
||
return []
|
||
|
||
|
||
def _merge_open_slots(
|
||
current_slots: list[str],
|
||
*,
|
||
clarified_slots: list[str],
|
||
fallback_slots: list[str],
|
||
) -> tuple[list[str], list[str]]:
|
||
specific_current_slots = [slot for slot in current_slots if slot and slot != "goal"]
|
||
specific_fallback_slots = [slot for slot in fallback_slots if slot and slot != "goal"]
|
||
if clarified_slots:
|
||
if clarified_slots == ["goal"]:
|
||
preserved_slots = specific_current_slots or specific_fallback_slots
|
||
if preserved_slots:
|
||
return preserved_slots, preserved_slots
|
||
return clarified_slots, clarified_slots
|
||
if specific_current_slots and specific_fallback_slots and set(specific_current_slots) == set(specific_fallback_slots):
|
||
return specific_current_slots, specific_current_slots
|
||
return fallback_slots, current_slots
|
||
|
||
|
||
def _looks_like_request_statement(text: str | None) -> bool:
|
||
normalized = _text_key(text)
|
||
if not normalized:
|
||
return False
|
||
markers = (
|
||
"интересует",
|
||
"как работает",
|
||
"какой график",
|
||
"мне надо",
|
||
"мне нужен",
|
||
"нужно",
|
||
"подключ",
|
||
"подскажите",
|
||
"узнать",
|
||
"хочу",
|
||
)
|
||
return any(marker in normalized for marker in markers)
|
||
|
||
|
||
def render_context_summary_text(raw: str | dict[str, Any] | None) -> str:
|
||
summary = load_context_summary(raw)
|
||
parts: list[str] = []
|
||
if summary["customer_name"]:
|
||
parts.append(f"Customer name: {summary['customer_name']}")
|
||
if summary["language"]:
|
||
parts.append(f"Language: {summary['language']}")
|
||
if summary["active_intent"]:
|
||
parts.append(f"Active intent: {summary['active_intent']}")
|
||
if summary["active_request_text"]:
|
||
parts.append(f"Active request: {summary['active_request_text']}")
|
||
facts = summary["confirmed_facts"]
|
||
fact_parts = []
|
||
if facts.get("city"):
|
||
fact_parts.append(f"city={facts['city']}")
|
||
if facts.get("branch_hint"):
|
||
fact_parts.append(f"branch={facts['branch_hint']}")
|
||
if facts.get("service_hint"):
|
||
fact_parts.append(f"service={facts['service_hint']}")
|
||
if facts.get("request_number"):
|
||
fact_parts.append(f"request_number={facts['request_number']}")
|
||
if fact_parts:
|
||
parts.append("Confirmed facts: " + ", ".join(fact_parts))
|
||
if summary["open_slots"]:
|
||
parts.append("Open slots: " + ", ".join(summary["open_slots"]))
|
||
if summary["scenario_state"]:
|
||
parts.append(f"Scenario state: {summary['scenario_state']}")
|
||
return " | ".join(parts)
|
||
|
||
|
||
def context_summary_context_texts(raw: str | dict[str, Any] | None) -> list[str]:
|
||
summary = load_context_summary(raw)
|
||
result: list[str] = []
|
||
if summary["active_request_text"]:
|
||
result.append(summary["active_request_text"])
|
||
facts = summary["confirmed_facts"]
|
||
if facts.get("city"):
|
||
result.append(f"город {facts['city']}")
|
||
if facts.get("branch_hint"):
|
||
result.append(facts["branch_hint"])
|
||
if facts.get("service_hint"):
|
||
result.append(facts["service_hint"])
|
||
if summary["active_intent"] == "schedule":
|
||
result.append("график работы")
|
||
elif summary["active_intent"] == "address":
|
||
result.append("филиал или адрес")
|
||
elif summary["active_intent"] == "price":
|
||
result.append("тариф или стоимость")
|
||
elif summary["active_intent"] == "status":
|
||
result.append("статус заявки")
|
||
elif summary["active_intent"] == "problem":
|
||
result.append("техническая проблема")
|
||
seen: set[str] = set()
|
||
deduped: list[str] = []
|
||
for item in result:
|
||
normalized = _text_key(item)
|
||
if not normalized or normalized in seen:
|
||
continue
|
||
seen.add(normalized)
|
||
deduped.append(str(item).strip())
|
||
return deduped
|
||
|
||
|
||
def update_context_summary_from_user_turn(
|
||
raw: str | dict[str, Any] | None,
|
||
*,
|
||
channel: str,
|
||
language: str | None,
|
||
customer_name: str | None,
|
||
text: str | None,
|
||
now: str | None = None,
|
||
) -> dict[str, Any]:
|
||
summary = load_context_summary(raw)
|
||
summary["channel"] = str(channel or summary["channel"] or "").strip() or summary["channel"]
|
||
if language:
|
||
summary["language"] = str(language).strip()
|
||
if customer_name:
|
||
summary["customer_name"] = str(customer_name).strip()
|
||
compact_text = re.sub(r"\s+", " ", str(text or "").strip())
|
||
if not compact_text or _is_low_signal_text(compact_text):
|
||
if now:
|
||
summary["updated_at"] = now
|
||
return summary
|
||
city = _extract_city(compact_text)
|
||
branch_hint = _extract_branch_hint(compact_text)
|
||
request_number = _extract_request_number(compact_text)
|
||
service_hint, inferred_intent = _extract_service_hint(compact_text)
|
||
should_refresh_request_text = (
|
||
bool(city or branch_hint or request_number or service_hint)
|
||
or _looks_like_request_statement(compact_text)
|
||
or not str(summary.get("active_request_text") or "").strip()
|
||
)
|
||
if should_refresh_request_text:
|
||
summary["active_request_text"] = compact_text[:500]
|
||
if city:
|
||
summary["confirmed_facts"]["city"] = city
|
||
if branch_hint:
|
||
summary["confirmed_facts"]["branch_hint"] = branch_hint
|
||
if request_number:
|
||
summary["confirmed_facts"]["request_number"] = request_number
|
||
if service_hint:
|
||
summary["confirmed_facts"]["service_hint"] = service_hint
|
||
if inferred_intent:
|
||
summary["active_intent"] = inferred_intent
|
||
summary["scenario_state"] = f"{inferred_intent}_in_progress"
|
||
summary["open_slots"] = _intent_open_slots(summary["active_intent"], summary["confirmed_facts"])
|
||
if now:
|
||
summary["updated_at"] = now
|
||
return summary
|
||
|
||
|
||
def update_context_summary_from_assistant_turn(
|
||
raw: str | dict[str, Any] | None,
|
||
*,
|
||
language: str | None,
|
||
customer_name: str | None,
|
||
reply_text: str | None,
|
||
decision_intent: str | None,
|
||
needs_handoff: bool,
|
||
handoff_reason: str | None,
|
||
now: str | None = None,
|
||
) -> dict[str, Any]:
|
||
summary = load_context_summary(raw)
|
||
if language:
|
||
summary["language"] = str(language).strip()
|
||
if customer_name:
|
||
summary["customer_name"] = str(customer_name).strip()
|
||
if decision_intent and decision_intent not in {"clarification", "kb_answer", "unknown"}:
|
||
summary["active_intent"] = str(decision_intent).strip()
|
||
if needs_handoff:
|
||
summary["scenario_state"] = "handoff_requested"
|
||
summary["open_slots"] = []
|
||
summary["requested_clarifications"] = []
|
||
else:
|
||
clarifications = _requested_clarifications_from_reply(reply_text)
|
||
fallback_slots = _intent_open_slots(summary["active_intent"], summary["confirmed_facts"])
|
||
merged_open_slots, merged_requested_clarifications = _merge_open_slots(
|
||
list(summary["open_slots"]),
|
||
clarified_slots=clarifications,
|
||
fallback_slots=fallback_slots,
|
||
)
|
||
summary["open_slots"] = merged_open_slots
|
||
summary["requested_clarifications"] = merged_requested_clarifications
|
||
if summary["active_intent"]:
|
||
summary["scenario_state"] = f"{summary['active_intent']}_in_progress"
|
||
if handoff_reason and needs_handoff:
|
||
summary["active_request_text"] = summary["active_request_text"] or str(handoff_reason).strip()
|
||
if now:
|
||
summary["updated_at"] = now
|
||
return summary
|