Dateien nach "app/core" hochladen
All checks were successful
Deploy mindnet to llm-node / deploy (push) Successful in 4s

This commit is contained in:
Lars 2025-11-17 21:42:26 +01:00
parent 7e8d6c1954
commit fb2de78d62

View File

@ -1,286 +1,423 @@
# app/core/derive_edges.py #!/usr/bin/env python3
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
""" """
Mindnet V2 Edge derivation Modul: app/core/derive_edges.py
Zweck:
- Bewahrt bestehende Edgelogik (belongs_to, prev/next, references, backlink)
- Ergänzt typenbasierte Default-Kanten (edge_defaults aus config/types.yaml)
- Unterstützt "typed inline relations" ([[rel:KIND | Target]] / [[rel:KIND Target]])
- Unterstützt Obsidian-Callouts (> [!edge] KIND: [[Target]] [[Target2]] ...)
Features preserved & extended: Kompatibilität:
- Structure edges: belongs_to, next, prev (deterministic order by index/ord/chunk_id) - build_edges_for_note(...) Signatur unverändert
- Explicit references from wikilinks [[Title]] -> kind="references", rule_id="explicit:wikilink", confidence=1.0 - rule_id Werte exakt wie zuvor erwartet (ohne Versionssuffix):
- Inline relations: * structure:belongs_to
A) [[rel:<kind> <Target>]] -> kind from link, rule_id="inline:rel", confidence=0.95 * structure:order
B) rel: <kind> [[Target1]] [[Target2]] ... -> supports multiple wikilinks on a single line * explicit:wikilink
- Callout relations: * inline:rel
> [!edge] <kind>: [[Target1]] [[Target2]] -> rule_id="callout:edge", confidence=0.9 * callout:edge
- Type-based defaults (from types.yaml): * edge_defaults:<type>:<relation>
For each note type's `edge_defaults`, derive additional edges for every explicit/inline/callout target. * derived:backlink
rule_id="edge_defaults:<note_type>:<kind>", confidence=0.7
Symmetric kinds ("related_to","similar_to") also add reversed edges.
- De-duplication across all sources by (kind, scope, source_id, target_id, rule_id)
- Backward-compatible function signature
Return: list[dict] with at least:
kind, scope, source_id, target_id, note_id, chunk_id, rule_id, confidence
""" """
from __future__ import annotations from __future__ import annotations
import os
import re import re
from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple from typing import Iterable, List, Optional, Tuple, Set, Dict
# Kanten, die symmetrisch interpretiert werden (Rückkante wird erzeugt) try:
SYMMETRIC_KINDS = {"related_to", "similar_to"} import yaml # optional, nur für types.yaml
except Exception: # pragma: no cover
yaml = None
# ----------------------------------------------------------------------------- # --------------------------------------------------------------------------- #
# Utilities # Utilities
# ----------------------------------------------------------------------------- # --------------------------------------------------------------------------- #
def _get(d: Dict[str, Any], *keys: str, default: Any = None) -> Any: def _get(d: dict, *keys, default=None):
for k in keys: for k in keys:
if isinstance(d, dict) and k in d: if k in d and d[k] is not None:
return d[k] return d[k]
return default return default
def _safe_int(x: Any, default: int = 10**9) -> int: def _chunk_text_for_refs(chunk: dict) -> str:
try: # bevorzugt 'window' → dann 'text' → 'content' → 'raw'
return int(x) return (
except Exception: _get(chunk, "window")
return default or _get(chunk, "text")
or _get(chunk, "content")
or _get(chunk, "raw")
or ""
)
def _sort_chunks(chs: Sequence[Dict[str, Any]]) -> List[Dict[str, Any]]: def _dedupe_seq(seq: Iterable[str]) -> List[str]:
"""Sort chunks primarily by index, then ord, then chunk_id to guarantee stable next/prev.""" seen: Set[str] = set()
return sorted(chs, key=lambda ch: ( out: List[str] = []
_get(ch, "index") is None, _safe_int(_get(ch, "index")), for s in seq:
_get(ch, "ord") is None, _safe_int(_get(ch, "ord")), if s not in seen:
str(_get(ch, "chunk_id") or _get(ch, "id") or "") seen.add(s)
)) out.append(s)
return out
def _make_edge(kind: str, def _edge(kind: str, scope: str, source_id: str, target_id: str, note_id: str, extra: Optional[dict] = None) -> dict:
source_id: str, pl = {
target_id: str,
note_id: str,
chunk_id: str,
rule_id: str,
confidence: float,
scope: str = "chunk") -> Dict[str, Any]:
return {
"kind": kind, "kind": kind,
"scope": scope, "relation": kind, # Alias (v2)
"scope": scope, # "chunk" | "note"
"source_id": source_id, "source_id": source_id,
"target_id": target_id, "target_id": target_id,
"note_id": note_id, "note_id": note_id, # Träger-Note der Kante
"chunk_id": chunk_id,
"rule_id": rule_id,
"confidence": float(confidence),
} }
if extra:
pl.update(extra)
return pl
def _dedupe_key(e: Dict[str, Any]) -> Tuple[Any, ...]: def _mk_edge_id(kind: str, s: str, t: str, scope: str, rule_id: Optional[str] = None) -> str:
return ( base = f"{kind}:{s}->{t}#{scope}"
e.get("kind"), if rule_id:
e.get("scope"), base += f"|{rule_id}"
e.get("source_id"), try:
e.get("target_id"), import hashlib
e.get("rule_id"), return hashlib.blake2s(base.encode("utf-8"), digest_size=12).hexdigest()
) except Exception: # pragma: no cover
return base
# ----------------------------------------------------------------------------- # --------------------------------------------------------------------------- #
# Parsing helpers # Typen-Registry (types.yaml)
# ----------------------------------------------------------------------------- # --------------------------------------------------------------------------- #
RE_WIKILINK = re.compile(r"\[\[([^\]]+?)\]\]") def _env(n: str, default: Optional[str] = None) -> str:
RE_INLINE_BRACKET = re.compile( v = os.getenv(n)
r"\[\[\s*rel\s*:\s*(?P<kind>[a-z_][a-z0-9_]*)\s+(?P<target>[^\]]+?)\s*\]\]", return v if v is not None else (default or "")
flags=re.IGNORECASE,
)
RE_INLINE_PREFIX = re.compile(
r"(?m)\brel\s*:\s*(?P<kind>[a-z_][a-z0-9_]*)\s+(?P<links>(?:\[\[[^\]]+\]\]\s*)+)",
flags=re.IGNORECASE,
)
RE_CALLOUT = re.compile(
r"(?m)^\s*>\s*\[!edge\]\s*(?P<kind>[a-z_][a-z0-9_]*)\s*:\s*(?P<body>.+)$",
flags=re.IGNORECASE,
)
def _extract_wikilinks(text: str) -> List[str]: def _load_types_registry() -> dict:
"""All [[Title]] except those starting with 'rel:' (reserved for inline relations).""" """Lädt die YAML-Registry aus MINDNET_TYPES_FILE oder ./config/types.yaml"""
targets: List[str] = [] p = _env("MINDNET_TYPES_FILE", "./config/types.yaml")
for m in RE_WIKILINK.finditer(text or ""): if not os.path.isfile(p) or yaml is None:
label = m.group(1).strip() return {}
if label.lower().startswith("rel:"): try:
# handled by explicit inline relation parser with open(p, "r", encoding="utf-8") as f:
continue data = yaml.safe_load(f) or {}
if label: return data
targets.append(label) except Exception:
return targets return {}
def _extract_inline_relations(text: str) -> List[Tuple[str, str]]: def _get_types_map(reg: dict) -> dict:
"""Find inline relations from: if isinstance(reg, dict) and isinstance(reg.get("types"), dict):
A) [[rel:kind Target]] return reg["types"]
B) rel: kind [[Target1]] [[Target2]] (supports multiple targets on one line) return reg if isinstance(reg, dict) else {}
def _edge_defaults_for(note_type: Optional[str], reg: dict) -> List[str]:
""" """
out: List[Tuple[str, str]] = [] Liefert die edge_defaults-Liste für den gegebenen Notiztyp.
Fallback-Reihenfolge:
1) reg['types'][note_type]['edge_defaults']
2) reg['defaults']['edge_defaults'] (oder 'default'/'global')
3) []
"""
types_map = _get_types_map(reg)
if note_type and isinstance(types_map, dict):
t = types_map.get(note_type)
if isinstance(t, dict) and isinstance(t.get("edge_defaults"), list):
return [str(x) for x in t["edge_defaults"] if isinstance(x, str)]
for key in ("defaults", "default", "global"):
v = reg.get(key)
if isinstance(v, dict) and isinstance(v.get("edge_defaults"), list):
return [str(x) for x in v["edge_defaults"] if isinstance(x, str)]
return []
# A) Bracket form # --------------------------------------------------------------------------- #
for m in RE_INLINE_BRACKET.finditer(text or ""): # Parser für Links / Relationen
kind = m.group("kind").strip().lower() # --------------------------------------------------------------------------- #
target = m.group("target").strip()
if kind and target:
out.append((kind, target))
# B) Prefix form with multiple wikilinks # Normale Wikilinks (Fallback)
for m in RE_INLINE_PREFIX.finditer(text or ""): _WIKILINK_RE = re.compile(r"\[\[(?:[^\|\]]+\|)?([a-zA-Z0-9_\-#:. ]+)\]\]")
kind = m.group("kind").strip().lower()
links = m.group("links")
for lm in RE_WIKILINK.finditer(links or ""):
target = lm.group(1).strip()
if target:
out.append((kind, target))
return out # Getypte Inline-Relationen:
# [[rel:depends_on | Target]]
# [[rel:related_to Target]]
_REL_PIPE = re.compile(r"\[\[\s*rel:(?P<kind>[a-z_]+)\s*\|\s*(?P<target>[^\]]+?)\s*\]\]", re.IGNORECASE)
_REL_SPACE = re.compile(r"\[\[\s*rel:(?P<kind>[a-z_]+)\s+(?P<target>[^\]]+?)\s*\]\]", re.IGNORECASE)
def _extract_callout_relations(text: str) -> List[Tuple[str, str]]: def _extract_typed_relations(text: str) -> Tuple[List[Tuple[str,str]], str]:
"""Find callout > [!edge] kind: [[Target1]] [[Target2]] | Target, Target2""" """
out: List[Tuple[str, str]] = [] Gibt Liste (kind, target) zurück und den Text mit entfernten getypten Relation-Links,
for m in RE_CALLOUT.finditer(text or ""): damit die generische Wikilink-Erkennung sie nicht doppelt zählt.
kind = (m.group("kind") or "").strip().lower() """
body = m.group("body") or "" pairs: List[Tuple[str,str]] = []
# Prefer wikilinks; fallback: split by comma def _collect(m):
wl = [mm.group(1).strip() for mm in RE_WIKILINK.finditer(body)] k = (m.group("kind") or "").strip().lower()
if wl: t = (m.group("target") or "").strip()
out.extend((kind, t) for t in wl if t) if k and t:
pairs.append((k, t))
return "" # Link entfernen
text = _REL_PIPE.sub(_collect, text)
text = _REL_SPACE.sub(_collect, text)
return pairs, text
# Obsidian Callout Parser
_CALLOUT_START = re.compile(r"^\s*>\s*\[!edge\]\s*(.*)$", re.IGNORECASE)
_REL_LINE = re.compile(r"^(?P<kind>[a-z_]+)\s*:\s*(?P<targets>.+?)\s*$", re.IGNORECASE)
_WIKILINKS_IN_LINE = re.compile(r"\[\[([^\]]+)\]\]")
def _extract_callout_relations(text: str) -> Tuple[List[Tuple[str,str]], str]:
"""
Findet [!edge]-Callouts und extrahiert (kind, target). Entfernt den gesamten
Callout-Block aus dem Text (damit Wikilinks daraus nicht zusätzlich als
"references" gezählt werden).
"""
if not text:
return [], text
lines = text.splitlines()
out_pairs: List[Tuple[str,str]] = []
keep_lines: List[str] = []
i = 0
while i < len(lines):
m = _CALLOUT_START.match(lines[i])
if not m:
keep_lines.append(lines[i])
i += 1
continue
block_lines: List[str] = []
first_rest = m.group(1) or ""
if first_rest.strip():
block_lines.append(first_rest)
i += 1
while i < len(lines) and lines[i].lstrip().startswith('>'):
block_lines.append(lines[i].lstrip()[1:].lstrip())
i += 1
for bl in block_lines:
mrel = _REL_LINE.match(bl)
if not mrel:
continue
kind = (mrel.group("kind") or "").strip().lower()
targets = mrel.group("targets") or ""
found = _WIKILINKS_IN_LINE.findall(targets)
if found:
for t in found:
t = t.strip()
if t:
out_pairs.append((kind, t))
else: else:
for raw in body.split(","): for raw in re.split(r"[,;]", targets):
t = raw.strip() t = raw.strip()
if t: if t:
out.append((kind, t)) out_pairs.append((kind, t))
return out
# ----------------------------------------------------------------------------- # Callout wird NICHT in keep_lines übernommen
# Edge builder
# -----------------------------------------------------------------------------
def _edge_defaults_for_type(types_cfg: Optional[Dict[str, Any]], note_type: Optional[str]) -> List[str]:
"""Read defaults from types.yaml structure: {'types': {<type>: {'edge_defaults': [..]}}}"""
if not types_cfg or not note_type:
return []
t = types_cfg.get("types", {}).get(note_type, {})
vals = t.get("edge_defaults", []) or []
return [str(v) for v in vals if isinstance(v, (str,))]
def _append_with_dedupe(edges: List[Dict[str, Any]], new_edges: Iterable[Dict[str, Any]]) -> None:
seen = { _dedupe_key(e) for e in edges }
for e in new_edges:
k = _dedupe_key(e)
if k in seen:
continue continue
edges.append(e)
seen.add(k)
def _structure_edges(note_id: str, title: str, chunks: Sequence[Dict[str, Any]]) -> List[Dict[str, Any]]: remainder = "\n".join(keep_lines)
edges: List[Dict[str, Any]] = [] return out_pairs, remainder
ordered = _sort_chunks(chunks)
# belongs_to def _extract_wikilinks(text: str) -> List[str]:
for ch in ordered: ids: List[str] = []
for m in _WIKILINK_RE.finditer(text or ""):
ids.append(m.group(1).strip())
return ids
# --------------------------------------------------------------------------- #
# Hauptfunktion
# --------------------------------------------------------------------------- #
def build_edges_for_note(
note_id: str,
chunks: List[dict],
note_level_references: Optional[List[str]] = None,
include_note_scope_refs: bool = False,
) -> List[dict]:
"""
Erzeugt Kanten für eine Note.
- belongs_to: für jeden Chunk (chunk -> note)
- next / prev: zwischen aufeinanderfolgenden Chunks
- references: pro Chunk aus window/text (via Wikilinks)
- typed inline relations: [[rel:KIND | Target]] oder [[rel:KIND Target]]
- Obsidian Callouts: > [!edge] KIND: [[Target]] [[Target2]]
- optional note-scope references/backlinks: dedupliziert über alle Chunk-Funde + note_level_references
- typenbasierte Default-Kanten (edge_defaults) je gefundener Referenz
"""
edges: List[dict] = []
# Note-Typ (aus erstem Chunk erwartet)
note_type = None
if chunks:
note_type = _get(chunks[0], "type")
# 1) belongs_to
for ch in chunks:
cid = _get(ch, "chunk_id", "id") cid = _get(ch, "chunk_id", "id")
if not cid: if not cid:
continue continue
edges.append(_make_edge("belongs_to", cid, note_id, note_id, cid, "structure:belongs_to", 1.0)) edges.append(_edge("belongs_to", "chunk", cid, note_id, note_id, {
# next/prev "chunk_id": cid,
for i in range(len(ordered) - 1): "edge_id": _mk_edge_id("belongs_to", cid, note_id, "chunk", "structure:belongs_to"),
a, b = ordered[i], ordered[i + 1] "provenance": "rule",
a_id = _get(a, "chunk_id", "id"); b_id = _get(b, "chunk_id", "id") "rule_id": "structure:belongs_to",
"confidence": 1.0,
}))
# 2) next / prev
for i in range(len(chunks) - 1):
a, b = chunks[i], chunks[i + 1]
a_id = _get(a, "chunk_id", "id")
b_id = _get(b, "chunk_id", "id")
if not a_id or not b_id: if not a_id or not b_id:
continue continue
edges.append(_make_edge("next", a_id, b_id, note_id, a_id, "structure:next", 1.0)) edges.append(_edge("next", "chunk", a_id, b_id, note_id, {
edges.append(_make_edge("prev", b_id, a_id, note_id, b_id, "structure:prev", 1.0)) "chunk_id": a_id,
return edges "edge_id": _mk_edge_id("next", a_id, b_id, "chunk", "structure:order"),
"provenance": "rule",
"rule_id": "structure:order",
"confidence": 0.95,
}))
edges.append(_edge("prev", "chunk", b_id, a_id, note_id, {
"chunk_id": b_id,
"edge_id": _mk_edge_id("prev", b_id, a_id, "chunk", "structure:order"),
"provenance": "rule",
"rule_id": "structure:order",
"confidence": 0.95,
}))
def _explicit_and_inline_edges(note_id: str, # 3) references + typed inline + callouts + defaults (chunk-scope)
title: str, reg = _load_types_registry()
note_type: Optional[str], defaults = _edge_defaults_for(note_type, reg)
chunks: Sequence[Dict[str, Any]], refs_all: List[str] = []
types_cfg: Optional[Dict[str, Any]]) -> Tuple[List[Dict[str, Any]], List[str]]:
"""Return (edges, referenced_targets)"""
edges: List[Dict[str, Any]] = []
referenced: List[str] = []
defaults_kinds = set(_edge_defaults_for_type(types_cfg, note_type)) for ch in chunks:
for ch in _sort_chunks(chunks):
cid = _get(ch, "chunk_id", "id")
text = _get(ch, "text", "window", default="") or ""
# --- Explicit wikilinks -> references
for target in _extract_wikilinks(text):
referenced.append(target)
edges.append(_make_edge("references", cid, target, note_id, cid, "explicit:wikilink", 1.0))
# --- Inline relations (both forms)
for kind, target in _extract_inline_relations(text):
referenced.append(target)
# forward
edges.append(_make_edge(kind, cid, target, note_id, cid, "inline:rel", 0.95))
# symmetric reverse
if kind in SYMMETRIC_KINDS:
edges.append(_make_edge(kind, target, cid, note_id, cid, "inline:rel", 0.95))
# --- Callout relations
for kind, target in _extract_callout_relations(text):
referenced.append(target)
edges.append(_make_edge(kind, cid, target, note_id, cid, "callout:edge", 0.90))
if kind in SYMMETRIC_KINDS:
edges.append(_make_edge(kind, target, cid, note_id, cid, "callout:edge", 0.90))
# --- Type-based defaults: apply to the union of referenced targets
if defaults_kinds and referenced:
uniq_targets = list(dict.fromkeys(referenced)) # preserve order, drop dups
for ch in _sort_chunks(chunks):
cid = _get(ch, "chunk_id", "id") cid = _get(ch, "chunk_id", "id")
if not cid: if not cid:
continue continue
for kind in defaults_kinds: raw = _chunk_text_for_refs(ch)
for target in uniq_targets:
edges.append(_make_edge(kind, cid, target, note_id, cid, f"edge_defaults:{note_type}:{kind}", 0.70))
if kind in SYMMETRIC_KINDS:
edges.append(_make_edge(kind, target, cid, note_id, cid, f"edge_defaults:{note_type}:{kind}", 0.70))
return edges, referenced # 3a) typed inline relations
typed, remainder = _extract_typed_relations(raw)
for kind, target in typed:
kind = kind.strip().lower()
if not kind or not target:
continue
edges.append(_edge(kind, "chunk", cid, target, note_id, {
"chunk_id": cid,
"edge_id": _mk_edge_id(kind, cid, target, "chunk", "inline:rel"),
"provenance": "explicit",
"rule_id": "inline:rel",
"confidence": 0.95,
}))
if kind in {"related_to", "similar_to"}:
edges.append(_edge(kind, "chunk", target, cid, note_id, {
"chunk_id": cid,
"edge_id": _mk_edge_id(kind, target, cid, "chunk", "inline:rel"),
"provenance": "explicit",
"rule_id": "inline:rel",
"confidence": 0.95,
}))
# ----------------------------------------------------------------------------- # 3b) callouts
# Public API (backward-compatible) call_pairs, remainder2 = _extract_callout_relations(remainder)
# ----------------------------------------------------------------------------- for kind, target in call_pairs:
k = (kind or "").strip().lower()
if not k or not target:
continue
edges.append(_edge(k, "chunk", cid, target, note_id, {
"chunk_id": cid,
"edge_id": _mk_edge_id(k, cid, target, "chunk", "callout:edge"),
"provenance": "explicit",
"rule_id": "callout:edge",
"confidence": 0.95,
}))
if k in {"related_to", "similar_to"}:
edges.append(_edge(k, "chunk", target, cid, note_id, {
"chunk_id": cid,
"edge_id": _mk_edge_id(k, target, cid, "chunk", "callout:edge"),
"provenance": "explicit",
"rule_id": "callout:edge",
"confidence": 0.95,
}))
def build_edges_for_note(*args, **kwargs) -> List[Dict[str, Any]]: # 3c) generische Wikilinks → references (+ defaults je Ref)
""" refs = _extract_wikilinks(remainder2)
Backward-compatible entry point. for r in refs:
edges.append(_edge("references", "chunk", cid, r, note_id, {
"chunk_id": cid,
"ref_text": r,
"edge_id": _mk_edge_id("references", cid, r, "chunk", "explicit:wikilink"),
"provenance": "explicit",
"rule_id": "explicit:wikilink",
"confidence": 1.0,
}))
for rel in defaults:
if rel == "references":
continue
edges.append(_edge(rel, "chunk", cid, r, note_id, {
"chunk_id": cid,
"edge_id": _mk_edge_id(rel, cid, r, "chunk", f"edge_defaults:{note_type}:{rel}"),
"provenance": "rule",
"rule_id": f"edge_defaults:{note_type}:{rel}",
"confidence": 0.7,
}))
if rel in {"related_to", "similar_to"}:
edges.append(_edge(rel, "chunk", r, cid, note_id, {
"chunk_id": cid,
"edge_id": _mk_edge_id(rel, r, cid, "chunk", f"edge_defaults:{note_type}:{rel}"),
"provenance": "rule",
"rule_id": f"edge_defaults:{note_type}:{rel}",
"confidence": 0.7,
}))
refs_all.extend(refs)
Supported call styles: # 4) optional note-scope refs/backlinks (+ defaults)
1) build_edges_for_note(note_id, title, note_type, chunks, types_cfg) if include_note_scope_refs:
2) build_edges_for_note(note_payload=dict, chunks=[...], types_cfg=dict) refs_note = list(refs_all or [])
where note_payload contains: note_id/id, title, type if note_level_references:
""" refs_note.extend([r for r in note_level_references if isinstance(r, str) and r])
# Detect signature form refs_note = _dedupe_seq(refs_note)
if args and isinstance(args[0], dict) and ("title" in args[0] or "type" in args[0] or "note_id" in args[0] or "id" in args[0]): for r in refs_note:
note = args[0] edges.append(_edge("references", "note", note_id, r, note_id, {
chunks = args[1] if len(args) > 1 else kwargs.get("chunks", []) "edge_id": _mk_edge_id("references", note_id, r, "note", "explicit:note_scope"),
types_cfg = args[2] if len(args) > 2 else kwargs.get("types_cfg") "provenance": "explicit",
note_id = _get(note, "note_id", "id") "rule_id": "explicit:note_scope",
title = _get(note, "title") or "" "confidence": 1.0,
note_type = _get(note, "type") }))
else: edges.append(_edge("backlink", "note", r, note_id, note_id, {
# Positional legacy form "edge_id": _mk_edge_id("backlink", r, note_id, "note", "derived:backlink"),
note_id = args[0] if len(args) > 0 else kwargs.get("note_id") "provenance": "rule",
title = args[1] if len(args) > 1 else kwargs.get("title", "") "rule_id": "derived:backlink",
note_type = args[2] if len(args) > 2 else kwargs.get("note_type") "confidence": 0.9,
chunks = args[3] if len(args) > 3 else kwargs.get("chunks", []) }))
types_cfg = args[4] if len(args) > 4 else kwargs.get("types_cfg") for rel in defaults:
if rel == "references":
continue
edges.append(_edge(rel, "note", note_id, r, note_id, {
"edge_id": _mk_edge_id(rel, note_id, r, "note", f"edge_defaults:{note_type}:{rel}"),
"provenance": "rule",
"rule_id": f"edge_defaults:{note_type}:{rel}",
"confidence": 0.7,
}))
if rel in {"related_to", "similar_to"}:
edges.append(_edge(rel, "note", r, note_id, note_id, {
"edge_id": _mk_edge_id(rel, r, note_id, "note", f"edge_defaults:{note_type}:{rel}"),
"provenance": "rule",
"rule_id": f"edge_defaults:{note_type}:{rel}",
"confidence": 0.7,
}))
chunks = list(chunks or []) # 5) De-Dupe (source_id, target_id, relation, rule_id)
seen: Set[Tuple[str,str,str,str]] = set()
# Structure edges out: List[dict] = []
edges = _structure_edges(note_id, title, chunks) for e in edges:
s = str(e.get("source_id") or "")
# Explicit / inline / callout / defaults t = str(e.get("target_id") or "")
more, _ = _explicit_and_inline_edges(note_id, title, note_type, chunks, types_cfg) rel = str(e.get("relation") or e.get("kind") or "edge")
_append_with_dedupe(edges, more) rule = str(e.get("rule_id") or "")
key = (s, t, rel, rule)
return edges if key in seen:
continue
seen.add(key)
out.append(e)
return out