Knowledge graphs and structured retrieval

Read a sample · Chapter 01 of 06

Start with the question you need to answer and the evidence you can inspect. A diagram can make relationships visible, but a readable picture does not establish that its edges are true.

Trace the evidence

All five recorded relationships

Edge	Relationship	Source
e1	Atlas uses Parser	s1
e2	Parser depends on Index	s2
e3	Index depends on Parser	s2
e4	Parser maintained by North	s3
e5	Index maintained by North	s3

• s1: Fictional Atlas manifest, revision 1.
• s2: Fictional package dependency note, revision 1.
• s3: Fictional maintenance register, revision 1.

Parser and Index form a directed dependency cycle (e2 and e3). Index also points to North (e5). Archive is isolated in this fixture. No recorded path to Archive does not prove that no real-world relationship exists. These source labels identify fictional records, not external evidence.

Our fictional Atlas project uses a Parser package. Parser depends on an Index package, and Index depends on Parser. Both packages are maintained by the North team. An Archive package has no recorded relationships. These invented records let us practise direction, cycles and missing information without making claims about Mickai systems or a real software supply chain.

The question is deliberately narrow: which entities can we reach from Atlas by following selected outgoing relationships for at most two steps? This is a reachability query, not a dependency-risk score or a judgement about the team. The answer must state the selected relation types, hop limit and graph snapshot. Asking which packages are safe would require a different definition and additional evidence.

Give each entity a stable identifier distinct from its display name. A package renamed in documentation can keep its identifier; two teams with the same display name must not be silently merged. In this lab, lowercase identifiers are local keys. In a shared system, decide the namespace and entity-resolution policy before combining sources. A similar name is a candidate match, not proof that two records describe the same entity.

Scroll sideways to see every column.

A source identifier attaches a relationship to a record. Here the source descriptions are explicitly fictional revision labels, not fabricated public URLs. In a real collection, retain a retrievable document version and the exact supporting passage. The graph may record what a source asserted without establishing that the assertion is current or correct.

RDF describes statements using subject, predicate and object. The W3C primer explains that this relationship is directed. Our JSON lab borrows that useful way to think, but is not RDF, JSON-LD or a SPARQL implementation. Its local type checks are application validation rules. Do not treat them as an implementation of RDF inference or assume a file becomes standards-compatible because it contains triples.

Create a new local folder for the four files in this workbook. Save the following fixture as graph.json. It has five nodes, three source descriptions and five uniquely identified edges. The short keys s, p and o mean subject, predicate and object. Keep these exact identifiers for the first run so you can compare your output with the worked result.

{
  "nodes": {
    "atlas": "project",
    "parser": "package",
    "index": "package",
    "north": "team",
    "archive": "package"
  },
  "sources": {
    "s1": "Fictional Atlas manifest, revision 1",
    "s2": "Fictional package dependency note, revision 1",
    "s3": "Fictional maintenance register, revision 1"
  },
  "edges": [
    {"id": "e1", "s": "atlas", "p": "uses", "o": "parser", "source": "s1"},
    {"id": "e2", "s": "parser", "p": "depends_on", "o": "index", "source": "s2"},
    {"id": "e3", "s": "index", "p": "depends_on", "o": "parser", "source": "s2"},
    {"id": "e4", "s": "parser", "p": "maintained_by", "o": "north", "source": "s3"},
    {"id": "e5", "s": "index", "p": "maintained_by", "o": "north", "source": "s3"}
  ]
}

Read a sample · Chapter 02 of 06

A malformed graph should fail clearly before producing an answer. Validation checks the contract of this small application; it does not verify a source document or solve entity resolution.

Save the next block as the first part of graph_lab.py. The following chapter supplies the rest of the same file. The schema allows projects, packages and teams, plus only the three named relations. Every endpoint must exist and have the right kind. For example, an edge from Atlas directly to North labelled uses is invalid because uses must end at a package.

