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/覆盖现有改动。此提交即该基线。
130 lines
4.8 KiB
Python
130 lines
4.8 KiB
Python
"""首版身份实现：显式切换 + 访客兜底。
|
||
|
||
对应方案 §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
|
||
visible_name = (
|
||
display_name.strip()
|
||
if display_name.strip()
|
||
else self._decision.display_name if prev == user_id else ""
|
||
)
|
||
self._decision = IdentityDecision(
|
||
user_id=user_id,
|
||
trust_level=TRUST_NORMAL,
|
||
confirmed=True,
|
||
reason=reason,
|
||
display_name=visible_name,
|
||
)
|
||
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"]
|