367 lines
15 KiB
Python
367 lines
15 KiB
Python
"""叩心枢（KnockingPivot）主循环。
|
|
|
|
人物主动对话守护进程：后台常驻循环（~30s tick），
|
|
遍历活跃角色 → 检查触发条件 → 生成开场 → 创建会话/消息 → 推一炁事件。
|
|
同时订阅一炁事件流（big_cycle.* / reflection.*）实现近实时快速路径。
|
|
|
|
叩心枢与辰正（时感）、灵核（情绪）并列——辰正提供时态语境，
|
|
灵核产出情绪状态，叩心判断「此时该不该开口」。
|
|
|
|
设计 docs/domain/core-domains.md §4。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import contextlib
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
|
|
|
from app.core.logging import logger
|
|
from app.db.session import AsyncSessionLocal
|
|
from app.models.character import Character
|
|
from app.models.chat import ChatMessage, ChatSession
|
|
from app.models.personality_emotion import CharacterEmotionState
|
|
from app.modules.knocking.conditions import (
|
|
AnniversaryCondition,
|
|
BigCycleCompletedCondition,
|
|
CuriosityCondition,
|
|
EmotionChangeCondition,
|
|
EvolutionAppliedCondition,
|
|
MemoryRecallCondition,
|
|
ProactiveCondition,
|
|
RelationshipMilestoneCondition,
|
|
TimeElapsedCondition,
|
|
TimeOfDayCondition,
|
|
UserReunionCondition,
|
|
)
|
|
from app.modules.knocking.creator import create_proactive_message
|
|
from app.modules.knocking.opener import ChatMessageBrief, generate_opener
|
|
from app.modules.personality.emotion import EmotionState
|
|
from app.modules.runtime.events import Event
|
|
from app.services.event_store import DbEventStore
|
|
from app.services.memory import memory_context as _fmt_memory_context
|
|
from app.services.personality import build_personality_summary
|
|
|
|
|
|
class KnockingPivot:
|
|
"""叩心枢主动对话守护进程。
|
|
|
|
用法：
|
|
pivot = KnockingPivot(event_store)
|
|
await pivot.start()
|
|
...
|
|
await pivot.stop()
|
|
"""
|
|
|
|
# 条件注册表（第一梯队 + 第二梯队 ▸ 按优先级排列）
|
|
conditions: list[ProactiveCondition] = [
|
|
EvolutionAppliedCondition(),
|
|
BigCycleCompletedCondition(),
|
|
TimeElapsedCondition(min_idle_minutes=360),
|
|
TimeOfDayCondition(windows=["morning", "late_night"]),
|
|
AnniversaryCondition(),
|
|
UserReunionCondition(),
|
|
EmotionChangeCondition(min_delta=0.3),
|
|
MemoryRecallCondition(),
|
|
RelationshipMilestoneCondition(),
|
|
CuriosityCondition(),
|
|
]
|
|
|
|
def __init__(
|
|
self,
|
|
event_store: DbEventStore,
|
|
session_factory: async_sessionmaker[AsyncSession] | None = None,
|
|
*,
|
|
tick_interval: float = 30.0,
|
|
cooldown_minutes: int = 30,
|
|
daily_limit: int = 3,
|
|
quiet_hours_start: int = 23,
|
|
quiet_hours_end: int = 8,
|
|
) -> None:
|
|
self._event_store = event_store
|
|
self._session_factory = session_factory or AsyncSessionLocal
|
|
self._tick_interval = tick_interval
|
|
self._cooldown_minutes = cooldown_minutes
|
|
self._daily_limit = daily_limit
|
|
self._quiet_hours_start = quiet_hours_start
|
|
self._quiet_hours_end = quiet_hours_end
|
|
|
|
self._task: asyncio.Task[None] | None = None
|
|
self._event_task: asyncio.Task[None] | None = None
|
|
self._stop_event = asyncio.Event()
|
|
self._event_queue: asyncio.Queue[Event] | None = None
|
|
|
|
# per-character 冷启动冷却（character_id → 下次可触发时间）
|
|
self._cooldowns: dict[str, datetime] = {}
|
|
# per-character 日计数（character_id → 今日日期 → 次数）
|
|
self._daily_counts: dict[str, tuple[str, int]] = {}
|
|
|
|
# ------------------------------------------------------------------
|
|
# 生命周期
|
|
# ------------------------------------------------------------------
|
|
|
|
async def start(self) -> None:
|
|
if self._task is not None:
|
|
logger.warning("knocking_pivot_already_running")
|
|
return
|
|
self._stop_event.clear()
|
|
|
|
# 订阅一炁事件流（大周天/小周天完成等），实现近实时快速路径
|
|
self._event_queue = self._event_store.subscribe_by_pattern("big_cycle.*,reflection.*")
|
|
self._event_task = asyncio.create_task(self._run_event_consumer())
|
|
|
|
self._task = asyncio.create_task(self._run_loop())
|
|
logger.info("knocking_pivot_started", tick_interval=self._tick_interval)
|
|
|
|
async def stop(self) -> None:
|
|
if self._task is None:
|
|
return
|
|
self._stop_event.set()
|
|
self._task.cancel()
|
|
with contextlib.suppress(asyncio.CancelledError):
|
|
await self._task
|
|
self._task = None
|
|
if self._event_task is not None:
|
|
self._event_task.cancel()
|
|
with contextlib.suppress(asyncio.CancelledError):
|
|
await self._event_task
|
|
self._event_task = None
|
|
if self._event_queue is not None:
|
|
self._event_store.unsubscribe_by_pattern("big_cycle.*,reflection.*", self._event_queue)
|
|
self._event_queue = None
|
|
logger.info("knocking_pivot_stopped")
|
|
|
|
# ------------------------------------------------------------------
|
|
# 主循环
|
|
# ------------------------------------------------------------------
|
|
|
|
async def _run_loop(self) -> None:
|
|
while not self._stop_event.is_set():
|
|
try:
|
|
await self._tick()
|
|
except asyncio.CancelledError:
|
|
break
|
|
except Exception as exc:
|
|
err_str = str(exc)
|
|
if "database is locked" in err_str:
|
|
logger.warning(
|
|
"knocking_pivot_db_locked_retry",
|
|
error=err_str[:120],
|
|
)
|
|
else:
|
|
logger.error("knocking_pivot_tick_error", error=err_str)
|
|
await asyncio.sleep(self._tick_interval)
|
|
|
|
async def _run_event_consumer(self) -> None:
|
|
"""事件消费者：订阅 big_cycle.* / reflection.* 事件，近实时触发叩心检查。
|
|
|
|
与 _run_loop 并行运行，事件驱动而非轮询驱动。
|
|
每个事件提取 character_id，立即做一次条件检查（跳过冷却/日限等节流逻辑用轮询管理）。
|
|
"""
|
|
queue = self._event_queue
|
|
if queue is None:
|
|
return
|
|
while not self._stop_event.is_set():
|
|
try:
|
|
event = await asyncio.wait_for(queue.get(), timeout=30)
|
|
except TimeoutError:
|
|
continue
|
|
except asyncio.CancelledError:
|
|
break
|
|
try:
|
|
await self._handle_event(event)
|
|
except Exception: # noqa: BLE001
|
|
logger.warning("knocking_pivot_event_error", event_type=event.type)
|
|
|
|
async def _handle_event(self, event: Event) -> None:
|
|
"""响应一炁事件：提取 character_id，尝试用对应条件触发。"""
|
|
character_id = event.character_id
|
|
if not character_id:
|
|
return
|
|
now = datetime.now(UTC)
|
|
if self._in_quiet_hours(now):
|
|
return
|
|
if not self._can_trigger(character_id, now):
|
|
return
|
|
|
|
# 大周天完成 → 快速检查 BigCycleCompletedCondition
|
|
if event.type == "big_cycle.completed":
|
|
async with self._session_factory() as session:
|
|
cond = BigCycleCompletedCondition()
|
|
result = await cond.check(character_id, session, now=now)
|
|
if result.triggered:
|
|
await self._execute_trigger(character_id, result, session, now)
|
|
self._mark_triggered(character_id, now)
|
|
return
|
|
|
|
# 反思完成 → 快速检查 MemoryRecallCondition
|
|
if event.type == "reflection.completed":
|
|
async with self._session_factory() as session:
|
|
cond = MemoryRecallCondition()
|
|
result = await cond.check(character_id, session, now=now)
|
|
if result.triggered:
|
|
await self._execute_trigger(character_id, result, session, now)
|
|
self._mark_triggered(character_id, now)
|
|
return
|
|
|
|
async def _tick(self) -> None:
|
|
"""一次 tick：找出活跃角色 → 逐角色检查条件。"""
|
|
now = datetime.now(UTC)
|
|
if self._in_quiet_hours(now):
|
|
return
|
|
|
|
async with self._session_factory() as session:
|
|
active_characters = await self._list_active_characters(session)
|
|
if not active_characters:
|
|
return
|
|
|
|
for character_id in active_characters:
|
|
if not self._can_trigger(character_id, now):
|
|
continue
|
|
if await self._check_and_trigger(character_id, session, now):
|
|
# 每次 tick 只触发一个角色——命中即停
|
|
break
|
|
|
|
# ------------------------------------------------------------------
|
|
# 内部
|
|
# ------------------------------------------------------------------
|
|
|
|
async def _list_active_characters(self, session: AsyncSession) -> list[str]:
|
|
"""返回全部 status='active' 的 character_id。"""
|
|
result = await session.execute(select(Character.id).where(Character.status == "active"))
|
|
return [row[0] for row in result.all()]
|
|
|
|
def _in_quiet_hours(self, now: datetime) -> bool:
|
|
hour = now.hour
|
|
if self._quiet_hours_start <= self._quiet_hours_end:
|
|
return self._quiet_hours_start <= hour < self._quiet_hours_end
|
|
return hour >= self._quiet_hours_start or hour < self._quiet_hours_end
|
|
|
|
def _can_trigger(self, character_id: str, now: datetime) -> bool:
|
|
# 冷却
|
|
cooldown_until = self._cooldowns.get(character_id)
|
|
if cooldown_until is not None and now < cooldown_until:
|
|
return False
|
|
# 日上限
|
|
today = now.date().isoformat()
|
|
last_date, count = self._daily_counts.get(character_id, ("", 0))
|
|
return not (last_date == today and count >= self._daily_limit)
|
|
|
|
def _mark_triggered(self, character_id: str, now: datetime) -> None:
|
|
self._cooldowns[character_id] = now + timedelta(minutes=self._cooldown_minutes)
|
|
today = now.date().isoformat()
|
|
last_date, count = self._daily_counts.get(character_id, ("", 0))
|
|
if last_date == today:
|
|
self._daily_counts[character_id] = (today, count + 1)
|
|
else:
|
|
self._daily_counts[character_id] = (today, 1)
|
|
|
|
async def _check_and_trigger(
|
|
self, character_id: str, session: AsyncSession, now: datetime
|
|
) -> bool:
|
|
"""遍历条件列表，任一触发则执行并返回 True；否则返回 False。"""
|
|
for condition in self.conditions:
|
|
result = await condition.check(character_id, session, now=now)
|
|
if not result.triggered:
|
|
continue
|
|
|
|
await self._execute_trigger(character_id, result, session, now)
|
|
self._mark_triggered(character_id, now)
|
|
return True
|
|
return False
|
|
|
|
async def _execute_trigger(
|
|
self,
|
|
character_id: str,
|
|
condition_result,
|
|
session: AsyncSession,
|
|
now: datetime,
|
|
) -> None:
|
|
"""条件触发后的执行链：人格注入 → 开场（带上下文）→ 追加/新建会话 → 推事件。"""
|
|
try:
|
|
# 1. 加载角色对象（供人格摘要使用）
|
|
result = await session.execute(select(Character).where(Character.id == character_id))
|
|
character = result.scalar_one_or_none()
|
|
if character is None:
|
|
logger.warning(
|
|
"knocking_pivot_character_not_found",
|
|
character_id=character_id,
|
|
)
|
|
return
|
|
|
|
# 2. 人格注入
|
|
summary = await build_personality_summary(session, character)
|
|
|
|
# 3. 读取情绪
|
|
emotion = await self._get_emotion(session, character_id)
|
|
|
|
# 4. 查找该角色已有的活跃会话（优先复用）
|
|
session_result = await session.execute(
|
|
select(ChatSession)
|
|
.where(
|
|
ChatSession.character_id == character_id,
|
|
ChatSession.status == "active",
|
|
)
|
|
.order_by(ChatSession.updated_at.desc())
|
|
.limit(1)
|
|
)
|
|
existing_session = session_result.scalar_one_or_none()
|
|
|
|
# 5. 读取最近对话记录作为上下文
|
|
chat_history: list[ChatMessageBrief] = []
|
|
if existing_session:
|
|
msg_result = await session.execute(
|
|
select(ChatMessage)
|
|
.where(ChatMessage.session_id == existing_session.id)
|
|
.order_by(ChatMessage.created_at.desc())
|
|
.limit(5)
|
|
)
|
|
raw = list(reversed(msg_result.scalars().all()))
|
|
chat_history = [ChatMessageBrief(role=m.role, content=m.content) for m in raw]
|
|
|
|
# 6. 检索角色的长期记忆作为上下文
|
|
memory_context: str | None = None
|
|
mc = await _fmt_memory_context(session, character_id, limit=5)
|
|
memory_context = mc if mc else None
|
|
|
|
# 7. 生成开场（携带上下文与记忆）
|
|
character_name = character.display_name or character.name or character_id
|
|
opener = await generate_opener(
|
|
character_id=character_id,
|
|
character_name=character_name,
|
|
personality_summary=summary.text if summary and summary.text else None,
|
|
emotion=emotion,
|
|
condition_result=condition_result,
|
|
chat_history=chat_history,
|
|
memory_context=memory_context,
|
|
now=now,
|
|
)
|
|
|
|
# 8. 追加消息（复用已有会话，或创建新会话）
|
|
existing_id = existing_session.id if existing_session else None
|
|
await create_proactive_message(
|
|
session=session,
|
|
event_store=self._event_store,
|
|
character_id=character_id,
|
|
session_id=existing_id,
|
|
opener_text=opener.text,
|
|
condition_name=condition_result.condition_name,
|
|
now=now,
|
|
)
|
|
await session.commit()
|
|
except Exception:
|
|
await session.rollback()
|
|
raise
|
|
|
|
async def _get_emotion(self, session: AsyncSession, character_id: str) -> EmotionState | None:
|
|
result = await session.execute(
|
|
select(CharacterEmotionState).where(CharacterEmotionState.character_id == character_id)
|
|
)
|
|
row = result.scalar_one_or_none()
|
|
if row is None:
|
|
return None
|
|
return EmotionState(mood=row.mood, intensity=row.intensity)
|