Each edge has an identifier of its own. Repeating an edge identifier is rejected because a witness path must refer to one unambiguous record. Different identifiers may describe the same relationship with different supporting sources. The lab permits that duplication, then chooses one witness per reachable node. It does not merge evidence, reconcile disagreement or list every possible explanation.

The graph object is capped at 256 nodes, 128 sources and 1,024 edges. Those are teaching limits chosen for a small local exercise, not measured production capacity. The query also has a traversal budget. Neither limit protects the earlier JSON parsing step from an arbitrarily large file: use only the supplied local fixture here. A public ingest service needs byte limits and a separate parsing boundary.

from collections import deque
import re

KINDS = {"project", "package", "team"}
RELATIONS = {
    "uses": ("project", "package"),
    "depends_on": ("package", "package"),
    "maintained_by": ("package", "team"),
}


def valid_id(value):
    return (type(value) is str and
            re.fullmatch(r"[a-z][a-z0-9-]{0,39}", value) is not None)


def validate(data):
    if type(data) is not dict or set(data) != {"nodes", "sources", "edges"}:
        raise ValueError("Expected nodes, sources and edges")
    nodes, sources, edges = data["nodes"], data["sources"], data["edges"]
    if type(nodes) is not dict or not 1 <= len(nodes) <= 256:
        raise ValueError("Expected 1 to 256 nodes")
    if any(not valid_id(k) or type(v) is not str or v not in KINDS
           for k, v in nodes.items()):
        raise ValueError("Invalid node identifier or kind")
    if type(sources) is not dict or not 1 <= len(sources) <= 128:
        raise ValueError("Expected 1 to 128 sources")
    if any(not valid_id(k) or type(v) is not str or not v.strip()
           for k, v in sources.items()):
        raise ValueError("Invalid source identifier or description")
    if type(edges) is not list or len(edges) > 1024:
        raise ValueError("Expected up to 1024 edges")
    seen, adjacency = set(), {node: [] for node in nodes}
    for edge in edges:
        if type(edge) is not dict or set(edge) != {"id", "s", "p", "o", "source"}:
            raise ValueError("Invalid edge fields")
        if any(type(v) is not str for v in edge.values()):
            raise ValueError("Edge values must be strings")
        eid, s, p, o = edge["id"], edge["s"], edge["p"], edge["o"]
        if not valid_id(eid) or eid in seen:
            raise ValueError("Invalid or repeated edge identifier")
        if s not in nodes or o not in nodes or p not in RELATIONS:
            raise ValueError("Unknown endpoint or relation")
        if (nodes[s], nodes[o]) != RELATIONS[p]:
            raise ValueError("Relation endpoints have the wrong kinds")
        if edge["source"] not in sources:
            raise ValueError("Unknown source")
        seen.add(eid)
        adjacency[s].append(edge)
    for outgoing in adjacency.values():
        outgoing.sort(key=lambda edge: edge["id"])
    return nodes, adjacency

Validation builds an adjacency list for each node. The list contains only outgoing edges, so a relationship never becomes bidirectional accidentally. Sorting each list by edge identifier makes traversal reproducible when the input edge rows are rearranged. The function sorts its own lists, not the original edge list; it does not rewrite the supplied graph.

The source dictionary requires a non-empty description and a known key for each edge. That is a structural check only. An attacker or a mistaken editor could still invent a convincing description. Before accepting real data, independently inspect the document, version and passage, record extraction confidence and arrange review of ambiguous relationships. Keep an unsupported candidate separate from accepted records.

Standard JSON decoding creates an in-memory object. Its default treatment of duplicate object keys is another reason this exercise is not a hostile-file ingestion boundary. The duplicate-edge test later checks explicit edge IDs in the edge list; it does not certify every property of an arbitrary JSON document.

Read a sample · Chapter 03 of 06

