#!/usr/bin/env python3"""リポジトリを対象に、エージェントが実際に受け取るコンテキスト量を測る。1タスク = 1ファイルを編集する仕事。渡るコンテキストは、そのファイルとローカル依存（1ホップ）の合計とみなす。"""import os, re, sys, jsonimport tiktokenROOT = sys.argv[1]SRC = os.path.join(ROOT, "src")ENC = tiktoken.get_encoding("o200k_base")IMPORT_RE = re.compile(r"""(?:from|import)\s+['"]([^'"]+)['"]""")EXTS = (".ts", ".tsx")def files(): out = [] for d, _, fs in os.walk(SRC): for f in fs: if f.endswith(EXTS): out.append(os.path.join(d, f)) return sorted(out)def resolve(spec, origin): """インポート指定をローカルのファイルパスへ解決する。外部パッケージは None。""" if spec.startswith("@/"): base = os.path.join(SRC, spec[2:]) elif spec.startswith("."): base = os.path.normpath(os.path.join(os.path.dirname(origin), spec)) else: return None for cand in (base + ".ts", base + ".tsx", os.path.join(base, "index.ts"), os.path.join(base, "index.tsx")): if os.path.isfile(cand): return cand return NoneCACHE = {}def toks(path): if path not in CACHE: with open(path, encoding="utf-8", errors="replace") as fh: CACHE[path] = len(ENC.encode(fh.read())) return CACHE[path]def context(path): with open(path, encoding="utf-8", errors="replace") as fh: src = fh.read() deps = set() for spec in IMPORT_RE.findall(src): r = resolve(spec, path) if r and r != path: deps.add(r) return sorted(deps), len(ENC.encode(src))def pct(sorted_vals, p): k = (len(sorted_vals) - 1) * p / 100 lo, hi = int(k), min(int(k) + 1, len(sorted_vals) - 1) return sorted_vals[lo] + (sorted_vals[hi] - sorted_vals[lo]) * (k - lo)tasks = []for f in files(): deps, own = context(f) total = own + sum(toks(d) for d in deps) tasks.append({"file": os.path.relpath(f, ROOT), "deps": len(deps), "own": own, "ctx": total})vals = sorted(t["ctx"] for t in tasks)total_tokens = sum(vals)print(f"tasks={len(tasks)} total_ctx_tokens={total_tokens}")for p in (50, 75, 90, 95, 99): print(f" p{p} = {pct(vals, p):,.0f}")print(f" max = {vals[-1]:,} mean = {total_tokens/len(vals):,.0f}")top5 = max(1, round(len(vals) * 0.05))print(f"top {top5} tasks (5%) hold {sum(vals[-top5:])/total_tokens*100:.1f}% of all context tokens")print("\nlargest 5:")for t in sorted(tasks, key=lambda x: -x["ctx"])[:5]: print(f" {t['ctx']:>7,} deps={t['deps']:<3} {t['file']}")json.dump(tasks, open("tasks.json", "w"))
import jsontasks = json.load(open("tasks.json")) # ctxsize.py の出力TOTAL = sum(t["ctx"] for t in tasks)def task_class(path): if "/app/" in path and path.endswith(("page.tsx", "route.ts", "layout.tsx")): return "route" # 画面・API の実装 if "/components/" in path: return "component" if "/lib/" in path or "/config/" in path: return "lib" if "/i18n/" in path or "/generated/" in path: return "data" return "other"PRO_CLASSES = {"route", "lib"} # 「設計を伴う実装は上位モデル」という素朴なルールpro = [t for t in tasks if task_class(t["file"]) in PRO_CLASSES]pro_n, pro_tok = len(pro), sum(t["ctx"] for t in pro)print(f"[class] Pro {pro_n}/{len(tasks)} = {pro_n/len(tasks)*100:.1f}%" f" | tokens {pro_tok/TOTAL*100:.1f}%")# 同じ「件数」をサイズ順に選んだら、何%のトークンを掴めるかby_size = sorted(tasks, key=lambda t: -t["ctx"])[:pro_n]print(f"[size ] same count {pro_n}: tokens {sum(t['ctx'] for t in by_size)/TOTAL*100:.1f}%" f" (threshold = {by_size[-1]['ctx']:,})")
def cost(T, R, e=0.0, biased=False): pro = [t for t in tasks if t["ctx"] >= T] fla = [t for t in tasks if t["ctx"] < T] c = sum(t["ctx"] for t in pro) * R + sum(t["ctx"] for t in fla) if e > 0 and fla: k = max(1, round(len(fla) * e)) # biased=True: やり直しは大きいタスクに偏る、という前提 pick = sorted(fla, key=lambda t: -t["ctx"])[:k] if biased else \ sorted(fla, key=lambda t: t["file"])[:k] c += sum(t["ctx"] for t in pick) * R # Flash 分は課金済み、Pro 分を上乗せ return c
R = 10、閾値 6,000 のときの結果です。全部 Pro に送った場合を 100% とした相対値です。