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/覆盖现有改动。此提交即该基线。
114 lines
4.5 KiB
Python
114 lines
4.5 KiB
Python
"""提示词构造。
|
|
|
|
关键约束（实测，P0-本地LLM验证报告.md §四）：
|
|
|
|
本机 vLLM **只接受一条 system 消息**，多条返回
|
|
400 "System message must be at the beginning."
|
|
|
|
所以记忆不能作为"第二条 system"注入，必须**并进同一条里、每轮重建**。
|
|
|
|
这其实比追加式更干净，也正好满足方案 §8 的要求：
|
|
"切换身份清理上一人的私有检索上下文"、"注入的记忆每轮替换，不是追加"——
|
|
每轮从零拼出来的 system 天然不会有上一人的残留。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Sequence
|
|
|
|
from .identity.base import IdentityDecision
|
|
from .memory.base import Memory
|
|
|
|
#: 语音助手的回复风格。方案 §5："默认简短自然回复"。
|
|
#: 这几条不是"风格偏好"，是**功能性的**——放出去的文本会直接进 TTS，
|
|
#: 带 Markdown 的话会被逐字念出来（"井号井号 第一步"）。
|
|
PERSONA_RULES = """\
|
|
你是一个桌面陪伴助手，通过语音和用户交流。用户听到的是合成语音，不是文字。
|
|
|
|
必须遵守：
|
|
1. 回复简短自然，适合朗读，默认一到两句话。不要长篇解释。
|
|
2. 绝对不要使用 Markdown、星号、井号、列表、表格、表情符号或括号注释。
|
|
3. 数字、日期、金额按中文口语写法，例如"一千二百元""下午三点半"。
|
|
4. 不确定的事直说不确定，不要编造。没做过的事不要说已经做了。
|
|
5. 需要操作设备时调用工具。不要假装已经执行，也不要说"正在为您打开"。
|
|
6. 用户打断时立刻停止，不要坚持把上一句说完。"""
|
|
|
|
|
|
def build_system_prompt(
|
|
identity: IdentityDecision,
|
|
memories: Sequence[Memory] = (),
|
|
*,
|
|
extra: str = "",
|
|
) -> str:
|
|
"""拼出本轮唯一的 system 消息。
|
|
|
|
每轮调用一次，不要缓存后复用——身份或记忆一变，缓存就会串。
|
|
"""
|
|
parts = [PERSONA_RULES]
|
|
|
|
if identity.confirmed:
|
|
if identity.display_name:
|
|
parts.append(
|
|
f"\n当前用户希望被称为“{identity.display_name}”。"
|
|
"这是用户可见称呼，可以在合适时自然使用。"
|
|
)
|
|
else:
|
|
parts.append("\n当前连接已绑定到一个用户档案，但尚未记录用户称呼。")
|
|
parts.append(
|
|
"内部 user_id 只用于数据隔离，绝不能把它当作用户名、姓名或称呼告诉用户。"
|
|
"如果用户问‘我是谁’而没有已知称呼，应坦白还不知道，而不是回答内部 ID。"
|
|
)
|
|
else:
|
|
# 方案 §6：未知身份只用访客上下文，不做个性化
|
|
parts.append("\n当前用户身份未确认，按访客对待。"
|
|
"不要提及任何私人信息，也不要假定你知道对方是谁。")
|
|
|
|
if memories:
|
|
lines = ["\n已知的用户信息（供参考，不要逐条复述）："]
|
|
for m in memories:
|
|
lines.append(f"- {m.text}")
|
|
parts.append("\n".join(lines))
|
|
|
|
if identity.suggestions and not identity.confirmed:
|
|
# 有人脸候选但不自动切换。让助手可以自然地问一句，
|
|
# 而不是自作主张切换档案（方案 §6："看到熟人不等于确认他正在说话"）。
|
|
names = "、".join(
|
|
s.candidate_user_id for s in identity.suggestions[:2])
|
|
parts.append(
|
|
f"\n注意：摄像头可能认出了 {names}，但这不代表对方正在说话。"
|
|
f"不要据此切换身份或提及该用户的私人信息。")
|
|
|
|
if extra:
|
|
parts.append("\n" + extra.strip())
|
|
|
|
return "\n".join(parts)
|
|
|
|
|
|
def trim_history(
|
|
history: Sequence[dict],
|
|
*,
|
|
max_turns: int,
|
|
max_chars: int = 4000,
|
|
) -> list[dict]:
|
|
"""裁掉过旧的对话轮次。
|
|
|
|
方案 §8 要求"对话摘要限长、按需检索"，不默认永久保存逐字稿。
|
|
这里做最简单的一档：保留最近 max_turns 轮，再按字符数兜底。
|
|
上下文越短，本地 LLM 的首 token 越短——这不是省资源，是直接买延迟。
|
|
"""
|
|
msgs = list(history)
|
|
# 一轮 = 一问一答，所以保留的消息数是轮数的两倍
|
|
keep = max_turns * 2
|
|
if len(msgs) > keep:
|
|
msgs = msgs[-keep:]
|
|
|
|
# 按字符数从旧到新裁，保证裁完仍然从 user 消息开始
|
|
while msgs and sum(len(m.get("content") or "") for m in msgs) > max_chars:
|
|
msgs.pop(0)
|
|
while msgs and msgs[0].get("role") != "user":
|
|
msgs.pop(0)
|
|
return msgs
|
|
|
|
|
|
__all__ = ["build_system_prompt", "trim_history", "PERSONA_RULES"]
|