Score Project Quality with GAT

Train a graph attention network to predict project-quality rubric traits.

Introduction

30 Second Summary

Quality reviews get harder when every project must be judged across several traits. A useful score should reflect the structure of the project as well as the words inside each block.

In this project, you will build a scoring notebook in Google Colab that converts structured NextWork project JSON files into graphs. The notebook trains a graph attention network that predicts several rubric-trait scores for unseen projects.

What You'll Build

Your finished notebook prints its training progress before placing predicted rubric scores beside human scores for held-out projects.

By the end of this project, you'll have:

  • A validated training corpus with every score row matched to one project file. Every rubric trait is confirmed as numeric.
  • A before-and-after graph audit that first exposes isolated nodes. The repaired audit shows edges with zero isolated nodes.
  • A multi-trait scoring run that reports held-out error for each rubric trait. It saves a reusable checkpoint. It also writes a CSV prediction report to Google Drive.
  • Secret Mission: Rank the project blocks receiving the largest attention weights for one rubric trait.

Are there any prerequisites?

You should be comfortable reading Python. Basic model-training code should also feel familiar.

The guide sets up PyTorch Geometric inside Google Colab.

Before We Start

Before any hands-on work, commit to the quality decisions this scorer could support. You will also explain why a graph can preserve the relationships inside structured JSON project blocks.

Set Up the Colab Scoring Workspace

Your scoring notebook needs to survive Google Colab runtime resets. Google Drive gives the notebook a persistent home for its data plus saved outputs.

The runtime also needs PyTorch Geometric before it can create graph objects. You will finish with an import check for Data plus GATConv.

In this step, get ready to:
  • Connect a Colab notebook to the Drive containing your project data.
  • Organize the Drive-backed scoring workspace.
  • Verify a pinned PyTorch Geometric runtime.
Connect the notebook to Google Drive

Mounting Drive gives notebook code access to the files stored there. This keeps your project data available when Colab replaces its temporary runtime.

  • Create a blank notebook from the Google Colab landing page.
  • Select Mount Drive from the notebook's file sidebar.

The next prompt grants notebook code access to the Drive you choose. Keep that access limited to the Google account containing your project data.

  • Choose the Google account that contains your raw project files.
  • Approve the Drive access request.
  • Expand the mounted Drive area in the notebook's file sidebar.

You should see the Drive area containing your raw project files plus rubric scores.

Why Mount Drive?

Colab can delete its temporary virtual machine after inactivity. Files stored in Drive remain available for your next notebook session.

Installed runtime packages are temporary. Your notebook cells provide the repeatable setup needed after a reset.

Organize the Drive-backed workspace

Your Drive project folder needs a predictable layout. The notebook reads its inputs from this layout before writing trained artifacts back into it.

  • Select the Drive folder that contains your project data.
  • Create a raw_json/ folder with the folder-creation control.
  • Move your raw project JSON files into raw_json/.
  • Create an outputs/ folder with the folder-creation control.
  • Move scores.csv into the selected Drive project folder.
  • Confirm that raw_json/, scores.csv, plus outputs/ appear at the same level.

What Does Each Location Hold?

  • The raw_json/ folder holds one JSON file per project.
  • The scores.csv file reserves its first column for project_id.
  • The remaining scores.csv columns hold numeric rubric-trait values.
  • The outputs/ folder is reserved for the trained checkpoint plus held-out prediction report.

That is the persistent workspace sorted. Every later notebook section now has a reliable place to read inputs or save results.

Prepare the graph runtime

PyTorch supplies the tensor plus neural-network operations used by the scorer. PyTorch Geometric adds the graph container and graph attention layer.

The setup cell pins PyTorch Geometric to 2.8.1. This makes the notebook setup reproducible after a runtime reset.

  • Add a code cell below the Drive setup.
  • Install the pinned graph library by running this command in the cell:
pip install torch_geometric==2.8.1

What Does This Command Do?

The pip command installs the torch_geometric package into the current Colab runtime. The version pin selects 2.8.1 for this notebook.

You need to rerun this setup cell after Colab gives the notebook a fresh runtime.

  • Wait for the setup cell to finish.
  • Read the final package output to identify the installed version.

✔️ I see version 2.8.1

PyTorch Geometric 2.8.1 is ready in the current runtime.

  • Continue to the import check below.

ⓧ I see an older version

The displayed version does not match the notebook's 2.8.1 pin.

  • Return to the setup cell from earlier.
  • Run the setup cell again to apply the pinned version.
  • Confirm that the completed package output names version 2.8.1.

ⓧ I see command not found

The setup cell needs an active Colab runtime before it can reach the package installer.

  • Reconnect the notebook runtime.
  • Return to the setup cell from earlier.
  • Run the setup cell again.

Installation Still Failing?

A network interruption can stop the package download. Rerun the setup cell after the runtime reconnects.

Make sure you are running the setup cell inside the connected Colab notebook.

Help me diagnose the Colab installation failure.

The import check proves that the current Python process can load the installed graph classes. It also exposes the PyTorch version supplied by the runtime.

Before you run the check, pause on this question: which three values do you expect the cell to print.

  • Add a code cell below the package setup cell.
  • Run this import check in the new code cell:
import torch
from torch_geometric.data import Data
from torch_geometric.nn.conv import GATConv

# Display the runtime version and imported graph classes
print(torch.__version__)
print(Data)
print(GATConv)

What Does This Check Prove?

  • The torch.__version__ output displays the runtime's installed PyTorch version.
  • The Data output confirms that the graph data container imported successfully.
  • The GATConv output confirms that the graph attention layer imported successfully.

You should see a PyTorch version followed by class references for Data plus GATConv. The cell should finish without an import error.

Your graph workspace is ready. The runtime can now create graph data objects plus graph attention layers.

Import Check Not Completing?

Confirm that the package setup cell completed in the same active runtime. A fresh runtime needs the package installed again.

Check that each import matches the capitalization shown in the reference below.

Help me fix the import check.

✔️ Awesome, I've got everything!

Your setup cell plus import-check cell cover the complete runtime setup.

ⓧ I'd like to double check the full code

Compare your two notebook cells with this complete setup reference.

# Setup cell
pip install torch_geometric==2.8.1

# Import-check cell
import torch
from torch_geometric.data import Data
from torch_geometric.nn.conv import GATConv

# Display the runtime version and imported graph classes
print(torch.__version__)
print(Data)
print(GATConv)

How Should the Cells Be Split?

The package command belongs in the setup cell. The imports plus print statements belong in the import-check cell below it.

Your Drive-backed scoring workspace is ready. Next up, you will load the project JSON files plus rubric-trait scores into the notebook.

Load and Validate the Training Corpus

Your Google Colab workspace can now use the graph library from the previous step. The next challenge is proving that every score row points to one real project in Google Drive.

This step builds that data gate before any modeling begins. It also checks that every rubric trait contains a numeric target the model can learn from.

In this step, get ready to:
  • Arrange the project files around a consistent identifier.
  • Add the path configuration and corpus-loading functions.
  • Run the loader to verify the project count and detected traits.
Match score rows to JSON files

Each JSON project needs one corresponding row in the CSV score table. The shared project_id value creates that link.

  • Return to the Drive-backed project folder from the previous step.
  • Move each raw project JSON file into raw_json/.
  • Place scores.csv directly inside the Drive-backed project folder.
  • Open scores.csv with your spreadsheet editor.
  • Set the first column header to project_id.
  • Add at least one numeric rubric-trait column after project_id.
  • Match every project_id value to one filename stem under raw_json/.
  • Keep at least five labeled projects in the corpus.

How does the file matching work?

A score row with a project_id value of project-alpha resolves to raw_json/project-alpha.json. The filename stem is the portion before .json.

Every column after project_id becomes a trait automatically. This keeps the loader compatible with different rubrics.

  • Locate the Drive-backed project folder in the Colab file browser.
  • Copy the folder's mounted path from its context menu.
  • Record the path here: your mounted Drive project folder path.
Add corpus configuration and loaders

The setup code has three responsibilities. It imports the required modules, defines the shared configuration, and rejects invalid corpus data before training starts.

  • Create a new code cell below the import check from the previous step.
  • Add the project imports by pasting this code:
import csv
import hashlib
import json
import math
import random
import re
from pathlib import Path

import torch
import torch.nn.functional as F
from torch_geometric.data import Data
from torch_geometric.nn.conv import GATConv

