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/覆盖现有改动。此提交即该基线。
109 lines
4.3 KiB
Python
109 lines
4.3 KiB
Python
#!/usr/bin/env python3
|
|
"""身份层测试。
|
|
|
|
最关键的一条：**人脸认出来不等于切身份**。
|
|
方案 §6 写得很直白："看到熟人不等于确认他正在说话"。
|
|
做错的后果是——路人从摄像头前走过，助手就把你的私人记忆念了出来。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent))
|
|
|
|
from app.identity import ( # noqa: E402
|
|
GUEST_USER_ID, IdentityClaim, ManualIdentityProvider,
|
|
)
|
|
|
|
OK, FAIL = [], []
|
|
|
|
|
|
def check(name: str, cond: bool, detail: str = "") -> None:
|
|
(OK if cond else FAIL).append(name)
|
|
print(f" [{'PASS' if cond else 'FAIL'}] {name}" + (f" — {detail}" if detail else ""))
|
|
|
|
|
|
def main() -> int:
|
|
print("=== 1. 初始状态 ===")
|
|
p = ManualIdentityProvider()
|
|
d = p.current()
|
|
check("默认是访客", d.user_id == GUEST_USER_ID and d.is_guest, d.user_id)
|
|
check("访客信任等级独立", d.trust_level == "guest")
|
|
|
|
print("\n=== 2. 显式切换（首版唯一可靠入口）===")
|
|
d = p.switch("u_ylhan", reason="对话确认")
|
|
check("切换生效", d.user_id == "u_ylhan" and d.confirmed)
|
|
check("信任等级为 normal", d.trust_level == "normal")
|
|
named = p.switch("u_ylhan", display_name="林", reason="用户自报")
|
|
check("显示名与内部 ID 分离", named.display_name == "林")
|
|
check("同一身份空显示名不覆盖", p.switch("u_ylhan").display_name == "林")
|
|
check("版本号 +1", p.version == 1, str(p.version))
|
|
v = p.version
|
|
p.switch("u_ylhan") # 切到同一个人
|
|
check("切到同一人不增版本号", p.version == v, f"{v} → {p.version}")
|
|
|
|
print("\n=== 3. 人脸候选不得自动切换（关键）===")
|
|
p2 = ManualIdentityProvider("u_ylhan")
|
|
before = p2.current().user_id
|
|
d = p2.resolve([IdentityClaim(source="face", candidate_user_id="u_other",
|
|
confidence=0.95, track_id="track_7")])
|
|
check("身份未被自动切换", d.user_id == before, f"{before} → {d.user_id}")
|
|
check("但仍保留建议", any(s.candidate_user_id == "u_other" for s in d.suggestions),
|
|
d.reason)
|
|
check("track_id 未污染 user_id", d.user_id != "track_7")
|
|
|
|
# 访客状态下也不应变
|
|
p3 = ManualIdentityProvider()
|
|
d3 = p3.resolve([IdentityClaim(source="face", candidate_user_id="u_ylhan",
|
|
confidence=0.99)])
|
|
check("访客状态也不自动切换", d3.user_id == GUEST_USER_ID and d3.is_guest, d3.reason)
|
|
|
|
print("\n=== 4. 低置信度连建议都不给 ===")
|
|
p4 = ManualIdentityProvider()
|
|
d4 = p4.resolve([IdentityClaim(source="face", candidate_user_id="u_x", confidence=0.3)])
|
|
check("低置信度无建议", len(d4.suggestions) == 0, f"{len(d4.suggestions)} 条")
|
|
|
|
print("\n=== 5. 权威来源可以切换 ===")
|
|
p5 = ManualIdentityProvider()
|
|
d5 = p5.resolve([IdentityClaim(source="client_login", candidate_user_id="u_phone",
|
|
confidence=1.0)])
|
|
check("客户端登录态可切身份", d5.user_id == "u_phone" and d5.confirmed, d5.reason)
|
|
|
|
d5b = p5.resolve([IdentityClaim(source="manual", candidate_user_id="u_ylhan")])
|
|
check("显式来源可切身份", d5b.user_id == "u_ylhan")
|
|
|
|
print("\n=== 6. 权威线索互相矛盾时不切 ===")
|
|
p6 = ManualIdentityProvider("u_ylhan")
|
|
d6 = p6.resolve([
|
|
IdentityClaim(source="manual", candidate_user_id="u_a", confidence=1.0),
|
|
IdentityClaim(source="client_login", candidate_user_id="u_b", confidence=1.0),
|
|
])
|
|
check("矛盾时保持原身份", d6.user_id == "u_ylhan", d6.reason)
|
|
|
|
print("\n=== 7. reset 回访客 ===")
|
|
p7 = ManualIdentityProvider("u_ylhan")
|
|
v7 = p7.version
|
|
d7 = p7.reset()
|
|
check("回到访客", d7.user_id == GUEST_USER_ID and d7.is_guest)
|
|
check("版本号 +1（上层据此丢弃旧上下文）", p7.version == v7 + 1,
|
|
f"{v7} → {p7.version}")
|
|
|
|
print("\n=== 8. 空 user_id 应报错而非静默接受 ===")
|
|
try:
|
|
p7.switch("")
|
|
check("拒绝空 user_id", False, "没报错")
|
|
except ValueError:
|
|
check("拒绝空 user_id", True)
|
|
|
|
print("\n" + "=" * 62)
|
|
print(f"通过 {len(OK)} / {len(OK) + len(FAIL)}")
|
|
for n in FAIL:
|
|
print(f" 未通过: {n}")
|
|
return 0 if not FAIL else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|