Mutual_Fund/tool/confidence_rank.py

"""客服记忆候选的综合置信分重排工具。"""

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"]