116 lines
4.0 KiB
Python
116 lines
4.0 KiB
Python
"""投顾聊天入口的轻量意图识别。"""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
|
|
from common.common_const import (
|
|
AGENT_INTENT_DATA_QUERY,
|
|
AGENT_INTENT_DIALOGUE_SCRIPT,
|
|
AGENT_INTENT_FUND_ANALYSIS,
|
|
AGENT_INTENT_REBALANCE,
|
|
AGENT_INTENT_RECOMMEND,
|
|
)
|
|
|
|
|
|
_QUERY_ACTIONS = (
|
|
"查询",
|
|
"查一下",
|
|
"查看",
|
|
"统计",
|
|
"列出",
|
|
"显示",
|
|
"多少",
|
|
"有哪些",
|
|
"明细",
|
|
)
|
|
_DATA_TERMS = (
|
|
"持仓",
|
|
"资产",
|
|
"资金",
|
|
"收益",
|
|
"交易记录",
|
|
"申购",
|
|
"赎回",
|
|
"余额",
|
|
"市值",
|
|
"份额",
|
|
"客户数据",
|
|
"账户",
|
|
)
|
|
_REBALANCE_TERMS = ("调仓", "再平衡", "组合调整", "配置偏离", "偏离目标", "降低仓位", "增加仓位")
|
|
_FUND_ANALYSIS_TERMS = ("基金分析", "分析基金", "基金表现", "净值走势", "最大回撤", "夏普比率", "年化波动")
|
|
_DIALOGUE_TERMS = ("话术", "沟通", "怎么跟客户说", "如何向客户解释", "安抚客户", "投诉处理")
|
|
_RECOMMEND_TERMS = ("推荐", "组合建议", "配置建议", "买什么", "适合配置", "筛选基金", "投资方案")
|
|
_CUSTOMER_IDENTITY_TERMS = ("是谁", "姓名", "实名", "基本信息", "联系方式", "手机号")
|
|
_CUSTOMER_ROSTER_TERMS = ("名单", "列表", "几个")
|
|
_CUSTOMER_RISK_TERMS = (
|
|
"风险评级", "风险等级", "风险类型", "激进型", "进取型", "平衡型", "稳健型", "保守型",
|
|
"C1", "C2", "C3", "C4", "C5",
|
|
)
|
|
_IMPLICIT_QUERY_TERMS = (
|
|
"谁", "哪个", "哪些", "多少", "最大", "最多", "最小", "最少", "最高", "最低",
|
|
"合计", "总额", "总资产", "排名", "前几", "前十", "超过", "低于", "是否", "有没有",
|
|
)
|
|
_SCOPE_ERROR_MESSAGE = "投顾范围查询仅支持客户数据查询"
|
|
|
|
|
|
def _contains_any(text: str, terms: tuple[str, ...]) -> bool:
|
|
return any(term in text for term in terms)
|
|
|
|
|
|
def recognize_advisor_intent(query: str | None, explicit_intent: str | None = None) -> str | None:
|
|
"""返回当前投顾聊天应使用的意图；无法判断时返回 ``None``。
|
|
|
|
数据查询采用保守规则：必须命中查询动作或客户数据表达，且不能明显是
|
|
推荐、调仓或话术请求，避免把生成类请求送入 NL2SQL。
|
|
"""
|
|
if explicit_intent:
|
|
return explicit_intent
|
|
text = re.sub(r"\s+", "", query or "")
|
|
if not text:
|
|
return None
|
|
if text == _SCOPE_ERROR_MESSAGE:
|
|
return None
|
|
|
|
# 先处理最明确的任务词，避免“查询基金收益”被误判成普通数据查询。
|
|
if _contains_any(text, _REBALANCE_TERMS):
|
|
return AGENT_INTENT_REBALANCE
|
|
if _contains_any(text, _FUND_ANALYSIS_TERMS):
|
|
return AGENT_INTENT_FUND_ANALYSIS
|
|
if _contains_any(text, _DIALOGUE_TERMS):
|
|
return AGENT_INTENT_DIALOGUE_SCRIPT
|
|
|
|
has_action = any(term in text for term in _QUERY_ACTIONS)
|
|
has_data = any(term in text for term in _DATA_TERMS)
|
|
has_customer_identity = (
|
|
"客户" in text
|
|
and _contains_any(text, _CUSTOMER_IDENTITY_TERMS)
|
|
and not _contains_any(text, _RECOMMEND_TERMS)
|
|
)
|
|
if has_customer_identity:
|
|
return AGENT_INTENT_DATA_QUERY
|
|
# 客户名册类提问（我名下有哪些客户/客户名单）也属于数据查询。
|
|
has_customer_roster = (
|
|
"客户" in text
|
|
and (has_action or _contains_any(text, _CUSTOMER_ROSTER_TERMS))
|
|
and not _contains_any(text, _RECOMMEND_TERMS)
|
|
)
|
|
if has_customer_roster:
|
|
return AGENT_INTENT_DATA_QUERY
|
|
if "客户" in text and _contains_any(text, _CUSTOMER_RISK_TERMS) and not _contains_any(text, _RECOMMEND_TERMS):
|
|
return AGENT_INTENT_DATA_QUERY
|
|
if has_data and (
|
|
has_action
|
|
or "客户" in text
|
|
or "近一年" in text
|
|
or "本月" in text
|
|
or _contains_any(text, _IMPLICIT_QUERY_TERMS)
|
|
):
|
|
return AGENT_INTENT_DATA_QUERY
|
|
if _contains_any(text, _RECOMMEND_TERMS):
|
|
return AGENT_INTENT_RECOMMEND
|
|
return None
|
|
|
|
|
|
__all__ = ["recognize_advisor_intent"]
|