Knowledge Graph Tutorial: Linking Records Across Collections

Collections are great for storing records, but “who works with whom”, “what does this document depend on”, “how are these two people connected” - those are relationship questions, and answering them with filters means rebuilding joins in your application. VSKI graphs exist for exactly that.

This tutorial builds a small company knowledge base: people, projects and documents live in collections, and a graph indexes how they relate. At the end you can answer relationship questions in one API call - and visualize the result in the dashboard.

Note This page covers the graph API itself. For the reference documentation see Graphs.

What You’ll Build

A kb graph over three collections, with three relationship types:

• works_on - person -> project (weight: hours per week)
• authored - person -> document
• depends_on - document -> document (weight: coupling, lower = tighter)

and the queries that come with it:

• who works on a project (and what else they work on)
• the full dependency impact of changing one document
• the closest connection between two people, weighted by involvement
• a neighborhood subgraph for the dashboard

Architecture

┌─────────────────┐      ┌──────────────────────────────┐
│   Script/Cli    │─────▶│         VSKI Server          │
│   (Python)      │ HTTP │                              │
└─────────────────┘      │  ┌────────────────────────┐  │
                         │  │ collections            │  │
                         │  │  users, projects, docs │  │
                         │  ├────────────────────────┤  │
                         │  │ graph kb               │  │
                         │  │  nodes + labelled      │  │
                         │  │  edges (SQLite)        │  │
                         │  │  + in-memory index     │  │
                         │  └────────────────────────┘  │
                         └──────────────────────────────┘

Records live in collections. The graph references them by properties.ref = { collection, recordId } and adds the relationship layer on top. Neither side owns the other: you can re-index the graph from the collections at any time.

Prerequisites

• Docker
• Python 3.11+
• The VSKI CLI:

  pip install "vski-sdk[cli]" --extra-index-url https://vski.sh/api/packages/x/pypi/simple/

Quick Start

1. Start VSKI

docker-compose.yml

services:
  vski:
    image: vski.sh/x/vski:latest-standalone
    ports:
      - "8347:8347"
    environment:
      - SERVER_PORT=8347
      - DATA_DIR=/app/data
      - JWT_SECRET=dev-secret-change-in-production
    volumes:
      - vski_data:/app/data

volumes:
  vski_data:

docker compose up -d

Create your admin at http://localhost:8347/installer, then log in with the command line:

vski login --url http://localhost:8347 --email admin@example.com --password yourpassword
vski db create kb

Step 1: Create the Collections

The data itself is ordinary collections. Only the fields the graph rules will look at later need to be there from the start. Declare the schema in the IaC YAML (see IaC) and apply it:

vski/kb/collections.yaml

version: 1
database: kb

collections:
  users:
    type: base
    fields:
      - {name: email, type: email, required: true}
      - {name: name, type: text, required: true}
  projects:
    type: base
    fields:
      - {name: title, type: text, required: true}
  documents:
    type: base
    fields:
      - {name: title, type: text, required: true}
      - {name: status, type: select, options: {values: [draft, review, final]}}

vski plan      # + collections users, projects, documents
vski apply     # creates all three via server-side migrations

And some records:

src/setup.py

import asyncio

from vski import VskiClient


async def main():
    async with VskiClient("http://localhost:8347") as client:
        res = await client.admins.auth_with_password(
            "admin@example.com", "yourpassword"
        )
        client.set_token(res["token"])
        db = client.db("kb")

        alice = await db.collection("users").create(
            {"email": "alice@example.com", "name": "Alice"}
        )
        bob = await db.collection("users").create(
            {"email": "bob@example.com", "name": "Bob"}
        )
        carol = await db.collection("users").create(
            {"email": "carol@example.com", "name": "Carol"}
        )

        atlas = await db.collection("projects").create({"title": "Atlas"})

        api = await db.collection("documents").create(
            {"title": "API design", "status": "review"}
        )
        db_schema = await db.collection("documents").create(
            {"title": "Database schema", "status": "final"}
        )
        onboarding = await db.collection("documents").create(
            {"title": "Onboarding guide", "status": "draft"}
        )


asyncio.run(main())

Step 2: Create the Graph

The graph is declared in the same vski/ layout:

