life-echo/api/app/agents/state_schema.py

"""
共享状态 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",
]