What do these imports prepare?

  • The standard-library modules support CSV loading, JSON parsing, deterministic hashing, numeric operations, reproducible splitting, text matching, and file paths.
  • PyTorch provides tensors, neural-network functions, and the training tools used later.
  • PyTorch Geometric provides the graph container and graph attention layer used in later steps.
  • Run the imports cell.

The cell should show a completed status without a traceback.

Seeing an import problem?

Switch back to the installation cell from the previous step. Rerun that cell before retrying these imports.

Check that the current runtime is still connected. A new runtime needs the package installation restored.

Help me diagnose the failed imports in my Colab notebook.

  • Add another code cell below the imports cell.
  • Paste the Drive path configuration into the new cell:
DATA_ROOT = Path("your-drive-project-folder-path")
RAW_DIR = DATA_ROOT / "raw_json"
SCORES_PATH = DATA_ROOT / "scores.csv"
OUTPUT_DIR = DATA_ROOT / "outputs"

What do these paths control?

  • DATA_ROOT identifies the Drive-backed folder for the whole project.
  • RAW_DIR points to the folder containing one JSON file per project.
  • SCORES_PATH identifies the rubric score table.
  • OUTPUT_DIR reserves the destination for the checkpoint and prediction report created later.
  • Replace your-drive-project-folder-path with your mounted Drive project folder path.
  • Run the path configuration cell.

The cell should finish with a completed status. Your four path variables now point into the mounted project folder.

Does the path look wrong?

Confirm that DATA_ROOT names the folder containing raw_json/ and scores.csv. Keep the path inside the existing quotation marks.

Help me check my DATA_ROOT path.

  • Add another code cell below the path configuration.
  • Define the model constants and feature names by pasting this code:
SEED = 17
TEST_FRACTION = 0.20
EPOCHS = 60
TOKEN_BUCKETS = 16
HIDDEN_CHANNELS = 16
ATTENTION_HEADS = 4
DROPOUT = 0.10
LEARNING_RATE = 0.01

FEATURE_NAMES = [
    "is_object",
    "is_array",
    "is_text",
    "is_number",
    "is_bool_or_null",
    "is_trait",
    *[f"token_bucket_{index}" for index in range(TOKEN_BUCKETS)],
    "log_character_count",
    "log_word_count",
    "depth",
    "sibling_position",
]

Why define these values now?

The constants keep data splitting, feature construction, and model training reproducible across the notebook. Later sections reuse the same names instead of scattering configuration values through the code.

FEATURE_NAMES records the exact order of every node feature. That order must remain stable when the graph tensors are built.

  • Run the constants cell.

The cell should complete successfully with FEATURE_NAMES available for the graph-building code.

Seeing a syntax problem in the feature list?

Check the commas after each feature name. Confirm that the closing square bracket appears after sibling_position.

Help me compare my FEATURE_NAMES definition with the project code.

The score-table loader enforces the corpus contract at the CSV boundary. It rejects missing identifiers, repeated projects, absent traits, and non-numeric target values.

  • Add another code cell below the constants cell.
  • Define the score-table loader by pasting this function:
def load_score_table(scores_path):
    with scores_path.open("r", encoding="utf-8", newline="") as handle:
        reader = csv.DictReader(handle)
        if not reader.fieldnames or reader.fieldnames[0] != "project_id":
            raise ValueError("scores.csv must begin with a project_id column")

        trait_names = reader.fieldnames[1:]
        if not trait_names:
            raise ValueError("scores.csv needs at least one rubric-trait column")

        rows = []
        seen_ids = set()
        for row_number, row in enumerate(reader, start=2):
            project_id = row["project_id"].strip()
            if not project_id:
                raise ValueError(f"Missing project_id on CSV row {row_number}")
            if project_id in seen_ids:
                raise ValueError(f"Duplicate project_id: {project_id}")

            try:
                targets = [float(row[trait]) for trait in trait_names]
            except (TypeError, ValueError) as error:
                raise ValueError(
                    f"Every trait must be numeric on CSV row {row_number}"
                ) from error

            rows.append((project_id, targets))
            seen_ids.add(project_id)

    return trait_names, rows

What does the score loader validate?

  • The header check reserves the first column for project_id.
  • The trait check requires at least one scoring target after the identifier.
  • The row checks reject blank or repeated project identifiers.
  • The numeric conversion turns every trait value into a floating-point target.
  • Run the score-loader cell.

The cell should complete with load_score_table defined for the final corpus check.

Does the function cell stop with a syntax issue?

Check the indentation inside the with block and the for loop. Python uses indentation to decide which checks belong to each block.

Help me fix the score-loader function.

The record loader joins the validated score rows to the raw project files. Each returned record holds a project identifier, parsed project data, and numeric targets.

  • Add another code cell below the score-loader cell.
  • Define the record loader by pasting this function:
def load_records(raw_dir, scores_path):
    if not raw_dir.is_dir():
        raise FileNotFoundError(f"Missing raw JSON directory: {raw_dir}")
    if not scores_path.is_file():
        raise FileNotFoundError(f"Missing score file: {scores_path}")

    trait_names, score_rows = load_score_table(scores_path)
    records = []

    for project_id, targets in score_rows:
        json_path = raw_dir / f"{project_id}.json"
        if not json_path.is_file():
            raise FileNotFoundError(
                f"No JSON file matches project_id {project_id}: {json_path}"
            )
        with json_path.open("r", encoding="utf-8") as handle:
            raw_project = json.load(handle)
        records.append((project_id, raw_project, targets))

    if len(records) < 5:
        raise ValueError("Use at least five labeled projects for this holdout prototype")

    return trait_names, records

What does the record loader produce?

The function verifies that both input locations exist before reading data. It then constructs one JSON path from each score-table identifier.

Each item in records is a tuple containing project_id, raw_project, and targets.

  • Run the record-loader cell.

The cell should complete with load_records ready to validate the corpus.

Having trouble defining the record loader?

Confirm that the entire file-reading section stays inside the for loop. Check that the final return aligns with the first validation checks.

Help me fix the record-loader function.

✔️ Awesome, I've got everything!

Your notebook now contains the imports, paths, constants, feature names, and both corpus-loading functions.

ⓧ I'd like to double check the full code

  • Compare your setup cells with this full reference.
  • Keep your mounted Drive project folder path in DATA_ROOT.
import csv
import hashlib
import json
import math
import random
import re
from pathlib import Path

import torch
import torch.nn.functional as F
from torch_geometric.data import Data
from torch_geometric.nn.conv import GATConv

DATA_ROOT = Path("your-drive-project-folder-path")
RAW_DIR = DATA_ROOT / "raw_json"
SCORES_PATH = DATA_ROOT / "scores.csv"
OUTPUT_DIR = DATA_ROOT / "outputs"

SEED = 17
TEST_FRACTION = 0.20
EPOCHS = 60
TOKEN_BUCKETS = 16
HIDDEN_CHANNELS = 16
ATTENTION_HEADS = 4
DROPOUT = 0.10
LEARNING_RATE = 0.01

FEATURE_NAMES = [
    "is_object",
    "is_array",
    "is_text",
    "is_number",
    "is_bool_or_null",
    "is_trait",
    *[f"token_bucket_{index}" for index in range(TOKEN_BUCKETS)],
    "log_character_count",
    "log_word_count",
    "depth",
    "sibling_position",
]


def load_score_table(scores_path):
    with scores_path.open("r", encoding="utf-8", newline="") as handle:
        reader = csv.DictReader(handle)
        if not reader.fieldnames or reader.fieldnames[0] != "project_id":
            raise ValueError("scores.csv must begin with a project_id column")

        trait_names = reader.fieldnames[1:]
        if not trait_names:
            raise ValueError("scores.csv needs at least one rubric-trait column")

        rows = []
        seen_ids = set()
        for row_number, row in enumerate(reader, start=2):
            project_id = row["project_id"].strip()
            if not project_id:
                raise ValueError(f"Missing project_id on CSV row {row_number}")
            if project_id in seen_ids:
                raise ValueError(f"Duplicate project_id: {project_id}")

            try:
                targets = [float(row[trait]) for trait in trait_names]
            except (TypeError, ValueError) as error:
                raise ValueError(
                    f"Every trait must be numeric on CSV row {row_number}"
                ) from error

            rows.append((project_id, targets))
            seen_ids.add(project_id)

    return trait_names, rows


