mirror of
https://github.com/teamchong/pxpipe.git
synced 2026-07-22 02:02:51 +02:00
teamchong
7af6c200c3
eval: - Add eval/lib/cci.py: drop-in interactive Claude session shim (pexpect/PTY) replacing headless `claude -p`. Drives the real TUI on Max-subscription auth (no ANTHROPIC_API_KEY), scrapes the reply + /context + /cost, and is argv/stdin/--output-format json compatible with the old call sites. - Wire every call site to cci.py: gist-recall (run/run2/run3), needle-haystack (sweep/crux), gsm8k (bench), swe-bench (run_pilot), swe-bench-pro (rep_nav/run_pro), and eval/lib/anthropic-client.mjs. - Isolation per call: --setting-sources project --strict-mcp-config --dangerously-skip-permissions. Recall evals keep --allowedTools Read; swe-bench keeps the full agentic toolset. Each trial is a fresh process. demo: - Resolve the `opus` shorthand to the full claude-opus-4-8[1m] model in cost-ab and effective-context a.sh/b.sh. setup.sh intentionally keeps the base id: the proxy scope gate strips [1m] from the request before matching, so [1m]-on-launch + base-in-scope is the correct pairing. docs: - FINDINGS.md: drop the 2026-06-16 Opus 4.8 re-test sections.
47 lines
2.2 KiB
Python
47 lines
2.2 KiB
Python
#!/usr/bin/env python3
|
|
# Gist-recall A/B runner. Both arms call claude CLI directly (proxy bypassed).
|
|
import json, os, subprocess, sys, glob
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
|
|
WORK = os.path.abspath('work')
|
|
MODEL = os.environ.get('MODEL', 'claude-fable-5')
|
|
CLAUDE = os.path.expanduser('~/.claude/local/claude')
|
|
CCI = os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', 'lib', 'cci.py')
|
|
probes = json.load(open(f'{WORK}/probes.json'))
|
|
env = {k: v for k, v in os.environ.items() if k != 'ANTHROPIC_BASE_URL'}
|
|
env['CCI_TIMEOUT'] = '210' # self-exit before the subprocess timeout=240 hard kill
|
|
|
|
def ask(prompt):
|
|
try:
|
|
r = subprocess.run([sys.executable, CCI, '--model', MODEL, '--allowedTools', 'Read', prompt],
|
|
capture_output=True, text=True, timeout=240, env=env)
|
|
return r.stdout.strip()
|
|
except Exception as e:
|
|
return f'<ERROR {e}>'
|
|
|
|
def one(job):
|
|
arm, p = job
|
|
sid, q = p['session'], p['q']
|
|
suffix = (f"\n\nQuestion: {q}\nIf the transcript does not contain the answer, "
|
|
f"reply exactly UNKNOWN. Reply with only the answer, nothing else.")
|
|
if arm == 'text':
|
|
prompt = (f"An earlier coding session transcript is saved at {WORK}/s{sid}.txt. "
|
|
f"Read that file (in chunks if needed, read all of it).") + suffix
|
|
else:
|
|
pngs = sorted(glob.glob(f'{WORK}/s{sid}_p*.png'))
|
|
prompt = (f"An earlier coding session transcript is rendered as {len(pngs)} images: "
|
|
+ ' '.join(pngs) + ". Read all of them in order; do not use any other tool "
|
|
"or write code, just read the images visually.") + suffix
|
|
ans = ask(prompt)
|
|
return dict(arm=arm, **p, answer=ans)
|
|
|
|
jobs = [(arm, p) for arm in ['text', 'image'] for p in probes]
|
|
print(f'{len(jobs)} calls, model {MODEL}', flush=True)
|
|
out = open(f'{WORK}/results.jsonl', 'w')
|
|
with ThreadPoolExecutor(max_workers=6) as ex:
|
|
for i, res in enumerate(ex.map(one, jobs)):
|
|
out.write(json.dumps(res) + '\n'); out.flush()
|
|
ok = '?'
|
|
print(f"[{i+1}/{len(jobs)}] {res['arm']:5s} s{res['session']} {res['type']:12s} -> {res['answer'][:60]!r}", flush=True)
|
|
print('done')
|