Add negative act form experiment

2026-08-20 12:19:55 +02:00
parent 97a22f3ebb
commit 0d4b426021
5 changed files with 593 additions and 0 deletions
@@ -0,0 +1,277 @@
+#!/usr/bin/env python3
+"""Isolated evidence-near Negative Act Form classification experiment."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import time
+from pathlib import Path
+from typing import Any
+
+from .experiment_h import (
+    DEFAULT_ENDPOINT,
+    DEFAULT_MODEL,
+    DerivationValidationError,
+    OBSERVATION_KEYS,
+    build_ollama_payload,
+    call_ollama,
+)
+
+
+GOLD_SCHEMA_VERSION = "experimental-negative-act-form-gold-v0"
+RECOGNITION_KEYS = {"observation_id", "negative_act_form", "normalized_action_text"}
+NEGATIVE_ACT_FORMS = {
+    "explicit_non_pursuit", "personal_preference", "recommendation",
+    "temporary_non_action", "none",
+}
+FORBIDDEN_LLM_KEYS = {
+    "rejection_form", "explicitly_rejected", "status", "decision", "outcome",
+    "topic_status", "responsible_person", "responsibility", "owner",
+    "requested_actor", "action_item", "protocol", "protocol_category",
+    "confidence", "relation", "relations", "graph", "unresolved_issue",
+}
+
+PROMPT_TEMPLATE = """Classify only the negative semantic form expressed by the candidate observation, using earlier supplied V3-style observations only as local context for pronouns or shortened references.
+
+The candidate observation is {candidate_observation_id}.
+
+Choose exactly one negative_act_form:
+- explicit_non_pursuit: explicitly states that an action, option, collaboration, or course will not be continued or pursued. This is stronger than preference, advice, or temporary delay.
+- personal_preference: the speaker states what they personally would or would not do, without establishing collective non-pursuit.
+- recommendation: the speaker advises for or against an action without establishing abandonment.
+- temporary_non_action: the action is postponed, deferred, or explicitly not done for now without abandonment.
+- none: none of those four forms is present, including mere concern, uncertainty, negative sentiment, or factual negation.
+
+Do not collapse non-pursuit into temporary non-action. Do not convert a personal conditional preference into collective non-pursuit. Do not convert advice into non-pursuit. Speaker identity does not change personal preference into collective non-pursuit.
+
+When the form is not none, return concise normalized action meaning. Resolve a pronoun only from the supplied local context. If its target is genuinely ambiguous, return none rather than guessing. When the form is none, normalized_action_text must be null. Keep normalized action text in the observation language.
+
+Do not derive or output rejection, status, decision, outcome, topic closure, responsibility, ownership, Action Item, protocol category, confidence, relations, graphs, or unresolved issues.
+
+Return exactly this JSON shape and no additional fields:
+{{
+  "observation_id": "{candidate_observation_id}",
+  "negative_act_form": "explicit_non_pursuit | personal_preference | recommendation | temporary_non_action | none",
+  "normalized_action_text": "concise action meaning" | null
+}}
+
+V3-style observations:
+{observations_json}
+"""
+
+
+def _exact_keys(value: dict[str, Any], required: set[str], location: str) -> None:
+    missing = required - value.keys()
+    unknown = value.keys() - required
+    if missing:
+        raise DerivationValidationError(f"{location} missing required keys: {sorted(missing)}")
+    if unknown:
+        raise DerivationValidationError(f"{location} has unknown keys: {sorted(unknown)}")
+
+
+def _nonempty_text(value: Any, location: str) -> str:
+    if not isinstance(value, str) or not value.strip():
+        raise DerivationValidationError(f"{location} must be a non-empty string")
+    return value.strip()
+
+
+def _validate_observations(observations: Any) -> None:
+    if not isinstance(observations, list) or not observations:
+        raise DerivationValidationError("observations must be a non-empty list")
+    seen_observations: set[str] = set()
+    seen_evidence: set[str] = set()
+    for index, observation in enumerate(observations):
+        location = f"observations[{index}]"
+        if not isinstance(observation, dict):
+            raise DerivationValidationError(f"{location} must be an object")
+        _exact_keys(observation, OBSERVATION_KEYS, location)
+        observation_id = _nonempty_text(observation["observation_id"], f"{location}.observation_id")
+        evidence_id = _nonempty_text(observation["evidence_id"], f"{location}.evidence_id")
+        if observation_id in seen_observations or evidence_id in seen_evidence:
+            raise DerivationValidationError("observation and evidence provenance must be unique")
+        seen_observations.add(observation_id)
+        seen_evidence.add(evidence_id)
+        _nonempty_text(observation["content"], f"{location}.content")
+        _nonempty_text(observation["speaker"], f"{location}.speaker")
+        for field in ("named_person", "addressee"):
+            if observation[field] is not None:
+                _nonempty_text(observation[field], f"{location}.{field}")
+
+
+def load_gold_cases(path: Path) -> list[dict[str, Any]]:
+    data = json.loads(path.read_text(encoding="utf-8-sig"))
+    if not isinstance(data, dict):
+        raise DerivationValidationError("Gold fixture must be an object")
+    _exact_keys(data, {"schema_version", "cases"}, "Gold fixture")
+    if data["schema_version"] != GOLD_SCHEMA_VERSION:
+        raise DerivationValidationError("unexpected Gold fixture schema_version")
+    cases = data["cases"]
+    if not isinstance(cases, list) or not cases:
+        raise DerivationValidationError("Gold fixture cases must be a non-empty list")
+    seen: set[str] = set()
+    for case in cases:
+        _exact_keys(case, {"case_id", "description", "observations", "expected"}, "Gold case")
+        case_id = _nonempty_text(case["case_id"], "Gold case.case_id")
+        if case_id in seen:
+            raise DerivationValidationError(f"duplicate case ID: {case_id}")
+        seen.add(case_id)
+        _validate_observations(case["observations"])
+        if len(case["observations"]) not in (1, 2):
+            raise DerivationValidationError("Negative Act cases require one or two observations")
+    return cases
+
+
+def build_prompt(case: dict[str, Any]) -> str:
+    observations = case["observations"]
+    _validate_observations(observations)
+    candidate_id = observations[-1]["observation_id"]
+    return PROMPT_TEMPLATE.format(
+        candidate_observation_id=candidate_id,
+        observations_json=json.dumps(observations, ensure_ascii=False, indent=2),
+    )
+
+
+def parse_model_json(raw_text: str) -> dict[str, Any]:
+    data = json.loads(raw_text)
+    if not isinstance(data, dict):
+        raise DerivationValidationError("semantic classification must be an object")
+    return data
+
+
+def _reject_forbidden_keys(value: Any, location: str = "output") -> None:
+    if isinstance(value, dict):
+        forbidden = FORBIDDEN_LLM_KEYS.intersection(value)
+        if forbidden:
+            raise DerivationValidationError(f"{location} contains forbidden semantic keys: {sorted(forbidden)}")
+        for key, item in value.items():
+            _reject_forbidden_keys(item, f"{location}.{key}")
+    elif isinstance(value, list):
+        for index, item in enumerate(value):
+            _reject_forbidden_keys(item, f"{location}[{index}]")
+
+
+def validate_classification(data: Any, observations: list[dict[str, Any]]) -> dict[str, Any]:
+    _validate_observations(observations)
+    if not isinstance(data, dict):
+        raise DerivationValidationError("semantic classification must be an object")
+    _reject_forbidden_keys(data)
+    _exact_keys(data, RECOGNITION_KEYS, "output")
+    observation_id = _nonempty_text(data["observation_id"], "output.observation_id")
+    if observation_id not in {item["observation_id"] for item in observations}:
+        raise DerivationValidationError("classification references unknown observation")
+    form = data["negative_act_form"]
+    if form not in NEGATIVE_ACT_FORMS:
+        raise DerivationValidationError("negative_act_form has an unsupported value")
+    action_text = data["normalized_action_text"]
+    if form == "none":
+        if action_text is not None:
+            raise DerivationValidationError("none form requires null normalized_action_text")
+    else:
+        _nonempty_text(action_text, "output.normalized_action_text")
+    return data
+
+
+def _concepts_present(text: str | None, concepts: list[list[str]]) -> bool:
+    if not concepts:
+        return text is None
+    if not isinstance(text, str):
+        return False
+    folded = text.casefold()
+    return all(any(alias.casefold() in folded for alias in alternatives) for alternatives in concepts)
+
+
+def evaluate_case(case: dict[str, Any], classification: dict[str, Any]) -> dict[str, Any]:
+    validate_classification(classification, case["observations"])
+    expected = case["expected"]
+    observation_correct = classification["observation_id"] == expected["observation_id"]
+    form_correct = classification["negative_act_form"] == expected["negative_act_form"]
+    action_correct = _concepts_present(classification["normalized_action_text"], expected["action_concepts"])
+    unsupported_strengthening = expected["negative_act_form"] == "none" and classification["negative_act_form"] != "none"
+    classification_label = "PASS" if observation_correct and form_correct and action_correct else ("PARTIAL" if observation_correct and form_correct else "FAIL")
+    return {
+        "case_id": case["case_id"], "classification": classification_label,
+        "expected_negative_act_form": expected["negative_act_form"],
+        "actual_negative_act_form": classification["negative_act_form"],
+        "observation_id_correct": observation_correct,
+        "normalized_action_meaning_correct": action_correct,
+        "unsupported_semantic_strengthening": unsupported_strengthening,
+        "normative_leakage": False,
+    }
+
+
+def _write_json(path: Path, value: Any) -> None:
+    path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+
+
+def run_experiment(args: argparse.Namespace) -> dict[str, Any]:
+    cases = load_gold_cases(args.cases)
+    args.output.mkdir(parents=True, exist_ok=False)
+    _write_json(args.output / "gold_cases.json", {"schema_version": GOLD_SCHEMA_VERSION, "cases": cases})
+    evaluations: list[dict[str, Any]] = []
+    successful_calls = 0
+    technical_failures = 0
+    started = time.perf_counter()
+    for case in cases:
+        case_dir = args.output / case["case_id"].lower()
+        case_dir.mkdir()
+        observations = case["observations"]
+        _write_json(case_dir / "v3_style_input_observations.json", observations)
+        prompt = build_prompt(case)
+        (case_dir / "prompt.txt").write_text(prompt, encoding="utf-8")
+        try:
+            raw, metadata = call_ollama(args.endpoint, args.model, prompt, args.timeout, args.num_ctx, args.num_predict)
+            successful_calls += 1
+        except Exception as exc:  # one recorded attempt; never retry
+            technical_failures += 1
+            failure = {"case_id": case["case_id"], "classification": "FAIL", "technical_failure": True, "error_type": type(exc).__name__, "error": str(exc)}
+            _write_json(case_dir / "ollama_metadata.json", {"model": args.model, "configuration": {"temperature": 0, "think": False, "num_ctx": args.num_ctx, "num_predict": args.num_predict, "retries": 0}, "technical_failure": failure})
+            _write_json(case_dir / "structural_validation.json", {"valid": False, "error": str(exc)})
+            _write_json(case_dir / "evaluation.json", failure)
+            evaluations.append(failure)
+            continue
+        (case_dir / "raw_model_response.txt").write_text(raw + "\n", encoding="utf-8")
+        _write_json(case_dir / "ollama_metadata.json", metadata)
+        try:
+            parsed = parse_model_json(raw)
+            _write_json(case_dir / "parsed_semantic_classification.json", parsed)
+            evaluation = evaluate_case(case, parsed)
+            validation = {"valid": True, "error": None}
+        except (DerivationValidationError, json.JSONDecodeError) as exc:
+            validation = {"valid": False, "error_type": type(exc).__name__, "error": str(exc)}
+            evaluation = {"case_id": case["case_id"], "classification": "FAIL", "error": str(exc), "normative_leakage": "forbidden" in str(exc)}
+        _write_json(case_dir / "structural_validation.json", validation)
+        _write_json(case_dir / "evaluation.json", evaluation)
+        evaluations.append(evaluation)
+    summary = {
+        "experiment": "negative_act_form_v0", "model": args.model,
+        "successful_llm_call_count": successful_calls,
+        "technical_failed_call_count": technical_failures,
+        "runtime_seconds": round(time.perf_counter() - started, 3),
+        "counts": {label: sum(item["classification"] == label for item in evaluations) for label in ("PASS", "PARTIAL", "FAIL")},
+        "evaluations": evaluations,
+    }
+    _write_json(args.output / "summary.json", summary)
+    return summary
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Run isolated Negative Act Form experiment")
+    parser.add_argument("cases", type=Path)
+    parser.add_argument("-o", "--output", type=Path, required=True)
+    parser.add_argument("--model", default=DEFAULT_MODEL)
+    parser.add_argument("--endpoint", default=DEFAULT_ENDPOINT)
+    parser.add_argument("--timeout", type=int, default=300)
+    parser.add_argument("--num-ctx", type=int, default=16384)
+    parser.add_argument("--num-predict", type=int, default=1024)
+    return parser.parse_args()
+
+
+def main() -> int:
+    summary = run_experiment(parse_args())
+    print(json.dumps(summary, ensure_ascii=False, indent=2))
+    return 0 if summary["counts"]["FAIL"] == 0 and summary["technical_failed_call_count"] == 0 else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())