Skip to content

datarobot_genai.eval.validation

validation

preflight_judge 

preflight_judge(judge_cfg: dict[str, Any]) -> None

Ping the judge endpoint with a minimal chat-completions call.

Raises RuntimeError on any non-200 response or network failure so callers can bail before running the full benchmark — catches missing / invalid / expired tokens, wrong model_id, and gateway outages that would otherwise surface only as a cascade of per-case CALL_ERROR results.

Source code in datarobot_genai/eval/validation.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def preflight_judge(judge_cfg: dict[str, Any]) -> None:
    """Ping the judge endpoint with a minimal chat-completions call.

    Raises RuntimeError on any non-200 response or network failure so callers
    can bail before running the full benchmark — catches missing / invalid /
    expired tokens, wrong model_id, and gateway outages that would otherwise
    surface only as a cascade of per-case CALL_ERROR results.
    """
    url = judge_cfg["url"].rstrip("/") + "/chat/completions"
    model_id = judge_cfg["model_id"]
    api_key_name = judge_cfg["api_key_name"]
    token = os.environ.get(api_key_name)
    if not token:
        raise RuntimeError(
            f"Judge preflight failed: env var {api_key_name} is not set "
            f"(judge url={judge_cfg['url']}, model={model_id})"
        )

    # Reasoning models (o-series, etc.) spend tokens on an internal reasoning
    # pass before emitting output, so a tight max_tokens budget yields an HTTP
    # 400 even when auth and model are fine — a false negative. Give the ping
    # enough headroom to finish reasoning and emit at least one output token.
    payload = json.dumps(
        {
            "model": model_id,
            "messages": [{"role": "user", "content": "ping"}],
            "max_tokens": 512,
            "temperature": 0,
        }
    ).encode()
    req = Request(
        url,
        data=payload,
        headers={
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json",
        },
        method="POST",
    )
    try:
        urlopen(req, timeout=15)
    except HTTPError as e:
        body = e.read().decode("utf-8", errors="replace")[:500]
        raise RuntimeError(
            f"Judge preflight failed: HTTP {e.code} from {url} "
            f"(model={model_id}, token env={api_key_name}). Response: {body}"
        ) from e
    except URLError as e:
        raise RuntimeError(
            f"Judge preflight failed: cannot reach {url} (model={model_id}): {e.reason}"
        ) from e

health_check 

health_check(endpoint_url: str) -> str | None

Return None if the server responds at all, else an error string.

Prefer a dedicated /health probe: dragent/DRUM-fronted agents expose a /health route (returning 200) at the host root, so probing it avoids the 404 log noise that pinging a bare /v1-style base URL generates on the agent side. The endpoint's path (e.g. /v1) is stripped so /health resolves against the host:port.

If /health isn't available (some deployed / production endpoints don't expose it), fall back to pinging the literal endpoint and treating any HTTP response (even 4xx/405) as "reachable" — a DRUM agentic-workflow server exposes /chat/completions but not necessarily /v1/models. Only a connection-level failure is fatal.

Source code in datarobot_genai/eval/validation.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def health_check(endpoint_url: str) -> str | None:
    """Return None if the server responds at all, else an error string.

    Prefer a dedicated /health probe: dragent/DRUM-fronted agents expose a
    /health route (returning 200) at the host root, so probing it avoids the
    404 log noise that pinging a bare /v1-style base URL generates on the
    agent side. The endpoint's path (e.g. /v1) is stripped so /health resolves
    against the host:port.

    If /health isn't available (some deployed / production endpoints don't
    expose it), fall back to pinging the literal endpoint and treating any HTTP
    response (even 4xx/405) as "reachable" — a DRUM agentic-workflow server
    exposes /chat/completions but not necessarily /v1/models. Only a
    connection-level failure is fatal.
    """
    parts = urlsplit(endpoint_url)
    if parts.scheme and parts.netloc:
        health_url = f"{parts.scheme}://{parts.netloc}/health"
        try:
            req = Request(health_url, headers={"Accept": "application/json"})
            urlopen(req, timeout=10)
            return None
        except (HTTPError, URLError, OSError):
            # No /health route, or it errored — fall back to the base ping,
            # which either confirms reachability or reports the failure.
            pass

    base = endpoint_url.rstrip("/")
    try:
        req = Request(base, headers={"Accept": "application/json"})
        urlopen(req, timeout=10)
        return None
    except HTTPError:
        # Server responded (e.g. 404/405 on the root path) — it's reachable.
        return None
    except URLError as e:
        return f"Endpoint not reachable at {base}: {e.reason}"
    except Exception as e:  # noqa: BLE001
        return f"Endpoint check failed: {e}"