A traversal needs a starting entity, allowed relation types and a stopping rule. Return evidence with the result instead of asking a language model to reconstruct the route afterwards.

Append this block directly after the first part of graph_lab.py. The queue holds a node, the edge IDs used to reach it and their source IDs in the same order. Python deque provides the queue operations used here. Taking from the front and adding at the back visits shallower paths before deeper ones.

def traverse(
    data, start, relations, *, max_hops=2, max_edges=100
):
    nodes, adjacency = validate(data)
    if type(start) is not str or start not in nodes:
        raise ValueError("Unknown start node")
    if (type(relations) not in (set, frozenset) or not relations or
            not relations <= RELATIONS.keys()):
        raise ValueError("Choose a non-empty set of known relations")
    if type(max_hops) is not int or not 0 <= max_hops <= 5:
        raise ValueError("max_hops must be an integer from 0 to 5")
    if type(max_edges) is not int or not 1 <= max_edges <= 1024:
        raise ValueError("max_edges must be an integer from 1 to 1024")
    queue = deque([(start, [], [])])
    seen, results, examined = {start}, [], 0
    while queue:
        node, path, sources = queue.popleft()
        if len(path) == max_hops:
            continue
        for edge in adjacency[node]:
            if edge["p"] not in relations:
                continue
            examined += 1
            if examined > max_edges:
                raise ValueError("Traversal edge budget exceeded")
            target = edge["o"]
            if target in seen:
                continue
            seen.add(target)
            next_path = path + [edge["id"]]
            next_sources = sources + [edge["source"]]
            results.append({
                "node": target, "edges": next_path,
                "sources": next_sources,
            })
            queue.append((target, next_path, next_sources))
    return results

The seen set starts with the starting node and is updated when a target is queued. Parser and Index can point back to one another, but neither is returned twice. The start node is not included in its own results. The hop limit counts edges in a witness path. With zero hops the answer is empty; with one hop from Atlas only Parser is reachable.

This breadth-first traversal returns one shortest witness, measured by the number of edges, for each reachable node. Sorted edge identifiers break ties deterministically. This is not a ranking by source quality, recency or probability. A shorter route can still rely on a weak source. If the task requires all supporting routes or a specific sequence of relations, this function is not a complete solution.

The relation selector is a set, so any chosen relation can be followed at any step. It does not express a pattern such as exactly one uses edge followed by exactly one maintained_by edge. Our two-hop Atlas result includes both Index and North. Filter or design an explicit query pattern when the question asks specifically for maintainers; do not call every reachable node a maintainer.

The edge budget counts eligible outgoing edges examined, including edges pointing to a node already seen. It does not count rejected relation types or all validation work. If another eligible edge would exceed the budget, the function raises an error instead of returning a partial answer as though it were complete. A production caller should turn that into an explicit incomplete-query state.

Save demo.py beside the other files. Its path is anchored to the script directory, so the fixture is found even if the terminal starts elsewhere. The file is local and no model, database server, account or paid tool is required.

import json
from pathlib import Path
from graph_lab import traverse

data = json.loads(Path(__file__).with_name("graph.json").read_text(encoding="utf-8"))
relations = {"uses", "depends_on", "maintained_by"}
for row in traverse(data, "atlas", relations, max_hops=2):
    print(row["node"], ">".join(row["edges"]), ",".join(row["sources"]))

python demo.py

parser e1 s1
index e1>e2 s1,s2
north e1>e4 s1,s3

Read a sample · Chapter 04 of 06

A retrieved path is an explanation of how the query reached a node. Preserve its edges and sources, then decide whether they support the exact sentence you intend to publish.

The North row contains e1 followed by e4. In the fixture, e1 says that Atlas uses Parser and e4 says that Parser is maintained by North. A careful answer is: the fictional register names North as maintainer of Parser, which the Atlas manifest lists as a used package. Cite both source records. Do not shorten this into North maintains Atlas, because the graph contains no such relationship.

