跳转至

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

CompareRunsDTO

Bases: BaseModel

model_config class-attribute instance-attribute

Python
model_config = ConfigDict(extra='forbid')

suite instance-attribute

Python
suite: str

current_run_id instance-attribute

Python
current_run_id: str

baseline_label class-attribute instance-attribute

Python
baseline_label: str = 'main'

GateDTO

Bases: BaseModel

model_config class-attribute instance-attribute

Python
model_config = ConfigDict(extra='forbid')

suite instance-attribute

Python
suite: str

current_run_id instance-attribute

Python
current_run_id: str

baseline_label class-attribute instance-attribute

Python
baseline_label: str = 'main'

min_score class-attribute instance-attribute

Python
min_score: float | None = None

max_drop class-attribute instance-attribute

Python
max_drop: float | None = None

require_baseline class-attribute instance-attribute

Python
require_baseline: bool = True

GateResultDTO

Bases: BaseModel

model_config class-attribute instance-attribute

Python
model_config = ConfigDict(extra='forbid')

suite instance-attribute

Python
suite: str

passed instance-attribute

Python
passed: bool

aggregate_score instance-attribute

Python
aggregate_score: float

baseline_score instance-attribute

Python
baseline_score: float | None

delta instance-attribute

Python
delta: float

reasons class-attribute instance-attribute

Python
reasons: list[str] = Field(default_factory=list)

RunSuiteDTO

Bases: BaseModel

model_config class-attribute instance-attribute

Python
model_config = ConfigDict(extra='forbid')

suite instance-attribute

Python
suite: str

agent class-attribute instance-attribute

Python
agent: str | None = None

metadata class-attribute instance-attribute

Python
metadata: dict[str, str] = Field(default_factory=dict)

case_ids class-attribute instance-attribute

Python
case_ids: list[str] | None = None

Subset of case ids to run; None runs every case.

SetBaselineDTO

Bases: BaseModel

model_config class-attribute instance-attribute

Python
model_config = ConfigDict(extra='forbid')

suite instance-attribute

Python
suite: str

run_id instance-attribute

Python
run_id: str

label class-attribute instance-attribute

Python
label: str = 'main'

Application · Use cases

CompareRunsUseCase

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
Python
def __init__(
    self,
    run_repository: IRunRepository,
    baseline_repository: IBaselineRepository,
) -> None:
    self._runs = run_repository
    self._baselines = baseline_repository

REGRESSION_EPS class-attribute instance-attribute

Python
REGRESSION_EPS = 0.005

Score deltas within ±0.5% are considered noise.

execute 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,
    )

GetRunUseCase

Python
GetRunUseCase(run_repository: IRunRepository)
Source code in apogee_ai_eval/application/use_cases/get_run_use_case.py
Python
def __init__(self, run_repository: IRunRepository) -> None:
    self._runs = run_repository

execute async

Python
execute(run_id: str) -> EvalRun
Source code in apogee_ai_eval/application/use_cases/get_run_use_case.py
Python
async def execute(self, run_id: str) -> EvalRun:
    run = await self._runs.find(run_id)
    if run is None:
        raise RunNotFoundException(run_id)
    return run

ListRunsUseCase

Python
ListRunsUseCase(run_repository: IRunRepository)
Source code in apogee_ai_eval/application/use_cases/list_runs_use_case.py
Python
def __init__(self, run_repository: IRunRepository) -> None:
    self._runs = run_repository

execute async

Python
execute(*, suite: str | None = None, limit: int | None = None) -> list[EvalRun]
Source code in apogee_ai_eval/application/use_cases/list_runs_use_case.py
Python
async def execute(
    self,
    *,
    suite: str | None = None,
    limit: int | None = None,
) -> list[EvalRun]:
    return await self._runs.list(suite=suite, limit=limit)

RegressionGateUseCase

Python
RegressionGateUseCase(suite_repository: ISuiteRepository, run_repository: IRunRepository, baseline_repository: IBaselineRepository)

Decides whether current_run_id is acceptable vs the baseline.

Combines
  • explicit min_score and max_drop from 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
Python
def __init__(
    self,
    suite_repository: ISuiteRepository,
    run_repository: IRunRepository,
    baseline_repository: IBaselineRepository,
) -> None:
    self._suites = suite_repository
    self._runs = run_repository
    self._baselines = baseline_repository

execute 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,
    )

RunSuiteUseCase

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
Python
def __init__(
    self,
    suite_repository: ISuiteRepository,
    run_repository: IRunRepository,
    invoker: IAgentInvoker,
    metrics: Mapping[str, IMetric],
) -> None:
    self._suites = suite_repository
    self._runs = run_repository
    self._invoker = invoker
    self._metrics = dict(metrics)

execute 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

SetBaselineUseCase

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
Python
def __init__(
    self,
    run_repository: IRunRepository,
    baseline_repository: IBaselineRepository,
) -> None:
    self._runs = run_repository
    self._baselines = baseline_repository

execute 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

Baseline 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.

suite instance-attribute

Python
suite: str

label class-attribute instance-attribute

Python
label: str = 'main'

run_id class-attribute instance-attribute

Python
run_id: str = ''

aggregate_score class-attribute instance-attribute

Python
aggregate_score: float = 0.0

case_scores class-attribute instance-attribute

Python
case_scores: dict[str, float] = field(default_factory=dict)

Map case_id → aggregate score for that case (mean of metrics).

metric_scores class-attribute instance-attribute

Python
metric_scores: dict[str, dict[str, float]] = field(default_factory=dict)

case_id → metric_kind → score.

created_at class-attribute instance-attribute

Python
created_at: datetime = field(default_factory=lambda: now(utc))

