Skip to content

zettelkasten.synapse.eval

zettelkasten.synapse.eval

CLI-first eval harness for synapse.

Two objective checks that replace a human approval gate with measurable signal:

  1. Connection impact on retrieval — for a set of questions, compare cross-store search WITHOUT vs WITH overlay expansion. Reports what the typed connections actually add (how many extra nodes surface, from which store, via which relation). If expansion adds nothing, the connections aren't earning their keep.

  2. Typed-edge precision spot-check — sample edges from the overlay for review, with the relation, confidence, rationale, and (optionally) resolved titles, plus the relation/confidence distribution. This is the objective substitute for eyeballing every edge.

Run it::

python -m zettelkasten.synapse.eval --sample 20
python -m zettelkasten.synapse.eval --questions-file q.txt --projects factors,execution

Everything is read-only. The retrieval-impact functions accept an injected search_fn so the harness (and its tests) run without the MCP server / an LLM.

retrieval_impact 

retrieval_impact(questions: list[str], zk_get_graph: GetGraph, projects: list[str] | None = None, top_k: int = 15, expand_min_conf: float = 0.6, search_fn: Callable[..., dict[str, Any]] | None = None) -> dict[str, Any]

Compare cross-store search without vs with overlay expansion, per question.

Source code in zettelkasten/synapse/eval.py 
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
def retrieval_impact(
    questions: list[str],
    zk_get_graph: GetGraph,
    projects: list[str] | None = None,
    top_k: int = 15,
    expand_min_conf: float = 0.6,
    search_fn: Callable[..., dict[str, Any]] | None = None,
) -> dict[str, Any]:
    """Compare cross-store search without vs with overlay expansion, per question."""
    search_fn = search_fn or retrieval.search_across_stores
    per_question: list[dict[str, Any]] = []
    total_added = 0
    for q in questions:
        base = search_fn(q, zk_get_graph, projects=projects, top_k=top_k)
        expanded = search_fn(
            q, zk_get_graph, projects=projects, top_k=top_k, expand=True, expand_min_conf=expand_min_conf
        )
        added = [r for r in expanded.get("results", []) if r.get("expanded_from")]
        total_added += len(added)
        per_question.append({
            "question": q,
            "base_count": len(base.get("results", [])),
            "added_by_expansion": len(added),
            "added": [
                {"id": r["id"], "store": r.get("store"), "relation": r.get("relation"),
                 "expanded_from": r.get("expanded_from")}
                for r in added
            ],
        })
    return {
        "question_count": len(questions),
        "total_added_by_expansion": total_added,
        "mean_added_per_question": round(total_added / len(questions), 3) if questions else 0.0,
        "per_question": per_question,
    }

edge_precision_sample 

edge_precision_sample(overlay: dict[str, Any] | None = None, sample_size: int = 20, seed: int = 0, zk_get_graph: GetGraph | None = None, projects: list[str] | None = None) -> dict[str, Any]

Sample overlay edges for a precision spot-check + report distributions.

Source code in zettelkasten/synapse/eval.py 
 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
def edge_precision_sample(
    overlay: dict[str, Any] | None = None,
    sample_size: int = 20,
    seed: int = 0,
    zk_get_graph: GetGraph | None = None,
    projects: list[str] | None = None,
) -> dict[str, Any]:
    """Sample overlay edges for a precision spot-check + report distributions."""
    if overlay is None:
        overlay = _overlay.load_overlay()
    edges = list(overlay.get("edges", []))

    relation_dist: dict[str, int] = {}
    conf_dist: dict[str, int] = {}
    conf_sum = 0.0
    for e in edges:
        relation_dist[e.get("relation")] = relation_dist.get(e.get("relation"), 0) + 1
        conf = float(e.get("confidence", 0.0))
        conf_sum += conf
        conf_dist[_confidence_bucket(conf)] = conf_dist.get(_confidence_bucket(conf), 0) + 1

    rng = random.Random(seed)
    sampled = edges if len(edges) <= sample_size else rng.sample(edges, sample_size)

    titles: dict[str, str] = {}
    if zk_get_graph is not None and sampled:
        _sources, get_graph = retrieval.resolve_scope(zk_get_graph, projects=projects)
        for e in sampled:
            try:
                mt = get_graph(MEMORY_SOURCE).notes.get(e["memory_id"])
                titles[e["memory_id"]] = getattr(mt, "title", "") if mt else ""
                zt = get_graph(e.get("zk_source", "")).notes.get(e["zk_id"])
                titles[e["zk_id"]] = getattr(zt, "title", "") if zt else ""
            except Exception:
                continue

    sample_rows = [
        {
            "memory_id": e["memory_id"],
            "memory_title": titles.get(e["memory_id"], ""),
            "zk_id": e["zk_id"],
            "zk_title": titles.get(e["zk_id"], ""),
            "relation": e.get("relation"),
            "confidence": e.get("confidence"),
            "rationale": e.get("rationale", ""),
        }
        for e in sampled
    ]

    return {
        "edge_count": len(edges),
        "relation_distribution": relation_dist,
        "confidence_distribution": conf_dist,
        "mean_confidence": round(conf_sum / len(edges), 4) if edges else 0.0,
        "sample_size": len(sample_rows),
        "sample": sample_rows,
    }