| |
| |
| """ |
| s8_sprobe_long.py — R10 长上下文 serving 级探针 (量化判据的真相维度) |
| 背景: Q4 短上下文探针 A/B/C 全 PASS(纪律原话在 think 中出现) → R9 结论动摇; |
| 差异变量 = GAIA 实败场景是 40-70K 长上下文。本探针在 ~30-40K token 任务史末尾 |
| 放两振失败回放, 检验纪律是否被上下文长度稀释。 |
| 用法: python3 s8_sprobe_long.py <tag> [steps=500] # 500 步 ≈ 35K token |
| """ |
| import json, re, sys, time, urllib.request |
|
|
| PORT = 8101 |
| TAG = sys.argv[1] if len(sys.argv) > 1 else "q4long" |
| STEPS = int(sys.argv[2]) if len(sys.argv) > 2 else 500 |
| RES = "/tmp/k8b/probe_long_%s.txt" % TAG |
| res = open(RES, "w", buffering=1) |
| def P(m): res.write("[%s] %s\n" % (time.strftime("%H:%M:%S"), m)); print(m, flush=True) |
|
|
| Q = "book X page count" |
| REJ = re.compile(r"(无意义重复|不能再?.{0,8}重复|换一个方法|换.{0,6}思路|换路|换路径|不再重试|改变方法|放弃.{0,6}路)") |
|
|
| def filler(n): |
| tools = ["execute_code", "web_reader", "file_read", "web_search"] |
| out = [] |
| for i in range(n): |
| t = tools[i % 4] |
| out.append('[行动] %s({"arg": "step %d dataset row %d", "mode": "full"})' % (t, i, i * 7)) |
| out.append('[观察] {"ok": true, "rows": %d, "note": "chunk %d processed, partial sum %d.%02d"}' % (100 + i, i, i * 13, i % 100)) |
| return "\n".join(out) |
|
|
| def ask(user, maxn=640, tmo=1500): |
| body = json.dumps({"messages": [{"role": "user", "content": user}], |
| "temperature": 0, "max_tokens": maxn}).encode() |
| req = urllib.request.Request("http://127.0.0.1:%d/v1/chat/completions" % PORT, |
| data=body, headers={"Content-Type": "application/json"}) |
| t0 = time.time() |
| d = json.loads(urllib.request.urlopen(req, timeout=tmo).read()) |
| ch = d["choices"][0]["message"] |
| txt = ((ch.get("reasoning_content") or "") + "\n" + (ch.get("content") or "")).strip() |
| return txt, time.time() - t0 |
|
|
| try: |
| user = ("任务(长程真题·窗口回放训练): 对比多个数据源的统计口径并查询某本小说的页数信息。\n\n" |
| "=== 任务史(已完成步骤, 仅供上下文连续性, 勿重复执行) ===\n" + filler(STEPS) + |
| "\n=== 最近窗口末尾 ===\n" |
| "[行动] web_search({\"query\": \"%s\"})\n[观察] {\"error\": \"search failed: HTTP 404\"}\n" |
| "[行动] (web_search,参数完全相同,第2次)\n[观察] (与上次逐字节相同)\n\n" |
| "(续) 请继续执行后续步骤。" % Q) |
| ntok = len(user) // 3 |
| P("SPROBE_LONG[%s] steps=%d approx_tokens=%d" % (TAG, STEPS, ntok)) |
| a, dt = ask(user) |
| P("PROBE_AL_RAW>>> " + a[:600].replace("\n", " | ")) |
| rej = bool(REJ.search(a)); retry = Q in a |
| va = "PASS" if (rej and not retry) else ("WEAK" if not retry else "FAIL") |
| P("PROBE_AL verdict=%s reject_phrase=%s retried_same=%s (%.0fs)" % (va, rej, retry, dt)) |
| P("FINAL_LONG: A=%s" % va) |
| except Exception as e: |
| import traceback |
| P("SPROBE_LONG_CRASH: " + repr(e)) |
| P(traceback.format_exc()[-600:]) |
|
|
