Document failed explicit rejection experiment
This commit is contained in:
@@ -0,0 +1,347 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Isolated explicit-action-rejection Gold reliability 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,
|
||||
call_ollama,
|
||||
)
|
||||
|
||||
|
||||
GOLD_SCHEMA_VERSION = "experimental-explicit-rejection-gold-v0"
|
||||
RECOGNITION_KEYS = {
|
||||
"rejection_observation_id", "target_observation_id", "rejection_form",
|
||||
"normalized_rejected_action_text",
|
||||
}
|
||||
REJECTION_FORMS = {"explicit_action_rejection", "none"}
|
||||
FORBIDDEN_LLM_KEYS = {
|
||||
"decision", "decision_status", "outcome", "topic_status", "closed",
|
||||
"agreement", "responsible_person", "responsibility", "responsibility_scope",
|
||||
"owner", "ownership", "assignee", "requested_actor", "status",
|
||||
"explicitly_rejected", "action_item", "protocol", "protocol_category",
|
||||
"confidence", "relation", "relations", "graph", "unresolved_issue",
|
||||
}
|
||||
|
||||
PROMPT_TEMPLATE = """Recognize only whether the candidate rejection observation explicitly rejects a concrete action, option, proposal, or future course of action in this small local set of V3-style observations.
|
||||
|
||||
Answer only:
|
||||
1. Does the candidate rejection observation explicitly reject, abandon, discontinue, or rule out a concrete action, option, proposal, or future course of action?
|
||||
2. If yes, which supplied observation identifies the rejected target?
|
||||
3. What is the concise normalized meaning of the rejected action or option?
|
||||
|
||||
The candidate rejection observation is {rejection_observation_id}.
|
||||
|
||||
Use explicit_action_rejection only for an asserted rejection, abandonment, discontinuation, or non-pursuit with a concrete locally resolvable target. Personal preference is not meeting-level explicit rejection. Concern or objection without refusal is not rejection. Uncertainty is not rejection. Negative recommendation or advice is not established rejection. Deferral is not rejection. "Not yet" or temporary non-action is not abandonment. Factual negation is not action rejection. Lack of commitment is not rejection.
|
||||
|
||||
The rejected target may be self-contained in the candidate observation or introduced by one earlier supplied observation. Choose only among supplied observation IDs. If the target is ambiguous or unresolved, return rejection_form none. Preserve material scope limitations in normalized_rejected_action_text. Ignore a separate positive alternative when describing the rejected target. Keep normalized text in the observation language.
|
||||
|
||||
Do not infer responsibility, ownership, decision status, final outcome, topic closure, protocol status, confidence, relations, graphs, or unresolved issues. Do not answer whether this was finally decided, what the meeting outcome was, who is responsible, or whether the topic is closed.
|
||||
|
||||
Return exactly this JSON shape and no additional fields:
|
||||
{{
|
||||
"rejection_observation_id": "{rejection_observation_id}",
|
||||
"target_observation_id": "supplied observation ID" | null,
|
||||
"rejection_form": "explicit_action_rejection | none",
|
||||
"normalized_rejected_action_text": "concise rejected target" | null
|
||||
}}
|
||||
|
||||
For rejection_form none, target_observation_id and normalized_rejected_action_text must both be 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:
|
||||
raise DerivationValidationError("observation IDs must be unique")
|
||||
if evidence_id in seen_evidence:
|
||||
raise DerivationValidationError("evidence provenance must be unique and consistent")
|
||||
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_recognition", "expected_result"}, "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("rejection Gold cases require one or two observations")
|
||||
return cases
|
||||
|
||||
|
||||
def build_prompt(case: dict[str, Any]) -> str:
|
||||
observations = case["observations"]
|
||||
_validate_observations(observations)
|
||||
rejection_observation_id = observations[-1]["observation_id"]
|
||||
return PROMPT_TEMPLATE.format(
|
||||
rejection_observation_id=rejection_observation_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 recognition 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_recognition(data: Any, observations: list[dict[str, Any]]) -> dict[str, Any]:
|
||||
_validate_observations(observations)
|
||||
if not isinstance(data, dict):
|
||||
raise DerivationValidationError("semantic recognition must be an object")
|
||||
_reject_forbidden_keys(data)
|
||||
_exact_keys(data, RECOGNITION_KEYS, "output")
|
||||
rejection_id = _nonempty_text(data["rejection_observation_id"], "output.rejection_observation_id")
|
||||
known_ids = {item["observation_id"] for item in observations}
|
||||
if rejection_id not in known_ids:
|
||||
raise DerivationValidationError("unknown rejection observation ID")
|
||||
form = data["rejection_form"]
|
||||
if form not in REJECTION_FORMS:
|
||||
raise DerivationValidationError("rejection_form has an unsupported value")
|
||||
target_id = data["target_observation_id"]
|
||||
action_text = data["normalized_rejected_action_text"]
|
||||
if form == "none":
|
||||
if target_id is not None:
|
||||
raise DerivationValidationError("none rejection must have null target_observation_id")
|
||||
if action_text is not None:
|
||||
raise DerivationValidationError("none rejection must have null normalized_rejected_action_text")
|
||||
else:
|
||||
target_id = _nonempty_text(target_id, "output.target_observation_id")
|
||||
if target_id not in known_ids:
|
||||
raise DerivationValidationError("unknown target observation ID")
|
||||
_nonempty_text(action_text, "output.normalized_rejected_action_text")
|
||||
return data
|
||||
|
||||
|
||||
def derive_rejection(
|
||||
observations: list[dict[str, Any]], recognition: dict[str, Any]
|
||||
) -> tuple[dict[str, bool], dict[str, Any] | None]:
|
||||
validate_recognition(recognition, observations)
|
||||
by_id = {item["observation_id"]: item for item in observations}
|
||||
positions = {item["observation_id"]: index for index, item in enumerate(observations)}
|
||||
rejection = by_id.get(recognition["rejection_observation_id"])
|
||||
target_id = recognition["target_observation_id"]
|
||||
target = by_id.get(target_id) if target_id is not None else None
|
||||
gates = {
|
||||
"recognition_schema_valid": True,
|
||||
"explicit_action_rejection": recognition["rejection_form"] == "explicit_action_rejection",
|
||||
"rejection_observation_exists": rejection is not None,
|
||||
"target_observation_exists": target is not None,
|
||||
"observation_ids_valid_and_unique": len(by_id) == len(observations),
|
||||
"evidence_provenance_valid_unique_consistent": len({item["evidence_id"] for item in observations}) == len(observations),
|
||||
"target_same_or_before_rejection": target is not None and rejection is not None and positions[target["observation_id"]] <= positions[rejection["observation_id"]],
|
||||
"normalized_rejected_action_present": isinstance(recognition["normalized_rejected_action_text"], str) and bool(recognition["normalized_rejected_action_text"].strip()),
|
||||
"target_local_to_case": target_id in by_id if target_id is not None else False,
|
||||
"schema_state_consistent": recognition["rejection_form"] == "explicit_action_rejection" and target_id is not None,
|
||||
"referenced_provenance_available": target is not None and rejection is not None and bool(target["evidence_id"]) and bool(rejection["evidence_id"]),
|
||||
}
|
||||
if not all(gates.values()):
|
||||
return gates, None
|
||||
return gates, {
|
||||
"rejection_id": "rejection_1",
|
||||
"content": recognition["normalized_rejected_action_text"].strip(),
|
||||
"status": "explicitly_rejected",
|
||||
"support": {
|
||||
"target": {"observation_id": target["observation_id"], "evidence_id": target["evidence_id"]},
|
||||
"rejection": {"observation_id": rejection["observation_id"], "evidence_id": rejection["evidence_id"]},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _concepts_present(text: str | None, concepts: list[list[str]]) -> bool:
|
||||
if not concepts:
|
||||
return True
|
||||
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 _contains_forbidden_concept(text: str | None, concepts: list[str]) -> bool:
|
||||
return isinstance(text, str) and any(concept.casefold() in text.casefold() for concept in concepts)
|
||||
|
||||
|
||||
def evaluate_case(case: dict[str, Any], recognition: dict[str, Any]) -> dict[str, Any]:
|
||||
validate_recognition(recognition, case["observations"])
|
||||
gates, result = derive_rejection(case["observations"], recognition)
|
||||
expected = case["expected_recognition"]
|
||||
expected_result = case["expected_result"]
|
||||
form_correct = recognition["rejection_form"] == expected["rejection_form"]
|
||||
rejection_observation_correct = recognition["rejection_observation_id"] == expected["rejection_observation_id"]
|
||||
target_correct = recognition["target_observation_id"] == expected["target_observation_id"]
|
||||
action_correct = _concepts_present(recognition["normalized_rejected_action_text"], expected["action_concepts"])
|
||||
qualifier_preserved = _concepts_present(recognition["normalized_rejected_action_text"], expected["qualifier_concepts"])
|
||||
alternative_absorbed = _contains_forbidden_concept(recognition["normalized_rejected_action_text"], expected["forbidden_action_concepts"])
|
||||
derived = result is not None
|
||||
final_correct = derived == expected_result["explicitly_rejected"]
|
||||
if result is not None:
|
||||
final_correct = final_correct and result["status"] == "explicitly_rejected"
|
||||
semantic_correct = form_correct and rejection_observation_correct and target_correct and action_correct and qualifier_preserved and not alternative_absorbed
|
||||
automatic_failure = (derived and not expected_result["explicitly_rejected"]) or (derived and not target_correct) or (derived and not qualifier_preserved) or alternative_absorbed
|
||||
classification = "FAIL" if automatic_failure or not final_correct else ("PASS" if semantic_correct else "PARTIAL")
|
||||
return {
|
||||
"case_id": case["case_id"], "classification": classification,
|
||||
"rejection_form_correct": form_correct,
|
||||
"rejection_observation_correct": rejection_observation_correct,
|
||||
"target_observation_correct": target_correct,
|
||||
"normalized_rejected_action_correct": action_correct,
|
||||
"material_qualifiers_preserved": qualifier_preserved,
|
||||
"positive_alternative_absorbed": alternative_absorbed,
|
||||
"deterministic_gates_correct": final_correct,
|
||||
"final_result_correct": final_correct,
|
||||
"unsupported_semantic_strengthening": recognition["rejection_form"] == "explicit_action_rejection" and expected["rejection_form"] == "none",
|
||||
"normative_leakage": False,
|
||||
"gates": gates, "result": result,
|
||||
}
|
||||
|
||||
|
||||
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_gold(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 / "deterministic_gate_results.json", {})
|
||||
_write_json(case_dir / "final_derived_result.json", None)
|
||||
_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_recognition.json", parsed)
|
||||
evaluation = evaluate_case(case, parsed)
|
||||
validation = {"valid": True, "error": None}
|
||||
gates, result = derive_rejection(observations, parsed)
|
||||
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)}
|
||||
gates, result = {}, None
|
||||
_write_json(case_dir / "structural_validation.json", validation)
|
||||
_write_json(case_dir / "deterministic_gate_results.json", gates)
|
||||
_write_json(case_dir / "final_derived_result.json", result)
|
||||
_write_json(case_dir / "evaluation.json", evaluation)
|
||||
evaluations.append(evaluation)
|
||||
summary = {
|
||||
"experiment": "explicit_rejection_gold_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 explicit-rejection Gold 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_gold(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())
|
||||
Reference in New Issue
Block a user