API reference¶
Generated from the apogee-ai-eval source with mkdocstrings. Every symbol below is exported from apogee_ai_eval, so it is part of the supported public surface.
Application · DTOs¶
Bases: BaseModel
Bases: BaseModel
Bases: BaseModel
Application · Use cases¶
Python
CompareRunsUseCase(run_repository: IRunRepository, baseline_repository: IBaselineRepository)
Diff a run against the recorded baseline.
Source code in apogee_ai_eval/application/use_cases/compare_runs_use_case.py
class-attribute
instance-attribute
¶
Score deltas within ±0.5% are considered noise.
async
¶
Python
execute(dto: CompareRunsDTO) -> Comparison
Source code in apogee_ai_eval/application/use_cases/compare_runs_use_case.py
Python
async def execute(self, dto: CompareRunsDTO) -> Comparison:
run = await self._runs.find(dto.current_run_id)
if run is None:
raise RunNotFoundException(dto.current_run_id)
baseline = await self._baselines.find(dto.suite, label=dto.baseline_label)
baseline_score = baseline.aggregate_score if baseline else None
delta = (run.aggregate_score - baseline_score) if baseline_score is not None else 0.0
status = self._classify(delta) if baseline_score is not None else RegressionStatus.NEW
cases: list[CaseComparison] = []
seen_ids: set[str] = set()
for case in run.cases:
seen_ids.add(case.case_id)
base_score = baseline.case_scores.get(case.case_id) if baseline else None
current = case.aggregate_score
d = (current - base_score) if base_score is not None else 0.0
case_status = self._classify(d) if base_score is not None else RegressionStatus.NEW
metric_comparisons = self._compare_metrics(case, baseline)
cases.append(
CaseComparison(
case_id=case.case_id,
metric_comparisons=metric_comparisons,
aggregate_baseline=base_score,
aggregate_current=current,
aggregate_delta=d,
status=case_status,
)
)
# Account for cases removed from current run but present in baseline
if baseline:
for case_id, base_score in baseline.case_scores.items():
if case_id in seen_ids:
continue
cases.append(
CaseComparison(
case_id=case_id,
metric_comparisons=(),
aggregate_baseline=base_score,
aggregate_current=None,
aggregate_delta=-base_score,
status=RegressionStatus.REMOVED,
)
)
return Comparison(
suite=dto.suite,
baseline_label=baseline.label if baseline else None,
current_run_id=run.id,
cases=tuple(sorted(cases, key=lambda c: c.case_id)),
aggregate_baseline=baseline_score,
aggregate_current=run.aggregate_score,
aggregate_delta=delta,
status=status,
)
Python
GetRunUseCase(run_repository: IRunRepository)
Source code in apogee_ai_eval/application/use_cases/get_run_use_case.py
Python
ListRunsUseCase(run_repository: IRunRepository)
Python
RegressionGateUseCase(suite_repository: ISuiteRepository, run_repository: IRunRepository, baseline_repository: IBaselineRepository)
Decides whether current_run_id is acceptable vs the baseline.
Combines
- explicit
min_scoreandmax_dropfrom the DTO - the suite's optional
RegressionGate
Returns GateResultDTO with passed and a list of human-readable
reasons. Use RegressionGateException if you prefer raising.
Source code in apogee_ai_eval/application/use_cases/regression_gate_use_case.py
async
¶
Python
execute(dto: GateDTO) -> GateResultDTO
Source code in apogee_ai_eval/application/use_cases/regression_gate_use_case.py
Python
async def execute(self, dto: GateDTO) -> GateResultDTO:
run = await self._runs.find(dto.current_run_id)
if run is None:
raise RunNotFoundException(dto.current_run_id)
suite = await self._suites.find(dto.suite)
baseline = await self._baselines.find(dto.suite, label=dto.baseline_label)
if baseline is None and dto.require_baseline:
raise BaselineNotFoundException(dto.suite, dto.baseline_label)
min_score = dto.min_score if dto.min_score is not None else (
suite.gate.min_score if suite and suite.gate else None
)
max_drop = dto.max_drop if dto.max_drop is not None else (
suite.gate.max_drop if suite and suite.gate else None
)
baseline_score = baseline.aggregate_score if baseline else None
delta = (run.aggregate_score - baseline_score) if baseline_score is not None else 0.0
reasons: list[str] = []
passed = True
if run.errors:
passed = False
reasons.append(f"{run.errors} case(s) errored")
if min_score is not None and run.aggregate_score < min_score:
passed = False
reasons.append(
f"aggregate score {run.aggregate_score:.3f} below min {min_score:.3f}"
)
if max_drop is not None and baseline_score is not None and -delta > max_drop:
passed = False
reasons.append(
f"score dropped {-delta:.3f} (max allowed {max_drop:.3f})"
)
return GateResultDTO(
suite=dto.suite,
passed=passed,
aggregate_score=run.aggregate_score,
baseline_score=baseline_score,
delta=delta,
reasons=reasons,
)
Python
RunSuiteUseCase(suite_repository: ISuiteRepository, run_repository: IRunRepository, invoker: IAgentInvoker, metrics: Mapping[str, IMetric])
Executes every case + metric in the suite, persists the run.
Source code in apogee_ai_eval/application/use_cases/run_suite_use_case.py
async
¶
Python
execute(dto: RunSuiteDTO) -> EvalRun
Source code in apogee_ai_eval/application/use_cases/run_suite_use_case.py
Python
async def execute(self, dto: RunSuiteDTO) -> EvalRun:
suite = await self._suites.find(dto.suite)
if suite is None:
raise SuiteNotFoundException(dto.suite)
case_ids = set(dto.case_ids) if dto.case_ids else None
case_results: list[CaseResult] = []
for case in suite.cases:
if case_ids is not None and case.id not in case_ids:
continue
invocation = await self._invoke(suite, case_id=case.id, case=case, dto=dto)
metric_results = await self._evaluate(suite, case=case, invocation=invocation)
case_results.append(
CaseResult(case_id=case.id, invocation=invocation, metrics=tuple(metric_results))
)
run = EvalRun(
suite=suite.name,
cases=tuple(case_results),
finished_at=datetime.now(timezone.utc),
metadata=dict(dto.metadata),
)
await self._runs.save(run)
return run
Python
SetBaselineUseCase(run_repository: IRunRepository, baseline_repository: IBaselineRepository)
Promote a run to be the baseline for label.
Source code in apogee_ai_eval/application/use_cases/set_baseline_use_case.py
async
¶
Python
execute(dto: SetBaselineDTO) -> Baseline
Source code in apogee_ai_eval/application/use_cases/set_baseline_use_case.py
Python
async def execute(self, dto: SetBaselineDTO) -> Baseline:
run = await self._runs.find(dto.run_id)
if run is None:
raise RunNotFoundException(dto.run_id)
case_scores: dict[str, float] = {}
metric_scores: dict[str, dict[str, float]] = {}
for case in run.cases:
case_scores[case.case_id] = case.aggregate_score
metric_scores[case.case_id] = {m.metric_name: m.score for m in case.metrics}
baseline = Baseline(
suite=dto.suite,
label=dto.label,
run_id=run.id,
aggregate_score=run.aggregate_score,
case_scores=case_scores,
metric_scores=metric_scores,
)
return await self._baselines.save(baseline)
Domain¶
dataclass
¶
Python
Baseline(suite: str, label: str = 'main', run_id: str = '', aggregate_score: float = 0.0, case_scores: dict[str, float] = dict(), metric_scores: dict[str, dict[str, float]] = dict(), created_at: datetime = (lambda: now(utc))())
Frozen reference run used as the regression target.
Stored per-suite, optionally tagged (label) so multiple lines such
as main, staging, release-1.4 can co-exist.
dataclass
¶
Python
CaseComparison(case_id: str, metric_comparisons: tuple[MetricComparison, ...], aggregate_baseline: float | None, aggregate_current: float | None, aggregate_delta: float, status: RegressionStatus)
dataclass
¶
Python
CaseInvocation(case_id: str, output: str, latency_ms: float = 0.0, cost_usd: float = 0.0, input_tokens: int = 0, output_tokens: int = 0, error: str | None = None, tool_trajectory: tuple[str, ...] = tuple())
The actual response captured for a case (output + observed metrics).
Returned by an :class:IAgentInvoker so metrics can be computed.
class-attribute
instance-attribute
¶
Actual tool-call sequence the agent took (tool names, in order).
dataclass
¶
Python
CaseResult(case_id: str, invocation: CaseInvocation, metrics: tuple[MetricResult, ...] = tuple())
class-attribute
instance-attribute
¶
Python
metrics: tuple[MetricResult, ...] = field(default_factory=tuple)
dataclass
¶
Python
Comparison(suite: str, baseline_label: str | None, current_run_id: str, cases: tuple[CaseComparison, ...] = tuple(), aggregate_baseline: float | None = None, aggregate_current: float | None = None, aggregate_delta: float = 0.0, status: RegressionStatus = UNCHANGED)
class-attribute
instance-attribute
¶
Python
cases: tuple[CaseComparison, ...] = field(default_factory=tuple)
class-attribute
instance-attribute
¶
dataclass
¶
Python
EvalCase(id: str, input: str, expected: str | None = None, expected_substrings: tuple[str, ...] = tuple(), expected_json: dict[str, Any] | None = None, context: tuple[str, ...] = tuple(), metadata: dict[str, str] = dict(), tags: tuple[str, ...] = tuple(), expected_tools: tuple[str, ...] = tuple())
A single input → expected output pair.
class-attribute
instance-attribute
¶
class-attribute
instance-attribute
¶
class-attribute
instance-attribute
¶
Reference passages used by faithfulness/RAG metrics.
class-attribute
instance-attribute
¶
class-attribute
instance-attribute
¶
class-attribute
instance-attribute
¶
Expected tool-call trajectory (tool names, in order) — used by the trajectory metric.
dataclass
¶
Python
EvalRun(id: str = (lambda: hex)(), suite: str = '', started_at: datetime = (lambda: now(utc))(), finished_at: datetime | None = None, cases: tuple[CaseResult, ...] = tuple(), metadata: dict[str, str] = dict())
Results of running a complete suite once.
class-attribute
instance-attribute
¶
class-attribute
instance-attribute
¶
Python
cases: tuple[CaseResult, ...] = field(default_factory=tuple)
class-attribute
instance-attribute
¶
dataclass
¶
Python
EvalSuite(name: str, agent: str | None = None, description: str | None = None, cases: tuple[EvalCase, ...] = tuple(), metrics: tuple[MetricSpec, ...] = tuple(), gate: RegressionGate | None = None, tags: tuple[str, ...] = tuple(), metadata: dict[str, str] = dict())
A named bundle of cases + metrics + (optional) regression gate.
class-attribute
instance-attribute
¶
Logical id of the system-under-test (a slug your invoker resolves).
class-attribute
instance-attribute
¶
Python
cases: tuple[EvalCase, ...] = field(default_factory=tuple)
class-attribute
instance-attribute
¶
Python
metrics: tuple[MetricSpec, ...] = field(default_factory=tuple)
class-attribute
instance-attribute
¶
class-attribute
instance-attribute
¶
Python
metrics_for_case(case_id: str) -> tuple[MetricSpec, ...]
dataclass
¶
Python
JudgeVerdict(score: float, reason: str = '', judge: str = 'rule-based', metadata: dict[str, str] = dict())
Decision returned by an LLM-as-judge.
dataclass
¶
Python
MetricComparison(metric_kind: str, baseline_score: float | None, current_score: float | None, delta: float, status: RegressionStatus)
dataclass
¶
Python
MetricResult(metric_name: str, metric_kind: str, case_id: str, score: float, verdict: Verdict, weight: float = 1.0, reason: str | None = None, details: dict[str, str] = dict())
Outcome of running one MetricSpec against one CaseInvocation.
class-attribute
instance-attribute
¶
dataclass
¶
Python
MetricSpec(kind: MetricKind, case_id: str | None = None, name: str | None = None, weight: float = 1.0, params: dict[str, Any] = dict())
Declaration of a metric inside a suite.
Bound to a specific case via case_id (None = applies to all
cases). params is metric-specific config (substrings, thresholds,
judge config, etc.).
class-attribute
instance-attribute
¶
dataclass
¶
Python
RegressionGate(min_score: float | None = None, max_drop: float | None = None, severity: Severity = CRITICAL)
Threshold definition that determines whether a run is acceptable.
min_score is an absolute lower bound. max_drop is the largest
score regression we tolerate vs the baseline (e.g. 0.02 means we accept
a 2 pp drop). Both are optional — at least one must be set.
Bases: str, Enum
class-attribute
instance-attribute
¶
Suite/case that didn't exist in the baseline.
dataclass
¶
Domain · Enums¶
Bases: str, Enum
Identifier for built-in metric implementations.
class-attribute
instance-attribute
¶
Domain · Exceptions¶
Bases: EvalError
Source code in apogee_ai_eval/domain/exceptions/eval_exceptions.py
Bases: EvalError
Source code in apogee_ai_eval/domain/exceptions/eval_exceptions.py
Bases: Exception
Base for all apogee-ai-eval errors.
Bases: EvalError
Source code in apogee_ai_eval/domain/exceptions/eval_exceptions.py
Domain · Protocols (ports)¶
Bases: Protocol
Bridges the eval framework with the system-under-test.
Implementations call apogee-ai, an HTTP API, a local function, etc. The invoker is responsible for measuring latency, cost and tokens.
async
¶
Python
invoke(*, agent: str | None, case: EvalCase) -> CaseInvocation
Bases: Protocol
LLM-as-judge: scores an output (0..1) given a case.
async
¶
Python
judge(*, case: EvalCase, output: str, criterion: str) -> JudgeVerdict
Bases: Protocol
Computes a single metric for one case + invocation pair.
instance-attribute
¶
Identifier matching the MetricKind enum value this metric handles.
async
¶
Python
evaluate(*, case: EvalCase, invocation: CaseInvocation, spec: MetricSpec) -> MetricResult
Bases: Protocol
Python
render_comparison(comparison: Comparison) -> str
Infrastructure¶
Python
AnthropicJudge(*, api_key: str | None = None, model: str = 'claude-haiku-4-5-20251001', max_tokens: int = 256)
LLM-as-judge backed by Anthropic Messages API.
Lazy-imports anthropic so installing apogee-ai-eval does not pull it
in unless [anthropic] extra is requested.
Source code in apogee_ai_eval/infrastructure/judges/anthropic_judge.py
Python
def __init__(
self,
*,
api_key: str | None = None,
model: str = "claude-haiku-4-5-20251001",
max_tokens: int = 256,
) -> None:
try:
import anthropic # type: ignore # noqa: F401
except ImportError as exc:
raise ImportError(
"AnthropicJudge requires `anthropic`. "
"Install with: pip install 'apogee-ai-eval[anthropic]'"
) from exc
self._api_key = api_key
self._model = model
self._max_tokens = max_tokens
async
¶
Python
judge(*, case: EvalCase, output: str, criterion: str) -> JudgeVerdict
Source code in apogee_ai_eval/infrastructure/judges/anthropic_judge.py
Python
async def judge(
self,
*,
case: EvalCase,
output: str,
criterion: str,
) -> JudgeVerdict:
try:
import anthropic # type: ignore
except ImportError as exc: # pragma: no cover
raise JudgeException(str(exc), judge=self.name) from exc
client = anthropic.AsyncAnthropic(api_key=self._api_key)
try:
response = await client.messages.create(
model=self._model,
max_tokens=self._max_tokens,
system=JUDGE_SYSTEM_PROMPT,
messages=[
{
"role": "user",
"content": build_user_prompt(
case=case, output=output, criterion=criterion
),
}
],
)
except Exception as exc: # noqa: BLE001 - surface as JudgeException
raise JudgeException(f"Anthropic API error: {exc}", judge=self.name) from exc
text = "".join(
block.text # type: ignore[attr-defined]
for block in response.content
if getattr(block, "type", None) == "text"
)
verdict = parse_judge_response(text, judge_name=self.name)
return JudgeVerdict(
score=verdict.score,
reason=verdict.reason,
judge=f"anthropic:{self._model}",
metadata={
"input_tokens": str(response.usage.input_tokens),
"output_tokens": str(response.usage.output_tokens),
},
)
Pass when the output contains every required substring.
Substrings come from
spec.params['expect']— single string, or list of strings- else from
case.expected_substrings
async
¶
Python
evaluate(*, case: EvalCase, invocation: CaseInvocation, spec: MetricSpec) -> MetricResult
Source code in apogee_ai_eval/infrastructure/metrics/contains_metric.py
Python
async def evaluate(
self,
*,
case: EvalCase,
invocation: CaseInvocation,
spec: MetricSpec,
) -> MetricResult:
substrings = self._substrings(case, spec)
if not substrings:
return MetricResult(
metric_name=spec.display_name,
metric_kind=self.kind,
case_id=case.id,
score=0.0,
verdict=Verdict.SKIP,
weight=spec.weight,
reason="No expected substrings provided",
)
case_sensitive = bool(spec.params.get("case_sensitive", False))
haystack = invocation.output if case_sensitive else invocation.output.lower()
needles = substrings if case_sensitive else [s.lower() for s in substrings]
matched = [n for n in needles if n in haystack]
score = len(matched) / len(needles)
verdict = Verdict.PASS if score >= 1.0 else Verdict.FAIL
return MetricResult(
metric_name=spec.display_name,
metric_kind=self.kind,
case_id=case.id,
score=score,
verdict=verdict,
weight=spec.weight,
reason=(
f"Matched {len(matched)}/{len(needles)} substrings"
if score < 1
else "All substrings present"
),
details={"missing": ", ".join(s for s in needles if s not in haystack)},
)
Pass when invocation.cost_usd <= threshold_usd.
async
¶
Python
evaluate(*, case: EvalCase, invocation: CaseInvocation, spec: MetricSpec) -> MetricResult
Source code in apogee_ai_eval/infrastructure/metrics/cost_metric.py
Python
async def evaluate(
self,
*,
case: EvalCase,
invocation: CaseInvocation,
spec: MetricSpec,
) -> MetricResult:
threshold = spec.params.get("threshold_usd")
if threshold is None or threshold < 0:
return MetricResult(
metric_name=spec.display_name,
metric_kind=self.kind,
case_id=case.id,
score=0.0,
verdict=Verdict.SKIP,
weight=spec.weight,
reason="No threshold_usd set",
)
cost = invocation.cost_usd
if cost <= threshold:
score = 1.0
verdict = Verdict.PASS
elif threshold == 0:
score = 0.0
verdict = Verdict.FAIL
else:
score = max(0.0, 1.0 - (cost - threshold) / threshold)
verdict = Verdict.FAIL
return MetricResult(
metric_name=spec.display_name,
metric_kind=self.kind,
case_id=case.id,
score=score,
verdict=verdict,
weight=spec.weight,
reason=f"cost=${cost:.6f} threshold=${threshold:.6f}",
)
Bridge to DeepEval metrics (Hallucination, Faithfulness, Toxicity, etc).
Lazy-imports deepeval. spec.params['metric'] selects which class
to instantiate: faithfulness | answer_relevancy | hallucination |
toxicity | bias.
Source code in apogee_ai_eval/infrastructure/bridges/deepeval_adapter.py
async
¶
Python
evaluate(*, case: EvalCase, invocation: CaseInvocation, spec: MetricSpec) -> MetricResult
Source code in apogee_ai_eval/infrastructure/bridges/deepeval_adapter.py
Python
async def evaluate(
self,
*,
case: EvalCase,
invocation: CaseInvocation,
spec: MetricSpec,
) -> MetricResult:
try:
from deepeval.metrics import ( # type: ignore
AnswerRelevancyMetric,
BiasMetric,
FaithfulnessMetric,
HallucinationMetric,
ToxicityMetric,
)
from deepeval.test_case import LLMTestCase # type: ignore
except ImportError as exc: # pragma: no cover
raise MetricException(str(exc), metric=spec.display_name) from exc
chosen = spec.params.get("metric", "faithfulness")
metric_cls = {
"faithfulness": FaithfulnessMetric,
"answer_relevancy": AnswerRelevancyMetric,
"hallucination": HallucinationMetric,
"toxicity": ToxicityMetric,
"bias": BiasMetric,
}.get(chosen, FaithfulnessMetric)
threshold = float(spec.params.get("threshold", 0.7))
metric = metric_cls(threshold=threshold)
test_case = LLMTestCase(
input=case.input,
actual_output=invocation.output,
expected_output=case.expected,
retrieval_context=list(case.context) or None,
context=list(case.context) or None,
)
try:
metric.measure(test_case)
except Exception as exc: # noqa: BLE001
raise MetricException(
f"deepeval error: {exc}", metric=spec.display_name
) from exc
score = float(getattr(metric, "score", 0.0) or 0.0)
score = max(0.0, min(1.0, score))
return MetricResult(
metric_name=spec.display_name,
metric_kind=self.kind,
case_id=case.id,
score=score,
verdict=Verdict.PASS if score >= threshold else Verdict.FAIL,
weight=spec.weight,
reason=f"deepeval.{chosen}={score:.2f}",
details={"reason": str(getattr(metric, "reason", "") or "")},
)
Toy invoker — echoes the input. Used by tests and demos.
async
¶
Python
invoke(*, agent: str | None, case: EvalCase) -> CaseInvocation
async
¶
Python
evaluate(*, case: EvalCase, invocation: CaseInvocation, spec: MetricSpec) -> MetricResult
Source code in apogee_ai_eval/infrastructure/metrics/equals_metric.py
Python
async def evaluate(
self,
*,
case: EvalCase,
invocation: CaseInvocation,
spec: MetricSpec,
) -> MetricResult:
expected = spec.params.get("expect", case.expected)
if expected is None:
return MetricResult(
metric_name=spec.display_name,
metric_kind=self.kind,
case_id=case.id,
score=0.0,
verdict=Verdict.SKIP,
weight=spec.weight,
reason="No expected value",
)
case_sensitive = bool(spec.params.get("case_sensitive", True))
strip = bool(spec.params.get("strip", True))
a = str(expected)
b = invocation.output
if strip:
a, b = a.strip(), b.strip()
if not case_sensitive:
a, b = a.lower(), b.lower()
equal = a == b
return MetricResult(
metric_name=spec.display_name,
metric_kind=self.kind,
case_id=case.id,
score=1.0 if equal else 0.0,
verdict=Verdict.PASS if equal else Verdict.FAIL,
weight=spec.weight,
reason="Equal" if equal else f"Expected {a!r}, got {b!r}",
)
Python
FunctionAgentInvoker(fn: Callable[[EvalCase], Any], *, invoker_name: str | None = None)
Wraps a Python callable so any function can act as an agent.
Pass either an awaitable returning a string or a sync function. Latency is measured around the call.
Source code in apogee_ai_eval/infrastructure/scorers/echo_agent_invoker.py
async
¶
Python
invoke(*, agent: str | None, case: EvalCase) -> CaseInvocation
Source code in apogee_ai_eval/infrastructure/scorers/echo_agent_invoker.py
Python
async def invoke(self, *, agent: str | None, case: EvalCase) -> CaseInvocation:
start = time.perf_counter()
try:
output = self._fn(case)
if asyncio.iscoroutine(output) or isinstance(output, Awaitable):
output = await output # type: ignore[assignment]
err = None
except Exception as exc: # noqa: BLE001
output = ""
err = str(exc)
latency = (time.perf_counter() - start) * 1000.0
return CaseInvocation(
case_id=case.id,
output=str(output),
latency_ms=latency,
error=err,
)
Source code in apogee_ai_eval/infrastructure/datasets/json_run_repository.py
async
¶
async
¶
Source code in apogee_ai_eval/infrastructure/datasets/json_run_repository.py
Python
async def list(self, *, suite: str | None = None, limit: int | None = None) -> list[EvalRun]:
items = sorted(self._store.values(), key=lambda r: r.started_at, reverse=True)
if suite is not None:
items = [r for r in items if r.suite == suite]
if limit is not None:
items = items[:limit]
return items
One JSON per <root>/<suite>/<label>.json.
Source code in apogee_ai_eval/infrastructure/datasets/json_baseline_repository.py
Pass when the output JSON is a superset of the expected payload.
spec.params['expect'] (or case.expected_json) is a dict whose
keys/values must exist in the parsed output. Lists are matched by
length-and-position; nested dicts recurse.
async
¶
Python
evaluate(*, case: EvalCase, invocation: CaseInvocation, spec: MetricSpec) -> MetricResult
Source code in apogee_ai_eval/infrastructure/metrics/json_match_metric.py
Python
async def evaluate(
self,
*,
case: EvalCase,
invocation: CaseInvocation,
spec: MetricSpec,
) -> MetricResult:
expected = spec.params.get("expect", case.expected_json)
if expected is None:
return MetricResult(
metric_name=spec.display_name,
metric_kind=self.kind,
case_id=case.id,
score=0.0,
verdict=Verdict.SKIP,
weight=spec.weight,
reason="No expected JSON",
)
try:
actual = json.loads(invocation.output)
except json.JSONDecodeError as exc:
return MetricResult(
metric_name=spec.display_name,
metric_kind=self.kind,
case_id=case.id,
score=0.0,
verdict=Verdict.FAIL,
weight=spec.weight,
reason=f"Output is not valid JSON: {exc}",
)
diff = self._missing(expected, actual, path="$")
score = 1.0 if not diff else 0.0
return MetricResult(
metric_name=spec.display_name,
metric_kind=self.kind,
case_id=case.id,
score=score,
verdict=Verdict.PASS if score == 1.0 else Verdict.FAIL,
weight=spec.weight,
reason="JSON matches" if score == 1.0 else f"Missing/mismatched at: {', '.join(diff)}",
)
Python
render_comparison(comparison: Comparison) -> str
One JSON file per run on <root>/<run_id>.json.
Index file <root>/_index.json keeps an ordered list of run ids → suite
mapping for fast list/latest queries.
Source code in apogee_ai_eval/infrastructure/datasets/json_run_repository.py
Python
JudgeCustomMetric(judge: IJudge, *, threshold: float | None = None)
Bases: _JudgeBackedMetric
Source code in apogee_ai_eval/infrastructure/metrics/judge_metric.py
class-attribute
instance-attribute
¶
Python
JudgeFaithfulnessMetric(judge: IJudge, *, threshold: float | None = None)
Bases: _JudgeBackedMetric
Source code in apogee_ai_eval/infrastructure/metrics/judge_metric.py
Python
JudgeRelevanceMetric(judge: IJudge, *, threshold: float | None = None)
Bases: _JudgeBackedMetric
Source code in apogee_ai_eval/infrastructure/metrics/judge_metric.py
Python
JudgeToxicityMetric(judge: IJudge, *, threshold: float | None = None)
Bases: _JudgeBackedMetric
Source code in apogee_ai_eval/infrastructure/metrics/judge_metric.py
JUnit XML reporter, consumable by GitHub Actions / GitLab CI.
Source code in apogee_ai_eval/infrastructure/reporters/junit_reporter.py
Python
def render_run(self, suite: EvalSuite, run: EvalRun) -> str:
ts = ET.Element(
"testsuite",
attrib={
"name": run.suite or suite.name,
"tests": str(run.total),
"failures": str(run.failed),
"errors": str(run.errors),
"time": "0",
},
)
for case in run.cases:
tc = ET.SubElement(
ts,
"testcase",
attrib={
"classname": run.suite or suite.name,
"name": case.case_id,
"time": f"{case.invocation.latency_ms / 1000.0:.3f}",
},
)
if case.verdict == Verdict.FAIL:
msg = "; ".join(
f"{m.metric_name}: {m.reason or ''}"
for m in case.metrics
if m.verdict == Verdict.FAIL
)
fail = ET.SubElement(tc, "failure", attrib={"message": msg or "fail"})
fail.text = msg
elif case.verdict == Verdict.ERROR:
msg = "; ".join(
m.reason or "" for m in case.metrics if m.verdict == Verdict.ERROR
)
err = ET.SubElement(tc, "error", attrib={"message": msg or "error"})
err.text = msg
elif case.verdict == Verdict.SKIP:
ET.SubElement(tc, "skipped")
return '<?xml version="1.0" encoding="UTF-8"?>\n' + ET.tostring(
ts, encoding="unicode"
)
Python
render_comparison(comparison: Comparison) -> str
Pass when invocation.latency_ms <= threshold_ms.
Produces a soft-score: 1.0 at or below threshold, 0.0 at 2x threshold, linearly interpolated in between.
async
¶
Python
evaluate(*, case: EvalCase, invocation: CaseInvocation, spec: MetricSpec) -> MetricResult
Source code in apogee_ai_eval/infrastructure/metrics/latency_metric.py
Python
async def evaluate(
self,
*,
case: EvalCase,
invocation: CaseInvocation,
spec: MetricSpec,
) -> MetricResult:
threshold = spec.params.get("threshold_ms")
if threshold is None or threshold <= 0:
return MetricResult(
metric_name=spec.display_name,
metric_kind=self.kind,
case_id=case.id,
score=0.0,
verdict=Verdict.SKIP,
weight=spec.weight,
reason="No threshold_ms set",
)
latency = invocation.latency_ms
if latency <= threshold:
score = 1.0
verdict = Verdict.PASS
elif latency >= 2 * threshold:
score = 0.0
verdict = Verdict.FAIL
else:
score = max(0.0, 1.0 - (latency - threshold) / threshold)
verdict = Verdict.FAIL
return MetricResult(
metric_name=spec.display_name,
metric_kind=self.kind,
case_id=case.id,
score=score,
verdict=verdict,
weight=spec.weight,
reason=f"latency={latency:.0f}ms threshold={threshold:.0f}ms",
details={"latency_ms": f"{latency:.2f}", "threshold_ms": f"{threshold:.2f}"},
)
Pass when output length sits within [min_chars, max_chars].
async
¶
Python
evaluate(*, case: EvalCase, invocation: CaseInvocation, spec: MetricSpec) -> MetricResult
Source code in apogee_ai_eval/infrastructure/metrics/length_metric.py
Python
async def evaluate(
self,
*,
case: EvalCase,
invocation: CaseInvocation,
spec: MetricSpec,
) -> MetricResult:
text = invocation.output
min_chars = int(spec.params.get("min_chars", 0))
max_chars = spec.params.get("max_chars")
actual = len(text)
too_short = actual < min_chars
too_long = max_chars is not None and actual > int(max_chars)
passed = not too_short and not too_long
if passed:
return MetricResult(
metric_name=spec.display_name,
metric_kind=self.kind,
case_id=case.id,
score=1.0,
verdict=Verdict.PASS,
weight=spec.weight,
reason=f"len={actual}",
)
return MetricResult(
metric_name=spec.display_name,
metric_kind=self.kind,
case_id=case.id,
score=0.0,
verdict=Verdict.FAIL,
weight=spec.weight,
reason=(
f"len={actual} too short (min={min_chars})"
if too_short
else f"len={actual} too long (max={max_chars})"
),
)
Source code in apogee_ai_eval/infrastructure/reporters/markdown_reporter.py
Python
def render_run(self, suite: EvalSuite, run: EvalRun) -> str:
lines = [
f"# Eval run — {run.suite}",
"",
f"- run id: `{run.id}`",
f"- aggregate score: **{run.aggregate_score:.3f}**",
f"- verdict: **{run.verdict.value}**",
f"- pass/fail/err: {run.passed}/{run.failed}/{run.errors} of {run.total}",
"",
"| Case | Score | Verdict | Metrics |",
"|---|---:|:---:|---|",
]
for case in run.cases:
metric_summary = ", ".join(
f"{m.metric_name}={m.score:.2f}" for m in case.metrics
)
lines.append(
f"| `{case.case_id}` | {case.aggregate_score:.2f} "
f"| {_VERDICT_EMOJI.get(case.verdict, '?')} {case.verdict.value} "
f"| {metric_summary} |"
)
return "\n".join(lines) + "\n"
Python
render_comparison(comparison: Comparison) -> str
Source code in apogee_ai_eval/infrastructure/reporters/markdown_reporter.py
Python
def render_comparison(self, comparison: Comparison) -> str:
lines = [
f"# Eval regression — {comparison.suite}",
"",
f"- baseline: `{comparison.baseline_label or '-'}`",
f"- current: `{comparison.current_run_id}`",
f"- aggregate baseline: {comparison.aggregate_baseline if comparison.aggregate_baseline is not None else '-'}",
f"- aggregate current: {comparison.aggregate_current if comparison.aggregate_current is not None else '-'}",
f"- delta: **{comparison.aggregate_delta:+.3f}** ({_REGRESSION_EMOJI.get(comparison.status, '?')} {comparison.status.value})",
"",
"| Case | Baseline | Current | Δ | Status |",
"|---|---:|---:|---:|---|",
]
for case in comparison.cases:
base = (
f"{case.aggregate_baseline:.3f}" if case.aggregate_baseline is not None else "-"
)
cur = (
f"{case.aggregate_current:.3f}" if case.aggregate_current is not None else "-"
)
lines.append(
f"| `{case.case_id}` | {base} | {cur} | {case.aggregate_delta:+.3f} "
f"| {_REGRESSION_EMOJI.get(case.status, '?')} {case.status.value} |"
)
return "\n".join(lines) + "\n"
Python
OpenAIJudge(*, api_key: str | None = None, model: str = 'gpt-4o-mini', max_tokens: int = 256, temperature: float = 0.0)
LLM-as-judge backed by OpenAI Chat Completions.
Lazy-imports openai so installing apogee-ai-eval does not pull it in
unless [openai] extra is requested.
Source code in apogee_ai_eval/infrastructure/judges/openai_judge.py
Python
def __init__(
self,
*,
api_key: str | None = None,
model: str = "gpt-4o-mini",
max_tokens: int = 256,
temperature: float = 0.0,
) -> None:
try:
import openai # type: ignore # noqa: F401
except ImportError as exc:
raise ImportError(
"OpenAIJudge requires `openai`. "
"Install with: pip install 'apogee-ai-eval[openai]'"
) from exc
self._api_key = api_key
self._model = model
self._max_tokens = max_tokens
self._temperature = temperature
async
¶
Python
judge(*, case: EvalCase, output: str, criterion: str) -> JudgeVerdict
Source code in apogee_ai_eval/infrastructure/judges/openai_judge.py
Python
async def judge(
self,
*,
case: EvalCase,
output: str,
criterion: str,
) -> JudgeVerdict:
try:
from openai import AsyncOpenAI # type: ignore
except ImportError as exc: # pragma: no cover
raise JudgeException(str(exc), judge=self.name) from exc
client = AsyncOpenAI(api_key=self._api_key)
try:
response = await client.chat.completions.create(
model=self._model,
max_tokens=self._max_tokens,
temperature=self._temperature,
response_format={"type": "json_object"},
messages=[
{"role": "system", "content": JUDGE_SYSTEM_PROMPT},
{
"role": "user",
"content": build_user_prompt(
case=case, output=output, criterion=criterion
),
},
],
)
except Exception as exc: # noqa: BLE001
raise JudgeException(f"OpenAI API error: {exc}", judge=self.name) from exc
text = response.choices[0].message.content or ""
verdict = parse_judge_response(text, judge_name=self.name)
usage = response.usage
return JudgeVerdict(
score=verdict.score,
reason=verdict.reason,
judge=f"openai:{self._model}",
metadata={
"input_tokens": str(usage.prompt_tokens) if usage else "0",
"output_tokens": str(usage.completion_tokens) if usage else "0",
},
)
Bridge to ragas metrics (faithfulness/answer_relevancy/context_precision).
Lazy-imports ragas so this is only required when used.
Configure via spec.params['metric'] ∈ {faithfulness, answer_relevancy,
context_precision, context_recall}.
Source code in apogee_ai_eval/infrastructure/bridges/ragas_adapter.py
async
¶
Python
evaluate(*, case: EvalCase, invocation: CaseInvocation, spec: MetricSpec) -> MetricResult
Source code in apogee_ai_eval/infrastructure/bridges/ragas_adapter.py
Python
async def evaluate(
self,
*,
case: EvalCase,
invocation: CaseInvocation,
spec: MetricSpec,
) -> MetricResult:
try:
score = await self._score(case=case, output=invocation.output, spec=spec)
except Exception as exc: # noqa: BLE001
raise MetricException(str(exc), metric=spec.display_name) from exc
threshold = float(spec.params.get("threshold", 0.7))
return MetricResult(
metric_name=spec.display_name,
metric_kind=self.kind,
case_id=case.id,
score=score,
verdict=Verdict.PASS if score >= threshold else Verdict.FAIL,
weight=spec.weight,
reason=f"ragas={score:.2f} threshold={threshold:.2f}",
details={"backend": "ragas"},
)
async
¶
Python
evaluate(*, case: EvalCase, invocation: CaseInvocation, spec: MetricSpec) -> MetricResult
Source code in apogee_ai_eval/infrastructure/metrics/regex_metric.py
Python
async def evaluate(
self,
*,
case: EvalCase,
invocation: CaseInvocation,
spec: MetricSpec,
) -> MetricResult:
pattern = spec.params.get("pattern")
if not pattern:
return MetricResult(
metric_name=spec.display_name,
metric_kind=self.kind,
case_id=case.id,
score=0.0,
verdict=Verdict.SKIP,
weight=spec.weight,
reason="No pattern",
)
flags = re.IGNORECASE if spec.params.get("ignore_case") else 0
try:
compiled = re.compile(pattern, flags)
except re.error as exc:
return MetricResult(
metric_name=spec.display_name,
metric_kind=self.kind,
case_id=case.id,
score=0.0,
verdict=Verdict.ERROR,
weight=spec.weight,
reason=f"Invalid regex: {exc}",
)
match = compiled.search(invocation.output)
return MetricResult(
metric_name=spec.display_name,
metric_kind=self.kind,
case_id=case.id,
score=1.0 if match else 0.0,
verdict=Verdict.PASS if match else Verdict.FAIL,
weight=spec.weight,
reason=f"Matched {match.group(0)!r}" if match else "Pattern not found",
)
Heuristic judge — no LLM call, deterministic, useful in tests/CI dry-runs.
For faithfulness: pass when output overlaps strongly with context.
For relevance: pass when output shares 30%+ tokens with the input.
For toxicity: penalize toxic tokens.
async
¶
Python
judge(*, case: EvalCase, output: str, criterion: str) -> JudgeVerdict
Source code in apogee_ai_eval/infrastructure/judges/rule_based_judge.py
Python
async def judge(
self,
*,
case: EvalCase,
output: str,
criterion: str,
) -> JudgeVerdict:
criterion_lc = criterion.lower()
if "faithful" in criterion_lc or "context" in criterion_lc:
return self._faithfulness(case, output)
if "relevan" in criterion_lc or "address" in criterion_lc:
return self._relevance(case, output)
if "toxic" in criterion_lc or "harm" in criterion_lc:
return self._toxicity(output)
return self._token_overlap(case.input, output, label="overlap")
Compares the agent's tool-call trajectory against an expected one.
Expected tools come from spec.params['expect_tools'] (list) or
case.expected_tools. The actual trajectory is invocation.tool_trajectory.
Match mode (spec.params['mode'], default strict):
- strict — exact sequence (order + repetition).
- unordered — same multiset of tools, any order.
- subset — every expected tool appears in the actual run (extras allowed).
- superset — only expected tools appear (some expected may be missing).
- ordered_subsequence — expected is an in-order subsequence of the actual run.
async
¶
Python
evaluate(*, case: EvalCase, invocation: CaseInvocation, spec: MetricSpec) -> MetricResult
Source code in apogee_ai_eval/infrastructure/metrics/trajectory_metric.py
Python
async def evaluate(
self,
*,
case: EvalCase,
invocation: CaseInvocation,
spec: MetricSpec,
) -> MetricResult:
expected = self._expected(case, spec)
if not expected:
return self._result(spec, case, 0.0, Verdict.SKIP, "No expected tools provided", {})
mode = str(spec.params.get("mode", "strict"))
if mode not in self._MODES:
return self._result(
spec, case, 0.0, Verdict.ERROR,
f"unknown trajectory mode {mode!r} (use {sorted(self._MODES)})", {},
)
actual = list(invocation.tool_trajectory)
score, passed = self._match(mode, expected, actual)
verdict = Verdict.PASS if passed else Verdict.FAIL
reason = (
f"trajectory {mode}: matched"
if passed
else f"trajectory {mode}: expected {expected} got {actual}"
)
details = {"mode": mode, "expected": ", ".join(expected), "actual": ", ".join(actual)}
return self._result(spec, case, score, verdict, reason, details)
Bridge to TruLens feedback functions.
Lazy-imports trulens-eval. Choose which feedback to run via
spec.params['feedback']: groundedness | relevance | qa_relevance.
Source code in apogee_ai_eval/infrastructure/bridges/trulens_adapter.py
async
¶
Python
evaluate(*, case: EvalCase, invocation: CaseInvocation, spec: MetricSpec) -> MetricResult
Source code in apogee_ai_eval/infrastructure/bridges/trulens_adapter.py
Python
async def evaluate(
self,
*,
case: EvalCase,
invocation: CaseInvocation,
spec: MetricSpec,
) -> MetricResult:
try:
from trulens_eval.feedback.provider.openai import OpenAI as TLOpenAI # type: ignore
except ImportError as exc: # pragma: no cover
raise MetricException(
"trulens-eval feedback provider unavailable", metric=spec.display_name
) from exc
feedback_kind = spec.params.get("feedback", "qa_relevance")
provider = TLOpenAI()
try:
if feedback_kind == "groundedness":
value = provider.groundedness_measure_with_cot_reasons(
" ".join(case.context), invocation.output
)
elif feedback_kind == "relevance":
value = provider.relevance_with_cot_reasons(case.input, invocation.output)
else:
value = provider.qs_relevance_with_cot_reasons(case.input, invocation.output)
except Exception as exc: # noqa: BLE001
raise MetricException(f"TruLens error: {exc}", metric=spec.display_name) from exc
score = float(value[0]) if isinstance(value, tuple) else float(value)
score = max(0.0, min(1.0, score))
threshold = float(spec.params.get("threshold", 0.7))
return MetricResult(
metric_name=spec.display_name,
metric_kind=self.kind,
case_id=case.id,
score=score,
verdict=Verdict.PASS if score >= threshold else Verdict.FAIL,
weight=spec.weight,
reason=f"trulens.{feedback_kind}={score:.2f}",
)
Return a dict {metric_kind_string: IMetric} ready for the runner.
Judge-backed metrics are only included when judge is supplied.
Source code in apogee_ai_eval/infrastructure/metrics/registry.py
Python
def default_metrics(*, judge: IJudge | None = None) -> dict[str, IMetric]:
"""Return a dict ``{metric_kind_string: IMetric}`` ready for the runner.
Judge-backed metrics are only included when ``judge`` is supplied.
"""
metrics: dict[str, IMetric] = {
ContainsMetric.kind: ContainsMetric(),
EqualsMetric.kind: EqualsMetric(),
RegexMetric.kind: RegexMetric(),
JsonMatchMetric.kind: JsonMatchMetric(),
LatencyMetric.kind: LatencyMetric(),
CostMetric.kind: CostMetric(),
LengthMetric.kind: LengthMetric(),
TrajectoryMetric.kind: TrajectoryMetric(),
}
if judge is not None:
metrics[JudgeFaithfulnessMetric.kind] = JudgeFaithfulnessMetric(judge)
metrics[JudgeRelevanceMetric.kind] = JudgeRelevanceMetric(judge)
metrics[JudgeToxicityMetric.kind] = JudgeToxicityMetric(judge)
metrics[JudgeCustomMetric.kind] = JudgeCustomMetric(judge)
return metrics