Skip to main content

evals.common

Shared, transport-agnostic helpers for the evals runners.

These helpers are pure functions used by both the in-process :class:agent_runtimes.evals.local.EvalRunner (local) and the platform orchestrator :func:agent_runtimes.evals.remote.runner.execute_evalset_spec (remote). Keeping them here avoids duplicating prompt extraction, output text coercion, and token-usage aggregation logic across runners.

@module agent_runtimes.evals.common

case_prompt​

def case_prompt(case: dict[str, Any]) -> str

Extract a prompt string from an evalset case's inputs.

@param case - Evalset case dict with an inputs field. @returns The best prompt string found, or a JSON dump of the inputs.

compose_case_prompt​

def compose_case_prompt(case: dict[str, Any], *, preamble: str = "") -> str

Build the effective case prompt with an optional preamble.

preamble lets a spec enforce task instructions (for example output format/constraints) without mutating every individual case input.

@param case - Evalset case dict. @param preamble - Optional instruction text prepended to the case prompt. @returns The composed prompt string.

extract_text​

def extract_text(payload: Any) -> str

Coerce an agent output payload into a plain text answer.

@param payload - Agent output (dict, str, or other). @returns A plain text representation of the payload.

usage_number​

def usage_number(value: Any) -> float | None

Coerce a usage value into a float, or None when not numeric.

@param value - Candidate usage value. @returns The numeric value as a float, or None.

usage_pick_number​

def usage_pick_number(usage: dict[str, Any], *keys: str) -> float | None

Return the first numeric value found among keys in usage.

@param usage - Usage mapping to inspect. @param keys - Candidate keys, checked in order. @returns The first numeric value found, or None.

extract_case_usage​

def extract_case_usage(chat_result: dict[str, Any]) -> dict[str, Any]

Extract a token-usage mapping from an agent chat result.

@param chat_result - Result payload from an agent chat call. @returns A copy of the usage mapping, or an empty dict when absent.

merge_run_usage​

def merge_run_usage(aggregate: dict[str, Any],
                    case_usage: dict[str, Any]) -> dict[str, Any]

Merge a single case's usage into a run-level aggregate in place.

Numeric token/credit fields are summed (credits kept as floats, token/ request/duration fields rounded to ints). Descriptive fields (provider, model, account/requester identifiers, timestamps) are taken from the first case that provides them.

@param aggregate - The run-level usage aggregate (mutated and returned). @param case_usage - The per-case usage mapping to merge in. @returns The updated aggregate mapping.