group_fqcd_jr/app/service/agent/customer_service_routing.py

"""一期客服的确定性路由，先处理安全和边界，再允许公开知识检索。"""

from collections.abc import Sequence
from dataclasses import dataclass


@dataclass(frozen=True)
class CustomerServiceRoute:
    intent: str
    knowledge_intents: tuple[str, ...] = ()
    is_chitchat: bool = False
    requires_context: bool = False


class CustomerServiceIntentRouter:
    _SECURITY_KEYWORDS = ("验证码", "密码泄露", "被盗", "诈骗", "非本人交易")
    # 明确的凭据披露和提示词注入必须在知识检索前拦截，避免把攻击内容当作普通 FAQ。
    _SECURITY_DISCLOSURE_PHRASES = ("密码是", "密码为", "我的密码", "验证码是", "验证码为")
    _PROMPT_INJECTION_KEYWORDS = (
        "忽略之前", "忽略所有规则", "系统提示词", "开发者消息", "泄露提示词", "越过限制",
        "不要遵守规则", "显示内部指令",
    )
    _COMPLIANCE_KEYWORDS = ("推荐", "收益最高", "稳赚", "保本", "帮我买", "替我交易")
    _ACCOUNT_KEYWORDS = ("持仓", "收益", "订单", "定投", "银行卡", "风险测评", "投诉进度")
    _HUMAN_TRANSFER_KEYWORDS = ("转人工", "人工客服", "投诉", "赔偿", "法律", "纠纷")
    _POLICY_KEYWORDS = (
        "申购", "赎回", "到账", "费率", "手续费", "确认份额", "交易日", "分红", "规则", "政策"
    )
    _PRODUCT_KEYWORDS = (
        "产品", "基金代码", "基金经理", "份额类别", "a类", "c类", "净值", "风险等级"
    )
    _CHITCHAT_MESSAGES = frozenset({
        "你好", "您好", "嗨", "哈喽", "在吗", "谢谢", "谢谢你", "再见", "拜拜",
        "你是谁", "你叫什么", "你今天开心吗",
    })
    _CHITCHAT_PHRASES = ("今天天气", "讲个笑话", "你几岁", "你开心吗", "你忙吗")
    _REFERENCE_PHRASES = ("这个", "那个", "它的", "刚才", "上面", "前面", "这只", "那只")

    @classmethod
    def classify(cls, message: str) -> CustomerServiceRoute:
        normalized = message.strip().lower()
        if (cls._contains(normalized, cls._SECURITY_KEYWORDS)
                or cls._contains(normalized, cls._SECURITY_DISCLOSURE_PHRASES)):
            return CustomerServiceRoute(intent="security_notice")
        if cls._contains(normalized, cls._PROMPT_INJECTION_KEYWORDS):
            return CustomerServiceRoute(intent="compliance_refusal")
        if cls._contains(normalized, cls._COMPLIANCE_KEYWORDS):
            return CustomerServiceRoute(intent="compliance_refusal")
        if cls._contains(normalized, cls._ACCOUNT_KEYWORDS):
            return CustomerServiceRoute(intent="account_entry")
        if cls._contains(normalized, cls._HUMAN_TRANSFER_KEYWORDS):
            return CustomerServiceRoute(intent="human_transfer")
        if cls._is_chitchat(normalized):
            return CustomerServiceRoute(intent="chitchat", is_chitchat=True)
        if cls._contains(normalized, cls._POLICY_KEYWORDS):
            return CustomerServiceRoute(
                intent="public_knowledge", knowledge_intents=("policy_explain",),
                requires_context=cls._contains(normalized, cls._REFERENCE_PHRASES),
            )
        if cls._contains(normalized, cls._PRODUCT_KEYWORDS):
            return CustomerServiceRoute(
                intent="public_knowledge", knowledge_intents=("product_inquiry",),
                requires_context=cls._contains(normalized, cls._REFERENCE_PHRASES),
            )
        return CustomerServiceRoute(intent="public_knowledge", knowledge_intents=("faq",))

    @classmethod
    def chitchat_streak(cls, prior_messages: Sequence[str], message: str) -> int:
        """返回当前消息在同一会话中连续闲聊的次数，最大只需记录到第五句。"""
        if not cls._is_chitchat(message.strip().lower()):
            return 0
        streak = 1
        for prior_message in reversed(prior_messages):
            if not cls._is_chitchat(prior_message.strip().lower()):
                break
            streak += 1
            if streak == 5:
                break
        return streak

    @staticmethod
    def _contains(message: str, keywords: tuple[str, ...]) -> bool:
        return any(keyword in message for keyword in keywords)

    @classmethod
    def _is_chitchat(cls, message: str) -> bool:
        return message in cls._CHITCHAT_MESSAGES or cls._contains(message, cls._CHITCHAT_PHRASES)