CaseComparison dataclass

Python
CaseComparison(case_id: str, metric_comparisons: tuple[MetricComparison, ...], aggregate_baseline: float | None, aggregate_current: float | None, aggregate_delta: float, status: RegressionStatus)

case_id instance-attribute

Python
case_id: str

metric_comparisons instance-attribute

Python
metric_comparisons: tuple[MetricComparison, ...]

aggregate_baseline instance-attribute

Python
aggregate_baseline: float | None

aggregate_current instance-attribute

Python
aggregate_current: float | None

aggregate_delta instance-attribute

Python
aggregate_delta: float

status instance-attribute

Python
status: RegressionStatus

CaseInvocation 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.

case_id instance-attribute

Python
case_id: str

output instance-attribute

Python
output: str

latency_ms class-attribute instance-attribute

Python
latency_ms: float = 0.0

cost_usd class-attribute instance-attribute

Python
cost_usd: float = 0.0

input_tokens class-attribute instance-attribute

Python
input_tokens: int = 0

output_tokens class-attribute instance-attribute

Python
output_tokens: int = 0

error class-attribute instance-attribute

Python
error: str | None = None

tool_trajectory class-attribute instance-attribute

Python
tool_trajectory: tuple[str, ...] = field(default_factory=tuple)

Actual tool-call sequence the agent took (tool names, in order).

CaseResult dataclass

Python
CaseResult(case_id: str, invocation: CaseInvocation, metrics: tuple[MetricResult, ...] = tuple())

case_id instance-attribute

Python
case_id: str

invocation instance-attribute

Python
invocation: CaseInvocation

metrics class-attribute instance-attribute

Python
metrics: tuple[MetricResult, ...] = field(default_factory=tuple)

aggregate_score property

Python
aggregate_score: float

verdict property

Python
verdict: Verdict

Comparison 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)

suite instance-attribute

Python
suite: str

baseline_label instance-attribute

Python
baseline_label: str | None

current_run_id instance-attribute

Python
current_run_id: str

cases class-attribute instance-attribute

Python
cases: tuple[CaseComparison, ...] = field(default_factory=tuple)

aggregate_baseline class-attribute instance-attribute

Python
aggregate_baseline: float | None = None

aggregate_current class-attribute instance-attribute

Python
aggregate_current: float | None = None

aggregate_delta class-attribute instance-attribute

Python
aggregate_delta: float = 0.0

status class-attribute instance-attribute

Python
status: RegressionStatus = UNCHANGED

EvalCase 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.

id instance-attribute

Python
id: str

input instance-attribute

Python
input: str

expected class-attribute instance-attribute

Python
expected: str | None = None

expected_substrings class-attribute instance-attribute

Python
expected_substrings: tuple[str, ...] = field(default_factory=tuple)

expected_json class-attribute instance-attribute

Python
expected_json: dict[str, Any] | None = None

context class-attribute instance-attribute

Python
context: tuple[str, ...] = field(default_factory=tuple)

Reference passages used by faithfulness/RAG metrics.

metadata class-attribute instance-attribute

Python
metadata: dict[str, str] = field(default_factory=dict)

tags class-attribute instance-attribute

Python
tags: tuple[str, ...] = field(default_factory=tuple)

expected_tools class-attribute instance-attribute

Python
expected_tools: tuple[str, ...] = field(default_factory=tuple)

Expected tool-call trajectory (tool names, in order) — used by the trajectory metric.

EvalRun 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.

id class-attribute instance-attribute

Python
id: str = field(default_factory=lambda: hex)

suite class-attribute instance-attribute

Python
suite: str = ''

started_at class-attribute instance-attribute

Python
started_at: datetime = field(default_factory=lambda: now(utc))

finished_at class-attribute instance-attribute

Python
finished_at: datetime | None = None

cases class-attribute instance-attribute

Python
cases: tuple[CaseResult, ...] = field(default_factory=tuple)

metadata class-attribute instance-attribute

Python
metadata: dict[str, str] = field(default_factory=dict)

aggregate_score property

Python
aggregate_score: float

verdict property

Python
verdict: Verdict

passed property

Python
passed: int

failed property

Python
failed: int

errors property

Python
errors: int

total property

Python
total: int

EvalSuite 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.

name instance-attribute

Python
name: str

agent class-attribute instance-attribute

Python
agent: str | None = None

Logical id of the system-under-test (a slug your invoker resolves).

description class-attribute instance-attribute

Python
description: str | None = None

cases class-attribute instance-attribute

Python
cases: tuple[EvalCase, ...] = field(default_factory=tuple)

metrics class-attribute instance-attribute

Python
metrics: tuple[MetricSpec, ...] = field(default_factory=tuple)

gate class-attribute instance-attribute

Python
gate: RegressionGate | None = None

tags class-attribute instance-attribute

Python
tags: tuple[str, ...] = field(default_factory=tuple)

metadata class-attribute instance-attribute

Python
metadata: dict[str, str] = field(default_factory=dict)

get_case

Python
get_case(case_id: str) -> EvalCase
Source code in apogee_ai_eval/domain/entities/eval_suite.py
Python
def get_case(self, case_id: str) -> EvalCase:
    for c in self.cases:
        if c.id == case_id:
            return c
    raise KeyError(f"Case {case_id!r} not found in suite {self.name!r}")

metrics_for_case

Python
metrics_for_case(case_id: str) -> tuple[MetricSpec, ...]
Source code in apogee_ai_eval/domain/entities/eval_suite.py
Python
def metrics_for_case(self, case_id: str) -> tuple[MetricSpec, ...]:
    return tuple(m for m in self.metrics if m.case_id is None or m.case_id == case_id)

