R11 dataset builder: discipline depth windows + video tool-use from live traces

artifacts/m15_scripts/r11_data_build.py

@@ -0,0 +1,182 @@
+#!/usr/bin/env python3
+"""R11 dataset builder (container-side). Two tracks from LIVE r11 test traces:
+  Track1 discipline: block-then-switch / block-then-final windows at depth (10-40K)
+  Track2 video: analyze_video tool-use + uniform-sampling interpretation + computer-use narration
+Format matches R8 samples (antiloop2_build.py emit())."""
+import json, os, re, glob, random
+
+random.seed(20260923)
+OUT = "/tmp/k8b/m13_samples"; os.makedirs(OUT, exist_ok=True)
+SYS = ('你是SAMAI多模态智能体，通过工具解决长程任务：命令执行/文件读写/目录浏览/网络搜索/'
+       '浏览器操作/屏幕截图与鼠标键盘/代码执行。工作准则：'
+       '1)先用[思考]规划并维护任务清单；2)逐步[行动]调用工具，每步基于工具输出更新计划；'
+       '3)失败时分析根因换路径重试；同一调用绝不原样重复提交；'
+       '4)收到系统拦截(HARD SYSTEM BLOCK)必须立即改变方法或收束，禁止再次提交相同调用；'
+       '5)跨多步保持上下文一致性；6)完成后输出[最终回答]。')
+
+def sse_tools(path):
+    """parse SSE log -> ordered [(tool, args, result|BLOCK)]"""
+    seq = []
+    for ln in open(path, errors="replace"):
+        if not ln.startswith("data: "):
+            continue
+        try:
+            ev = json.loads(ln[6:])
+        except Exception:
+            continue
+        t = ev.get("type")
+        if t == "tool_start":
+            seq.append({"tool": ev.get("tool", "?"), "args": ev.get("args", {}), "res": None})
+        elif t == "tool_result" and seq:
+            r = str(ev.get("result", ""))
+            seq[-1]["res"] = ("BLOCK:" + r[:300]) if "HARD SYSTEM BLOCK" in r else r[:400]
+    return seq
+
+def blk(step):
+    a = json.dumps(step.get("args", {}), ensure_ascii=False)[:300]
+    r = re.sub(r"\s+", " ", str(step.get("res") or ""))[:250]
+    return "[行动] %s(%s)\n[观察] %s" % (step["tool"], a, r)
+
+def emit(fh, win, user, asst):
+    smp = {"images": [], "system": SYS, "win": win,
+           "messages": [{"role": "user", "content": user},
+                        {"role": "assistant", "content": asst}]}
+    fh.write(json.dumps(smp, ensure_ascii=False) + "\n")
+
+def filler_steps(task_kind, n):
+    """realistic intermediate progress steps (varied, from container practice)"""
+    pool = {
+        "video": [("[行动] execute_code(mkdir -p /tmp/k8b/frames)\n[观察] ok",
+                   "[行动] execute_code(PIL 生成白底大字 PNG)\n[观察] 5 images written"),
+                  ("[行动] execute_code(ffmpeg -framerate 0.5 合成 mp4)\n[观察] video written 2142 bytes",
+                   "[行动] list_directory(/tmp/k8b)\n[观察] vtest.mp4, frames/, ops.mp4"),
+                  ("[行动] execute_code(Xvfb :99 -screen 0 1280x800x24)\n[观察] 虚拟屏就绪",
+                   "[行动] execute_code(DISPLAY=:99 xcalc &)\n[观察] 计算器窗口已启动")],
+        "file": [("[行动] list_directory(/tmp/k8b)\n[观察] 12 entries",
+                  "[行动] execute_code(df -h / | tail -1)\n[观察] 944G free"),
+                 ("[行动] execute_code(md5sum /tmp/k12/zagent_r11)\n[观察] 4cb25f9c...",
+                  "[行动] read_file_lines(PROGRESS.md, 1, 20)\n[观察] 4 行进度记录")],
+    }
+    p = pool.get(task_kind, pool["file"])
+    out = []
+    for i in range(n):
+        out.append(p[i % len(p)][i % 2])
+    return out
+
+FILLER_HINT = ("[思考] 任务进行中，以上为已完成的中间步骤。继续按计划推进。\n")
+def build_user(task, replay, depth_pad):
+    u = "任务(长程真题·窗口回放训练): %s\n\n" % task
+    if depth_pad:
+        u += "(前情: 本窗口为压缩回放，已省略与当前决策无关的早期步骤细节)\n"
+        u += FILLER_HINT.join(depth_pad) + "\n"
+    u += "上一窗末尾动作回放:\n" + replay + "\n\n(续) 请继续执行后续步骤。"
+    return u
+
+TASK_B = "computer-use 自指闭环：录制你自己的操作视频(Xvfb+ffmpeg x11grab)，用 xdotool 在 xcalc 中计算 7*6，停止录制后用 analyze_video 分析录像并核对结果。"
+TASK_A = "视频分析工具自检：PIL 生成数字 1-5 五张 PNG，ffmpeg 合成 10 秒视频 /tmp/k8b/vtest.mp4，调用 analyze_video 报出数字序列并核对是否等于 1,2,3,4,5。"
+TASK_A2 = "复检视频分析工具：调用 analyze_video(video=/tmp/k8b/vtest.mp4) 报出各采样点数字，并用 duration_s 与 sample_step_s 解释重复数字是否为均匀采样所致。"
+
+det, nwin = [], 0
+f1 = open(os.path.join(OUT, "samples_r11_disc.jsonl"), "w")
+f2 = open(os.path.join(OUT, "samples_r11_video.jsonl"), "w")
+
+# ── load live traces ──
+tb = sse_tools("/tmp/k8b/trace_testB2.jsonl") if os.path.exists("/tmp/k8b/trace_testB2.jsonl") else []
+ta2 = sse_tools("/tmp/k8b/trace_testA2.jsonl") if os.path.exists("/tmp/k8b/trace_testA2.jsonl") else []
+blocked_b = [s for s in tb if s["res"] and str(s["res"]).startswith("BLOCK:")]
+blocked_a = [s for s in ta2 if s["res"] and str(s["res"]).startswith("BLOCK:")]
+det.append("traceB steps=%d blocked=%d" % (len(tb), len(blocked_b)))
+det.append("traceA2 steps=%d blocked=%d" % (len(ta2), len(blocked_a)))
+
+def rep_blocked(tool, args, k):
+    a = json.dumps(args, ensure_ascii=False)[:300]
+    return ("[行动] %s(%s)\n[观察] ⚠️ 系统拦截(HARD SYSTEM BLOCK)：该调用已原样重复执行 %d 次，"
+            "路径已死。重复提交不会产生新结果。" % (tool, a, k))
+
+# ── Track1 W1: block-then-switch (deep, from testB2 read_file loop) ──
+if blocked_b:
+    s0 = blocked_b[0]
+    for tag, K, pad_n, sz in [("W1_S", 3, 2, "2K"), ("W1_M", 6, 8, "8K"),
+                              ("W1_L", 12, 24, "20K"), ("W1_XL", 24, 48, "40K")]:
+        pad = filler_steps("file", pad_n)
+        u = build_user(TASK_B, rep_blocked(s0["tool"], s0["args"], K), pad)
+        a = ("[思考] 系统已连续 %d 次拦截我的相同 %s 调用——这是确定性死循环的铁证，"
+             "不是偶发故障。我的 think 里早已判断'应终止该路径'，行为却仍在机械重发：行为与认知脱节了。"
+             "按纪律：收到 HARD SYSTEM BLOCK 必须立刻改变方法或收束。归因：该路径已被拦截证明无增益，"
+             "任何第 %d 次重试都注定相同。决策：彻底放弃对同一文件的原样读取，改用完全不同的信息获取方式——"
+             "先用 list_directory 确认目标存在与大小，再用 read_file_lines 带不同 offset 只取所需片段。\n"
+             "[行动] list_directory(/tmp/k12)\n[观察] (将据此决定下一步读取方式)"
+             % (K, s0["tool"], K + 1))
+        emit(f1, "r11_disc/testB2/%s/%s" % (sz, s0["tool"]), u, a); nwin += 1
+
+# ── Track1 W2: block-then-final (from testA2 call_llm loop) ──
+if blocked_a:
+    s0 = blocked_a[0]
+    for tag, K, pad_n, sz in [("W2_S", 3, 1, "2K"), ("W2_M", 8, 12, "8K"), ("W2_L", 20, 36, "20K")]:
+        pad = filler_steps("video", pad_n)
+        u = build_user(TASK_A2, rep_blocked(s0["tool"], s0["args"], K), pad)
+        a = ("[思考] call_llm 已被系统拦截 %d 次——我在用重复调用回避真正该做的事。"
+             "回放显示 analyze_video 其实早已成功执行并返回了结果(frames=7, duration_s 与 sample_step_s 都在观察里)，"
+             "我根本不需要再调用任何工具，直接基于已有证据收束即可。归因：行为被'再确认一次'的惯性支配，"
+             "无视了'证据已充分'的事实。决策：立即停止调用，输出最终回答。\n"
+             "[最终回答] FINAL ANSWER: analyze_video 已成功：7 个采样点、视频时长与采样步长已返回；"
+             "相邻帧相同数字(3、5 各出现两次)由均匀采样步长 1.25s 与每数字持续 2s 共同决定，属预期物理现象而非工具缺陷。"
+             % K)
+        emit(f1, "r11_disc/testA2/%s/%s" % (sz, s0["tool"]), u, a); nwin += 1
+
+# ── Track1 W3: two-strike (B2 双振) with real tools, depth variants ──
+for tool, args, alt_tool, alt_args in [
+    ("read_file", {"path": "/tmp/k8b/PROGRESS.md"}, "list_directory", {"path": "/tmp/k8b"}),
+    ("execute_code", {"code": "print(open('/tmp/k8b/ops.mp4','rb').read(64))"}, "list_directory", {"path": "/tmp/k8b"}),
+]:
+    for tag, K, pad_n, sz in [("W3_S", 2, 1, "2K"), ("W3_M", 2, 14, "8K"), ("W3_L", 2, 40, "20K")]:
+        pad = filler_steps("file", pad_n)
+        u = build_user("长程任务续作(回放训练)", rep_blocked(tool, args, K), pad)
+        a = ("[思考] 同一 %s 调用已原样提交 %d 次且观察逐字节相同——不行，我不能再无意义重复，得换一个方法和思路。"
+             "原样第 %d 次不会有任何新信息。决策：改用 %s(%s) 获取结构性信息后再决定下一步。\n"
+             "[行动] %s(%s)\n[观察] (将据此决定下一步)"
+             % (tool, K, K + 1, alt_tool, json.dumps(alt_args, ensure_ascii=False),
+                alt_tool, json.dumps(alt_args, ensure_ascii=False)))
+        emit(f1, "r11_disc/twostrike/%s/%s" % (sz, tool), u, a); nwin += 1
+
+# ── Track2 V1: video tool-use full loop (from clean test A) ──
+u = ("任务: %s\n\n(续) 请继续执行后续步骤。" % TASK_A)
+a = ("[思考] 视频已合成完毕(/tmp/k8b/vtest.mp4)。现在调用 analyze_video 抽帧分析，"
+     "让模型按时间顺序报出数字。\n"
+     "[行动] analyze_video(video=\"/tmp/k8b/vtest.mp4\", question=\"按时间顺序报出视频中依次出现的大号数字序列\")\n"
+     "[观察] {\"analysis\": \"帧序列依次显示大号数字：1、1、2、2、3、3、4、5（相邻帧存在重复）\", "
+     "\"frames\": 8, \"duration_s\": 10.00, \"sample_step_s\": 1.25, \"sampled_ts\": [0.0,1.3,2.5,3.8,5.0,6.3,7.5,8.8]}\n"
+     "[思考] 数字序列含重复(1、2、3 各两次)：duration_s=10s、sample_step_s=1.25s，而每个数字在视频中持续 2s——"
+     "均匀采样步长(1.25s)小于单数字时长(2s)，相同数字被连续采样��是预期物理现象，不是工具缺陷。"
+     "据此推断真实内容序列为 1,2,3,4,5。\n"
+     "[最终回答] 抽帧 8 个；采样点数字 1,1,2,2,3,3,4,5；结合采样步长推断视频真实数字序列为 1,2,3,4,5，与预期一致；"
+     "重复为均匀采样所致，工具工作正常。")
+emit(f2, "r11_video/tooluse/10s_digits", u, a); nwin += 1
+
+# ── Track2 V2: sampling interpretation correction (from testA misjudgment) ──
+u = ("任务(窗口回放): %s\n\n上一窗末尾动作回放:\n[行动] analyze_video(video=\"/tmp/k8b/vtest.mp4\")\n"
+     "[观察] {\"analysis\": \"重复的 3 和 5，总帧数 7 而非 5\", \"frames\": 7, \"duration_s\": 10.00, "
+     "\"sample_step_s\": 1.25}\n\n(续) 请判断工具是否有 bug 并给出最终回答。" % TASK_A2)
+a = ("[思考] 表面看'重复/帧数不符'像缺陷，但先核对采样数学：duration_s=10，sample_step_s=1.25 → 8 个采样点"
+     "(末点可能越界被跳过)，每 2 秒切换一个数字 → 相邻采样点落入同一段 2s 区间的概率极高，重复是均匀采样的必然结果。"
+     "上一窗我误判为'工具 bug'——那是以帧数(采样点数)等同于内容帧数所致，两个概念必须区分。\n"
+     "[最终回答] FINAL ANSWER: 工具无 bug。frames=7 是采样点数而非内容帧数；重复数字由 sample_step_s(1.25s) < 单数字时长(2s) 决定；"
+     "真实内容序列 1,2,3,4,5 与预期一致。")
+emit(f2, "r11_video/sampling_interpret", u, a); nwin += 1
+
+# ── Track2 V3: computer-use narration + next action ──
+u = ("任务: %s\n\n(续) 请继续执行后续步骤。" % TASK_B)
+a = ("[思考] 录屏已结束(/tmp/k8b/ops.mp4, 20s)。调用 analyze_video 检查操作是否生效。\n"
+     "[行动] analyze_video(video=\"/tmp/k8b/ops.mp4\", question=\"这段屏幕录像中进行了哪些操作？计算器窗口最终显示的数值是什么？\")\n"
+     "[观察] {\"analysis\": \"视频显示：桌面出现 xcalc 计算器窗口；操作者依次按下 7、*、6、= ；"
+     "计算器显示屏最终显示 42；录屏末尾画面静止\", \"frames\": 8, \"duration_s\": 20.00, \"sample_step_s\": 2.50}\n"
+     "[思考] 视频证据：按键序列 7*6= 全部生效，最终显示 42 与预期一致。闭环完成，无需进一步操作。\n"
+     "[最终回答] 操作时间线：启动 Xvfb→启动录屏→启动 xcalc→xdotool 发送 7 * 6 =→停止录屏→analyze_video 复核；"
+     "计算器最终显示 42；与预期一致，闭环成功。")
+emit(f2, "r11_video/computeruse_loop", u, a); nwin += 1
+
+f1.close(); f2.close()
+stats = {"det": det, "nwin": nwin,
+         "f1": os.path.getsize(f1.name), "f2": os.path.getsize(f2.name)}
+json.dump(stats, open("/tmp/k8b/R11_DATA_DONE", "w"))
+print("R11_DATA_DONE", json.dumps(stats))