def load_records(raw_dir, scores_path):
    if not raw_dir.is_dir():
        raise FileNotFoundError(f"Missing raw JSON directory: {raw_dir}")
    if not scores_path.is_file():
        raise FileNotFoundError(f"Missing score file: {scores_path}")

    trait_names, score_rows = load_score_table(scores_path)
    records = []

    for project_id, targets in score_rows:
        json_path = raw_dir / f"{project_id}.json"
        if not json_path.is_file():
            raise FileNotFoundError(
                f"No JSON file matches project_id {project_id}: {json_path}"
            )
        with json_path.open("r", encoding="utf-8") as handle:
            raw_project = json.load(handle)
        records.append((project_id, raw_project, targets))

    if len(records) < 5:
        raise ValueError("Use at least five labeled projects for this holdout prototype")

    return trait_names, records

How should you use this reference?

This block combines the setup code created in this step. The reference keeps the original path placeholder visible, so your own DATA_ROOT line should contain the mounted path you recorded earlier.

Run the corpus validation

The loader now has enough information to test the whole data contract. A successful run proves that every score row resolves to one parseable project with numeric targets.

Before you run the check, do you expect the loaded project count to match the number of score rows?

  • Add a final code cell below the loader functions.
  • Run the corpus validation by pasting this code:
trait_names, records = load_records(RAW_DIR, SCORES_PATH)
print(f"Loaded {len(records)} projects with traits: {trait_names}")
print(f"First project identifier: {records[0][0]}")

What does this check prove?

The first line calls both loaders and stores their results. The next two lines expose the corpus size, dynamically detected traits, and first matched identifier.

A successful result means records contains one tuple for every validated score row. The graph-building step can now consume those tuples.

You should see a project count that matches the rows in scores.csv. You should also see the rubric columns followed by the first project identifier.

Did the corpus validation stop?

A missing-directory or missing-file message points to the DATA_ROOT value or folder layout. Confirm that raw_json/ and scores.csv sit under the same project folder.

A project-matching message means one project_id lacks a JSON file with the same stem. A numeric-trait message means the reported CSV row contains a value that cannot become a number.

Help me diagnose my corpus validation result.

That is the data gate cleared. Your notebook can now identify every project and every rubric target before graph construction begins.

Your training corpus is matched and validated. Next up, you will turn one project into nodes and expose why isolated nodes cannot support message passing.

Expose the Disconnected-Graph Problem

Your training records now pair each project JSON file with its numeric rubric targets. The next challenge is preserving the structure inside each project.

A collection of nodes has no message-passing routes until edges connect it. This step deliberately builds the zero-edge version so you can see that limitation in the audit.

In this step, get ready to:
  • Convert JSON values and rubric traits into graph nodes with numeric features.
  • Build the first project as a deliberately disconnected graph.
  • Audit the graph to prove that every node is isolated.
Create stable node features

A node feature vector gives every JSON value a numeric representation. Stable token hashing keeps the same token in the same feature bucket across notebook runs.

  • Switch back to project_quality_gat.py in the Colab editor.
  • Place the node-feature helpers below load_records by copying this code:
def node_type_vector(value, is_trait=False):
    if is_trait:
        return [0.0, 0.0, 0.0, 0.0, 0.0, 1.0]
    if isinstance(value, dict):
        return [1.0, 0.0, 0.0, 0.0, 0.0, 0.0]
    if isinstance(value, list):
        return [0.0, 1.0, 0.0, 0.0, 0.0, 0.0]
    if isinstance(value, str):
        return [0.0, 0.0, 1.0, 0.0, 0.0, 0.0]
    if isinstance(value, bool) or value is None:
        return [0.0, 0.0, 0.0, 0.0, 1.0, 0.0]
    if isinstance(value, (int, float)):
        return [0.0, 0.0, 0.0, 1.0, 0.0, 0.0]
    return [0.0, 0.0, 1.0, 0.0, 0.0, 0.0]


def scalar_text(value):
    if isinstance(value, (dict, list)) or value is None:
        return ""
    return str(value)


def token_bucket_features(text):
    buckets = [0.0] * TOKEN_BUCKETS
    tokens = re.findall(r"[A-Za-z0-9_]+", text.lower())
    for token in tokens:
        digest = hashlib.sha256(token.encode("utf-8")).digest()
        bucket = int.from_bytes(digest[:4], "big") % TOKEN_BUCKETS
        buckets[bucket] += 1.0

    total = sum(buckets)
    if total:
        buckets = [value / total for value in buckets]
    return buckets


def make_node_features(value, label, depth, position, is_trait=False):
    text = label if is_trait else f"{label} {scalar_text(value)}".strip()
    word_count = len(re.findall(r"[A-Za-z0-9_]+", text))
    return [
        *node_type_vector(value, is_trait=is_trait),
        *token_bucket_features(text),
        math.log1p(len(text)) / 10.0,
        math.log1p(word_count) / 5.0,
        min(depth, 20) / 20.0,
        min(position, 20) / 20.0,
    ]


def make_node_label(value, label, is_trait=False):
    if is_trait:
        return f"trait:{label}"
    preview = scalar_text(value).replace("\n", " ")[:40]
    return f"{label}:{preview}" if preview else label

What do these helpers create?

  • The node_type_vector() helper identifies objects, arrays, text, numbers, Boolean or null values, and trait nodes.
  • The token_bucket_features() helper maps matching tokens into stable SHA-256 buckets.
  • The make_node_features() helper adds text size, word count, JSON depth, and sibling position.
  • The make_node_label() helper stores readable labels for later inspection.

The feature vector captures node type and lightweight text signals. It also records where each value sits in the JSON hierarchy.

  • Place the recursive graph builder directly below make_node_label() by copying this code:
def build_graph(project_id, raw_project, targets, trait_names, connected):
    features = []
    node_labels = []
    edges = []

    def add_edge(source, target):
        edges.append((source, target))

    def visit(value, label, depth, position, parent_id=None):
        node_id = len(features)
        features.append(make_node_features(value, label, depth, position))
        node_labels.append(make_node_label(value, label))

        if connected and parent_id is not None:
            add_edge(parent_id, node_id)
            add_edge(node_id, parent_id)

        if isinstance(value, dict):
            children = [(str(key), child) for key, child in value.items()]
        elif isinstance(value, list):
            children = [(f"{label}[{index}]", child) for index, child in enumerate(value)]
        else:
            children = []

        previous_child_id = None
        for child_position, (child_label, child_value) in enumerate(children):
            child_id = visit(
                child_value,
                child_label,
                depth + 1,
                child_position,
                node_id,
            )
            if connected and previous_child_id is not None:
                add_edge(previous_child_id, child_id)
                add_edge(child_id, previous_child_id)
            previous_child_id = child_id

        return node_id

    visit(raw_project, "project", 0, 0)
    content_node_count = len(features)
    trait_start = content_node_count

    for trait_name in trait_names:
        features.append(
            make_node_features(None, trait_name, 0, 0, is_trait=True)
        )
        node_labels.append(make_node_label(None, trait_name, is_trait=True))

    if connected:
        trait_ids = list(range(trait_start, len(features)))
        for trait_id in trait_ids:
            for content_id in range(content_node_count):
                add_edge(trait_id, content_id)
                add_edge(content_id, trait_id)
            for other_trait_id in trait_ids:
                if trait_id != other_trait_id:
                    add_edge(trait_id, other_trait_id)

    edge_index = torch.tensor(
        [
            [source for source, _ in edges],
            [target for _, target in edges],
        ],
        dtype=torch.long,
    )

    graph = Data(
        x=torch.tensor(features, dtype=torch.float32),
        edge_index=edge_index,
        y=torch.tensor(targets, dtype=torch.float32),
    )
    graph.project_id = project_id
    graph.trait_start = trait_start
    graph.node_labels = node_labels
    return graph

How does the graph builder work?

  • The recursive visit() helper creates one content node for each JSON object, array, and scalar value.
  • The trait loop appends one separate node for every dynamically detected rubric column.
  • The trait_start value records where the trait nodes begin.
  • The connected argument controls whether the builder adds relationships between nodes.
  • The returned Data object stores features in x and targets in y.
Build the disconnected graph

The first build uses connected=False to leave every node isolated. The readable labels let you confirm that the traversal created nodes from real project content.

  • Save project_quality_gat.py.
  • Add the first-record graph build below the loading output by copying this code:
first_project_id, first_raw_project, first_targets = records[0]
naive_graph = build_graph(
    first_project_id,
    first_raw_project,
    first_targets,
    trait_names,
    connected=False,
)
print("First five node labels:", naive_graph.node_labels[:5])