JudgeVerdict dataclass

Python
JudgeVerdict(score: float, reason: str = '', judge: str = 'rule-based', metadata: dict[str, str] = dict())

Decision returned by an LLM-as-judge.

score instance-attribute

Python
score: float

reason class-attribute instance-attribute

Python
reason: str = ''

judge class-attribute instance-attribute

Python
judge: str = 'rule-based'

metadata class-attribute instance-attribute

Python
metadata: dict[str, str] = field(default_factory=dict)

MetricComparison dataclass

Python
MetricComparison(metric_kind: str, baseline_score: float | None, current_score: float | None, delta: float, status: RegressionStatus)

metric_kind instance-attribute

Python
metric_kind: str

baseline_score instance-attribute

Python
baseline_score: float | None

current_score instance-attribute

Python
current_score: float | None

delta instance-attribute

Python
delta: float

status instance-attribute

Python
status: RegressionStatus

MetricResult 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.

metric_name instance-attribute

Python
metric_name: str

metric_kind instance-attribute

Python
metric_kind: str

case_id instance-attribute

Python
case_id: str

score instance-attribute

Python
score: float

verdict instance-attribute

Python
verdict: Verdict

weight class-attribute instance-attribute

Python
weight: float = 1.0

reason class-attribute instance-attribute

Python
reason: str | None = None

details class-attribute instance-attribute

Python
details: dict[str, str] = field(default_factory=dict)

MetricSpec 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.).

kind instance-attribute

Python
kind: MetricKind

case_id class-attribute instance-attribute

Python
case_id: str | None = None

name class-attribute instance-attribute

Python
name: str | None = None

weight class-attribute instance-attribute

Python
weight: float = 1.0

params class-attribute instance-attribute

Python
params: dict[str, Any] = field(default_factory=dict)

display_name property

Python
display_name: str

RegressionGate 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.

min_score class-attribute instance-attribute

Python
min_score: float | None = None

max_drop class-attribute instance-attribute

Python
max_drop: float | None = None

severity class-attribute instance-attribute

Python
severity: Severity = CRITICAL

RegressionStatus

Bases: str, Enum

IMPROVED class-attribute instance-attribute

Python
IMPROVED = 'improved'

UNCHANGED class-attribute instance-attribute

Python
UNCHANGED = 'unchanged'

REGRESSED class-attribute instance-attribute

Python
REGRESSED = 'regressed'

NEW class-attribute instance-attribute

Python
NEW = 'new'

Suite/case that didn't exist in the baseline.

REMOVED class-attribute instance-attribute

Python
REMOVED = 'removed'

Score dataclass

Python
Score(value: float, samples: int = 1)

Numeric score in [0, 1] with optional sample size for statistics.

value instance-attribute

Python
value: float

samples class-attribute instance-attribute

Python
samples: int = 1

Severity

Bases: str, Enum

INFO class-attribute instance-attribute

Python
INFO = 'info'

WARNING class-attribute instance-attribute

Python
WARNING = 'warning'

CRITICAL class-attribute instance-attribute

Python
CRITICAL = 'critical'

Verdict

Bases: str, Enum

PASS class-attribute instance-attribute

Python
PASS = 'pass'

FAIL class-attribute instance-attribute

Python
FAIL = 'fail'

SKIP class-attribute instance-attribute

Python
SKIP = 'skip'

ERROR class-attribute instance-attribute

Python
ERROR = 'error'

Domain · Enums

MetricKind

Bases: str, Enum

Identifier for built-in metric implementations.

CONTAINS class-attribute instance-attribute

Python
CONTAINS = 'contains'

EQUALS class-attribute instance-attribute

Python
EQUALS = 'equals'

JSON_MATCH class-attribute instance-attribute

Python
JSON_MATCH = 'json_match'

LATENCY class-attribute instance-attribute

Python
LATENCY = 'latency'

COST class-attribute instance-attribute

Python
COST = 'cost'

LENGTH class-attribute instance-attribute

Python
LENGTH = 'length'

REGEX class-attribute instance-attribute

Python
REGEX = 'regex'

TRAJECTORY class-attribute instance-attribute

Python
TRAJECTORY = 'trajectory'

JUDGE_FAITHFULNESS class-attribute instance-attribute

Python
JUDGE_FAITHFULNESS = 'judge_faithfulness'

JUDGE_RELEVANCE class-attribute instance-attribute

Python
JUDGE_RELEVANCE = 'judge_relevance'

JUDGE_TOXICITY class-attribute instance-attribute

Python
JUDGE_TOXICITY = 'judge_toxicity'

JUDGE_CUSTOM class-attribute instance-attribute

Python
JUDGE_CUSTOM = 'judge_custom'

RAGAS class-attribute instance-attribute

Python
RAGAS = 'ragas'

TRULENS class-attribute instance-attribute

Python
TRULENS = 'trulens'

DEEPEVAL class-attribute instance-attribute

Python
DEEPEVAL = 'deepeval'

Domain · Exceptions

BaselineNotFoundException

Python
BaselineNotFoundException(suite: str, label: str = 'main')

Bases: EvalError

Source code in apogee_ai_eval/domain/exceptions/eval_exceptions.py
Python
def __init__(self, suite: str, label: str = "main") -> None:
    super().__init__(f"Baseline ({label!r}) not found for suite {suite!r}")
    self.suite = suite
    self.label = label

suite instance-attribute

Python
suite = suite

label instance-attribute

Python
label = label

CaseNotFoundException

Python
CaseNotFoundException(suite: str, case_id: str)

Bases: EvalError

