feat(v2): T5 WorkerLoop + 接地验证（代码沙箱/facts对照/结构检查）

2026-08-30 21:08:45 +08:00
parent b3143e2914
commit 4fbbdb5290
5 changed files with 502 additions and 2 deletions
@@ -0,0 +1,135 @@
+"""接地验证器（D4）—— Worker 自验证的"接地"来源。
+
+验证分层优先级（D4）：
+  可执行验证（跑代码/跑测试） > facts 对照 > 结构检查 > 模型自由判断（最后手段）
+
+- 可执行验证：code 域 .py 工件在临时目录子进程沙箱运行（timeout、-I 隔离、捕获输出）。
+- facts 对照：把回答/工件与知识库 facts（statement/keywords）比对覆盖度。
+- 结构检查：工件非空、长度达标、不含纯占位。
+
+说明：沙箱目前做"临时目录 + 超时 + 解释器隔离"，Windows 下真正禁网需系统级工具，
+此处以超时与隔离为主要护栏（文档如实记录）。验证器为纯逻辑，可注入 runner 便于单测。
+"""
+from __future__ import annotations
+
+import subprocess
+import sys
+import tempfile
+from pathlib import Path
+from typing import Any, Callable, Dict, List, Optional, Tuple
+
+# 判定为"含代码"的启发标记
+_CODE_HINTS = ("def ", "class ", "import ", "return ", "if __name__", "print(")
+
+
+def detect_artifact_language(name: str) -> str:
+    """按文件名推断工件语言：python / json / text。"""
+    suffix = Path(name).suffix.lower()
+    if suffix in (".py", ".pyw"):
+        return "python"
+    if suffix in (".json",):
+        return "json"
+    return "text"
+
+
+def extract_code_block(text: str) -> str:
+    """从模型输出提取 python 代码块（剥除 markdown 围栏），无则返回原文本。"""
+    t = text.strip()
+    fence = chr(96) * 3  # 三个反引号
+    marker = fence + "python"
+    start = t.find(marker)
+    if start == -1:
+        return t
+    body_start = start + len(marker)
+    end = t.find(fence, body_start)
+    if end == -1:
+        return t[body_start:].strip()
+    return t[body_start:end].strip()
+
+
+def run_code_sandbox(code: str, timeout_s: float = 10.0,
+                     runner: Optional[Callable[[List[str], Dict[str, str], float], Tuple[int, str, str]]] = None
+                     ) -> Tuple[int, str, str]:
+    """在临时目录子进程运行 python 代码。返回 (returncode, stdout, stderr)。
+
+    隔离措施：临时工作目录、-I 隔离模式、timeout 超时强杀、捕获输出。
+    runner 可注入（测试用假执行器，避免真跑任意代码）。
+    """
+    if runner is not None:
+        return runner([sys.executable, "-I"], {}, timeout_s)
+    with tempfile.TemporaryDirectory(prefix="v2_sandbox_") as tmp:
+        script = Path(tmp) / "main.py"
+        script.write_text(code, encoding="utf-8")
+        try:
+            proc = subprocess.run(
+                [sys.executable, "-I", str(script)],
+                capture_output=True, text=True, timeout=timeout_s,
+                cwd=tmp,
+                creationflags=subprocess.CREATE_NO_WINDOW,
+            )
+            return proc.returncode, proc.stdout or "", proc.stderr or ""
+        except subprocess.TimeoutExpired:
+            return -1, "", "timeout exceeded"
+
+
+class Verifier:
+    """按 D4 分层执行接地验证。"""
+
+    def __init__(self, code_timeout_s: float = 10.0,
+                 sandbox_runner: Optional[Callable[..., Tuple[int, str, str]]] = None):
+        self.code_timeout_s = code_timeout_s
+        self._sandbox_runner = sandbox_runner
+
+    def verify(self, domain: str, artifact_name: str, artifact_text: str,
+               query: str, kb: Any = None) -> Tuple[bool, List[str]]:
+        """返回 (passed, details)。kb 为 KnowledgeBase（可 None，跳过 facts 层）。"""
+        lang = detect_artifact_language(artifact_name)
+        details: List[str] = []
+
+        # 1) 可执行验证（D4 最高优先级）：code + python 工件且含代码
+        if lang == "python" and any(h in artifact_text for h in _CODE_HINTS):
+            rc, out, err = run_code_sandbox(artifact_text, self.code_timeout_s,
+                                            runner=self._sandbox_runner)
+            if rc == 0:
+                details.append("代码沙箱运行通过 (rc=0)")
+                return True, details
+            # 可执行验证失败 -> 直接拒绝（不落回结构检查，避免"假通过"）
+            details.append(f"代码沙箱运行失败 rc={rc}: {(err or out)[:120]}")
+            return False, details
+        elif lang == "python":
+            details.append("工件不含可执行代码（跳过沙箱，进入 facts/结构检查）")
+
+        # 2) facts 对照
+        if kb is not None:
+            fact_hits = self._check_facts(domain, artifact_text, kb)
+            if fact_hits:
+                details.append(f"facts 对照命中 {fact_hits}")
+                return True, details
+
+        # 3) 结构检查
+        text = artifact_text.strip()
+        if len(text) < 20:
+            details.append(f"工件过短（{len(text)} 字符）")
+            return False, details
+        if text.lower() in ("pass", "none", "todo", "待实现", "略"):
+            details.append("工件为占位内容")
+            return False, details
+        details.append("结构检查通过（非空、长度达标）")
+        return True, details
+
+    def _check_facts(self, domain: str, text: str, kb: Any) -> int:
+        """统计工件文本命中知识库事实的数量。"""
+        facts = kb.facts(domain) or []
+        if not facts:
+            return 0
+        hits = 0
+        for f in facts:
+            kw = f.get("keywords") or []
+            if any(str(k) in text for k in kw):
+                hits += 1
+        return hits
+
+
+def build_verifier(cfg: Dict[str, Any]) -> Verifier:
+    """cfg 为 config.worker 段。"""
+    return Verifier(code_timeout_s=float(cfg.get("code_timeout_s", 10)))