In this tutorial, we explore how we can decode linguistic features directly from brain signals using a modern neuroAI pipeline. We work with MEG data and build an end-to-end system that transforms raw neural activity into meaningful predictions, in this case, estimating word length from brain responses. We set up the environment, load and process neural events, design a custom feature extractor, and construct a structured data pipeline using NeuralSet. From there, we train a convolutional neural network to learn patterns in the temporal and spatial structure of MEG signals. Throughout the process, we focus on building a clean, modular workflow that mirrors real-world neuroAI research practices.
import subprocess, sys, importlib, pkgutil
def pip_install(*pkgs):
print(f"pip install {' '.join(pkgs)} ...")
r = subprocess.run([sys.executable, "-m", "pip", "install", "-q", *pkgs],
capture_output=True, text=True)
if r.returncode != 0:
print("pip STDOUT:", r.stdout[-2000:])
print("pip STDERR:", r.stderr[-2000:])
raise RuntimeError("pip install failed; see output above.")
print(" ok")
pip_install("numpy>=2.0,<2.3")
pip_install("neuralset")
pip_install("neuralfetch")
import numpy as np
from numpy._core.umath import _center
print(f"numpy {np.__version__} OK")
import warnings, typing as tp
warnings.filterwarnings("ignore")
import pandas as pd
import torch
import torch.nn as nn
import torch.nn.functional as F
from torch.utils.data import DataLoader
import matplotlib.pyplot as plt
import neuralset as ns
from neuralset import extractors as ext_modWe install and validate all required dependencies, ensuring critical packages such as NumPy and NeuralSet are properly configured. We perform a quick NumPy check to avoid runtime issues later in the pipeline. We then import all core libraries needed for data processing, modeling, and visualization.
def deep_import(pkg_name: str):
try:
pkg = importlib.import_module(pkg_name)
except Exception as e:
print(f"⚠️ could not import {pkg_name}: {e}")
return
if not hasattr(pkg, "__path__"):
return
for m in pkgutil.walk_packages(pkg.__path__, prefix=pkg_name + "."):
try:
importlib.import_module(m.name)
except Exception:
pass
deep_import("neuralfetch")
deep_import("neuralset")
torch.manual_seed(0); np.random.seed(0)
catalog = ns.Study.catalog()
print(f"\n{len(catalog)} studies registered.")
preferred = ["Fake2025Meg", "Test2025Meg", "Test2023Meg"]
study_name = next((n for n in preferred if n in catalog), None)
if study_name is None:
meg_studies = [n for n, c in catalog.items() if "Meg" in c.neuro_types()]
study_name = meg_studies[0] if meg_studies else None
if study_name is None:
raise RuntimeError(
"No MEG study available. Catalog: "
f"{sorted(catalog.keys())[:20]}… "
"Install neuralfetch correctly (pip install neuralfetch) and re-run."
)
print(f"→ Using study: {study_name}")We dynamically import all submodules from NeuralFetch and NeuralSet to ensure that all available studies are properly registered. We seed the random number generator for reproducibility and inspect the study catalog to identify available MEG datasets. We then select an appropriate study to use as the foundation for our pipeline.
class CharCount(ext_mod.BaseStatic):
event_types: tp.Literal["Word"] = "Word"
def get_static(self, event) -> torch.Tensor:
return torch.tensor([float(len(event.text))], dtype=torch.float32)
print("\nBuilding chain...")
chain = ns.Chain(steps=[
{"name": study_name, "path": str(ns.CACHE_FOLDER)},
{"name": "QueryEvents", "query": "type in ['Word', 'Meg']"},
])
events = chain.run()
print(f" → {len(events)} events; types={sorted(events.type.unique().tolist())}")
print(f" → Words: {(events.type=='Word').sum()} | "
f"timelines: {events.timeline.nunique()}")
print("\nSample words:")
print(events[events.type=='Word'][["start","duration","text","timeline"]]
.head(5).to_string(index=False))
print("\nBuilding segmenter...")
segmenter = ns.dataloader.Segmenter(
extractors={
"meg": {"name": "MegExtractor", "frequency": 100.0},
"char_count": CharCount(aggregation="trigger"),
},
trigger_query="type == 'Word'",
start=-0.2, duration=0.8,
drop_incomplete=True,
)
dataset = segmenter.apply(events)
print(f" → SegmentDataset: {len(dataset)} segments")
s0 = dataset[0]
print(f"\nSingle item:\n meg : {tuple(s0.data['meg'].shape)}")
print(f" char_count : {s0.data['char_count'].item()} "
f"(word: {s0.segments[0].trigger.text!r})")We define a custom extractor that computes the character count of each word event, enabling us to create a supervised learning target. We build a processing chain to load and filter relevant events from the selected study. We then segment the MEG signals around word events and construct a dataset ready for modeling.
rng = np.random.RandomState(42)
perm = rng.permutation(len(dataset))
n_tr, n_va = int(0.70*len(dataset)), int(0.15*len(dataset))
train_ds = dataset.select(perm[:n_tr])
val_ds = dataset.select(perm[n_tr:n_tr+n_va])
test_ds = dataset.select(perm[n_tr+n_va:])
print(f"\nSplit | train={len(train_ds)} val={len(val_ds)} test={len(test_ds)}")
mk = lambda d, sh: DataLoader(d, batch_size=32, shuffle=sh,
collate_fn=d.collate_fn, drop_last=False)
train_loader, val_loader, test_loader = mk(train_ds, True), mk(val_ds, False), mk(test_ds, False)
probe = next(iter(train_loader))
n_ch, n_t = probe.data["meg"].shape[-2:]
print(f" → batch[meg] shape: {tuple(probe.data['meg'].shape)}")
print(f" → batch[char] shape: {tuple(probe.data['char_count'].shape)}")
class MEGDecoder(nn.Module):
def __init__(self, n_channels: int, mid: int = 64):
super().__init__()
self.spatial = nn.Conv1d(n_channels, mid, 1)
self.bn0 = nn.BatchNorm1d(mid)
self.temporal1 = nn.Conv1d(mid, mid, 7, padding=3)
self.bn1 = nn.BatchNorm1d(mid)
self.temporal2 = nn.Conv1d(mid, mid//2, 7, padding=3)
self.bn2 = nn.BatchNorm1d(mid//2)
self.pool = nn.AdaptiveAvgPool1d(1)
self.head = nn.Linear(mid//2, 1)
self.drop = nn.Dropout(0.3)
def forward(self, x):
x = F.gelu(self.bn0(self.spatial(x)))
x = F.gelu(self.bn1(self.temporal1(x)))
x = self.drop(x)
x = F.gelu(self.bn2(self.temporal2(x)))
return self.head(self.pool(x).squeeze(-1)).squeeze(-1)
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model = MEGDecoder(n_channels=n_ch).to(device)
print(f"\nDevice: {device} | params: {sum(p.numel() for p in model.parameters()):,}")
train_targets = torch.cat([b.data["char_count"].squeeze(-1) for b in train_loader])
y_mean, y_std = train_targets.mean().item(), train_targets.std().item() + 1e-6
print(f"Target μ={y_mean:.2f} σ={y_std:.2f}")
def prep(batch):
x = batch.data["meg"].to(device).float()
y = batch.data["char_count"].squeeze(-1).to(device).float()
x = (x - x.mean(-1, keepdim=True)) / (x.std(-1, keepdim=True) + 1e-6)
y = (y - y_mean) / y_std
return x, yWe split the dataset into training, validation, and test sets to ensure proper model evaluation. We create data loaders and inspect batch shapes to confirm correct data formatting. We then define a convolutional neural network and prepare normalized inputs and targets for stable training.
EPOCHS = 15
opt = torch.optim.AdamW(model.parameters(), lr=1e-3, weight_decay=1e-4)
sched = torch.optim.lr_scheduler.CosineAnnealingLR(opt, T_max=EPOCHS)
loss_fn = nn.MSELoss()
hist = {"tr": [], "va": [], "r": []}
def pearson(a, b):
a, b = a - a.mean(), b - b.mean()
return (a*b).sum() / (a.norm()*b.norm() + 1e-8)
print("\n" + "="*64)
print(f"{'Epoch':>5} | {'train':>9} | {'val':>9} | {'val_r':>7}")
print("="*64)
for ep in range(EPOCHS):
model.train(); tr = []
for batch in train_loader:
x, y = prep(batch)
loss = loss_fn(model(x), y)
opt.zero_grad(); loss.backward()
torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)
opt.step(); tr.append(loss.item())
sched.step()
model.eval(); va, P, T = [], [], []
with torch.no_grad():
for batch in val_loader:
x, y = prep(batch); p = model(x)
va.append(loss_fn(p, y).item()); P.append(p.cpu()); T.append(y.cpu())
P, T = torch.cat(P), torch.cat(T)
r = pearson(P, T).item()
hist["tr"].append(np.mean(tr)); hist["va"].append(np.mean(va)); hist["r"].append(r)
print(f"{ep+1:>5d} | {np.mean(tr):>9.4f} | {np.mean(va):>9.4f} | {r:>+7.3f}")
model.eval(); P, T = [], []
with torch.no_grad():
for batch in test_loader:
x, y = prep(batch)
P.append(model(x).cpu()); T.append(y.cpu())
P, T = torch.cat(P), torch.cat(T)
test_r = pearson(P, T).item()
test_mse = ((P - T) ** 2).mean().item()
print(f"\nTEST | Pearson r = {test_r:+.3f} MSE = {test_mse:.3f}")
print(f"(Synthetic-MEG signals are random by design — small/zero r is expected.)")
fig, ax = plt.subplots(1, 3, figsize=(15, 4))
ax[0].plot(hist["tr"], label="train"); ax[0].plot(hist["va"], label="val")
ax[0].set(xlabel="Epoch", ylabel="MSE", title="Loss curves"); ax[0].legend(); ax[0].grid(alpha=.3)
ax[1].plot(hist["r"], color="C2"); ax[1].axhline(0, color="k", ls="--", alpha=.4)
ax[1].set(xlabel="Epoch", ylabel="Pearson r", title="Validation correlation"); ax[1].grid(alpha=.3)
m = float(max(T.abs().max(), P.abs().max()))
ax[2].scatter(T.numpy(), P.numpy(), s=10, alpha=.35)
ax[2].plot([-m, m], [-m, m], "k--", alpha=.4)
ax[2].set(xlabel="True (z-scored char count)", ylabel="Predicted",
title=f"Test predictions (r = {test_r:+.3f})"); ax[2].grid(alpha=.3)
plt.tight_layout(); plt.show()
print("\n✅ Tutorial complete!")
print(f" • Study used : {study_name}")
print(f" • Pipeline : Chain → Segmenter → SegmentDataset → DataLoader")
print(f" • Custom extractor : CharCount (subclass of BaseStatic)")
print(f" • Built-in extractor: MegExtractor @ 100 Hz")
print(f" • Model : 1×1 spatial conv + 2 temporal convs + linear head")We train the neural network using a structured training loop with loss tracking and learning rate scheduling. We evaluate the model on the validation and test sets using metrics such as MSE and Pearson’s correlation. Also, we visualize training performance and predictions to understand how well the model learns from the data.
In conclusion, we demonstrated how we can bridge neural data and language understanding using deep learning. We implemented a full pipeline, from raw event extraction to model training and evaluation, while maintaining flexibility through reusable components like chains, segmenters, and extractors. Although we worked with synthetic MEG signals, the framework we built is directly applicable to real-world datasets and more complex decoding tasks. This exercise highlights how we can combine neuroscience, machine learning, and structured pipelines to advance interpretable brain decoding systems, laying a strong foundation for more advanced neuroAI applications.
Check out the Full Codes and Notebook here. Also, feel free to follow us on Twitter and don’t forget to join our 130k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
Need to partner with us for promoting your GitHub Repo OR Hugging Face Page OR Product Release OR Webinar etc.? Connect with us
Sana Hassan, a consulting intern at Marktechpost and dual-degree student at IIT Madras, is passionate about applying technology and AI to address real-world challenges. With a keen interest in solving practical problems, he brings a fresh perspective to the intersection of AI and real-life solutions.
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。