vski/kb/graphs.yaml

version: 1
database: kb

graphs:
  kb:
    options:
      directed: true
      maxNodes: 10000   # this graph stays small; keep the caps tight
      maxEdges: 50000

vski apply

kb = db.graph("kb")

Exceeding a cap returns a clean 400 (graph edge cap exceeded) instead of growing forever - set caps to what the graph is allowed to be, not what it is today.

Step 3: Mirror the Records as Nodes

One node per record, with ref pointing back at the source of truth:

nodes = [
    {"id": "u:alice", "label": "person",
     "properties": {"ref": {"collection": "users", "recordId": alice["id"]}}},
    {"id": "u:bob", "label": "person",
     "properties": {"ref": {"collection": "users", "recordId": bob["id"]}}},
    {"id": "u:carol", "label": "person",
     "properties": {"ref": {"collection": "users", "recordId": carol["id"]}}},
    {"id": "p:atlas", "label": "project",
     "properties": {"ref": {"collection": "projects", "recordId": atlas["id"]}}},
    {"id": "d:api", "label": "document",
     "properties": {"ref": {"collection": "documents", "recordId": api["id"]}}},
    {"id": "d:schema", "label": "document",
     "properties": {"ref": {"collection": "documents", "recordId": db_schema["id"]}}},
    {"id": "d:onboarding", "label": "document",
     "properties": {"ref": {"collection": "documents", "recordId": onboarding["id"]}}},
]

await kb.create_nodes(nodes)  # one bulk request

The u: / p: / d: prefixes make ids globally unique inside the graph (collection record ids alone could collide) and make substring search useful - q: "d:" finds every document node.

Step 4: Add the Relationships

await kb.create_edges([
    # person -> project, weight = hours per week
    {"source": "u:alice", "target": "p:atlas", "label": "works_on", "weight": 20},
    {"source": "u:bob", "target": "p:atlas", "label": "works_on", "weight": 30},
    {"source": "u:carol", "target": "p:atlas", "label": "works_on", "weight": 10},

    # person -> document
    {"source": "u:alice", "target": "d:api", "label": "authored"},
    {"source": "u:bob", "target": "d:schema", "label": "authored"},
    {"source": "u:carol", "target": "d:onboarding", "label": "authored"},

    # document -> document, weight = coupling (lower = tighter)
    {"source": "d:api", "target": "d:schema", "label": "depends_on", "weight": 1},
    {"source": "d:onboarding", "target": "d:api", "label": "depends_on", "weight": 2},
])

Edges require both endpoints to exist - a missing node fails the whole request with node not found.

Step 5: Ask Relationship Questions

Who works on Atlas?

One hop, into the project node:

team = await kb.neighbors("p:atlas", direction="in")
print([n["id"] for n in team["nodes"]])
# ["u:alice", "u:bob", "u:carol"]

What breaks if the database schema changes?

Walk depends_on edges pointing at the schema document - everything that (transitively) depends on it is affected, whatever the distance:

impact = await kb.traverse(
    "d:schema",
    depth=10,            # hard server maximum
    direction="in",      # things that depend on it
    labels=["depends_on"],
)
print([n["id"] for n in impact["nodes"]])

Note the direction: depends_on points from dependent to dependency, so “what is affected by X” is an inbound walk.

How are Alice and Carol connected?

The path finder answers this in one call. With cost: "hops" you get the fewest-step route; with cost: "weight" Dijkstra minimizes the summed weight:

by_hops = await kb.path("u:alice", "u:carol", cost="hops")
print([n["id"] for n in by_hops["nodes"]])
# ["u:alice", "p:atlas", "u:carol"] - 2 hops through the shared project
print(by_hops["cost"])   # 30 (Alice's 20 + Carol's 10 weekly hours)

by_weight = await kb.path("u:alice", "u:carol", cost="weight")
print([n["id"] for n in by_weight["nodes"]])
# ["u:alice", "d:api", "d:onboarding", "u:carol"] - cheaper through documents
print(by_weight["cost"])  # 4 (authored = 1, depends_on = 2, authored = 1)

Two path semantics worth knowing: the finder follows edges in both directions by default (unlike neighbors/traverse, which respect the graph’s own direction; pass direction="out" or "in" to enforce it), and unweighted edges count as weight 1 - which is why the document route wins once weights are summed.