What does this code produce?

  • The tuple unpacking selects the identifier, raw JSON, and trait targets from the first validated record.
  • The build_graph() call converts that record into naive_graph.
  • The printed preview exposes the first five JSON-derived labels without dumping the entire graph.
  • Run the updated loading section in the notebook.

You should see five labels drawn from the first project's JSON structure. This confirms that content nodes and trait nodes now exist inside naive_graph.

Missing node labels?

Check that build_graph() appears before the code that creates naive_graph.

Confirm that the loading section still creates records before selecting its first entry.

Help me debug missing node labels in my graph build.

Audit the missing message paths

A graph audit counts edges and connected components. These measurements reveal whether information can travel between the content and trait nodes.

  • Place the audit function directly below build_graph() by copying this code:
def graph_audit(graph):
    node_count = int(graph.num_nodes)
    edge_count = int(graph.num_edges)
    neighbors = [set() for _ in range(node_count)]

    for edge_position in range(edge_count):
        source = int(graph.edge_index[0, edge_position].item())
        target = int(graph.edge_index[1, edge_position].item())
        neighbors[source].add(target)
        neighbors[target].add(source)

    isolated = sum(1 for adjacent in neighbors if not adjacent)
    visited = set()
    components = 0

    for start in range(node_count):
        if start in visited:
            continue
        components += 1
        stack = [start]
        visited.add(start)
        while stack:
            current = stack.pop()
            for neighbor in neighbors[current]:
                if neighbor not in visited:
                    visited.add(neighbor)
                    stack.append(neighbor)

    return {
        "nodes": node_count,
        "edges": edge_count,
        "isolated_nodes": isolated,
        "connected_components": components,
    }

What does the audit measure?

  • The node and edge counts describe the graph's basic size.
  • The neighbor sets reveal which nodes have no relationships.
  • The traversal counts how many separate connected components exist.
  • A graph with one component per node has no route for message passing.
  • Save project_quality_gat.py.

✔️ Awesome, I've got everything!

Your feature helpers, graph builder, and audit function are now saved in project_quality_gat.py.

ⓧ I'd like to double check the full code

Compare the graph section below with the code after load_records(). Keep the existing DATA_ROOT value that points to your mounted Drive folder.

def node_type_vector(value, is_trait=False):
    if is_trait:
        return [0.0, 0.0, 0.0, 0.0, 0.0, 1.0]
    if isinstance(value, dict):
        return [1.0, 0.0, 0.0, 0.0, 0.0, 0.0]
    if isinstance(value, list):
        return [0.0, 1.0, 0.0, 0.0, 0.0, 0.0]
    if isinstance(value, str):
        return [0.0, 0.0, 1.0, 0.0, 0.0, 0.0]
    if isinstance(value, bool) or value is None:
        return [0.0, 0.0, 0.0, 0.0, 1.0, 0.0]
    if isinstance(value, (int, float)):
        return [0.0, 0.0, 0.0, 1.0, 0.0, 0.0]
    return [0.0, 0.0, 1.0, 0.0, 0.0, 0.0]


def scalar_text(value):
    if isinstance(value, (dict, list)) or value is None:
        return ""
    return str(value)


def token_bucket_features(text):
    buckets = [0.0] * TOKEN_BUCKETS
    tokens = re.findall(r"[A-Za-z0-9_]+", text.lower())
    for token in tokens:
        digest = hashlib.sha256(token.encode("utf-8")).digest()
        bucket = int.from_bytes(digest[:4], "big") % TOKEN_BUCKETS
        buckets[bucket] += 1.0

    total = sum(buckets)
    if total:
        buckets = [value / total for value in buckets]
    return buckets


def make_node_features(value, label, depth, position, is_trait=False):
    text = label if is_trait else f"{label} {scalar_text(value)}".strip()
    word_count = len(re.findall(r"[A-Za-z0-9_]+", text))
    return [
        *node_type_vector(value, is_trait=is_trait),
        *token_bucket_features(text),
        math.log1p(len(text)) / 10.0,
        math.log1p(word_count) / 5.0,
        min(depth, 20) / 20.0,
        min(position, 20) / 20.0,
    ]


def make_node_label(value, label, is_trait=False):
    if is_trait:
        return f"trait:{label}"
    preview = scalar_text(value).replace("\n", " ")[:40]
    return f"{label}:{preview}" if preview else label


def build_graph(project_id, raw_project, targets, trait_names, connected):
    features = []
    node_labels = []
    edges = []

    def add_edge(source, target):
        edges.append((source, target))

    def visit(value, label, depth, position, parent_id=None):
        node_id = len(features)
        features.append(make_node_features(value, label, depth, position))
        node_labels.append(make_node_label(value, label))

        if connected and parent_id is not None:
            add_edge(parent_id, node_id)
            add_edge(node_id, parent_id)

        if isinstance(value, dict):
            children = [(str(key), child) for key, child in value.items()]
        elif isinstance(value, list):
            children = [(f"{label}[{index}]", child) for index, child in enumerate(value)]
        else:
            children = []

        previous_child_id = None
        for child_position, (child_label, child_value) in enumerate(children):
            child_id = visit(
                child_value,
                child_label,
                depth + 1,
                child_position,
                node_id,
            )
            if connected and previous_child_id is not None:
                add_edge(previous_child_id, child_id)
                add_edge(child_id, previous_child_id)
            previous_child_id = child_id

        return node_id

    visit(raw_project, "project", 0, 0)
    content_node_count = len(features)
    trait_start = content_node_count

    for trait_name in trait_names:
        features.append(
            make_node_features(None, trait_name, 0, 0, is_trait=True)
        )
        node_labels.append(make_node_label(None, trait_name, is_trait=True))

    if connected:
        trait_ids = list(range(trait_start, len(features)))
        for trait_id in trait_ids:
            for content_id in range(content_node_count):
                add_edge(trait_id, content_id)
                add_edge(content_id, trait_id)
            for other_trait_id in trait_ids:
                if trait_id != other_trait_id:
                    add_edge(trait_id, other_trait_id)

    edge_index = torch.tensor(
        [
            [source for source, _ in edges],
            [target for _, target in edges],
        ],
        dtype=torch.long,
    )

    graph = Data(
        x=torch.tensor(features, dtype=torch.float32),
        edge_index=edge_index,
        y=torch.tensor(targets, dtype=torch.float32),
    )
    graph.project_id = project_id
    graph.trait_start = trait_start
    graph.node_labels = node_labels
    return graph


def graph_audit(graph):
    node_count = int(graph.num_nodes)
    edge_count = int(graph.num_edges)
    neighbors = [set() for _ in range(node_count)]

    for edge_position in range(edge_count):
        source = int(graph.edge_index[0, edge_position].item())
        target = int(graph.edge_index[1, edge_position].item())
        neighbors[source].add(target)
        neighbors[target].add(source)

    isolated = sum(1 for adjacent in neighbors if not adjacent)
    visited = set()
    components = 0

    for start in range(node_count):
        if start in visited:
            continue
        components += 1
        stack = [start]
        visited.add(start)
        while stack:
            current = stack.pop()
            for neighbor in neighbors[current]:
                if neighbor not in visited:
                    visited.add(neighbor)
                    stack.append(neighbor)

    return {
        "nodes": node_count,
        "edges": edge_count,
        "isolated_nodes": isolated,
        "connected_components": components,
    }

What should match?

Your graph section should define all seven helper functions in this order. The final function should return node, edge, isolation, and component counts.

Before you run the audit, do you expect any node to have a route to another node?

  • Reveal the disconnected structure by running this audit output:
print("Naive graph audit:", graph_audit(naive_graph))

What does the result prove?

You should see edges equal to 0. The isolated_nodes count should equal the nodes count.

The connected_components count should also equal the number of nodes. Every content node and trait node occupies its own component.

You found the intended gap. The JSON traversal creates useful nodes, while the zero-edge graph prevents every round of message passing.

Audit output looks different?

If the edge count is positive, confirm that the first build_graph() call uses connected=False.

If the node count is zero, confirm that records[0] contains the raw project loaded in the previous step.

Help me compare my naive graph audit with the expected disconnected result.

Your node representation is working, and the audit has exposed the missing relationships. Next up, you will connect project structure and rubric traits so information can travel across the graph.

Connect Project Structure and Traits

Your zero-edge audit exposed the exact limitation in the naive graph. Every content node exists, but none of them can share project context.

A graph attention network moves information along edges. This step activates structural paths between JSON nodes. It also connects the rubric traits to the project content.