Scroll sideways to see every column.

The edge IDs and source IDs are parallel lists. Reading each edge in order should begin at the query start and end at the returned node. Every relation must belong to the allowed set, and every cited source must be the source recorded on that edge. This is more inspectable than a list of unexplained entity names, but the validity of the underlying claims still depends on the sources.

The W3C PROV overview treats provenance as information about the entities, activities and people involved in producing something. Our source labels are a much smaller teaching device, not a PROV representation. A real extraction pipeline should also record what transformation created an edge, which document version it used and who or what reviewed it. A label alone is insufficient to replay that process.

Missing edges require careful language. No path to Archive means that the selected graph and query limits produced no path. It does not establish that Atlas never used Archive. The relationship might be absent from the source, missed by extraction, excluded by a filter or beyond the chosen hop bound. Record which of those possibilities you checked rather than inventing a negative fact.

Conflicting sources need a policy before answer generation. If two accepted records name different maintainers at different dates, keep the dates and versions, decide the relevant time and show disagreement when it cannot be resolved. Our fixture has no temporal schema and the traversal chooses just one path per node. Do not extend its claims to current ownership, trust or authorisation.

If you later add a language model, supply the exact permitted witness bundle and require claim-by-claim attribution. Treat retrieved source text as data, not as instructions to run tools or change the query. Check the resulting sentence against the bundle. Merely including citations in a prompt does not demonstrate that the answer faithfully uses them.

Read a sample · Chapter 05 of 06

Choose expected answers from the fixture before trusting the implementation. Test both which entities are returned and whether their explanation paths actually connect the right records.

Save the following as test_graph.py. It uses the standard-library test runner and the exact local fixture. Run the command below from the folder containing all four files. If an import fails or the fixture is missing, fix that environment error before interpreting anything as a graph test result.

import copy
import json
from pathlib import Path
import unittest
from graph_lab import traverse

DATA = json.loads(Path(__file__).with_name("graph.json").read_text(encoding="utf-8"))
ALL = {"uses", "depends_on", "maintained_by"}


class GraphTests(unittest.TestCase):
    def test_expected_witnesses(self):
        self.assertEqual(traverse(DATA, "atlas", ALL), [
            {"node": "parser", "edges": ["e1"], "sources": ["s1"]},
            {"node": "index", "edges": ["e1", "e2"], "sources": ["s1", "s2"]},
            {"node": "north", "edges": ["e1", "e4"], "sources": ["s1", "s3"]},
        ])

    def test_direction_and_filter(self):
        self.assertEqual(traverse(DATA, "north", ALL), [])
        self.assertEqual(traverse(DATA, "atlas", {"depends_on"}), [])
        self.assertEqual(traverse(DATA, "archive", ALL), [])

    def test_hops_and_cycles(self):
        self.assertEqual(traverse(DATA, "atlas", ALL, max_hops=0), [])
        rows = traverse(DATA, "atlas", ALL, max_hops=1)
        self.assertEqual([r["node"] for r in rows], ["parser"])
        rows = traverse(DATA, "parser", ALL, max_hops=5)
        self.assertEqual([r["node"] for r in rows], ["index", "north"])

    def test_order_and_input_are_preserved(self):
        snapshot = copy.deepcopy(DATA)
        reversed_rows = copy.deepcopy(DATA)
        reversed_rows["edges"].reverse()
        self.assertEqual(traverse(reversed_rows, "atlas", ALL), traverse(DATA, "atlas", ALL))
        self.assertEqual(DATA, snapshot)

    def test_invalid_edge(self):
        for field, value in [("o", "absent"), ("p", "owns"), ("source", "absent"), ("o", "north")]:
            with self.subTest(field=field, value=value):
                broken = copy.deepcopy(DATA)
                broken["edges"][0][field] = value
                with self.assertRaises(ValueError):
                    traverse(broken, "atlas", ALL)

    def test_duplicate_id(self):
        broken = copy.deepcopy(DATA)
        broken["edges"][1]["id"] = "e1"
        with self.assertRaises(ValueError):
            traverse(broken, "atlas", ALL)

    def test_query_arguments(self):
        for args in [{"max_hops": True}, {"max_hops": -1}, {"max_hops": 6}, {"max_edges": 0}]:
            with self.subTest(args=args), self.assertRaises(ValueError):
                traverse(DATA, "atlas", ALL, **args)
        with self.assertRaises(ValueError):
            traverse(DATA, "missing", ALL)
        with self.assertRaises(ValueError):
            traverse(DATA, "atlas", "uses")

    def test_budget_rejects_incomplete_answer(self):
        with self.assertRaisesRegex(ValueError, "budget"):
            traverse(DATA, "atlas", ALL, max_edges=2)
        self.assertEqual(len(traverse(DATA, "atlas", ALL, max_edges=3)), 3)


