Mutual_Fund/tool/confidence.py

"""客服对话记忆的基础置信度计算工具。"""

from __future__ import annotations

from typing import Any


class BaseConfidenceCalcTool:
    """根据来源、证据和时间计算单条客服记忆的长期置信度。"""

    SOURCE_INITIAL = {
        "dialogue_confirmed": 0.75,
        "dialogue_stated": 0.50,
        "dialogue_inferred": 0.45,
    }
    MEMORY_THRESHOLDS = {
        "PROFILE_FACT": 0.75,
        "CUSTOMER_PREFERENCE": 0.65,
        "INVESTMENT_GOAL": 0.70,
        "SERVICE_FACT": 0.75,
    }
    DEFAULT_THRESHOLD = 0.80
    VERSION = "confidence-v2"

    def calc(
        self,
        tag: str,
        source: str,
        evidence_count: int,
        age_days: int,
    ) -> float:
        """计算基础置信度分数，返回范围为 0 到 1 的浮点数。"""
        self._validate(tag, source, evidence_count, age_days)
        base = self.SOURCE_INITIAL[source]
        gain = min(evidence_count * 0.05, 0.30)
        decay = max(0.80, 1 - age_days / 365 * 0.20)
        return max(0.0, min(1.0, (base + gain) * decay))

    def evaluate(
        self,
        *,
        tag: str,
        source: str,
        evidence_count: int,
        age_days: int,
        memory_type: str | None = None,
        threshold: float | None = None,
    ) -> dict[str, Any]:
        """返回可供记忆模块保存的完整置信度评估结果。"""
        score = self.calc(tag, source, evidence_count, age_days)
        if threshold is None:
            threshold = self.MEMORY_THRESHOLDS.get(
                memory_type or "", self.DEFAULT_THRESHOLD
            )
        if not 0.0 <= threshold <= 1.0:
            raise ValueError("threshold 必须在 [0.0, 1.0] 范围内")
        base = self.SOURCE_INITIAL[source]
        status = "confirmed" if score >= threshold else "candidate"
        return {
            "source_confidence": base,
            "confidence": score,
            "status": status,
            "evidence_count": evidence_count,
            "age_days": age_days,
            "threshold": threshold,
            "confidence_reason": self._reason(
                source, evidence_count, age_days
            ),
            "confidence_version": self.VERSION,
        }

    def batch_calc(self, tags: list[dict[str, Any]]) -> list[float]:
        """批量计算基础分数。"""
        return [self.calc(**tag) for tag in tags]

    def batch_evaluate(self, items: list[dict[str, Any]]) -> list[dict[str, Any]]:
        """批量生成完整评估结果。"""
        return [self.evaluate(**item) for item in items]

    @classmethod
    def _validate(
        cls,
        tag: str,
        source: str,
        evidence_count: int,
        age_days: int,
    ) -> None:
        """校验工具输入，避免非法计数污染记忆分数。"""
        if not tag or not tag.strip():
            raise ValueError("tag 不能为空")
        if source not in cls.SOURCE_INITIAL:
            raise ValueError(f"不支持的客服对话来源: {source}")
        for name, value in (
            ("evidence_count", evidence_count),
            ("age_days", age_days),
        ):
            if not isinstance(value, int) or isinstance(value, bool) or value < 0:
                raise ValueError(f"{name} 必须是非负整数")

    @staticmethod
    def _reason(source: str, evidence_count: int, age_days: int) -> str:
        """生成便于审计和排查的评分原因。"""
        return (
            f"来源={source}; 支持证据={evidence_count}; "
            f"存在天数={age_days}; 采用证据增益和时间衰减"
        )


__all__ = ["BaseConfidenceCalcTool"]