141 lines
5.0 KiB
Python
141 lines
5.0 KiB
Python
"""投顾聊天意图分类：LLM 优先，规则识别兜底。"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import re
|
|
from dataclasses import dataclass
|
|
|
|
from common.common_const import (
|
|
AGENT_INTENT_CASUAL_CHAT,
|
|
AGENT_INTENT_DATA_QUERY,
|
|
AGENT_INTENT_DIALOGUE_SCRIPT,
|
|
AGENT_INTENT_FUND_ANALYSIS,
|
|
AGENT_INTENT_REBALANCE,
|
|
AGENT_INTENT_RECOMMEND,
|
|
)
|
|
from agent.advisor_agent.intent.recognizer import recognize_advisor_intent
|
|
|
|
|
|
VALID_INTENTS = frozenset({
|
|
AGENT_INTENT_RECOMMEND,
|
|
AGENT_INTENT_REBALANCE,
|
|
AGENT_INTENT_FUND_ANALYSIS,
|
|
AGENT_INTENT_DIALOGUE_SCRIPT,
|
|
AGENT_INTENT_DATA_QUERY,
|
|
AGENT_INTENT_CASUAL_CHAT,
|
|
})
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class IntentClassification:
|
|
intent: str
|
|
confidence: float
|
|
source: str
|
|
reason: str = ""
|
|
|
|
|
|
_CLASSIFIER_PROMPT = """你是基金投顾工作台的意图分类器，只负责分类，不回答用户问题。
|
|
只能从以下分类中选择一个：
|
|
- recommend：基金推荐、组合配置或投资方案
|
|
- rebalance：组合偏离、调仓、再平衡、仓位调整
|
|
- fund_analysis：单只基金分析、净值、收益、回撤、波动率、夏普比率
|
|
- dialogue-script：给客户准备沟通话术、解释、安抚、投诉或风险提醒
|
|
- data_query：查询客户持仓、资产、余额、收益、交易、账户明细
|
|
- casual_chat：问候、闲聊、感谢、身份询问或无法归入业务分类的内容
|
|
|
|
只输出 JSON，不要 Markdown，不要额外文字：
|
|
{"intent":"分类值","confidence":0到1之间的数字,"reason":"不超过30字的原因"}
|
|
"""
|
|
|
|
_RECOMMENDATION_TERMS = ("推荐", "组合建议", "配置建议", "买什么", "适合配置", "筛选基金", "投资方案")
|
|
_SCOPE_ERROR_MESSAGE = "投顾范围查询仅支持客户数据查询"
|
|
|
|
|
|
def _rule_fallback(query: str | None) -> IntentClassification:
|
|
intent = recognize_advisor_intent(query)
|
|
if intent:
|
|
return IntentClassification(intent=intent, confidence=0.72, source="rule", reason="关键词规则匹配")
|
|
return IntentClassification(intent=AGENT_INTENT_CASUAL_CHAT, confidence=0.0, source="fallback", reason="无法匹配业务意图")
|
|
|
|
|
|
def _parse_model_result(raw: str) -> IntentClassification | None:
|
|
text = raw.strip()
|
|
fenced = re.search(r"\{.*\}", text, re.DOTALL)
|
|
if fenced:
|
|
text = fenced.group(0)
|
|
try:
|
|
payload = json.loads(text)
|
|
except (TypeError, json.JSONDecodeError):
|
|
return None
|
|
intent = payload.get("intent")
|
|
if intent not in VALID_INTENTS:
|
|
return None
|
|
try:
|
|
confidence = max(0.0, min(1.0, float(payload.get("confidence", 0.0))))
|
|
except (TypeError, ValueError):
|
|
confidence = 0.0
|
|
return IntentClassification(
|
|
intent=intent,
|
|
confidence=confidence,
|
|
source="llm",
|
|
reason=str(payload.get("reason") or "模型分类"),
|
|
)
|
|
|
|
|
|
async def classify_advisor_intent(
|
|
query: str | None,
|
|
llm_client=None,
|
|
*,
|
|
explicit_intent: str | None = None,
|
|
timeout: float = 2.0,
|
|
) -> IntentClassification:
|
|
"""分类用户意图；显式意图兼容旧客户端，模型失败时安全回退规则。"""
|
|
if explicit_intent in VALID_INTENTS:
|
|
return IntentClassification(explicit_intent, 1.0, "explicit", "客户端显式指定")
|
|
if not query or not query.strip():
|
|
return IntentClassification("", 0.0, "fallback", "空输入")
|
|
if re.sub(r"\s+", "", query) == _SCOPE_ERROR_MESSAGE:
|
|
return IntentClassification(
|
|
AGENT_INTENT_CASUAL_CHAT,
|
|
1.0,
|
|
"rule",
|
|
"识别为系统提示文本而非业务查询",
|
|
)
|
|
if llm_client is not None:
|
|
try:
|
|
raw = await asyncio.wait_for(
|
|
llm_client.chat(
|
|
[
|
|
{"role": "system", "content": _CLASSIFIER_PROMPT},
|
|
{"role": "user", "content": query.strip()},
|
|
],
|
|
temperature=0,
|
|
max_tokens=120,
|
|
),
|
|
timeout=timeout,
|
|
)
|
|
parsed = _parse_model_result(raw)
|
|
if parsed is not None:
|
|
# LLM 偶尔会把“查询持仓/资产”等只读请求误判为推荐；
|
|
# 对明确的查询动作以规则结果为准，避免误进入草稿生成分支。
|
|
rule_intent = recognize_advisor_intent(query)
|
|
if (
|
|
rule_intent == AGENT_INTENT_DATA_QUERY
|
|
and parsed.intent != AGENT_INTENT_DATA_QUERY
|
|
and not any(term in (query or "") for term in _RECOMMENDATION_TERMS)
|
|
):
|
|
return IntentClassification(
|
|
intent=rule_intent,
|
|
confidence=max(parsed.confidence, 0.9),
|
|
source="rule_override",
|
|
reason="明确查询类关键词覆盖模型误判",
|
|
)
|
|
return parsed
|
|
except Exception:
|
|
pass
|
|
return _rule_fallback(query)
|
|
|
|
|
|
__all__ = ["IntentClassification", "classify_advisor_intent"]
|