Release vulcan-decision-shell-v1

f42e39c verified 16 days ago

4.53 kB

"""Load the native Vulcan first-pass shell decision model without generating text or executing tools.
加载原生Vulcan Shell初审判定模型，不生成文本也不执行工具。
"""

import argparse
import hashlib
import json
from pathlib import Path
import sys

# Runtime modules are bundled with this repository; network code loading is not required.
# 运行模块随本仓库提供，无需联网加载代码。
RUNTIME = Path(__file__).resolve().parent / "runtime"
sys.path.insert(0, str(RUNTIME))
from risk_judge import RiskJudge, render_call


class DecisionShell:
    """Own an explicit checkpoint and its fixed-option, three-decision inference runtime.
    管理明确检查点及固定选项、三种判决的推理运行时。
    This class accepts complete tool-call text, never a task-intent guess or executable callback.
    本类接收完整工具调用文本，不接收推测的任务意图或可执行回调。
    """

    def __init__(self, root: str | Path, *, device: str = "cuda", precision: str = "original",
                 allow_experimental_precision: bool = False) -> None:
        """Verify selected weight bytes and initialize one persistent model instance.
        核验选定权重字节，初始化一个持久模型实例。
        root is the unpacked repository; original means FP32 storage with FP16 CUDA autocast.
        root为解压后的仓库，original表示FP32存储并在CUDA使用FP16自动混合精度。
        Experimental FP16/BF16 storage requires explicit opt-in; return nothing and never fall back.
        FP16及BF16实验性存储需要显式选择，无返回值，绝不自动降级。
        """
        self.root = Path(root).resolve()
        if precision not in {"original", "fp16", "bf16"}:
            raise ValueError("Unsupported weight format.")
        if precision != "original" and not allow_experimental_precision:
            raise ValueError("Converted storage is experimental; read evaluation/PRECISION.md before explicit opt-in.")
        # The recorded format identity is authoritative; filename similarity is insufficient.
        # 记录的格式身份是权威，仅文件名相似不足以建立身份。
        name = "fp32" if precision == "original" else precision
        declaration = json.loads((self.root / "weights/formats.json").read_text(encoding="utf-8"))[name]
        path = self.root / "weights" / name / "model.safetensors"
        with path.open("rb") as stream:
            checksum = hashlib.file_digest(stream, "sha256").hexdigest()
        if path.stat().st_size != declaration["bytes"] or checksum != declaration["sha256"]:
            raise ValueError("Model weight identity verification failed.")
        self.judge = RiskJudge(self.root / "runtime", backend="torch", device=device, precision=precision)

    def predict(self, samples: list[dict]) -> list[dict]:
        """Return ordered decisions for complete settings/rendered records, without executing them.
        对完整settings及rendered记录返回有序判决，不执行其内容。
        samples must use the documented fixed options; runtime errors must never become approvals.
        samples必须使用文档中的固定选项，运行错误绝不能转为放行。
        """
        return self.judge.predict(samples)


def main() -> None:
    """Read an explicit JSON input file and print decisions using the chosen local checkpoint.
    读取明确的JSON输入文件，使用选定本地检查点打印判决。
    CLI accepts root, input, device and precision; reviewed commands are always inert data.
    CLI接收root、input、device和precision，受审命令始终是惰性数据。
    """
    parser = argparse.ArgumentParser()
    parser.add_argument("--root", type=Path, default=Path(__file__).resolve().parent)
    parser.add_argument("--input", type=Path, required=True)
    parser.add_argument("--device", choices=["cuda", "cpu"], default="cuda")
    parser.add_argument("--precision", choices=["original", "fp16", "bf16"], default="original")
    parser.add_argument("--allow-experimental-precision", action="store_true")
    args = parser.parse_args()
    model = DecisionShell(args.root, device=args.device, precision=args.precision,
                          allow_experimental_precision=args.allow_experimental_precision)
    samples = json.loads(args.input.read_text(encoding="utf-8"))
    print(json.dumps(model.predict(samples), ensure_ascii=False, allow_nan=False))


