Org Chart from JSON Exports

You have your org data as plain exports — one JSON file listing companies and their departments, another listing people and where they work — and you want it in memory as a connected graph of companies, departments, and employees, without an LLM re-deriving structure you already know.

What You’ll Build

Two JSON files — companies.json and people.json — are loaded and mapped onto typed DataPoint classes, then written straight into the graph and vector stores by a two-task custom pipeline: no chunking, no extraction, no cognify pass. What comes out is an org chart you can open in a browser: Company nodes linked to their Department nodes, each department linked to the Person nodes that work in it, with every name embedded for semantic search. Node IDs are derived from the data itself, so the same department name reaching the pipeline twice lands on one node rather than two. The demo ships as two variants of the same scenario, so this page walks through both: a high-level script that lets identity_fields deduplicate for it, and a low-level script that assembles and deduplicates the graph by hand. The complete runnable scripts are in examples/demos/custom_pipelines/organizational_hierarchy — this page walks through their key moments rather than reproducing them.

Features in Play

DataPoints — Person, Department, Company, and CompanyType are the node types; index_fields decides what gets embedded and identity_fields derives the deterministic ID that collapses repeats
Custom Data Models — add_data_points persists the typed objects and their nesting as nodes and edges, with no LLM in the path
Custom Tasks and Pipelines — two Task objects run by run_tasks against a dataset: one maps JSON to DataPoints, one stores them
Graph Visualization — both scripts close by rendering the org chart to an HTML file you can open
Search — the low-level variant finishes with a GRAPH_COMPLETION query, proving the hand-built nodes answer questions like any other memory

How It Looks

This is the graph the high-level script writes, loaded from a real run of it. Drag a node to pull it out of the tangle, scroll to zoom, and hover one to fade out everything it does not touch — hovering a department is the quickest way to see exactly who works in it.

The org chart from a real run: one CompanyType node at the centre, five companies around it, each linked to the departments it declares, and each department to the people in it.

Every node and edge here comes from the run’s visualize_graph export, so the shape is exactly what lands in your graph store: Research & Development sits on GreenFuture with no one attached to it, because the company declares the department but no one in people.json works there.

What to Expect

The lines below come from one real run of the high-level script, trimmed. Nothing in this demo calls an LLM, so the node and edge counts are identical on every run — only the UUIDs, timestamps, and paths change. A first run also prints a long block of relational migrations as cognee builds its tables; that block is setup, not part of the demo. The pipeline starts with both JSON files already in hand. build_lightweight_data_object has wrapped the two files into a single LightweightData DataPoint, so the payload printed at the start of the run is the whole input — five companies and ten people — before any task has touched it.

