"""首版身份实现:显式切换 + 访客兜底。 对应方案 §6:"首版可靠身份入口:手动/对话确认切换到已建档的 user_id; 无法确认时使用 unknown/guest。" 行为规则: 1. 显式切换(`switch`)立即生效,这是唯一能单独定身份的方式。 2. 人脸/声纹线索只生成**建议**,不自动改身份。上层可以选择性地 问一句"要切换成 X 吗",但不能替用户决定。 3. 没有任何线索时是访客,且访客不能落私人记忆(由记忆层用 owner_user_id 隔离保证)。 4. 切换身份返回一个"版本号",上层据此判断要不要丢弃上一人的上下文。 """ from __future__ import annotations import logging from typing import Sequence from .base import ( GUEST_USER_ID, TRUST_GUEST, TRUST_NORMAL, IdentityClaim, IdentityDecision, IdentityProvider, ) log = logging.getLogger(__name__) #: 人脸候选被采纳为"建议"的最低置信度。 #: 低于这个值连建议都不给——减少无意义的打扰。 SUGGEST_MIN_CONFIDENCE = 0.6 class ManualIdentityProvider(IdentityProvider): def __init__(self, initial_user: str | None = None): self._decision = IdentityDecision( user_id=initial_user or GUEST_USER_ID, trust_level=TRUST_NORMAL if initial_user else TRUST_GUEST, confirmed=bool(initial_user), reason="初始身份" if initial_user else "初始为访客", ) self._version = 0 # ---------------------------------------------------------- 查询 def current(self) -> IdentityDecision: return self._decision @property def version(self) -> int: """身份版本号。每次真正变更时 +1。 上层用它判断"上一轮的私有检索上下文是否过期"—— 方案 §8 要求切换身份时清理上一人的上下文。 """ return self._version # ---------------------------------------------------------- 变更 def switch(self, user_id: str, *, display_name: str = "", reason: str = "显式切换") -> IdentityDecision: if not user_id: raise ValueError("user_id 不能为空") prev = self._decision.user_id self._decision = IdentityDecision( user_id=user_id, trust_level=TRUST_NORMAL, confirmed=True, reason=reason, ) if prev != user_id: self._version += 1 log.info("身份切换 %s → %s(%s)", prev, user_id, reason) return self._decision def reset(self, reason: str = "回到待机") -> IdentityDecision: if self._decision.user_id != GUEST_USER_ID: self._version += 1 self._decision = IdentityDecision( user_id=GUEST_USER_ID, trust_level=TRUST_GUEST, confirmed=False, reason=reason, ) return self._decision def resolve(self, claims: Sequence[IdentityClaim]) -> IdentityDecision: """裁决。 顺序很重要:先看有没有权威来源(显式切换 / 客户端登录态), 有就采用;人脸之类的只作为建议附在后面。 """ authoritative = [c for c in claims if c.is_authoritative and c.candidate_user_id] suggestions = [ c for c in claims if not c.is_authoritative and c.candidate_user_id and c.confidence >= SUGGEST_MIN_CONFIDENCE ] if authoritative: # 同一批里有多个权威来源且互相矛盾——宁可不切,报出来让人处理 ids = {c.candidate_user_id for c in authoritative} if len(ids) > 1: log.warning("权威身份线索互相矛盾 %s,保持当前身份", ids) d = self._decision d.reason = f"线索矛盾 {ids},保持 {d.user_id}" d.suggestions = suggestions return d c = max(authoritative, key=lambda x: x.confidence) assert c.candidate_user_id is not None self.switch(c.candidate_user_id, reason=f"来源 {c.source}") else: # 没有权威来源:即便人脸认出来了也不切,只给建议 self._decision.suggestions = suggestions if suggestions: self._decision.reason = ( f"仅有人脸候选 {suggestions[0].candidate_user_id}" f"(置信度 {suggestions[0].confidence:.2f}),未自动切换" ) return self._decision __all__ = ["ManualIdentityProvider", "SUGGEST_MIN_CONFIDENCE"]