""" 访谈轮次编排（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", ]