ylhan
20cfaa6ef7
这是进入 Pipecat Flows 改造前的回滚点。 提交时已核对本地与线上代码 md5 一致（engine.py 7e97bad8、 bot_server.py 1cf543ae），所以这个提交就是当前线上状态。 内容（此前一直在线上运行但未提交的改动）： - 姓名档案：用户自报"我叫林"同步写 confirmed 姓名，重连后恢复 - 疑问句防误写："谁吗"之类的疑问句不再被覆盖成姓名 - 新增 test_profile_memory.py（姓名档案专项回归） - 新增 test_profile_live.py（需要真实 LLM 的现场测试） - 搜索、身份、记忆测试的相应补充 背景：下一步要按 P1-Pipecat Flows Agent化与记忆改造交接.md 渐进引入 Pipecat Flows，交接文档明确要求先保护基线、不得 reset/checkout/覆盖现有改动。此提交即该基线。
113 lines
4.0 KiB
Python
113 lines
4.0 KiB
Python
"""身份抽象层。
|
|
|
|
方案 §6 的核心约束，逐条落进接口：
|
|
|
|
* "首版可靠身份入口：手动/对话确认切换到已建档的 user_id；
|
|
无法确认时使用 unknown/guest。"
|
|
|
|
* **"看到熟人不等于确认他正在说话"** —— 人脸只提供候选，
|
|
不自动切换身份。这条最容易做错：摄像头看到谁就切谁的记忆，
|
|
结果是"路人经过就把你的私人记忆念出来"。
|
|
|
|
* "显式报名字用于个性化档案选择，**不是身份认证**，不能据此解锁敏感信息"
|
|
—— 所以 trust_level 与 user_id 分开存，将来加认证时不用改偏好存储。
|
|
|
|
* "预留 IdentityProvider 接口：输入时间戳、track_id、候选身份、置信度与来源；
|
|
输出确认的 user_id 或 unknown。后续声纹模块可接入此接口，不改记忆表结构。"
|
|
|
|
* "视觉 track_id、会话 session_id、永久 user_id 分开" ——
|
|
所以 IdentityClaim 里 track_id 是线索，不是身份。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from abc import ABC, abstractmethod
|
|
from dataclasses import dataclass, field
|
|
from typing import Sequence
|
|
|
|
#: 无法确认身份时使用的固定 id。永远不落私人记忆。
|
|
GUEST_USER_ID = "guest"
|
|
|
|
#: 信任等级。与 user_id 分开，将来加认证不改这里。
|
|
TRUST_NORMAL = "normal"
|
|
TRUST_GUEST = "guest"
|
|
|
|
|
|
@dataclass
|
|
class IdentityClaim:
|
|
"""一条身份线索。
|
|
|
|
source 取值（首版只用 manual，其余留给后续）：
|
|
"manual" 用户显式切换/确认——**唯一可单独决定身份的来源**
|
|
"face" 人脸识别候选（本地特征，需用户主动录入）
|
|
"client_login" 客户端登录态（手机 App / 浏览器）
|
|
"voiceprint" 声纹（后续阶段）
|
|
"""
|
|
source: str
|
|
candidate_user_id: str | None = None
|
|
confidence: float = 0.0
|
|
track_id: str | None = None
|
|
timestamp: float = field(default_factory=time.time)
|
|
note: str = ""
|
|
|
|
#: 只有这些来源可以单独把身份定下来。人脸不在其中——
|
|
#: 方案明确要求人脸仅作候选、经确认后才启用对应记忆。
|
|
AUTHORITATIVE_SOURCES = ("manual", "client_login")
|
|
|
|
@property
|
|
def is_authoritative(self) -> bool:
|
|
return self.source in self.AUTHORITATIVE_SOURCES
|
|
|
|
|
|
@dataclass
|
|
class IdentityDecision:
|
|
user_id: str
|
|
trust_level: str
|
|
confirmed: bool # True = 经确认；False = 默认访客
|
|
reason: str
|
|
# 用户可见称呼，与内部稳定 user_id 分开；自报姓名不能提升信任等级。
|
|
display_name: str = ""
|
|
#: 未被采纳的候选，保留下来供上层提示"是否要切换成 X"
|
|
suggestions: list[IdentityClaim] = field(default_factory=list)
|
|
|
|
@property
|
|
def is_guest(self) -> bool:
|
|
return not self.confirmed
|
|
|
|
|
|
class IdentityProvider(ABC):
|
|
"""身份裁决者。
|
|
|
|
实现要点：**宁可拒判，不要猜**。判错的代价是把 A 的私人记忆
|
|
用在 B 身上，比"这次不认识"严重得多。
|
|
"""
|
|
|
|
@abstractmethod
|
|
def current(self) -> IdentityDecision:
|
|
"""当前身份。任何时候都必须能回答。"""
|
|
|
|
@abstractmethod
|
|
def switch(self, user_id: str, *, display_name: str = "",
|
|
reason: str = "显式切换") -> IdentityDecision:
|
|
"""显式切换身份——首版唯一可靠的入口。"""
|
|
|
|
@abstractmethod
|
|
def resolve(self, claims: Sequence[IdentityClaim]) -> IdentityDecision:
|
|
"""根据一组线索裁决身份。"""
|
|
|
|
@abstractmethod
|
|
def reset(self, reason: str = "回到待机") -> IdentityDecision:
|
|
"""回到访客身份，并清掉上一人的身份状态。
|
|
|
|
方案 §8 要求"切换身份清理上一人的私有检索上下文"——
|
|
调用方在收到身份变化后必须重新构造 system 消息（含记忆），
|
|
不能沿用上一轮的。
|
|
"""
|
|
|
|
|
|
__all__ = [
|
|
"IdentityProvider", "IdentityClaim", "IdentityDecision",
|
|
"GUEST_USER_ID", "TRUST_NORMAL", "TRUST_GUEST",
|
|
]
|