Document controlled rejection V1 baseline
This commit is contained in:
@@ -0,0 +1,100 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Isolated controlled rejection derivation V1 experiment."""
|
||||
from __future__ import annotations
|
||||
import argparse, json, time
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from .experiment_h import DEFAULT_ENDPOINT, DEFAULT_MODEL, DerivationValidationError, OBSERVATION_KEYS, call_ollama
|
||||
from .experiment_negative_act import build_prompt as build_negative_prompt, parse_model_json, validate_classification
|
||||
|
||||
SCHEMA_VERSION="experimental-controlled-rejection-v1"
|
||||
TARGET_KEYS={"candidate_observation_id","target_observation_id","normalized_target_text"}
|
||||
FORBIDDEN={"rejection_form","negative_act_form","explicitly_rejected","status","decision","outcome","topic_status","closed","responsible_person","responsibility","owner","requested_actor","action_item","protocol_category","confidence","relation","relations","graph","unresolved_issue"}
|
||||
PROMPT="""Resolve only the concrete local action or option referred to by the candidate negative act. The candidate is {candidate}. Choose only a supplied observation ID. Use the same observation for a self-contained target. If no unique local target exists, return null for both target fields. Preserve source-language meaning and material scope such as purpose and location. Preserve continuation when non-pursuit concerns continuing something. Ignore any separate positive alternative. Do not classify the negative act and do not output rejection, status, decision, outcome, responsibility, protocol concepts, confidence, relations, or graphs. Return exactly JSON with candidate_observation_id, target_observation_id, normalized_target_text and no other fields.\nObservations:\n{observations}"""
|
||||
|
||||
def _keys(v,r,loc):
|
||||
if not isinstance(v,dict): raise DerivationValidationError(f"{loc} must be an object")
|
||||
if set(v)!=r: raise DerivationValidationError(f"{loc} keys invalid: missing={sorted(r-set(v))}, unknown={sorted(set(v)-r)}")
|
||||
def _text(v,loc):
|
||||
if not isinstance(v,str) or not v.strip(): raise DerivationValidationError(f"{loc} must be non-empty")
|
||||
return v.strip()
|
||||
def _forbidden(v,loc="output"):
|
||||
if isinstance(v,dict):
|
||||
bad=FORBIDDEN & set(v)
|
||||
if bad: raise DerivationValidationError(f"{loc} contains forbidden fields: {sorted(bad)}")
|
||||
for k,x in v.items(): _forbidden(x,f"{loc}.{k}")
|
||||
elif isinstance(v,list):
|
||||
for i,x in enumerate(v): _forbidden(x,f"{loc}[{i}]")
|
||||
def validate_observations(obs):
|
||||
if not isinstance(obs,list) or not obs: raise DerivationValidationError("observations must be non-empty")
|
||||
ids=set(); evid=set()
|
||||
for i,o in enumerate(obs):
|
||||
_keys(o,OBSERVATION_KEYS,f"observations[{i}]"); oid=_text(o["observation_id"],"observation_id"); eid=_text(o["evidence_id"],"evidence_id")
|
||||
if oid in ids or eid in evid: raise DerivationValidationError("provenance must be unique")
|
||||
ids.add(oid); evid.add(eid); _text(o["content"],"content"); _text(o["speaker"],"speaker")
|
||||
return ids
|
||||
def validate_target(data,obs):
|
||||
ids=validate_observations(obs); _forbidden(data); _keys(data,TARGET_KEYS,"target output")
|
||||
candidate=_text(data["candidate_observation_id"],"candidate_observation_id")
|
||||
if candidate not in ids: raise DerivationValidationError("unknown candidate observation")
|
||||
target=data["target_observation_id"]; normalized=data["normalized_target_text"]
|
||||
if target is None:
|
||||
if normalized is not None: raise DerivationValidationError("null target requires null text")
|
||||
else:
|
||||
target=_text(target,"target_observation_id")
|
||||
if target not in ids: raise DerivationValidationError("unknown target observation")
|
||||
_text(normalized,"normalized_target_text")
|
||||
return data
|
||||
def build_target_prompt(case):
|
||||
validate_observations(case["observations"])
|
||||
return PROMPT.format(candidate=case["expected"]["candidate_observation_id"],observations=json.dumps(case["observations"],ensure_ascii=False,indent=2))
|
||||
def derive(obs,negative,target):
|
||||
ids=validate_observations(obs); validate_classification(negative,obs); validate_target(target,obs)
|
||||
candidate=negative["observation_id"]
|
||||
if target["candidate_observation_id"]!=candidate: raise DerivationValidationError("candidate outputs disagree")
|
||||
positions={o["observation_id"]:i for i,o in enumerate(obs)}; tid=target["target_observation_id"]
|
||||
gates={"negative_act_valid":True,"eligible_explicit_non_pursuit":negative["negative_act_form"]=="explicit_non_pursuit","candidate_exists":candidate in ids,"target_valid":True,"target_present":tid is not None,"target_exists":tid in ids if tid else False,"target_not_after_candidate":positions[tid]<=positions[candidate] if tid else False,"provenance_valid_unique":True,"normalized_target_nonempty":bool(target["normalized_target_text"] and target["normalized_target_text"].strip()),"same_isolated_case":tid in ids if tid else False,"no_forbidden_fields":True}
|
||||
established=all(gates.values())
|
||||
result=None
|
||||
if established:
|
||||
byid={o["observation_id"]:o for o in obs}
|
||||
result={"rejection_id":"rejection_1","content":target["normalized_target_text"].strip(),"status":"explicitly_rejected","support":{"target":{"observation_id":tid,"evidence_id":byid[tid]["evidence_id"]},"negative_act":{"observation_id":candidate,"evidence_id":byid[candidate]["evidence_id"]}}}
|
||||
return {"gates":gates,"derived_result":result}
|
||||
def load_cases(path):
|
||||
data=json.loads(path.read_text(encoding="utf-8")); _keys(data,{"schema_version","cases"},"fixture")
|
||||
if data["schema_version"]!=SCHEMA_VERSION: raise DerivationValidationError("wrong schema version")
|
||||
return data["cases"]
|
||||
def _concepts(text,groups):
|
||||
folded=(text or "").casefold(); return all(any(x.casefold() in folded for x in g) for g in groups)
|
||||
def evaluate(case,negative,target,derivation):
|
||||
e=case["expected"]; text=target["normalized_target_text"]
|
||||
form=negative["negative_act_form"]==e["negative_act_form"]; target_ok=target["target_observation_id"]==e["target_observation_id"]
|
||||
action=_concepts(text,e["action_concepts"]); material=_concepts(text,e["material_concepts"]); forbidden=any(x.casefold() in (text or "").casefold() for x in e["forbidden_concepts"])
|
||||
final=(derivation["derived_result"] is not None)==e["explicitly_rejected"]
|
||||
label="PASS" if form and target_ok and action and material and not forbidden and final else ("PARTIAL" if form and target_ok and material and not forbidden and final else "FAIL")
|
||||
return {"case_id":case["case_id"],"classification":label,"expected_negative_act_form":e["negative_act_form"],"actual_negative_act_form":negative["negative_act_form"],"expected_target_observation_id":e["target_observation_id"],"actual_target_observation_id":target["target_observation_id"],"normalized_target_text":text,"normalized_action_correct":action,"material_scope_preserved":material,"alternative_absorbed":forbidden,"final_rejection_correct":final}
|
||||
def _write(p,v): p.write_text(json.dumps(v,ensure_ascii=False,indent=2)+"\n",encoding="utf-8")
|
||||
def run(args):
|
||||
cases=load_cases(args.cases); args.output.mkdir(parents=True,exist_ok=False); _write(args.output/"gold_cases.json",{"schema_version":SCHEMA_VERSION,"cases":cases})
|
||||
evals=[]; naf_calls=target_calls=technical_failures=structural_failures=0; start=time.perf_counter()
|
||||
for case in cases:
|
||||
d=args.output/case["case_id"].lower(); d.mkdir(); obs=case["observations"]; _write(d/"v3_style_input_observations.json",obs)
|
||||
try:
|
||||
source=case["negative_act_source"]
|
||||
if source=="live":
|
||||
np=build_negative_prompt(case); (d/"negative_act_prompt.txt").write_text(np,encoding="utf-8"); raw,nmeta=call_ollama(args.endpoint,args.model,np,args.timeout,args.num_ctx,args.num_predict); naf_calls+=1; (d/"negative_act_raw_response.txt").write_text(raw+"\n",encoding="utf-8"); negative=parse_model_json(raw)
|
||||
_write(d/"negative_act_ollama_metadata.json",nmeta); _write(d/"negative_act_source.json",{"kind":"live_call"})
|
||||
else:
|
||||
sd=args.negative_act_artifacts/source.lower(); accepted=json.loads((sd/"v3_style_input_observations.json").read_text());
|
||||
if accepted!=obs: raise DerivationValidationError(f"{source} observations do not exactly match")
|
||||
negative=json.loads((sd/"parsed_semantic_classification.json").read_text()); _write(d/"negative_act_source.json",{"kind":"accepted_artifact_reuse","case_id":source,"path":str(sd)})
|
||||
validate_classification(negative,obs); _write(d/"negative_act_classification.json",negative)
|
||||
tp=build_target_prompt(case); (d/"target_prompt.txt").write_text(tp,encoding="utf-8"); traw,tmeta=call_ollama(args.endpoint,args.model,tp,args.timeout,args.num_ctx,args.num_predict); target_calls+=1; (d/"target_raw_response.txt").write_text(traw+"\n",encoding="utf-8"); _write(d/"target_ollama_metadata.json",tmeta); target=parse_model_json(traw); _write(d/"target_recognition.json",target); validate_target(target,obs)
|
||||
derivation=derive(obs,negative,target); _write(d/"deterministic_gate_results.json",derivation["gates"]); _write(d/"final_derived_result.json",derivation["derived_result"]); ev=evaluate(case,negative,target,derivation)
|
||||
_write(d/"structural_validation.json",{"valid":True})
|
||||
except Exception as exc:
|
||||
structural_failures+=1; ev={"case_id":case["case_id"],"classification":"FAIL","technical_or_validation_failure":str(exc)}; _write(d/"structural_validation.json",{"valid":False,"error":str(exc)})
|
||||
_write(d/"evaluation.json",ev); evals.append(ev)
|
||||
summary={"experiment":"controlled_rejection_v1","model":args.model,"negative_act_llm_call_count":naf_calls,"reused_negative_act_count":len(cases)-naf_calls,"target_resolution_llm_call_count":target_calls,"technical_failed_call_count":technical_failures,"structural_validation_failure_count":structural_failures,"runtime_seconds":round(time.perf_counter()-start,3),"counts":{x:sum(e["classification"]==x for e in evals) for x in ["PASS","PARTIAL","FAIL"]},"evaluations":evals}; _write(args.output/"summary.json",summary); return summary
|
||||
def main():
|
||||
p=argparse.ArgumentParser(); p.add_argument("cases",type=Path); p.add_argument("-o","--output",type=Path,required=True); p.add_argument("--negative-act-artifacts",type=Path,default=Path("artifacts/experiments/negative_act_form_v0/20260820_qwen35_9b_single_run")); p.add_argument("--model",default=DEFAULT_MODEL); p.add_argument("--endpoint",default=DEFAULT_ENDPOINT); p.add_argument("--timeout",type=int,default=300); p.add_argument("--num-ctx",type=int,default=16384); p.add_argument("--num-predict",type=int,default=1024); args=p.parse_args(); print(json.dumps(run(args),ensure_ascii=False,indent=2)); return 0
|
||||
Reference in New Issue
Block a user