89 lines
3.1 KiB
Python
89 lines
3.1 KiB
Python
"""客服记忆候选的综合置信分重排工具。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from copy import deepcopy
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
|
|
class FinalConfidenceRankTool:
|
|
"""仅服务客服记忆召回的临时重排工具。"""
|
|
|
|
WEIGHTS = {
|
|
"semantic": 0.30,
|
|
"timeliness": 0.20,
|
|
"accuracy": 0.20,
|
|
"base": 0.30,
|
|
}
|
|
INVALID_STATUSES = frozenset({"expired", "rejected", "archived"})
|
|
|
|
def rank(
|
|
self,
|
|
memory_units: list[dict[str, Any]],
|
|
*,
|
|
top_k: int | None = None,
|
|
now: datetime | None = None,
|
|
) -> list[dict[str, Any]]:
|
|
"""过滤无效候选并按当前客服召回分数降序返回副本。"""
|
|
if top_k is not None and (not isinstance(top_k, int) or top_k < 0):
|
|
raise ValueError("top_k 必须是非负整数或 None")
|
|
now = now or datetime.now()
|
|
ranked = []
|
|
for original in memory_units:
|
|
unit = self._as_dict(original)
|
|
if self._is_invalid(unit, now):
|
|
continue
|
|
semantic = self._bounded(unit.get("semantic_similarity", 0.5), 0.5)
|
|
timeliness = self._calc_timeliness(unit.get("age_days", 0))
|
|
accuracy = self._bounded(unit.get("historical_accuracy", 0.5), 0.5)
|
|
base = self._bounded(unit.get("confidence", 0.5), 0.5)
|
|
final_score = (
|
|
self.WEIGHTS["semantic"] * semantic
|
|
+ self.WEIGHTS["timeliness"] * timeliness
|
|
+ self.WEIGHTS["accuracy"] * accuracy
|
|
+ self.WEIGHTS["base"] * base
|
|
)
|
|
unit["final_score"] = max(0.0, min(1.0, final_score))
|
|
ranked.append(unit)
|
|
ranked.sort(key=lambda item: item["final_score"], reverse=True)
|
|
return ranked if top_k is None else ranked[:top_k]
|
|
|
|
@staticmethod
|
|
def _as_dict(unit: dict[str, Any] | Any) -> dict[str, Any]:
|
|
"""兼容字典和 Pydantic/ORM 风格候选对象。"""
|
|
if isinstance(unit, dict):
|
|
return deepcopy(unit)
|
|
if hasattr(unit, "model_dump"):
|
|
return deepcopy(unit.model_dump())
|
|
return deepcopy(vars(unit))
|
|
|
|
@classmethod
|
|
def _is_invalid(cls, unit: dict[str, Any], now: datetime) -> bool:
|
|
"""过滤拒绝、归档和已过有效期的记忆。"""
|
|
if unit.get("status") in cls.INVALID_STATUSES:
|
|
return True
|
|
valid_until = unit.get("valid_until")
|
|
return valid_until is not None and valid_until <= now
|
|
|
|
@staticmethod
|
|
def _bounded(value: Any, default: float) -> float:
|
|
"""将缺失或异常评分转换为保守默认值。"""
|
|
try:
|
|
value = float(value)
|
|
except (TypeError, ValueError):
|
|
return default
|
|
return max(0.0, min(1.0, value))
|
|
|
|
@staticmethod
|
|
def _calc_timeliness(age_days: Any) -> float:
|
|
"""按每年 20% 计算平滑时效分，最低保留 0.8。"""
|
|
try:
|
|
age_days = max(0, int(age_days))
|
|
except (TypeError, ValueError):
|
|
age_days = 0
|
|
return max(0.80, 1 - age_days / 365 * 0.20)
|
|
|
|
|
|
__all__ = ["FinalConfidenceRankTool"]
|