status='PipelineRunStarted' pipeline_run_id=UUID('40da9a6f-3948-43be-a90f-e551570c40dd') dataset_id=UUID('94add405-52f0-5e30-871b-b2e8eeda31c7') dataset_name='test_dataset' payload=[LightweightData(id=UUID('c22f2fa6-5eee-5a20-a8d5-f3a1d10dbbbd'), ...

The two tasks finish in about three seconds. ingest_files maps the payload and add_data_points writes the result. There is no extraction step between them, which is why the whole pipeline costs less than a single LLM call would.

2026-09-11T15:24:05.784975 [info     ] Function task started: `ingest_files` [run_tasks_base]
2026-09-11T15:24:05.792215 [info     ] Coroutine task started: `add_data_points` [run_tasks_base]
...
2026-09-11T15:24:08.755657 [info     ] Coroutine task completed: `add_data_points` [run_tasks_base]
2026-09-11T15:24:08.762174 [info     ] Function task completed: `ingest_files` [run_tasks_base]
2026-09-11T15:24:08.768537 [info     ] Pipeline run completed: `07f15429-bcc9-56e7-8058-6ff9d3fd6372` [run_tasks_with_telemetry()]

The graph that lands is 27 nodes and 26 edges, and the arithmetic is what tells you the mapping was right. Five Company nodes, eleven Department nodes, ten Person nodes, and the one CompanyType all five companies point at make 27. Eleven company-to-department edges, ten department-to-employee edges, and five is_type edges make 26. Had a department name landed as two nodes, both numbers would be higher.

2026-09-11T15:24:08.791366 [info     ] Retrieved 27 nodes and 26 edges in 0.00 seconds [cognee.shared.logging_utils]
2026-09-11T15:24:08.797151 [info     ] Neighborhood retrieval (2-hop): 27 nodes and 26 edges in 0.01s [cognee.shared.logging_utils]

One warning is expected and does not mean the run failed. add_data_points did embed every index_fields value — the Person_name, Department_name, and Company_name collections are written and searchable. The warning comes from the renderer’s semantic-map layer, which only recognizes cognee’s built-in node types and so has nothing to plot for custom DataPoints. The graph view itself is complete.

2026-09-11T15:24:08.798832 [info     ] fetch_node_embeddings: resolved 0/27 node embeddings across 0 collection(s) [embedding_join]
2026-09-11T15:24:08.798890 [warning  ] fetch_node_embeddings: no embeddings resolved — the semantic map will be empty. Missing collections: none. Unmapped node types: ['Department', 'Company', 'Person', 'CompanyType']. [embedding_join]

Before You Start

Complete Quickstart to understand basic operations
Ensure you have LLM Providers configured — add_data_points embeds every index_fields value, so an embedding provider is required even though neither script extracts anything with an LLM; the low-level variant’s closing query also needs a completion model
Run both scripts from a checkout of the cognee repo: they read companies.json and people.json from the data/ folder next to them, and write their HTML renders into a sibling .artifacts/ folder
The high-level script starts with prune.prune_data() and prune.prune_system(metadata=True), which wipes the configured instance — point it at a scratch instance rather than memory you want to keep. The low-level script pins its system root to a .cognee_system folder beside itself before pruning, so it stays self-contained

How It Works

Stage 1: Model the Org Chart as Deduplicating DataPoints

Source: examples/demos/custom_pipelines/organizational_hierarchy/organizational_hierarchy_pipeline_example.py

class Person(DataPoint):
    name: str
    # "index_fields": fields to embed for vector search
    # "identity_fields": fields used to generate deterministic IDs (deduplication)
    metadata: dict = {"index_fields": ["name"], "identity_fields": ["name"]}


class Department(DataPoint):
    name: str
    employees: list[Person]
    metadata: dict = {"index_fields": ["name"], "identity_fields": ["name"]}

These classes are the entire schema of the resulting graph — Company and CompanyType follow the same pattern below them. Two things happen per class: index_fields names the field that gets embedded, so the graph is searchable by name; identity_fields makes the node’s UUID a deterministic function of that name, so building Person(name="John Doe") twice lands on one node instead of two. The employees field is what makes this a graph rather than two tables — a list of Person objects on a Department becomes edges from that department to those people.

Stage 2: Map the JSON Rows onto Nested DataPoints

Source: examples/demos/custom_pipelines/organizational_hierarchy/organizational_hierarchy_pipeline_example.py

def ingest_files(data: list[Any]) -> list[Company]:
    # With identity_fields, DataPoints with the same name automatically get the same UUID.
    # No manual dict-based deduplication needed — just create instances freely.
    all_companies: list[Company] = []

    # Single CompanyType node shared across all data items (deterministic ID via identity_fields)
    company_type = CompanyType()

    for data_item in data:
        people = data_item.people
        companies = data_item.companies

        # Build departments with their employees
        dept_employees: dict[str, list[Person]] = {}
        for person in people:
            dept_name = person["department"]
            if dept_name not in dept_employees:
                dept_employees[dept_name] = []
            dept_employees[dept_name].append(Person(name=person["name"]))

        departments = {
            name: Department(name=name, employees=employees)
            for name, employees in dept_employees.items()
        }

        for company in companies:
            company_departments = [
                departments.get(dept_name, Department(name=dept_name, employees=[]))
                for dept_name in company["departments"]
            ]
            all_companies.append(
                Company(name=company["name"], departments=company_departments, is_type=company_type)
            )

    return all_companies

This is the first pipeline task, and it is ordinary Python: group people by their department field, build one Department per group, then hang the departments each company declares off a Company. Nothing here checks whether a node already exists. A company that declares a department nobody works in still gets a Department node — created on the spot with no employees — and if another company declares the same one, identity_fields gives both objects the same UUID, so the two collapse into a single shared node when they are stored. The function returns only the companies; the departments and people reach the graph because they hang off them.

Stage 3: Run the Two-Task Pipeline

Source: examples/demos/custom_pipelines/organizational_hierarchy/organizational_hierarchy_pipeline_example.py

    # Run tasks expects a list of data even if it is just one document
    data = [{"companies": companies, "people": people}]

    pipeline = run_tasks(
        [Task(ingest_files), Task(add_data_points)],
        dataset_id=datasets[0].id,
        data=build_lightweight_data_object(data),
        incremental_loading=False,
    )

main() prunes, calls setup() to create the relational tables, resolves the default user, and creates a test_dataset with load_or_create_datasets before this point — the dataset is what the pipeline run is tracked against. run_tasks then chains the two tasks: ingest_files returns Company objects and add_data_points writes them, their nested departments and people, and the edges between them into the graph and vector stores. build_lightweight_data_object wraps the raw dict in a LightweightData DataPoint with a uuid5 ID, because the pipeline expects DataPoints rather than plain dicts. The run ends with visualize_graph, which writes the org chart to .artifacts/organizational_hierarchy_pipeline_example.html.

Stage 4: Model the Same Graph Without identity_fields

Source: examples/demos/custom_pipelines/organizational_hierarchy/organizational_hierarchy_pipeline_low_level_example.py

class Person(DataPoint):
    """Represent a person."""

    name: str
    metadata: dict = {"index_fields": ["name"]}


class Department(DataPoint):
    """Represent a department."""

    name: str
    employees: list[Person]
    metadata: dict = {"index_fields": ["name"]}

The low-level variant declares the same four node types, with one difference that shapes the rest of the script: no identity_fields. Names are still embedded, but each constructed instance gets its own random UUID, so two Person(name="John Doe") objects are two nodes. Everything the next stage does by hand exists to prevent that.

Stage 5: Deduplicate People and Departments by Hand

Source: examples/demos/custom_pipelines/organizational_hierarchy/organizational_hierarchy_pipeline_low_level_example.py

def build_people_nodes(people: Iterable[Mapping[str, Any]]) -> dict:
    """Build person nodes keyed by name."""
    nodes = {p["name"]: Person(name=p["name"]) for p in people if p.get("name")}
    return nodes


def group_people_by_department(people: Iterable[Mapping[str, Any]]) -> dict:
    """Group person names by department."""
    groups = defaultdict(list)
    for person in people:
        name = person.get("name")
        if not name:
            continue
        dept = person.get("department", "Unknown")
        groups[dept].append(name)
    return groups

Every node here is created once and then referenced by name from a dict — that keying, not the model, is what makes the graph deduplicate. build_companies chains the helpers in order: collect people and companies from the payloads, build the person and department node dicts, then attach_departments_to_companies and attach_employees_to_departments wire the objects together by looking each name up. remove_duplicates_preserve_order is needed because the same person can be listed twice for a department, and collect_declared_departments unions the departments people work in with the ones companies declare, so a department with no employees still gets a node.

Stage 6: Store the Graph, Index Its Edges, and Query It

Source: examples/demos/custom_pipelines/organizational_hierarchy/organizational_hierarchy_pipeline_low_level_example.py

    # Build and run pipeline
    tasks = [Task(ingest_payloads), Task(add_data_points)]
    pipeline = run_tasks(tasks, dataset_id, None, user, "demo_pipeline")
    async for status in pipeline:
        logger.info("Pipeline status: %s", status)

    # Post-process: index graph edges and visualize
    await index_graph_edges()
    await visualize_graph(str(GRAPH_HTML))

    # Run query against graph
    completion = await search(
        query_text="Who works for GreenFuture Solutions?",
        query_type=SearchType.GRAPH_COMPLETION,
    )

The pipeline is the same two-task shape, run against a demo_dataset with None as the data argument — ingest_payloads falls back to loading the bundled JSON files itself. After the nodes land, index_graph_edges embeds the relationship names so edges are retrievable and not just traversable, and visualize_graph writes the HTML render. The closing GRAPH_COMPLETION search is the point of the whole exercise: nothing about the query knows the graph was assembled by hand, so asking who works for GreenFuture Solutions walks company → department → employee edges and answers from them.

Run It

uv run python examples/demos/custom_pipelines/organizational_hierarchy/organizational_hierarchy_pipeline_example.py
uv run python examples/demos/custom_pipelines/organizational_hierarchy/organizational_hierarchy_pipeline_low_level_example.py

The high-level run is the one walked through above; it closes by logging the path of the render it wrote to .artifacts/organizational_hierarchy_pipeline_example.html, next to the script — open that file to see companies linked to departments linked to people. The low-level run logs its pipeline statuses through logging as Pipeline status: ... lines, writes .artifacts/graph_visualization.html, and ends by logging Graph completion result: ... with the LLM’s answer to “Who works for GreenFuture Solutions?” — the proof that the hand-built graph is queryable.

Choosing Between the Two Variants

Start from the high-level script. Declaring identity_fields moves deduplication into the node’s identity, so the mapping code is free to construct objects wherever it is convenient and the store collapses the repeats — the same property that makes re-running the import idempotent rather than doubling the graph. The low-level script is worth reading when you need to see what that buys you: it is the same org chart, with keyed dicts, ordered-unique helpers, and attach passes standing in for the deterministic IDs. Reach for its shape when identity is not a field you have — when nodes are distinguished by something you compute rather than something in the data — and for index_graph_edges, which either variant can call once the nodes are stored.

DataPoints

How index_fields, identity_fields, and nested models shape the graph a DataPoint becomes.

Custom Data Models

More on modeling your own node types and storing them with add_data_points.

Custom Tasks and Pipelines

Writing your own tasks and running them as a pipeline, step by step.

Reading the Visualization

What each tab of the rendered org chart shows, and which one to reach for.

Getting Started

Core Concepts

Setup Configuration

Guides

Examples

CLI

OSS

What You’ll Build

Features in Play

How It Looks

What to Expect

Before You Start

How It Works

Stage 1: Model the Org Chart as Deduplicating DataPoints

Stage 2: Map the JSON Rows onto Nested DataPoints

Stage 3: Run the Two-Task Pipeline

Stage 4: Model the Same Graph Without identity_fields

Stage 5: Deduplicate People and Departments by Hand

Stage 6: Store the Graph, Index Its Edges, and Query It

Run It

Choosing Between the Two Variants

DataPoints

Custom Data Models

Custom Tasks and Pipelines

Reading the Visualization

​What You’ll Build

​Features in Play

​How It Looks

​What to Expect

​Before You Start

​How It Works

​Stage 1: Model the Org Chart as Deduplicating DataPoints

​Stage 2: Map the JSON Rows onto Nested DataPoints

​Stage 3: Run the Two-Task Pipeline

​Stage 4: Model the Same Graph Without identity_fields

​Stage 5: Deduplicate People and Departments by Hand

​Stage 6: Store the Graph, Index Its Edges, and Query It

​Run It

​Choosing Between the Two Variants

DataPoints

Custom Data Models

Custom Tasks and Pipelines

Reading the Visualization

What You’ll Build

Features in Play

How It Looks

What to Expect

Before You Start

How It Works

Stage 1: Model the Org Chart as Deduplicating DataPoints

Stage 2: Map the JSON Rows onto Nested DataPoints

Stage 3: Run the Two-Task Pipeline

Stage 4: Model the Same Graph Without identity_fields

Stage 5: Deduplicate People and Departments by Hand

Stage 6: Store the Graph, Index Its Edges, and Query It

Run It

Choosing Between the Two Variants