191 lines
7.3 KiB
Python
191 lines
7.3 KiB
Python
"""叩心枢开场消息生成器。
|
||
|
||
设计 docs/domain/core-domains.md §6。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from dataclasses import dataclass
|
||
from datetime import UTC, datetime
|
||
|
||
from app.ai_provider.manager import ProviderManagerFactory
|
||
from app.core.config import settings
|
||
from app.core.logging import logger
|
||
from app.modules.knocking.conditions import ConditionResult
|
||
from app.modules.personality.emotion import EmotionState, to_prompt_text
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 回退模板
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_FALLBACK_TEMPLATES: dict[str, str] = {
|
||
"idle": "{{name}}默默地看了你一眼。",
|
||
"time_of_day": "{{name}}正望着窗外出神。",
|
||
"anniversary": "今日似乎是个特别的日子。",
|
||
"reunion": "{{name}}注意到你回来了。",
|
||
"emotion": "{{name}}轻轻叹了口气。",
|
||
"curiosity": "{{name}}似乎有什么话想说。",
|
||
"memory_recall": "{{name}}若有所思地望向远方。",
|
||
"big_cycle_completed": "{{name}}似乎有所感悟。",
|
||
"evolution_applied": "{{name}}看上去有些不一样了。",
|
||
"relationship_milestone": "{{name}}望着你，嘴角泛起一丝笑意。",
|
||
"default": "{{name}}若有所思。",
|
||
}
|
||
|
||
_FALLBACK = "{{name}}正望着窗外出神。"
|
||
|
||
|
||
@dataclass
|
||
class ChatMessageBrief:
|
||
"""简略对话记录（只携带必要字段，避免依赖 ORM）。"""
|
||
|
||
role: str = ""
|
||
content: str = ""
|
||
|
||
|
||
@dataclass
|
||
class OpenerResult:
|
||
"""开场消息生成结果。"""
|
||
|
||
text: str
|
||
used_fallback: bool = False
|
||
|
||
|
||
async def generate_opener(
|
||
character_id: str,
|
||
character_name: str,
|
||
personality_summary: str | None,
|
||
emotion: EmotionState | None,
|
||
condition_result: ConditionResult,
|
||
*,
|
||
chat_history: list[ChatMessageBrief] | None = None,
|
||
memory_context: str | None = None,
|
||
now: datetime | None = None,
|
||
) -> OpenerResult:
|
||
"""根据条件触发结果生成一句开场白。
|
||
|
||
依次尝试：
|
||
1. 调用神识（LLM）生成带人格注入的开场白
|
||
2. 失败时使用模板化回退
|
||
|
||
若提供 ``chat_history``，会将其拼入 prompt 使开场更贴合最近对话脉络。
|
||
若提供 ``memory_context``，会将其作为角色的长期记忆注入 prompt。
|
||
"""
|
||
now = now or datetime.now(UTC)
|
||
timestamp = now.strftime("%Y-%m-%d %H:%M")
|
||
|
||
# 构造 prompt
|
||
emotion_text = to_prompt_text(emotion) if emotion else "平静"
|
||
summary = personality_summary or f"你是一个名叫{character_name}的角色。{_DEFAULT_PERSONA}"
|
||
trigger_reason = condition_result.trigger_reason or ""
|
||
|
||
context_parts = _build_context_parts(condition_result)
|
||
|
||
system_prompt = _build_system_prompt(summary, emotion_text, timestamp)
|
||
user_prompt = _build_user_prompt(
|
||
character_name, trigger_reason, context_parts, chat_history, memory_context
|
||
)
|
||
|
||
try:
|
||
manager = ProviderManagerFactory.create()
|
||
result = await manager.chat(
|
||
manager.to_message_list([system_prompt], [("user", user_prompt)]),
|
||
model_hint=settings.big_cycle_model,
|
||
)
|
||
text = result.content.strip()
|
||
if not text:
|
||
raise ValueError("模型返回空开场文本")
|
||
return OpenerResult(text=text, used_fallback=False)
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"knocking_pivot_opener_fallback",
|
||
character_id=character_id,
|
||
condition=condition_result.condition_name,
|
||
error=str(exc),
|
||
)
|
||
template = _FALLBACK_TEMPLATES.get(condition_result.condition_name, _FALLBACK)
|
||
fallback = template.replace("{{name}}", character_name)
|
||
return OpenerResult(text=fallback, used_fallback=True)
|
||
|
||
|
||
def _build_context_parts(condition_result: ConditionResult) -> list[str]:
|
||
"""组装条件相关的上下文片段。"""
|
||
parts: list[str] = []
|
||
if condition_result.context.get("anniversary_kind"):
|
||
kind_label = condition_result.context.get("anniversary_label", "")
|
||
parts.append(f"今天是一个特别的日子：{kind_label}。")
|
||
if condition_result.context.get("curiosity_question"):
|
||
parts.append(
|
||
f"你一直想问用户一个问题：「{condition_result.context['curiosity_question']}」"
|
||
)
|
||
if condition_result.context.get("time_of_day") == "late_night":
|
||
parts.append("夜深了，万籁俱寂。")
|
||
if condition_result.context.get("memory_title"):
|
||
title = condition_result.context["memory_title"]
|
||
parts.append(f"你最近想起了一件事——「{title}」。")
|
||
if condition_result.context.get("big_cycle_summary"):
|
||
parts.append("你刚刚审视过自己的内心，有了一些新的思考。")
|
||
if condition_result.context.get("evolution_title"):
|
||
title = condition_result.context["evolution_title"]
|
||
parts.append(f"你感觉自己有了一些新的变化：{title}。")
|
||
if condition_result.context.get("milestone_type") == "messages":
|
||
count = condition_result.context["milestone_value"]
|
||
parts.append(f"你和用户的对话已经累积了 {count} 条。")
|
||
if condition_result.context.get("milestone_type") == "days":
|
||
days = condition_result.context["milestone_value"]
|
||
parts.append(f"从你们第一次对话至今，已经过去了 {days} 天。")
|
||
return parts
|
||
|
||
|
||
def _format_chat_history(history: list[ChatMessageBrief]) -> str:
|
||
"""将最近对话历史格式化为文本。"""
|
||
lines = ["\n[最近对话记录]"]
|
||
for msg in history:
|
||
speaker = "你" if msg.role == "assistant" else "用户"
|
||
lines.append(f"{speaker}：{msg.content[:200]}")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def _build_system_prompt(personality_summary: str, emotion_text: str, timestamp: str) -> str:
|
||
return (
|
||
"你是一位角色扮演助手。以下是你的人物设定：\n"
|
||
f"{personality_summary}\n\n"
|
||
"[当前状态]\n"
|
||
f"- 当前时间: {timestamp}\n"
|
||
f"- 情绪状态: {emotion_text}\n"
|
||
)
|
||
|
||
|
||
def _build_user_prompt(
|
||
character_name: str,
|
||
trigger_reason: str,
|
||
context_parts: list[str],
|
||
chat_history: list[ChatMessageBrief] | None = None,
|
||
memory_context: str | None = None,
|
||
) -> str:
|
||
lines = ["[任务]"]
|
||
lines.append(f"根据以上设定，生成一句自然、贴合{character_name}性格和当前情绪的开场白。")
|
||
lines.append("要求：")
|
||
lines.append("1. 一句话，不超过 80 字")
|
||
lines.append("2. 符合角色的表达风格和说话方式、语气")
|
||
lines.append("3. 反映当前情绪状态")
|
||
lines.append("4. 不要问「你还在吗？」之类的问题")
|
||
lines.append("5. 直接输出开场白，不要任何前缀或引号")
|
||
if trigger_reason:
|
||
lines.append(f"\n触发原因：{trigger_reason}")
|
||
if context_parts:
|
||
lines.append(f"\n{' '.join(context_parts)}")
|
||
if memory_context:
|
||
lines.append(f"\n{memory_context}")
|
||
if chat_history:
|
||
lines.append(_format_chat_history(chat_history))
|
||
lines.append("\n注意：延续上次对话的语境自然地开启新话题，不要重复上一句的内容。")
|
||
return "\n".join(lines)
|
||
|
||
|
||
_DEFAULT_PERSONA = (
|
||
"你对用户保持着温和的关注，"
|
||
"平日里话不多，但每次开口都带着真诚。"
|
||
"你善于观察，能从细微之处感知对方的情绪。"
|
||
)
|