Decision-tree walker over a corpus of conditional requirement notes.
A normative corpus (a building code, an engineering standard, a quant model's
assumption set) extracted under the reference-standard schema becomes a graph
of requirement notes wired by the structural backbone (component-of /
depends-on / qualifies / ...). Each conditional note carries an
applies_when block (see :class:zettelkasten.graph.Note):
{"condition": "<human-readable>", "predicate": {<param>: "<op value>"}?}
Given a caller-supplied parameter context (e.g. a design point
{"seismic_category": "D", "height_m": 32}), :func:walk traverses that graph
and classifies every decision node as applies, excluded (its predicate is
definitively false, so its whole subtree is pruned), or unresolved (it has a
condition but no machine predicate, or a referenced parameter is absent — the
caller/LLM must decide). The result is the compact, cited bundle a simulator
feeds itself instead of re-reading the source, and the step-through a dashboard
walkthrough renders.
The evaluator is a small whitelisted comparator (mirroring the data-grounding
derivation expr whitelist) — never :func:eval. Design is intentionally
decoupled from the spine internals: it reads plain :class:~zettelkasten.graph.Note
links, so it works on a schema-materialized spine and a hand-built requirement
graph alike.
parse_operator
parse_operator(expr: str) -> tuple[str, str]
Split a predicate expression into (operator, target).
Recognizes the ordering/equality operators in :data:_ORDER_OPS and the
membership forms in / not in. A bare value (no operator) is treated as
an equality test, so "D" is equivalent to "== D".
Source code in zettelkasten/decision_tree.py
| def parse_operator(expr: str) -> tuple[str, str]:
"""Split a predicate expression into ``(operator, target)``.
Recognizes the ordering/equality operators in :data:`_ORDER_OPS` and the
membership forms ``in`` / ``not in``. A bare value (no operator) is treated as
an equality test, so ``"D"`` is equivalent to ``"== D"``.
"""
raw = str(expr).strip()
low = raw.lower()
if low.startswith("not in ") or low.startswith("not-in "):
return "not in", raw[7:].strip()
if low.startswith("in ") or low.startswith("in:"):
return "in", raw[3:].strip()
for op in _ORDER_OPS:
if raw.startswith(op):
return op, raw[len(op):].strip()
return "==", raw
|
evaluate_predicate
evaluate_predicate(predicate: dict[str, Any] | None, params: dict[str, Any]) -> bool | None
Evaluate a predicate map against a parameter context.
All parameters in predicate are ANDed together. Returns:
True — every referenced parameter is present and satisfied.False — a present parameter definitively fails (the node is excluded,
even if other referenced parameters are missing).None — no predicate, or every present parameter passes but at least one
referenced parameter is absent/indeterminate (the node is unresolved).
Source code in zettelkasten/decision_tree.py
| def evaluate_predicate(
predicate: dict[str, Any] | None, params: dict[str, Any]
) -> bool | None:
"""Evaluate a predicate map against a parameter context.
All parameters in ``predicate`` are ANDed together. Returns:
* ``True`` — every referenced parameter is present and satisfied.
* ``False`` — a present parameter definitively fails (the node is excluded,
even if other referenced parameters are missing).
* ``None`` — no predicate, or every present parameter passes but at least one
referenced parameter is absent/indeterminate (the node is unresolved).
"""
if not predicate:
return None
saw_indeterminate = False
for param, expr in predicate.items():
if param not in params:
saw_indeterminate = True
continue
op, target = parse_operator(expr)
result = _compare(op, params[param], target)
if result is False:
return False
if result is None:
saw_indeterminate = True
return None if saw_indeterminate else True
|
validate_predicate
validate_predicate(predicate: Any) -> list[str]
Return a list of human-readable problems with a predicate map (empty = ok).
Structural check for authoring time: each value must be a non-empty
expression, and membership operators need a non-empty target list. Does not
require the referenced parameters to exist (those are supplied at query time).
Source code in zettelkasten/decision_tree.py
| def validate_predicate(predicate: Any) -> list[str]:
"""Return a list of human-readable problems with a predicate map (empty = ok).
Structural check for authoring time: each value must be a non-empty
expression, and membership operators need a non-empty target list. Does not
require the referenced parameters to exist (those are supplied at query time).
"""
errors: list[str] = []
if predicate is None:
return errors
if not isinstance(predicate, dict):
return ["predicate must be a mapping of {param: 'op value'}"]
for param, expr in predicate.items():
if not str(param).strip():
errors.append("predicate has an empty parameter name")
continue
op, target = parse_operator(expr)
if not str(target).strip():
errors.append(f"predicate['{param}'] has no value after '{op}'")
elif op in ("in", "not in") and not _split_targets(target):
errors.append(f"predicate['{param}'] '{op}' needs a comma-separated list")
return errors
|
walk
walk(graph: str | list[str] = '', params: dict[str, Any] | None = None, *, project: str = '', get_graph: Callable[[str], Any] | None = None, load_project: Callable[[str], dict] | None = None) -> dict[str, Any]
Traverse a requirement corpus and prune it against params.
Supply either graph (one name or a list) or project (its sources
are walked as a forest). params is the design context predicates are
evaluated against. Excluded nodes cut their subtree; a node reachable through
another kept parent survives. Node ids in the result are namespaced
"<graph>::<id>" so multiple graphs never collide.
Returns a bundle with nodes (uid -> node record incl. per-node
evidence for kept nodes and a children uid list), roots, the flat
applicable / unresolved / excluded uid lists, and a summary.
get_graph / load_project are injectable for testing; they default to
the zettelkasten server loaders.
Source code in zettelkasten/decision_tree.py
| def walk(
graph: str | list[str] = "",
params: dict[str, Any] | None = None,
*,
project: str = "",
get_graph: Callable[[str], Any] | None = None,
load_project: Callable[[str], dict] | None = None,
) -> dict[str, Any]:
"""Traverse a requirement corpus and prune it against ``params``.
Supply either ``graph`` (one name or a list) or ``project`` (its ``sources``
are walked as a forest). ``params`` is the design context predicates are
evaluated against. Excluded nodes cut their subtree; a node reachable through
another kept parent survives. Node ids in the result are namespaced
``"<graph>::<id>"`` so multiple graphs never collide.
Returns a bundle with ``nodes`` (uid -> node record incl. per-node
``evidence`` for kept nodes and a ``children`` uid list), ``roots``, the flat
``applicable`` / ``unresolved`` / ``excluded`` uid lists, and a ``summary``.
``get_graph`` / ``load_project`` are injectable for testing; they default to
the zettelkasten server loaders.
"""
params = dict(params or {})
if get_graph is None:
from zettelkasten.server import _get_graph # lazy import: avoid cycle
get_graph = _get_graph
if load_project is None:
from zettelkasten.graph_io import load_project as _load_project
load_project = _load_project
if project and not graph:
graph_names = list(load_project(project).get("sources", []) or [])
elif isinstance(graph, (list, tuple)):
graph_names = [str(g) for g in graph if g]
else:
graph_names = [str(graph)] if graph else []
out_nodes: dict[str, dict[str, Any]] = {}
roots: list[str] = []
applicable: list[str] = []
unresolved: list[str] = []
excluded: list[str] = []
def uid(gname: str, nid: str) -> str:
return f"{gname}::{nid}"
for gname in graph_names:
try:
zg = get_graph(gname)
except Exception:
continue
nodes, children, local_roots = _build_forest(zg, params)
visited: set[str] = set()
def visit(nid: str) -> None:
if nid in visited:
return
visited.add(nid)
rec = nodes[nid]
status = rec["status"]
node_uid = uid(gname, nid)
record = {
**rec,
"uid": node_uid,
"id": nid,
"children": [],
}
out_nodes[node_uid] = record
if status == STATUS_EXCLUDED:
excluded.append(node_uid)
return # cut the subtree
# Kept node: attach evidence and recurse into children.
record["evidence"] = _gather_evidence(zg, zg.notes[nid])
if status == STATUS_UNRESOLVED:
unresolved.append(node_uid)
else:
applicable.append(node_uid)
kept_children: list[str] = []
for child in sorted(children.get(nid, [])):
visit(child)
kept_children.append(uid(gname, child))
record["children"] = kept_children
for root in local_roots:
roots.append(uid(gname, root))
visit(root)
return {
"params": params,
"graphs": graph_names,
"roots": roots,
"nodes": out_nodes,
"applicable": applicable,
"unresolved": unresolved,
"excluded": excluded,
"summary": {
"decision_nodes": len(out_nodes),
"applicable": len(applicable),
"unresolved": len(unresolved),
"excluded": len(excluded),
},
}
|
next_decision
next_decision(walk_result: dict[str, Any]) -> dict[str, Any] | None
Return the next unresolved decision node from a :func:walk result.
The interactive step-through primitive: pick the shallowest unresolved node
(roots first) so a walkthrough asks the most general open question next.
Returns None when nothing is unresolved (the tree is fully decided).
Source code in zettelkasten/decision_tree.py
| def next_decision(walk_result: dict[str, Any]) -> dict[str, Any] | None:
"""Return the next unresolved decision node from a :func:`walk` result.
The interactive step-through primitive: pick the shallowest unresolved node
(roots first) so a walkthrough asks the most general open question next.
Returns ``None`` when nothing is unresolved (the tree is fully decided).
"""
unresolved = walk_result.get("unresolved") or []
if not unresolved:
return None
nodes = walk_result.get("nodes") or {}
depth = _depth_index(walk_result)
best = min(unresolved, key=lambda u: (depth.get(u, 1_000_000), u))
return nodes.get(best)
|