run.py

이 글의 원장 예제에 포함된 코드입니다. 압축을 푼 폴더의 experiments/inspect/run.py와 같습니다.

전체 예제 ZIP / 실행법

"""Real Inspect Task/MockLLM/scorer loop with synthetic artifacts; no paid model."""
from __future__ import annotations

import hashlib
import json
import os
from pathlib import Path
import socket
import sys

HERE = Path(__file__).resolve().parent
EXPERIMENTS = HERE.parent
sys.path.insert(0, str(EXPERIMENTS))
from common import cases, grade_artifact, input_hashes  # noqa: E402

# This is an application-level guard, not an OS sandbox. No secrets are needed.
for key in tuple(os.environ):
    if any(marker in key for marker in ("API_KEY", "ACCESS_TOKEN", "AUTH_TOKEN")):
        os.environ.pop(key)
os.environ["INSPECT_TELEMETRY_ENABLED"] = "false"
_original_connect = socket.socket.connect
_original_connect_ex = socket.socket.connect_ex


def guarded_connect(self, address):
    if self.family in (socket.AF_INET, socket.AF_INET6):
        raise RuntimeError("experiment blocks IP network connections")
    return _original_connect(self, address)


def guarded_connect_ex(self, address):
    if self.family in (socket.AF_INET, socket.AF_INET6):
        raise RuntimeError("experiment blocks IP network connections")
    return _original_connect_ex(self, address)


socket.socket.connect = guarded_connect
socket.socket.connect_ex = guarded_connect_ex

from inspect_ai import Task, eval  # noqa: E402
from inspect_ai.dataset import Sample  # noqa: E402
from inspect_ai.model import ModelOutput, ModelUsage, get_model  # noqa: E402
from inspect_ai.scorer import Score, mean, scorer  # noqa: E402
from inspect_ai.solver import generate  # noqa: E402

PIN = "456d982ef0b2344ca436b4263cebb45492a85edb"
CONTROLS = {case.name: case for case in cases()}


def mock_output(messages, tools, tool_choice, config):
    case_id = messages[-1].text.splitlines()[0].removeprefix("CONTROL=")
    if case_id == "injected_model_error":
        raise RuntimeError("synthetic model transport failure")
    result = ModelOutput.from_content("mockllm", CONTROLS[case_id].output)
    # Synthetic zero usage prevents tokenizer downloads. This is not measured cost.
    result.usage = ModelUsage(input_tokens=0, output_tokens=0, total_tokens=0)
    return result


@scorer(metrics=[mean()])
def artifact_contract():
    async def score(state, target):
        result = grade_artifact(state.output.completion)
        return Score(value=result.reward, answer=state.output.completion,
                     explanation=result.reason)
    return score


def main():
    instruction = (EXPERIMENTS / "fixtures/instruction.txt").read_text()
    ledger = (EXPERIMENTS / "fixtures/ledger.json").read_text()
    samples = [Sample(id=case.name,
                      input=f"CONTROL={case.name}\n{instruction}\n{ledger}",
                      target="independent golden fixture",
                      metadata={"expected_reward": case.expected_reward, "kind": case.kind})
               for case in CONTROLS.values()]
    task = Task(name="synthetic_artifact_contract", dataset=samples,
                solver=generate(), scorer=artifact_contract(), version=1,
                metadata={"study": "synthetic conformance, not model ability"})
    model = get_model("mockllm/model", custom_outputs=mock_output)
    logs = eval(task, model=model, epochs=2, max_samples=2,
                log_dir=str(HERE / "runtime"), display="none", retry_on_error=0)
    log = logs[0]
    rows = []
    for sample in log.samples or []:
        score = sample.scores["artifact_contract"] if sample.scores else None
        case = CONTROLS[str(sample.id)]
        rows.append({"case": sample.id, "epoch": sample.epoch,
                     "expected_reward": case.expected_reward,
                     "reward": score.value if score else None,
                     "reason": score.explanation if score else None,
                     "has_framework_error": sample.error is not None,
                     "artifact_sha256": hashlib.sha256(sample.output.completion.encode()).hexdigest(),
                     "matches_expectation": score is not None and score.value == case.expected_reward
                     and sample.error is None})
    failing_task = Task(name="synthetic_error_boundary",
                        dataset=[Sample(id="injected_model_error", input="CONTROL=injected_model_error")],
                        solver=generate(), scorer=artifact_contract())
    failed = eval(failing_task, model=model, max_samples=1, fail_on_error=False,
                  score_on_error=False, retry_on_error=0,
                  log_dir=str(HERE / "runtime"), display="none")[0]
    fs = failed.samples[0]
    error_boundary = {"kind": "injected_model_error", "has_sample_error": fs.error is not None,
                      "score_present": bool(fs.scores),
                      "not_a_wrong_answer": fs.error is not None and not fs.scores}
    result = {"execution_kind": "framework_execution", "framework": "Inspect AI",
              "commit": PIN, "model": "mockllm/model", "model_ability_measured": False,
              "network_guard": "Python AF_INET/AF_INET6 connect blocked; not an OS sandbox",
              "inputs": input_hashes(), "repeats": 2, "cases": rows,
              "framework_status": log.status, "error_boundary": error_boundary,
              "all_match": len(rows) == 26 and all(r["matches_expectation"] for r in rows)
              and error_boundary["not_a_wrong_answer"]}
    path = EXPERIMENTS / "results/inspect.json"
    path.write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n")
    print(json.dumps({"framework_status": log.status, "all_match": result["all_match"],
                      "cases": len(rows), "error_boundary": error_boundary}))
    if not result["all_match"]:
        raise SystemExit(1)


if __name__ == "__main__":
    main()

SHA-256: 2b4b91241965f5ab3893c7ad5202958d46077065495b4b0b5b26223fcbf1064d