feat: complete voice session safety and confirmation flow

backend/app/services/voice_session_safety.py

@@ -0,0 +1,135 @@
+"""Safety helpers for child-friendly voice co-creation sessions."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+UNSAFE_KEYWORD_GROUPS: dict[str, tuple[str, ...]] = {
+    "violence": (
+        "打死",
+        "杀掉",
+        "砍伤",
+        "流很多血",
+        "炸弹",
+        "爆炸",
+        "开枪",
+        "刀子",
+        "互相打",
+    ),
+    "horror": (
+        "鬼屋",
+        "鬼怪",
+        "僵尸",
+        "诅咒",
+        "恶魔",
+        "吃人",
+        "恐怖",
+        "吓死人",
+    ),
+    "danger": (
+        "毒药",
+        "绑架",
+        "自杀",
+        "跳楼",
+        "伤害自己",
+        "把人关起来",
+    ),
+    "adult": (
+        "色情",
+        "裸",
+        "亲热",
+        "不穿衣服",
+    ),
+}
+
+
+@dataclass(frozen=True)
+class VoiceSafetyResult:
+    """Result of one voice safety evaluation."""
+
+    is_safe: bool
+    flags: list[str]
+    replacement_text: str | None = None
+    message: str | None = None
+
+
+def _collect_safety_flags(text: str) -> list[str]:
+    normalized = text.replace(" ", "").strip()
+    flags: list[str] = []
+    if not normalized:
+        return flags
+
+    for flag, keywords in UNSAFE_KEYWORD_GROUPS.items():
+        if any(keyword in normalized for keyword in keywords):
+            flags.append(flag)
+    return flags
+
+
+def _redirect_prefix(flags: list[str]) -> str:
+    if "adult" in flags:
+        return "这个方向不适合小朋友的睡前故事。"
+    if "danger" in flags or "violence" in flags:
+        return "这个方向有点太危险了。"
+    if "horror" in flags:
+        return "这个方向有点太吓人了。"
+    return "这个方向现在不太适合继续讲下去。"
+
+
+def build_child_safe_redirect(flags: list[str]) -> str:
+    """Build a child-friendly redirect prompt after an unsafe request."""
+
+    return (
+        f"{_redirect_prefix(flags)}"
+        "我们把它改成温柔、安全、适合小朋友的冒险吧。"
+        "你可以试试说：让小伙伴一起想办法、让事情变得更明亮，或者让新朋友来帮忙。"
+    )
+
+
+def build_safe_story_fallback(*, premise: str | None = None) -> str:
+    """Build a safe replacement narrative segment for unsafe assistant output."""
+
+    subject = (premise or "小伙伴们").strip()
+    if len(subject) > 12:
+        subject = subject[:12]
+
+    return (
+        f"{subject}决定把眼前的难题变成一次温柔又勇敢的冒险。"
+        "大家先停下来想一想，再一起找到一个善良、安全、让人安心的解决办法，"
+        "故事也朝着明亮的方向继续展开。"
+    )
+
+
+def check_user_transcript_safety(transcript_text: str) -> VoiceSafetyResult:
+    """Screen user transcript text before it enters the story flow."""
+
+    flags = _collect_safety_flags(transcript_text)
+    if not flags:
+        return VoiceSafetyResult(is_safe=True, flags=[])
+
+    message = build_child_safe_redirect(flags)
+    return VoiceSafetyResult(
+        is_safe=False,
+        flags=flags,
+        replacement_text=message,
+        message=message,
+    )
+
+
+def check_assistant_output_safety(
+    assistant_text: str,
+    *,
+    premise: str | None = None,
+) -> VoiceSafetyResult:
+    """Screen assistant output and replace it with a child-safe segment when needed."""
+
+    flags = _collect_safety_flags(assistant_text)
+    if not flags:
+        return VoiceSafetyResult(is_safe=True, flags=[])
+
+    replacement_text = build_safe_story_fallback(premise=premise)
+    return VoiceSafetyResult(
+        is_safe=False,
+        flags=flags,
+        replacement_text=replacement_text,
+        message="系统已把不适合孩子的内容改写为更温和安全的版本。",
+    )