File size: 5,033 Bytes
d9099a0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 | """Path featurization for the C / A / Q scorers.
Only the algorithmic-relevance model A consumes the file path. It is turned
into a fixed-width vector by a hashing trick and concatenated to the code
embedding:
features = [ embedding (1024) | path_hash (256) * weight (0.4) ]
The algorithm is named ``crc32_signed_tokens_legacy`` in the checkpoints.
"legacy" is a name inherited from the training pipeline, not a deprecation --
this is the only path featurizer, and every shipped scorer set uses it.
Two details matter for reproducing training-time behaviour exactly:
* Benchmark names (``leetcode``, ``humaneval``, ``mbpp``, ``codeforces`` ...)
are stripped from the path first, so the model cannot shortcut on them.
* Tokens are a bag of path components plus 3-character prefixes of the longer
ones. Position is discarded, which is what makes it tolerant of repository
layout differences.
This is reproduced verbatim from the training-time implementation so a scorer
set is self-contained.
"""
from __future__ import annotations
import json
import re
import zlib
from typing import Sequence
import numpy as np
PATH_HASH_ALGORITHM = "crc32_signed_tokens_legacy"
# --------------------------------------------------------------------------- #
# read repository-relative paths out of a pyarrow table
# --------------------------------------------------------------------------- #
def _path_from_meta(value: object) -> str:
if isinstance(value, str):
try:
value = json.loads(value)
except (TypeError, ValueError):
return ""
if isinstance(value, dict):
path = value.get("file_path")
return path if isinstance(path, str) else ""
return ""
def extract_relative_paths(table, path_col: str = "relative_path") -> list[str]:
"""Read paths from ``path_col``, falling back to ``meta.file_path``."""
count = table.num_rows
if path_col in table.column_names:
paths = [
"" if value is None else str(value)
for value in table.column(path_col).to_pylist()
]
else:
paths = [""] * count
if "meta" in table.column_names and not all(paths):
paths = [
path or _path_from_meta(meta)
for path, meta in zip(paths, table.column("meta").to_pylist())
]
return paths
# --------------------------------------------------------------------------- #
# benchmark-name stripping
# --------------------------------------------------------------------------- #
_SEPARATOR = r"[-_./\\\s]*"
_BENCHMARK_PATTERNS = (
rf"human{_SEPARATOR}eval(?:{_SEPARATOR}plus|{_SEPARATOR}x)?",
rf"mbpp(?:{_SEPARATOR}plus)?",
r"multipl[-_./\\\s]+e",
rf"ds{_SEPARATOR}1000",
rf"crux{_SEPARATOR}eval",
rf"big{_SEPARATOR}code{_SEPARATOR}bench",
rf"live{_SEPARATOR}code{_SEPARATOR}bench",
rf"code{_SEPARATOR}contests?",
r"leetcode",
r"codeforces",
r"atcoder",
r"acm",
)
_BENCHMARK_RE = re.compile(
rf"(?<![a-z0-9])(?:{'|'.join(_BENCHMARK_PATTERNS)})(?=$|[^a-z0-9])",
re.IGNORECASE,
)
_TOKEN_SPLIT = re.compile(r"[/\\._\-]+")
def sanitize_benchmark_path(relative_path: str) -> str:
"""Remove benchmark names so the model cannot shortcut on them."""
path = str(relative_path or "").replace("\\", "/").lower()
return _BENCHMARK_RE.sub("/", path)
# --------------------------------------------------------------------------- #
# hashing trick
# --------------------------------------------------------------------------- #
def path_tokens(relative_path: str) -> list[str]:
"""Lowercased path components plus 3-char prefixes of longer tokens."""
if not relative_path:
return []
raw = _TOKEN_SPLIT.split(sanitize_benchmark_path(relative_path))
tokens = [t for t in raw if t]
extra = [t[:3] for t in tokens if len(t) > 3]
return tokens + extra
def path_hash_vector(relative_path: str, dim: int) -> np.ndarray:
"""crc32 hashing-trick bag-of-tokens, L2-normalized, shape [dim]."""
vec = np.zeros(dim, dtype=np.float32)
for tok in path_tokens(relative_path):
encoded = tok.encode("utf-8")
bucket = zlib.crc32(encoded) % dim
sign = 1.0 if (zlib.crc32(b"s:" + encoded) & 1) == 0 else -1.0
vec[bucket] += sign
norm = float(np.linalg.norm(vec))
if norm > 0.0:
vec /= norm
return vec
def build_feature_matrix(
embeddings: Sequence[Sequence[float]],
paths: Sequence[str],
*,
path_hash_dim: int,
path_feature_weight: float,
) -> np.ndarray:
"""Build ``[ embedding | path_hash * weight ]``, shape [N, D+H]."""
rows = []
for embedding, path in zip(embeddings, paths):
emb = np.asarray(embedding, dtype=np.float32).reshape(-1)
hashed = path_hash_vector(path, path_hash_dim) * float(path_feature_weight)
rows.append(np.concatenate([emb, hashed.astype(np.float32)]))
if not rows:
return np.zeros((0, 0), dtype=np.float32)
return np.stack(rows).astype(np.float32)
|