Source code in apogee_ai_eval/domain/exceptions/eval_exceptions.py
Python
def __init__(self, suite: str, case_id: str) -> None:
    super().__init__(f"Case {case_id!r} not found in suite {suite!r}")
    self.suite = suite
    self.case_id = case_id

suite instance-attribute

Python
suite = suite

case_id instance-attribute

Python
case_id = case_id

DatasetParseException

Bases: EvalError

EvalError

Bases: Exception

Base for all apogee-ai-eval errors.

InvokerException

Bases: EvalError

JudgeException

Python
JudgeException(message: str, judge: str | None = None)

Bases: EvalError

Source code in apogee_ai_eval/domain/exceptions/eval_exceptions.py
Python
def __init__(self, message: str, judge: str | None = None) -> None:
    super().__init__(message)
    self.judge = judge

judge instance-attribute

Python
judge = judge

MetricException

Python
MetricException(message: str, metric: str | None = None)

Bases: EvalError

Source code in apogee_ai_eval/domain/exceptions/eval_exceptions.py
Python
def __init__(self, message: str, metric: str | None = None) -> None:
    super().__init__(message)
    self.metric = metric

metric instance-attribute

Python
metric = metric

RegressionGateException

Python
RegressionGateException(suite: str, reason: str)

Bases: EvalError

Source code in apogee_ai_eval/domain/exceptions/eval_exceptions.py
Python
def __init__(self, suite: str, reason: str) -> None:
    super().__init__(f"Regression gate failed for suite {suite!r}: {reason}")
    self.suite = suite
    self.reason = reason

suite instance-attribute

Python
suite = suite

reason instance-attribute

Python
reason = reason

RunNotFoundException

Python
RunNotFoundException(run_id: str)

Bases: EvalError

Source code in apogee_ai_eval/domain/exceptions/eval_exceptions.py
Python
def __init__(self, run_id: str) -> None:
    super().__init__(f"Run {run_id!r} not found")
    self.run_id = run_id

run_id instance-attribute

Python
run_id = run_id

SuiteNotFoundException

Python
SuiteNotFoundException(name: str)

Bases: EvalError

Source code in apogee_ai_eval/domain/exceptions/eval_exceptions.py
Python
def __init__(self, name: str) -> None:
    super().__init__(f"Suite {name!r} not found")
    self.name = name

name instance-attribute

Python
name = name

Domain · Protocols (ports)

IAgentInvoker

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.

name instance-attribute

Python
name: str

invoke async

Python
invoke(*, agent: str | None, case: EvalCase) -> CaseInvocation
Source code in apogee_ai_eval/domain/services/i_agent_invoker.py
Python
async def invoke(self, *, agent: str | None, case: EvalCase) -> CaseInvocation:
    ...

IBaselineRepository

Bases: Protocol

get async

Python
get(suite: str, *, label: str = 'main') -> Baseline
Source code in apogee_ai_eval/domain/repositories/i_baseline_repository.py
Python
async def get(self, suite: str, *, label: str = "main") -> Baseline:
    ...

find async

Python
find(suite: str, *, label: str = 'main') -> Baseline | None
Source code in apogee_ai_eval/domain/repositories/i_baseline_repository.py
Python
async def find(self, suite: str, *, label: str = "main") -> Baseline | None:
    ...

save async

Python
save(baseline: Baseline) -> Baseline
Source code in apogee_ai_eval/domain/repositories/i_baseline_repository.py
Python
async def save(self, baseline: Baseline) -> Baseline:
    ...

list async

Python
list(suite: str) -> list[Baseline]
Source code in apogee_ai_eval/domain/repositories/i_baseline_repository.py
Python
async def list(self, suite: str) -> list[Baseline]:
    ...

delete async

Python
delete(suite: str, *, label: str = 'main') -> None
Source code in apogee_ai_eval/domain/repositories/i_baseline_repository.py
Python
async def delete(self, suite: str, *, label: str = "main") -> None:
    ...

IDatasetRepository

Bases: Protocol

Storage for golden datasets — collections of :class:EvalCase.

Datasets are versioned by name (a slug). Implementations may store them as JSONL/CSV/HF datasets/etc.

load async

Python
load(name: str) -> list[EvalCase]
Source code in apogee_ai_eval/domain/repositories/i_dataset_repository.py
Python
async def load(self, name: str) -> list[EvalCase]:
    ...

save async

Python
save(name: str, cases: list[EvalCase]) -> None
Source code in apogee_ai_eval/domain/repositories/i_dataset_repository.py
Python
async def save(self, name: str, cases: list[EvalCase]) -> None:
    ...

list async

Python
list() -> list[str]
Source code in apogee_ai_eval/domain/repositories/i_dataset_repository.py
Python
async def list(self) -> list[str]:
    ...

IJudge

Bases: Protocol

LLM-as-judge: scores an output (0..1) given a case.

name instance-attribute

Python
name: str

judge async

Python
judge(*, case: EvalCase, output: str, criterion: str) -> JudgeVerdict
Source code in apogee_ai_eval/domain/services/i_judge.py
Python
async def judge(
    self,
    *,
    case: EvalCase,
    output: str,
    criterion: str,
) -> JudgeVerdict:
    ...

IMetric

Bases: Protocol

Computes a single metric for one case + invocation pair.

kind instance-attribute

Python
kind: str

Identifier matching the MetricKind enum value this metric handles.

evaluate async

Python
evaluate(*, case: EvalCase, invocation: CaseInvocation, spec: MetricSpec) -> MetricResult
Source code in apogee_ai_eval/domain/services/i_metric.py
Python
async def evaluate(
    self,
    *,
    case: EvalCase,
    invocation: CaseInvocation,
    spec: MetricSpec,
) -> MetricResult:
    ...

IReporter

Bases: Protocol

name instance-attribute

