136 lines
5.5 KiB
Python
136 lines
5.5 KiB
Python
"""接地验证器（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)))
|