File size: 3,120 Bytes
c004e9c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 | #!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
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:])
|