Python
name: str

extension instance-attribute

Python
extension: str

render_run

Python
render_run(suite: EvalSuite, run: EvalRun) -> str
Source code in apogee_ai_eval/domain/services/i_reporter.py
Python
def render_run(self, suite: EvalSuite, run: EvalRun) -> str:
    ...

render_comparison

Python
render_comparison(comparison: Comparison) -> str
Source code in apogee_ai_eval/domain/services/i_reporter.py
Python
def render_comparison(self, comparison: Comparison) -> str:
    ...

IRunRepository

Bases: Protocol

get async

Python
get(run_id: str) -> EvalRun
Source code in apogee_ai_eval/domain/repositories/i_run_repository.py
Python
async def get(self, run_id: str) -> EvalRun:
    ...

find async

Python
find(run_id: str) -> EvalRun | None
Source code in apogee_ai_eval/domain/repositories/i_run_repository.py
Python
async def find(self, run_id: str) -> EvalRun | None:
    ...

save async

Python
save(run: EvalRun) -> EvalRun
Source code in apogee_ai_eval/domain/repositories/i_run_repository.py
Python
async def save(self, run: EvalRun) -> EvalRun:
    ...

list async

Python
list(*, suite: str | None = None, limit: int | None = None) -> list[EvalRun]
Source code in apogee_ai_eval/domain/repositories/i_run_repository.py
Python
async def list(self, *, suite: str | None = None, limit: int | None = None) -> list[EvalRun]:
    ...

latest async

Python
latest(suite: str) -> EvalRun | None
Source code in apogee_ai_eval/domain/repositories/i_run_repository.py
Python
async def latest(self, suite: str) -> EvalRun | None:
    ...

ISuiteRepository

Bases: Protocol

get async

Python
get(name: str) -> EvalSuite
Source code in apogee_ai_eval/domain/repositories/i_suite_repository.py
Python
async def get(self, name: str) -> EvalSuite:
    ...

find async

Python
find(name: str) -> EvalSuite | None
Source code in apogee_ai_eval/domain/repositories/i_suite_repository.py
Python
async def find(self, name: str) -> EvalSuite | None:
    ...

list async

Python
list() -> list[EvalSuite]
Source code in apogee_ai_eval/domain/repositories/i_suite_repository.py
Python
async def list(self) -> list[EvalSuite]:
    ...

save async

Python
save(suite: EvalSuite) -> EvalSuite
Source code in apogee_ai_eval/domain/repositories/i_suite_repository.py
Python
async def save(self, suite: EvalSuite) -> EvalSuite:
    ...

exists async

Python
exists(name: str) -> bool
Source code in apogee_ai_eval/domain/repositories/i_suite_repository.py
Python
async def exists(self, name: str) -> bool:
    ...

Infrastructure

AnthropicJudge

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

name class-attribute instance-attribute

Python
name = 'anthropic'

judge 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),
        },
    )

ContainsMetric

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

kind class-attribute instance-attribute

Python
kind = CONTAINS.value

evaluate 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)},
    )

CostMetric

Pass when invocation.cost_usd <= threshold_usd.

kind class-attribute instance-attribute

Python
kind = COST.value

evaluate 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}",
    )

DeepEvalAdapter

Python
DeepEvalAdapter()

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
Python
def __init__(self) -> None:
    try:
        import deepeval  # type: ignore  # noqa: F401
    except ImportError as exc:
        raise ImportError(
            "DeepEvalAdapter requires `deepeval`. "
            "Install with: pip install 'apogee-ai-eval[deepeval]'"
        ) from exc

kind class-attribute instance-attribute

Python
kind = DEEPEVAL.value

evaluate 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 "")},
    )

EchoAgentInvoker

Toy invoker — echoes the input. Used by tests and demos.

name class-attribute instance-attribute

Python
name = 'echo'

invoke 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:
    return CaseInvocation(case_id=case.id, output=case.input, latency_ms=1.0)

EqualsMetric

kind class-attribute instance-attribute

Python
kind = EQUALS.value

evaluate 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}",
    )

FunctionAgentInvoker

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
Python
def __init__(
    self,
    fn: Callable[[EvalCase], Any],
    *,
    invoker_name: str | None = None,
) -> None:
    self._fn = fn
    if invoker_name:
        self.name = invoker_name

name class-attribute instance-attribute

Python
name = 'function'

invoke 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,
    )

InMemoryBaselineRepository

Python
InMemoryBaselineRepository()
Source code in apogee_ai_eval/infrastructure/datasets/json_baseline_repository.py
Python
def __init__(self) -> None:
    self._store: dict[tuple[str, str], Baseline] = {}

name class-attribute instance-attribute

Python
name = 'memory'

get async

Python
get(suite: str, *, label: str = 'main') -> Baseline
Source code in apogee_ai_eval/infrastructure/datasets/json_baseline_repository.py
Python
async def get(self, suite: str, *, label: str = "main") -> Baseline:
    key = (suite, label)
    if key not in self._store:
        raise BaselineNotFoundException(suite, label)
    return self._store[key]

find async

Python
find(suite: str, *, label: str = 'main') -> Baseline | None
Source code in apogee_ai_eval/infrastructure/datasets/json_baseline_repository.py
Python
async def find(self, suite: str, *, label: str = "main") -> Baseline | None:
    return self._store.get((suite, label))

save async

Python
save(baseline: Baseline) -> Baseline
Source code in apogee_ai_eval/infrastructure/datasets/json_baseline_repository.py
Python
async def save(self, baseline: Baseline) -> Baseline:
    self._store[(baseline.suite, baseline.label)] = baseline
    return baseline

list async

