File size: 2,768 Bytes
d91fdcc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 | """Synthetic feature -> frozen router -> selected replacement example.
No pretrained verifier is loaded. All observations/text are invented toy inputs;
there is no benchmark data, gold label, metric evaluation or fitting.
"""
from pathlib import Path
import sys
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
import numpy as np
from factcascade.features import build_features
from factcascade.utility_model import UtilityModel
from factcascade.routing import route_batch, apply_replacement
from factcascade.quota import half_up_quota
def run_demo():
selected = np.array([0.80, 0.70, 0.95, 0.30, 0.65, 0.90, 0.55, 0.80])
full = np.array([0.90, 0.75, 0.95, 0.45, 0.70, 0.95, 0.80, 0.85])
deberta = np.array([0.70, 0.20, 0.80, 0.70, 0.60, 0.10, 0.40, 0.85])
batches = ['toy-a'] * 4 + ['toy-b'] * 4
keys = [f'toy-{i:02d}' for i in range(8)]
x = np.vstack([
build_features(
claim=f'Example Town has {i + 1} bridges.',
selected_evidence=f'The invented example lists {i + 1} bridges in Example Town.',
full_evidence=f'The invented example lists {i + 1} bridges in Example Town. This is synthetic.',
selected_ft5_score=float(selected[i]), full_ft5_score=float(full[i]),
deberta_score=float(deberta[i]), selected_chunks=1, full_chunks=1,
selected_source_count=1, source_coverage=1.0, adapter='external',
) for i in range(8)
])
model = UtilityModel.load(ROOT / 'router.json')
routed = route_batch(
selected_ft5_scores=selected, deberta_scores=deberta,
batch_ids=batches, features=x, utility_model=model, row_keys=keys,
)
# In a real application, call Bespoke only for these indices, using the
# frozen native input protocol. Here substitute invented observations.
idx = np.flatnonzero(routed.route_mask)
expensive = np.full(8, np.nan)
expensive[idx] = np.array([0.82, 0.28])
final = apply_replacement(selected, expensive, routed.route_mask)
for batch in set(batches):
member = np.array([b == batch for b in batches])
assert int(routed.route_mask[member].sum()) == half_up_quota(int(member.sum()))
assert np.array_equal(final[~routed.route_mask], (selected > 0.763)[~routed.route_mask])
assert np.array_equal(final[routed.route_mask], expensive[routed.route_mask] > 0.5)
assert x.shape == (8, 31) and len(idx) == 2
return dict(features_shape=list(x.shape), components=2, selected_rows=int(len(idx)),
selective_replacement_pass=True, unrouted_fallback_unchanged=True,
native_verifier_calls=0, fitting_calls=0, benchmark_rows=0)
if __name__ == '__main__':
import json
print(json.dumps(run_demo(), indent=2))
|