Add a directed edge between two nodes. Args: source_document: doc_id from documents_store (provenance pointer). supporting_text: The exact text span that supports this relation. chunk_id: The specific chunk_id the supporting text came from.
(
source_id: str,
target_id: str,
relation: str,
confidence: float,
source_document: str | None = None,
supporting_text: str | None = None,
chunk_id: str | None = None,
)
| 81 | |
| 82 | |
| 83 | def add_edge( |
| 84 | source_id: str, |
| 85 | target_id: str, |
| 86 | relation: str, |
| 87 | confidence: float, |
| 88 | source_document: str | None = None, |
| 89 | supporting_text: str | None = None, |
| 90 | chunk_id: str | None = None, |
| 91 | ) -> None: |
| 92 | """ |
| 93 | Add a directed edge between two nodes. |
| 94 | |
| 95 | Args: |
| 96 | source_document: doc_id from documents_store (provenance pointer). |
| 97 | supporting_text: The exact text span that supports this relation. |
| 98 | chunk_id: The specific chunk_id the supporting text came from. |
| 99 | """ |
| 100 | graph = _load() |
| 101 | |
| 102 | # Deduplicate edges by source + target + relation |
| 103 | relation_lower = relation.strip().lower() |
| 104 | for edge in graph["edges"]: |
| 105 | if ( |
| 106 | edge["source"] == source_id |
| 107 | and edge["target"] == target_id |
| 108 | and edge["type"] == relation_lower |
| 109 | ): |
| 110 | changed = False |
| 111 | if confidence > edge["confidence"]: |
| 112 | edge["confidence"] = confidence |
| 113 | changed = True |
| 114 | if source_document and edge.get("source_document") is None: |
| 115 | edge["source_document"] = source_document |
| 116 | changed = True |
| 117 | if supporting_text and edge.get("supporting_text") is None: |
| 118 | edge["supporting_text"] = supporting_text |
| 119 | changed = True |
| 120 | if chunk_id and edge.get("chunk_id") is None: |
| 121 | edge["chunk_id"] = chunk_id |
| 122 | changed = True |
| 123 | if changed: |
| 124 | _save(graph) |
| 125 | return |
| 126 | |
| 127 | graph["edges"].append({ |
| 128 | "source": source_id, |
| 129 | "target": target_id, |
| 130 | "type": relation_lower, |
| 131 | "confidence": confidence, |
| 132 | "source_document": source_document, |
| 133 | "supporting_text": supporting_text, |
| 134 | "chunk_id": chunk_id, |
| 135 | }) |
| 136 | _save(graph) |
| 137 | |
| 138 | |
| 139 | def get_neighbors(node_id: str, min_confidence: float = 0.0) -> list[str]: |