Add request-acceptance gold experiment
This commit is contained in:
@@ -0,0 +1,178 @@
|
||||
import json
|
||||
import tempfile
|
||||
import unittest
|
||||
from copy import deepcopy
|
||||
from pathlib import Path
|
||||
|
||||
from src.meeting_lab.controlled_semantic_derivation.experiment_gold import (
|
||||
RECOGNITION_SCHEMA_VERSION,
|
||||
DerivationValidationError,
|
||||
build_prompt,
|
||||
derive_action,
|
||||
evaluate_case,
|
||||
load_gold_cases,
|
||||
validate_recognition,
|
||||
)
|
||||
|
||||
|
||||
GOLD_PATH = Path("tests/gold/request_acceptance_v0/cases.json")
|
||||
|
||||
|
||||
def recognition_for(case):
|
||||
expected = case["expected_recognition"]
|
||||
request = None
|
||||
acceptance = None
|
||||
if expected["request"]:
|
||||
request = {
|
||||
"observation_id": "obs_1",
|
||||
"is_concrete_request": True,
|
||||
"normalized_action_text": "Auswertung der Messwerte bis Dienstag",
|
||||
}
|
||||
if expected["commitment"]:
|
||||
acceptance = {
|
||||
"observation_id": "obs_2",
|
||||
"is_explicit_commitment": True,
|
||||
"same_requested_work": expected["same_work"],
|
||||
"normalized_action_text": (
|
||||
"Auswertung der Messwerte" if expected["same_work"] else "Präsentation"
|
||||
),
|
||||
}
|
||||
return {
|
||||
"schema_version": RECOGNITION_SCHEMA_VERSION,
|
||||
"request": request,
|
||||
"acceptance": acceptance,
|
||||
}
|
||||
|
||||
|
||||
class RequestAcceptanceGoldExperimentTests(unittest.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
cls.cases = load_gold_cases(GOLD_PATH)
|
||||
cls.by_id = {case["case_id"]: case for case in cls.cases}
|
||||
|
||||
def test_fixture_has_exactly_required_ten_cases(self):
|
||||
self.assertEqual(list(self.by_id), [f"RA-{number:02d}" for number in range(1, 11)])
|
||||
|
||||
def test_all_cases_use_minimal_v3_style_observations(self):
|
||||
required = {"observation_id", "evidence_id", "content", "speaker", "named_person", "addressee"}
|
||||
for case in self.cases:
|
||||
with self.subTest(case=case["case_id"]):
|
||||
self.assertGreaterEqual(len(case["observations"]), 1)
|
||||
self.assertLessEqual(len(case["observations"]), 2)
|
||||
self.assertTrue(all(set(observation) == required for observation in case["observations"]))
|
||||
|
||||
def test_positive_cases_establish_exact_action_person_due_and_provenance(self):
|
||||
for case_id in ("RA-01", "RA-02"):
|
||||
case = self.by_id[case_id]
|
||||
gates, result = derive_action(case["observations"], recognition_for(case))
|
||||
with self.subTest(case=case_id):
|
||||
self.assertTrue(all(gates.values()))
|
||||
self.assertEqual(result["content"], "Auswertung der Messwerte")
|
||||
self.assertEqual(result["requested_actor"], "Clara")
|
||||
self.assertEqual(result["responsible_person"], "Clara")
|
||||
self.assertEqual(result["due"], "Dienstag")
|
||||
self.assertEqual(result["support"]["request"], {"observation_id": "obs_1", "evidence_id": "e1"})
|
||||
self.assertEqual(result["support"]["acceptance"], {"observation_id": "obs_2", "evidence_id": "e2"})
|
||||
|
||||
def test_paraphrase_does_not_require_lexical_identity(self):
|
||||
case = self.by_id["RA-02"]
|
||||
recognition = recognition_for(case)
|
||||
recognition["acceptance"]["normalized_action_text"] = "darum kümmern und fertigstellen"
|
||||
gates, result = derive_action(case["observations"], recognition)
|
||||
self.assertTrue(gates["same_requested_work"])
|
||||
self.assertIsNotNone(result)
|
||||
|
||||
def test_acknowledgement_is_unestablished(self):
|
||||
self._assert_unestablished("RA-03", "acceptance_semantic_positive")
|
||||
|
||||
def test_tentative_response_is_unestablished(self):
|
||||
self._assert_unestablished("RA-04", "acceptance_semantic_positive")
|
||||
|
||||
def test_different_responder_is_not_personal_acceptance(self):
|
||||
self._assert_unestablished("RA-05", "acceptance_semantic_positive")
|
||||
case = self.by_id["RA-05"]
|
||||
recognition = recognition_for(self.by_id["RA-01"])
|
||||
gates, result = derive_action(case["observations"], recognition)
|
||||
self.assertFalse(gates["acceptance_speaker_matches_addressee"])
|
||||
self.assertIsNone(result)
|
||||
|
||||
def test_different_work_fails_same_work_gate(self):
|
||||
self._assert_unestablished("RA-06", "same_requested_work")
|
||||
|
||||
def test_request_without_response_is_unestablished(self):
|
||||
self._assert_unestablished("RA-07", "acceptance_observation_exists")
|
||||
|
||||
def test_collective_impersonal_and_suggestion_controls_are_unestablished(self):
|
||||
for case_id in ("RA-08", "RA-09", "RA-10"):
|
||||
with self.subTest(case=case_id):
|
||||
self._assert_unestablished(case_id, "request_semantic_positive")
|
||||
|
||||
def test_named_person_without_request_cannot_create_responsibility(self):
|
||||
case = self.by_id["RA-10"]
|
||||
self.assertEqual(case["observations"][0]["named_person"], "Dirk Textor")
|
||||
_, result = derive_action(case["observations"], recognition_for(case))
|
||||
self.assertIsNone(result)
|
||||
|
||||
def test_all_expected_recognitions_evaluate_as_pass(self):
|
||||
for case in self.cases:
|
||||
evaluation = evaluate_case(case, recognition_for(case))
|
||||
with self.subTest(case=case["case_id"]):
|
||||
self.assertEqual(evaluation["classification"], "PASS")
|
||||
|
||||
def test_equivalent_translated_normalized_action_does_not_fail_structure(self):
|
||||
case = self.by_id["RA-01"]
|
||||
recognition = recognition_for(case)
|
||||
recognition["request"]["normalized_action_text"] = "evaluate measurement values"
|
||||
evaluation = evaluate_case(case, recognition)
|
||||
self.assertEqual(evaluation["classification"], "PASS")
|
||||
self.assertEqual(evaluation["result"]["content"], "evaluate measurement values")
|
||||
|
||||
def test_forbidden_llm_fields_are_rejected_recursively(self):
|
||||
case = self.by_id["RA-01"]
|
||||
for field in (
|
||||
"responsible_person", "responsibility", "requested_actor", "status",
|
||||
"established", "action_item", "protocol_category", "confidence", "graph",
|
||||
):
|
||||
recognition = recognition_for(case)
|
||||
recognition["request"][field] = "forbidden"
|
||||
with self.subTest(field=field), self.assertRaisesRegex(DerivationValidationError, "forbidden semantic keys"):
|
||||
validate_recognition(recognition, case["observations"])
|
||||
|
||||
def test_unknown_observation_reference_is_rejected(self):
|
||||
case = self.by_id["RA-01"]
|
||||
recognition = recognition_for(case)
|
||||
recognition["acceptance"]["observation_id"] = "obs_99"
|
||||
with self.assertRaisesRegex(DerivationValidationError, "unknown observation"):
|
||||
validate_recognition(recognition, case["observations"])
|
||||
|
||||
def test_duplicate_evidence_provenance_is_rejected(self):
|
||||
fixture = json.loads(GOLD_PATH.read_text())
|
||||
fixture["cases"][0]["observations"][1]["evidence_id"] = "e1"
|
||||
with tempfile.TemporaryDirectory() as temporary:
|
||||
path = Path(temporary) / "cases.json"
|
||||
path.write_text(json.dumps(fixture), encoding="utf-8")
|
||||
with self.assertRaisesRegex(DerivationValidationError, "must be unique"):
|
||||
load_gold_cases(path)
|
||||
|
||||
def test_prompt_is_fixed_narrow_and_contains_observations_only(self):
|
||||
prompt = build_prompt(self.by_id["RA-01"]["observations"])
|
||||
self.assertIn("V3-style observations", prompt)
|
||||
self.assertNotIn("expected_result", prompt)
|
||||
self.assertNotIn("Who is responsible", prompt)
|
||||
|
||||
def test_conflicting_weekdays_fail_deadline_gate(self):
|
||||
case = deepcopy(self.by_id["RA-01"])
|
||||
case["observations"][1]["content"] = "Clara: Ja, ich übernehme die Auswertung bis Mittwoch."
|
||||
gates, result = derive_action(case["observations"], recognition_for(case))
|
||||
self.assertFalse(gates["deadline_consistent"])
|
||||
self.assertIsNone(result)
|
||||
|
||||
def _assert_unestablished(self, case_id, failed_gate):
|
||||
case = self.by_id[case_id]
|
||||
gates, result = derive_action(case["observations"], recognition_for(case))
|
||||
self.assertFalse(gates[failed_gate])
|
||||
self.assertIsNone(result)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user