Python
list(suite: str) -> list[Baseline]
Source code in apogee_ai_eval/infrastructure/datasets/json_baseline_repository.py
Python
async def list(self, suite: str) -> list[Baseline]:
    return [b for (s, _), b in self._store.items() if s == suite]

delete async

Python
delete(suite: str, *, label: str = 'main') -> None
Source code in apogee_ai_eval/infrastructure/datasets/json_baseline_repository.py
Python
async def delete(self, suite: str, *, label: str = "main") -> None:
    self._store.pop((suite, label), None)

InMemoryDatasetRepository

Python
InMemoryDatasetRepository()
Source code in apogee_ai_eval/infrastructure/datasets/jsonl_dataset_repository.py
Python
def __init__(self) -> None:
    self._store: dict[str, list[EvalCase]] = {}

name class-attribute instance-attribute

Python
name = 'memory'

load async

Python
load(name: str) -> list[EvalCase]
Source code in apogee_ai_eval/infrastructure/datasets/jsonl_dataset_repository.py
Python
async def load(self, name: str) -> list[EvalCase]:
    return list(self._store.get(name, []))

save async

Python
save(name: str, cases: list[EvalCase]) -> None
Source code in apogee_ai_eval/infrastructure/datasets/jsonl_dataset_repository.py
Python
async def save(self, name: str, cases: list[EvalCase]) -> None:
    self._store[name] = list(cases)

list async

Python
list() -> list[str]
Source code in apogee_ai_eval/infrastructure/datasets/jsonl_dataset_repository.py
Python
async def list(self) -> list[str]:
    return sorted(self._store.keys())

InMemoryRunRepository

Python
InMemoryRunRepository()
Source code in apogee_ai_eval/infrastructure/datasets/json_run_repository.py
Python
def __init__(self) -> None:
    self._store: dict[str, EvalRun] = {}

name class-attribute instance-attribute

Python
name = 'memory'

get async

Python
get(run_id: str) -> EvalRun
Source code in apogee_ai_eval/infrastructure/datasets/json_run_repository.py
Python
async def get(self, run_id: str) -> EvalRun:
    if run_id not in self._store:
        raise RunNotFoundException(run_id)
    return self._store[run_id]

find async

Python
find(run_id: str) -> EvalRun | None
Source code in apogee_ai_eval/infrastructure/datasets/json_run_repository.py
Python
async def find(self, run_id: str) -> EvalRun | None:
    return self._store.get(run_id)

save async

Python
save(run: EvalRun) -> EvalRun
Source code in apogee_ai_eval/infrastructure/datasets/json_run_repository.py
Python
async def save(self, run: EvalRun) -> EvalRun:
    self._store[run.id] = run
    return run

list async

Python
list(*, suite: str | None = None, limit: int | None = None) -> list[EvalRun]
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

latest async

Python
latest(suite: str) -> EvalRun | None
Source code in apogee_ai_eval/infrastructure/datasets/json_run_repository.py
Python
async def latest(self, suite: str) -> EvalRun | None:
    for run in sorted(self._store.values(), key=lambda r: r.started_at, reverse=True):
        if run.suite == suite:
            return run
    return None

InMemorySuiteRepository

Python
InMemorySuiteRepository()
Source code in apogee_ai_eval/infrastructure/datasets/in_memory_suite_repository.py
Python
def __init__(self) -> None:
    self._store: dict[str, EvalSuite] = {}

name class-attribute instance-attribute

Python
name = 'memory'

get async

Python
get(name: str) -> EvalSuite
Source code in apogee_ai_eval/infrastructure/datasets/in_memory_suite_repository.py
Python
async def get(self, name: str) -> EvalSuite:
    if name not in self._store:
        raise SuiteNotFoundException(name)
    return self._store[name]

find async

Python
find(name: str) -> EvalSuite | None
Source code in apogee_ai_eval/infrastructure/datasets/in_memory_suite_repository.py
Python
async def find(self, name: str) -> EvalSuite | None:
    return self._store.get(name)

list async

Python
list() -> list[EvalSuite]
Source code in apogee_ai_eval/infrastructure/datasets/in_memory_suite_repository.py
Python
async def list(self) -> list[EvalSuite]:
    return sorted(self._store.values(), key=lambda s: s.name)

save async

Python
save(suite: EvalSuite) -> EvalSuite
Source code in apogee_ai_eval/infrastructure/datasets/in_memory_suite_repository.py
Python
async def save(self, suite: EvalSuite) -> EvalSuite:
    self._store[suite.name] = deepcopy(suite)
    return self._store[suite.name]

exists async

Python
exists(name: str) -> bool
Source code in apogee_ai_eval/infrastructure/datasets/in_memory_suite_repository.py
Python
async def exists(self, name: str) -> bool:
    return name in self._store

JsonBaselineRepository

Python
JsonBaselineRepository(root: str | Path)

One JSON per <root>/<suite>/<label>.json.

Source code in apogee_ai_eval/infrastructure/datasets/json_baseline_repository.py
Python
def __init__(self, root: str | Path) -> None:
    self._root = Path(root)

name class-attribute instance-attribute

Python
name = 'json'

get async

Python
get(suite: str, *, label: str = 'main') -> Baseline
Source code in apogee_ai_eval/infrastructure/datasets/json_baseline_repository.py
Python
async def get(self, suite: str, *, label: str = "main") -> Baseline:
    baseline = await self.find(suite, label=label)
    if baseline is None:
        raise BaselineNotFoundException(suite, label)
    return baseline

find async

Python
find(suite: str, *, label: str = 'main') -> Baseline | None
Source code in apogee_ai_eval/infrastructure/datasets/json_baseline_repository.py
Python
async def find(self, suite: str, *, label: str = "main") -> Baseline | None:
    return await asyncio.to_thread(self._read_one, suite, label)

