|
#!/usr/bin/env -S uv run --script |
|
# /// script |
|
# requires-python = ">=3.13" |
|
# dependencies = [ |
|
# "sentence-transformers>=3.0", |
|
# "numpy>=1.26", |
|
# "tqdm>=4.66", |
|
# ] |
|
# /// |
|
""" |
|
Find the signature expressions of Claude, of Codex, and of you (Tal) across all |
|
your AI coding sessions: Claude Code (~/.claude/projects) and Codex (~/.codex). |
|
|
|
Two analyses, both built on sentence embeddings: |
|
|
|
1. RECURRING EXPRESSIONS (sentence level) |
|
Dedup sentences, embed them with a sentence-transformer, greedily cluster |
|
near-paraphrases by cosine similarity, then rank clusters by how often the |
|
expression (in any phrasing) recurs. Surfaces signature *turns of phrase* |
|
like "let me look this up instead of relying on memory". |
|
|
|
2. SIGNATURE PHRASES (n-gram level) |
|
Count 1..N-gram phrases per speaker, score each phrase by how |
|
*distinctive* it is to one speaker vs the other (weighted log-odds, |
|
Monroe et al. 2008), then embed + dedup the top phrases so variants |
|
("load bearing" / "load-bearing" / "is load bearing") collapse. |
|
Surfaces idioms like "load bearing", "smoke test", "spine", "hand waving". |
|
|
|
Usage: |
|
uv run analyze_ai_expressions.py # full run, Claude + Codex |
|
uv run analyze_ai_expressions.py --sources claude --limit-files 40 # quick look |
|
uv run analyze_ai_expressions.py --sources codex --out report # Codex only |
|
uv run analyze_ai_expressions.py --out report # also write md + json |
|
|
|
Speaker corpora: |
|
claude = Claude Code assistant text (add --include-thinking for reasoning) |
|
codex = Codex agent_message text (event_msg payloads only) |
|
tal = your typed user messages from both (slash-command / tool noise stripped) |
|
|
|
Distinctiveness is one-vs-rest: each speaker's phrases are scored against all the |
|
others combined, so with both assistants you see what's distinctly Claude vs |
|
distinctly Codex vs distinctly you. |
|
""" |
|
|
|
from __future__ import annotations |
|
|
|
import argparse |
|
import json |
|
import math |
|
import os |
|
import re |
|
import sys |
|
from collections import Counter, defaultdict |
|
from dataclasses import dataclass, field |
|
from pathlib import Path |
|
|
|
import numpy as np |
|
from tqdm import tqdm |
|
|
|
# ---------------------------------------------------------------------------- |
|
# Text cleaning & segmentation |
|
# ---------------------------------------------------------------------------- |
|
|
|
# XML-ish wrapper blocks that are harness noise, not anyone's prose. |
|
_NOISY_TAGS = ( |
|
"system-reminder", "command-message", "command-name", "command-args", |
|
"command-stdout", "local-command-stdout", "local-command-stderr", |
|
"task-notification", "task-id", "tool-use-id", "output-file", |
|
"environment_details", "function_results", "function_calls", |
|
"thinking", "attachment", "user-prompt-submit-hook", |
|
) |
|
_PAIRED_TAG_RE = re.compile( |
|
r"<(" + "|".join(_NOISY_TAGS) + r")\b[^>]*>.*?</\1>", re.DOTALL | re.IGNORECASE |
|
) |
|
_ANY_TAG_RE = re.compile(r"<[^>\n]{1,120}>") |
|
_FENCE_RE = re.compile(r"```.*?```", re.DOTALL) |
|
_INLINE_CODE_RE = re.compile(r"`[^`\n]*`") |
|
_URL_RE = re.compile(r"https?://\S+|www\.\S+") |
|
_EMAIL_RE = re.compile(r"\b[\w.\-]+@[\w.\-]+\b") |
|
_DOMAIN_RE = re.compile( |
|
r"\b[\w\-]+(?:\.[\w\-]+)*\.(?:com|org|net|io|dev|ai|sh|app|xyz|co|gov|edu|me|tv|" |
|
r"cloud|run|eth)\b", re.IGNORECASE) |
|
_PATH_RE = re.compile(r"(?:/[\w.\-]+){2,}/?") # /a/b/c style paths |
|
_MD_PREFIX_RE = re.compile(r"^[ \t]{0,8}(?:[-*+]\s+|\d+[.)]\s+|#{1,6}\s+|>\s+)", re.MULTILINE) |
|
_WS_RE = re.compile(r"\s+") |
|
|
|
# --- pasted tool / build / log output detection (per line) ------------------- |
|
# People paste git, test-runner and build logs into chat; that text is no one's |
|
# voice, so drop those lines before mining sentences and n-grams. |
|
_OUTPUT_GLYPHS = set("✓✔✗✘→←↑↓•·│┃┌┐└┘├┤─━❯❮»«✅❌⚠") |
|
# filename:line references (stack traces, console errors, lint output) |
|
_FILELINE_RE = re.compile( |
|
r"\b[\w\-]+\.(?:js|ts|tsx|jsx|mjs|cjs|py|go|rs|rb|java|json|ya?ml|toml|sh|sql|" |
|
r"html|css|vue|svelte|c|cpp|h):\d+", re.IGNORECASE) |
|
_GIT_NOISE_RE = re.compile( |
|
r"^\s*(?:Enumerating|Counting|Compressing|Writing|Resolving|Receiving|Unpacking|" |
|
r"Delta|Total)\b|^\s*remote:|^\s*To (?:github\.com|git@|https)|" |
|
r"\b[0-9a-f]{7,40}\.\.[0-9a-f]{7,40}\b", re.IGNORECASE) |
|
_TEST_NOISE_RE = re.compile( |
|
r"\(\d+\s+tests?\)|\b\d+\s*ms\b|\b\d+\s+(?:passed|failed|skipped|pending)\b|" |
|
r"^\s*(?:PASS|FAIL|ok|not ok|✓|✗)\b", re.IGNORECASE) |
|
_BRACKET_LOG_RE = re.compile(r"^\s*\[[\w.\-/]+\]") # "[run402-astro] wrote ..." |
|
_COLUMN_RE = re.compile(r"\S {2,}\S") # table-ish column gaps |
|
_FLAG_RE = re.compile(r"(?:^|\s)--[a-z][\w-]*") # CLI flag: --loopback, --json |
|
# shell prompt: "host % cmd", "user@host:", leading "$ "/"# " |
|
_SHELL_RE = re.compile(r"(?:^|\s)%\s+\S|^\s*[\w.\-]+@[\w.\-]+[:#~]|^\s*[$#]\s+\S") |
|
# command-execution / build output (Codex exec, npm, gh, browser, CI) |
|
_EXEC_OUTPUT_RE = re.compile( |
|
r"^\s*(?:Wall time:|Original token count|Process exited with code|Chunk ID:|" |
|
r"npm (?:notice|warn|error|WARN|ERR)|Triggered via push|Refreshing run status|" |
|
r"Press Ctrl\+C|Current URL:|WebSocket is (?:closed|closing)|" |
|
r"Process running with session|Some conversation entries were omitted)", |
|
re.IGNORECASE) |
|
_PATCH_RE = re.compile(r"^\*\*\* (?:Begin|End) Patch|^\*\*\* (?:Update|Add|Delete) File") |
|
_GIT_STATUS_RE = re.compile(r"^[MADRCU]{1,2} [\w./\-]+$") # porcelain: "M src/x.ts" |
|
# browser / JS console errors pasted from the in-app browser or extensions |
|
_CONSOLE_RE = re.compile( |
|
r"WebSocket (?:is closed|connection to)|Unchecked runtime\.lastError|" |
|
r"Cannot create item with duplicate id|Uncaught (?:TypeError|ReferenceError|Error)|" |
|
r"Failed to load resource|net::ERR_|Content Security Policy|style-src|" |
|
r"is not entitled|couldn't generate a task port|elapsedCPUTimeForFrontBoard|" |
|
r"app group identifier", re.IGNORECASE) |
|
|
|
|
|
# Unfenced code / JSON that people paste inline (not wrapped in ``` ```). |
|
_CODE_HINT_RE = re.compile( |
|
r"=>|[{};]|\b(?:const|let|var|function|return|await|import|export|typeof)\b|" |
|
r"JSON\.|console\.|\w+\.\w+\(|\)\s*[;{,]?\s*$") |
|
_JSON_KEY_RE = re.compile(r'"[^"]{1,40}"\s*[:,]') |
|
|
|
|
|
def is_code_line(line: str) -> bool: |
|
s = line.strip() |
|
if len(s) < 4: |
|
return False |
|
if _CODE_HINT_RE.search(s) or _JSON_KEY_RE.search(s): |
|
return True |
|
if sum(c in "{}[]();=<>" for c in s) >= 3: |
|
return True |
|
return False |
|
|
|
|
|
def is_output_line(line: str) -> bool: |
|
s = line.strip() |
|
if len(s) < 4: |
|
return False |
|
if any(ch in _OUTPUT_GLYPHS for ch in s): |
|
return True |
|
if (_GIT_NOISE_RE.search(s) or _TEST_NOISE_RE.search(s) |
|
or _BRACKET_LOG_RE.match(s) or _FILELINE_RE.search(s) |
|
or _EXEC_OUTPUT_RE.match(s) or _PATCH_RE.match(s) |
|
or _GIT_STATUS_RE.match(s) or _CONSOLE_RE.search(s)): |
|
return True |
|
if len(_COLUMN_RE.findall(s)) >= 2: |
|
return True |
|
if re.search(r"\S \| \S", s): # markdown / pipe-delimited table row |
|
return True |
|
if s.endswith("…") or s.endswith("..."): # CLI spinner / status / truncation |
|
return True |
|
if _FLAG_RE.search(s) or _SHELL_RE.search(s): # command line / shell prompt |
|
return True |
|
alpha = sum(c.isalpha() or c.isspace() for c in s) |
|
if len(s) >= 8 and alpha / len(s) < 0.55: # symbol/number-heavy → a log line |
|
return True |
|
return False |
|
|
|
# A message is pure harness/command noise if it's basically only these. |
|
_NOISE_MARKERS = ( |
|
"<command-name>", "<command-message>", "<task-notification>", |
|
"<local-command-stdout>", "<local-command-stderr>", |
|
"[Request interrupted", "Caveat: The messages below", |
|
) |
|
|
|
# Whole-message markers of automated review/approval prompts (a bot feeding the |
|
# agent, e.g. the Codex approval reviewer) — programmatic, not a human typing. |
|
_AUTOMATION_MARKERS = ( |
|
"codex agent history", "planned action json", "assess the exact planned action", |
|
"my request for codex", "continue the same review conversation", |
|
"reviewed codex session id", "read-only tool checks when local state", |
|
"the user has the in-app browser open", "the following is the codex agent", |
|
# templated workflow / bug-fix prompts fed to the agent (not a human typing) |
|
"you are not alone in the codebase", "never use command substitution", |
|
"if any step fails, do not commit", "run npm run lint from the repo root", |
|
# Codex automation config injection |
|
"automation memory:", "automation id:", |
|
) |
|
|
|
|
|
def is_automation_prompt(text: str) -> bool: |
|
low = text.lower() |
|
return any(m in low for m in _AUTOMATION_MARKERS) |
|
|
|
|
|
def is_noise_message(text: str) -> bool: |
|
"""True for slash-command invocations, task notifications, interrupts, and |
|
automated review/approval prompts — anything not actually typed by a human.""" |
|
head = text.lstrip()[:80] |
|
if any(m in head or m in text[:200] for m in _NOISE_MARKERS): |
|
return True |
|
return is_automation_prompt(text) |
|
|
|
|
|
# User-editable denylist of pasted/templated text that is no one's voice but |
|
# has no reliable structural signal (e.g. "proposed wording", "dns record"). |
|
# Populated from --stoplist; entries are normalized lowercase prefixes. |
|
_STOPLIST: list[str] = [] |
|
|
|
|
|
def norm_stoplisted(norm: str) -> bool: |
|
"""True if an already-normalized string matches a stoplist entry.""" |
|
return any(norm == e or norm.startswith(e + " ") for e in _STOPLIST) |
|
|
|
|
|
def is_stoplisted(line: str) -> bool: |
|
return bool(_STOPLIST) and norm_stoplisted(normalize(line)) |
|
|
|
|
|
def load_stoplist(path: Path) -> list[str]: |
|
if not path or not path.exists(): |
|
return [] |
|
out = [] |
|
for raw in path.read_text(encoding="utf-8").splitlines(): |
|
raw = raw.strip() |
|
if not raw or raw.startswith("#"): |
|
continue |
|
norm = normalize(raw) |
|
if norm: |
|
out.append(norm) |
|
return out |
|
|
|
|
|
def clean_text(text: str) -> str: |
|
"""Strip code, harness tags, urls, pasted logs and markdown decoration. |
|
|
|
Newlines are preserved as sentence boundaries; only spaces/tabs collapse. |
|
""" |
|
text = _FENCE_RE.sub("\n", text) |
|
text = _PAIRED_TAG_RE.sub(" ", text) |
|
text = _INLINE_CODE_RE.sub(" ", text) |
|
text = _ANY_TAG_RE.sub(" ", text) |
|
text = _URL_RE.sub(" ", text) |
|
text = _EMAIL_RE.sub(" ", text) |
|
text = _DOMAIN_RE.sub(" ", text) |
|
text = _PATH_RE.sub(" ", text) |
|
# drop pasted tool/build/log output, unfenced code, and stoplisted text |
|
kept = [ln for ln in text.splitlines() |
|
if not is_output_line(ln) and not is_code_line(ln) |
|
and not is_stoplisted(ln)] |
|
text = "\n".join(kept) |
|
text = _MD_PREFIX_RE.sub("", text) |
|
text = text.replace("*", "").replace("_", " ").replace("#", " ") |
|
text = re.sub(r"[ \t]+", " ", text) |
|
text = re.sub(r"\n{2,}", "\n", text) |
|
return text |
|
|
|
|
|
# Split into sentences: newlines are hard boundaries, then split on . ! ? ; : |
|
_SENT_SPLIT_RE = re.compile(r"(?<=[.!?;:])\s+|\n+") |
|
|
|
|
|
def split_sentences(text: str) -> list[str]: |
|
out: list[str] = [] |
|
for chunk in _SENT_SPLIT_RE.split(text): |
|
s = chunk.strip(" \t\r\n\"'`()[]{}-—–•|") |
|
if s: |
|
out.append(s) |
|
return out |
|
|
|
|
|
# words start with a letter; digits allowed inside so "e2e", "s3", "v1" stay whole |
|
_WORD_RE = re.compile(r"[a-z][a-z0-9'\-]*[a-z0-9]|[a-z]") |
|
|
|
|
|
def words(text: str) -> list[str]: |
|
return _WORD_RE.findall(text.lower()) |
|
|
|
|
|
def normalize(sentence: str) -> str: |
|
"""Canonical key for dedup: lowercase, de-punctuated, whitespace-collapsed.""" |
|
s = sentence.lower().strip() |
|
s = re.sub(r"[^\w'\- ]+", " ", s) |
|
s = _WS_RE.sub(" ", s).strip() |
|
return s |
|
|
|
|
|
def is_good_sentence(norm: str, min_words: int, max_words: int) -> bool: |
|
toks = norm.split() |
|
n = len(toks) |
|
if n < min_words or n > max_words: |
|
return False |
|
alpha = sum(1 for t in toks if t.isalpha()) |
|
if alpha < max(2, n // 2): # at least half real words, >=2 |
|
return False |
|
# reject lines dominated by ids / hex / numbers |
|
if sum(c.isdigit() for c in norm) > len(norm) * 0.3: |
|
return False |
|
return True |
|
|
|
|
|
# ---------------------------------------------------------------------------- |
|
# Stopwords (inline; no nltk download) |
|
# ---------------------------------------------------------------------------- |
|
|
|
STOPWORDS = set(""" |
|
a an and are as at be been being but by for from had has have he her his i if in |
|
into is it its me my no nor not of off on once only or our out over own re s so |
|
some such t than that the their them then there these they this those to too up |
|
us very was we were what when where which while who whom why will with would you |
|
your yours d ll m o ve ain aren couldn didn doesn don hadn hasn haven isn ma |
|
mightn mustn needn shan shouldn wasn weren won wouldn am about above after again |
|
against all any because before below between both did do does doing down during |
|
each few further here how itself more most other same she should under until |
|
""".split()) |
|
|
|
|
|
# ---------------------------------------------------------------------------- |
|
# Ingest |
|
# ---------------------------------------------------------------------------- |
|
|
|
@dataclass |
|
class Corpus: |
|
name: str |
|
# normalized sentence -> count, and -> first (raw_excerpt, session) |
|
sent_counts: Counter = field(default_factory=Counter) |
|
sent_example: dict = field(default_factory=dict) |
|
# n -> Counter(ngram_tuple_joined -> count) |
|
ngrams: dict = field(default_factory=lambda: defaultdict(Counter)) |
|
total_ngrams: dict = field(default_factory=lambda: defaultdict(int)) |
|
n_messages: int = 0 |
|
|
|
def add_sentence(self, raw: str, norm: str, session: str): |
|
self.sent_counts[norm] += 1 |
|
if norm not in self.sent_example: |
|
self.sent_example[norm] = (raw[:240], session) |
|
|
|
def add_ngrams(self, toks: list[str], nmax: int): |
|
L = len(toks) |
|
for n in range(1, nmax + 1): |
|
cnt = self.ngrams[n] |
|
tot = 0 |
|
for i in range(L - n + 1): |
|
gram = toks[i:i + n] |
|
if n == 1: |
|
if gram[0] in STOPWORDS or len(gram[0]) < 3: |
|
continue |
|
else: |
|
if gram[0] in STOPWORDS or gram[-1] in STOPWORDS: |
|
continue |
|
cnt[" ".join(gram)] += 1 |
|
tot += 1 |
|
self.total_ngrams[n] += tot |
|
|
|
|
|
def prune_singletons(corpus: Corpus, n: int, cap: int): |
|
"""Keep memory bounded: drop count==1 high-order n-grams when a vocab is huge.""" |
|
c = corpus.ngrams[n] |
|
if len(c) > cap: |
|
for g in [g for g, v in c.items() if v == 1]: |
|
del c[g] |
|
|
|
|
|
def merge_corpora(corpora: list[Corpus], name: str) -> Corpus: |
|
"""Combine several corpora into one (for one-vs-rest distinctiveness).""" |
|
m = Corpus(name) |
|
for c in corpora: |
|
m.sent_counts.update(c.sent_counts) |
|
for norm, ex in c.sent_example.items(): |
|
m.sent_example.setdefault(norm, ex) |
|
for n, cnt in c.ngrams.items(): |
|
m.ngrams[n].update(cnt) |
|
for n, t in c.total_ngrams.items(): |
|
m.total_ngrams[n] += t |
|
m.n_messages += c.n_messages |
|
return m |
|
|
|
|
|
# --- per-source line parsers: yield ("assistant"|"user", raw_text) ----------- |
|
|
|
def parse_claude(obj: dict, include_thinking: bool): |
|
"""Claude Code session line. assistant = text blocks; user = typed prose. |
|
|
|
Skips harness-injected content (isMeta = slash-command/skill bodies, |
|
isApiErrorMessage = rate-limit placeholders, isCompactSummary, sidechains). |
|
""" |
|
if obj.get("isMeta") or obj.get("isApiErrorMessage") or obj.get("isCompactSummary"): |
|
return |
|
msg = obj.get("message") |
|
if not isinstance(msg, dict): |
|
return |
|
role, content = msg.get("role"), msg.get("content") |
|
if role == "assistant" and isinstance(content, list): |
|
for b in content: |
|
if not isinstance(b, dict): |
|
continue |
|
if b.get("type") == "text": |
|
yield "assistant", b.get("text", "") |
|
elif include_thinking and b.get("type") == "thinking": |
|
yield "assistant", b.get("thinking", "") |
|
elif role == "user" and not obj.get("isSidechain"): |
|
if isinstance(content, str): |
|
if not is_noise_message(content): |
|
yield "user", content |
|
elif isinstance(content, list): |
|
for b in content: |
|
if isinstance(b, dict) and b.get("type") == "text": |
|
t = b.get("text", "") |
|
if not is_noise_message(t): |
|
yield "user", t |
|
|
|
|
|
def parse_codex(obj: dict, include_thinking: bool): |
|
"""Codex rollout line. Only event_msg payloads carry the final, de-duplicated |
|
human-readable text; response_item entries duplicate them with tool noise.""" |
|
if obj.get("type") != "event_msg": |
|
return |
|
p = obj.get("payload") |
|
if not isinstance(p, dict): |
|
return |
|
pt = p.get("type") |
|
if pt == "agent_message": |
|
yield "assistant", p.get("message", "") |
|
elif pt == "agent_reasoning" and include_thinking: |
|
yield "assistant", p.get("text", "") |
|
elif pt == "user_message": |
|
msg = p.get("message", "") |
|
# "[$skill](path)" is a Codex slash-command invocation, not prose |
|
if not is_noise_message(msg) and not msg.lstrip().startswith("[$"): |
|
yield "user", msg |
|
|
|
|
|
# source name -> (file finder, line parser, assistant-corpus name) |
|
SOURCES = { |
|
"claude": (lambda r: sorted(r.rglob("*.jsonl")), parse_claude, "claude"), |
|
"codex": (lambda r: sorted(r.rglob("rollout-*.jsonl")), parse_codex, "codex"), |
|
} |
|
|
|
|
|
def ingest(sources: list[str], roots: dict[str, Path], limit_files: int | None, |
|
include_thinking: bool, min_words: int, max_words: int, |
|
ngram_max: int) -> dict[str, Corpus]: |
|
"""Scan the given sources; route assistant text per-source, user text into 'tal'.""" |
|
corpora: dict[str, Corpus] = {"you": Corpus("you")} |
|
me = corpora["you"] |
|
prune_every, scanned = 50, 0 |
|
for src in sources: |
|
files_fn, parse_fn, asst_name = SOURCES[src] |
|
asst = corpora.setdefault(asst_name, Corpus(asst_name)) |
|
files = files_fn(roots[src]) |
|
if limit_files: |
|
files = files[:limit_files] |
|
for fpath in tqdm(files, desc=f"{src} sessions", unit="file"): |
|
session = fpath.stem |
|
try: |
|
with fpath.open(encoding="utf-8", errors="replace") as fh: |
|
for line in fh: |
|
line = line.strip() |
|
if not line: |
|
continue |
|
try: |
|
obj = json.loads(line) |
|
except json.JSONDecodeError: |
|
continue |
|
for speaker, raw in parse_fn(obj, include_thinking): |
|
if not raw or not raw.strip(): |
|
continue |
|
target = asst if speaker == "assistant" else me |
|
target.n_messages += 1 |
|
cleaned = clean_text(raw) |
|
target.add_ngrams(words(cleaned), ngram_max) |
|
for sent in split_sentences(cleaned): |
|
norm = normalize(sent) |
|
# re-apply line filters: sentence-splitting can |
|
# isolate output fragments (e.g. "Error: WebSocket…") |
|
if (is_good_sentence(norm, min_words, max_words) |
|
and not norm_stoplisted(norm) |
|
and not is_output_line(sent) |
|
and not is_code_line(sent)): |
|
target.add_sentence(sent, norm, session) |
|
except OSError: |
|
continue |
|
scanned += 1 |
|
if scanned % prune_every == 0: |
|
for corp in corpora.values(): |
|
for n in range(2, ngram_max + 1): |
|
prune_singletons(corp, n, cap=2_000_000) |
|
return corpora |
|
|
|
|
|
# ---------------------------------------------------------------------------- |
|
# Embedding + clustering |
|
# ---------------------------------------------------------------------------- |
|
|
|
def load_model(model_name: str, device: str | None): |
|
from sentence_transformers import SentenceTransformer |
|
import torch |
|
|
|
if device is None: |
|
if torch.backends.mps.is_available(): |
|
device = "mps" |
|
elif torch.cuda.is_available(): |
|
device = "cuda" |
|
else: |
|
device = "cpu" |
|
print(f"[embed] loading {model_name} on {device}", file=sys.stderr) |
|
return SentenceTransformer(model_name, device=device) |
|
|
|
|
|
def embed(model, texts: list[str], batch_size: int) -> np.ndarray: |
|
return model.encode( |
|
texts, |
|
batch_size=batch_size, |
|
convert_to_numpy=True, |
|
normalize_embeddings=True, |
|
show_progress_bar=True, |
|
).astype(np.float32) |
|
|
|
|
|
def cluster_by_similarity(emb: np.ndarray, threshold: float, block: int = 512) -> list[int]: |
|
"""Greedy union-find clustering on cosine similarity (embeddings are L2-normed).""" |
|
n = len(emb) |
|
parent = list(range(n)) |
|
|
|
def find(x: int) -> int: |
|
root = x |
|
while parent[root] != root: |
|
root = parent[root] |
|
while parent[x] != root: |
|
parent[x], x = root, parent[x] |
|
return root |
|
|
|
def union(a: int, b: int): |
|
ra, rb = find(a), find(b) |
|
if ra != rb: |
|
parent[max(ra, rb)] = min(ra, rb) |
|
|
|
for start in range(0, n, block): |
|
sims = emb[start:start + block] @ emb.T # (b, n) |
|
for i in range(sims.shape[0]): |
|
gi = start + i |
|
js = np.nonzero(sims[i] >= threshold)[0] |
|
for j in js: |
|
j = int(j) |
|
if j > gi: |
|
union(gi, j) |
|
return [find(i) for i in range(n)] |
|
|
|
|
|
# ---------------------------------------------------------------------------- |
|
# Analysis 1: recurring expressions (sentence clusters) |
|
# ---------------------------------------------------------------------------- |
|
|
|
@dataclass |
|
class ExprCluster: |
|
rep: str # representative raw sentence (most frequent member) |
|
count: int # total occurrences across the cluster |
|
variants: int # distinct normalized phrasings merged |
|
example: str # an illustrative raw excerpt |
|
session: str |
|
|
|
|
|
def recurring_expressions(corpus: Corpus, model, *, min_count: int, |
|
max_candidates: int, sim_threshold: float, |
|
batch_size: int) -> list[ExprCluster]: |
|
items = [(norm, c) for norm, c in corpus.sent_counts.items() if c >= min_count] |
|
items.sort(key=lambda x: x[1], reverse=True) |
|
items = items[:max_candidates] |
|
if not items: |
|
return [] |
|
norms = [n for n, _ in items] |
|
counts = {n: c for n, c in items} |
|
print(f"[{corpus.name}] embedding {len(norms)} recurring sentences", file=sys.stderr) |
|
emb = embed(model, norms, batch_size) |
|
labels = cluster_by_similarity(emb, sim_threshold) |
|
|
|
groups: dict[int, list[str]] = defaultdict(list) |
|
for norm, lab in zip(norms, labels): |
|
groups[lab].append(norm) |
|
|
|
clusters: list[ExprCluster] = [] |
|
for members in groups.values(): |
|
members.sort(key=lambda m: counts[m], reverse=True) |
|
total = sum(counts[m] for m in members) |
|
rep_norm = members[0] |
|
raw, session = corpus.sent_example.get(rep_norm, (rep_norm, "")) |
|
clusters.append(ExprCluster( |
|
rep=raw, count=total, variants=len(members), |
|
example=raw, session=session, |
|
)) |
|
clusters.sort(key=lambda c: c.count, reverse=True) |
|
return clusters |
|
|
|
|
|
# ---------------------------------------------------------------------------- |
|
# Analysis 2: signature phrases (distinctive n-grams, embedding-deduped) |
|
# ---------------------------------------------------------------------------- |
|
|
|
@dataclass |
|
class Phrase: |
|
text: str |
|
z: float |
|
own: int |
|
other: int |
|
merged: list[str] = field(default_factory=list) |
|
|
|
|
|
def weighted_log_odds(a: Corpus, b: Corpus, n: int, alpha: float) -> dict[str, tuple[float, int, int]]: |
|
"""Monroe et al. weighted log-odds with uninformative Dirichlet prior. |
|
|
|
Returns phrase -> (z, count_in_a, count_in_b). Positive z => distinctive to `a`. |
|
""" |
|
ca, cb = a.ngrams[n], b.ngrams[n] |
|
vocab = set(ca) | set(cb) |
|
a0 = alpha * len(vocab) |
|
na = a.total_ngrams[n] |
|
nb = b.total_ngrams[n] |
|
out: dict[str, tuple[float, int, int]] = {} |
|
for w in vocab: |
|
ya, yb = ca.get(w, 0), cb.get(w, 0) |
|
la = math.log((ya + alpha) / (na + a0 - ya - alpha)) |
|
lb = math.log((yb + alpha) / (nb + a0 - yb - alpha)) |
|
delta = la - lb |
|
var = 1.0 / (ya + alpha) + 1.0 / (yb + alpha) |
|
out[w] = (delta / math.sqrt(var), ya, yb) |
|
return out |
|
|
|
|
|
def signature_phrases(a: Corpus, b: Corpus, model, *, ngram_min: int, ngram_max: int, |
|
alpha: float, min_own: int, top_pool: int, |
|
sim_threshold: float, batch_size: int) -> list[Phrase]: |
|
"""Distinctive phrases of speaker `a` vs `b`, deduped via embeddings. |
|
|
|
ngram_min defaults to 2 so the result is multiword *expressions* ("load |
|
bearing", "root cause") rather than raw function-word frequency gaps |
|
("let", "now"). Set ngram_min=1 to also surface distinctive single words. |
|
""" |
|
scored: list[Phrase] = [] |
|
for n in range(ngram_min, ngram_max + 1): |
|
wlo = weighted_log_odds(a, b, n, alpha) |
|
for w, (z, ya, yb) in wlo.items(): |
|
if ya >= min_own and z > 0: |
|
scored.append(Phrase(text=w, z=z, own=ya, other=yb)) |
|
# higher-order phrases are inherently rarer; rank by z but give a mild |
|
# length bonus so multiword idioms aren't buried under unigrams |
|
scored.sort(key=lambda p: p.z + 0.15 * (len(p.text.split()) - 1), reverse=True) |
|
pool = scored[:top_pool] |
|
if not pool: |
|
return [] |
|
|
|
emb = embed(model, [p.text for p in pool], batch_size) |
|
labels = cluster_by_similarity(emb, sim_threshold) |
|
groups: dict[int, list[Phrase]] = defaultdict(list) |
|
for p, lab in zip(pool, labels): |
|
groups[lab].append(p) |
|
|
|
merged: list[Phrase] = [] |
|
for members in groups.values(): |
|
members.sort(key=lambda p: p.own, reverse=True) |
|
head = members[0] |
|
head.merged = [m.text for m in members[1:]] |
|
head.own = sum(m.own for m in members) |
|
head.other = sum(m.other for m in members) |
|
head.z = max(m.z for m in members) |
|
merged.append(head) |
|
merged.sort(key=lambda p: p.z + 0.15 * (len(p.text.split()) - 1), reverse=True) |
|
return merged |
|
|
|
|
|
# ---------------------------------------------------------------------------- |
|
# Reporting |
|
# ---------------------------------------------------------------------------- |
|
|
|
def fmt_expr_table(title: str, clusters: list[ExprCluster], top: int) -> str: |
|
lines = [f"\n## {title}", ""] |
|
for i, c in enumerate(clusters[:top], 1): |
|
v = f" (+{c.variants - 1} variants)" if c.variants > 1 else "" |
|
lines.append(f"{i:>2}. [{c.count:>4}×]{v} {c.rep}") |
|
return "\n".join(lines) |
|
|
|
|
|
def fmt_phrase_table(title: str, phrases: list[Phrase], top: int) -> str: |
|
lines = [f"\n## {title}", ""] |
|
for i, p in enumerate(phrases[:top], 1): |
|
extra = f" ~ {', '.join(p.merged[:3])}" if p.merged else "" |
|
ratio = f"{p.own}/{p.other}" if p.other else f"{p.own}/0" |
|
lines.append(f"{i:>2}. [z={p.z:5.1f} {ratio:>9}] “{p.text}”{extra}") |
|
return "\n".join(lines) |
|
|
|
|
|
def main(): |
|
ap = argparse.ArgumentParser(description=__doc__, |
|
formatter_class=argparse.RawDescriptionHelpFormatter) |
|
ap.add_argument("--sources", nargs="+", choices=["claude", "codex"], |
|
default=["claude", "codex"], |
|
help="which assistants' sessions to scan (default: both)") |
|
ap.add_argument("--claude-root", type=Path, |
|
default=Path.home() / ".claude" / "projects") |
|
ap.add_argument("--codex-root", type=Path, default=Path.home() / ".codex") |
|
ap.add_argument("--model", default="sentence-transformers/all-MiniLM-L6-v2") |
|
ap.add_argument("--device", default=None, help="cpu / mps / cuda (auto if unset)") |
|
ap.add_argument("--limit-files", type=int, default=None, |
|
help="only scan the first N session files per source (quick runs)") |
|
ap.add_argument("--include-thinking", action="store_true", |
|
help="also mine the assistants' reasoning, not just visible text") |
|
ap.add_argument("--top", type=int, default=30) |
|
ap.add_argument("--me-label", default="You", |
|
help="how to label your own (user) corpus in the report") |
|
ap.add_argument("--batch-size", type=int, default=256) |
|
# sentence-cluster knobs |
|
ap.add_argument("--min-count", type=int, default=3, |
|
help="min occurrences for a sentence to count as recurring") |
|
ap.add_argument("--max-candidates", type=int, default=8000, |
|
help="cap embedded sentences per speaker") |
|
ap.add_argument("--sent-sim", type=float, default=0.78) |
|
ap.add_argument("--min-words", type=int, default=2) |
|
ap.add_argument("--max-words", type=int, default=18) |
|
# phrase knobs |
|
ap.add_argument("--ngram-min", type=int, default=2, |
|
help="1 = also include distinctive single words") |
|
ap.add_argument("--ngram-max", type=int, default=4) |
|
ap.add_argument("--alpha", type=float, default=0.1, help="log-odds prior") |
|
ap.add_argument("--min-own", type=int, default=6, |
|
help="min count for a phrase to be a signature candidate") |
|
ap.add_argument("--phrase-pool", type=int, default=400, |
|
help="top distinctive phrases to embed+dedup per speaker") |
|
ap.add_argument("--phrase-sim", type=float, default=0.72) |
|
ap.add_argument("--stoplist", type=Path, |
|
default=Path(__file__).with_name("stoplist.txt"), |
|
help="file of phrases to drop (pasted/templated text); one per line") |
|
ap.add_argument("--out", type=Path, default=None, |
|
help="directory to also write report.md + report.json") |
|
args = ap.parse_args() |
|
|
|
roots = {"claude": args.claude_root, "codex": args.codex_root} |
|
sources = [s for s in args.sources if roots[s].exists()] |
|
if not sources: |
|
sys.exit("no session directories found for " + ", ".join(args.sources)) |
|
|
|
global _STOPLIST |
|
_STOPLIST = load_stoplist(args.stoplist) |
|
if _STOPLIST: |
|
print(f"[stoplist] {len(_STOPLIST)} phrases from {args.stoplist}", file=sys.stderr) |
|
|
|
corpora = ingest(sources, roots, args.limit_files, args.include_thinking, |
|
args.min_words, args.max_words, args.ngram_max) |
|
|
|
# speaker order: each assistant present, then Tal |
|
LABELS = {"claude": "Claude", "codex": "Codex", "you": args.me_label} |
|
speakers = [s for s in ("claude", "codex") if s in corpora] + ["you"] |
|
print("\nmessages mined " + " ".join( |
|
f"{s}={corpora[s].n_messages:,}" for s in speakers), file=sys.stderr) |
|
|
|
model = load_model(args.model, args.device) |
|
|
|
def recurring(name): |
|
return recurring_expressions( |
|
corpora[name], model, min_count=args.min_count, |
|
max_candidates=args.max_candidates, sim_threshold=args.sent_sim, |
|
batch_size=args.batch_size) |
|
|
|
def distinctive(name): |
|
# one-vs-rest: this speaker against everyone else combined |
|
rest = merge_corpora([corpora[o] for o in speakers if o != name], "rest") |
|
return signature_phrases( |
|
corpora[name], rest, model, ngram_min=args.ngram_min, |
|
ngram_max=args.ngram_max, alpha=args.alpha, min_own=args.min_own, |
|
top_pool=args.phrase_pool, sim_threshold=args.phrase_sim, |
|
batch_size=args.batch_size) |
|
|
|
rec = {s: recurring(s) for s in speakers} |
|
sig = {s: distinctive(s) for s in speakers} |
|
|
|
head = [ |
|
"# Signature expressions across your AI coding sessions", |
|
f"\nsources: {', '.join(sources)}", |
|
"messages: " + " ".join(f"{LABELS[s]}={corpora[s].n_messages:,}" for s in speakers), |
|
] |
|
body = [] |
|
for s in speakers: |
|
body.append(fmt_expr_table(f"{LABELS[s]} — top recurring expressions", |
|
rec[s], args.top)) |
|
for s in speakers: |
|
against = "the others" if len(speakers) > 2 else \ |
|
LABELS[next(o for o in speakers if o != s)] |
|
body.append(fmt_phrase_table( |
|
f"{LABELS[s]} — most distinctive phrases (vs {against})", sig[s], args.top)) |
|
report = "\n".join(head + body) |
|
print(report) |
|
|
|
if args.out: |
|
args.out.mkdir(parents=True, exist_ok=True) |
|
(args.out / "report.md").write_text(report, encoding="utf-8") |
|
payload = { |
|
"sources": sources, |
|
"messages": {s: corpora[s].n_messages for s in speakers}, |
|
"recurring": {s: [vars(c) for c in rec[s][:args.top]] for s in speakers}, |
|
"distinctive": {s: [vars(p) for p in sig[s][:args.top]] for s in speakers}, |
|
} |
|
(args.out / "report.json").write_text( |
|
json.dumps(payload, indent=2, ensure_ascii=False), encoding="utf-8") |
|
print(f"\nwrote {args.out}/report.md and report.json", file=sys.stderr) |
|
|
|
|
|
if __name__ == "__main__": |
|
main() |
