Setting the file. One moment.
Audit Intentions · Exploring MCP Tool Original User Motive · PostHog/skills · Skills Docs
Back to the top of the page scripts/audit_intentions.py
scripts/ audit_intentions.py
Python · 111 lines · 5 KB
16 and still worth separating — "verify a feature flag configuration" and "manage
17 feature flag rollout" sit near each other and belong in different themes, because
18 one is checking state and the other is changing it. That call is yours.
19
20 export OPENAI_API_KEY=$(op read "$(grep -E '^OPENAI_API_KEY=' .env.local | cut -d= -f2- | tr -d ' \" ')")
21 python audit_intentions.py facets.jsonl --threshold 0.80
22
23 Note the model name differs from the backend's `text-embedding-3-small-1536`.
24 That is a PostHog gateway alias carrying a dimension count; calling OpenAI
25 directly needs the plain model id.
26 """
27
28 import argparse
29 import json
30 import os
31 import sys
32 from collections import Counter
33 from pathlib import Path
34
35 import numpy as np
36 from openai import OpenAI
37
38 EMBEDDING_MODEL = "text-embedding-3-small"
39 BATCH = 256
40
41
42 def embed (client: OpenAI, texts: list[ str ]) -> np.ndarray:
43 vectors: list[list[ float ]] = []
44 for start in range ( 0 , len (texts), BATCH ):
45 chunk = texts[start : start + BATCH ]
46 response = client.embeddings.create( model = EMBEDDING_MODEL , input = chunk)
47 vectors.extend(item.embedding for item in response.data)
48 matrix = np.array(vectors, dtype = np.float64)
49 return matrix / np.linalg.norm(matrix, axis = 1 , keepdims = True )
50
51
52 def main () -> None :
53 parser = argparse.ArgumentParser( description = __doc__ )
54 parser.add_argument( "facets" , type = Path, help = "JSONL from extract_facets.py" )
55 parser.add_argument( "--threshold" , type = float , default = 0.80 , help = "cosine similarity to flag at" )
56 parser.add_argument( "--field" , default = "goal" , help = "field holding the intention" )
57 parser.add_argument( "--top" , type = int , default = 15 , help = "pairs to show when nothing crosses the threshold" )
58 args = parser.parse_args()
59
60 if not os.environ.get( "OPENAI_API_KEY" ):
61 sys.exit( "OPENAI_API_KEY is unset. Resolve it from .env.local with `op read` first — see the module docstring." )
62
63 records = [json.loads(line) for line in args.facets.read_text().splitlines() if line.strip()]
64 counts = Counter(r[args.field] for r in records)
65 intentions = sorted (counts)
66 print ( f " { len (records) } sessions, { len (intentions) } distinct intentions" )
67 if len (intentions) < 2 :
68 # No pairs exist, so there is nothing to audit. Say that rather than
69 # reporting a clean result, which would read as "checked, found nothing".
70 print ( "Fewer than two distinct intentions; no pairs to compare." )
71 return
72
73 client = OpenAI()
74 vectors = embed(client, intentions)
75 # Apple's Accelerate BLAS raises divide-by-zero and overflow on this matmul
76 # whatever the dtype, and the result is correct regardless — norms are 1.0
77 # and the diagonal comes back 1.0. Left unsuppressed the warnings sit right
78 # above a "nothing flagged" line and read like corrupt data.
79 with np.errstate( all = "ignore" ):
80 similarity = vectors @ vectors.T
81
82 ranked = sorted (
83 (
84 ( float (similarity[i, j]), intentions[i], intentions[j])
85 for i in range ( len (intentions))
86 for j in range (i + 1 , len (intentions))
87 ),
88 reverse = True ,
89 )
90 flagged = [p for p in ranked if p[ 0 ] >= args.threshold]
91
92 # Always show the closest pairs, threshold or not. An empty flag list means
93 # "nothing crossed the line you chose", which is not the same as "verified
94 # clean" — print the ceiling so the operator can judge the threshold itself.
95 shown = flagged or ranked[: args.top]
96 header = (
97 f " { len (flagged) } pairs at or above { args.threshold } , "
98 f "touching { sum (counts[a] + counts[b] for _, a, b in flagged) } sessions"
99 if flagged
100 else f "Nothing reached { args.threshold } . Closest { len (shown) } pairs, highest similarity { ranked[ 0 ][ 0 ] :.3f} "
101 )
102 print ( f " \n{ header } " )
103 print ( "Merging a pair moves both counts onto one label. Semantic closeness is" )
104 print ( "not a merge instruction — check that the two describe the same job. \n " )
105 for score, a, b in shown:
106 print ( f " { score :.3f} { a } ( { counts[a] } )" )
107 print ( f " { b } ( { counts[b] } )" )
108
109
110 if __name__ == "__main__" :
111 main()