save async

Python
save(baseline: Baseline) -> Baseline
Source code in apogee_ai_eval/infrastructure/datasets/json_baseline_repository.py
Python
async def save(self, baseline: Baseline) -> Baseline:
    await asyncio.to_thread(self._write_one, baseline)
    return baseline

list async

Python
list(suite: str) -> list[Baseline]
Source code in apogee_ai_eval/infrastructure/datasets/json_baseline_repository.py
Python
async def list(self, suite: str) -> list[Baseline]:
    return await asyncio.to_thread(self._read_all_for_suite, suite)

delete async

Python
delete(suite: str, *, label: str = 'main') -> None
Source code in apogee_ai_eval/infrastructure/datasets/json_baseline_repository.py
Python
async def delete(self, suite: str, *, label: str = "main") -> None:
    await asyncio.to_thread(self._delete_one, suite, label)

JsonMatchMetric

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.

kind class-attribute instance-attribute

Python
kind = JSON_MATCH.value

evaluate 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)}",
    )

JsonReporter

name class-attribute instance-attribute

Python
name = 'json'

extension class-attribute instance-attribute

Python
extension = 'json'

render_run

Python
render_run(suite: EvalSuite, run: EvalRun) -> str
Source code in apogee_ai_eval/infrastructure/reporters/json_reporter.py
Python
def render_run(self, suite: EvalSuite, run: EvalRun) -> str:
    return json.dumps(run_to_dict(run), ensure_ascii=False, indent=2, default=_default)

render_comparison

Python
render_comparison(comparison: Comparison) -> str
Source code in apogee_ai_eval/infrastructure/reporters/json_reporter.py
Python
def render_comparison(self, comparison: Comparison) -> str:
    return json.dumps(comparison, ensure_ascii=False, indent=2, default=_default)

JsonRunRepository

Python
JsonRunRepository(root: str | Path)

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
def __init__(self, root: str | Path) -> None:
    self._root = Path(root)

name class-attribute instance-attribute

Python
name = 'json'

get async

Python
get(run_id: str) -> EvalRun
Source code in apogee_ai_eval/infrastructure/datasets/json_run_repository.py
Python
async def get(self, run_id: str) -> EvalRun:
    run = await self.find(run_id)
    if run is None:
        raise RunNotFoundException(run_id)
    return run

find async

Python
find(run_id: str) -> EvalRun | None
Source code in apogee_ai_eval/infrastructure/datasets/json_run_repository.py
Python
async def find(self, run_id: str) -> EvalRun | None:
    return await asyncio.to_thread(self._read_one, run_id)

save async

Python
save(run: EvalRun) -> EvalRun
Source code in apogee_ai_eval/infrastructure/datasets/json_run_repository.py
Python
async def save(self, run: EvalRun) -> EvalRun:
    await asyncio.to_thread(self._write_one, run)
    return run

list async

Python
list(*, suite: str | None = None, limit: int | None = None) -> list[EvalRun]
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]:
    return await asyncio.to_thread(self._read_index, suite, limit)

latest async

Python
latest(suite: str) -> EvalRun | None
Source code in apogee_ai_eval/infrastructure/datasets/json_run_repository.py
Python
async def latest(self, suite: str) -> EvalRun | None:
    items = await self.list(suite=suite, limit=1)
    return items[0] if items else None

JsonlDatasetRepository

Python
JsonlDatasetRepository(root: str | Path)

Datasets stored as <root>/<name>.jsonl (1 case per line).

Source code in apogee_ai_eval/infrastructure/datasets/jsonl_dataset_repository.py
Python
def __init__(self, root: str | Path) -> None:
    self._root = Path(root)

name class-attribute instance-attribute

Python
name = 'jsonl'

load async

Python
load(name: str) -> list[EvalCase]
Source code in apogee_ai_eval/infrastructure/datasets/jsonl_dataset_repository.py
Python
async def load(self, name: str) -> list[EvalCase]:
    return await asyncio.to_thread(self._read, name)

save async

Python
save(name: str, cases: list[EvalCase]) -> None
Source code in apogee_ai_eval/infrastructure/datasets/jsonl_dataset_repository.py
Python
async def save(self, name: str, cases: list[EvalCase]) -> None:
    await asyncio.to_thread(self._write, name, cases)

list async

Python
list() -> list[str]
Source code in apogee_ai_eval/infrastructure/datasets/jsonl_dataset_repository.py
Python
async def list(self) -> list[str]:
    return await asyncio.to_thread(self._list)

JudgeCustomMetric

Python
JudgeCustomMetric(judge: IJudge, *, threshold: float | None = None)

Bases: _JudgeBackedMetric

Source code in apogee_ai_eval/infrastructure/metrics/judge_metric.py
Python
def __init__(self, judge: IJudge, *, threshold: float | None = None) -> None:
    self._judge = judge
    if threshold is not None:
        self.threshold = threshold

kind class-attribute instance-attribute

Python
kind = JUDGE_CUSTOM.value

criterion class-attribute instance-attribute

Python
criterion = 'Score the output between 0 and 1.'

JudgeFaithfulnessMetric

Python
JudgeFaithfulnessMetric(judge: IJudge, *, threshold: float | None = None)

Bases: _JudgeBackedMetric

Source code in apogee_ai_eval/infrastructure/metrics/judge_metric.py
Python
def __init__(self, judge: IJudge, *, threshold: float | None = None) -> None:
    self._judge = judge
    if threshold is not None:
        self.threshold = threshold

kind class-attribute instance-attribute

Python
kind = JUDGE_FAITHFULNESS.value