if __name__ == "__main__":
    unittest.main()

python -m unittest -v test_graph

The eight test methods cover exact witnesses, direction and filters, hop limits and cycles, row order and input preservation, invalid edges, duplicate identifiers, query arguments and the budget boundary. Assertions about sources matter as much as assertions about returned nodes. A graph could retrieve North correctly while attaching the wrong source to the explanation.

The release also ran an independent recursive reachability oracle across 210 combinations of start node, relation subset and hop bound. It checked every returned path and tried all 120 permutations of the five input edges. Together with rejection checks, this produced 950 additional assertions. These are recorded results for this tiny fixture, not measurements of general retrieval accuracy or production scale.

Three isolated mutations were tested: following the subject as the target, inventing the source ID s1 for every hop, and allowing an extra hop. All were caught. The extra-hop defect also exhausted an edge budget that the correct implementation satisfies. This illustrates why a test suite should check limits and evidence, not just one successful demo.

Read a sample · Chapter 06 of 06

Use structure where relationships are explicit and valuable. Keep documents available for details that the graph does not represent, and evaluate the combined system against questions with known evidence.

A practical workflow separates ingestion, candidate extraction, review, graph validation, query planning, retrieval and answer checking. Each boundary has a different job. An extractor proposes entities and relationships from documents; a reviewer accepts or rejects them; validation checks the graph shape; retrieval follows permitted edges; answer checking compares each sentence with the retrieved evidence.

Text retrieval and graph retrieval solve different parts of the problem. A graph query can enforce a relation pattern when entity IDs and edges are reliable. Text search can locate supporting passages, definitions or information that was never represented as an edge. Combining them can help, but adds work: the entity mapping, source version and passage must all refer to the same thing.

For the Atlas question, first resolve Atlas to its exact project identifier. Retrieve a bounded path, collect its source IDs, then fetch only the corresponding reviewed passages from the same snapshot. If the passage no longer supports the edge, stop or mark the answer unresolved. Do not silently replace it with a similar passage from another revision.

Build a small evaluation sheet before adding more automation. Include a direct relationship, a two-hop relationship, a reverse-direction trap, a cycle, an isolated node, ambiguous entity names, conflicting revisions and a budget failure. Record expected entities, required evidence and the allowed abstention wording. Report retrieval correctness separately from whether a generated answer faithfully states it.

This lab has no persistence layer, authorisation model, source-fetching code or language-model call. The JSON is loaded into memory and structurally validated on every query. Before real deployment, define access restrictions, snapshot/version handling, byte limits, logging that avoids sensitive source text, update and withdrawal procedures, and performance measurements on representative data. Do not present this small script as a complete private knowledge-base service.

Finish with a handover another person can reproduce. Include the four files, interpreter version, exact query, expected output and tests actually run. Name the fictional nature of the fixture and the one-witness limitation. Record deferred decisions separately from completed checks. A graph drawing can support the explanation, but retain the adjacency table and evidence paths as accessible text.