"""《分析任务单》数据模型（理解层产物，intent-understanding-layer 任务 1）。 对应 design.md「Components and Interfaces / 1. Data model」。本模块定义理解层 （``UnderstandingLayer``）在正式分析（``compute``）之前产出的**强结构化、可审计、 可序列化**的中间契约——《分析任务单》（:class:`AnalysisTaskSheet`），以及其内嵌的 文档画像、提取项与澄清决策类型。 设计取向（与需求 / 设计一致）： - **不含任何分析数值计算**：任务单只承载「文档画像 + 意图 + 来源可定位的提取项 + 缺失/歧义 + 澄清决策」，所有分析数值由后续 ``compute`` 阶段（纯 Python）产出 （需求 1.4、Property 4）。 - **可序列化且往返一致**：:meth:`AnalysisTaskSheet.to_dict` / :meth:`AnalysisTaskSheet.from_dict` 互为逆操作（``from_dict(to_dict(x)) == x``， 需求 9.3、Property 9），便于会话状态保存、审计与无 UI / 无 LLM 的单元测试。 - **仅依赖标准库**：纯 dataclass + enum，不触网、无重依赖，便于稳定测试。 枚举值采用稳定的字符串字面量，作为对外序列化契约的一部分（不可随意更名）。 """ from __future__ import annotations from dataclasses import dataclass, field from enum import Enum from typing import Any, Optional # --------------------------------------------------------------------------- # 枚举：意图、表类型、澄清原因码 # --------------------------------------------------------------------------- class Intent(str, Enum): """用户请求意图分类（需求 3）。 - ``DESCRIPTIVE_SUMMARY``：描述性质量属性梳理（分组/统计/限度符合性，不外推）。 - ``SHELF_LIFE``：货架期外推 / 动力学建模。 - ``UNKNOWN``：无法自信分类 → 进入澄清模式（需求 3.4 / 6.4）。 """ DESCRIPTIVE_SUMMARY = "descriptive_summary" SHELF_LIFE = "shelf_life_extrapolation" COMPATIBILITY = "compatibility" UNKNOWN = "unknown" class SlotState(str, Enum): """意图槽位的**来源态**（意图保真核心元信息）。 意图是潜变量——系统只能从「用户话语 + 文档上下文」观测它。为确保**用户真实意图 被采纳**，每个槽位必须显式携带其取值来源，杜绝「系统凭空具象化用户从未表达的意图」： - ``STATED``：用户在指令中**明示**——最高优先级，经确认后锁定，永不被重推断覆盖。 - ``INFERRED``：由文档画像 / 输入信号**语用推断**——可用，但必须向用户**回显来源** 并允许一键改正（推断可以激进，但绝不隐瞒「这是推断」）。 - ``MISSING``：既无明示也无可靠推断——触发**选择性澄清**，绝不填默认值。 注意：**刻意不设 ``ASSUMED`` 态**。静默默认值（如规格 0.5 / 主 CQA「总杂质」/ 目标时间点 [24,36]）是篡改用户意图的暗门，在本保真模型中被禁止——缺失即 ``MISSING``。 """ STATED = "stated" INFERRED = "inferred" MISSING = "missing" class TableType(str, Enum): """文档中检测到的表类型（需求 2.2）。""" RELATED_SUBSTANCES = "related_substances" ASSAY = "assay" DISSOLUTION = "dissolution" CONTENT_UNIFORMITY = "content_uniformity" PHYSICAL_PROPERTIES = "physical_properties" STABILITY_TIME_SERIES = "stability_time_series" UNKNOWN = "unknown" #: 澄清原因码（需求 6）：无 LLM / 意图未知 / 数据与意图不匹配。 CLARIFY_NO_LLM = "no_llm" CLARIFY_INTENT_UNKNOWN = "intent_unknown" CLARIFY_DATA_INTENT_MISMATCH = "data_intent_mismatch" def _coerce_enum(enum_cls, value, default): """把字符串 / 枚举安全归一为 ``enum_cls`` 成员；无法识别回退 ``default``。""" if isinstance(value, enum_cls): return value try: return enum_cls(str(value)) except ValueError: return default # --------------------------------------------------------------------------- # 文档画像 # --------------------------------------------------------------------------- @dataclass class DetectedTable: """文档中检测到的单张表（需求 2.1 / 2.2 / 2.3）。""" table_type: TableType title: str = "" source_document: str = "" # 多文档聚合：记录来源文件名（需求 2.2） is_stability_time_series: bool = False columns: list[str] = field(default_factory=list) sample_rows: list[list[str]] = field(default_factory=list) def to_dict(self) -> dict: return { "table_type": self.table_type.value, "title": self.title, "source_document": self.source_document, "is_stability_time_series": bool(self.is_stability_time_series), "columns": list(self.columns), "sample_rows": [list(r) for r in self.sample_rows], } @staticmethod def from_dict(d: dict) -> "DetectedTable": d = d or {} return DetectedTable( table_type=_coerce_enum(TableType, d.get("table_type"), TableType.UNKNOWN), title=str(d.get("title", "")), source_document=str(d.get("source_document", "")), is_stability_time_series=bool(d.get("is_stability_time_series", False)), columns=[str(c) for c in (d.get("columns") or [])], sample_rows=[[str(c) for c in row] for row in (d.get("sample_rows") or [])], ) @dataclass class DocumentProfile: """文档画像：检测到的表集合 + 是否含稳定性时序（需求 2）。""" tables: list[DetectedTable] = field(default_factory=list) has_stability_time_series: bool = False notes: list[str] = field(default_factory=list) def to_dict(self) -> dict: return { "tables": [t.to_dict() for t in self.tables], "has_stability_time_series": bool(self.has_stability_time_series), "notes": list(self.notes), } @staticmethod def from_dict(d: dict) -> "DocumentProfile": d = d or {} return DocumentProfile( tables=[DetectedTable.from_dict(t) for t in (d.get("tables") or [])], has_stability_time_series=bool(d.get("has_stability_time_series", False)), notes=[str(n) for n in (d.get("notes") or [])], ) # --------------------------------------------------------------------------- # 提取项与澄清决策 # --------------------------------------------------------------------------- @dataclass class ExtractedItem: """单个提取项，必须可回溯到原文（需求 4）。 - ``source_ref``：可在原文定位的片段 / 单元格内容（回填校验依据，需求 4.1）。 - ``located``：是否在原文成功定位；``False`` 表示被标记缺失、不得带入捏造值 （需求 4.2 / Property 1）。 - ``group``：分组维度（如 ``{"strength": "20μg", "batch": "B1", "attribute": "总杂"}``）， 供描述性梳理按规格/批次/属性组织（需求 7.2）。 """ field: str value: str = "" source_ref: str = "" located: bool = True group: dict = field(default_factory=dict) source: str = "llm" # "llm"（模型提取）| "user_edited"（用户手动修改/新增） def to_dict(self) -> dict: return { "field": self.field, "value": self.value, "source_ref": self.source_ref, "located": bool(self.located), "group": dict(self.group), "source": self.source, } @staticmethod def from_dict(d: dict) -> "ExtractedItem": d = d or {} return ExtractedItem( field=str(d.get("field", "")), value=str(d.get("value", "")), source_ref=str(d.get("source_ref", "")), located=bool(d.get("located", True)), group=dict(d.get("group") or {}), source=str(d.get("source", "llm")), ) @dataclass class IntentSlot: """单个**意图槽位**及其保真元信息（意图保真核心载体）。 把「意图」从一次性扁平分类，升级为「带来源、带置信、可锁定、缺失即问」的可确认 结构。每个槽位回答三个保真关键问题：这是用户**说的**还是系统**猜的**（``state`` / ``evidence``）？系统有**多确定**（``confidence``）？用户**改得动**吗（``locked``）？ - ``name``：槽位名（如 ``primary_cqa`` / ``spec_limit`` / ``target_timepoints``）。 - ``value``：取值；``MISSING`` 态下应为 ``None``（绝不填默认）。 - ``state``：来源态（:class:`SlotState`）。 - ``confidence``：0..1 置信度，驱动选择性澄清与确认策略（STATED≈1.0 / INFERRED 居中 / MISSING=0）。 - ``evidence``：明示 / 推断的原文依据片段（对话接地溯源，供回显）。 - ``locked``：用户确认 / 编辑后置 ``True``——**任何重推断都不得覆盖**（采纳真实意图的硬保证）。 - ``affects_result``：是否影响计算结论，驱动「信息价值」澄清决策（关键槽缺失才必问）。 """ name: str value: Any = None state: SlotState = SlotState.MISSING confidence: float = 0.0 evidence: str = "" locked: bool = False affects_result: bool = True def needs_clarification(self) -> bool: """该槽位是否应触发澄清：缺失、且影响结论、且未被用户锁定。""" return ( self.state is SlotState.MISSING and self.affects_result and not self.locked ) def to_dict(self) -> dict: return { "name": self.name, "value": self.value, "state": self.state.value, "confidence": float(self.confidence), "evidence": self.evidence, "locked": bool(self.locked), "affects_result": bool(self.affects_result), } @staticmethod def from_dict(d: dict) -> "IntentSlot": d = d or {} return IntentSlot( name=str(d.get("name", "")), value=d.get("value"), state=_coerce_enum(SlotState, d.get("state"), SlotState.MISSING), confidence=float(d.get("confidence", 0.0) or 0.0), evidence=str(d.get("evidence", "")), locked=bool(d.get("locked", False)), affects_result=bool(d.get("affects_result", True)), ) @dataclass class ClarificationDecision: """澄清决策（需求 6）：是否需要向用户澄清及其原因。""" needed: bool = False reason_code: str = "" # CLARIFY_* 之一 message_key: str = "" # i18n key（界面提示文案） options: list[str] = field(default_factory=list) def to_dict(self) -> dict: return { "needed": bool(self.needed), "reason_code": self.reason_code, "message_key": self.message_key, "options": list(self.options), } @staticmethod def from_dict(d: dict) -> "ClarificationDecision": d = d or {} return ClarificationDecision( needed=bool(d.get("needed", False)), reason_code=str(d.get("reason_code", "")), message_key=str(d.get("message_key", "")), options=[str(o) for o in (d.get("options") or [])], ) # --------------------------------------------------------------------------- # 分析任务单 # --------------------------------------------------------------------------- @dataclass class AnalysisTaskSheet: """《分析任务单》：理解层的强结构化产物，经用户确认后驱动分析（需求 5）。 字段： - ``intent``：分类意图（需求 3.5）。 - ``document_profile``：文档画像（需求 2）。 - ``extracted_items``：来源可定位的提取项（需求 4）。 - ``proposed_skill_id``：建议路由的 Skill id（用户可编辑，需求 5.3 / 8 路由）。 - ``missing_items`` / ``ambiguous_items``：缺失与歧义清单（需求 4.5）。 - ``clarification``：澄清决策；``None`` 表示无需澄清（需求 6）。 - ``source_method``：``"llm"`` | ``"clarification"``，标注任务单来源。 """ intent: Intent = Intent.UNKNOWN document_profile: DocumentProfile = field(default_factory=DocumentProfile) extracted_items: list[ExtractedItem] = field(default_factory=list) proposed_skill_id: str = "" missing_items: list[str] = field(default_factory=list) ambiguous_items: list[str] = field(default_factory=list) clarification: Optional[ClarificationDecision] = None source_method: str = "llm" #: 意图槽位（保真核心）：带来源态 / 置信度 / 锁定的可确认意图维度（需求 3 / 5）。 #: 向后兼容默认空列表——旧任务单与未启用槽位的技能不受影响。 intent_slots: list[IntentSlot] = field(default_factory=list) def needs_clarification(self) -> bool: """是否处于澄清状态（澄清决策存在且 needed=True）。""" return self.clarification is not None and bool(self.clarification.needed) def slots_needing_clarification(self) -> list[IntentSlot]: """返回应触发澄清的关键意图槽位（缺失、影响结论、未锁定）。 供选择性澄清（信息价值驱动）使用：只对真正影响结果且不确定的槽位发问， 避免过度澄清。返回空列表表示无需就意图槽位追问。 """ return [s for s in self.intent_slots if s.needs_clarification()] def get_slot(self, name: str) -> Optional[IntentSlot]: """按名取意图槽位；不存在返回 ``None``。""" for s in self.intent_slots: if s.name == name: return s return None def to_dict(self) -> dict: return { "intent": self.intent.value, "document_profile": self.document_profile.to_dict(), "extracted_items": [it.to_dict() for it in self.extracted_items], "proposed_skill_id": self.proposed_skill_id, "missing_items": list(self.missing_items), "ambiguous_items": list(self.ambiguous_items), "clarification": self.clarification.to_dict() if self.clarification else None, "source_method": self.source_method, "intent_slots": [s.to_dict() for s in self.intent_slots], } @staticmethod def from_dict(d: dict) -> "AnalysisTaskSheet": d = d or {} clar = d.get("clarification") return AnalysisTaskSheet( intent=_coerce_enum(Intent, d.get("intent"), Intent.UNKNOWN), document_profile=DocumentProfile.from_dict(d.get("document_profile") or {}), extracted_items=[ExtractedItem.from_dict(it) for it in (d.get("extracted_items") or [])], proposed_skill_id=str(d.get("proposed_skill_id", "")), missing_items=[str(m) for m in (d.get("missing_items") or [])], ambiguous_items=[str(a) for a in (d.get("ambiguous_items") or [])], clarification=ClarificationDecision.from_dict(clar) if clar else None, source_method=str(d.get("source_method", "llm")), intent_slots=[IntentSlot.from_dict(s) for s in (d.get("intent_slots") or [])], ) __all__ = [ "Intent", "SlotState", "TableType", "CLARIFY_NO_LLM", "CLARIFY_INTENT_UNKNOWN", "CLARIFY_DATA_INTENT_MISMATCH", "DetectedTable", "DocumentProfile", "ExtractedItem", "IntentSlot", "ClarificationDecision", "AnalysisTaskSheet", ]