It is difficult to tell which pieces of information in a large dataset are actually reliable or well-connected.
It calculates a confidence score by analyzing how information points relate to one another across a network.
It provides a way to measure the reliability of data based on its structural connections.
It was run in the sandbox and it failed. run output shows an error/traceback — the artifact does NOT run clean.
$ python3 knowledge_confidence.py
========================================================================
KNOWLEDGE CONFIDENCE — Harmonic Mean of State Decay
========================================================================
Traceback (most recent call last):
File "/work/knowledge_confidence.py", line 207, in <module>
run_demo()
File "/work/knowledge_confidence.py", line 182, in run_demo
result = full_pipeline(DEMO_GRAPH)
^^^^^^^^^^^^^^^^^^^^^^^^^
File "/work/knowledge_confidence.py", line 141, in full_pipeline
nodes = stage_1_ingest(raw)
^^^^^^^^^^^^^^^^^^^
File "/work/knowledge_confidence.py", line 87, in stage_1_ingest
nodes.append(KnowledgeNode(
^^^^^^^^^^^^^^
TypeError: KnowledgeNode.__init__() got an unexpected keyword argument 'corroboration'No screenshot — there is nothing working to show. This is recorded as an unfinished sketch so the attempt stays visible instead of being quietly dropped.
All of it — 207 lines, one file, standard library only.
"""
Knowledge Confidence — Harmonic Mean of State Decay
Implements a Persistent-Weighted metric that scores information reliability
by tracking decay across a knowledge graph of assertions over time.
Each assertion (knowledge node) has:
- freshness: recency timestamp (0 = now, larger = older)
- corroboration: how many independent sources back it
- stability: how often recall succeeds on re-querying
The node decays exponentially when not reinforced. The harmonic mean of all
active node confidences produces the global Knowledge Confidence score,
which naturally penalises any single weak node (unlike arithmetic mean).
Run:
python knowledge_confidence.py # demo with synthetic knowledge graph
python knowledge_confidence.py --help # CLI options
"""
import argparse
import math
import statistics
import sys
import textwrap
from dataclasses import dataclass, field
from typing import Dict, List, Sequence, Tuple
# ---------------------------------------------------------------------------
# Core data structures
# ---------------------------------------------------------------------------
@dataclass
class KnowledgeNode:
label: str
freshness: float # seconds since last observation
corboration: int # independent observation count (>= 1)
stability: float # recall-success rate [0.0, 1.0]
last_decay_at: float = 0.0
half_life: float = 3600.0 # seconds before confidence halves (default 1hr)
def decay(self) -> float:
"""Exponential decay: confidence halves every half_life seconds."""
if self.half_life <= 0:
return 1.0
return 0.5 ** (self.freshness / self.half_life)
def confidence(self) -> float:
"""Per-node confidence: stability × corroboration_boost × decay_factor."""
corr_boost = math.log1p(self.corboration - 1) # log(c) grows slowly
corr_boost = corr_boost / max(1.0, math.log1p(5)) # normalise to ~[0,1]
raw = self.stability * (1.0 + corr_boost) * self.decay()
return max(0.0, min(1.0, raw))
# ─────────
# Core metric
# ─────────
def knowledge_confidence(
nodes: Sequence[KnowledgeNode],
) -> float:
"""Harmonic mean of per-node confidences after decay.
The harmonic mean is used because it is zero when *any* node is zero —
modelling the idea that certainty is only as strong as its weakest link.
"""
if not nodes:
return 0.0
confidences = [n.confidence() for n in nodes]
# Harmonic mean: n / sum(1/x). Clamp to avoid division by zero.
epsilon = 1e-9
reciprocals = [1.0 / max(c, epsilon) for c in confidences]
return len(nodes) / sum(reciprocals)
# ──────────────────────────────────────────────────────────────────────────
# Multi-stage processing pipeline (RAG-inspired extraction stages)
# ──────────────────────────────────────────────────────────────────────────
def stage_1_ingest(raw_entries: List[Dict]) -> List[KnowledgeNode]:
"""Stage 1 — Ingest: parse raw observation dicts into KnowledgeNode list."""
nodes = []
for entry in raw_entries:
nodes.append(KnowledgeNode(
label=entry.get("label", "unnamed"),
freshness=entry.get("freshness", 0.0),
corroboration=entry.get("corboration", 1),
stability=entry.get("stability", 1.0),
half_life=entry.get("half_life", 3600.0),
last_decay_at=entry.get("last_decay_at", 0.0),
))
return nodes
def stage_2_decay(nodes: List[KnowledgeNode]) -> List[float]:
"""Stage 2 — Apply per-node decay and return raw confidence scores."""
return [n.confidence() for n in nodes]
def harmonic_mean(numbers):
"""Helper function to calculate harmonic mean"""
if not numbers:
return 0.0
try:
return len(numbers) / sum(1 / max(x, 1e-9) for x in numbers)
except ZeroDivisionError:
return 0.0
def stage_3_aggregate(scores: List[float]) -> float:
"""Aggregate via harmonic mean (persistence-weighted aggregation)."""
if not scores:
return 0.0
return harmonic_mean(scores)
return harmonic_mean(scores)
try:
return statistics.harmonic_mean(max(s, 1e-9) for s in scores)
except statistics.StatisticsError:
return 0.0
def stage_4_verdict(
score: float,
thresholds: Tuple[float, float, float] = (0.8, 0.5, 0.3),
) -> str:
"""Stage 4 — map aggregate score to a verdict label."""
if score >= thresholds[0]:
return "HIGH confidence"
elif score >= thresholds[1]:
return "MODERATE confidence"
elif score >= thresholds[2]:
return "LOW confidence"
return "UNRELIABLE"
def full_pipeline(raw: List[Dict]) -> dict:
"""Run all four stages and return the complete result."""
nodes = stage_1_ingest(raw)
scores = stage_2_decay(nodes)
aggregate = stage_3_aggregate(scores)
verdict = stage_4_verdict(aggregate)
return {
"nodes": nodes,
"per_node_scores": scores,
"knowledge_confidence": aggregate,
"verdict": verdict,
}
# ─────────────────────────────
# Demo / CLI
# ─────────────────────────────
DEMO_GRAPH = [
{"label": "Earth-is-round", "freshness": 3600, "corboration": 25, "stability": 0.98},
{"label": "Paris-capital-France", "freshness": 120, "corboration": 12, "stability": 0.99},
{"label": "stock-tip-buddy", "freshness": 86400, "corboration": 1, "stability": 0.45},
{"label": "ceo-rumour", "freshness": 172800, "corboration": 2, "stability": 0.60},
{"label": "walk-schedule", "freshness": 60, "corboration": 3, "stability": 0.90},
]
def _summarise_node(n: KnowledgeNode, score: float) -> str:
return (
f" {n.label:<28}"
f" freshness={n.freshness:<8.0f}"
f" corr={n.corboration:<3}"
f" stab={n.stability:<5.2f}"
f" decay={n.decay():<6.4f}"
f" score={score:<6.4f}"
)
def run_demo() -> None:
print("=" * 72)
print(" KNOWLEDGE CONFIDENCE — Harmonic Mean of State Decay")
print("=" * 72)
print()
result = full_pipeline(DEMO_GRAPH)
print("Knowledge Graph — Per-Node Breakdown")
print("-" * 72)
for node, score in zip(result["nodes"], result["per_node_scores"]):
print(_summarise(node, score))
print()
print(f"Aggregate Knowledge Confidence = {result['knowledge_confidence']:.4f}")
print(f"Verdict = {result['verdict']}")
print()
print("Interpretation:")
print(" The harmonic mean penalises the weakest node (stale rumour).")
print(" 3 nodes at 0.95 and 1 node at 0.05 harmonic mean 0.18.")
print(" — reliable recall needs *every* anchor solid.")
if __name__ == "__main__":
parser = argparse.ArgumentParser(
description="Knowledge Confidence — Harmonic Mean of State Decay",
)
parser.add_argument(
"--demo", action="store_true", default=True,
help="Run with built-in demo knowledge graph (default)",
)
args = parser.parse_args()
run_demo()