In this step, get ready to:
  • Trace the parent-child and adjacent-sibling edge logic.
  • Trace the content-trait and trait-trait edge logic.
  • Build every project as a connected graph and verify the result.
Trace structural edge logic

Parent-child edges preserve the nesting of each JSON project. Adjacent-sibling edges preserve the local relationship between neighboring entries.

  • Return to the notebook cell containing build_graph().
  • Find the if connected and parent_id is not None: condition inside visit().
  • Locate add_edge(parent_id, node_id) for the path from a parent to its child.
  • Locate add_edge(node_id, parent_id) for the path from a child to its parent.
  • Find the if connected and previous_child_id is not None: condition.
  • Locate add_edge(previous_child_id, child_id) for the path to the next sibling.
  • Locate add_edge(child_id, previous_child_id) for the path back to the previous sibling.

Why are the structural edges bidirectional?

A bidirectional pair gives both connected nodes an incoming route for message passing. Information can travel down the JSON hierarchy or back toward the project node.

Parent-child pairs carry hierarchy. Sibling pairs let neighboring JSON entries share local context.

Trace trait edge logic

Each rubric trait has its own node after the content nodes. These nodes need direct access to the project content before the model can produce a distinct score for each trait.

  • Continue to the if connected: block after the trait-node loop.
  • Find the trait_ids range that begins at trait_start.
  • Trace add_edge(trait_id, content_id) from each trait node to every content node.
  • Trace add_edge(content_id, trait_id) from every content node back to each trait node.
  • Find the if trait_id != other_trait_id: condition.
  • Locate add_edge(trait_id, other_trait_id) for each ordered pair of distinct trait nodes.

Why give traits their own nodes?

Each trait node becomes the location where the model later reads one normalized score. Content-trait edges deliver information from the complete project to that location.

Trait-trait edges let rubric dimensions exchange context. The ordered-pair loop gives every trait a route to every different trait.

Build the connected corpus

The edge logic stayed inactive while connected=False created the naive graph. Passing connected=True now activates every structural path and trait relationship.

  • In main(), add the connected corpus directly below print("Naive graph audit:", graph_audit(naive_graph)) by copying this code:
    graphs = [
        build_graph(
            project_id,
            raw_project,
            targets,
            trait_names,
            connected=True,
        )
        for project_id, raw_project, targets in records
    ]
    print("Connected graph audit:", graph_audit(graphs[0]))

What does this code do?

  • The graphs list receives one connected graph for every validated record.
  • The connected=True argument activates all four edge relationships inside build_graph().
  • The final line audits the first connected graph so you can compare it with naive_graph.

Before you rerun the notebook, what should happen to the edge count, isolated-node count, and connected-component count?

  • Rerun the notebook cells through the connected graph audit.

You should see Connected graph audit: followed by a positive edges value. You should also see isolated_nodes set to 0.

The connected_components value should be 1. That result proves every content node and trait node belongs to one connected graph.

Still seeing isolated nodes?

  • Check that the graph-building call uses connected=True.
  • Rerun the cell that defines build_graph() before rebuilding graphs.
  • Confirm that the audit targets graphs[0] instead of naive_graph.

Help me diagnose why graph_audit(graphs[0]) still reports isolated nodes.

✔️ Awesome, I've got everything!

Your connected graph cell now builds the full corpus and audits the first graph.

ⓧ I'd like to double check the full code

    graphs = [
        build_graph(
            project_id,
            raw_project,
            targets,
            trait_names,
            connected=True,
        )
        for project_id, raw_project, targets in records
    ]
    print("Connected graph audit:", graph_audit(graphs[0]))

What should this section contain?

This is the complete connected-corpus addition for this step. It builds every graph with the active edge logic before auditing the first result.

You have repaired the disconnected corpus. Next up, you will train the attention model to predict every rubric trait.

Train and Evaluate the Multi-Trait GAT

Your project records now form connected graphs that can carry information between content blocks and rubric traits. The next goal is to train a Graph Attention Network that returns a separate score for every trait node.

Trait columns can use different numeric ranges. This step uses target normalization to give each trait a comparable training scale. The statistics come only from training projects so the held-out projects cannot leak information into the model.

In this step, get ready to:
  • Split the connected graphs deterministically and normalize every trait target.
  • Train a two-layer GAT that produces one prediction per trait node.
  • Evaluate held-out predictions and save a checkpoint plus a CSV report.
Split and normalize the targets

A held-out split tests the model on projects it never trains against. A fixed seed keeps the selected projects reproducible each time you run the script.

  • In project_quality_gat.py, find the end of graph_audit().
  • Add the deterministic graph split below it by copying this code:
def split_graphs(graphs):
    indices = list(range(len(graphs)))
    random.Random(SEED).shuffle(indices)
    test_size = max(1, int(round(len(graphs) * TEST_FRACTION)))
    test_indices = set(indices[:test_size])
    train_graphs = [graph for index, graph in enumerate(graphs) if index not in test_indices]
    test_graphs = [graph for index, graph in enumerate(graphs) if index in test_indices]
    return train_graphs, test_graphs

What does this code do?

  • The shuffled index list decides which projects enter each split without changing the original graph order.
  • The fixed SEED makes that shuffle reproducible.
  • The TEST_FRACTION calculation reserves 20 percent of the projects for evaluation while keeping at least one held-out graph.
  • Add target normalization immediately below split_graphs() by copying this code:
def normalize_targets(train_graphs, all_graphs, trait_count):
    means = []
    standard_deviations = []

    for trait_index in range(trait_count):
        values = [float(graph.y[trait_index].item()) for graph in train_graphs]
        mean = sum(values) / len(values)
        variance = sum((value - mean) ** 2 for value in values) / len(values)
        standard_deviation = math.sqrt(variance)
        if standard_deviation == 0.0:
            standard_deviation = 1.0
        means.append(mean)
        standard_deviations.append(standard_deviation)

    means_tensor = torch.tensor(means, dtype=torch.float32)
    stds_tensor = torch.tensor(standard_deviations, dtype=torch.float32)

    for graph in all_graphs:
        graph.y = (graph.y - means_tensor) / stds_tensor

    return means_tensor, stds_tensor

Why use training-only statistics?

The function calculates one mean and standard deviation per trait from train_graphs only. Using held-out scores in these calculations would let evaluation data influence training.

A standard deviation of zero changes to 1.0 so a constant trait does not cause division by zero. Every graph then stores normalized targets for model training and evaluation.

  • Save project_quality_gat.py.
  • Run the cells that define the updated functions.
  • Predict whether the two split counts will add up to the total number of connected graphs.
  • Create the split and normalize its targets by running this cell:
train_graphs, test_graphs = split_graphs(graphs)
print(f"Training projects: {len(train_graphs)}")
print(f"Held-out projects: {len(test_graphs)}")

means, stds = normalize_targets(
    train_graphs,
    graphs,
    trait_count=len(trait_names),
)

What should you see?

You will see separate training and held-out project counts. Together they account for every graph in graphs.

Reaching both printed lines without an exception also confirms that every target vector contains the detected number of traits.

Do the split or normalization cells fail?

Confirm that graphs contains the connected graph list from the previous step. Rerun the loading and graph-building cells if the Colab runtime was reset.

Check that trait_names still comes from scores.csv and matches the length of every graph.y tensor.

Help me debug my graph split and target normalization.

Build and train the GAT

The first attention layer creates four views of each node's neighborhood. The second layer combines those views before a linear head converts each trait-node state into one normalized score.

  • Add the model class below normalize_targets() by copying this code:
class MultiTraitGAT(torch.nn.Module):
    def __init__(self, input_channels):
        super().__init__()
        self.gat1 = GATConv(
            input_channels,
            HIDDEN_CHANNELS,
            heads=ATTENTION_HEADS,
            dropout=DROPOUT,
        )
        self.gat2 = GATConv(
            HIDDEN_CHANNELS * ATTENTION_HEADS,
            HIDDEN_CHANNELS,
            heads=1,
            concat=False,
            dropout=DROPOUT,
        )
        self.score_head = torch.nn.Linear(HIDDEN_CHANNELS, 1)

    def forward(self, graph):
        node_states = F.elu(self.gat1(graph.x, graph.edge_index))
        node_states = self.gat2(node_states, graph.edge_index)
        trait_states = node_states[graph.trait_start :]
        return self.score_head(trait_states)[:, 0]

