55 lines
2.2 KiB
Python
55 lines
2.2 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Dict
|
|
|
|
|
|
class ResponsePlanner:
|
|
def choose_reply_mode(self, trigger: Dict, flow_state: str) -> str:
|
|
if trigger.get("is_question"):
|
|
return "qa_with_context" if flow_state in {"engaged", "deep_engaged"} else "qa_fast"
|
|
if trigger.get("is_followup"):
|
|
return "qa_with_context"
|
|
if trigger.get("is_social_call"):
|
|
return "social_short"
|
|
if trigger.get("is_returning_member"):
|
|
return "social_short"
|
|
return "social_short" if flow_state in {"deep_engaged"} else "refuse_or_skip"
|
|
|
|
def should_reply(
|
|
self,
|
|
trigger: Dict,
|
|
flow_state: str,
|
|
allow_proactive: bool,
|
|
acceptance_state: str = "neutral",
|
|
conversation_hints: Dict | None = None,
|
|
) -> bool:
|
|
conversation_hints = conversation_hints or {}
|
|
trigger_type = str(trigger.get("trigger_type", "") or "")
|
|
if trigger.get("is_at") or trigger_type == "at_trigger":
|
|
return True
|
|
if trigger_type == "quote_followup_trigger" and trigger.get("is_directed"):
|
|
return True
|
|
if trigger.get("is_question") and conversation_hints.get("has_recent_human_solver") and flow_state != "deep_engaged":
|
|
return False
|
|
if trigger.get("is_question"):
|
|
if trigger.get("is_directed"):
|
|
return True
|
|
if acceptance_state == "warm" and flow_state == "deep_engaged" and trigger.get("priority", 0) >= 0.95:
|
|
return True
|
|
return False
|
|
if trigger.get("is_followup"):
|
|
return False
|
|
if trigger.get("is_social_call"):
|
|
if acceptance_state == "cold":
|
|
return False
|
|
return flow_state in {"engaged", "deep_engaged"}
|
|
if trigger.get("is_returning_member"):
|
|
return False
|
|
if not allow_proactive:
|
|
return False
|
|
if acceptance_state == "cold":
|
|
return False
|
|
if acceptance_state == "neutral":
|
|
return flow_state in {"deep_engaged"} and trigger.get("priority", 0) >= 0.8
|
|
return flow_state in {"engaged", "deep_engaged"} and trigger.get("priority", 0) >= 0.65
|