Working with Graphs

This guide explains the core concepts of the Knowledge Platform and how to work with graphs, nodes, and edges — either through the UI or programmatically via the Python API.

Core Concepts

The Knowledge Graph Model

graph LR
    W["Workspace"] -->|contains| G1["Graph (outline)"]
    W -->|contains| G2["Graph (outline)"]
    G1 -->|nodes| N1["Node: OutlineItem"]
    G1 -->|nodes| N2["Node: OutlineItem"]
    G1 -->|edges| E1["Edge: ParentOf"]
    E1 -->|from| N1
    E1 -->|to| N2

The platform organises knowledge as a hierarchy of containers:

Concept Description
Workspace A named project container. Holds one or more graphs. Each workspace has its own SQLite database on disk.
Graph A typed collection of nodes and edges. The graph type (e.g. "outline") defines what nodes and edges are valid.
Node A single entity in a graph. Has a type name (e.g. "OutlineItem") and a key-value attributes dictionary.
Edge A directed relationship between two nodes. Has a type name (e.g. "ParentOf") and optionally attributes.
Graph Type A schema that declares which node types and edge types are valid, along with required/optional attributes for each.

Workspaces

A workspace maps to a SQLite database on disk under the platform workspace directory. The application keeps one workspace active at a time, and you can create or open workspaces from the running UI.

Graphs within a Workspace

Each graph belongs to exactly one workspace and has exactly one type_name. The type name determines which module can edit the graph. You cannot mix node types from different graph types within the same graph.

Immutable Nodes and Edges

Nodes and edges are immutable value objects. When you "update" a node, you are actually producing a new version of it with a higher version counter. The original is replaced in the graph's in-memory store and in the database.

Using the Service API (Programmatic)

The following examples show how to interact with graphs via the Python service layer. This is useful for scripting, testing, or building new UI components.

Setup

from knowledge_platform.core.engine import GraphEngine
from knowledge_platform.domain.registry import TypeRegistry
from knowledge_platform.modules.outline.graph_type import OutlineGraphType
from knowledge_platform.persistence.store import SqliteGraphRepository
from knowledge_platform.services.graph_service import GraphService
from knowledge_platform.services.workspace_service import WorkspaceService
from pathlib import Path

# Wire the stack
repo = SqliteGraphRepository(":memory:")
engine = GraphEngine(repo)

type_registry = TypeRegistry()
type_registry.register(OutlineGraphType())

graph_service = GraphService(engine, type_registry)
workspace_service = WorkspaceService(base_directory=Path("/tmp/my_workspaces"))

Creating a Workspace and Graph

# Create a workspace
workspace = workspace_service.create_workspace("Research Notes")

# Create a graph in that workspace
graph = graph_service.create_graph(
    workspace_id=workspace.id,
    type_name="outline",
    name="Chapter 1",
)
print(graph.id)        # GraphId("3f2504e0-...")
print(graph.version)   # 1

Adding Nodes

# Add a node — attributes are validated against the OutlineGraphType schema
node = graph_service.add_node(
    graph_id=graph.id,
    type_name="OutlineItem",
    attributes={"title": "Introduction", "content": "Opening paragraph.", "position": 0},
)
print(node.id)      # NodeId("...")
print(node.version) # 1

Updating a Node

update_node performs a partial merge — unspecified attributes are preserved:

updated = graph_service.update_node(
    node_id=node.id,
    attributes={"content": "Revised opening paragraph."},
)
# title and position are unchanged
print(updated.version)                   # 2
print(updated.attributes["content"])     # "Revised opening paragraph."
print(updated.attributes["title"])       # "Introduction"

Adding Edges

child_node = graph_service.add_node(
    graph.id, "OutlineItem",
    {"title": "Section 1.1", "position": 0},
)

edge = graph_service.add_edge(
    graph_id=graph.id,
    source_id=node.id,
    target_id=child_node.id,
    type_name="ParentOf",
    attributes={},
)

Removing Nodes and Edges

# Remove edge only
graph_service.remove_edge(graph.id, edge.id)

# Remove node — also cascades and removes all incident edges
graph_service.remove_node(graph.id, node.id)

Querying a Graph

After mutations, retrieve the latest in-memory state:

graph = graph_service.get_graph(graph.id)

# Count nodes and edges
print(graph.node_count())  # int
print(graph.edge_count())  # int

# Iterate all nodes of a type
for n in graph.nodes(type_name="OutlineItem"):
    print(n.attributes["title"])

# Iterate all edges of a type
for e in graph.edges(type_name="ParentOf"):
    print(f"{e.source_id} -> {e.target_id}")

# Structural queries
for e in graph.outgoing_edges(node_id, type_name="ParentOf"):
    print(e.target_id)   # children of node_id

for e in graph.incoming_edges(node_id, type_name="ParentOf"):
    print(e.source_id)   # parents of node_id

Listing and Deleting Graphs

# List all graphs in a workspace
graphs = graph_service.list_graphs(workspace.id)
for g in graphs:
    print(f"{g.name} ({g.type_name})")

# Permanently delete a graph
graph_service.delete_graph(graph.id)

Using Transactions

For operations that must succeed or fail together, use the transaction() context manager:

from knowledge_platform.core.transaction import transaction

with transaction(graph.id) as tx:
    tx.record_add_node(parent_node)
    tx.record_add_node(child_node)
    tx.record_add_edge(edge)
    engine.commit(tx)
# All three operations are persisted atomically

If an exception is raised before engine.commit(), the transaction is marked rolled_back and no changes are applied:

try:
    with transaction(graph.id) as tx:
        tx.record_add_node(node)
        raise ValueError("abort!")
except ValueError:
    pass

assert tx.rolled_back
assert graph.node_count() == 0  # nothing was applied

Validation Errors

The service layer validates all mutations against the active GraphType. If validation fails, ValueError is raised with a human-readable message:

try:
    graph_service.add_node(graph.id, "OutlineItem", {})  # missing required "title"
except ValueError as e:
    print(e)  # "Missing required attribute 'title' on node type 'OutlineItem'"

Common validation errors:

Error Cause
Missing required attribute 'X' A required field was not provided
Attribute 'X' expected str got int Wrong Python type for a field
Edge type 'X' does not allow source type 'Y' The source node type is not in allowed_source_types
Node type 'X' is not registered on graph type 'Y' Unknown node type for this graph

Cleanup

Always close repositories when you are done:

workspace_service.close()  # disposes all open SQLite connection pools