if __name__ == "__main__":
    main()

	"""Load the native Vulcan first-pass shell decision model without generating text or executing tools.
	加载原生Vulcan Shell初审判定模型，不生成文本也不执行工具。
	"""

	import argparse
	import hashlib
	import json
	from pathlib import Path
	import sys

	# Runtime modules are bundled with this repository; network code loading is not required.
	# 运行模块随本仓库提供，无需联网加载代码。
	RUNTIME = Path(__file__).resolve().parent / "runtime"
	sys.path.insert(0, str(RUNTIME))
	from risk_judge import RiskJudge, render_call


	class DecisionShell:
	"""Own an explicit checkpoint and its fixed-option, three-decision inference runtime.
	管理明确检查点及固定选项、三种判决的推理运行时。
	This class accepts complete tool-call text, never a task-intent guess or executable callback.
	本类接收完整工具调用文本，不接收推测的任务意图或可执行回调。
	"""

	def __init__(self, root: str \| Path, *, device: str = "cuda", precision: str = "original",
	allow_experimental_precision: bool = False) -> None:
	"""Verify selected weight bytes and initialize one persistent model instance.
	核验选定权重字节，初始化一个持久模型实例。
	root is the unpacked repository; original means FP32 storage with FP16 CUDA autocast.
	root为解压后的仓库，original表示FP32存储并在CUDA使用FP16自动混合精度。
	Experimental FP16/BF16 storage requires explicit opt-in; return nothing and never fall back.
	FP16及BF16实验性存储需要显式选择，无返回值，绝不自动降级。
	"""
	self.root = Path(root).resolve()
	if precision not in {"original", "fp16", "bf16"}:
	raise ValueError("Unsupported weight format.")
	if precision != "original" and not allow_experimental_precision:
	raise ValueError("Converted storage is experimental; read evaluation/PRECISION.md before explicit opt-in.")
	# The recorded format identity is authoritative; filename similarity is insufficient.
	# 记录的格式身份是权威，仅文件名相似不足以建立身份。
	name = "fp32" if precision == "original" else precision
	declaration = json.loads((self.root / "weights/formats.json").read_text(encoding="utf-8"))[name]
	path = self.root / "weights" / name / "model.safetensors"
	with path.open("rb") as stream:
	checksum = hashlib.file_digest(stream, "sha256").hexdigest()
	if path.stat().st_size != declaration["bytes"] or checksum != declaration["sha256"]:
	raise ValueError("Model weight identity verification failed.")
	self.judge = RiskJudge(self.root / "runtime", backend="torch", device=device, precision=precision)

	def predict(self, samples: list[dict]) -> list[dict]:
	"""Return ordered decisions for complete settings/rendered records, without executing them.
	对完整settings及rendered记录返回有序判决，不执行其内容。
	samples must use the documented fixed options; runtime errors must never become approvals.
	samples必须使用文档中的固定选项，运行错误绝不能转为放行。
	"""
	return self.judge.predict(samples)


	def main() -> None:
	"""Read an explicit JSON input file and print decisions using the chosen local checkpoint.
	读取明确的JSON输入文件，使用选定本地检查点打印判决。
	CLI accepts root, input, device and precision; reviewed commands are always inert data.
	CLI接收root、input、device和precision，受审命令始终是惰性数据。
	"""
	parser = argparse.ArgumentParser()
	parser.add_argument("--root", type=Path, default=Path(__file__).resolve().parent)
	parser.add_argument("--input", type=Path, required=True)
	parser.add_argument("--device", choices=["cuda", "cpu"], default="cuda")
	parser.add_argument("--precision", choices=["original", "fp16", "bf16"], default="original")
	parser.add_argument("--allow-experimental-precision", action="store_true")
	args = parser.parse_args()
	model = DecisionShell(args.root, device=args.device, precision=args.precision,
	allow_experimental_precision=args.allow_experimental_precision)
	samples = json.loads(args.input.read_text(encoding="utf-8"))
	print(json.dumps(model.predict(samples), ensure_ascii=False, allow_nan=False))


	if __name__ == "__main__":
	main()