Restricting the walk to certain relationship types keeps the answer meaningful:

via_work = await kb.path(
    "u:alice", "u:carol",
    cost="weight",
    labels=["works_on"],  # collaboration edges only
)
# ["u:alice", "p:atlas", "u:carol"] - the documents are excluded

Pull a neighborhood for visualization

The subgraph endpoint exists for rendering - seed nodes plus everything within a few hops, capped at 5 hops / 1000 nodes:

hood = await kb.subgraph(["u:alice"], hops=2)
print(f"{len(hood['nodes'])} nodes, {len(hood['edges'])} edges")
# "4 nodes, 3 edges": alice, p:atlas, d:api and d:schema

Like neighbors and traverse, the walk follows outbound edges by default - pass direction="both" to pull inbound relationships too (useful when the seed is a “sink”, e.g. a project everyone points at).

And check the shape of the whole graph without touching a node:

stats = await kb.stats()
print(stats["edgeLabels"])  # {"works_on": 3, "authored": 3, "depends_on": 2}
print(f"{stats['avgDegree']:.2f}", stats["maxDegree"])

Step 6: Keep the Graph in Sync

Records change; the graph should follow. Because nodes carry ref, sync is a small loop:

# src/sync.py - run whenever records change (e.g. from a trigger)
async def sync_user(record: dict) -> None:
    node_id = f"u:{record['id']}"
    existing = await kb.get_nodes(q=node_id)
    if existing["totalItems"] == 0:
        await kb.create_nodes({
            "id": node_id,
            "label": "person",
            "properties": {"ref": {"collection": "users", "recordId": record["id"]}},
        })

For deletions, remember that deleting a node cascades its edges - removing u:bob when Bob leaves cleanly disconnects him from every project and document. The reverse also holds: deleting a collection record does not touch the graph, so sync hooks (or the re-index script below) should handle that side.

Re-indexing from scratch is the same three steps as the initial build: create a fresh graph, mirror records, replay relationships.

Step 7: Open It Up (Rules)

Graphs use the collection rule model, and rules are part of the graph’s IaC block. This config keeps structure edits admin-only but lets any authenticated user explore:

# vski/kb/graphs.yaml - the kb block, now with rules
version: 1
database: kb

graphs:
  kb:
    options:
      directed: true
      maxNodes: 10000
      maxEdges: 50000
    rules:
      list: "@request.auth.id != ''"
      view: "@request.auth.id != ''"   # gates neighbors/traverse/path/subgraph
      create: null                     # admin-only
      update: null
      delete: null

vski apply

Per-item rules see node fields, so a read-only layer is one line away:

list: "label = 'document'"
# every non-admin now sees just the three document nodes; persons and
# projects stay admin-only, and the response carries filtered: true

Step 8: Explore in the Dashboard

Open http://localhost:8347 - the sidebar has a Graphs section. The kb explorer gives you:

• a searchable center-node picker (substring over ids and labels, backed by the same q parameter used above)
• hops (1-5), limit and direction controls
• click-to-select with Center here / Expand actions
• a from/to shortest-path finder with hops or weight cost
• per-label edge colors and counts, node/edge create-edit-delete

Everything the tutorial did through the SDK is available interactively - handy for verifying the graph after each sync run.

Full Script

# src/main.py - the whole tutorial as one runnable script
# Schema first: `vski apply` creates the collections and the kb graph from
# the YAML files above; this script seeds the data and runs the queries.
import asyncio

from vski import VskiClient


