705fe951b3
- interview_turn_plan: 识别低信息短回复，引导 AI 承接后主动追问新话题 - development.sh / docker-compose.dev: Postgres/Redis 端口与 .env 对齐，补充宿主机端口监听检查 - Alembic: 补回 0016 memory pipeline status、0017 segment narrative defer - app-expo: api/ws URL 去掉末尾斜杠，避免 WS 双斜杠；更新 .env.staging Co-authored-by: Cursor <cursoragent@cursor.com>
705 lines
27 KiB
Python
705 lines
27 KiB
Python
"""
|
||
访谈轮次编排（Option B 定位：InterviewTurnPlan 是**本轮行为模式的唯一决策源**）。
|
||
|
||
约束说明：
|
||
- 主 prompt（prompt_layers）只提供跨轮通用的承接-深挖-串联节奏与身份守则；
|
||
- 本轮是否「情绪优先」「跟话头」「回忆推进」完全由 `plan_interview_turn()` 决定；
|
||
- 对 LLM 的硬指令由 `InterviewTurnPlan.render_system_directive()` 输出，主 prompt **不得**再
|
||
针对具体模式另立软约束。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from dataclasses import dataclass
|
||
from typing import Literal
|
||
|
||
from app.agents.chat.prompts_conversation import SLOT_NAME_MAP
|
||
from app.agents.stage_constants import STAGE_KEYWORD_WEIGHTS, STAGE_SLOT_KEYS
|
||
|
||
InterviewTurnMode = Literal[
|
||
"emotion_first",
|
||
"clarify_first",
|
||
"memoir_push",
|
||
"follow_user_only",
|
||
]
|
||
SubjectOwner = Literal["user_only"]
|
||
MemoryUsage = Literal["none", "allowed_with_attribution"]
|
||
ReplyShape = Literal["flexible", "ack_only", "ack_then_question"]
|
||
AnchorSourceKind = Literal["user_message", "memory", "none"]
|
||
|
||
# 本轮承接焦点（规则基线或 focus planner LLM 输出；供 directive 与日志）
|
||
FocusPrimary = Literal[
|
||
"emotion",
|
||
"relationship",
|
||
"identity",
|
||
"scene",
|
||
"memoir_gap",
|
||
"follow_user",
|
||
]
|
||
FocusSource = Literal["rule", "llm", "fallback"]
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class InterviewTurnPlan:
|
||
"""单轮访谈的硬目标（优先级高于主 prompt 的一般性建议）。
|
||
|
||
作为唯一决策源，对 LLM 的落地由 `render_system_directive()` 输出；其它调用方应通过
|
||
`is_emotion_first` / `is_clarify_first` / `is_follow_user_only` / `is_memoir_push` /
|
||
`requires_explicit_question` 等语义属性读取决策，避免在主 prompt 里重复立法。
|
||
|
||
扩展字段承载「主语归属 / 记忆引用策略 / 回复形状」，供防上下文污染与双阶段 planner 合并。
|
||
"""
|
||
|
||
mode: InterviewTurnMode
|
||
anchor_slot_key: str | None
|
||
anchor_slot_readable: str
|
||
anchor_snippet: str
|
||
anchor_source_kind: AnchorSourceKind = "none"
|
||
assistant_identity_question: bool = False
|
||
subject_owner: SubjectOwner = "user_only"
|
||
memory_usage: MemoryUsage = "none"
|
||
memory_reference_style: str = "你之前提过"
|
||
forbid_first_person_experience: bool = True
|
||
reply_shape: ReplyShape = "flexible"
|
||
# 本轮承接重点（单一事实源的一部分；与 mode 配合，由 directive 落地）
|
||
primary_focus: FocusPrimary = "memoir_gap"
|
||
secondary_focus: FocusPrimary | None = None
|
||
focus_summary: str = ""
|
||
focus_source: FocusSource = "rule"
|
||
low_information_reply: bool = False
|
||
|
||
# ---- 语义属性：供 prompt_layers / interview_agent 等调用方消费，禁止重复立法 ----
|
||
|
||
@property
|
||
def is_emotion_first(self) -> bool:
|
||
return self.mode == "emotion_first"
|
||
|
||
@property
|
||
def is_clarify_first(self) -> bool:
|
||
return self.mode == "clarify_first"
|
||
|
||
@property
|
||
def is_follow_user_only(self) -> bool:
|
||
return self.mode == "follow_user_only"
|
||
|
||
@property
|
||
def is_memoir_push(self) -> bool:
|
||
return self.mode == "memoir_push"
|
||
|
||
@property
|
||
def requires_explicit_question(self) -> bool:
|
||
"""memoir_push 要求恰好一个开放式追问；其它模式允许不问或尽量不问。"""
|
||
return self.mode == "memoir_push"
|
||
|
||
@property
|
||
def allows_zero_question(self) -> bool:
|
||
"""emotion_first / clarify_first 整轮允许不问；follow_user_only 仅允许最多一个轻问。"""
|
||
return self.mode in ("emotion_first", "clarify_first", "follow_user_only")
|
||
|
||
def render_system_directive(self) -> str:
|
||
"""对 LLM 的硬指令（供注入 system prompt 顶部）。"""
|
||
return format_interview_turn_directive_block(self)
|
||
|
||
|
||
def _focus_primary_for_mode(mode: InterviewTurnMode) -> FocusPrimary:
|
||
if mode == "emotion_first":
|
||
return "emotion"
|
||
if mode == "clarify_first":
|
||
return "follow_user"
|
||
if mode == "follow_user_only":
|
||
return "follow_user"
|
||
return "memoir_gap"
|
||
|
||
|
||
def apply_safe_mode_override(
|
||
baseline: InterviewTurnMode,
|
||
override: str | None,
|
||
*,
|
||
primary_focus: str | None,
|
||
) -> InterviewTurnMode | None:
|
||
"""若 planner 建议覆盖 mode，在此做安全收敛；返回 None 表示保持 baseline。"""
|
||
if not override or override == baseline:
|
||
return None
|
||
if override not in (
|
||
"emotion_first",
|
||
"clarify_first",
|
||
"memoir_push",
|
||
"follow_user_only",
|
||
):
|
||
return None
|
||
# 情绪优先一旦成立，不因 planner 改回强推素材
|
||
if baseline == "emotion_first" and override != "emotion_first":
|
||
return None
|
||
# 跟话头 / 模糊先澄清：不允许改回问卷式 memoir_push
|
||
if baseline in ("follow_user_only", "clarify_first") and override == "memoir_push":
|
||
return None
|
||
# memoir_push -> emotion_first：仅当焦点偏情绪/关系/身份（或空，由模型判断）
|
||
if baseline == "memoir_push" and override == "emotion_first":
|
||
pf = (primary_focus or "").strip()
|
||
if pf in ("", "emotion", "relationship", "identity"):
|
||
return "emotion_first"
|
||
return None
|
||
if baseline == "memoir_push" and override == "follow_user_only":
|
||
return "follow_user_only"
|
||
if baseline == "follow_user_only" and override == "emotion_first":
|
||
return "emotion_first"
|
||
if baseline == "follow_user_only" and override == "clarify_first":
|
||
return "clarify_first"
|
||
if baseline == "memoir_push" and override == "clarify_first":
|
||
pf = (primary_focus or "").strip()
|
||
if pf in ("", "emotion", "relationship", "identity", "follow_user"):
|
||
return "clarify_first"
|
||
return None
|
||
if baseline == "clarify_first" and override == "emotion_first":
|
||
return "emotion_first"
|
||
if baseline == "clarify_first" and override == "follow_user_only":
|
||
return "follow_user_only"
|
||
return None
|
||
|
||
|
||
def _focus_directive_lines(plan: InterviewTurnPlan) -> str:
|
||
"""本轮承接重点：focus_summary 仅作追问角度，不支配正文措辞。"""
|
||
if (plan.focus_summary or "").strip():
|
||
summary = (plan.focus_summary or "").strip()
|
||
if len(summary) > 200:
|
||
summary = summary[:199].rstrip() + "…"
|
||
return (
|
||
"- **本轮追问/承接角度（非正文提纲；主回复仍须从用户本轮原话长出来）**："
|
||
+ summary
|
||
+ "\n"
|
||
)
|
||
labels: dict[FocusPrimary, str] = {
|
||
"emotion": "情绪与感受",
|
||
"relationship": "关系与他人（谁在场、谁在看、和谁较劲）",
|
||
"identity": "身份与面子（自我形象、怕丢脸、要强）",
|
||
"scene": "现场与感官（画面、光线、身体感受）",
|
||
"memoir_gap": "叙述槽与成稿素材（当前主追问方向）",
|
||
"follow_user": "顺着用户话头自然展开",
|
||
}
|
||
pri = labels.get(plan.primary_focus, labels["memoir_gap"])
|
||
if plan.secondary_focus and plan.secondary_focus != plan.primary_focus:
|
||
sec = labels.get(plan.secondary_focus, "")
|
||
if sec:
|
||
return (
|
||
f"- **本轮承接重点**：先锚定「{pri}」，必要时再轻点「{sec}」作辅助。\n"
|
||
)
|
||
return f"- **本轮承接重点**：优先锚定「{pri}」。\n"
|
||
|
||
|
||
def primary_empty_slot(stage: str, empty_slots: list[str]) -> str | None:
|
||
"""按 STAGE_SLOT_KEYS 顺序取第一个仍空的槽。"""
|
||
if not empty_slots:
|
||
return None
|
||
order = STAGE_SLOT_KEYS.get(stage, ())
|
||
for key in order:
|
||
if key in empty_slots:
|
||
return key
|
||
return empty_slots[0]
|
||
|
||
|
||
def _strip_scene_hint(memory_evidence_text: str) -> str:
|
||
raw = (memory_evidence_text or "").strip()
|
||
if "[场景氛围提示" in raw:
|
||
raw = raw.split("[场景氛围提示", 1)[0].strip()
|
||
return raw
|
||
|
||
|
||
def extract_anchor_snippet(
|
||
*,
|
||
memory_evidence_text: str,
|
||
user_message: str,
|
||
max_chars: int = 180,
|
||
) -> str:
|
||
"""优先用户本轮原话，其次极短记忆线索（用于追问挂钩，非事实断言）。
|
||
|
||
旧记忆不得压过用户当前话头；仅当本轮原话过短时再借 `memory_anchor_source` 挂钩。
|
||
`memory_evidence_text` 此处仅为**一条短线索**（非整段 `[M…]` 摘录块）。
|
||
"""
|
||
um = (user_message or "").strip()
|
||
if len(um) >= 10:
|
||
return um[:max_chars].strip()
|
||
mem = _strip_scene_hint(memory_evidence_text)
|
||
if mem and len(mem) >= 4:
|
||
for line in mem.splitlines():
|
||
s = line.strip()
|
||
if s.startswith("[M") and "]" in s:
|
||
rest = s.split("]", 1)[1].strip()
|
||
if rest and len(rest) >= 4:
|
||
return rest[:max_chars].strip()
|
||
return mem[:max_chars].strip()
|
||
if len(um) >= 4:
|
||
return um[:max_chars].strip()
|
||
return ""
|
||
|
||
|
||
def determine_anchor_source_kind(
|
||
*,
|
||
memory_evidence_text: str,
|
||
user_message: str,
|
||
) -> AnchorSourceKind:
|
||
"""标记挂钩线索来源，避免把用户本轮原话误标成旧记忆。"""
|
||
um = (user_message or "").strip()
|
||
if len(um) >= 10:
|
||
return "user_message"
|
||
mem = _strip_scene_hint(memory_evidence_text)
|
||
if mem and len(mem) >= 4:
|
||
for line in mem.splitlines():
|
||
s = line.strip()
|
||
if s.startswith("[M") and "]" in s:
|
||
rest = s.split("]", 1)[1].strip()
|
||
if rest and len(rest) >= 4:
|
||
return "memory"
|
||
return "memory"
|
||
if len(um) >= 4:
|
||
return "user_message"
|
||
return "none"
|
||
|
||
|
||
_EMOTION_MARKERS: tuple[str, ...] = (
|
||
"哭",
|
||
"难受",
|
||
"委屈",
|
||
"害怕",
|
||
"后悔",
|
||
"恨",
|
||
"舍不得",
|
||
"崩溃",
|
||
"绝望",
|
||
"心疼",
|
||
"哽咽",
|
||
"咽不下",
|
||
"睡不着",
|
||
"想哭",
|
||
"好难",
|
||
"太难了",
|
||
"挺不住",
|
||
"扛不住",
|
||
"放不下",
|
||
"意难平",
|
||
"难堪",
|
||
"丢脸",
|
||
"怕丢人",
|
||
"臊得慌",
|
||
)
|
||
|
||
# 说不清 / 暧昧放慢：不触发 memoir_push 式强推槽位
|
||
_AMBIGUITY_MARKERS: tuple[str, ...] = (
|
||
"说不清",
|
||
"说不明白",
|
||
"说不上来",
|
||
"说不上",
|
||
"说不好",
|
||
"不知道怎么说",
|
||
"不知道咋说",
|
||
"好像",
|
||
"也不是",
|
||
"不完全是",
|
||
"不太确定",
|
||
"不确定",
|
||
"有点模糊",
|
||
"很模糊",
|
||
"模模糊糊",
|
||
"难说",
|
||
"有一点，但我不确定",
|
||
)
|
||
_SOFT_SLOW_MARKERS: tuple[str, ...] = (
|
||
"害羞",
|
||
"羞涩",
|
||
"不好意思",
|
||
"脸红",
|
||
"暧昧",
|
||
)
|
||
|
||
_ASSISTANT_IDENTITY_QUESTION_MARKERS: tuple[str, ...] = (
|
||
"你是哪里人",
|
||
"你哪的人",
|
||
"你是哪里长大的",
|
||
"你在哪长大",
|
||
"你的童年",
|
||
"你小时候",
|
||
"你家里",
|
||
"你爸妈",
|
||
"你父母",
|
||
"你也有过",
|
||
"你也是",
|
||
)
|
||
|
||
_LOW_INFORMATION_REPLY_MAX_CHARS = 8
|
||
_LOW_INFORMATION_REPLY_CHARS: frozenset[str] = frozenset(
|
||
# 这不是短语白名单，而是一组低信息的应声/确认/语气字符。
|
||
# 只要短回复里出现不在此集合中的字，就会被视为有潜在叙事信号。
|
||
"嗯唔呃哦噢喔啊呀呢嘛吧哈"
|
||
"对是的了好行可允许以"
|
||
"没错确实不否无有还太很挺就"
|
||
"什么这样那样当当然"
|
||
)
|
||
_ACK_STRIP_CHARS = " \t\r\n，。！？!?、,.；;：:~～…"
|
||
|
||
|
||
def _is_emotion_heavy(text: str) -> bool:
|
||
t = (text or "").strip()
|
||
if not t:
|
||
return False
|
||
if any(m in t for m in _EMOTION_MARKERS):
|
||
return True
|
||
if (
|
||
len(t) >= 40
|
||
and ("！" in t or "!" in t)
|
||
and (".." in t or "…" in t or "唉" in t)
|
||
):
|
||
return True
|
||
return False
|
||
|
||
|
||
def _is_ambiguous_or_needs_slow_pace(text: str) -> bool:
|
||
"""模糊表达、暧昧/羞涩等：先澄清或放慢，不强推叙述槽。"""
|
||
t = (text or "").strip()
|
||
if not t:
|
||
return True
|
||
if any(m in t for m in _AMBIGUITY_MARKERS):
|
||
return True
|
||
if any(m in t for m in _SOFT_SLOW_MARKERS):
|
||
return True
|
||
return False
|
||
|
||
|
||
def _normalized_short_reply(text: str) -> str:
|
||
return "".join(ch for ch in (text or "").strip() if ch not in _ACK_STRIP_CHARS)
|
||
|
||
|
||
def _has_substantive_short_reply_signal(compact: str) -> bool:
|
||
"""短回复里的叙事实质信号：地点/人事词、阶段关键词、年份数字等。"""
|
||
if any(ch.isdigit() for ch in compact):
|
||
return True
|
||
if any(("a" <= ch.lower() <= "z") for ch in compact):
|
||
return True
|
||
for weighted_keywords in STAGE_KEYWORD_WEIGHTS.values():
|
||
if any(keyword and keyword in compact for keyword, _ in weighted_keywords):
|
||
return True
|
||
return any(ch not in _LOW_INFORMATION_REPLY_CHARS for ch in compact)
|
||
|
||
|
||
def _is_low_information_reply(text: str) -> bool:
|
||
"""识别短、低信息、没有新增叙事素材的确认/应声回复。"""
|
||
compact = _normalized_short_reply(text)
|
||
if not compact:
|
||
return False
|
||
if len(compact) > _LOW_INFORMATION_REPLY_MAX_CHARS:
|
||
return False
|
||
return not _has_substantive_short_reply_signal(compact)
|
||
|
||
|
||
def _is_too_vague_for_memoir_push(text: str) -> bool:
|
||
"""过短或仍含糊时，不进入 memoir_push。"""
|
||
t = (text or "").strip()
|
||
if len(t) < 12:
|
||
return True
|
||
return _is_ambiguous_or_needs_slow_pace(t)
|
||
|
||
|
||
def _is_asking_assistant_identity_or_life(text: str) -> bool:
|
||
t = (text or "").strip()
|
||
if not t:
|
||
return False
|
||
return any(marker in t for marker in _ASSISTANT_IDENTITY_QUESTION_MARKERS)
|
||
|
||
|
||
def plan_interview_turn(
|
||
*,
|
||
current_stage: str,
|
||
empty_slots: list[str],
|
||
normalized_user_message: str,
|
||
memory_evidence_text: str,
|
||
stage_switched_this_turn: bool,
|
||
) -> InterviewTurnPlan:
|
||
"""
|
||
粗规则（可迭代）：
|
||
- 情绪浓：先共情，不强推叙述槽搜集问。
|
||
- 说不清/暧昧羞涩等：模糊先澄清，禁止替用户下结论或强推槽位。
|
||
- 刚切换人生阶段：跟着用户节奏，不做「新阶段问卷首开」。
|
||
- 当前阶段无空槽：深度跟进，不重启盘点。
|
||
- 默认：memoir_push（仅当本轮话头足够具体、非含糊短答时）。
|
||
"""
|
||
snippet = extract_anchor_snippet(
|
||
memory_evidence_text=memory_evidence_text,
|
||
user_message=normalized_user_message,
|
||
)
|
||
anchor_source_kind = determine_anchor_source_kind(
|
||
memory_evidence_text=memory_evidence_text,
|
||
user_message=normalized_user_message,
|
||
)
|
||
mem_trim = _strip_scene_hint(memory_evidence_text).strip()
|
||
mem_use: MemoryUsage = (
|
||
"allowed_with_attribution" if mem_trim else "none"
|
||
)
|
||
um = (normalized_user_message or "").strip()
|
||
asks_assistant_identity = _is_asking_assistant_identity_or_life(um)
|
||
low_information_reply = _is_low_information_reply(um)
|
||
reply_shape: ReplyShape = "flexible"
|
||
if any(
|
||
k in um
|
||
for k in ("讲讲", "说说", "她的故事", "他的故事", "后来呢", "然后呢")
|
||
):
|
||
reply_shape = "ack_then_question"
|
||
|
||
if low_information_reply:
|
||
slot = primary_empty_slot(current_stage, empty_slots)
|
||
if slot:
|
||
readable = SLOT_NAME_MAP.get(slot, slot)
|
||
return InterviewTurnPlan(
|
||
mode="memoir_push",
|
||
anchor_slot_key=slot,
|
||
anchor_slot_readable=readable,
|
||
anchor_snippet=snippet,
|
||
anchor_source_kind=anchor_source_kind,
|
||
assistant_identity_question=asks_assistant_identity,
|
||
memory_usage=mem_use,
|
||
reply_shape="ack_then_question",
|
||
primary_focus=_focus_primary_for_mode("memoir_push"),
|
||
focus_summary=(
|
||
f"用户只做了简短确认；短接一句后，不澄清“{um}”，"
|
||
f"主动从「{readable}」打开一个具体、好回答的新回忆话题"
|
||
),
|
||
focus_source="rule",
|
||
low_information_reply=True,
|
||
)
|
||
return InterviewTurnPlan(
|
||
mode="follow_user_only",
|
||
anchor_slot_key=None,
|
||
anchor_slot_readable="（本阶段主要叙述槽已有素材）请回到上文最近的人/事/地方或情绪线，主动打开一个纵深问题",
|
||
anchor_snippet=snippet,
|
||
anchor_source_kind=anchor_source_kind,
|
||
assistant_identity_question=asks_assistant_identity,
|
||
memory_usage=mem_use,
|
||
reply_shape="ack_then_question",
|
||
primary_focus=_focus_primary_for_mode("follow_user_only"),
|
||
focus_summary=(
|
||
f"用户只做了简短确认；短接一句后，不澄清“{um}”，"
|
||
"回到上文最近的具体线索，主动递一个新的回忆追问"
|
||
),
|
||
focus_source="rule",
|
||
low_information_reply=True,
|
||
)
|
||
|
||
if _is_emotion_heavy(normalized_user_message):
|
||
slot = primary_empty_slot(current_stage, empty_slots)
|
||
readable = (
|
||
SLOT_NAME_MAP.get(slot, slot or "")
|
||
if slot
|
||
else "（情绪优先时可暂不强绑某一槽位）"
|
||
)
|
||
return InterviewTurnPlan(
|
||
mode="emotion_first",
|
||
anchor_slot_key=slot,
|
||
anchor_slot_readable=readable,
|
||
anchor_snippet=snippet,
|
||
anchor_source_kind=anchor_source_kind,
|
||
assistant_identity_question=asks_assistant_identity,
|
||
memory_usage=mem_use,
|
||
reply_shape=reply_shape,
|
||
primary_focus=_focus_primary_for_mode("emotion_first"),
|
||
focus_source="rule",
|
||
)
|
||
|
||
if stage_switched_this_turn:
|
||
return InterviewTurnPlan(
|
||
mode="follow_user_only",
|
||
anchor_slot_key=None,
|
||
anchor_slot_readable="（刚自然谈到本阶段，先顺着对方语势，勿问卷式首开）",
|
||
anchor_snippet=snippet,
|
||
anchor_source_kind=anchor_source_kind,
|
||
assistant_identity_question=asks_assistant_identity,
|
||
memory_usage=mem_use,
|
||
reply_shape=reply_shape,
|
||
primary_focus=_focus_primary_for_mode("follow_user_only"),
|
||
focus_source="rule",
|
||
)
|
||
|
||
if not empty_slots:
|
||
return InterviewTurnPlan(
|
||
mode="follow_user_only",
|
||
anchor_slot_key=None,
|
||
anchor_slot_readable="（本阶段主要叙述槽已有素材）请 depth-first：接续画面或情绪线，别重启童年在哪长大式盘点",
|
||
anchor_snippet=snippet,
|
||
anchor_source_kind=anchor_source_kind,
|
||
assistant_identity_question=asks_assistant_identity,
|
||
memory_usage=mem_use,
|
||
reply_shape=reply_shape,
|
||
primary_focus=_focus_primary_for_mode("follow_user_only"),
|
||
focus_source="rule",
|
||
)
|
||
|
||
if _is_too_vague_for_memoir_push(um):
|
||
slot = primary_empty_slot(current_stage, empty_slots)
|
||
readable = (
|
||
SLOT_NAME_MAP.get(slot, slot or "")
|
||
if slot
|
||
else "（先澄清感受时可暂不强绑某一槽位）"
|
||
)
|
||
return InterviewTurnPlan(
|
||
mode="clarify_first",
|
||
anchor_slot_key=slot,
|
||
anchor_slot_readable=readable,
|
||
anchor_snippet=snippet,
|
||
anchor_source_kind=anchor_source_kind,
|
||
assistant_identity_question=asks_assistant_identity,
|
||
memory_usage=mem_use,
|
||
reply_shape=reply_shape,
|
||
primary_focus=_focus_primary_for_mode("clarify_first"),
|
||
focus_source="rule",
|
||
)
|
||
|
||
slot = primary_empty_slot(current_stage, empty_slots)
|
||
assert slot is not None
|
||
return InterviewTurnPlan(
|
||
mode="memoir_push",
|
||
anchor_slot_key=slot,
|
||
anchor_slot_readable=SLOT_NAME_MAP.get(slot, slot),
|
||
anchor_snippet=snippet,
|
||
anchor_source_kind=anchor_source_kind,
|
||
assistant_identity_question=asks_assistant_identity,
|
||
memory_usage=mem_use,
|
||
reply_shape=reply_shape,
|
||
primary_focus=_focus_primary_for_mode("memoir_push"),
|
||
focus_source="rule",
|
||
)
|
||
|
||
|
||
def format_interview_turn_directive_block(plan: InterviewTurnPlan) -> str:
|
||
"""注入 guided prompt 顶部的硬指令块。"""
|
||
snippet_line = (
|
||
plan.anchor_snippet
|
||
if plan.anchor_snippet
|
||
else "（无可用摘录时，必须从用户本轮原话里抽词作挂钩，禁止编造）"
|
||
)
|
||
|
||
if plan.memory_usage == "allowed_with_attribution":
|
||
mem_block = (
|
||
"- **过往记忆线索**：主 Context 里若有一条极短线索，**只**用于帮你选追问角度；"
|
||
"**不是**正文提纲，**不是**长素材。\n"
|
||
"- **禁止**用「"
|
||
+ plan.memory_reference_style
|
||
+ "…」开场或整段复述旧记忆；若轻勾连，**最多半句**，且须让本轮原话占主位。\n"
|
||
"- **禁止**把线索写成助手亲历；**禁止**把线索当「咱俩共同回忆」。\n"
|
||
)
|
||
else:
|
||
mem_block = (
|
||
"- **过往记忆线索**：本轮无可用线索（或已策略性不注入）；"
|
||
"**不要**编造「你说过…」式假归因。\n"
|
||
)
|
||
|
||
perspective_block = (
|
||
"- **主语与视角**：用户可见回复主线必须是**用户**的经历与感受；"
|
||
"助手**没有**真实人生传记。\n"
|
||
)
|
||
if plan.forbid_first_person_experience:
|
||
perspective_block += (
|
||
"- **禁止助手自传式措辞**：不得使用「我小时候」「我演过/我扮演」「我当时暗恋」"
|
||
"「我爸妈」等自称亲历表述。\n"
|
||
)
|
||
if plan.assistant_identity_question:
|
||
perspective_block += (
|
||
"- **本轮用户在问助手本人**：必须明确守住边界，不能假装自己有籍贯、童年、家庭或恋爱经历；"
|
||
"若要借记忆继续聊，只能明确归因到用户，例如「你刚提到上海」「你前面说过……」。\n"
|
||
)
|
||
|
||
if plan.reply_shape == "ack_only":
|
||
shape_block = "- **本轮回复形状**：以短承接为主；若无必要可整轮不问。\n"
|
||
elif plan.reply_shape == "ack_then_question":
|
||
shape_block = (
|
||
"- **本轮回复形状**：先短承接，再尽量带**一条**贴用户话头的开放式问（仍守全篇最多一个问句）。\n"
|
||
)
|
||
else:
|
||
shape_block = ""
|
||
|
||
ack_block = ""
|
||
if plan.low_information_reply:
|
||
ack_block = (
|
||
"- **低信息短回复处理**：用户本轮只是简短确认、应声或泛泛回应，没有新增叙事素材。"
|
||
"不要把这个短答本身当成需要澄清的内容，不要反复追问「你是说……吗」，"
|
||
"也不要停在原地等用户继续补充。\n"
|
||
"- 先用半句自然接住，再主动从「主追问方向」、上文最近的具体人/事/地方，"
|
||
"或过往记忆线索里挑一个**具体、好回答**的新回忆话题；若挂钩线索为空，"
|
||
"允许直接借当前阶段未聊方向起问，但禁止编造用户没说过的细节。\n"
|
||
)
|
||
|
||
if plan.mode == "emotion_first":
|
||
mode_rules = (
|
||
"- **情绪优先**：本轮以承接、并肩与安全感为主，**不要**为推进「叙述槽大纲」而追加信息搜集型追问。\n"
|
||
"- 若末尾带问，只能是**贴着用户当前情绪或原词**的极轻一句；禁止切到盘点式下一题。\n"
|
||
"- 参考主槽「"
|
||
+ plan.anchor_slot_readable
|
||
+ "」仅供你心里知道后续方向，**不要**在本轮用问卷口吻硬推该槽。"
|
||
)
|
||
elif plan.mode == "clarify_first":
|
||
mode_rules = (
|
||
"- **模糊先澄清**：用户正在表达不确定、说不清，或暧昧/羞涩等未命名感受；"
|
||
"以承接、并列为先，**禁止**把模糊感受改写成确定关系、命运、动机或人生结论。\n"
|
||
"- 若带问句，最多**一个**，且须邀请对方**用自己的词**继续摸索或澄清（例如更接近哪一种、还是也不完全是），"
|
||
"禁止封闭式逼认、禁止替用户命名关系或事件结局。\n"
|
||
"- 允许整轮**只承接不问**；参考主槽「"
|
||
+ plan.anchor_slot_readable
|
||
+ "」仅供你心里知道后续方向，**不要**问卷式硬推该槽。"
|
||
)
|
||
elif plan.mode == "follow_user_only":
|
||
mode_rules = (
|
||
"- **跟话头**：本轮禁止问卷式首开、禁止重启式盘点；顺着用户刚展开的画面、人物或情绪自然往下。\n"
|
||
"- 若带问句，最多**一个**，且必须**从用户原词或下面摘录**长出来，禁止空泛「还有吗」。"
|
||
)
|
||
if plan.low_information_reply:
|
||
mode_rules = (
|
||
"- **跟话头 + 主动续话**：用户本轮只是简短确认，没有新素材；"
|
||
"不要围着短答本身澄清，也不要重复上一问等对方补充。\n"
|
||
"- 若带问句，最多**一个**，优先从上文最近的具体人/事/地方或下面摘录长出来；"
|
||
"若无可用摘录，就从当前阶段已聊内容的纵深处打开一个新回忆话题。"
|
||
)
|
||
else:
|
||
if plan.low_information_reply:
|
||
mode_rules = (
|
||
"- **回忆推进（memoir_push）**：用户本轮只是简短确认，"
|
||
"对用户可见回复中须有**恰好一个**开放式回忆追问，主动把话头往下递。\n"
|
||
" 追问可挂住**上文最近具体细节**、**主追问方向**或**挂钩摘录**；"
|
||
"不要要求从低信息短答里抽词。\n"
|
||
"- 禁止用日常寒暄替代该追问；仍遵守全篇「最多一个问句」「禁止晚会硬切」。"
|
||
)
|
||
else:
|
||
mode_rules = (
|
||
"- **回忆推进（memoir_push）**：对用户可见回复中须有**恰好一个**开放式回忆追问，\n"
|
||
" 且意图明显在补足下面「主追问方向」；问句必须挂住**用户本轮原词**或**挂钩摘录**（二者至少其一，且**优先原词**）。\n"
|
||
"- **禁止**把用户仍含糊的表达改写成确定事实或关系；禁止用日常寒暄替代该追问；"
|
||
"仍遵守全篇「最多一个问句」「禁止晚会硬切」。"
|
||
)
|
||
|
||
focus_block = _focus_directive_lines(plan)
|
||
|
||
if plan.anchor_source_kind == "user_message":
|
||
anchor_label = "（仅作追问挂钩，来自用户本轮原话摘录；优先把这句里的原词接住）"
|
||
elif plan.anchor_source_kind == "memory":
|
||
anchor_label = "（仅作追问挂钩，来自检索到的用户过往口述/摘要；不是用户本轮新说的内容）"
|
||
else:
|
||
anchor_label = "（无可用挂钩时，必须从用户本轮原话里抽词承接，禁止编造）"
|
||
|
||
return f"""## 本轮编排指令（硬规则，优先于后文一般性建议）
|
||
{mode_rules}
|
||
{focus_block}{ack_block}{mem_block}{perspective_block}{shape_block}- **主追问方向（叙述槽）**：{plan.anchor_slot_readable}
|
||
- **挂钩线索**{anchor_label}：{snippet_line}
|
||
"""
|
||
|
||
|
||
__all__ = [
|
||
"FocusPrimary",
|
||
"FocusSource",
|
||
"InterviewTurnMode",
|
||
"InterviewTurnPlan",
|
||
"MemoryUsage",
|
||
"ReplyShape",
|
||
"SubjectOwner",
|
||
"AnchorSourceKind",
|
||
"apply_safe_mode_override",
|
||
"determine_anchor_source_kind",
|
||
"extract_anchor_snippet",
|
||
"format_interview_turn_directive_block",
|
||
"plan_interview_turn",
|
||
"primary_empty_slot",
|
||
]
|