
Tribe V2 Agent Alignment
- 1 installs
- 3 repo stars
- Updated August 5, 2026
- broomva/skills
tribe-v2-agent-alignment is a Claude skill that uses Meta's TRIBE v2 brain encoder to benchmark AI encoders by cortical alignment and inform model selection in an agent OS.
About
tribe-v2-agent-alignment is a Claude skill that uses Meta's TRIBE v2 brain encoder to score how well an AI encoder's representations align with human cortical processing. It runs alignment for text, video, or audio encoders and compares candidates by their language- or visual-cortex R2 fit. A developer uses it to pick the most brain-aligned encoder before wiring it into agent model routing. It also validates that a fine-tuned model has not lost biological plausibility.
- Scores text, video, and audio encoders by cortical alignment using TRIBE v2
- Compares encoders (LLaMA, Mistral, BERT) by language-cortex R2 fit
- Feeds neuro-alignment scores into agent model routing
Tribe V2 Agent Alignment by the numbers
- 1 all-time installs (skills.sh)
- Ranked #14,098 of 16,546 AI & Agent Building skills by installs in the Skillselion catalog
- Data as of Aug 5, 2026 (Skillselion catalog sync)
tribe-v2-agent-alignment capabilities & compatibility
- Capabilities
- model selection · encoder benchmark · model routing
- Use cases
- research · data analysis · orchestration
- Runs
- Runs locally
- Pricing
- Free
What tribe-v2-agent-alignment says it does
Validate whether your AI encoders — text, video, or audio — represent information the way human brains do, using Meta's TRIBE v2 cortical predictor.
A high alignment score (R² > 0.25) means the encoder has learned representations that are geometrically similar to what the human language, visual, or auditory cortex computes
Run a full alignment score for any encoder in 5 commands:
npx skills add https://github.com/broomva/skills --skill tribe-v2-agent-alignmentAdd your badge
Show developers this skill is listed on Skillselion. Paste this into your README.
| Installs | 1 |
|---|---|
| repo stars | ★ 3 |
| Last updated | August 5, 2026 |
| Repository | broomva/skills ↗ |
What it does
Benchmark AI encoders by cortical alignment and route models by their neuro-alignment score.
Who is it for?
Comparing candidate encoders by brain alignment before committing them to a routing stack.
Skip if: Production inference; it is an offline evaluation and model-selection tool.
When should I use this skill?
Benchmarking whether a new encoder aligns with human cortex or comparing encoders by alignment score.
What you get
Per-encoder cortical alignment scores that drive neuro-informed model routing.
- Per-encoder alignment JSON
- Encoder comparison table
- Routing recommendation
By the numbers
- Full alignment score in 5 commands
- TRIBE v2 predicts ~20k fsaverage5 cortical vertices
- Alignment threshold R2 > 0.25
Files
TRIBE v2 Agent Alignment
Validate whether your AI encoders — text, video, or audio — represent information the way human brains do, using Meta's TRIBE v2 cortical predictor. Use the resulting alignment scores to drive neuro-informed model routing in Life/Arcan.
Concept
Cortical alignment measures how well an AI encoder's hidden states predict actual fMRI brain activity in response to the same stimulus. TRIBE v2 (TRansformer for In-silico Brain Experiments) was trained on thousands of hours of naturalistic fMRI data and can predict activity across the full cortical surface (~20k vertices on the fsaverage5 mesh) for any text, video, or audio input. A high alignment score (R² > 0.25) means the encoder has learned representations that are geometrically similar to what the human language, visual, or auditory cortex computes — without any explicit neuroscience objective. This matters for model selection in an agent OS: a text encoder with higher language cortex alignment tends to generalize better to novel linguistic contexts, is more robust to distribution shift, and exhibits better zero-shot transfer. TRIBE v2 proved that LLaMA 3.2-3B spontaneously developed such alignment, validating its representations neurologically. The same benchmark can be applied to any candidate encoder before committing it to Arcan's routing stack.
Quick Start
Run a full alignment score for any encoder in 5 commands:
# 1. Install dependencies
pip install tribev2 transformers torch scikit-learn numpy
# 2. Prepare a stimulus directory (video files for video, text files for text, wav for audio)
mkdir -p ~/stimuli/text && echo "The model routed the task to the visual cortex." > ~/stimuli/text/s1.txt
# 3. Run alignment against LLaMA 3.2-3B (text encoder baseline)
python scripts/align_encoder.py \
--encoder-type text \
--encoder-model meta-llama/Llama-3.2-3B \
--stimulus-dir ~/stimuli/text \
--output ~/results/llama_alignment.json
# 4. Run alignment against a competing encoder
python scripts/align_encoder.py \
--encoder-type text \
--encoder-model bert-base-uncased \
--stimulus-dir ~/stimuli/text \
--output ~/results/bert_alignment.json
# 5. Compare scores
python -c "
import json
llama = json.load(open('~/results/llama_alignment.json'))['alignment_score']
bert = json.load(open('~/results/bert_alignment.json'))['alignment_score']
winner = 'LLaMA 3.2' if llama > bert else 'BERT'
print(f'LLaMA 3.2: {llama:.3f} | BERT: {bert:.3f} | Winner: {winner}')
"Workflow A: Text Encoder Alignment
Compare any two text encoders by their language cortex alignment score. Language cortex vertices cover Broca's area (~vertex 15000-18000, left hemisphere) and Wernicke's area (~vertex 12000-15000, left hemisphere) on the fsaverage5 mesh.
Step 1 — Prepare Text Stimuli
Text stimuli should be naturalistic sentences or paragraphs (not short keywords). TRIBE v2 was trained on narrative speech transcripts; similar inputs yield the most reliable alignment estimates.
mkdir -p ~/stimuli/text
cat > ~/stimuli/text/naturalistic_en.txt << 'EOF'
The surgeon carefully examined the patient before the procedure.
Language emerges from a distributed network spanning frontal and temporal lobes.
The model predicted activation in Broca's area when processing syntactically complex sentences.
EOFStep 2 — Run Alignment for Each Encoder
# Baseline: TRIBE v2's own text encoder (LLaMA 3.2-3B) — expect ~0.40 R²
python scripts/align_encoder.py \
--encoder-type text \
--encoder-model meta-llama/Llama-3.2-3B \
--stimulus-dir ~/stimuli/text \
--output ~/results/llama32_align.json
# Candidate A: Mistral 7B
python scripts/align_encoder.py \
--encoder-type text \
--encoder-model mistralai/Mistral-7B-v0.1 \
--stimulus-dir ~/stimuli/text \
--output ~/results/mistral7b_align.json
# Candidate B: sentence-transformers (smaller, faster)
python scripts/align_encoder.py \
--encoder-type text \
--encoder-model sentence-transformers/all-mpnet-base-v2 \
--stimulus-dir ~/stimuli/text \
--output ~/results/mpnet_align.jsonStep 3 — Interpret and Route
import json, pathlib
results = {}
for p in pathlib.Path("~/results").expanduser().glob("*_align.json"):
d = json.loads(p.read_text())
results[d["encoder"]] = d["alignment_score"]
best = max(results, key=results.get)
print("Alignment scores (language cortex R²):")
for enc, score in sorted(results.items(), key=lambda x: -x[1]):
flag = " <-- route here" if enc == best else ""
print(f" {enc:55s} {score:.3f}{flag}")Text Encoder Comparison Table
| Encoder | Type | Expected R² | Language Cortex Fit |
|---|---|---|---|
| LLaMA 3.2-3B | Autoregressive LM | ~0.40 | Excellent |
| Mistral 7B | Autoregressive LM | ~0.35-0.40 | Excellent |
| GPT-2 (medium) | Autoregressive LM | ~0.25-0.30 | Good |
| BERT-base | Masked LM | ~0.15-0.22 | Moderate |
| all-mpnet-base-v2 | Sentence encoder | ~0.10-0.18 | Moderate |
| Random linear encoder | Baseline | ~0.00-0.03 | Poor |
Workflow B: Video Encoder Alignment
Compare video encoders by their visual cortex alignment. Visual cortex vertices cover V1-V4 (~vertex 1000-5000) and motion-selective areas MT/MST (~vertex 5000-8000) on fsaverage5.
Step 1 — Prepare Video Stimuli
Use naturalistic video clips (not slideshows). MP4 format, 1-5 minutes each. TRIBE v2 segments at 5-second windows internally.
mkdir -p ~/stimuli/video
# Download a CC-licensed short clip, or use any .mp4 you have:
# ffmpeg -i source.mp4 -t 120 -c copy ~/stimuli/video/clip01.mp4Step 2 — Run Alignment
# Baseline: TRIBE v2's own video encoder (V-JEPA2 ViT-G) — expect high visual cortex alignment
python scripts/align_encoder.py \
--encoder-type video \
--encoder-model facebook/vjepa2-vitg-fpc64-256 \
--stimulus-dir ~/stimuli/video \
--output ~/results/vjepa2_align.json
# Candidate: CLIP ViT-L/14
python scripts/align_encoder.py \
--encoder-type video \
--encoder-model openai/clip-vit-large-patch14 \
--stimulus-dir ~/stimuli/video \
--output ~/results/clip_vitl_align.json
# Candidate: VideoMAE-v2 (ViT-G, action recognition)
python scripts/align_encoder.py \
--encoder-type video \
--encoder-model MCG-NJU/videomae-huge \
--stimulus-dir ~/stimuli/video \
--output ~/results/videomae_align.jsonStep 3 — Check Emergent Networks
TRIBE v2 spontaneously recovers 5 functional brain networks. Verify the visual encoder activates the correct one:
import json
result = json.load(open("~/results/vjepa2_align.json"))
print(f"Alignment score: {result['alignment_score']:.3f}")
print(f"Top cortical regions: {result['top_regions']}")
# Expected output for video: top_regions includes 'visual_cortex' vertices 1000-8000Video Encoder Comparison Table
| Encoder | Architecture | Expected Visual R² | Motion Sensitivity |
|---|---|---|---|
| V-JEPA2 (ViT-G) | Masked video prediction | High (>0.35) | High |
| VideoMAE-v2 (ViT-H) | Masked video prediction | High (>0.30) | High |
| CLIP ViT-L/14 | Contrastive image-text | Moderate (0.20-0.28) | Low |
| DINO ViT-B/16 | Self-supervised image | Moderate (0.15-0.22) | Low |
| Random CNN baseline | — | Near-zero | None |
Workflow C: Arcan Integration
Use alignment scores stored in Lago to configure Arcan's model routing at task dispatch time.
Step 1 — Cache Scores in Lago
After running align_encoder.py, push scores into Lago's alignment table:
import json, datetime
import lago # Life/Lago Python client
scores = {}
for path in ["llama32_align.json", "mistral7b_align.json"]:
d = json.load(open(path))
scores[d["encoder"]] = {
"modality": d["modality"],
"alignment_score": d["alignment_score"],
"top_regions": d["top_regions"],
"evaluated_at": datetime.datetime.utcnow().isoformat(),
}
lago.write("broomva.arcan.encoder_alignment", scores)Step 2 — Declare Routing Weights in Arcan Config
Add alignment-driven routing to ~/.config/arcan/routing.toml:
[routing.text]
strategy = "neuro_alignment"
alignment_table = "broomva.arcan.encoder_alignment"
modality = "text"
fallback = "meta-llama/Llama-3.2-3B"
min_score = 0.15 # reject encoders below this threshold
[routing.video]
strategy = "neuro_alignment"
alignment_table = "broomva.arcan.encoder_alignment"
modality = "video"
fallback = "facebook/vjepa2-vitg-fpc64-256"
min_score = 0.20
[routing.audio]
strategy = "neuro_alignment"
alignment_table = "broomva.arcan.encoder_alignment"
modality = "audio"
fallback = "facebook/w2v-bert-2.0"
min_score = 0.10Step 3 — Routing Logic (Pseudocode)
# arcan/src/routing/neuro_alignment.py
def select_encoder(task: Task, alignment_table: dict) -> str:
"""Return the highest-alignment encoder for this task's modality."""
modality = task.modality # "text", "video", or "audio"
candidates = {
enc: data["alignment_score"]
for enc, data in alignment_table.items()
if data["modality"] == modality
and data["alignment_score"] >= MIN_SCORE[modality]
}
if not candidates:
return FALLBACK[modality]
return max(candidates, key=candidates.get)
# Called at every task dispatch:
encoder = select_encoder(task, lago.read("broomva.arcan.encoder_alignment"))
result = arcan.run(task, encoder=encoder)Step 4 — Re-Evaluation Triggers
| Trigger | Action |
|---|---|
| New model release | Run align_encoder.py, update Lago table |
| Fine-tune completes | Re-run alignment; validate score did not degrade |
| Score staleness > 30 days | Scheduled re-evaluation via Autonomic |
| Alignment score drops > 0.05 | Alert via Autonomic + rollback to previous encoder |
# Autonomic watchdog (add to autonomic/config/watches.toml):
# [watch.encoder_alignment]
# table = "broomva.arcan.encoder_alignment"
# check = "alignment_score"
# threshold_drop = 0.05
# action = "rollback_and_alert"Alignment Score Interpretation
| R² Range | Label | Interpretation | Action |
|---|---|---|---|
| > 0.40 | Excellent | Encoder matches cortical representations at TRIBE v2 baseline level | Use as primary encoder |
| 0.25 – 0.40 | Good | Meaningful alignment; encoder captures most modality-relevant features | Use; monitor over time |
| 0.10 – 0.25 | Moderate | Partial alignment; encoder may miss higher-level semantic features | Use only if no better option |
| < 0.10 | Poor | Near-random; encoder does not capture brain-relevant information | Do not use for this modality |
Important caveats:
- Scores are population-average predictions from TRIBE v2's training cohort. Individual subject variability can shift scores ±0.05.
- TRIBE v2 operates on 5-second temporal windows. Encoders that produce token-level representations need temporal pooling before probing.
- The linear ridge regression probe (see
scripts/align_encoder.py) measures linear decodability, not representational isomorphism. High R² means the encoder's representations are linearly predictive of cortical activity, which is the standard encoding model benchmark in computational neuroscience. - License constraint: TRIBE v2 is CC BY-NC 4.0. Alignment scores derived from it cannot be used in commercial products without a separate agreement with Meta.
Reference Files
- references/encoder-alignment.md — Methodology, known baseline scores, modality-to-region mapping, limitations
- references/arcan-integration.md — Full integration guide, TOML config schema, Lago caching, re-evaluation workflow
Arcan Integration — Neuro-Informed Model Routing
How to wire TRIBE v2 alignment scores into the Life/Arcan model routing layer for biologically-informed encoder selection at task dispatch time.
---
1. Overview
Arcan is the AI orchestration layer in the Life monorepo (core/life/arcan/). It receives tasks from aiOS, selects the appropriate model/encoder, calls tools, and returns results. Today, routing decisions are driven by:
- Task type (generation, embedding, classification)
- Cost budget
- Latency SLA
- Model capability flags (context window, multimodal support)
TRIBE v2 alignment scores add a fifth dimension: biological plausibility — does this encoder represent the stimulus the way a human brain would? Encoders with higher language cortex alignment tend to generalize better to novel linguistic inputs, while high visual cortex alignment predicts better zero-shot scene understanding.
The integration flow:
New model released
|
v
run align_encoder.py --> alignment_score JSON
|
v
lago.write("arcan.encoder_alignment")
|
v
Arcan routing reads alignment table at task dispatch
|
v
Route to highest-alignment encoder meeting cost/latency constraintsAlignment scores are computed offline (not per-request) and cached in Lago. The routing decision adds negligible latency (a single Lago table read, typically < 1ms local).
---
2. Configuration Format
Declare alignment-aware routing in Arcan's routing configuration file (~/.config/arcan/routing.toml or core/life/arcan/config/routing.toml):
# routing.toml — Arcan model routing configuration
[routing.defaults]
cost_budget = "medium" # low / medium / high
latency_sla_ms = 2000
enable_neuro_alignment = true # set false to disable alignment-based routing
# Text task routing
[routing.text]
strategy = "neuro_alignment"
alignment_table = "broomva.arcan.encoder_alignment"
modality = "text"
fallback_model = "meta-llama/Llama-3.2-3B"
min_alignment_score = 0.15 # reject models below this R^2
# Video task routing
[routing.video]
strategy = "neuro_alignment"
alignment_table = "broomva.arcan.encoder_alignment"
modality = "video"
fallback_model = "facebook/vjepa2-vitg-fpc64-256"
min_alignment_score = 0.20
# Audio task routing
[routing.audio]
strategy = "neuro_alignment"
alignment_table = "broomva.arcan.encoder_alignment"
modality = "audio"
fallback_model = "facebook/w2v-bert-2.0"
min_alignment_score = 0.10
# Model registry: declare all available encoders with their metadata
[models.text]
"meta-llama/Llama-3.2-3B" = { cost = "low", latency = "medium", context = 131072 }
"mistralai/Mistral-7B-v0.1" = { cost = "medium", latency = "medium", context = 32768 }
"bert-base-uncased" = { cost = "low", latency = "low", context = 512 }
[models.video]
"facebook/vjepa2-vitg-fpc64-256" = { cost = "high", latency = "high", fps = 64 }
"openai/clip-vit-large-patch14" = { cost = "medium", latency = "medium", fps = 1 }
"MCG-NJU/videomae-huge" = { cost = "high", latency = "high", fps = 16 }
[models.audio]
"facebook/w2v-bert-2.0" = { cost = "medium", latency = "medium" }
"facebook/wav2vec2-large-960h" = { cost = "medium", latency = "medium" }YAML Alternative
If your Arcan config uses YAML:
routing:
defaults:
cost_budget: medium
latency_sla_ms: 2000
enable_neuro_alignment: true
text:
strategy: neuro_alignment
alignment_table: broomva.arcan.encoder_alignment
modality: text
fallback_model: meta-llama/Llama-3.2-3B
min_alignment_score: 0.15
video:
strategy: neuro_alignment
alignment_table: broomva.arcan.encoder_alignment
modality: video
fallback_model: facebook/vjepa2-vitg-fpc64-256
min_alignment_score: 0.20
audio:
strategy: neuro_alignment
alignment_table: broomva.arcan.encoder_alignment
modality: audio
fallback_model: facebook/w2v-bert-2.0
min_alignment_score: 0.10---
3. Routing Logic
The neuro-alignment routing strategy implemented in Arcan:
# core/life/arcan/src/routing/neuro_alignment.py
from dataclasses import dataclass
from typing import Optional
import lago # Life/Lago Python client
MIN_SCORE = {"text": 0.15, "video": 0.20, "audio": 0.10}
FALLBACK_MODEL = {
"text": "meta-llama/Llama-3.2-3B",
"video": "facebook/vjepa2-vitg-fpc64-256",
"audio": "facebook/w2v-bert-2.0",
}
@dataclass
class Task:
modality: str # "text", "video", "audio"
cost_budget: str # "low", "medium", "high"
latency_sla_ms: int
requires_language: bool = False
requires_motion: bool = False
def select_encoder(task: Task, alignment_table: dict, model_registry: dict) -> str:
"""
Select the highest-alignment encoder that fits within cost/latency constraints.
Falls back to the configured fallback model if no alignment data is available.
"""
modality = task.modality
min_score = MIN_SCORE.get(modality, 0.0)
available_models = model_registry.get(modality, {})
# Filter: must have alignment data and meet minimum score
candidates = []
for model_id, meta in available_models.items():
if model_id not in alignment_table:
continue
score = alignment_table[model_id].get("alignment_score", 0.0)
if score < min_score:
continue
# Filter by cost budget
model_cost = meta.get("cost", "high")
if not _cost_fits(model_cost, task.cost_budget):
continue
# Filter by latency
model_latency_ms = meta.get("latency_ms", 5000)
if model_latency_ms > task.latency_sla_ms:
continue
candidates.append((model_id, score))
if not candidates:
return FALLBACK_MODEL.get(modality, "meta-llama/Llama-3.2-3B")
# Sort by alignment score descending; pick the top candidate
candidates.sort(key=lambda x: x[1], reverse=True)
selected, score = candidates[0]
return selected
def _cost_fits(model_cost: str, budget: str) -> bool:
"""Return True if model cost tier fits within the task budget."""
tiers = {"low": 0, "medium": 1, "high": 2}
return tiers.get(model_cost, 2) <= tiers.get(budget, 1)
# Called at task dispatch time in Arcan's main routing loop:
def route_task(task: Task) -> str:
alignment_table = lago.read("broomva.arcan.encoder_alignment")
model_registry = lago.read("broomva.arcan.model_registry")
return select_encoder(task, alignment_table, model_registry)Decision Tree
Task arrives at Arcan dispatcher
├── Determine modality (text / video / audio)
├── Read alignment_table from Lago cache
│ └── If cache_age > 30 days: trigger async re-evaluation, use stale data
├── Filter candidates by min_alignment_score
├── Filter candidates by cost_budget and latency_sla
├── Sort remaining by alignment_score descending
├── Return top candidate
└── If no candidates: return FALLBACK_MODEL[modality]---
4. Score Caching in Lago
Alignment scores are stable until a new model version is released or fine-tuning occurs. Cache them in Lago's broomva.arcan.encoder_alignment table:
# scripts/push_alignment_to_lago.py
# Run after align_encoder.py to update the Lago routing table.
import json
import datetime
import pathlib
import lago
RESULTS_DIR = pathlib.Path("./results")
TABLE_NAME = "broomva.arcan.encoder_alignment"
def push_results():
current = {}
try:
current = lago.read(TABLE_NAME) or {}
except Exception:
pass # Table may not exist yet
for result_path in RESULTS_DIR.glob("*_align.json"):
data = json.loads(result_path.read_text())
encoder_id = data["encoder"]
current[encoder_id] = {
"alignment_score": data["alignment_score"],
"modality": data["modality"],
"interpretation": data["interpretation"],
"roi_label": data["roi_label"],
"top_regions": data["top_regions"],
"evaluated_at": datetime.datetime.utcnow().isoformat(),
"n_stimuli": data["n_stimuli"],
"tribe_model": data["tribe_model"],
}
lago.write(TABLE_NAME, current)
print("Pushed {} encoder alignment records to Lago.".format(len(current)))
for enc, rec in sorted(current.items(), key=lambda x: -x[1]["alignment_score"]):
print(" {:.3f} {} ({})".format(rec["alignment_score"], enc, rec["modality"]))
if __name__ == "__main__":
push_results()Lago Table Schema
broomva.arcan.encoder_alignment
├── key: encoder_id (str) -- HuggingFace model ID
├── alignment_score (float) -- mean R^2 from ridge probe
├── modality (str) -- text / video / audio
├── interpretation (str) -- poor / moderate / good / excellent
├── roi_label (str) -- language_cortex / visual_cortex / auditory_cortex
├── top_regions (list[dict]) -- top-5 vertices with highest alignment
├── evaluated_at (ISO timestamp) -- when the score was computed
├── n_stimuli (int) -- stimulus count used for evaluation
└── tribe_model (str) -- "facebook/tribev2"---
5. Re-Evaluation Triggers
| Trigger | Action | Automation |
|---|---|---|
| New model release (HuggingFace) | Run align_encoder.py for new model; push to Lago | Manual (prompted by BRO ticket) |
| Fine-tuning completes | Re-run alignment on fine-tuned checkpoint; compare to pre-tune score | CI hook in training pipeline |
| Score staleness > 30 days | Scheduled re-evaluation for all models in routing table | Autonomic watchdog |
| Alignment score drops > 0.05 vs previous | Alert + automatic rollback to previous best encoder | Autonomic watchdog |
| New modality added to task space | Run alignment for that modality's ROI | Manual |
Autonomic Watchdog Configuration
Add to core/life/autonomic/config/watches.toml:
[watch.encoder_alignment_staleness]
description = "Re-evaluate alignment if any encoder score is stale"
table = "broomva.arcan.encoder_alignment"
check_field = "evaluated_at"
max_age_days = 30
action = "trigger_skill"
skill = "tribe-v2-agent-alignment"
command = "python scripts/align_encoder.py --encoder-type {modality} --encoder-model {encoder_id} --stimulus-dir ./stimuli/{modality} --output ./results/{encoder_id_safe}_align.json"
[watch.encoder_alignment_regression]
description = "Alert and rollback if alignment score drops significantly"
table = "broomva.arcan.encoder_alignment"
check_field = "alignment_score"
threshold_drop = 0.05
action = "rollback_and_alert"
alert_channel = "agent-logs"
rollback_table = "broomva.arcan.encoder_alignment_history"---
6. End-to-End Workflow: New Model Integration
Full workflow when a new text encoder (e.g., LLaMA 3.3-8B) is released:
# Step 1: Pull the new model (or it will be downloaded by align_encoder.py)
huggingface-cli download meta-llama/Llama-3.3-8B --local-dir ./models/llama-3.3-8b
# Step 2: Run cortical alignment benchmark
python scripts/align_encoder.py \
--encoder-type text \
--encoder-model meta-llama/Llama-3.3-8B \
--stimulus-dir ./stimuli/text \
--output ./results/llama33_align.json \
--cv-splits 5
# Step 3: Inspect the result
python -c "
import json
r = json.load(open('./results/llama33_align.json'))
print('Encoder:', r['encoder'])
print('Alignment score:', r['alignment_score'], '(' + r['interpretation'] + ')')
print('Top regions:', r['top_regions'][:2])
"
# Step 4: Compare to current best encoder
python -c "
import json, glob
scores = {}
for p in glob.glob('./results/*_align.json'):
d = json.load(open(p))
if d['modality'] == 'text':
scores[d['encoder']] = d['alignment_score']
for enc, s in sorted(scores.items(), key=lambda x: -x[1]):
print(f'{s:.3f} {enc}')
"
# Step 5: Push to Lago
python scripts/push_alignment_to_lago.py
# Step 6: Arcan will automatically pick up the new scores on next task dispatch.
# Verify by checking the routing decision for a test task:
arcan route --task-type language_understanding --modality text --dry-run
# Expected output: Selected encoder: meta-llama/Llama-3.3-8B (score: 0.XXX)
# Step 7: Monitor for regressions via Autonomic (automatically watches alignment table)
arcan status --watch encoder_alignmentRollback Procedure
If alignment regression is detected after switching to a new model:
# View alignment history
lago query "SELECT encoder, alignment_score, evaluated_at FROM broomva.arcan.encoder_alignment_history ORDER BY evaluated_at DESC LIMIT 20"
# Manual rollback: restore previous best encoder to routing table
python -c "
import lago
history = lago.read('broomva.arcan.encoder_alignment_history')
current = lago.read('broomva.arcan.encoder_alignment')
# Find the previous best text encoder
prev_best = max(
((enc, rec) for enc, rec in history.items() if rec['modality'] == 'text'),
key=lambda x: x[1]['alignment_score']
)
current[prev_best[0]] = prev_best[1]
lago.write('broomva.arcan.encoder_alignment', current)
print('Rolled back to:', prev_best[0], 'score:', prev_best[1]['alignment_score'])
"---
7. Integration Testing
Before deploying alignment-driven routing to production, verify the integration:
# tests/test_arcan_alignment_routing.py
import pytest
from core.life.arcan.routing.neuro_alignment import select_encoder, Task
MOCK_ALIGNMENT_TABLE = {
"meta-llama/Llama-3.2-3B": {"alignment_score": 0.40, "modality": "text"},
"bert-base-uncased": {"alignment_score": 0.18, "modality": "text"},
"facebook/vjepa2-vitg": {"alignment_score": 0.38, "modality": "video"},
"openai/clip-vit-l14": {"alignment_score": 0.24, "modality": "video"},
}
MOCK_MODEL_REGISTRY = {
"text": {
"meta-llama/Llama-3.2-3B": {"cost": "low", "latency_ms": 500},
"bert-base-uncased": {"cost": "low", "latency_ms": 100},
},
"video": {
"facebook/vjepa2-vitg": {"cost": "high", "latency_ms": 3000},
"openai/clip-vit-l14": {"cost": "medium", "latency_ms": 500},
},
}
def test_selects_highest_alignment_text_encoder():
task = Task(modality="text", cost_budget="medium", latency_sla_ms=2000)
result = select_encoder(task, MOCK_ALIGNMENT_TABLE, MOCK_MODEL_REGISTRY)
assert result == "meta-llama/Llama-3.2-3B"
def test_falls_back_on_latency_constraint():
task = Task(modality="video", cost_budget="medium", latency_sla_ms=1000)
# vjepa2 latency_ms=3000 exceeds SLA; should select clip
result = select_encoder(task, MOCK_ALIGNMENT_TABLE, MOCK_MODEL_REGISTRY)
assert result == "openai/clip-vit-l14"
def test_falls_back_when_no_alignment_data():
task = Task(modality="audio", cost_budget="medium", latency_sla_ms=2000)
result = select_encoder(task, {}, {})
assert result == "facebook/w2v-bert-2.0" # hardcoded fallbackRun with: cargo test -p arcan (Rust) or pytest tests/test_arcan_alignment_routing.py (Python bindings).
Encoder Alignment — Methodology and Reference
Technical reference for interpreting cortical alignment scores produced by scripts/align_encoder.py.
---
1. Methodology: Linear Probing via Ridge Regression
Cortical alignment is measured using the linear encoding model paradigm from computational neuroscience. The protocol:
1. Encoder forward pass — Run each stimulus through the candidate AI encoder. Extract the final hidden state and mean-pool over the spatial/temporal dimension to get a single vector per stimulus. This yields a matrix X of shape (n_stimuli, encoder_dim).
2. TRIBE v2 forward pass — Run the same stimuli through TRIBE v2. Extract predicted fMRI activity over the modality-relevant ROI (e.g., language cortex vertices 12000-18000 for text). This yields Y of shape (n_stimuli, n_roi_vertices).
3. Ridge regression probe — Fit a ridge regression Y_hat = X @ W + b where W is (encoder_dim, n_roi_vertices). Train and test on disjoint stimulus sets via k-fold cross-validation (default 5 folds). Regularization prevents overfitting to encoder-specific quirks.
4. R-squared as alignment score — Report the mean coefficient of determination (R²) across vertices and CV folds. R² = 1 - (sum of squared residuals / total variance). This measures the fraction of TRIBE v2 cortical variance that the encoder's representations can linearly explain.
Why R² and Not Pearson r
R² accounts for both correlation and scaling. An encoder whose activations are correlated but differently scaled from cortical activity will score lower than one with matched magnitude. In practice, StandardScaler is applied to both X and Y before fitting, so the difference from using r vs. R² is small, but R² is the conventional reporting metric in encoding model benchmarks (see Scotti et al., Brain-Score 2, 2024).
Why Ridge and Not LASSO or OLS
Ridge regression handles the high-dimensional encoder_dim >> n_stimuli regime (common with LLMs) without overfitting. The regularization strength alpha defaults to 1.0; sweep over [0.01, 0.1, 1.0, 10.0, 100.0] and pick the best by inner CV when n_stimuli is large enough.
---
2. Known Alignment Scores (from TRIBE v2 paper and related work)
These are reference values from the TRIBE v2 publication (Benchetrit et al., 2025) and related encoding model studies. Use them to sanity-check your benchmark runs.
Text Encoders — Language Cortex (vertices 12000-18000)
| Encoder | Architecture | R² (approx) | Source |
|---|---|---|---|
| LLaMA 3.2-3B | Autoregressive LM | ~0.40 | TRIBE v2 paper |
| LLaMA 3.1-8B | Autoregressive LM | ~0.38-0.42 | Benchetrit et al. |
| Mistral 7B | Autoregressive LM | ~0.35-0.40 | Community benchmarks |
| GPT-2 (large) | Autoregressive LM | ~0.28-0.33 | Encoding model literature |
| BERT-base | Masked LM | ~0.18-0.22 | Multiple sources |
| RoBERTa-large | Masked LM | ~0.20-0.26 | Multiple sources |
| all-mpnet-base-v2 | Sentence BERT | ~0.12-0.18 | Approximate |
| Random linear encoder | Baseline | ~0.00-0.03 | Lower bound |
Key finding from TRIBE v2: Autoregressive LMs (LLaMA family) consistently outperform masked LMs (BERT family) on language cortex alignment. This matches the finding that language cortex processes text in a predictive (left-to-right) manner, not bidirectionally.
Video Encoders — Visual Cortex (vertices 1000-8000)
| Encoder | Architecture | R² (visual cortex) | Notes |
|---|---|---|---|
| V-JEPA2 (ViT-G) | Masked video prediction | High (~0.35-0.45) | TRIBE v2's native video encoder |
| VideoMAE-v2 (ViT-H) | Masked video prediction | ~0.30-0.40 | Strong motion encoding |
| CLIP ViT-L/14 | Contrastive image-text | ~0.20-0.28 | Good static visual, weak motion |
| DINO ViT-B/16 | Self-supervised image | ~0.15-0.22 | Limited temporal sensitivity |
| ResNet-50 | Supervised image classification | ~0.10-0.15 | Old baseline |
| Random CNN | Baseline | ~0.00-0.05 | Lower bound |
Key finding from TRIBE v2: Self-supervised video models trained with masked prediction (V-JEPA2, VideoMAE) strongly align with motion-selective cortex (MT/MST, vertices 5000-8000). CLIP-style models align better with ventral visual stream (V1-V4) than dorsal motion stream.
Audio Encoders — Auditory Cortex (vertices 8000-11000)
| Encoder | Architecture | R² (auditory cortex) | Notes |
|---|---|---|---|
| Wav2Vec-BERT 2.0 | Self-supervised speech | High (~0.30-0.40) | TRIBE v2's native audio encoder |
| wav2vec 2.0 (large) | Self-supervised speech | ~0.25-0.35 | Strong speech alignment |
| HuBERT (large) | Self-supervised speech | ~0.25-0.32 | Similar to wav2vec 2.0 |
| Whisper (encoder only) | Supervised ASR | ~0.20-0.28 | Task supervision reduces alignment |
| CLAP (audio-text) | Contrastive audio-text | ~0.15-0.22 | General audio, not speech-specific |
| Random waveform features | Baseline | ~0.00-0.04 | Lower bound |
---
3. Interpretation Thresholds
| R² Range | Label | Recommendation |
|---|---|---|
| > 0.40 | Excellent | Use as primary encoder for this modality in Arcan routing |
| 0.25 – 0.40 | Good | Use; monitor for regression after fine-tuning |
| 0.10 – 0.25 | Moderate | Consider as fallback; may lack higher-level semantic features |
| 0.05 – 0.10 | Poor | Not recommended; likely encoding only low-level statistical patterns |
| < 0.05 | Chance | Do not use; representations are not capturing modality-relevant information |
Practical guidance: A 0.05 R² difference between two encoders is typically not meaningful given fMRI noise. Treat differences < 0.05 as ties and prefer the faster/smaller model.
---
4. Modality-to-Region Mapping
The ROI vertex ranges below are approximate boundaries on the fsaverage5 surface (~20k vertices total, both hemispheres combined). Left hemisphere is vertices 0-9999, right hemisphere 10000-19999.
| Modality | ROI Label | Vertex Range | Cortical Areas |
|---|---|---|---|
| Text | language_cortex | 12000 – 18000 | IFG (Broca, ~15000-18000), STG posterior (Wernicke, ~13000-15000) |
| Video | visual_cortex | 1000 – 8000 | V1/V2 (~1000-3000), V3/V4 (~3000-5000), MT/MST motion (~5000-8000) |
| Audio | auditory_cortex | 8000 – 11000 | Primary auditory cortex A1 (~8000-10000), belt areas (~10000-11000) |
For more precise parcellation, use the HCP MMP1.0 atlas projected onto fsaverage5. The vertex ranges above are suitable for bulk alignment scoring but not for fine-grained region analysis.
Emergent Functional Networks
TRIBE v2 spontaneously recovers 5 functional brain networks from its training data alone. Verify your encoder activates the expected network:
| Network | ROI Vertices (approx) | Encoder That Best Drives It |
|---|---|---|
| Primary auditory | 8000 – 10000 | Wav2Vec-BERT 2.0 |
| Language | 12000 – 18000 | LLaMA 3.2-3B |
| Motion | 5000 – 8000 | V-JEPA2 |
| Default mode | 18000 – 20000 | Broad contextual encoders |
| Visual | 1000 – 5000 | V-JEPA2, CLIP |
If your encoder achieves high R² outside its expected network (e.g., a video encoder scoring high on language cortex vertices), investigate for data leakage or multimodal contamination in training.
---
5. Limitations
Population-Average Predictions
TRIBE v2 predicts population-average cortical responses from its training cohort. Individual subjects show ~0.05 R² variability around the population mean. Alignment scores therefore reflect how well an encoder matches the "average human", not any particular subject.
5-Second Temporal Window
TRIBE v2 processes stimuli in non-overlapping 5-second windows. Encoders that capture phenomena at shorter timescales (e.g., phoneme-level in audio) may be slightly disadvantaged. Temporal pooling (mean over the window) is applied before probing.
Linear Probe Assumption
Ridge regression measures linear decodability only. An encoder could have high non-linear alignment with cortex but low linear R². For a fuller picture, use representational similarity analysis (RSA) in addition to linear probing — but linear R² is sufficient for model selection.
Stimulus Distribution Shift
Alignment scores are sensitive to stimulus statistics. TRIBE v2 was trained on naturalistic video/audio/text; synthetic or highly structured stimuli (e.g., word lists, artificial tones) will yield lower and less reliable alignment estimates. Use naturalistic stimuli when possible.
CC BY-NC 4.0 License Constraint
TRIBE v2 is released under Creative Commons Attribution-NonCommercial 4.0. Alignment scores derived from its predictions cannot be used in commercial products or to gate commercial model routing decisions without a separate agreement with Meta. For internal R&D and open-source projects within the Broomva stack, CC BY-NC permits use.
Stimulus Count Requirement
Reliable R² estimates require at least 20 stimuli (more is better). With fewer than 10 stimuli, the cross-validation estimate has high variance; treat results as directional only. The TRIBE v2 paper used thousands of fMRI trial repetitions; align_encoder.py is a lightweight proxy using TRIBE v2 predictions as a surrogate ground truth.
#!/usr/bin/env python3
"""
align_encoder.py -- Compute cortical alignment score for any HuggingFace encoder
using TRIBE v2 as the ground-truth cortical predictor.
Methodology:
1. Load the target encoder (text/video/audio).
2. Load TRIBE v2 to predict cortical responses to the same stimuli.
3. Extract encoder hidden states and TRIBE v2 cortical predictions for each stimulus.
4. Fit a ridge regression probe: encoder hidden states -> TRIBE v2 predictions (modality ROI).
5. Report R-squared as the alignment score.
Usage:
python align_encoder.py \
--encoder-type text \
--encoder-model meta-llama/Llama-3.2-3B \
--stimulus-dir ./stimuli \
--output ./results/alignment.json
"""
import argparse
import json
import logging
import sys
import time
from pathlib import Path
from typing import Optional
import numpy as np
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)-8s %(message)s",
datefmt="%H:%M:%S",
)
log = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Cortical region vertex ranges on fsaverage5 (~20k vertices total)
# Based on standard atlas parcellations; left hemisphere offsets used here.
# ---------------------------------------------------------------------------
CORTICAL_ROI = {
"text": {
"label": "language_cortex",
"vertex_start": 12000,
"vertex_end": 18000,
"description": "Broca + Wernicke areas, left hemisphere",
},
"video": {
"label": "visual_cortex",
"vertex_start": 1000,
"vertex_end": 8000,
"description": "V1-V4 + MT/MST motion areas",
},
"audio": {
"label": "auditory_cortex",
"vertex_start": 8000,
"vertex_end": 11000,
"description": "Primary + belt auditory cortex, bilateral",
},
}
SUPPORTED_ENCODER_TYPES = ("text", "video", "audio")
TEXT_EXTENSIONS = {".txt", ".md"}
VIDEO_EXTENSIONS = {".mp4", ".avi", ".mov", ".mkv", ".webm"}
AUDIO_EXTENSIONS = {".wav", ".mp3", ".flac", ".ogg"}
EXTENSION_MAP = {
"text": TEXT_EXTENSIONS,
"video": VIDEO_EXTENSIONS,
"audio": AUDIO_EXTENSIONS,
}
# Approximate fsaverage5 region boundaries for top-region labeling
REGION_ATLAS = [
(1000, 3000, "V1/V2"),
(3000, 5000, "V3/V4"),
(5000, 8000, "MT/MST_motion"),
(8000, 10000, "primary_auditory"),
(10000, 11000, "auditory_belt"),
(11000, 13000, "posterior_STG"),
(13000, 15000, "Wernicke"),
(15000, 18000, "Broca"),
(18000, 20000, "parietal_association"),
]
# ---------------------------------------------------------------------------
# Stimulus discovery
# ---------------------------------------------------------------------------
def discover_stimuli(stimulus_dir: Path, encoder_type: str) -> list:
"""Return all stimulus files of the appropriate type from stimulus_dir."""
valid_exts = EXTENSION_MAP[encoder_type]
files = [
p for p in stimulus_dir.iterdir()
if p.is_file() and p.suffix.lower() in valid_exts
]
if not files:
raise FileNotFoundError(
"No {} stimuli found in {}. Expected extensions: {}".format(
encoder_type, stimulus_dir, sorted(valid_exts)
)
)
files.sort()
log.info("Found %d %s stimulus file(s) in %s", len(files), encoder_type, stimulus_dir)
return files
# ---------------------------------------------------------------------------
# TRIBE v2 cortical predictions
# ---------------------------------------------------------------------------
def load_tribe_model(cache_folder: str = "./cache"):
"""Load TRIBE v2 from HuggingFace (downloads on first call, ~10 GB)."""
try:
from tribev2 import TribeModel # type: ignore
except ImportError as exc:
raise ImportError(
"tribev2 package not found. Install with: pip install tribev2"
) from exc
log.info("Loading TRIBE v2 from facebook/tribev2 (may download ~10 GB on first run)...")
model = TribeModel.from_pretrained("facebook/tribev2", cache_folder=cache_folder)
log.info("TRIBE v2 loaded.")
return model
def get_tribe_predictions(tribe_model, stimuli: list, encoder_type: str, roi: dict) -> np.ndarray:
"""
Run TRIBE v2 on all stimuli and return mean ROI activation per stimulus.
Returns
-------
np.ndarray of shape (n_stimuli, n_vertices_in_roi)
"""
roi_slice = slice(roi["vertex_start"], roi["vertex_end"])
all_preds = []
for stimulus_path in stimuli:
log.info("TRIBE v2 predicting for: %s", stimulus_path.name)
kwargs = {}
if encoder_type == "text":
kwargs["text_path"] = str(stimulus_path)
elif encoder_type == "video":
kwargs["video_path"] = str(stimulus_path)
elif encoder_type == "audio":
kwargs["audio_path"] = str(stimulus_path)
events_df = tribe_model.get_events_dataframe(**kwargs)
preds, _segments = tribe_model.predict(events=events_df)
# preds: (n_timesteps, n_vertices) -- mean over time for a single stimulus
stimulus_mean = np.mean(preds, axis=0) # (n_vertices,)
all_preds.append(stimulus_mean[roi_slice])
return np.vstack(all_preds) # (n_stimuli, n_roi_vertices)
# ---------------------------------------------------------------------------
# Encoder hidden state extraction
# ---------------------------------------------------------------------------
def extract_text_hidden_states(model_id: str, stimuli: list) -> np.ndarray:
"""Extract mean-pooled last-layer hidden states from a HuggingFace text model."""
import torch
from transformers import AutoModel, AutoTokenizer
log.info("Loading text encoder: %s", model_id)
tokenizer = AutoTokenizer.from_pretrained(model_id, trust_remote_code=True)
model = AutoModel.from_pretrained(
model_id, trust_remote_code=True, output_hidden_states=True
)
model.eval()
device = "cuda" if torch.cuda.is_available() else "cpu"
model.to(device)
log.info("Text encoder loaded on %s.", device)
all_states = []
for path in stimuli:
text = path.read_text(encoding="utf-8").strip()
inputs = tokenizer(
text, return_tensors="pt", truncation=True, max_length=512
).to(device)
with torch.no_grad():
outputs = model(**inputs)
# Use last hidden state, mean-pool over token dimension
last_hidden = outputs.last_hidden_state # (1, seq_len, hidden_dim)
pooled = last_hidden.mean(dim=1).squeeze(0).cpu().numpy() # (hidden_dim,)
all_states.append(pooled)
return np.vstack(all_states) # (n_stimuli, hidden_dim)
def _sample_video_frames(path: Path, n_frames: int = 16) -> list:
"""Sample n_frames uniformly from a video file using OpenCV."""
try:
import cv2 # type: ignore
except ImportError as exc:
raise ImportError(
"OpenCV not found. Install with: pip install opencv-python-headless"
) from exc
cap = cv2.VideoCapture(str(path))
total = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
if total <= 0:
raise ValueError("Could not read frames from {}".format(path))
indices = np.linspace(0, total - 1, n_frames, dtype=int)
frames = []
for idx in indices:
cap.set(cv2.CAP_PROP_POS_FRAMES, int(idx))
ret, frame = cap.read()
if ret:
frames.append(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
cap.release()
if not frames:
raise ValueError("No frames could be decoded from {}".format(path))
return frames
def extract_video_hidden_states(model_id: str, stimuli: list) -> np.ndarray:
"""
Extract mean-pooled visual features from a video/image encoder.
For video inputs, uniformly sample frames and run the image encoder on each,
then average. Supports CLIP-style models via transformers and V-JEPA2 via its API.
"""
import torch
all_states = []
# Try V-JEPA2 first if model_id indicates it
if "vjepa" in model_id.lower():
try:
from vjepa2 import VJEPA2 # type: ignore
log.info("Loading V-JEPA2 encoder: %s", model_id)
vjepa = VJEPA2.from_pretrained(model_id)
vjepa.eval()
device = "cuda" if torch.cuda.is_available() else "cpu"
vjepa.to(device)
for path in stimuli:
log.info("V-JEPA2 encoding: %s", path.name)
features = vjepa.encode_video(str(path)) # (T, D) or (D,)
if hasattr(features, "ndim") and features.ndim == 2:
features = features.mean(axis=0)
all_states.append(np.array(features))
return np.vstack(all_states)
except ImportError:
log.warning("vjepa2 package not available; falling back to CLIP-style extraction.")
# General CLIP / ViT approach via transformers
from transformers import CLIPModel, CLIPProcessor, AutoFeatureExtractor, AutoModel
log.info("Loading video/image encoder via transformers: %s", model_id)
model = None
processor = None
use_clip = False
try:
processor = CLIPProcessor.from_pretrained(model_id)
model = CLIPModel.from_pretrained(model_id)
use_clip = True
except Exception:
processor = AutoFeatureExtractor.from_pretrained(model_id, trust_remote_code=True)
model = AutoModel.from_pretrained(model_id, trust_remote_code=True)
model.eval()
device = "cuda" if torch.cuda.is_available() else "cpu"
model.to(device)
from PIL import Image
for path in stimuli:
log.info("Encoding video frames from: %s", path.name)
frames = _sample_video_frames(path, n_frames=16)
frame_features = []
for frame in frames:
img = Image.fromarray(frame)
if use_clip:
inputs = processor(images=img, return_tensors="pt").to(device)
with torch.no_grad():
feat = model.get_image_features(**inputs) # (1, D)
else:
inputs = processor(images=img, return_tensors="pt").to(device)
with torch.no_grad():
out = model(**inputs)
feat = out.last_hidden_state.mean(dim=1) # (1, D)
frame_features.append(feat.squeeze(0).cpu().numpy())
video_rep = np.mean(frame_features, axis=0) # (D,)
all_states.append(video_rep)
return np.vstack(all_states)
def extract_audio_hidden_states(model_id: str, stimuli: list) -> np.ndarray:
"""Extract mean-pooled hidden states from a Wav2Vec-style audio encoder."""
import torch
from transformers import AutoProcessor, AutoModel
log.info("Loading audio encoder: %s", model_id)
processor = AutoProcessor.from_pretrained(model_id)
model = AutoModel.from_pretrained(model_id)
model.eval()
device = "cuda" if torch.cuda.is_available() else "cpu"
model.to(device)
log.info("Audio encoder loaded on %s.", device)
all_states = []
for path in stimuli:
log.info("Encoding audio: %s", path.name)
import soundfile as sf # type: ignore
waveform, sample_rate = sf.read(str(path))
if waveform.ndim == 2:
waveform = waveform.mean(axis=1) # mix to mono
inputs = processor(
waveform, sampling_rate=sample_rate, return_tensors="pt", padding=True
).to(device)
with torch.no_grad():
outputs = model(**inputs)
last_hidden = outputs.last_hidden_state # (1, T, D)
pooled = last_hidden.mean(dim=1).squeeze(0).cpu().numpy() # (D,)
all_states.append(pooled)
return np.vstack(all_states)
EXTRACTOR_MAP = {
"text": extract_text_hidden_states,
"video": extract_video_hidden_states,
"audio": extract_audio_hidden_states,
}
# ---------------------------------------------------------------------------
# Ridge regression probe
# ---------------------------------------------------------------------------
def compute_alignment_score(
encoder_states: np.ndarray,
tribe_predictions: np.ndarray,
n_splits: int = 5,
alpha: float = 1.0,
) -> tuple:
"""
Fit a ridge regression probe from encoder_states to tribe_predictions and
return (mean_r2, per_vertex_r2) across cross-validation splits.
Parameters
----------
encoder_states : (n_stimuli, encoder_dim)
tribe_predictions: (n_stimuli, n_roi_vertices)
n_splits : number of cross-validation folds
alpha : Ridge regularization strength
Returns
-------
mean_r2 : scalar float -- the alignment score
per_vertex_r2 : (n_roi_vertices,) array
"""
from sklearn.linear_model import Ridge
from sklearn.model_selection import KFold
from sklearn.preprocessing import StandardScaler
n_stimuli = encoder_states.shape[0]
if n_stimuli < n_splits:
# Not enough stimuli for full CV; use simple 2-fold
n_splits = max(2, n_stimuli // 2)
log.warning(
"Reduced CV splits to %d because only %d stimuli are available.", n_splits, n_stimuli
)
kf = KFold(n_splits=n_splits, shuffle=True, random_state=42)
scaler_X = StandardScaler()
scaler_y = StandardScaler()
all_r2_per_vertex = []
for fold_idx, (train_idx, test_idx) in enumerate(kf.split(encoder_states)):
X_train = scaler_X.fit_transform(encoder_states[train_idx])
X_test = scaler_X.transform(encoder_states[test_idx])
y_train = scaler_y.fit_transform(tribe_predictions[train_idx])
y_test = scaler_y.transform(tribe_predictions[test_idx])
ridge = Ridge(alpha=alpha, fit_intercept=True)
ridge.fit(X_train, y_train)
y_pred = ridge.predict(X_test)
# Per-vertex R^2: 1 - SS_res / SS_tot
ss_res = np.sum((y_test - y_pred) ** 2, axis=0)
ss_tot = np.sum((y_test - y_test.mean(axis=0)) ** 2, axis=0)
with np.errstate(divide="ignore", invalid="ignore"):
r2_vertex = np.where(ss_tot > 1e-10, 1.0 - ss_res / ss_tot, 0.0)
all_r2_per_vertex.append(r2_vertex)
log.info("Fold %d/%d mean R^2: %.4f", fold_idx + 1, n_splits, float(r2_vertex.mean()))
per_vertex_r2 = np.mean(all_r2_per_vertex, axis=0)
mean_r2 = float(np.mean(per_vertex_r2))
return mean_r2, per_vertex_r2
# ---------------------------------------------------------------------------
# Top region identification
# ---------------------------------------------------------------------------
def vertex_to_region_label(vertex: int) -> str:
"""Approximate a cortical region name from a global fsaverage5 vertex index."""
for start, end, label in REGION_ATLAS:
if start <= vertex < end:
return label
return "other_cortex"
def identify_top_regions(per_vertex_r2: np.ndarray, roi: dict, top_k: int = 5) -> list:
"""
Return the top_k vertices with highest alignment scores within the ROI.
Each entry: {"vertex": int, "r2": float, "region": str}
"""
offset = roi["vertex_start"]
top_local_indices = np.argsort(per_vertex_r2)[::-1][:top_k]
top_regions = []
for local_idx in top_local_indices:
global_vertex = int(offset + local_idx)
r2_val = float(per_vertex_r2[local_idx])
label = vertex_to_region_label(global_vertex)
top_regions.append({"vertex": global_vertex, "r2": round(r2_val, 4), "region": label})
return top_regions
# ---------------------------------------------------------------------------
# Main pipeline
# ---------------------------------------------------------------------------
def run_alignment(
encoder_type: str,
encoder_model: str,
stimulus_dir: Path,
output_path: Path,
tribe_cache: str = "./cache",
ridge_alpha: float = 1.0,
cv_splits: int = 5,
) -> dict:
"""Full alignment pipeline. Returns the results dict."""
start_time = time.time()
roi = CORTICAL_ROI[encoder_type]
log.info("=== TRIBE v2 Cortical Alignment ===")
log.info("Encoder type : %s", encoder_type)
log.info("Encoder model : %s", encoder_model)
log.info("Stimulus dir : %s", stimulus_dir)
log.info("Target ROI : %s (%s)", roi["label"], roi["description"])
# 1. Discover stimuli
stimuli = discover_stimuli(stimulus_dir, encoder_type)
# 2. Extract encoder hidden states
log.info("Extracting encoder hidden states...")
extractor = EXTRACTOR_MAP[encoder_type]
encoder_states = extractor(encoder_model, stimuli)
log.info("Encoder states shape: %s", encoder_states.shape)
# 3. Load TRIBE v2 and get cortical predictions
log.info("Running TRIBE v2 cortical predictions...")
tribe_model = load_tribe_model(cache_folder=tribe_cache)
tribe_preds = get_tribe_predictions(tribe_model, stimuli, encoder_type, roi)
log.info("TRIBE v2 predictions shape: %s", tribe_preds.shape)
# 4. Compute alignment score via ridge regression probe
log.info("Fitting ridge regression probe (alpha=%.3f, cv_splits=%d)...", ridge_alpha, cv_splits)
mean_r2, per_vertex_r2 = compute_alignment_score(
encoder_states, tribe_preds, n_splits=cv_splits, alpha=ridge_alpha
)
log.info("Alignment score (mean R-squared): %.4f", mean_r2)
# 5. Identify top cortical regions
top_regions = identify_top_regions(per_vertex_r2, roi, top_k=5)
# 6. Interpret score
if mean_r2 >= 0.40:
interpretation = "excellent"
elif mean_r2 >= 0.25:
interpretation = "good"
elif mean_r2 >= 0.10:
interpretation = "moderate"
else:
interpretation = "poor"
elapsed = time.time() - start_time
results = {
"encoder": encoder_model,
"encoder_type": encoder_type,
"modality": encoder_type,
"alignment_score": round(mean_r2, 4),
"interpretation": interpretation,
"roi_label": roi["label"],
"roi_description": roi["description"],
"top_regions": top_regions,
"n_stimuli": len(stimuli),
"cv_splits": cv_splits,
"ridge_alpha": ridge_alpha,
"encoder_dim": int(encoder_states.shape[1]),
"n_roi_vertices": int(tribe_preds.shape[1]),
"elapsed_seconds": round(elapsed, 1),
"tribe_model": "facebook/tribev2",
}
# Write output
output_path.parent.mkdir(parents=True, exist_ok=True)
output_path.write_text(json.dumps(results, indent=2))
log.info("Results written to: %s", output_path)
log.info("Final alignment score: %.4f (%s)", mean_r2, interpretation)
return results
def parse_args(argv: Optional[list] = None) -> argparse.Namespace:
parser = argparse.ArgumentParser(
prog="align_encoder",
description=(
"Compute cortical alignment score between an AI encoder and "
"TRIBE v2 cortical predictions via ridge regression probe."
),
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
python align_encoder.py --encoder-type text --encoder-model meta-llama/Llama-3.2-3B \\
--stimulus-dir ./stimuli/text --output ./results/llama_align.json
python align_encoder.py --encoder-type video --encoder-model openai/clip-vit-large-patch14 \\
--stimulus-dir ./stimuli/video --output ./results/clip_align.json
python align_encoder.py --encoder-type audio --encoder-model facebook/w2v-bert-2.0 \\
--stimulus-dir ./stimuli/audio --output ./results/wav2vec_align.json
Output JSON fields:
encoder - HuggingFace model ID
modality - text / video / audio
alignment_score - mean R-squared across CV folds (0.0 to 1.0)
interpretation - poor / moderate / good / excellent
top_regions - list of top-5 cortical vertices with highest alignment
n_stimuli - number of stimuli processed
""",
)
parser.add_argument(
"--encoder-type",
required=True,
choices=SUPPORTED_ENCODER_TYPES,
help="Modality of the encoder: text, video, or audio.",
)
parser.add_argument(
"--encoder-model",
required=True,
help="HuggingFace model ID or local path of the encoder to benchmark.",
)
parser.add_argument(
"--stimulus-dir",
required=True,
type=Path,
help="Directory containing stimulus files (.txt/.mp4/.wav etc.).",
)
parser.add_argument(
"--output",
required=True,
type=Path,
help="Path to write the JSON results file.",
)
parser.add_argument(
"--tribe-cache",
default="./cache",
help="Directory to cache TRIBE v2 model weights (default: ./cache).",
)
parser.add_argument(
"--ridge-alpha",
type=float,
default=1.0,
help="Ridge regression regularization strength (default: 1.0).",
)
parser.add_argument(
"--cv-splits",
type=int,
default=5,
help="Number of cross-validation folds (default: 5).",
)
return parser.parse_args(argv)
if __name__ == "__main__":
args = parse_args()
stimulus_dir = Path(args.stimulus_dir).expanduser().resolve()
if not stimulus_dir.exists():
log.error("Stimulus directory not found: %s", stimulus_dir)
sys.exit(1)
output_path = Path(args.output).expanduser().resolve()
try:
results = run_alignment(
encoder_type=args.encoder_type,
encoder_model=args.encoder_model,
stimulus_dir=stimulus_dir,
output_path=output_path,
tribe_cache=args.tribe_cache,
ridge_alpha=args.ridge_alpha,
cv_splits=args.cv_splits,
)
print(json.dumps(results, indent=2))
sys.exit(0)
except FileNotFoundError as exc:
log.error("Input error: %s", exc)
sys.exit(1)
except ImportError as exc:
log.error("Missing dependency: %s", exc)
sys.exit(1)
except Exception as exc:
log.exception("Unexpected error: %s", exc)
sys.exit(1)
Related skills
FAQ
What does the alignment score mean?
It measures how well an encoder's hidden states predict fMRI brain activity; R2 above 0.25 means representations resemble the cortex.
Which encoders can it score?
Any text, video, or audio encoder, including LLaMA, Mistral, BERT, V-JEPA2, and Wav2Vec.