criterion class-attribute instance-attribute

Python
criterion = _FAITHFULNESS_CRITERION

JudgeRelevanceMetric

Python
JudgeRelevanceMetric(judge: IJudge, *, threshold: float | None = None)

Bases: _JudgeBackedMetric

Source code in apogee_ai_eval/infrastructure/metrics/judge_metric.py
Python
def __init__(self, judge: IJudge, *, threshold: float | None = None) -> None:
    self._judge = judge
    if threshold is not None:
        self.threshold = threshold

kind class-attribute instance-attribute

Python
kind = JUDGE_RELEVANCE.value

criterion class-attribute instance-attribute

Python
criterion = _RELEVANCE_CRITERION

JudgeToxicityMetric

Python
JudgeToxicityMetric(judge: IJudge, *, threshold: float | None = None)

Bases: _JudgeBackedMetric

Source code in apogee_ai_eval/infrastructure/metrics/judge_metric.py
Python
def __init__(self, judge: IJudge, *, threshold: float | None = None) -> None:
    self._judge = judge
    if threshold is not None:
        self.threshold = threshold

kind class-attribute instance-attribute

Python
kind = JUDGE_TOXICITY.value

criterion class-attribute instance-attribute

Python
criterion = _TOXICITY_CRITERION

JunitReporter

JUnit XML reporter, consumable by GitHub Actions / GitLab CI.

name class-attribute instance-attribute

Python
name = 'junit'

extension class-attribute instance-attribute

Python
extension = 'xml'

render_run

Python
render_run(suite: EvalSuite, run: EvalRun) -> str
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"
    )

render_comparison

Python
render_comparison(comparison: Comparison) -> str
Source code in apogee_ai_eval/infrastructure/reporters/junit_reporter.py
Python
def render_comparison(self, comparison: Comparison) -> str:
    # JUnit is only meaningful for runs; comparisons have a separate report
    return f"<!-- comparison for {comparison.suite} (use markdown/json reporter) -->"

LatencyMetric

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.

kind class-attribute instance-attribute

Python
kind = LATENCY.value

evaluate 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}"},
    )

LengthMetric

Pass when output length sits within [min_chars, max_chars].

kind class-attribute instance-attribute

Python
kind = LENGTH.value

evaluate 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})"
        ),
    )

MarkdownReporter

name class-attribute instance-attribute

Python
name = 'markdown'

extension class-attribute instance-attribute

Python
extension = 'md'

render_run

Python
render_run(suite: EvalSuite, run: EvalRun) -> str
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"

render_comparison

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"

OpenAIJudge

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

name class-attribute instance-attribute

Python
name = 'openai'

judge 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",
        },
    )

RagasAdapter

Python
RagasAdapter()

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
Python
def __init__(self) -> None:
    try:
        import ragas  # type: ignore  # noqa: F401
    except ImportError as exc:
        raise ImportError(
            "RagasAdapter requires `ragas`. "
            "Install with: pip install 'apogee-ai-eval[ragas]'"
        ) from exc

kind class-attribute instance-attribute

Python
kind = RAGAS.value

evaluate 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"},
    )

RegexMetric

kind class-attribute instance-attribute

Python
kind = REGEX.value

evaluate 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",
    )

RuleBasedJudge

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.

name class-attribute instance-attribute

Python
name = 'rule-based'

judge 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")

TrajectoryMetric

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.

kind class-attribute instance-attribute

Python
kind = TRAJECTORY.value

evaluate 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)

TruLensAdapter

Python
TruLensAdapter()

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
Python
def __init__(self) -> None:
    try:
        import trulens_eval  # type: ignore  # noqa: F401
    except ImportError as exc:
        raise ImportError(
            "TruLensAdapter requires `trulens-eval`. "
            "Install with: pip install 'apogee-ai-eval[trulens]'"
        ) from exc

kind class-attribute instance-attribute

Python
kind = TRULENS.value

evaluate 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}",
    )

YamlSuiteRepository

Python
YamlSuiteRepository(root: str | Path)

One <root>/<suite>.yml file per suite.

Source code in apogee_ai_eval/infrastructure/datasets/yaml_suite_repository.py
Python
def __init__(self, root: str | Path) -> None:
    self._root = Path(root)

name class-attribute instance-attribute

Python
name = 'yaml'

get async

Python
get(name: str) -> EvalSuite
Source code in apogee_ai_eval/infrastructure/datasets/yaml_suite_repository.py
Python
async def get(self, name: str) -> EvalSuite:
    suite = await self.find(name)
    if suite is None:
        raise SuiteNotFoundException(name)
    return suite

find async

Python
find(name: str) -> EvalSuite | None
Source code in apogee_ai_eval/infrastructure/datasets/yaml_suite_repository.py
Python
async def find(self, name: str) -> EvalSuite | None:
    return await asyncio.to_thread(self._read_one, name)

list async

Python
list() -> list[EvalSuite]
Source code in apogee_ai_eval/infrastructure/datasets/yaml_suite_repository.py
Python
async def list(self) -> list[EvalSuite]:
    return await asyncio.to_thread(self._read_all)

save async

Python
save(suite: EvalSuite) -> EvalSuite
Source code in apogee_ai_eval/infrastructure/datasets/yaml_suite_repository.py
Python
async def save(self, suite: EvalSuite) -> EvalSuite:
    await asyncio.to_thread(self._write_one, suite)
    return suite

exists async

Python
exists(name: str) -> bool
Source code in apogee_ai_eval/infrastructure/datasets/yaml_suite_repository.py
Python
async def exists(self, name: str) -> bool:
    return await asyncio.to_thread(self._path(name).is_file)

default_metrics

Python
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.

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