Setting the file. One moment.
Anisotropy Check · Embedding Anisotropy Check · github/awesome-copilot · Skills Docs
Back to the top of the page scripts/ anisotropy_check.py
Python · 150 lines · 7 KB
15 Centring subtracts the mean of the unit-normalised rows of the fit population and normalises again. By default each
16 side is centred by its own mean, fitted on the compared rows themselves; pass --fit-a and --fit-b to fit on a train
17 split of the same domain, model and modality instead, and --shared when A and B come from one space and one
18 population. Needs numpy only; prints one JSON object.
19 """
20 import argparse
21 import json
22 import sys
23
24 import numpy as np
25
26 BLOCK = 2048
27 # Below this a norm or a spread is rounding noise; cosines are bounded by 1.
28 TINY = 1e-9
29
30
31 def load_rows (path, what):
32 x = np.load(path).astype(np.float64)
33 if x.ndim != 2 :
34 sys.exit( f " { what } must be an (n, d) array, got shape { x.shape } " )
35 if len (x) == 0 :
36 sys.exit( f " { what } must contain at least one row" )
37 bad = ~ np.isfinite(x).all( axis = 1 )
38 if bad.any():
39 sys.exit( f " { what } has { int (bad.sum()) } row(s) with NaN or infinite values; drop or fix them first" )
40 zero = np.linalg.norm(x, axis = 1 ) < 1e-12
41 if zero.any():
42 sys.exit( f " { what } has { int (zero.sum()) } zero-length row(s), which have no direction; drop them first" )
43 return x
44
45
46 def unit (x):
47 return x / np.linalg.norm(x, axis = 1 , keepdims = True )
48
49
50 def anisotropy (u):
51 """Mean cosine over all pairs of different rows of unit rows u: near 0 with no shared direction, near 1 for one."""
52 n = len (u)
53 s = u.sum( 0 )
54 return float ((s @ s - n) / (n * (n - 1 )))
55
56
57 def derangement (n, rng):
58 while True :
59 p = rng.permutation(n)
60 if not np.any(p == np.arange(n)):
61 return p
62
63
64 def partner_first (ua, ub):
65 """Share of A rows whose own partner is the nearest B row, in blocks of both sides so memory stays bounded."""
66 hits = 0
67 for i in range ( 0 , len (ua), BLOCK ):
68 qa = ua[i:i + BLOCK ]
69 best, arg = np.full( len (qa), - np.inf), np.zeros( len (qa), dtype = np.int64)
70 for j in range ( 0 , len (ub), BLOCK ):
71 s = qa @ ub[j:j + BLOCK ].T
72 k = s.argmax( axis = 1 )
73 v = s[np.arange( len (qa)), k]
74 better = v > best
75 best[better], arg[better] = v[better], k[better] + j
76 hits += int (np.sum(arg == np.arange(i, i + len (qa))))
77 return hits / len (ua)
78
79
80 def compare (ua, ub, perms):
81 matched = np.sum(ua * ub, axis = 1 )
82 floor = np.concatenate([np.sum(ua * ub[p], axis = 1 ) for p in perms])
83 gap, sd = float (matched.mean() - floor.mean()), float (floor.std())
84 pooled = float (np.sqrt((matched.var() + floor.var()) / 2 ))
85 return { "matched" : float (matched.mean()), "floor" : float (floor.mean()), "gap" : gap,
86 "floor_sd" : sd, "gap_over_floor_sd" : float (gap / sd) if sd > TINY else None ,
87 "cohens_d" : float (gap / pooled) if pooled > TINY else None ,
88 "partner_ranked_first" : float (partner_first(ua, ub))}
89
90
91 def main ():
92 ap = argparse.ArgumentParser( description = __doc__ , formatter_class = argparse.RawDescriptionHelpFormatter)
93 ap.add_argument( "a" , help = "(n, d) .npy, row i paired with row i of B" )
94 ap.add_argument( "b" , help = "(n, d) .npy in the same coordinate space as A" )
95 ap.add_argument( "--fit-a" , help = "rows to fit A's mean on; default A itself" )
96 ap.add_argument( "--fit-b" , help = "rows to fit B's mean on; default B itself" )
97 ap.add_argument( "--shared" , action = "store_true" , help = "one mean over both sides' fit rows" )
98 ap.add_argument( "--perms" , type = int , default = 20 , help = "derangements for the floor (default 20)" )
99 ap.add_argument( "--seed" , type = int , default = 0 )
100 o = ap.parse_args()
101 if o.perms < 1 :
102 ap.error( "--perms must be at least 1" )
103 a, b = load_rows(o.a, "A" ), load_rows(o.b, "B" )
104 if a.shape != b.shape or len (a) < 3 :
105 sys.exit( f "A and B must be paired arrays of one shape with n >= 3, got { a.shape } and { b.shape } " )
106 rng = np.random.default_rng(o.seed)
107 perms = [derangement( len (b), rng) for _ in range (o.perms)]
108 ua, ub = unit(a), unit(b)
109 fa = unit(load_rows(o.fit_a, "--fit-a" )) if o.fit_a else ua
110 fb = unit(load_rows(o.fit_b, "--fit-b" )) if o.fit_b else ub
111 for f, what in ((fa, "--fit-a" ), (fb, "--fit-b" )):
112 if f.shape[ 1 ] != a.shape[ 1 ]:
113 sys.exit( f " { what } has dimension { f.shape[ 1 ] } , the compared rows { a.shape[ 1 ] } " )
114 if o.shared:
115 mu_a = mu_b = np.vstack([fa, fb]).mean( 0 )
116 else :
117 mu_a, mu_b = fa.mean( 0 ), fb.mean( 0 )
118 ca, cb = ua - mu_a, ub - mu_b
119 for c, what in ((ca, "A" ), (cb, "B" )):
120 empty = np.linalg.norm(c, axis = 1 ) < TINY
121 if empty.any():
122 sys.exit( f "centring leaves { int (empty.sum()) } row(s) of { what } with no length: they equal the fitted mean, as when "
123 "every row points one way or the fit rows are a single row. Fit the mean on more rows, or drop those rows" )
124 raw = compare(ua, ub, perms)
125 cen = compare(unit(ca), unit(cb), perms)
126 an = { "a" : anisotropy(ua), "b" : anisotropy(ub)}
127 share = raw[ "floor" ] / raw[ "matched" ] if raw[ "matched" ] > 0 else None
128 source = { "a" : f "--fit-a, { len (fa) } rows" if o.fit_a else f "the compared rows, { len (a) } " ,
129 "b" : f "--fit-b, { len (fb) } rows" if o.fit_b else f "the compared rows, { len (b) } " }
130 means = ({ "shared" : f "one mean over both sides' fit rows ( { source[ 'a' ] } ; { source[ 'b' ] } )" } if o.shared
131 else { "a" : f "fitted on { source[ 'a' ] } " , "b" : f "fitted on { source[ 'b' ] } " })
132 apart = ( f " { cen[ 'gap_over_floor_sd' ] :.2f} floor sd apart" if cen[ "gap_over_floor_sd" ] is not None
133 else f "a gap of { cen[ 'gap' ] :.4g} , which has no size in floor sd because the floor has no spread" )
134 out = { "n" : len (a), "dim" : a.shape[ 1 ], "means" : means,
135 "anisotropy" : an, "raw" : raw, "centred" : cen, "partner_first_by_chance" : 1 / len (a),
136 "reading" : ( f "Unrelated pairs score { raw[ 'floor' ] :.4f} raw" + ( f ", { share :.0%} of the matched { raw[ 'matched' ] :.4f} " if share is not None else "" )
137 + f "; centred, matched { cen[ 'matched' ] :.4f} against a floor of { cen[ 'floor' ] :.4f} , "
138 f " { apart } . Report both the raw and centred pairs with their floors, never a raw matched cosine alone." )}
139 if max (an.values()) > 0.3 :
140 out[ "warning" ] = ( "anisotropy above 0.3 on at least one side: rows within that side cluster around a mean direction; "
141 "check raw.floor to see how much of that direction is shared across sides" )
142 notes = [ f " { side } 's mean is fitted on { len (x) } compared rows; fit it on a train split of the same population when you have one"
143 for side, given, x in (( "A" , o.fit_a, a), ( "B" , o.fit_b, b)) if not given and len (x) < 1000 ]
144 if notes:
145 out[ "note" ] = notes
146 print (json.dumps(out, indent = 1 , allow_nan = False ))
147
148
149 if __name__ == "__main__" :
150 main()