Spaces:
Running on Zero
Running on Zero
File size: 10,473 Bytes
c8fbdf1 | 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 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 | """
Echo / Perspective-Collapse Detector
Detects two failure modes in multi-perspective reasoning:
1. Echo: A perspective output is nearly identical to the raw input prompt
— the "lens" is just relabeling the question rather than analyzing it.
2. Collapse: All perspective outputs are near-identical to each other
— the multi-perspective call collapsed into a single viewpoint.
Uses token-based cosine similarity (no ML dependencies).
Thresholds:
echo_threshold: similarity(output, prompt) > 0.70 → echo risk
collapse_threshold: mean pairwise sim across perspectives > 0.80 → collapse
"""
from __future__ import annotations
import re
from collections import Counter
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple
# ── Text utilities ────────────────────────────────────────────────────────────
_STOP = frozenset(
"a an the is are was were be been being have has had do does did "
"will would could should may might shall this that these those with "
"from for of in on at to by as it its i you he she we they and or "
"but not what when where which how who".split()
)
def _tokenize(text: str) -> List[str]:
return [
w for w in re.findall(r'\b[a-zA-Z]{3,}\b', text.lower())
if w not in _STOP
]
def _term_vec(text: str) -> Counter:
return Counter(_tokenize(text))
def _cosine(a: Counter, b: Counter) -> float:
if not a or not b:
return 0.0
dot = sum(a[k] * b[k] for k in a if k in b)
mag_a = sum(v * v for v in a.values()) ** 0.5
mag_b = sum(v * v for v in b.values()) ** 0.5
if mag_a == 0 or mag_b == 0:
return 0.0
return round(dot / (mag_a * mag_b), 4)
def _novelty_ratio(output: str, prompt: str) -> float:
"""Fraction of output tokens NOT present in prompt (0=pure echo, 1=all novel)."""
out_tokens = set(_tokenize(output))
prompt_tokens = set(_tokenize(prompt))
if not out_tokens:
return 0.0
novel = out_tokens - prompt_tokens
return round(len(novel) / len(out_tokens), 4)
# ── Result types ──────────────────────────────────────────────────────────────
@dataclass
class PerspectiveEchoResult:
"""Per-perspective echo analysis."""
name: str
similarity_to_prompt: float # cosine similarity to raw prompt
novelty_ratio: float # fraction of unique tokens vs prompt
token_count: int
is_echo: bool # True if similarity_to_prompt > threshold
is_too_short: bool # True if suspiciously short output
@dataclass
class EchoCollapseResult:
"""Full echo + collapse report for a multi-perspective generation."""
echo_risk: str # 'low' | 'medium' | 'high' | 'unknown'
perspective_collapse_detected: bool
per_perspective: List[PerspectiveEchoResult] = field(default_factory=list)
mean_prompt_similarity: float = 0.0
mean_pairwise_similarity: float = 0.0
collapse_pairs: List[Tuple[str, str, float]] = field(default_factory=list)
summary: str = ""
def to_dict(self) -> dict:
return {
"echo_risk": self.echo_risk,
"perspective_collapse_detected": self.perspective_collapse_detected,
"mean_prompt_similarity": self.mean_prompt_similarity,
"mean_pairwise_similarity": self.mean_pairwise_similarity,
"collapse_pairs": [
{"a": a, "b": b, "similarity": s}
for a, b, s in self.collapse_pairs
],
"per_perspective": [
{
"name": p.name,
"similarity_to_prompt": p.similarity_to_prompt,
"novelty_ratio": p.novelty_ratio,
"token_count": p.token_count,
"is_echo": p.is_echo,
"is_too_short": p.is_too_short,
}
for p in self.per_perspective
],
"summary": self.summary,
}
# ── Detector ──────────────────────────────────────────────────────────────────
class EchoCollapseDetector:
"""Detect echo and collapse in multi-perspective outputs.
Args:
echo_threshold: similarity(output, prompt) above which output is flagged as echo.
collapse_threshold: mean pairwise similarity above which collapse is flagged.
min_token_count: outputs shorter than this are flagged as suspiciously short.
"""
def __init__(
self,
echo_threshold: float = 0.70,
collapse_threshold: float = 0.80,
min_token_count: int = 15,
):
self.echo_threshold = echo_threshold
self.collapse_threshold = collapse_threshold
self.min_token_count = min_token_count
def check(
self,
prompt: str,
perspective_outputs: Dict[str, str],
) -> EchoCollapseResult:
"""Analyze perspective_outputs for echo and collapse.
Args:
prompt: The raw user query / input prompt.
perspective_outputs: {perspective_name: output_text}
Returns:
EchoCollapseResult with echo_risk, collapse flag, and per-perspective detail.
"""
if not perspective_outputs:
return EchoCollapseResult(
echo_risk="unknown",
perspective_collapse_detected=False,
summary="No perspective outputs to analyze.",
)
prompt_vec = _term_vec(prompt)
per_perspective = []
output_vecs: Dict[str, Counter] = {}
for name, output in perspective_outputs.items():
tokens = _tokenize(output)
out_vec = _term_vec(output)
output_vecs[name] = out_vec
sim = _cosine(out_vec, prompt_vec)
novelty = _novelty_ratio(output, prompt)
n_tokens = len(tokens)
per_perspective.append(PerspectiveEchoResult(
name=name,
similarity_to_prompt=sim,
novelty_ratio=novelty,
token_count=n_tokens,
is_echo=(sim > self.echo_threshold),
is_too_short=(n_tokens < self.min_token_count),
))
# ── Echo risk ─────────────────────────────────────────────────────────
mean_prompt_sim = (
sum(p.similarity_to_prompt for p in per_perspective) / len(per_perspective)
)
n_echo = sum(1 for p in per_perspective if p.is_echo)
echo_fraction = n_echo / len(per_perspective)
if echo_fraction >= 0.6 or mean_prompt_sim > 0.80:
echo_risk = "high"
elif echo_fraction >= 0.3 or mean_prompt_sim > 0.65:
echo_risk = "medium"
else:
echo_risk = "low"
# ── Collapse detection ────────────────────────────────────────────────
names = list(output_vecs.keys())
pairwise_sims = []
collapse_pairs = []
for i in range(len(names)):
for j in range(i + 1, len(names)):
a, b = names[i], names[j]
sim = _cosine(output_vecs[a], output_vecs[b])
pairwise_sims.append(sim)
if sim > self.collapse_threshold:
collapse_pairs.append((a, b, sim))
mean_pairwise = (
sum(pairwise_sims) / len(pairwise_sims) if pairwise_sims else 0.0
)
collapse_detected = (
mean_pairwise > self.collapse_threshold
or len(collapse_pairs) / max(len(pairwise_sims), 1) > 0.5
)
# ── Summary ───────────────────────────────────────────────────────────
parts = []
if echo_risk == "high":
parts.append(
f"{n_echo}/{len(per_perspective)} perspectives are echoing the prompt "
f"(mean similarity={mean_prompt_sim:.2f})"
)
if collapse_detected:
parts.append(
f"Perspective collapse detected: mean pairwise similarity={mean_pairwise:.2f}, "
f"{len(collapse_pairs)} collapsing pairs"
)
if not parts:
parts.append(
f"No echo/collapse. Mean prompt-sim={mean_prompt_sim:.2f}, "
f"mean pairwise-sim={mean_pairwise:.2f}"
)
summary = "; ".join(parts)
return EchoCollapseResult(
echo_risk=echo_risk,
perspective_collapse_detected=collapse_detected,
per_perspective=per_perspective,
mean_prompt_similarity=round(mean_prompt_sim, 4),
mean_pairwise_similarity=round(mean_pairwise, 4),
collapse_pairs=collapse_pairs,
summary=summary,
)
def check_single(self, prompt: str, output: str, name: str = "output") -> PerspectiveEchoResult:
"""Quick echo check for a single output (no collapse analysis)."""
prompt_vec = _term_vec(prompt)
out_vec = _term_vec(output)
tokens = _tokenize(output)
sim = _cosine(out_vec, prompt_vec)
return PerspectiveEchoResult(
name=name,
similarity_to_prompt=sim,
novelty_ratio=_novelty_ratio(output, prompt),
token_count=len(tokens),
is_echo=(sim > self.echo_threshold),
is_too_short=(len(tokens) < self.min_token_count),
)
|