Brummel
ded28e2f75
Ran the two core formats (plain Form-A, annotated parens) on Llama-3.1-405B (405B dense, ~135x Qwen3-Coder-Next's 3B active) to separate format-effect from small-model-effect. Scripts parametrized by PROBE_MODEL/PROBE_TAG (Qwen default preserved). Watch-out caught: Llama-3.1 on IONOS leaks its chat template (repeats "assistant", generates endless variants to max_tokens — an IONOS-documented quirk). First run read 0/8; that was the instrument. Fixed with a stop sequence and a balanced-first-module extractor. Result reverses the effect: Qwen3-Coder (3B): plain 6/8, annotated 7/8 Llama-405B (405B): plain 7/8, annotated 6/8 Llama solves L4 (Qwen's bracket wall) with plain parens, so that wall was a small-model effect. Annotated parens help the small model and hurt the large one (at L1 Llama invented a spurious (lit 3) wrapper under the annotated dialect). Both still fail L7 (SMA) — a genuine complexity ceiling. Synthesis: redundant explicit structure is a crutch for weak models. A frontier model tracks plain brackets fine and the unfamiliar scaffolding only adds noise. The right surface is model-relative. Synthesis in format-findings.md; raw ladders llama405-*.md.
184 lines
8.7 KiB
Python
184 lines
8.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Format probe 4 — annotated parens. Does giving every paren its nesting
|
|
depth (`(#N ` ... ` #N)`) help Qwen keep brackets balanced at depth?
|
|
|
|
The model writes the annotated dialect; we STRIP the annotations with a
|
|
round-trip-verified converter (see qwen_fmt4 verification in session) back to
|
|
plain Form-A, then `ail check` / `ail run`. Strip-variant: we trust the bare
|
|
parens, the annotation is the model's self-discipline scaffold. Same ladder as
|
|
qwen_ablation.py so results are comparable.
|
|
|
|
Reads $IONOS_API_TOKEN (never printed). Live IONOS — run with consent.
|
|
"""
|
|
from __future__ import annotations
|
|
import json, os, re, subprocess, urllib.request
|
|
from pathlib import Path
|
|
|
|
ENDPOINT = "https://openai.inference.de-txl.ionos.com/v1/chat/completions"
|
|
MODEL = os.environ.get("PROBE_MODEL","Qwen/Qwen3-Coder-Next")
|
|
REPO = Path(__file__).resolve().parents[2]
|
|
AIL = os.environ.get("AIL_BIN", str(REPO / "target/debug/ail"))
|
|
DOC = REPO / f"experiments/2026-05-12-cross-model-authoring/{os.environ.get('PROBE_TAG','qwen')}-fmt4.md"
|
|
|
|
def annotate(src):
|
|
out=[]; depth=0; in_str=False; esc=False
|
|
for ch in src:
|
|
if in_str:
|
|
out.append(ch)
|
|
if esc: esc=False
|
|
elif ch=='\\': esc=True
|
|
elif ch=='"': in_str=False
|
|
continue
|
|
if ch=='"': in_str=True; out.append(ch)
|
|
elif ch=='(': out.append(f'(#{depth} '); depth+=1
|
|
elif ch==')': depth-=1; out.append(f' #{depth})')
|
|
else: out.append(ch)
|
|
return ''.join(out)
|
|
|
|
def strip_ann(src):
|
|
out=[]; i=0; in_str=False; esc=False
|
|
while i<len(src):
|
|
ch=src[i]
|
|
if in_str:
|
|
out.append(ch)
|
|
if esc: esc=False
|
|
elif ch=='\\': esc=True
|
|
elif ch=='"': in_str=False
|
|
i+=1; continue
|
|
if ch=='"': in_str=True; out.append(ch); i+=1; continue
|
|
m=re.match(r'\(\s*#\d+\s+', src[i:])
|
|
if m: out.append('('); i+=m.end(); continue
|
|
m=re.match(r'\s+#\d+\s*\)', src[i:])
|
|
if m: out.append(')'); i+=m.end(); continue
|
|
out.append(ch); i+=1
|
|
return ''.join(out)
|
|
|
|
A = """(module demo_a
|
|
(fn double
|
|
(type (fn-type (params (own (con Int))) (ret (own (con Int)))))
|
|
(params x)
|
|
(body (app + x x)))
|
|
(fn main
|
|
(type (fn-type (params) (ret (own (con Unit))) (effects IO)))
|
|
(params)
|
|
(body (seq (app print (app double 21)) (do io/print_str "\\n")))))"""
|
|
B = """(module demo_b
|
|
(data IntList (ctor Nil) (ctor Cons (con Int) (con IntList)))
|
|
(fn sum
|
|
(type (fn-type (params (own (con IntList))) (ret (own (con Int)))))
|
|
(params xs)
|
|
(body
|
|
(match xs
|
|
(case (pat-ctor Nil) 0)
|
|
(case (pat-ctor Cons h t) (app + h (app sum t))))))
|
|
(fn main
|
|
(type (fn-type (params) (ret (own (con Unit))) (effects IO)))
|
|
(params)
|
|
(body
|
|
(let xs (term-ctor IntList Cons 1 (term-ctor IntList Cons 2 (term-ctor IntList Nil)))
|
|
(seq (app print (app sum xs)) (do io/print_str "\\n"))))))"""
|
|
S = """(module demo_s
|
|
(fn main
|
|
(type (fn-type (params) (ret (own (con Unit))) (effects IO)))
|
|
(params)
|
|
(body
|
|
(let s (new Series (con Float) 3)
|
|
(let s2 (app Series.push s 1.0)
|
|
(let s3 (app Series.push s2 2.0)
|
|
(seq (app print (app Series.len s3)) (do io/print_str "\\n"))))))))"""
|
|
|
|
INTRO = (
|
|
"This is a Lisp-like language written as S-expressions, BUT in an annotated "
|
|
"dialect: every parenthesis carries its nesting depth. An opening paren is "
|
|
"`(#N ` and its matching closing paren is ` #N)`, where N is the depth "
|
|
"(outermost = 0). Openers and closers at the same depth must pair up. Study "
|
|
"these complete, valid examples and write the requested module in the SAME "
|
|
"annotated style. Return ONLY the module source — no prose, no markdown.\n\n")
|
|
|
|
BASE = INTRO + "Example 1:\n" + annotate(A) + "\n\nExample 2:\n" + annotate(B) + "\n"
|
|
SERIES = (BASE + "\nExample 3 (Series ring-buffer library: "
|
|
"`(#k new Series (#k1 con Float #k1) N #k)` allocates capacity N; "
|
|
"Series.push (own series)(own elem)->own series appends & returns; "
|
|
"Series.at (borrow series)(own Int)->own elem reads index back from "
|
|
"newest (0=newest); Series.len / Series.total_count "
|
|
"(borrow series)->own Int):\n" + annotate(S) + "\n")
|
|
|
|
SMA_OUT = "3.0\n5.33333\n5.66667\n5.33333\n"
|
|
SMA_TASK = ("Write a module `m` computing a simple moving average, window 3, over "
|
|
"the float stream 1.0, 5.0, 3.0, 8.0, 6.0, 2.0 using the Series library: "
|
|
"create a Float Series lookback 3, push each value in turn, and after each "
|
|
"push once at least 3 values have been pushed, print the average of the "
|
|
"most recent 3 (their sum / 3.0) on its own line. Expected:\n" + SMA_OUT)
|
|
|
|
STAGES = [
|
|
("L0_const", BASE, "Write a module `m` with a function `answer` taking no parameters that returns the Int 42.", None),
|
|
("L1_arith", BASE, "Write a module `m` with a function `triple` taking one Int parameter, returning it multiplied by 3.", None),
|
|
("L2_print", BASE, "Write a module `m` whose `main` prints the Int 7 on its own line.", "7\n"),
|
|
("L3_adt", BASE, "Write a module `m` with an ADT `Light` (ctors Red, Green) and a function `code` returning Int 0 for Red and 1 for Green via match.", None),
|
|
("L4_rec", BASE, "Write a module `m` with an IntList ADT (Nil/Cons), a function `length` counting elements by recursion, and a `main` that prints the length of the list [1,2,3] on its own line.", "3\n"),
|
|
("L5_series_len", SERIES, "Write a module `m` using the Series library: create a Float Series lookback 3, push 1.0 then 2.0 then 3.0, then print its Series.len on its own line.", "3\n"),
|
|
("L6_series_at", SERIES, "Write a module `m` using the Series library: create a Float Series lookback 3, push 1.0 then 2.0 then 3.0, then print (Series.at on index 0, the newest) on its own line.", "3.0\n"),
|
|
("L7_sma", SERIES, SMA_TASK, SMA_OUT),
|
|
]
|
|
|
|
def call(system, task):
|
|
body=json.dumps({"model":MODEL,"temperature":0.2,"max_tokens":1000,"stop":["assistant"],
|
|
"messages":[{"role":"system","content":system},{"role":"user","content":task}]}).encode()
|
|
req=urllib.request.Request(ENDPOINT,data=body,headers={
|
|
"Authorization":f"Bearer {os.environ['IONOS_API_TOKEN']}","Content-Type":"application/json"})
|
|
with urllib.request.urlopen(req,timeout=300) as r:
|
|
d=json.loads(r.read())
|
|
return d["choices"][0]["message"]["content"], d.get("usage",{})
|
|
|
|
def extract(resp):
|
|
m=re.search(r"```(?:[a-zA-Z]*)?\n(.*?)```",resp,re.S)
|
|
code=m.group(1) if m else resp
|
|
i=code.find("(#0")
|
|
if i<0: i=code.find("(module")
|
|
if i<0: return code.strip()
|
|
depth=0
|
|
for j in range(i,len(code)):
|
|
if code[j]=="(": depth+=1
|
|
elif code[j]==")":
|
|
depth-=1
|
|
if depth==0: return code[i:j+1]
|
|
return code[i:].strip()
|
|
|
|
def evaluate(annotated_code, expected):
|
|
plain = strip_ann(annotated_code) # <-- verified converter
|
|
m=re.search(r"\(module\s+([A-Za-z_][A-Za-z0-9_]*)", plain)
|
|
name=m.group(1) if m else "m"
|
|
f=Path(f"/tmp/abl/{name}.ail"); f.parent.mkdir(exist_ok=True); f.write_text(plain+"\n")
|
|
chk=subprocess.run([AIL,"check",str(f)],capture_output=True,text=True)
|
|
if chk.returncode!=0: return "check",(chk.stderr or chk.stdout).strip(), plain
|
|
if expected is None: return "green","(check-only)", plain
|
|
run=subprocess.run([AIL,"run",str(f)],capture_output=True,text=True)
|
|
if run.returncode!=0: return "run",(run.stderr or run.stdout).strip(), plain
|
|
if run.stdout!=expected: return "output",f"got {run.stdout!r}, expected {expected!r}", plain
|
|
return "green",run.stdout, plain
|
|
|
|
def main():
|
|
log=["# Qwen format probe 4 — annotated parens `(#N ... #N)`\n",
|
|
f"**Model:** {MODEL}; single-shot per level; annotated-paren dialect; "
|
|
"output stripped via round-trip-verified converter then ail-checked.\n","---\n"]
|
|
summary=[]
|
|
for sid,ctx,task,exp in STAGES:
|
|
try:
|
|
resp,usage=call(ctx,task)
|
|
code=extract(resp)
|
|
stage,fb,plain=evaluate(code,exp)
|
|
except Exception as e:
|
|
usage={}; code=""; plain=""; stage="api-error"; fb=str(e)[:200]
|
|
ok=stage=="green"; mark="✅ green" if ok else f"❌ {stage}"
|
|
summary.append((sid,mark)); print(f"{sid}: {mark}")
|
|
log+=[f"## {sid} — {mark}",
|
|
f"\ntokens: prompt={usage.get('prompt_tokens','?')} completion={usage.get('completion_tokens','?')}\n",
|
|
"annotated output:\n```\n"+code+"\n```\n",
|
|
"stripped to Form-A:\n```\n"+plain+"\n```\n",
|
|
("" if ok else f"**`ail {stage}`:**\n```\n{fb[:700]}\n```\n")]
|
|
log+=["\n---\n## Ladder\n"]+[f"- {s}: {m}" for s,m in summary]
|
|
DOC.write_text("\n".join(log)); print("doc →",DOC)
|
|
|
|
if __name__=="__main__":
|
|
main()
|