#!/usr/bin/env python3 """Launch inkreach-teable skill evals (iteration-4 A/B rerun). Usage: python3 run_evals4.py [eval-ids] config-spec: with_skill=/root/eval-rerun/skill-new:/root/eval-rerun/bin-new/bin,old_skill=/root/eval-rerun/skill-old:/root/eval-rerun/bin-old/bin Differences vs the original run_evals.py (eval-integrity fixes, see LESSONS.md): - evals.json + sandbox-env.json live in THIS harness dir (outside the agent's cwd tree) - per-config sandbox BASE via INKREACH_TEABLE_BASE_ID (no cross-config contamination) - per-config PATH prepended with that config's CLI version dir (0.1.4 vs 0.1.5) - pi runs with an isolated HOME (no user-level skills discovered) - grading.json written only afterwards by grade_evals4.py (no answers near agents) """ import json import os import re import subprocess import sys import time from concurrent.futures import ThreadPoolExecutor from pathlib import Path HARNESS = Path(__file__).parent # evals/ itself ROOT = Path(os.environ.get("EVAL_ROOT", "/root/eval-rerun")) MODEL = os.environ.get("EVAL_MODEL", "opencode-go/deepseek-v4.1-flash") RUN_TIMEOUT = 600 PI_HOME = os.environ.get("EVAL_PI_HOME", "/root/eval-rerun/pi-home") env_all = json.loads(Path(os.environ.get('EVAL_ENV', str(HARNESS / 'sandbox-env.json'))).read_text()) PROMPT_TMPL = """你在操作一个已部署的 inkreach-teable 实例(Teable fork,自研自动化引擎)。 环境(已配置好,直接使用): - CLI 命令: `inkreach-teable`(已在 PATH)。endpoint / token / base 已通过环境变量 INKREACH_TEABLE_ENDPOINT / INKREACH_TEABLE_TOKEN / INKREACH_TEABLE_BASE_ID 注入,不要执行 auth login,不要改配置文件。 - 只操作环境变量指定的这个基地(评估沙箱);严禁触碰其他基地。 - 工作目录就是当前目录,把中间产物写在这里。 {SKILL_LINE} 任务:{TASK} 要求: 1. 逐步执行任务,用 CLI 完成实际操作(不是只给命令)。 2. 完成后把最终答复写入 final_answer.md:做了什么、关键命令、结果、需要用户知道的事项。""" def extract_final_text(jsonl_path: Path) -> str: texts = [] for line in jsonl_path.read_text(errors="replace").splitlines(): line = line.strip() if not line: continue try: ev = json.loads(line) except json.JSONDecodeError: continue if ev.get("type") == "turn_end" and ev.get("message", {}).get("role") == "assistant": for c in ev["message"].get("content", []): if c.get("type") == "text" and c.get("text", "").strip(): texts.append(c["text"]) return texts[-1] if texts else "" def extract_usage(jsonl_path: Path) -> dict: usage = {} for line in jsonl_path.read_text(errors="replace").splitlines(): m = re.search(r'"usage":\{[^{}]*"totalTokens":(\d+)', line) if m: usage["totalTokens"] = int(m.group(1)) return usage def run_one(task, config, skill_path, bin_dir, cfg_env, run_dir: Path): run_dir.mkdir(parents=True, exist_ok=True) if config == "auto_skill": # trigger test: no skill hint at all; the skill must be DISCOVERED from # the user-level skills dir of the isolated (skilled) HOME. skill_line = "" elif skill_path: skill_line = ( f"- 先完整阅读技能说明 {skill_path}/SKILL.md 并严格按其路由规则与关键约束操作;\n" f" 需要细节时再按其中链接查阅 guides/ 与 api-reference/ 文档。" ) else: skill_line = "- 环境里没有额外文档,自行判断。" prompt = PROMPT_TMPL.format(SKILL_LINE=skill_line, TASK=task) (run_dir / "prompt.txt").write_text(prompt) full_env = { **os.environ, "HOME": (os.environ.get("EVAL_PI_HOME_SKILLED", "/root/eval-rerun/pi-home-skilled") if config == "auto_skill" else PI_HOME), "PATH": f"{bin_dir}:{os.environ['PATH']}", **{k: env_all[k] for k in ("INKREACH_TEABLE_ENDPOINT", "INKREACH_TEABLE_TOKEN")}, **cfg_env, } cmd = ["timeout", str(RUN_TIMEOUT), "pi", "-p", "--mode", "json", "--model", MODEL, prompt] t0 = time.time() with open(run_dir / "pi.jsonl", "w") as out: proc = subprocess.run(cmd, stdout=out, stderr=subprocess.DEVNULL, cwd=run_dir, env=full_env) dur = time.time() - t0 timing = {"duration_ms": int(dur * 1000), "total_duration_seconds": round(dur, 1), "total_tokens": extract_usage(run_dir / "pi.jsonl").get("totalTokens"), "exit_code": proc.returncode, "cli": Path(bin_dir).name, "config": config} (run_dir / "timing.json").write_text(json.dumps(timing, indent=2)) (run_dir / "assistant_final.md").write_text(extract_final_text(run_dir / "pi.jsonl") or "(no final text captured)") return config, run_dir.name, proc.returncode, round(dur, 1) def main(): iteration = sys.argv[1] tag = sys.argv[2] iteration_dir = ROOT / iteration configs = {} for part in sys.argv[3].split(","): name, _, rest = part.partition("=") skill_path, _, bin_dir = rest.partition(":") configs[name] = (skill_path, bin_dir) evals = json.loads(Path(os.environ.get('EVAL_EVALS', str(HARNESS / 'teable-evals.json'))).read_text())["evals"] if len(sys.argv) > 4 and sys.argv[4]: only = {int(x) for x in sys.argv[4].split(",")} evals = [e for e in evals if e["id"] in only] jobs = [] for ev in evals: name = f"eval-{ev['id']}-{ev['name']}" for config, (skill_path, bin_dir) in configs.items(): cfg = env_all.get(config) or env_all['with_skill'] cfg_short = "W" if config == "with_skill" else "O" task = (ev["prompt"] .replace("{TAG}", f"{tag}{cfg_short}") .replace("{WFID}", cfg["WF_ID"]) .replace("{TBL5}", cfg["TBL5_NAME"]) .replace("{ORD}", cfg["ORD"])) meta = {"eval_id": ev["id"], "eval_name": ev["name"], "config": config, "prompt": task, "assertions": []} rd = iteration_dir / name / config / "run-1" rd.parent.mkdir(parents=True, exist_ok=True) (rd.parent / "eval_metadata.json").write_text(json.dumps(meta, indent=2, ensure_ascii=False)) cfg_env = { "INKREACH_TEABLE_BASE_ID": cfg["INKREACH_TEABLE_BASE_ID"], "INKREACH_TEABLE_EVAL_TBL": cfg["TBL5_ID"], "INKREACH_TEABLE_EVAL_ORD": cfg["ORD"], } if ev["id"] == 12: cfg_env["INKREACH_TEABLE_TOKEN"] = env_all["LOWPERM_TOKEN"] jobs.append((task, config, skill_path, bin_dir, cfg_env, rd)) print(f"{len(jobs)} runs -> {iteration_dir}", flush=True) with ThreadPoolExecutor(max_workers=6) as ex: futs = [ex.submit(run_one, *j) for j in jobs] for f in futs: print(f.result(), flush=True) if __name__ == "__main__": main()