""" 共享状态 Schema（对话 Agent 与后台 Agent 共用）。 Option B 的 schema 分层： - `NarrativeCoverageState`：叙述覆盖视图 —— 阶段推进、叙述槽、每阶段的完成情况；**不掺控制元信息**。 - `InterviewControlState`：访谈控制视图 —— 已确认事实、人物主线、最近已问；用于访谈控场，**不参与成稿槽位**。 - `MemoirStateSchema`：持久化 DTO，只承载字段；消费方显式投影到两个视图。 """ from __future__ import annotations from typing import Dict, List, Optional from pydantic import BaseModel, Field from app.agents.stage_constants import CHAT_STAGES class SlotData(BaseModel): """Slot 数据结构""" snippet: Optional[str] = None segment_ids: List[str] = Field(default_factory=list) class KnownFact(BaseModel): """会话级已知事实：供 prompt 明确声明“不要再问这些”。""" label: str value: str source: str = "" stage: str = "" slot_name: str | None = None def prompt_line(self) -> str: prefix = f"{self.label}：".strip("：") if prefix: return f"{prefix} {self.value}".strip() return self.value.strip() class PersonaThread(BaseModel): """跨轮人物主线：用于持续呼应用户的稳定特质与动机。""" trait: str evidence: str = "" source: str = "" stage: str = "" def prompt_line(self) -> str: if self.evidence: return f"{self.trait}（依据：{self.evidence}）" return self.trait # ============================================================================= # Narrative 视图：叙述覆盖、阶段推进、叙述槽 # ============================================================================= class NarrativeCoverageState(BaseModel): """叙述覆盖视图。 只承载「人生叙事覆盖」相关信息：阶段顺序、当前阶段、覆盖过的阶段、每阶段的叙述槽。 **禁止**在此视图承载访谈控场数据（已确认事实、人物主线、最近已问），那些数据属于 `InterviewControlState`。 """ stage_order: List[str] current_stage: str covered_stages: List[str] slots: Dict[str, Dict[str, SlotData]] def empty_slots_for_current_stage(self) -> List[str]: stage_slots = self.slots.get(self.current_stage, {}) return [key for key, value in stage_slots.items() if not value.snippet] def empty_slots_for_stage(self, stage: str) -> List[str]: stage_slots = self.slots.get(stage, {}) return [key for key, value in stage_slots.items() if not value.snippet] def filled_slots_for_stage(self, stage: str) -> Dict[str, str]: stage_slots = self.slots.get(stage, {}) return { key: value.snippet for key, value in stage_slots.items() if value.snippet } def all_stages_coverage(self) -> Dict[str, Dict]: coverage: Dict[str, Dict] = {} for stage in self.stage_order: stage_slots = self.slots.get(stage, {}) total = len(stage_slots) filled = sum(1 for v in stage_slots.values() if v.snippet) coverage[stage] = { "total": total, "filled": filled, "empty": total - filled, "ratio": filled / total if total > 0 else 0, } return coverage # ============================================================================= # Interview Control 视图：访谈控场（已知事实 / 人物主线 / 最近已问） # ============================================================================= class InterviewControlState(BaseModel): """访谈控制视图。 承载仅与「控场 / 去重问 / 人物呼应」相关的信息。这些字段**不应**出现在叙述覆盖计算里，也 **不应**写入 `slots`。 """ known_facts: List[KnownFact] = Field(default_factory=list) persona_threads: List[PersonaThread] = Field(default_factory=list) recent_questions: List[str] = Field(default_factory=list) def prompt_known_fact_lines(self, *, limit: int = 10) -> List[str]: xs: List[str] = [] for fact in self.known_facts[-limit:]: line = fact.prompt_line().strip() if line: xs.append(line) return xs def prompt_persona_thread_lines(self, *, limit: int = 6) -> List[str]: xs: List[str] = [] for item in self.persona_threads[-limit:]: line = item.prompt_line().strip() if line: xs.append(line) return xs def prompt_recent_question_lines(self, *, limit: int = 4) -> List[str]: out: List[str] = [] seen: set[str] = set() for item in self.recent_questions[-limit:]: s = str(item).strip() if not s or s in seen: continue seen.add(s) out.append(s) return out def blocked_slot_names_for_stage(self, stage: str) -> set[str]: """已被 known_facts 覆盖的槽位名：追问时应避开。""" return { fact.slot_name for fact in self.known_facts if fact.slot_name and (not fact.stage or fact.stage == stage) } def prompt_empty_slots_for_stage( self, narrative: NarrativeCoverageState, stage: str ) -> List[str]: """生成 prompt 时可追问的槽位，排除已被 known_facts 覆盖的方向。""" blocked = self.blocked_slot_names_for_stage(stage) return [ key for key in narrative.empty_slots_for_stage(stage) if key not in blocked ] # ============================================================================= # 持久化 DTO：不承载视图委托逻辑 # ============================================================================= class MemoirStateSchema(BaseModel): """回忆录状态持久化 DTO。视图逻辑由显式 state projector 提供。""" stage_order: List[str] current_stage: str covered_stages: List[str] slots: Dict[str, Dict[str, SlotData]] known_facts: List[KnownFact] = Field(default_factory=list) persona_threads: List[PersonaThread] = Field(default_factory=list) recent_questions: List[str] = Field(default_factory=list) DEFAULT_STAGE_ORDER: list[str] = list(CHAT_STAGES) def narrative_coverage_state(state: MemoirStateSchema) -> NarrativeCoverageState: return NarrativeCoverageState( stage_order=state.stage_order, current_stage=state.current_stage, covered_stages=state.covered_stages, slots=state.slots, ) def interview_control_state(state: MemoirStateSchema) -> InterviewControlState: return InterviewControlState( known_facts=state.known_facts, persona_threads=state.persona_threads, recent_questions=state.recent_questions, ) def prompt_empty_slots_for_current_stage(state: MemoirStateSchema) -> List[str]: narrative = narrative_coverage_state(state) control = interview_control_state(state) return control.prompt_empty_slots_for_stage(narrative, state.current_stage) def default_slots() -> Dict[str, Dict[str, SlotData]]: return { "childhood": { "place": SlotData(), "people": SlotData(), "daily_life": SlotData(), "emotion": SlotData(), "turning_event": SlotData(), }, "education": { "school": SlotData(), "city": SlotData(), "motivation": SlotData(), "challenge": SlotData(), "change": SlotData(), }, "career": { "job": SlotData(), "environment": SlotData(), "decision": SlotData(), "pressure": SlotData(), "growth": SlotData(), }, "family": { "relationship": SlotData(), "conflict": SlotData(), "support": SlotData(), "responsibility": SlotData(), "change": SlotData(), }, "belief": { "value": SlotData(), "regret": SlotData(), "pride": SlotData(), "lesson": SlotData(), }, } def default_state() -> MemoirStateSchema: return MemoirStateSchema( stage_order=DEFAULT_STAGE_ORDER, current_stage=DEFAULT_STAGE_ORDER[0], covered_stages=[], slots=default_slots(), ) __all__ = [ "DEFAULT_STAGE_ORDER", "InterviewControlState", "KnownFact", "MemoirStateSchema", "NarrativeCoverageState", "PersonaThread", "SlotData", "default_slots", "default_state", "interview_control_state", "narrative_coverage_state", "prompt_empty_slots_for_current_stage", ]