async def main():
    async with VskiClient("http://localhost:8347") as client:
        res = await client.admins.auth_with_password(
            "admin@example.com", "yourpassword"
        )
        client.set_token(res["token"])
        db = client.db("kb")

        # 1. records
        alice = await db.collection("users").create(
            {"email": "alice@example.com", "name": "Alice"}
        )
        bob = await db.collection("users").create(
            {"email": "bob@example.com", "name": "Bob"}
        )
        carol = await db.collection("users").create(
            {"email": "carol@example.com", "name": "Carol"}
        )
        atlas = await db.collection("projects").create({"title": "Atlas"})
        api = await db.collection("documents").create(
            {"title": "API design", "status": "review"}
        )
        schema = await db.collection("documents").create(
            {"title": "Database schema", "status": "final"}
        )
        onboarding = await db.collection("documents").create(
            {"title": "Onboarding guide", "status": "draft"}
        )

        # 2. nodes (ref -> collection records)
        def ref(collection, record_id):
            return {"collection": collection, "recordId": record_id}

        kb = db.graph("kb")
        await kb.create_nodes([
            {"id": "u:alice", "label": "person",
             "properties": {"ref": ref("users", alice["id"])}},
            {"id": "u:bob", "label": "person",
             "properties": {"ref": ref("users", bob["id"])}},
            {"id": "u:carol", "label": "person",
             "properties": {"ref": ref("users", carol["id"])}},
            {"id": "p:atlas", "label": "project",
             "properties": {"ref": ref("projects", atlas["id"])}},
            {"id": "d:api", "label": "document",
             "properties": {"ref": ref("documents", api["id"])}},
            {"id": "d:schema", "label": "document",
             "properties": {"ref": ref("documents", schema["id"])}},
            {"id": "d:onboarding", "label": "document",
             "properties": {"ref": ref("documents", onboarding["id"])}},
        ])

        # 3. relationships
        await kb.create_edges([
            {"source": "u:alice", "target": "p:atlas", "label": "works_on", "weight": 20},
            {"source": "u:bob", "target": "p:atlas", "label": "works_on", "weight": 30},
            {"source": "u:carol", "target": "p:atlas", "label": "works_on", "weight": 10},
            {"source": "u:alice", "target": "d:api", "label": "authored"},
            {"source": "u:bob", "target": "d:schema", "label": "authored"},
            {"source": "u:carol", "target": "d:onboarding", "label": "authored"},
            {"source": "d:api", "target": "d:schema", "label": "depends_on", "weight": 1},
            {"source": "d:onboarding", "target": "d:api", "label": "depends_on", "weight": 2},
        ])

        # 4. queries
        team = await kb.neighbors("p:atlas", direction="in")
        print("Atlas team:", [n["id"] for n in team["nodes"]])

        impact = await kb.traverse(
            "d:schema", depth=10, direction="in", labels=["depends_on"]
        )
        print("Schema change affects:", [n["id"] for n in impact["nodes"]])

        route = await kb.path("u:alice", "u:carol", cost="weight")
        print("Alice -> Carol:", [n["id"] for n in route["nodes"]], "cost", route["cost"])
        # -> ["u:alice", "d:api", "d:onboarding", "u:carol"] cost 4

        stats = await kb.stats()
        print("Labels:", stats["edgeLabels"],
              "avgDegree", f"{stats['avgDegree']:.2f}")


asyncio.run(main())

Run it with:

python src/main.py

(The SDK is the plain pip package from Prerequisites - pip install "vski-sdk[cli]" - so any Python 3.11+ interpreter can run the script without further setup.)

Troubleshooting

node not found when creating edges

Both source and target must exist in the graph before the edge is created. Check with graph.get_nodes(q="u:") - remember that graph node ids (u:alice) are not collection record ids.

graph edge cap exceeded (400)

The graph hit options.maxEdges (or the server-wide GRAPH_MAX_EDGES). Raise the per-graph cap in vski/kb/graphs.yaml and re-run vski apply - the write that failed did not partially apply.

graph not found (404) although it exists

Graphs are db-scoped with strict resolution: client.graph("kb") targets the client’s default database. Use client.db("kb").graph("kb") (as in this tutorial) or client.graph("kb", "kb").

Traversal returns empty results

Two usual causes: direction (depends_on impact analysis is an inbound walk) and labels filtering (labels must match exactly, including case and empty strings).

Next Steps

• Wire the sync loop from Step 6 into Triggers so record changes update the graph automatically
• Subscribe to db:kb realtime events to live-update a visualization (Realtime)
• Combine graph traversal with collection queries: resolve a route’s node ids to full records via properties.ref
• Explore weighting schemes - works_on hours, review status, whatever makes cost: "weight" meaningful in your domain

Resources

• Graphs reference
• Collections
• Rules and Access Control
• VSKI Documentation