| |
| """ |
| example.py — Usage demonstrations for hv_quorum_inference. |
| |
| NumPy only. |
| """ |
|
|
| import numpy as np |
|
|
| from hv_quorum_inference import ( |
| QuorumConfig, quorum, majority_vote, dominance, |
| pairwise_divergence, closest_to_centroid, |
| SyntheticModel, |
| ) |
|
|
|
|
| def demo_simple_use(): |
| print("=" * 72) |
| print("Demo 1 — simplest usage") |
| print("=" * 72) |
| print() |
|
|
| |
| model = SyntheticModel(mode_collapse_rate=0.8, seed=0) |
| prompt = model.make_prompt(concentration=2.0) |
|
|
| cfg = QuorumConfig( |
| n_samples=64, |
| max_resamples=1, |
| dominance_threshold=0.70, |
| ) |
|
|
| |
| r = quorum(prompt, model, cfg) |
| print(f" collapse flag: {r.collapsed}") |
| print(f" candidate: {r.candidate}") |
| print(f" dominance: {r.final_dominance:.3f}") |
| print(f" divergence: {r.final_divergence:.3f}") |
| print(f" samples used: {r.n_samples_used}") |
| print() |
|
|
| |
| print(f" correct answer: {prompt['correct']}") |
| print() |
|
|
|
|
| def demo_clean_vs_collapsed(): |
| print("=" * 72) |
| print("Demo 2 — clean model vs collapsed model") |
| print("=" * 72) |
| print() |
|
|
| cfg = QuorumConfig(n_samples=64, max_resamples=1, |
| dominance_threshold=0.70) |
|
|
| for label, rate in [('clean (0% collapse)', 0.0), |
| ('moderate (50%)', 0.5), |
| ('severe (80%)', 0.8)]: |
| model = SyntheticModel(mode_collapse_rate=rate, seed=0) |
| flag_count = 0 |
| correct_count = 0 |
| for _ in range(50): |
| prompt = model.make_prompt(concentration=2.0) |
| r = quorum(prompt, model, cfg) |
| flag_count += int(r.collapsed) |
| correct_count += int(model.is_correct(r.candidate, prompt)) |
| print(f" {label}") |
| print(f" flagged: {flag_count / 50:.2f}") |
| print(f" correct: {correct_count / 50:.2f}") |
| print() |
|
|
|
|
| def demo_threshold_effect(): |
| print("=" * 72) |
| print("Demo 3 — how threshold affects flag rate") |
| print("=" * 72) |
| print() |
|
|
| print(f" {'threshold':>10s} {'flag@0%':>8s} {'flag@70%':>9s}") |
| print(" " + "-" * 32) |
| for th in [0.55, 0.60, 0.65, 0.70, 0.75, 0.80]: |
| cfg = QuorumConfig(n_samples=64, max_resamples=1, |
| dominance_threshold=th) |
| flags = {} |
| for rate in [0.0, 0.70]: |
| m = SyntheticModel(mode_collapse_rate=rate, seed=42) |
| count = 0 |
| for _ in range(100): |
| p = m.make_prompt(concentration=2.0) |
| r = quorum(p, m, cfg) |
| count += int(r.collapsed) |
| flags[rate] = count / 100 |
| print(f" {th:>10.2f} {flags[0.0]:>8.2f} {flags[0.70]:>9.2f}") |
| print() |
| print(" Lower threshold → more flags, more false positives.") |
| print() |
|
|
|
|
| def demo_majority_vs_quorum(): |
| print("=" * 72) |
| print("Demo 4 — majority vs quorum at 80% collapse") |
| print("=" * 72) |
| print() |
|
|
| cfg = QuorumConfig(n_samples=64, max_resamples=1, |
| dominance_threshold=0.70) |
| model = SyntheticModel(mode_collapse_rate=0.8, seed=0) |
| prompt = model.make_prompt(concentration=2.0) |
|
|
| |
| r_maj = majority_vote(prompt, model, n_samples=64) |
| print(f" Majority vote:") |
| print(f" candidate: {r_maj.candidate}") |
| print(f" correct: {model.is_correct(r_maj.candidate, prompt)}") |
| print(f" dominance: {r_maj.final_dominance:.3f}") |
| print() |
|
|
| |
| model2 = SyntheticModel(mode_collapse_rate=0.8, seed=0) |
| prompt2 = model2.make_prompt(concentration=2.0) |
| r_q = quorum(prompt2, model2, cfg) |
| print(f" Quorum:") |
| print(f" candidate: {r_q.candidate}") |
| print(f" collapsed: {r_q.collapsed}") |
| print(f" dominance: {r_q.final_dominance:.3f}") |
| print() |
| print(f" Correct answer: {prompt['correct']}") |
| print() |
|
|
|
|
| def demo_signal_components(): |
| print("=" * 72) |
| print("Demo 5 — inspect the signal values") |
| print("=" * 72) |
| print() |
|
|
| print(f" {'scenario':<24s} {'dominance':>10s} {'divergence':>12s}") |
| print(" " + "-" * 50) |
|
|
| for label, rate in [('clean', 0.0), ('moderate', 0.4), |
| ('severe', 0.8)]: |
| model = SyntheticModel(mode_collapse_rate=rate, seed=1) |
| prompt = model.make_prompt(concentration=2.0) |
| samples = model.sample(prompt, 64) |
| emb = model.embed(samples) |
| dom = dominance(samples) |
| div = pairwise_divergence(emb) |
| print(f" {label:<24s} {dom:>10.3f} {div:>12.3f}") |
| print() |
| print(" Dominance rises with collapse; divergence falls.") |
| print(" Either could be a detector. This release uses dominance.") |
| print() |
|
|
|
|
| def demo_custom_backend(): |
| print("=" * 72) |
| print("Demo 6 — using a custom backend") |
| print("=" * 72) |
| print() |
|
|
| |
| class Wrapped: |
| def __init__(self, model): |
| self.model = model |
| def sample(self, prompt, n): |
| return self.model.sample(prompt, n) |
| def embed(self, candidates): |
| return self.model.embed(candidates) |
| def is_correct(self, candidate, prompt): |
| return self.model.is_correct(candidate, prompt) |
|
|
| base = SyntheticModel(mode_collapse_rate=0.6, seed=7) |
| backend = Wrapped(base) |
| prompt = base.make_prompt(concentration=2.0) |
|
|
| cfg = QuorumConfig(n_samples=64, max_resamples=1, |
| dominance_threshold=0.70) |
| r = quorum(prompt, backend, cfg) |
| print(f" collapsed: {r.collapsed}") |
| print(f" candidate: {r.candidate}") |
| print(f" dominance: {r.final_dominance:.3f}") |
| print() |
| print(" To use a real LLM, replace the Wrapped class with a") |
| print(" HuggingFaceBackend or API wrapper. The quorum algorithm") |
| print(" only calls sample() and embed().") |
| print() |
|
|
|
|
| def main(): |
| demo_simple_use() |
| demo_clean_vs_collapsed() |
| demo_threshold_effect() |
| demo_majority_vs_quorum() |
| demo_signal_components() |
| demo_custom_backend() |
|
|
|
|
| if __name__ == "__main__": |
| main() |
