| """意图保真：意图槽位（IntentSlot）派生与不变量测试。 |
| |
| 覆盖「确保用户真实意图被采纳」的核心保真规则： |
| - 用户明示 → STATED；仅可推断 → INFERRED；缺失 / 歧义 → MISSING（**绝不填默认**）。 |
| - 选择性澄清：仅对「影响结论且不确定且未锁定」的槽位追问。 |
| - 用户锁定后不再被澄清（采纳真实意图的硬保证）。 |
| - 序列化往返一致（含 intent_slots）。 |
| """ |
|
|
| from __future__ import annotations |
|
|
| from kernel.task_sheet import ( |
| AnalysisTaskSheet, |
| ExtractedItem, |
| Intent, |
| IntentSlot, |
| SlotState, |
| ) |
| from kernel.understanding import ( |
| SLOT_PRIMARY_CQA, |
| SLOT_SPEC_LIMIT, |
| SLOT_TARGET_TIMEPOINTS, |
| build_intent_slots, |
| ) |
|
|
|
|
| def _slot(slots, name): |
| return next((s for s in slots if s.name == name), None) |
|
|
|
|
| |
| |
| |
|
|
| def test_explicit_goal_yields_stated_slots(): |
| """用户明示主 CQA 与目标时间点 → STATED，置信 1.0。""" |
| slots = build_intent_slots( |
| goal="请把总杂质外推到36个月", intent=Intent.SHELF_LIFE, extracted_items=[] |
| ) |
| tp = _slot(slots, SLOT_TARGET_TIMEPOINTS) |
| cqa = _slot(slots, SLOT_PRIMARY_CQA) |
| assert tp.state is SlotState.STATED and tp.value == [36] and tp.confidence == 1.0 |
| assert cqa.state is SlotState.STATED and cqa.value == "总杂" |
|
|
|
|
| def test_missing_slots_have_no_default_values(): |
| """指令与文档都未给出关键槽位 → MISSING 且 value 为 None（禁止 0.5 / 总杂质 / [24,36]）。""" |
| slots = build_intent_slots( |
| goal="预测一下稳定性", intent=Intent.SHELF_LIFE, extracted_items=[] |
| ) |
| tp = _slot(slots, SLOT_TARGET_TIMEPOINTS) |
| cqa = _slot(slots, SLOT_PRIMARY_CQA) |
| assert tp.state is SlotState.MISSING and tp.value is None |
| assert cqa.state is SlotState.MISSING and cqa.value is None |
| |
| assert all(s.state is not SlotState.MISSING or s.value is None for s in slots) |
|
|
|
|
| def test_single_attribute_infers_cqa(): |
| """文档仅含单一属性 → 主 CQA 推断为 INFERRED（带来源 evidence）。""" |
| items = [ExtractedItem(field="总杂", value="0.5", group={"attribute": "总杂"})] |
| slots = build_intent_slots( |
| goal="外推到24个月", intent=Intent.SHELF_LIFE, extracted_items=items |
| ) |
| cqa = _slot(slots, SLOT_PRIMARY_CQA) |
| assert cqa.state is SlotState.INFERRED |
| assert cqa.value == "总杂" |
| assert cqa.evidence |
|
|
|
|
| def test_multiple_attributes_are_ambiguous_not_guessed(): |
| """文档含多个属性且指令未指定 → 主 CQA 视为歧义 → MISSING（不臆断择一）。""" |
| items = [ |
| ExtractedItem(field="总杂", value="0.5", group={"attribute": "总杂"}), |
| ExtractedItem(field="含量", value="99.5", group={"attribute": "含量"}), |
| ] |
| slots = build_intent_slots( |
| goal="外推到24个月", intent=Intent.SHELF_LIFE, extracted_items=items |
| ) |
| cqa = _slot(slots, SLOT_PRIMARY_CQA) |
| assert cqa.state is SlotState.MISSING and cqa.value is None |
|
|
|
|
| def test_spec_limit_inferred_from_items(): |
| items = [ExtractedItem(field="spec_limit", value="总杂≤2.0%", source_ref="限度 总杂≤2.0%")] |
| slots = build_intent_slots( |
| goal="把总杂外推到36个月", intent=Intent.SHELF_LIFE, extracted_items=items |
| ) |
| spec = _slot(slots, SLOT_SPEC_LIMIT) |
| assert spec.state is SlotState.INFERRED and spec.value == "总杂≤2.0%" |
|
|
|
|
| def test_non_shelf_life_intent_yields_no_slots(): |
| """相容性 / 未知意图暂不强加槽位，保持向后兼容（描述性梳理见下方专项测试）。""" |
| assert build_intent_slots("相容性评估", Intent.COMPATIBILITY, []) == [] |
| assert build_intent_slots("随便问问", Intent.UNKNOWN, []) == [] |
|
|
|
|
| |
| |
| |
|
|
| def test_selective_clarification_targets_only_result_affecting_missing(): |
| """缺失的目标时间点 / 主 CQA（影响结论）应被追问；缺失规格（可降级）不强制追问。""" |
| slots = build_intent_slots("预测稳定性", Intent.SHELF_LIFE, []) |
| sheet = AnalysisTaskSheet(intent=Intent.SHELF_LIFE, intent_slots=slots) |
| names = {s.name for s in sheet.slots_needing_clarification()} |
| assert SLOT_TARGET_TIMEPOINTS in names |
| assert SLOT_PRIMARY_CQA in names |
| assert SLOT_SPEC_LIMIT not in names |
|
|
|
|
| def test_locked_slot_is_not_reclarified(): |
| """用户锁定的缺失槽位不再触发澄清（采纳真实意图，模型不得推翻用户决定）。""" |
| s = IntentSlot(name=SLOT_TARGET_TIMEPOINTS, value=None, state=SlotState.MISSING, |
| affects_result=True, locked=True) |
| assert s.needs_clarification() is False |
|
|
|
|
| def test_stated_slot_does_not_need_clarification(): |
| s = IntentSlot(name=SLOT_PRIMARY_CQA, value="总杂", state=SlotState.STATED, |
| confidence=1.0, affects_result=True) |
| assert s.needs_clarification() is False |
|
|
|
|
| |
| |
| |
|
|
| def test_intent_slot_roundtrip(): |
| s = IntentSlot(name=SLOT_TARGET_TIMEPOINTS, value=[24, 36], state=SlotState.STATED, |
| confidence=1.0, evidence="36个月", locked=True, affects_result=True) |
| assert IntentSlot.from_dict(s.to_dict()) == s |
|
|
|
|
| def test_sheet_with_slots_roundtrip(): |
| slots = build_intent_slots("把总杂外推到36个月", Intent.SHELF_LIFE, []) |
| sheet = AnalysisTaskSheet(intent=Intent.SHELF_LIFE, intent_slots=slots, |
| proposed_skill_id="stability") |
| assert AnalysisTaskSheet.from_dict(sheet.to_dict()) == sheet |
|
|
|
|
| def test_get_slot_helper(): |
| slots = build_intent_slots("外推到36个月", Intent.SHELF_LIFE, []) |
| sheet = AnalysisTaskSheet(intent=Intent.SHELF_LIFE, intent_slots=slots) |
| assert sheet.get_slot(SLOT_TARGET_TIMEPOINTS) is not None |
| assert sheet.get_slot("nonexistent") is None |
|
|
|
|
| |
| |
| |
|
|
| from kernel.understanding import SLOT_TARGET_ATTRIBUTES |
|
|
|
|
| def test_descriptive_target_attributes_stated_when_goal_names_them(): |
| """指令点名质量属性 → target_attributes 为 STATED，值为属性列表。""" |
| slots = build_intent_slots("梳理总杂和含量", Intent.DESCRIPTIVE_SUMMARY, []) |
| s = next(s for s in slots if s.name == SLOT_TARGET_ATTRIBUTES) |
| assert s.state is SlotState.STATED |
| assert "总杂" in s.value and "含量" in s.value |
| |
| assert s.affects_result is False |
|
|
|
|
| def test_descriptive_target_attributes_missing_when_unspecified(): |
| """未点名属性 → MISSING 且 value 为 None（不臆造默认），但不强制澄清。""" |
| slots = build_intent_slots("帮我梳理一下检测数据", Intent.DESCRIPTIVE_SUMMARY, []) |
| s = next(s for s in slots if s.name == SLOT_TARGET_ATTRIBUTES) |
| assert s.state is SlotState.MISSING |
| assert s.value is None |
| assert s.affects_result is False |
| |
| sheet = AnalysisTaskSheet(intent=Intent.DESCRIPTIVE_SUMMARY, intent_slots=slots) |
| assert sheet.slots_needing_clarification() == [] |
|
|
|
|
| def test_descriptive_skill_filters_to_confirmed_attributes(): |
| """描述性技能尊重已确认/锁定的 target_attributes，按属性过滤观测（用户说了算）。""" |
| from kernel.skill_base import RawInput |
| from kernel.services import Services |
| from skills.descriptive_summary.skill import SKILL |
|
|
| task_sheet = { |
| "extracted_items": [ |
| {"field": "总杂", "value": "0.30", "group": {"strength": "20μg", "attribute": "总杂"}}, |
| {"field": "含量", "value": "99.5", "group": {"strength": "20μg", "attribute": "含量"}}, |
| {"field": "水分", "value": "3.1", "group": {"strength": "20μg", "attribute": "水分"}}, |
| ], |
| "intent_slots": [ |
| {"name": SLOT_TARGET_ATTRIBUTES, "value": ["总杂"], |
| "state": "stated", "locked": True}, |
| ], |
| } |
| raw = RawInput(goal="梳理总杂", extra={"task_sheet": task_sheet}) |
| data = SKILL.extract(raw, Services()) |
| attrs = {o["attribute"] for o in data.payload["observations"]} |
| assert attrs == {"总杂"} |
| assert "filtered to user-confirmed attributes" in data.notes |
|
|
|
|
| def test_descriptive_skill_no_confirmation_keeps_all(): |
| """未确认 target_attributes → 梳理全部观测（向后兼容、行为零变化）。""" |
| from kernel.skill_base import RawInput |
| from kernel.services import Services |
| from skills.descriptive_summary.skill import SKILL |
|
|
| task_sheet = { |
| "extracted_items": [ |
| {"field": "总杂", "value": "0.30", "group": {"attribute": "总杂"}}, |
| {"field": "含量", "value": "99.5", "group": {"attribute": "含量"}}, |
| ], |
| "intent_slots": [ |
| {"name": SLOT_TARGET_ATTRIBUTES, "value": None, "state": "missing", "locked": False}, |
| ], |
| } |
| raw = RawInput(goal="梳理数据", extra={"task_sheet": task_sheet}) |
| data = SKILL.extract(raw, Services()) |
| attrs = {o["attribute"] for o in data.payload["observations"]} |
| assert attrs == {"总杂", "含量"} |
|
|