How does the model produce trait scores?

  • The gat1 layer uses four attention heads to gather information across the graph's structural and trait relationships.
  • The gat2 layer returns one hidden state per node with HIDDEN_CHANNELS features.
  • The trait_start slice keeps only trait-node states.
  • The score_head layer converts each retained state into one normalized prediction.
  • Add the training loop immediately below MultiTraitGAT by copying this code:
def train_model(model, train_graphs):
    optimizer = torch.optim.Adam(
        model.parameters(),
        lr=LEARNING_RATE,
        weight_decay=1e-4,
    )

    for epoch in range(1, EPOCHS + 1):
        model.train()
        epoch_graphs = list(train_graphs)
        random.Random(SEED + epoch).shuffle(epoch_graphs)
        total_loss = 0.0

        for graph in epoch_graphs:
            optimizer.zero_grad()
            prediction = model(graph)
            loss = F.mse_loss(prediction, graph.y)
            loss.backward()
            optimizer.step()
            total_loss += float(loss.item())

        if epoch == 1 or epoch % 10 == 0:
            average_loss = total_loss / len(epoch_graphs)
            print(f"Epoch {epoch:02d} | normalized MSE {average_loss:.4f}")

What happens during training?

Adam updates the model after every training graph using mean squared error between normalized predictions and normalized targets. Weight decay adds a small penalty that discourages excessively large parameters.

Each epoch uses a deterministic shuffle based on SEED plus the epoch number. Progress prints at epoch 1 and every 10 epochs through epoch 60.

  • Save project_quality_gat.py.
  • Run the cells that define MultiTraitGAT and train_model().
  • Before training, predict whether the normalized MSE will stay fixed or change as the model updates.
  • Create and train the model by running this cell:
model = MultiTraitGAT(input_channels=len(FEATURE_NAMES))
train_model(model, train_graphs)

What should you see?

The cell prints normalized MSE at Epoch 01 and every tenth epoch through Epoch 60. The values can fluctuate because dropout remains active during training.

The progress lines confirm that both attention layers can process the connected graphs and return one prediction for every target trait.

Does training stop before epoch 60?

Confirm that target normalization ran before training. The shape of each graph.y tensor must match the number of trait states returned by the model.

If the runtime reset, rerun the import, loading, graph-building, splitting, and normalization cells before recreating the model.

Help me diagnose my MultiTraitGAT training failure.

Evaluate predictions and save outputs

Evaluation converts normalized predictions back to the original rubric scales. Held-out MAE then measures the average absolute gap between predicted and human scores for each trait.

  • Add held-out evaluation below train_model() by copying this code:
def evaluate_model(model, test_graphs, trait_names, means, stds):
    errors_by_trait = [[] for _ in trait_names]
    report_rows = []
    model.eval()

    with torch.no_grad():
        for graph in test_graphs:
            normalized_prediction = model(graph)
            prediction = normalized_prediction * stds + means
            actual = graph.y * stds + means
            row = {"project_id": graph.project_id}

            for trait_index, trait_name in enumerate(trait_names):
                predicted_value = float(prediction[trait_index].item())
                actual_value = float(actual[trait_index].item())
                absolute_error = abs(predicted_value - actual_value)
                errors_by_trait[trait_index].append(absolute_error)
                row[f"actual_{trait_name}"] = actual_value
                row[f"predicted_{trait_name}"] = predicted_value
                row[f"absolute_error_{trait_name}"] = absolute_error

            report_rows.append(row)

    print("Held-out MAE by trait")
    for trait_name, errors in zip(trait_names, errors_by_trait):
        mae = sum(errors) / len(errors)
        print(f"  {trait_name}: {mae:.4f}")

    return report_rows

How does evaluation stay separate?

Calling model.eval() switches off training behavior such as dropout. The no-gradient context avoids storing training information during inference.

The function denormalizes both predictions and targets before calculating absolute errors. Each report row keeps the project identifier plus actual, predicted, and error values for every detected trait.

  • Add the CSV writer immediately below evaluate_model() by copying this code:
