feat:客户agent以及记忆模块功能开发

tool/confidence.py

@@ -0,0 +1,115 @@
+"""客服对话记忆的基础置信度计算工具。"""
+
+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-v1"
+
+    def calc(
+        self,
+        tag: str,
+        source: str,
+        evidence_count: int,
+        conflict_count: int,
+        age_days: int,
+    ) -> float:
+        """计算基础置信度分数，返回范围为 0 到 1 的浮点数。"""
+        self._validate(tag, source, evidence_count, conflict_count, age_days)
+        base = self.SOURCE_INITIAL[source]
+        gain = min(evidence_count * 0.05, 0.30)
+        penalty = min(conflict_count * 0.10, 0.50)
+        decay = max(0.80, 1 - age_days / 365 * 0.20)
+        return max(0.0, min(1.0, (base + gain - penalty) * decay))
+
+    def evaluate(
+        self,
+        *,
+        tag: str,
+        source: str,
+        evidence_count: int,
+        conflict_count: int,
+        age_days: int,
+        memory_type: str | None = None,
+        threshold: float | None = None,
+    ) -> dict[str, Any]:
+        """返回可供记忆模块保存的完整置信度评估结果。"""
+        score = self.calc(tag, source, evidence_count, conflict_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,
+            "conflict_count": conflict_count,
+            "age_days": age_days,
+            "threshold": threshold,
+            "confidence_reason": self._reason(
+                source, evidence_count, conflict_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,
+        conflict_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),
+            ("conflict_count", conflict_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, conflict_count: int, age_days: int) -> str:
+        """生成便于审计和排查的评分原因。"""
+        return (
+            f"来源={source}; 支持证据={evidence_count}; 冲突证据={conflict_count}; "
+            f"存在天数={age_days}; 采用证据增益、冲突惩罚和时间衰减"
+        )
+
+
+__all__ = ["BaseConfidenceCalcTool"]