113 lines
3.7 KiB
Python
113 lines
3.7 KiB
Python
|
"""reply_planner：JSON 合并与安全边界。"""
|
|||
|
|
|
|||
|
|
import json
|
|||
|
|
|
|||
|
|
from app.agents.chat.interview_turn_plan import InterviewTurnPlan
|
|||
|
|
from app.agents.chat.reply_planner import merge_reply_planner_json_into_turn_plan
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _base_plan(**kwargs) -> InterviewTurnPlan:
|
|||
|
|
defaults = dict(
|
|||
|
|
mode="memoir_push",
|
|||
|
|
anchor_slot_key="place",
|
|||
|
|
anchor_slot_readable="成长的地方",
|
|||
|
|
anchor_snippet="河边",
|
|||
|
|
memory_usage="allowed_with_attribution",
|
|||
|
|
reply_shape="flexible",
|
|||
|
|
memory_reference_style="你之前提过",
|
|||
|
|
)
|
|||
|
|
defaults.update(kwargs)
|
|||
|
|
return InterviewTurnPlan(**defaults)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_merge_does_not_upgrade_memory_from_none_to_allowed():
|
|||
|
|
plan = _base_plan(memory_usage="none")
|
|||
|
|
raw = json.dumps(
|
|||
|
|
{
|
|||
|
|
"memory_usage": "allowed_with_attribution",
|
|||
|
|
"reply_shape": "ack_then_question",
|
|||
|
|
"memory_reference_style": "你说过",
|
|||
|
|
"forbid_first_person_experience": True,
|
|||
|
|
}
|
|||
|
|
)
|
|||
|
|
merged = merge_reply_planner_json_into_turn_plan(plan, raw)
|
|||
|
|
assert merged.memory_usage == "none"
|
|||
|
|
assert merged.reply_shape == "ack_then_question"
|
|||
|
|
assert merged.memory_reference_style == "你说过"
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_merge_allows_downgrade_memory_usage_to_none():
|
|||
|
|
plan = _base_plan(memory_usage="allowed_with_attribution")
|
|||
|
|
raw = json.dumps({"memory_usage": "none"})
|
|||
|
|
merged = merge_reply_planner_json_into_turn_plan(plan, raw)
|
|||
|
|
assert merged.memory_usage == "none"
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_merge_invalid_json_returns_original():
|
|||
|
|
plan = _base_plan()
|
|||
|
|
merged = merge_reply_planner_json_into_turn_plan(plan, "not json")
|
|||
|
|
assert merged == plan
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_merge_ignores_non_dict_json():
|
|||
|
|
plan = _base_plan()
|
|||
|
|
merged = merge_reply_planner_json_into_turn_plan(plan, "[1,2]")
|
|||
|
|
assert merged == plan
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_merge_trims_memory_reference_style():
|
|||
|
|
plan = _base_plan()
|
|||
|
|
raw = json.dumps({"memory_reference_style": " 你刚讲到 "})
|
|||
|
|
merged = merge_reply_planner_json_into_turn_plan(plan, raw)
|
|||
|
|
assert merged.memory_reference_style == "你刚讲到"
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_merge_sets_focus_and_summary():
|
|||
|
|
plan = _base_plan()
|
|||
|
|
raw = json.dumps(
|
|||
|
|
{
|
|||
|
|
"primary_focus": "relationship",
|
|||
|
|
"focus_summary": "先接住用户提到的在场关系与面子压力",
|
|||
|
|
"forbid_first_person_experience": True,
|
|||
|
|
}
|
|||
|
|
)
|
|||
|
|
merged = merge_reply_planner_json_into_turn_plan(plan, raw)
|
|||
|
|
assert merged.primary_focus == "relationship"
|
|||
|
|
assert "关系" in merged.focus_summary
|
|||
|
|
assert merged.focus_source == "llm"
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_merge_mode_override_memoir_to_emotion_when_focus_supports():
|
|||
|
|
plan = _base_plan(mode="memoir_push")
|
|||
|
|
raw = json.dumps(
|
|||
|
|
{
|
|||
|
|
"mode_override": "emotion_first",
|
|||
|
|
"primary_focus": "identity",
|
|||
|
|
"forbid_first_person_experience": True,
|
|||
|
|
}
|
|||
|
|
)
|
|||
|
|
merged = merge_reply_planner_json_into_turn_plan(plan, raw)
|
|||
|
|
assert merged.mode == "emotion_first"
|
|||
|
|
assert merged.focus_source == "llm"
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_merge_rejects_unsafe_mode_override_emotion_to_memoir():
|
|||
|
|
plan = _base_plan(mode="emotion_first")
|
|||
|
|
raw = json.dumps(
|
|||
|
|
{
|
|||
|
|
"mode_override": "memoir_push",
|
|||
|
|
"primary_focus": "memoir_gap",
|
|||
|
|
"forbid_first_person_experience": True,
|
|||
|
|
}
|
|||
|
|
)
|
|||
|
|
merged = merge_reply_planner_json_into_turn_plan(plan, raw)
|
|||
|
|
assert merged.mode == "emotion_first"
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_merge_omitted_secondary_focus_unchanged():
|
|||
|
|
plan = _base_plan()
|
|||
|
|
raw = json.dumps({"reply_shape": "ack_only", "forbid_first_person_experience": True})
|
|||
|
|
merged = merge_reply_planner_json_into_turn_plan(plan, raw)
|
|||
|
|
assert merged.reply_shape == "ack_only"
|
|||
|
|
assert merged.focus_source == "rule"
|