def write_prediction_report(report_rows, trait_names, output_path):
    fieldnames = ["project_id"]
    for trait_name in trait_names:
        fieldnames.extend(
            [
                f"actual_{trait_name}",
                f"predicted_{trait_name}",
                f"absolute_error_{trait_name}",
            ]
        )

    with output_path.open("w", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(report_rows)

What goes into the report?

The header begins with project_id. It then adds an actual score, predicted score, and absolute error column for each trait discovered in scores.csv.

Building the columns from trait_names keeps the report compatible with different rubric schemas.

  • Save project_quality_gat.py.
  • Run the cells that define evaluate_model() and write_prediction_report().
  • Predict whether the MAE table will contain one row or one row per detected rubric trait.
  • Evaluate the model by running this cell:
report_rows = evaluate_model(model, test_graphs, trait_names, means, stds)

What should you see?

You will see Held-out MAE by trait followed by one numeric MAE value for every detected rubric trait.

These values are an instructional held-out signal. A production evaluation would need more data and broader validation.

  • In main(), find the connected-audit line that prints graph_audit(graphs[0]).
  • Add the split, normalization, training, and evaluation flow directly below that line:
    train_graphs, test_graphs = split_graphs(graphs)
    print(f"Training projects: {len(train_graphs)}")
    print(f"Held-out projects: {len(test_graphs)}")

    means, stds = normalize_targets(
        train_graphs,
        graphs,
        trait_count=len(trait_names),
    )

    model = MultiTraitGAT(input_channels=len(FEATURE_NAMES))
    train_model(model, train_graphs)
    report_rows = evaluate_model(model, test_graphs, trait_names, means, stds)

How does the complete flow fit together?

The split happens before normalization so the target statistics use training projects only. The model then trains against normalized targets before evaluation restores the original scoring scales.

Keeping this sequence inside main() makes the full experiment reproducible from a single run.

  • Continue main() by adding the output creation and return code directly below report_rows:
    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
    checkpoint_path = OUTPUT_DIR / "gat_project_quality.pt"
    report_path = OUTPUT_DIR / "held_out_predictions.csv"

    torch.save(
        {
            "model_state_dict": model.state_dict(),
            "trait_names": trait_names,
            "target_means": [float(value.item()) for value in means],
            "target_standard_deviations": [float(value.item()) for value in stds],
            "feature_names": FEATURE_NAMES,
        },
        checkpoint_path,
    )
    write_prediction_report(report_rows, trait_names, report_path)

    print(f"Saved checkpoint: {checkpoint_path}")
    print(f"Saved predictions: {report_path}")
    return model, test_graphs, trait_names


model, test_graphs, trait_names = main()

What gets saved?

The checkpoint stores the trained parameters plus the trait names, feature names, means, and standard deviations needed to interpret future predictions.

The CSV stores one row per held-out project. Returning model, test_graphs, and trait_names keeps them available for notebook inspection.

✔️ Awesome, I've got everything!

Great. Save project_quality_gat.py before running the complete training flow.

ⓧ I'd like to double check the full code

import csv
import hashlib
import json
import math
import random
import re
from pathlib import Path

import torch
import torch.nn.functional as F
from torch_geometric.data import Data
from torch_geometric.nn.conv import GATConv

DATA_ROOT = Path("your-drive-project-folder-path")
RAW_DIR = DATA_ROOT / "raw_json"
SCORES_PATH = DATA_ROOT / "scores.csv"
OUTPUT_DIR = DATA_ROOT / "outputs"

SEED = 17
TEST_FRACTION = 0.20
EPOCHS = 60
TOKEN_BUCKETS = 16
HIDDEN_CHANNELS = 16
ATTENTION_HEADS = 4
DROPOUT = 0.10
LEARNING_RATE = 0.01

FEATURE_NAMES = [
    "is_object",
    "is_array",
    "is_text",
    "is_number",
    "is_bool_or_null",
    "is_trait",
    *[f"token_bucket_{index}" for index in range(TOKEN_BUCKETS)],
    "log_character_count",
    "log_word_count",
    "depth",
    "sibling_position",
]


def load_score_table(scores_path):
    with scores_path.open("r", encoding="utf-8", newline="") as handle:
        reader = csv.DictReader(handle)
        if not reader.fieldnames or reader.fieldnames[0] != "project_id":
            raise ValueError("scores.csv must begin with a project_id column")

        trait_names = reader.fieldnames[1:]
        if not trait_names:
            raise ValueError("scores.csv needs at least one rubric-trait column")

        rows = []
        seen_ids = set()
        for row_number, row in enumerate(reader, start=2):
            project_id = row["project_id"].strip()
            if not project_id:
                raise ValueError(f"Missing project_id on CSV row {row_number}")
            if project_id in seen_ids:
                raise ValueError(f"Duplicate project_id: {project_id}")

            try:
                targets = [float(row[trait]) for trait in trait_names]
            except (TypeError, ValueError) as error:
                raise ValueError(
                    f"Every trait must be numeric on CSV row {row_number}"
                ) from error

            rows.append((project_id, targets))
            seen_ids.add(project_id)

    return trait_names, rows


def load_records(raw_dir, scores_path):
    if not raw_dir.is_dir():
        raise FileNotFoundError(f"Missing raw JSON directory: {raw_dir}")
    if not scores_path.is_file():
        raise FileNotFoundError(f"Missing score file: {scores_path}")

    trait_names, score_rows = load_score_table(scores_path)
    records = []

    for project_id, targets in score_rows:
        json_path = raw_dir / f"{project_id}.json"
        if not json_path.is_file():
            raise FileNotFoundError(
                f"No JSON file matches project_id {project_id}: {json_path}"
            )
        with json_path.open("r", encoding="utf-8") as handle:
            raw_project = json.load(handle)
        records.append((project_id, raw_project, targets))

    if len(records) < 5:
        raise ValueError("Use at least five labeled projects for this holdout prototype")

    return trait_names, records


def node_type_vector(value, is_trait=False):
    if is_trait:
        return [0.0, 0.0, 0.0, 0.0, 0.0, 1.0]
    if isinstance(value, dict):
        return [1.0, 0.0, 0.0, 0.0, 0.0, 0.0]
    if isinstance(value, list):
        return [0.0, 1.0, 0.0, 0.0, 0.0, 0.0]
    if isinstance(value, str):
        return [0.0, 0.0, 1.0, 0.0, 0.0, 0.0]
    if isinstance(value, bool) or value is None:
        return [0.0, 0.0, 0.0, 0.0, 1.0, 0.0]
    if isinstance(value, (int, float)):
        return [0.0, 0.0, 0.0, 1.0, 0.0, 0.0]
    return [0.0, 0.0, 1.0, 0.0, 0.0, 0.0]


def scalar_text(value):
    if isinstance(value, (dict, list)) or value is None:
        return ""
    return str(value)


def token_bucket_features(text):
    buckets = [0.0] * TOKEN_BUCKETS
    tokens = re.findall(r"[A-Za-z0-9_]+", text.lower())
    for token in tokens:
        digest = hashlib.sha256(token.encode("utf-8")).digest()
        bucket = int.from_bytes(digest[:4], "big") % TOKEN_BUCKETS
        buckets[bucket] += 1.0

    total = sum(buckets)
    if total:
        buckets = [value / total for value in buckets]
    return buckets


def make_node_features(value, label, depth, position, is_trait=False):
    text = label if is_trait else f"{label} {scalar_text(value)}".strip()
    word_count = len(re.findall(r"[A-Za-z0-9_]+", text))
    return [
        *node_type_vector(value, is_trait=is_trait),
        *token_bucket_features(text),
        math.log1p(len(text)) / 10.0,
        math.log1p(word_count) / 5.0,
        min(depth, 20) / 20.0,
        min(position, 20) / 20.0,
    ]


def make_node_label(value, label, is_trait=False):
    if is_trait:
        return f"trait:{label}"
    preview = scalar_text(value).replace("\n", " ")[:40]
    return f"{label}:{preview}" if preview else label


def build_graph(project_id, raw_project, targets, trait_names, connected):
    features = []
    node_labels = []
    edges = []

    def add_edge(source, target):
        edges.append((source, target))

    def visit(value, label, depth, position, parent_id=None):
        node_id = len(features)
        features.append(make_node_features(value, label, depth, position))
        node_labels.append(make_node_label(value, label))

        if connected and parent_id is not None:
            add_edge(parent_id, node_id)
            add_edge(node_id, parent_id)

        if isinstance(value, dict):
            children = [(str(key), child) for key, child in value.items()]
        elif isinstance(value, list):
            children = [(f"{label}[{index}]", child) for index, child in enumerate(value)]
        else:
            children = []

        previous_child_id = None
        for child_position, (child_label, child_value) in enumerate(children):
            child_id = visit(
                child_value,
                child_label,
                depth + 1,
                child_position,
                node_id,
            )
            if connected and previous_child_id is not None:
                add_edge(previous_child_id, child_id)
                add_edge(child_id, previous_child_id)
            previous_child_id = child_id

        return node_id

    visit(raw_project, "project", 0, 0)
    content_node_count = len(features)
    trait_start = content_node_count

    for trait_name in trait_names:
        features.append(
            make_node_features(None, trait_name, 0, 0, is_trait=True)
        )
        node_labels.append(make_node_label(None, trait_name, is_trait=True))

    if connected:
        trait_ids = list(range(trait_start, len(features)))
        for trait_id in trait_ids:
            for content_id in range(content_node_count):
                add_edge(trait_id, content_id)
                add_edge(content_id, trait_id)
            for other_trait_id in trait_ids:
                if trait_id != other_trait_id:
                    add_edge(trait_id, other_trait_id)

    edge_index = torch.tensor(
        [
            [source for source, _ in edges],
            [target for _, target in edges],
        ],
        dtype=torch.long,
    )

    graph = Data(
        x=torch.tensor(features, dtype=torch.float32),
        edge_index=edge_index,
        y=torch.tensor(targets, dtype=torch.float32),
    )
    graph.project_id = project_id
    graph.trait_start = trait_start
    graph.node_labels = node_labels
    return graph


def graph_audit(graph):
    node_count = int(graph.num_nodes)
    edge_count = int(graph.num_edges)
    neighbors = [set() for _ in range(node_count)]

    for edge_position in range(edge_count):
        source = int(graph.edge_index[0, edge_position].item())
        target = int(graph.edge_index[1, edge_position].item())
        neighbors[source].add(target)
        neighbors[target].add(source)

    isolated = sum(1 for adjacent in neighbors if not adjacent)
    visited = set()
    components = 0

    for start in range(node_count):
        if start in visited:
            continue
        components += 1
        stack = [start]
        visited.add(start)
        while stack:
            current = stack.pop()
            for neighbor in neighbors[current]:
                if neighbor not in visited:
                    visited.add(neighbor)
                    stack.append(neighbor)

    return {
        "nodes": node_count,
        "edges": edge_count,
        "isolated_nodes": isolated,
        "connected_components": components,
    }


def split_graphs(graphs):
    indices = list(range(len(graphs)))
    random.Random(SEED).shuffle(indices)
    test_size = max(1, int(round(len(graphs) * TEST_FRACTION)))
    test_indices = set(indices[:test_size])
    train_graphs = [graph for index, graph in enumerate(graphs) if index not in test_indices]
    test_graphs = [graph for index, graph in enumerate(graphs) if index in test_indices]
    return train_graphs, test_graphs


def normalize_targets(train_graphs, all_graphs, trait_count):
    means = []
    standard_deviations = []

    for trait_index in range(trait_count):
        values = [float(graph.y[trait_index].item()) for graph in train_graphs]
        mean = sum(values) / len(values)
        variance = sum((value - mean) ** 2 for value in values) / len(values)
        standard_deviation = math.sqrt(variance)
        if standard_deviation == 0.0:
            standard_deviation = 1.0
        means.append(mean)
        standard_deviations.append(standard_deviation)

    means_tensor = torch.tensor(means, dtype=torch.float32)
    stds_tensor = torch.tensor(standard_deviations, dtype=torch.float32)

    for graph in all_graphs:
        graph.y = (graph.y - means_tensor) / stds_tensor

    return means_tensor, stds_tensor


class MultiTraitGAT(torch.nn.Module):
    def __init__(self, input_channels):
        super().__init__()
        self.gat1 = GATConv(
            input_channels,
            HIDDEN_CHANNELS,
            heads=ATTENTION_HEADS,
            dropout=DROPOUT,
        )
        self.gat2 = GATConv(
            HIDDEN_CHANNELS * ATTENTION_HEADS,
            HIDDEN_CHANNELS,
            heads=1,
            concat=False,
            dropout=DROPOUT,
        )
        self.score_head = torch.nn.Linear(HIDDEN_CHANNELS, 1)

    def forward(self, graph):
        node_states = F.elu(self.gat1(graph.x, graph.edge_index))
        node_states = self.gat2(node_states, graph.edge_index)
        trait_states = node_states[graph.trait_start :]
        return self.score_head(trait_states)[:, 0]


def train_model(model, train_graphs):
    optimizer = torch.optim.Adam(
        model.parameters(),
        lr=LEARNING_RATE,
        weight_decay=1e-4,
    )

    for epoch in range(1, EPOCHS + 1):
        model.train()
        epoch_graphs = list(train_graphs)
        random.Random(SEED + epoch).shuffle(epoch_graphs)
        total_loss = 0.0

        for graph in epoch_graphs:
            optimizer.zero_grad()
            prediction = model(graph)
            loss = F.mse_loss(prediction, graph.y)
            loss.backward()
            optimizer.step()
            total_loss += float(loss.item())

        if epoch == 1 or epoch % 10 == 0:
            average_loss = total_loss / len(epoch_graphs)
            print(f"Epoch {epoch:02d} | normalized MSE {average_loss:.4f}")


def evaluate_model(model, test_graphs, trait_names, means, stds):
    errors_by_trait = [[] for _ in trait_names]
    report_rows = []
    model.eval()

    with torch.no_grad():
        for graph in test_graphs:
            normalized_prediction = model(graph)
            prediction = normalized_prediction * stds + means
            actual = graph.y * stds + means
            row = {"project_id": graph.project_id}

            for trait_index, trait_name in enumerate(trait_names):
                predicted_value = float(prediction[trait_index].item())
                actual_value = float(actual[trait_index].item())
                absolute_error = abs(predicted_value - actual_value)
                errors_by_trait[trait_index].append(absolute_error)
                row[f"actual_{trait_name}"] = actual_value
                row[f"predicted_{trait_name}"] = predicted_value
                row[f"absolute_error_{trait_name}"] = absolute_error

            report_rows.append(row)

    print("Held-out MAE by trait")
    for trait_name, errors in zip(trait_names, errors_by_trait):
        mae = sum(errors) / len(errors)
        print(f"  {trait_name}: {mae:.4f}")

    return report_rows


def write_prediction_report(report_rows, trait_names, output_path):
    fieldnames = ["project_id"]
    for trait_name in trait_names:
        fieldnames.extend(
            [
                f"actual_{trait_name}",
                f"predicted_{trait_name}",
                f"absolute_error_{trait_name}",
            ]
        )

    with output_path.open("w", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(report_rows)


def main():
    torch.manual_seed(SEED)
    random.seed(SEED)

    trait_names, records = load_records(RAW_DIR, SCORES_PATH)
    print(f"Loaded {len(records)} projects with traits: {trait_names}")
    print(f"First project identifier: {records[0][0]}")

    first_project_id, first_raw_project, first_targets = records[0]
    naive_graph = build_graph(
        first_project_id,
        first_raw_project,
        first_targets,
        trait_names,
        connected=False,
    )
    print("First five node labels:", naive_graph.node_labels[:5])
    print("Naive graph audit:", graph_audit(naive_graph))

    graphs = [
        build_graph(
            project_id,
            raw_project,
            targets,
            trait_names,
            connected=True,
        )
        for project_id, raw_project, targets in records
    ]
    print("Connected graph audit:", graph_audit(graphs[0]))

    train_graphs, test_graphs = split_graphs(graphs)
    print(f"Training projects: {len(train_graphs)}")
    print(f"Held-out projects: {len(test_graphs)}")

    means, stds = normalize_targets(
        train_graphs,
        graphs,
        trait_count=len(trait_names),
    )

    model = MultiTraitGAT(input_channels=len(FEATURE_NAMES))
    train_model(model, train_graphs)
    report_rows = evaluate_model(model, test_graphs, trait_names, means, stds)

    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
    checkpoint_path = OUTPUT_DIR / "gat_project_quality.pt"
    report_path = OUTPUT_DIR / "held_out_predictions.csv"

    torch.save(
        {
            "model_state_dict": model.state_dict(),
            "trait_names": trait_names,
            "target_means": [float(value.item()) for value in means],
            "target_standard_deviations": [float(value.item()) for value in stds],
            "feature_names": FEATURE_NAMES,
        },
        checkpoint_path,
    )
    write_prediction_report(report_rows, trait_names, report_path)

    print(f"Saved checkpoint: {checkpoint_path}")
    print(f"Saved predictions: {report_path}")
    return model, test_graphs, trait_names


model, test_graphs, trait_names = main()

How to use this reference

Compare the reference with your saved project_quality_gat.py file. Pay close attention to the order of splitting, normalization, training, evaluation, and saving.

Before the final run, predict which three signals will prove the complete pipeline worked: training progress, held-out evaluation, or saved output paths.

  • Save the completed project_quality_gat.py file.
  • Run the complete project_quality_gat.py cell in the Drive-connected Colab notebook.

You will see loss updates through Epoch 60. You will then see the Held-out MAE by trait table.

The final lines print Drive paths for gat_project_quality.pt and held_out_predictions.csv.

  • Return to the Drive project folder from earlier.
  • Open the outputs folder.
  • Confirm that gat_project_quality.pt appears in the folder.
  • Confirm that held_out_predictions.csv appears in the folder.
  • Open held_out_predictions.csv and confirm that each trait has actual, predicted, and absolute-error columns.

Are the output files missing?

Check that DATA_ROOT still points to the mounted Drive folder containing raw_json and scores.csv.

Confirm that the run reaches both saved-path lines. An earlier loading, graph, or training failure prevents the output section from running.

Help me find why my GAT checkpoint or prediction CSV was not saved.

You have completed the full scoring pipeline. Your connected graphs now produce held-out rubric predictions plus reusable files in Google Drive.

Secret mission

Inspect Which Blocks Receive the Most Attention

Your scorer can predict several rubric traits for an unseen project. This challenge traces its second attention layer to rank the five JSON-derived content labels emphasized most for one trait. The ranking stays grounded as a model-inspection signal with no causal claim.

Clean Up Your Resources

Clean Up Your Resources

The project used the free-of-charge Google Colab tier with no paid cloud resources. Choose whether to keep the Google Drive files or disconnect the runtime or delete the saved artifacts.

Resources you used:

  • A Google Colab notebook stored in Google Drive with the training script and attention-inspection cell.
  • A PyTorch checkpoint at outputs/gat_project_quality.pt.
  • A CSV evaluation report at outputs/held_out_predictions.csv.

Keep everything running

No action is needed. Choose this option if you want to continue scoring projects or inspect more attention rankings.

  • Leave the Google Colab notebook in its current Google Drive location.
  • Keep outputs/gat_project_quality.pt for future scoring experiments.
  • Keep outputs/held_out_predictions.csv for future held-out comparisons.
  • Rerun the setup cell when a new Colab runtime starts.

Pause - I'll come back to this later

Shut down the active Colab session while keeping every Google Drive file. Runtime packages need to be restored when you return.

  • Disconnect the Google Colab runtime using the runtime controls.
  • Confirm the notebook no longer has an active runtime session.
  • Leave the notebook in its current Google Drive location.
  • Leave the outputs/ folder inside the Drive project folder assigned to DATA_ROOT.

Delete - I don't want to use this again

Removing the saved notebook and generated outputs gives you a clean start. Your source raw_json/ folder and scores.csv stay in place because this project did not create them.

  • In Google Drive, move the outputs/ folder inside the Drive project folder assigned to DATA_ROOT to the bin.
  • Move the Google Colab notebook used for this project to the bin.
  • Permanently delete the moved outputs/ folder from the Google Drive bin.
  • Permanently delete the moved notebook from the Google Drive bin.
  • Confirm outputs/ no longer appears inside the Drive project folder assigned to DATA_ROOT.
  • Confirm the notebook no longer appears in its saved Google Drive location.

Nice Work!

Nice Work!

You did it! Your Google Colab notebook now scores structured NextWork projects with a Graph Attention Network.

You've learned how to:

  • Build connected graph datasets from raw JSON projects after proving that isolated nodes cannot pass messages.
  • Train a multi-trait Graph Attention Network with targets normalized from training data only. Produce one predicted score for each detected rubric trait.
  • Measure held-out performance with per-trait MAE. Save a reusable PyTorch checkpoint to Google Drive. Write each human score beside its prediction and absolute error in a CSV report.
  • Secret Mission: Inspected the content blocks with the largest incoming second-layer attention weights for one trait. You kept the ranking framed as a model-inspection signal with no causal claim.

Ready to quiz yourself?