53e0065e3e
配置 SSOT（TOML + .env） 统一错误契约 Auth 与事务边界 Redis / Celery 可靠性:业务 Redis（DB/0）与 Celery broker/backend（DB/1）显式拆分；连接池、sync client 可观测性（OpenTelemetry + LGTM）
307 lines
10 KiB
Python
307 lines
10 KiB
Python
"""interview_turn_plan：轮次模式与主槽选择（服务端硬编排）。"""
|
||
|
||
from app.agents.chat.interview_turn_plan import (
|
||
InterviewTurnPlan,
|
||
apply_safe_mode_override,
|
||
extract_anchor_snippet,
|
||
format_interview_turn_directive_block,
|
||
plan_interview_turn,
|
||
primary_empty_slot,
|
||
)
|
||
|
||
|
||
def test_primary_empty_slot_order():
|
||
assert primary_empty_slot("childhood", ["emotion", "place"]) == "place"
|
||
assert primary_empty_slot("childhood", ["emotion"]) == "emotion"
|
||
|
||
|
||
def test_extract_anchor_snippet_prefers_user_when_long_enough():
|
||
mem = "摘录的一段记忆\n\n[场景氛围提示"
|
||
um = "用户说很长一句" * 5
|
||
sn = extract_anchor_snippet(memory_evidence_text=mem, user_message=um)
|
||
assert sn.startswith("用户说")
|
||
assert "摘录" not in sn
|
||
|
||
|
||
def test_extract_anchor_snippet_prefers_first_m_line():
|
||
mem = (
|
||
"【相关记忆摘录·聊天专用】\n"
|
||
"说明行……\n"
|
||
"[M1] 你在校园演出里饰演罗密欧。\n"
|
||
)
|
||
sn = extract_anchor_snippet(memory_evidence_text=mem, user_message="短")
|
||
assert "校园演出" in sn
|
||
assert "【相关" not in sn
|
||
|
||
|
||
def test_plan_sets_memory_usage_when_evidence_present():
|
||
p = plan_interview_turn(
|
||
current_stage="childhood",
|
||
empty_slots=["place"],
|
||
normalized_user_message="嗯。",
|
||
memory_evidence_text="【头】\n[M1] 你提过河边。",
|
||
stage_switched_this_turn=False,
|
||
)
|
||
assert p.memory_usage == "allowed_with_attribution"
|
||
|
||
|
||
def test_plan_marks_assistant_identity_question() -> None:
|
||
p = plan_interview_turn(
|
||
current_stage="childhood",
|
||
empty_slots=["place"],
|
||
normalized_user_message="你是哪里人，你的童年是什么样的？",
|
||
memory_evidence_text="用户曾说：「我小时候在上海长大。」",
|
||
stage_switched_this_turn=False,
|
||
)
|
||
assert p.assistant_identity_question is True
|
||
|
||
|
||
def test_plan_reply_shape_ack_then_question_on_her_story_followup():
|
||
p = plan_interview_turn(
|
||
current_stage="education",
|
||
empty_slots=["school"],
|
||
normalized_user_message="那你讲讲她的故事吧。",
|
||
memory_evidence_text="",
|
||
stage_switched_this_turn=False,
|
||
)
|
||
assert p.reply_shape == "ack_then_question"
|
||
|
||
|
||
def test_directive_includes_attribution_when_memory_allowed():
|
||
plan = InterviewTurnPlan(
|
||
mode="memoir_push",
|
||
anchor_slot_key="place",
|
||
anchor_slot_readable="成长的地方",
|
||
anchor_snippet="挂钩",
|
||
memory_usage="allowed_with_attribution",
|
||
memory_reference_style="你之前提过",
|
||
reply_shape="ack_then_question",
|
||
)
|
||
block = format_interview_turn_directive_block(plan)
|
||
assert "线索" in block
|
||
assert "你之前提过" in block
|
||
assert "真实人生传记" in block
|
||
assert "ack" not in block.lower()
|
||
assert "承接" in block
|
||
assert (
|
||
"本轮追问" in block
|
||
or "承接角度" in block
|
||
or "本轮承接重点" in block
|
||
)
|
||
|
||
|
||
def test_directive_includes_focus_summary_when_set():
|
||
plan = InterviewTurnPlan(
|
||
mode="memoir_push",
|
||
anchor_slot_key="place",
|
||
anchor_slot_readable="成长的地方",
|
||
anchor_snippet="挂钩",
|
||
focus_summary="先接住「芳芳」与怕丢脸这条线",
|
||
focus_source="llm",
|
||
primary_focus="relationship",
|
||
)
|
||
block = format_interview_turn_directive_block(plan)
|
||
assert "芳芳" in block
|
||
assert "本轮追问" in block or "承接角度" in block
|
||
|
||
|
||
def test_directive_marks_user_message_anchor_source_correctly():
|
||
plan = InterviewTurnPlan(
|
||
mode="memoir_push",
|
||
anchor_slot_key="place",
|
||
anchor_slot_readable="成长的地方",
|
||
anchor_snippet="那年夏天我总往河边跑",
|
||
anchor_source_kind="user_message",
|
||
)
|
||
block = format_interview_turn_directive_block(plan)
|
||
assert "来自用户本轮原话摘录" in block
|
||
assert "不是用户本轮新说的内容" not in block
|
||
|
||
|
||
def test_directive_marks_memory_anchor_source_correctly():
|
||
plan = InterviewTurnPlan(
|
||
mode="memoir_push",
|
||
anchor_slot_key="place",
|
||
anchor_slot_readable="成长的地方",
|
||
anchor_snippet="你小时候常在河边玩",
|
||
anchor_source_kind="memory",
|
||
)
|
||
block = format_interview_turn_directive_block(plan)
|
||
assert "来自检索到的用户过往口述/摘要" in block
|
||
assert "不是用户本轮新说的内容" in block
|
||
|
||
|
||
def test_directive_adds_boundary_for_assistant_identity_question():
|
||
plan = InterviewTurnPlan(
|
||
mode="memoir_push",
|
||
anchor_slot_key="place",
|
||
anchor_slot_readable="成长的地方",
|
||
anchor_snippet="用户曾说：「我小时候在上海长大。」",
|
||
assistant_identity_question=True,
|
||
)
|
||
block = format_interview_turn_directive_block(plan)
|
||
assert "本轮用户在问助手本人" in block
|
||
assert "你刚提到上海" in block
|
||
|
||
|
||
def test_apply_safe_mode_override_blocks_emotion_downgrade():
|
||
assert apply_safe_mode_override("emotion_first", "memoir_push", primary_focus="emotion") is None
|
||
|
||
|
||
def test_apply_safe_mode_override_allows_memoir_to_emotion():
|
||
assert (
|
||
apply_safe_mode_override("memoir_push", "emotion_first", primary_focus="relationship")
|
||
== "emotion_first"
|
||
)
|
||
|
||
|
||
def test_apply_safe_mode_override_clarify_blocks_memoir():
|
||
assert apply_safe_mode_override("clarify_first", "memoir_push", primary_focus="emotion") is None
|
||
|
||
|
||
def test_apply_safe_mode_override_clarify_to_emotion():
|
||
assert (
|
||
apply_safe_mode_override("clarify_first", "emotion_first", primary_focus="emotion")
|
||
== "emotion_first"
|
||
)
|
||
|
||
|
||
def test_plan_clarify_first_when_ambiguous():
|
||
p = plan_interview_turn(
|
||
current_stage="childhood",
|
||
empty_slots=["place", "people"],
|
||
normalized_user_message="那种喜欢我也说不清，好像有又好像没有。",
|
||
memory_evidence_text="",
|
||
stage_switched_this_turn=False,
|
||
)
|
||
assert p.mode == "clarify_first"
|
||
|
||
|
||
def test_plan_clarify_first_when_very_short():
|
||
p = plan_interview_turn(
|
||
current_stage="childhood",
|
||
empty_slots=["place"],
|
||
normalized_user_message="有点说不清",
|
||
memory_evidence_text="",
|
||
stage_switched_this_turn=False,
|
||
)
|
||
assert p.mode == "clarify_first"
|
||
|
||
|
||
def test_plan_low_information_reply_pushes_next_topic_when_slots_remain():
|
||
p = plan_interview_turn(
|
||
current_stage="childhood",
|
||
empty_slots=["place", "people"],
|
||
normalized_user_message="嗯。",
|
||
memory_evidence_text="",
|
||
stage_switched_this_turn=False,
|
||
)
|
||
assert p.mode == "memoir_push"
|
||
assert p.anchor_slot_key == "place"
|
||
assert p.reply_shape == "ack_then_question"
|
||
assert p.low_information_reply is True
|
||
|
||
|
||
def test_low_information_directive_asks_for_proactive_topic_not_clarification():
|
||
p = plan_interview_turn(
|
||
current_stage="childhood",
|
||
empty_slots=["place"],
|
||
normalized_user_message="对",
|
||
memory_evidence_text="",
|
||
stage_switched_this_turn=False,
|
||
)
|
||
block = format_interview_turn_directive_block(p)
|
||
assert "低信息短回复处理" in block
|
||
assert "不要把这个短答本身当成需要澄清的内容" in block
|
||
assert "恰好一个" in block
|
||
assert "主动" in block
|
||
|
||
|
||
def test_plan_low_information_reply_uses_follow_when_no_slots_remain():
|
||
p = plan_interview_turn(
|
||
current_stage="childhood",
|
||
empty_slots=[],
|
||
normalized_user_message="是的",
|
||
memory_evidence_text="",
|
||
stage_switched_this_turn=False,
|
||
)
|
||
assert p.mode == "follow_user_only"
|
||
assert p.reply_shape == "ack_then_question"
|
||
assert p.low_information_reply is True
|
||
|
||
|
||
def test_short_substantive_reply_is_not_treated_as_low_information():
|
||
p = plan_interview_turn(
|
||
current_stage="childhood",
|
||
empty_slots=["place", "people"],
|
||
normalized_user_message="上海",
|
||
memory_evidence_text="",
|
||
stage_switched_this_turn=False,
|
||
)
|
||
assert p.low_information_reply is False
|
||
|
||
|
||
def test_plan_memoir_push():
|
||
p = plan_interview_turn(
|
||
current_stage="childhood",
|
||
empty_slots=["place", "people"],
|
||
normalized_user_message="我小时候住在河边，夏天常去玩水。",
|
||
memory_evidence_text="",
|
||
stage_switched_this_turn=False,
|
||
)
|
||
assert p.mode == "memoir_push"
|
||
assert p.anchor_slot_key == "place"
|
||
assert p.anchor_snippet
|
||
|
||
|
||
def test_plan_emotion_first():
|
||
p = plan_interview_turn(
|
||
current_stage="childhood",
|
||
empty_slots=["place"],
|
||
normalized_user_message="想起来还是很难受，忍不住想哭。",
|
||
memory_evidence_text="",
|
||
stage_switched_this_turn=False,
|
||
)
|
||
assert p.mode == "emotion_first"
|
||
|
||
|
||
def test_plan_follow_on_stage_switch():
|
||
p = plan_interview_turn(
|
||
current_stage="education",
|
||
empty_slots=["school", "city"],
|
||
normalized_user_message="后来我去省城读中学了。",
|
||
memory_evidence_text="",
|
||
stage_switched_this_turn=True,
|
||
)
|
||
assert p.mode == "follow_user_only"
|
||
|
||
|
||
def test_plan_follow_when_no_empty_slots():
|
||
p = plan_interview_turn(
|
||
current_stage="childhood",
|
||
empty_slots=[],
|
||
normalized_user_message="嗯。",
|
||
memory_evidence_text="",
|
||
stage_switched_this_turn=False,
|
||
)
|
||
assert p.mode == "follow_user_only"
|
||
assert p.low_information_reply is True
|
||
|
||
|
||
def test_build_topic_chips_english_uses_slot_name_map_en():
|
||
from app.agents.chat.prompts_conversation import build_topic_chips
|
||
|
||
chips = build_topic_chips("childhood", ["place"], max_chips=2, language="en")
|
||
assert len(chips) >= 1
|
||
place_chip = next(c for c in chips if c["id"] == "place")
|
||
assert place_chip["label"] == "where you grew up"
|
||
assert place_chip["text"].startswith("I'd like to talk about")
|
||
|
||
|
||
def test_build_topic_chips_belief_stage_not_empty():
|
||
from app.agents.chat.prompts_conversation import build_topic_chips
|
||
|
||
chips = build_topic_chips("belief", [], max_chips=4, language="en")
|
||
assert len(chips) == 4
|
||
assert chips[0]["label"]
|