Skip to content

Experiments

Versioned agent configurations, runs, baselines, and regression detection. Workflow: Experiments guide.

uaef.experiments.models

Experiment data models for versioning and comparison.

AgentDesign

Bases: BaseModel

Agent design configuration for reproducibility.

Attributes:

Name Type Description
framework str

Agent framework (e.g., "langgraph", "bedrock", "langchain")

model str

LLM model identifier (e.g., "us.anthropic.claude-sonnet-4-6")

tools List[str]

List of tool names available to the agent

system_prompt Optional[str]

System prompt used for the agent

parameters Dict[str, Any]

Additional agent parameters (temperature, max_tokens, etc.)

Source code in uaef/experiments/models.py
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
class AgentDesign(BaseModel):
    """
    Agent design configuration for reproducibility.

    Attributes:
        framework: Agent framework (e.g., "langgraph", "bedrock", "langchain")
        model: LLM model identifier (e.g., "us.anthropic.claude-sonnet-4-6")
        tools: List of tool names available to the agent
        system_prompt: System prompt used for the agent
        parameters: Additional agent parameters (temperature, max_tokens, etc.)
    """

    framework: str = Field(..., description="Agent framework")
    model: str = Field(..., description="LLM model identifier")
    tools: List[str] = Field(
        default_factory=list,
        description="List of tool names available to the agent"
    )
    system_prompt: Optional[str] = Field(
        None,
        description="System prompt used for the agent"
    )
    parameters: Dict[str, Any] = Field(
        default_factory=dict,
        description="Additional agent parameters"
    )

    class Config:
        """Pydantic configuration."""
        json_schema_extra = {
            "example": {
                "framework": "langgraph",
                "model": "us.anthropic.claude-sonnet-4-6",
                "tools": ["search", "calculator", "database_query"],
                "system_prompt": "You are a helpful assistant...",
                "parameters": {
                    "temperature": 0.7,
                    "max_tokens": 2048,
                    "top_p": 0.9
                }
            }
        }

Config

Pydantic configuration.

Source code in uaef/experiments/models.py
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
class Config:
    """Pydantic configuration."""
    json_schema_extra = {
        "example": {
            "framework": "langgraph",
            "model": "us.anthropic.claude-sonnet-4-6",
            "tools": ["search", "calculator", "database_query"],
            "system_prompt": "You are a helpful assistant...",
            "parameters": {
                "temperature": 0.7,
                "max_tokens": 2048,
                "top_p": 0.9
            }
        }
    }

Experiment

Bases: BaseModel

Experiment for tracking agent versions and evaluations.

Attributes:

Name Type Description
experiment_id UUID

Unique identifier for the experiment

name str

Human-readable experiment name

description Optional[str]

Detailed description of the experiment

created_at datetime

When the experiment was created

metadata Dict[str, Any]

Additional metadata (tags, owner, project, etc.)

baseline_run_id Optional[UUID]

ID of the run designated as baseline for comparison

Source code in uaef/experiments/models.py
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
class Experiment(BaseModel):
    """
    Experiment for tracking agent versions and evaluations.

    Attributes:
        experiment_id: Unique identifier for the experiment
        name: Human-readable experiment name
        description: Detailed description of the experiment
        created_at: When the experiment was created
        metadata: Additional metadata (tags, owner, project, etc.)
        baseline_run_id: ID of the run designated as baseline for comparison
    """

    experiment_id: UUID = Field(
        default_factory=uuid4,
        description="Unique identifier for the experiment"
    )
    name: str = Field(..., description="Human-readable experiment name")
    description: Optional[str] = Field(
        None,
        description="Detailed description of the experiment"
    )
    created_at: datetime = Field(
        default_factory=datetime.utcnow,
        description="When the experiment was created"
    )
    metadata: Dict[str, Any] = Field(
        default_factory=dict,
        description="Additional metadata (tags, owner, project, etc.)"
    )
    baseline_run_id: Optional[UUID] = Field(
        None,
        description="ID of the run designated as baseline for comparison"
    )

    @field_validator("experiment_id")
    @classmethod
    def validate_experiment_id(cls, v: UUID) -> UUID:
        """Validate that experiment_id is not None."""
        if v is None:
            raise ValueError("experiment_id cannot be None")
        return v

    @field_validator("name")
    @classmethod
    def validate_name(cls, v: str) -> str:
        """Validate that name is not empty."""
        if not v or not v.strip():
            raise ValueError("Experiment name cannot be empty")
        return v.strip()

    class Config:
        """Pydantic configuration."""
        json_schema_extra = {
            "example": {
                "experiment_id": "550e8400-e29b-41d4-a716-446655440000",
                "name": "Customer Support Agent v2",
                "description": "Testing improved tool selection logic",
                "created_at": "2024-01-15T10:00:00Z",
                "metadata": {
                    "owner": "ml-team",
                    "project": "customer-support",
                    "tags": ["production", "tool-calling"]
                },
                "baseline_run_id": "660e8400-e29b-41d4-a716-446655440001"
            }
        }

Config

Pydantic configuration.

Source code in uaef/experiments/models.py
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
class Config:
    """Pydantic configuration."""
    json_schema_extra = {
        "example": {
            "experiment_id": "550e8400-e29b-41d4-a716-446655440000",
            "name": "Customer Support Agent v2",
            "description": "Testing improved tool selection logic",
            "created_at": "2024-01-15T10:00:00Z",
            "metadata": {
                "owner": "ml-team",
                "project": "customer-support",
                "tags": ["production", "tool-calling"]
            },
            "baseline_run_id": "660e8400-e29b-41d4-a716-446655440001"
        }
    }

validate_experiment_id(v) classmethod

Validate that experiment_id is not None.

Source code in uaef/experiments/models.py
92
93
94
95
96
97
98
@field_validator("experiment_id")
@classmethod
def validate_experiment_id(cls, v: UUID) -> UUID:
    """Validate that experiment_id is not None."""
    if v is None:
        raise ValueError("experiment_id cannot be None")
    return v

validate_name(v) classmethod

Validate that name is not empty.

Source code in uaef/experiments/models.py
100
101
102
103
104
105
106
@field_validator("name")
@classmethod
def validate_name(cls, v: str) -> str:
    """Validate that name is not empty."""
    if not v or not v.strip():
        raise ValueError("Experiment name cannot be empty")
    return v.strip()

ExperimentRun

Bases: BaseModel

A specific run within an experiment with config snapshot.

Attributes:

Name Type Description
run_id UUID

Unique identifier for this run

experiment_id UUID

ID of the parent experiment

run_name str

Human-readable name for this run

agent_design AgentDesign

Complete agent design configuration

config_snapshot Dict[str, Any]

Full configuration snapshot for reproducibility

git_commit Optional[str]

Git commit hash for code versioning

timestamp datetime

When the run was created

evaluation_count int

Number of evaluations in this run

aggregate_metrics Dict[str, float]

Aggregate metrics across all evaluations

metadata Dict[str, Any]

Additional run metadata

Source code in uaef/experiments/models.py
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
class ExperimentRun(BaseModel):
    """
    A specific run within an experiment with config snapshot.

    Attributes:
        run_id: Unique identifier for this run
        experiment_id: ID of the parent experiment
        run_name: Human-readable name for this run
        agent_design: Complete agent design configuration
        config_snapshot: Full configuration snapshot for reproducibility
        git_commit: Git commit hash for code versioning
        timestamp: When the run was created
        evaluation_count: Number of evaluations in this run
        aggregate_metrics: Aggregate metrics across all evaluations
        metadata: Additional run metadata
    """

    run_id: UUID = Field(
        default_factory=uuid4,
        description="Unique identifier for this run"
    )
    experiment_id: UUID = Field(
        ...,
        description="ID of the parent experiment"
    )
    run_name: str = Field(..., description="Human-readable name for this run")
    agent_design: AgentDesign = Field(
        ...,
        description="Complete agent design configuration"
    )
    config_snapshot: Dict[str, Any] = Field(
        ...,
        description="Full configuration snapshot for reproducibility"
    )
    git_commit: Optional[str] = Field(
        None,
        description="Git commit hash for code versioning"
    )
    timestamp: datetime = Field(
        default_factory=datetime.utcnow,
        description="When the run was created"
    )
    evaluation_count: int = Field(
        default=0,
        description="Number of evaluations in this run",
        ge=0
    )
    aggregate_metrics: Dict[str, float] = Field(
        default_factory=dict,
        description="Aggregate metrics across all evaluations"
    )
    metadata: Dict[str, Any] = Field(
        default_factory=dict,
        description="Additional run metadata"
    )

    @field_validator("run_id", "experiment_id")
    @classmethod
    def validate_uuid(cls, v: UUID) -> UUID:
        """Validate that UUIDs are not None."""
        if v is None:
            raise ValueError("UUID cannot be None")
        return v

    @field_validator("run_name")
    @classmethod
    def validate_run_name(cls, v: str) -> str:
        """Validate that run_name is not empty."""
        if not v or not v.strip():
            raise ValueError("Run name cannot be empty")
        return v.strip()

    @field_validator("git_commit")
    @classmethod
    def validate_git_commit(cls, v: Optional[str]) -> Optional[str]:
        """Validate git commit hash format (40 hex characters)."""
        if v is not None:
            v = v.strip()
            if len(v) != 40 or not all(c in "0123456789abcdef" for c in v.lower()):
                raise ValueError(
                    f"Invalid git commit hash: {v}. Must be 40 hexadecimal characters"
                )
        return v

    @field_validator("evaluation_count")
    @classmethod
    def validate_evaluation_count(cls, v: int) -> int:
        """Validate that evaluation_count is non-negative."""
        if v < 0:
            raise ValueError(f"evaluation_count must be non-negative, got {v}")
        return v

    @field_validator("aggregate_metrics")
    @classmethod
    def validate_aggregate_metrics(cls, v: Dict[str, float]) -> Dict[str, float]:
        """Validate that all metric scores are between 0 and 1."""
        for metric_name, score in v.items():
            if not 0.0 <= score <= 1.0:
                raise ValueError(
                    f"Metric '{metric_name}' score must be between 0 and 1, got {score}"
                )
        return v

    class Config:
        """Pydantic configuration."""
        json_schema_extra = {
            "example": {
                "run_id": "770e8400-e29b-41d4-a716-446655440002",
                "experiment_id": "550e8400-e29b-41d4-a716-446655440000",
                "run_name": "baseline-v1",
                "agent_design": {
                    "framework": "langgraph",
                    "model": "us.anthropic.claude-sonnet-4-6",
                    "tools": ["search", "calculator"],
                    "system_prompt": "You are a helpful assistant...",
                    "parameters": {"temperature": 0.7}
                },
                "config_snapshot": {
                    "metric_set": ["tool_accuracy", "answer_relevance"],
                    "thresholds": {"Tool Calling": 0.8},
                    "evaluation_mode": "offline"
                },
                "git_commit": "a1b2c3d4e5f6g7h8i9j0k1l2m3n4o5p6q7r8s9t0",  # pragma: allowlist secret
                "timestamp": "2024-01-15T10:30:00Z",
                "evaluation_count": 100,
                "aggregate_metrics": {
                    "tool_accuracy": 0.85,
                    "answer_relevance": 0.92
                },
                "metadata": {
                    "environment": "staging",
                    "test_suite": "customer_support_v1"
                }
            }
        }

Config

Pydantic configuration.

Source code in uaef/experiments/models.py
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
class Config:
    """Pydantic configuration."""
    json_schema_extra = {
        "example": {
            "run_id": "770e8400-e29b-41d4-a716-446655440002",
            "experiment_id": "550e8400-e29b-41d4-a716-446655440000",
            "run_name": "baseline-v1",
            "agent_design": {
                "framework": "langgraph",
                "model": "us.anthropic.claude-sonnet-4-6",
                "tools": ["search", "calculator"],
                "system_prompt": "You are a helpful assistant...",
                "parameters": {"temperature": 0.7}
            },
            "config_snapshot": {
                "metric_set": ["tool_accuracy", "answer_relevance"],
                "thresholds": {"Tool Calling": 0.8},
                "evaluation_mode": "offline"
            },
            "git_commit": "a1b2c3d4e5f6g7h8i9j0k1l2m3n4o5p6q7r8s9t0",  # pragma: allowlist secret
            "timestamp": "2024-01-15T10:30:00Z",
            "evaluation_count": 100,
            "aggregate_metrics": {
                "tool_accuracy": 0.85,
                "answer_relevance": 0.92
            },
            "metadata": {
                "environment": "staging",
                "test_suite": "customer_support_v1"
            }
        }
    }

validate_aggregate_metrics(v) classmethod

Validate that all metric scores are between 0 and 1.

Source code in uaef/experiments/models.py
218
219
220
221
222
223
224
225
226
227
@field_validator("aggregate_metrics")
@classmethod
def validate_aggregate_metrics(cls, v: Dict[str, float]) -> Dict[str, float]:
    """Validate that all metric scores are between 0 and 1."""
    for metric_name, score in v.items():
        if not 0.0 <= score <= 1.0:
            raise ValueError(
                f"Metric '{metric_name}' score must be between 0 and 1, got {score}"
            )
    return v

validate_evaluation_count(v) classmethod

Validate that evaluation_count is non-negative.

Source code in uaef/experiments/models.py
210
211
212
213
214
215
216
@field_validator("evaluation_count")
@classmethod
def validate_evaluation_count(cls, v: int) -> int:
    """Validate that evaluation_count is non-negative."""
    if v < 0:
        raise ValueError(f"evaluation_count must be non-negative, got {v}")
    return v

validate_git_commit(v) classmethod

Validate git commit hash format (40 hex characters).

Source code in uaef/experiments/models.py
198
199
200
201
202
203
204
205
206
207
208
@field_validator("git_commit")
@classmethod
def validate_git_commit(cls, v: Optional[str]) -> Optional[str]:
    """Validate git commit hash format (40 hex characters)."""
    if v is not None:
        v = v.strip()
        if len(v) != 40 or not all(c in "0123456789abcdef" for c in v.lower()):
            raise ValueError(
                f"Invalid git commit hash: {v}. Must be 40 hexadecimal characters"
            )
    return v

validate_run_name(v) classmethod

Validate that run_name is not empty.

Source code in uaef/experiments/models.py
190
191
192
193
194
195
196
@field_validator("run_name")
@classmethod
def validate_run_name(cls, v: str) -> str:
    """Validate that run_name is not empty."""
    if not v or not v.strip():
        raise ValueError("Run name cannot be empty")
    return v.strip()

validate_uuid(v) classmethod

Validate that UUIDs are not None.

Source code in uaef/experiments/models.py
182
183
184
185
186
187
188
@field_validator("run_id", "experiment_id")
@classmethod
def validate_uuid(cls, v: UUID) -> UUID:
    """Validate that UUIDs are not None."""
    if v is None:
        raise ValueError("UUID cannot be None")
    return v

uaef.experiments.manager

Experiment manager for creating and managing experiments and runs.

ExperimentManager

Manager for creating and managing experiments and runs.

Provides methods for: - Creating experiments and runs - Recording evaluation results - Setting baselines - Retrieving experiment data

Source code in uaef/experiments/manager.py
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
class ExperimentManager:
    """
    Manager for creating and managing experiments and runs.

    Provides methods for:
    - Creating experiments and runs
    - Recording evaluation results
    - Setting baselines
    - Retrieving experiment data
    """

    def __init__(self):
        """Initialize the experiment manager."""
        # In-memory storage for now (will be replaced with database in Task 6.8)
        self._experiments: Dict[UUID, Experiment] = {}
        self._runs: Dict[UUID, ExperimentRun] = {}
        self._evaluations: Dict[UUID, List[EvaluationResult]] = {}
        logger.info("ExperimentManager initialized")

    def create_experiment(
        self,
        name: str,
        description: Optional[str] = None,
        metadata: Optional[Dict[str, Any]] = None
    ) -> Experiment:
        """
        Create a new experiment.

        Args:
            name: Human-readable experiment name
            description: Detailed description of the experiment
            metadata: Additional metadata (tags, owner, project, etc.)

        Returns:
            Created Experiment instance

        Example:
            >>> manager = ExperimentManager()
            >>> experiment = manager.create_experiment(
            ...     name="Customer Support Agent v2",
            ...     description="Testing improved tool selection",
            ...     metadata={"owner": "ml-team", "tags": ["production"]}
            ... )
        """
        experiment = Experiment(
            name=name,
            description=description,
            metadata=metadata or {}
        )

        self._experiments[experiment.experiment_id] = experiment
        logger.info(
            f"Created experiment '{name}' with ID {experiment.experiment_id}"
        )

        return experiment

    def create_run(
        self,
        experiment_id: UUID,
        run_name: str,
        agent_design: AgentDesign,
        config: Dict[str, Any],
        git_commit: Optional[str] = None,
        metadata: Optional[Dict[str, Any]] = None
    ) -> ExperimentRun:
        """
        Create a new run within an experiment.

        Args:
            experiment_id: ID of the parent experiment
            run_name: Human-readable name for this run
            agent_design: Complete agent design configuration
            config: Full configuration snapshot for reproducibility
            git_commit: Git commit hash for code versioning
            metadata: Additional run metadata

        Returns:
            Created ExperimentRun instance

        Raises:
            ValueError: If experiment_id doesn't exist

        Example:
            >>> design = AgentDesign(
            ...     framework="langgraph",
            ...     model="us.anthropic.claude-sonnet-4-6",
            ...     tools=["search", "calculator"]
            ... )
            >>> run = manager.create_run(
            ...     experiment_id=experiment.experiment_id,
            ...     run_name="baseline-v1",
            ...     agent_design=design,
            ...     config={"metric_set": ["tool_accuracy"]},
            ...     git_commit="a1b2c3d4..."
            ... )
        """
        # Validate experiment exists
        if experiment_id not in self._experiments:
            raise ValueError(f"Experiment {experiment_id} not found")

        run = ExperimentRun(
            experiment_id=experiment_id,
            run_name=run_name,
            agent_design=agent_design,
            config_snapshot=config,
            git_commit=git_commit,
            metadata=metadata or {}
        )

        self._runs[run.run_id] = run
        self._evaluations[run.run_id] = []

        logger.info(
            f"Created run '{run_name}' with ID {run.run_id} "
            f"for experiment {experiment_id}"
        )

        return run

    def record_evaluation(
        self,
        run_id: UUID,
        evaluation_result: EvaluationResult
    ) -> None:
        """
        Record an evaluation result for a run.

        Args:
            run_id: ID of the run
            evaluation_result: Evaluation result to record

        Raises:
            ValueError: If run_id doesn't exist

        Example:
            >>> manager.record_evaluation(
            ...     run_id=run.run_id,
            ...     evaluation_result=evaluation_result
            ... )
        """
        # Validate run exists
        if run_id not in self._runs:
            raise ValueError(f"Run {run_id} not found")

        # Add evaluation to storage
        self._evaluations[run_id].append(evaluation_result)

        # Update run statistics
        run = self._runs[run_id]
        run.evaluation_count = len(self._evaluations[run_id])

        # Update aggregate metrics
        self._update_aggregate_metrics(run_id)

        logger.debug(
            f"Recorded evaluation {evaluation_result.evaluation_id} "
            f"for run {run_id}"
        )

    def _update_aggregate_metrics(self, run_id: UUID) -> None:
        """
        Update aggregate metrics for a run.

        Args:
            run_id: ID of the run to update
        """
        evaluations = self._evaluations[run_id]
        if not evaluations:
            return

        run = self._runs[run_id]

        # Collect all metric scores across evaluations
        metric_scores: Dict[str, List[float]] = {}

        for evaluation in evaluations:
            for dimension_result in evaluation.dimension_results:
                for metric_score in dimension_result.scored_metrics():
                    metric_name = metric_score.metric_name
                    if metric_name not in metric_scores:
                        metric_scores[metric_name] = []
                    metric_scores[metric_name].append(metric_score.score)

        # Calculate mean for each metric
        run.aggregate_metrics = {
            metric_name: sum(scores) / len(scores)
            for metric_name, scores in metric_scores.items()
        }

        logger.debug(f"Updated aggregate metrics for run {run_id}")

    def set_baseline(
        self,
        experiment_id: UUID,
        run_id: UUID
    ) -> None:
        """
        Set a run as the baseline for an experiment.

        Args:
            experiment_id: ID of the experiment
            run_id: ID of the run to set as baseline

        Raises:
            ValueError: If experiment or run doesn't exist, or run doesn't belong to experiment

        Example:
            >>> manager.set_baseline(
            ...     experiment_id=experiment.experiment_id,
            ...     run_id=run.run_id
            ... )
        """
        # Validate experiment exists
        if experiment_id not in self._experiments:
            raise ValueError(f"Experiment {experiment_id} not found")

        # Validate run exists
        if run_id not in self._runs:
            raise ValueError(f"Run {run_id} not found")

        # Validate run belongs to experiment
        run = self._runs[run_id]
        if run.experiment_id != experiment_id:
            raise ValueError(
                f"Run {run_id} does not belong to experiment {experiment_id}"
            )

        # Set baseline
        experiment = self._experiments[experiment_id]
        experiment.baseline_run_id = run_id

        logger.info(
            f"Set run {run_id} as baseline for experiment {experiment_id}"
        )

    def get_experiment(self, experiment_id: UUID) -> Optional[Experiment]:
        """
        Get an experiment by ID.

        Args:
            experiment_id: ID of the experiment

        Returns:
            Experiment instance or None if not found

        Example:
            >>> experiment = manager.get_experiment(experiment_id)
        """
        return self._experiments.get(experiment_id)

    def get_run(self, run_id: UUID) -> Optional[ExperimentRun]:
        """
        Get a run by ID.

        Args:
            run_id: ID of the run

        Returns:
            ExperimentRun instance or None if not found

        Example:
            >>> run = manager.get_run(run_id)
        """
        return self._runs.get(run_id)

    def list_experiments(self) -> List[Experiment]:
        """
        List all experiments.

        Returns:
            List of all Experiment instances

        Example:
            >>> experiments = manager.list_experiments()
        """
        return list(self._experiments.values())

    def list_runs(
        self,
        experiment_id: UUID
    ) -> List[ExperimentRun]:
        """
        List all runs for an experiment.

        Args:
            experiment_id: ID of the experiment

        Returns:
            List of ExperimentRun instances for the experiment

        Raises:
            ValueError: If experiment doesn't exist

        Example:
            >>> runs = manager.list_runs(experiment_id)
        """
        # Validate experiment exists
        if experiment_id not in self._experiments:
            raise ValueError(f"Experiment {experiment_id} not found")

        # Filter runs by experiment_id
        return [
            run for run in self._runs.values()
            if run.experiment_id == experiment_id
        ]

    def get_evaluations(
        self,
        run_id: UUID
    ) -> List[EvaluationResult]:
        """
        Get all evaluation results for a run.

        Args:
            run_id: ID of the run

        Returns:
            List of EvaluationResult instances for the run

        Raises:
            ValueError: If run doesn't exist

        Example:
            >>> evaluations = manager.get_evaluations(run_id)
        """
        # Validate run exists
        if run_id not in self._runs:
            raise ValueError(f"Run {run_id} not found")

        return self._evaluations.get(run_id, [])

    def get_baseline_run(
        self,
        experiment_id: UUID
    ) -> Optional[ExperimentRun]:
        """
        Get the baseline run for an experiment.

        Args:
            experiment_id: ID of the experiment

        Returns:
            Baseline ExperimentRun instance or None if no baseline set

        Raises:
            ValueError: If experiment doesn't exist

        Example:
            >>> baseline = manager.get_baseline_run(experiment_id)
        """
        # Validate experiment exists
        if experiment_id not in self._experiments:
            raise ValueError(f"Experiment {experiment_id} not found")

        experiment = self._experiments[experiment_id]
        if experiment.baseline_run_id is None:
            return None

        return self._runs.get(experiment.baseline_run_id)

create_experiment(name, description=None, metadata=None)

Create a new experiment.

Parameters:

Name Type Description Default
name str

Human-readable experiment name

required
description Optional[str]

Detailed description of the experiment

None
metadata Optional[Dict[str, Any]]

Additional metadata (tags, owner, project, etc.)

None

Returns:

Type Description
Experiment

Created Experiment instance

Example

manager = ExperimentManager() experiment = manager.create_experiment( ... name="Customer Support Agent v2", ... description="Testing improved tool selection", ... metadata={"owner": "ml-team", "tags": ["production"]} ... )

Source code in uaef/experiments/manager.py
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def create_experiment(
    self,
    name: str,
    description: Optional[str] = None,
    metadata: Optional[Dict[str, Any]] = None
) -> Experiment:
    """
    Create a new experiment.

    Args:
        name: Human-readable experiment name
        description: Detailed description of the experiment
        metadata: Additional metadata (tags, owner, project, etc.)

    Returns:
        Created Experiment instance

    Example:
        >>> manager = ExperimentManager()
        >>> experiment = manager.create_experiment(
        ...     name="Customer Support Agent v2",
        ...     description="Testing improved tool selection",
        ...     metadata={"owner": "ml-team", "tags": ["production"]}
        ... )
    """
    experiment = Experiment(
        name=name,
        description=description,
        metadata=metadata or {}
    )

    self._experiments[experiment.experiment_id] = experiment
    logger.info(
        f"Created experiment '{name}' with ID {experiment.experiment_id}"
    )

    return experiment

create_run(experiment_id, run_name, agent_design, config, git_commit=None, metadata=None)

Create a new run within an experiment.

Parameters:

Name Type Description Default
experiment_id UUID

ID of the parent experiment

required
run_name str

Human-readable name for this run

required
agent_design AgentDesign

Complete agent design configuration

required
config Dict[str, Any]

Full configuration snapshot for reproducibility

required
git_commit Optional[str]

Git commit hash for code versioning

None
metadata Optional[Dict[str, Any]]

Additional run metadata

None

Returns:

Type Description
ExperimentRun

Created ExperimentRun instance

Raises:

Type Description
ValueError

If experiment_id doesn't exist

Example

design = AgentDesign( ... framework="langgraph", ... model="us.anthropic.claude-sonnet-4-6", ... tools=["search", "calculator"] ... ) run = manager.create_run( ... experiment_id=experiment.experiment_id, ... run_name="baseline-v1", ... agent_design=design, ... config={"metric_set": ["tool_accuracy"]}, ... git_commit="a1b2c3d4..." ... )

Source code in uaef/experiments/manager.py
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def create_run(
    self,
    experiment_id: UUID,
    run_name: str,
    agent_design: AgentDesign,
    config: Dict[str, Any],
    git_commit: Optional[str] = None,
    metadata: Optional[Dict[str, Any]] = None
) -> ExperimentRun:
    """
    Create a new run within an experiment.

    Args:
        experiment_id: ID of the parent experiment
        run_name: Human-readable name for this run
        agent_design: Complete agent design configuration
        config: Full configuration snapshot for reproducibility
        git_commit: Git commit hash for code versioning
        metadata: Additional run metadata

    Returns:
        Created ExperimentRun instance

    Raises:
        ValueError: If experiment_id doesn't exist

    Example:
        >>> design = AgentDesign(
        ...     framework="langgraph",
        ...     model="us.anthropic.claude-sonnet-4-6",
        ...     tools=["search", "calculator"]
        ... )
        >>> run = manager.create_run(
        ...     experiment_id=experiment.experiment_id,
        ...     run_name="baseline-v1",
        ...     agent_design=design,
        ...     config={"metric_set": ["tool_accuracy"]},
        ...     git_commit="a1b2c3d4..."
        ... )
    """
    # Validate experiment exists
    if experiment_id not in self._experiments:
        raise ValueError(f"Experiment {experiment_id} not found")

    run = ExperimentRun(
        experiment_id=experiment_id,
        run_name=run_name,
        agent_design=agent_design,
        config_snapshot=config,
        git_commit=git_commit,
        metadata=metadata or {}
    )

    self._runs[run.run_id] = run
    self._evaluations[run.run_id] = []

    logger.info(
        f"Created run '{run_name}' with ID {run.run_id} "
        f"for experiment {experiment_id}"
    )

    return run

get_baseline_run(experiment_id)

Get the baseline run for an experiment.

Parameters:

Name Type Description Default
experiment_id UUID

ID of the experiment

required

Returns:

Type Description
Optional[ExperimentRun]

Baseline ExperimentRun instance or None if no baseline set

Raises:

Type Description
ValueError

If experiment doesn't exist

Example

baseline = manager.get_baseline_run(experiment_id)

Source code in uaef/experiments/manager.py
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
def get_baseline_run(
    self,
    experiment_id: UUID
) -> Optional[ExperimentRun]:
    """
    Get the baseline run for an experiment.

    Args:
        experiment_id: ID of the experiment

    Returns:
        Baseline ExperimentRun instance or None if no baseline set

    Raises:
        ValueError: If experiment doesn't exist

    Example:
        >>> baseline = manager.get_baseline_run(experiment_id)
    """
    # Validate experiment exists
    if experiment_id not in self._experiments:
        raise ValueError(f"Experiment {experiment_id} not found")

    experiment = self._experiments[experiment_id]
    if experiment.baseline_run_id is None:
        return None

    return self._runs.get(experiment.baseline_run_id)

get_evaluations(run_id)

Get all evaluation results for a run.

Parameters:

Name Type Description Default
run_id UUID

ID of the run

required

Returns:

Type Description
List[EvaluationResult]

List of EvaluationResult instances for the run

Raises:

Type Description
ValueError

If run doesn't exist

Example

evaluations = manager.get_evaluations(run_id)

Source code in uaef/experiments/manager.py
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
def get_evaluations(
    self,
    run_id: UUID
) -> List[EvaluationResult]:
    """
    Get all evaluation results for a run.

    Args:
        run_id: ID of the run

    Returns:
        List of EvaluationResult instances for the run

    Raises:
        ValueError: If run doesn't exist

    Example:
        >>> evaluations = manager.get_evaluations(run_id)
    """
    # Validate run exists
    if run_id not in self._runs:
        raise ValueError(f"Run {run_id} not found")

    return self._evaluations.get(run_id, [])

get_experiment(experiment_id)

Get an experiment by ID.

Parameters:

Name Type Description Default
experiment_id UUID

ID of the experiment

required

Returns:

Type Description
Optional[Experiment]

Experiment instance or None if not found

Example

experiment = manager.get_experiment(experiment_id)

Source code in uaef/experiments/manager.py
252
253
254
255
256
257
258
259
260
261
262
263
264
265
def get_experiment(self, experiment_id: UUID) -> Optional[Experiment]:
    """
    Get an experiment by ID.

    Args:
        experiment_id: ID of the experiment

    Returns:
        Experiment instance or None if not found

    Example:
        >>> experiment = manager.get_experiment(experiment_id)
    """
    return self._experiments.get(experiment_id)

get_run(run_id)

Get a run by ID.

Parameters:

Name Type Description Default
run_id UUID

ID of the run

required

Returns:

Type Description
Optional[ExperimentRun]

ExperimentRun instance or None if not found

Example

run = manager.get_run(run_id)

Source code in uaef/experiments/manager.py
267
268
269
270
271
272
273
274
275
276
277
278
279
280
def get_run(self, run_id: UUID) -> Optional[ExperimentRun]:
    """
    Get a run by ID.

    Args:
        run_id: ID of the run

    Returns:
        ExperimentRun instance or None if not found

    Example:
        >>> run = manager.get_run(run_id)
    """
    return self._runs.get(run_id)

list_experiments()

List all experiments.

Returns:

Type Description
List[Experiment]

List of all Experiment instances

Example

experiments = manager.list_experiments()

Source code in uaef/experiments/manager.py
282
283
284
285
286
287
288
289
290
291
292
def list_experiments(self) -> List[Experiment]:
    """
    List all experiments.

    Returns:
        List of all Experiment instances

    Example:
        >>> experiments = manager.list_experiments()
    """
    return list(self._experiments.values())

list_runs(experiment_id)

List all runs for an experiment.

Parameters:

Name Type Description Default
experiment_id UUID

ID of the experiment

required

Returns:

Type Description
List[ExperimentRun]

List of ExperimentRun instances for the experiment

Raises:

Type Description
ValueError

If experiment doesn't exist

Example

runs = manager.list_runs(experiment_id)

Source code in uaef/experiments/manager.py
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
def list_runs(
    self,
    experiment_id: UUID
) -> List[ExperimentRun]:
    """
    List all runs for an experiment.

    Args:
        experiment_id: ID of the experiment

    Returns:
        List of ExperimentRun instances for the experiment

    Raises:
        ValueError: If experiment doesn't exist

    Example:
        >>> runs = manager.list_runs(experiment_id)
    """
    # Validate experiment exists
    if experiment_id not in self._experiments:
        raise ValueError(f"Experiment {experiment_id} not found")

    # Filter runs by experiment_id
    return [
        run for run in self._runs.values()
        if run.experiment_id == experiment_id
    ]

record_evaluation(run_id, evaluation_result)

Record an evaluation result for a run.

Parameters:

Name Type Description Default
run_id UUID

ID of the run

required
evaluation_result EvaluationResult

Evaluation result to record

required

Raises:

Type Description
ValueError

If run_id doesn't exist

Example

manager.record_evaluation( ... run_id=run.run_id, ... evaluation_result=evaluation_result ... )

Source code in uaef/experiments/manager.py
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def record_evaluation(
    self,
    run_id: UUID,
    evaluation_result: EvaluationResult
) -> None:
    """
    Record an evaluation result for a run.

    Args:
        run_id: ID of the run
        evaluation_result: Evaluation result to record

    Raises:
        ValueError: If run_id doesn't exist

    Example:
        >>> manager.record_evaluation(
        ...     run_id=run.run_id,
        ...     evaluation_result=evaluation_result
        ... )
    """
    # Validate run exists
    if run_id not in self._runs:
        raise ValueError(f"Run {run_id} not found")

    # Add evaluation to storage
    self._evaluations[run_id].append(evaluation_result)

    # Update run statistics
    run = self._runs[run_id]
    run.evaluation_count = len(self._evaluations[run_id])

    # Update aggregate metrics
    self._update_aggregate_metrics(run_id)

    logger.debug(
        f"Recorded evaluation {evaluation_result.evaluation_id} "
        f"for run {run_id}"
    )

set_baseline(experiment_id, run_id)

Set a run as the baseline for an experiment.

Parameters:

Name Type Description Default
experiment_id UUID

ID of the experiment

required
run_id UUID

ID of the run to set as baseline

required

Raises:

Type Description
ValueError

If experiment or run doesn't exist, or run doesn't belong to experiment

Example

manager.set_baseline( ... experiment_id=experiment.experiment_id, ... run_id=run.run_id ... )

Source code in uaef/experiments/manager.py
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
def set_baseline(
    self,
    experiment_id: UUID,
    run_id: UUID
) -> None:
    """
    Set a run as the baseline for an experiment.

    Args:
        experiment_id: ID of the experiment
        run_id: ID of the run to set as baseline

    Raises:
        ValueError: If experiment or run doesn't exist, or run doesn't belong to experiment

    Example:
        >>> manager.set_baseline(
        ...     experiment_id=experiment.experiment_id,
        ...     run_id=run.run_id
        ... )
    """
    # Validate experiment exists
    if experiment_id not in self._experiments:
        raise ValueError(f"Experiment {experiment_id} not found")

    # Validate run exists
    if run_id not in self._runs:
        raise ValueError(f"Run {run_id} not found")

    # Validate run belongs to experiment
    run = self._runs[run_id]
    if run.experiment_id != experiment_id:
        raise ValueError(
            f"Run {run_id} does not belong to experiment {experiment_id}"
        )

    # Set baseline
    experiment = self._experiments[experiment_id]
    experiment.baseline_run_id = run_id

    logger.info(
        f"Set run {run_id} as baseline for experiment {experiment_id}"
    )

uaef.experiments.comparison

Comparison engine for comparing experiment runs.

ComparisonEngine

Engine for comparing experiment runs.

Provides methods for: - Comparing two runs metric-by-metric - Identifying regressions and improvements - Generating comparison reports

Source code in uaef/experiments/comparison.py
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
class ComparisonEngine:
    """
    Engine for comparing experiment runs.

    Provides methods for:
    - Comparing two runs metric-by-metric
    - Identifying regressions and improvements
    - Generating comparison reports
    """

    def __init__(
        self,
        regression_threshold: float = 5.0,
        improvement_threshold: float = 5.0
    ):
        """
        Initialize the comparison engine.

        Args:
            regression_threshold: Percentage threshold for regression detection (default: 5.0%)
            improvement_threshold: Percentage threshold for improvement detection (default: 5.0%)
        """
        self.regression_threshold = regression_threshold
        self.improvement_threshold = improvement_threshold
        logger.info(
            f"ComparisonEngine initialized with regression_threshold={regression_threshold}%, "
            f"improvement_threshold={improvement_threshold}%"
        )

    def compare_runs(
        self,
        baseline_run: ExperimentRun,
        current_run: ExperimentRun,
        regression_threshold: Optional[float] = None,
        improvement_threshold: Optional[float] = None
    ) -> ComparisonReport:
        """
        Compare two experiment runs metric-by-metric.

        Args:
            baseline_run: Baseline run for comparison
            current_run: Current run to compare against baseline
            regression_threshold: Custom regression threshold (overrides default)
            improvement_threshold: Custom improvement threshold (overrides default)

        Returns:
            ComparisonReport with detailed comparison results

        Example:
            >>> engine = ComparisonEngine()
            >>> report = engine.compare_runs(baseline_run, current_run)
            >>> print(f"Regressions: {report.regressions}")
            >>> print(f"Improvements: {report.improvements}")
        """
        # Use custom thresholds if provided
        reg_threshold = regression_threshold or self.regression_threshold
        imp_threshold = improvement_threshold or self.improvement_threshold

        # Get aggregate metrics from both runs
        baseline_metrics = baseline_run.aggregate_metrics
        current_metrics = current_run.aggregate_metrics

        # Find common metrics
        common_metrics = set(baseline_metrics.keys()) & set(current_metrics.keys())

        if not common_metrics:
            logger.warning(
                f"No common metrics found between runs {baseline_run.run_id} "
                f"and {current_run.run_id}"
            )

        # Compare each metric
        metric_comparisons: List[MetricComparison] = []
        regressions: List[str] = []
        improvements: List[str] = []
        unchanged: List[str] = []

        for metric_name in sorted(common_metrics):
            baseline_score = baseline_metrics[metric_name]
            current_score = current_metrics[metric_name]

            comparison = self._compare_metric(
                metric_name=metric_name,
                baseline_score=baseline_score,
                current_score=current_score,
                regression_threshold=reg_threshold,
                improvement_threshold=imp_threshold
            )

            metric_comparisons.append(comparison)

            # Categorize the change
            if comparison.is_regression:
                regressions.append(metric_name)
            elif comparison.is_improvement:
                improvements.append(metric_name)
            else:
                unchanged.append(metric_name)

        # Generate summary
        summary = {
            "total_metrics": len(common_metrics),
            "regression_count": len(regressions),
            "improvement_count": len(improvements),
            "unchanged_count": len(unchanged),
            "regression_threshold": reg_threshold,
            "improvement_threshold": imp_threshold
        }

        report = ComparisonReport(
            baseline_run_id=baseline_run.run_id,
            current_run_id=current_run.run_id,
            metric_comparisons=metric_comparisons,
            regressions=regressions,
            improvements=improvements,
            unchanged=unchanged,
            summary=summary
        )

        logger.info(
            f"Compared runs {baseline_run.run_id} and {current_run.run_id}: "
            f"{len(regressions)} regressions, {len(improvements)} improvements, "
            f"{len(unchanged)} unchanged"
        )

        return report

    def _compare_metric(
        self,
        metric_name: str,
        baseline_score: float,
        current_score: float,
        regression_threshold: float,
        improvement_threshold: float
    ) -> MetricComparison:
        """
        Compare a single metric between baseline and current.

        Args:
            metric_name: Name of the metric
            baseline_score: Score from baseline run
            current_score: Score from current run
            regression_threshold: Threshold for regression detection
            improvement_threshold: Threshold for improvement detection

        Returns:
            MetricComparison instance
        """
        # Calculate delta
        delta = current_score - baseline_score

        # Calculate percent change (handle division by zero)
        if baseline_score == 0:
            if current_score == 0:
                percent_change = 0.0
            else:
                # If baseline is 0 and current is not, treat as 100% change
                percent_change = 100.0 if current_score > 0 else -100.0
        else:
            percent_change = (delta / baseline_score) * 100.0

        # Determine if regression or improvement
        is_regression = percent_change < -regression_threshold
        is_improvement = percent_change > improvement_threshold

        return MetricComparison(
            metric_name=metric_name,
            baseline_score=baseline_score,
            current_score=current_score,
            delta=delta,
            percent_change=percent_change,
            is_regression=is_regression,
            is_improvement=is_improvement
        )

    def compare_multiple_runs(
        self,
        baseline_run: ExperimentRun,
        current_runs: List[ExperimentRun],
        regression_threshold: Optional[float] = None,
        improvement_threshold: Optional[float] = None
    ) -> List[ComparisonReport]:
        """
        Compare multiple runs against a baseline.

        Args:
            baseline_run: Baseline run for comparison
            current_runs: List of runs to compare against baseline
            regression_threshold: Custom regression threshold (overrides default)
            improvement_threshold: Custom improvement threshold (overrides default)

        Returns:
            List of ComparisonReport instances, one for each current run

        Example:
            >>> engine = ComparisonEngine()
            >>> reports = engine.compare_multiple_runs(baseline_run, [run1, run2, run3])
            >>> for report in reports:
            ...     print(f"Run {report.current_run_id}: {len(report.regressions)} regressions")
        """
        reports = []

        for current_run in current_runs:
            report = self.compare_runs(
                baseline_run=baseline_run,
                current_run=current_run,
                regression_threshold=regression_threshold,
                improvement_threshold=improvement_threshold
            )
            reports.append(report)

        logger.info(
            f"Compared baseline {baseline_run.run_id} against {len(current_runs)} runs"
        )

        return reports

    def get_comparison_summary(
        self,
        report: ComparisonReport
    ) -> str:
        """
        Generate a human-readable summary of a comparison report.

        Args:
            report: ComparisonReport to summarize

        Returns:
            Human-readable summary string

        Example:
            >>> summary = engine.get_comparison_summary(report)
            >>> print(summary)
        """
        lines = [
            f"Comparison Report",
            f"=================",
            f"Baseline Run: {report.baseline_run_id}",
            f"Current Run: {report.current_run_id}",
            f"",
            f"Summary:",
            f"  Total Metrics: {report.summary['total_metrics']}",
            f"  Regressions: {report.summary['regression_count']}",
            f"  Improvements: {report.summary['improvement_count']}",
            f"  Unchanged: {report.summary['unchanged_count']}",
            f""
        ]

        if report.regressions:
            lines.append("Regressions:")
            for metric_name in report.regressions:
                comparison = next(
                    c for c in report.metric_comparisons
                    if c.metric_name == metric_name
                )
                lines.append(
                    f"  - {metric_name}: {comparison.baseline_score:.3f} → "
                    f"{comparison.current_score:.3f} ({comparison.percent_change:+.1f}%)"
                )
            lines.append("")

        if report.improvements:
            lines.append("Improvements:")
            for metric_name in report.improvements:
                comparison = next(
                    c for c in report.metric_comparisons
                    if c.metric_name == metric_name
                )
                lines.append(
                    f"  - {metric_name}: {comparison.baseline_score:.3f} → "
                    f"{comparison.current_score:.3f} ({comparison.percent_change:+.1f}%)"
                )
            lines.append("")

        return "\n".join(lines)

compare_multiple_runs(baseline_run, current_runs, regression_threshold=None, improvement_threshold=None)

Compare multiple runs against a baseline.

Parameters:

Name Type Description Default
baseline_run ExperimentRun

Baseline run for comparison

required
current_runs List[ExperimentRun]

List of runs to compare against baseline

required
regression_threshold Optional[float]

Custom regression threshold (overrides default)

None
improvement_threshold Optional[float]

Custom improvement threshold (overrides default)

None

Returns:

Type Description
List[ComparisonReport]

List of ComparisonReport instances, one for each current run

Example

engine = ComparisonEngine() reports = engine.compare_multiple_runs(baseline_run, [run1, run2, run3]) for report in reports: ... print(f"Run {report.current_run_id}: {len(report.regressions)} regressions")

Source code in uaef/experiments/comparison.py
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
def compare_multiple_runs(
    self,
    baseline_run: ExperimentRun,
    current_runs: List[ExperimentRun],
    regression_threshold: Optional[float] = None,
    improvement_threshold: Optional[float] = None
) -> List[ComparisonReport]:
    """
    Compare multiple runs against a baseline.

    Args:
        baseline_run: Baseline run for comparison
        current_runs: List of runs to compare against baseline
        regression_threshold: Custom regression threshold (overrides default)
        improvement_threshold: Custom improvement threshold (overrides default)

    Returns:
        List of ComparisonReport instances, one for each current run

    Example:
        >>> engine = ComparisonEngine()
        >>> reports = engine.compare_multiple_runs(baseline_run, [run1, run2, run3])
        >>> for report in reports:
        ...     print(f"Run {report.current_run_id}: {len(report.regressions)} regressions")
    """
    reports = []

    for current_run in current_runs:
        report = self.compare_runs(
            baseline_run=baseline_run,
            current_run=current_run,
            regression_threshold=regression_threshold,
            improvement_threshold=improvement_threshold
        )
        reports.append(report)

    logger.info(
        f"Compared baseline {baseline_run.run_id} against {len(current_runs)} runs"
    )

    return reports

compare_runs(baseline_run, current_run, regression_threshold=None, improvement_threshold=None)

Compare two experiment runs metric-by-metric.

Parameters:

Name Type Description Default
baseline_run ExperimentRun

Baseline run for comparison

required
current_run ExperimentRun

Current run to compare against baseline

required
regression_threshold Optional[float]

Custom regression threshold (overrides default)

None
improvement_threshold Optional[float]

Custom improvement threshold (overrides default)

None

Returns:

Type Description
ComparisonReport

ComparisonReport with detailed comparison results

Example

engine = ComparisonEngine() report = engine.compare_runs(baseline_run, current_run) print(f"Regressions: {report.regressions}") print(f"Improvements: {report.improvements}")

Source code in uaef/experiments/comparison.py
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
def compare_runs(
    self,
    baseline_run: ExperimentRun,
    current_run: ExperimentRun,
    regression_threshold: Optional[float] = None,
    improvement_threshold: Optional[float] = None
) -> ComparisonReport:
    """
    Compare two experiment runs metric-by-metric.

    Args:
        baseline_run: Baseline run for comparison
        current_run: Current run to compare against baseline
        regression_threshold: Custom regression threshold (overrides default)
        improvement_threshold: Custom improvement threshold (overrides default)

    Returns:
        ComparisonReport with detailed comparison results

    Example:
        >>> engine = ComparisonEngine()
        >>> report = engine.compare_runs(baseline_run, current_run)
        >>> print(f"Regressions: {report.regressions}")
        >>> print(f"Improvements: {report.improvements}")
    """
    # Use custom thresholds if provided
    reg_threshold = regression_threshold or self.regression_threshold
    imp_threshold = improvement_threshold or self.improvement_threshold

    # Get aggregate metrics from both runs
    baseline_metrics = baseline_run.aggregate_metrics
    current_metrics = current_run.aggregate_metrics

    # Find common metrics
    common_metrics = set(baseline_metrics.keys()) & set(current_metrics.keys())

    if not common_metrics:
        logger.warning(
            f"No common metrics found between runs {baseline_run.run_id} "
            f"and {current_run.run_id}"
        )

    # Compare each metric
    metric_comparisons: List[MetricComparison] = []
    regressions: List[str] = []
    improvements: List[str] = []
    unchanged: List[str] = []

    for metric_name in sorted(common_metrics):
        baseline_score = baseline_metrics[metric_name]
        current_score = current_metrics[metric_name]

        comparison = self._compare_metric(
            metric_name=metric_name,
            baseline_score=baseline_score,
            current_score=current_score,
            regression_threshold=reg_threshold,
            improvement_threshold=imp_threshold
        )

        metric_comparisons.append(comparison)

        # Categorize the change
        if comparison.is_regression:
            regressions.append(metric_name)
        elif comparison.is_improvement:
            improvements.append(metric_name)
        else:
            unchanged.append(metric_name)

    # Generate summary
    summary = {
        "total_metrics": len(common_metrics),
        "regression_count": len(regressions),
        "improvement_count": len(improvements),
        "unchanged_count": len(unchanged),
        "regression_threshold": reg_threshold,
        "improvement_threshold": imp_threshold
    }

    report = ComparisonReport(
        baseline_run_id=baseline_run.run_id,
        current_run_id=current_run.run_id,
        metric_comparisons=metric_comparisons,
        regressions=regressions,
        improvements=improvements,
        unchanged=unchanged,
        summary=summary
    )

    logger.info(
        f"Compared runs {baseline_run.run_id} and {current_run.run_id}: "
        f"{len(regressions)} regressions, {len(improvements)} improvements, "
        f"{len(unchanged)} unchanged"
    )

    return report

get_comparison_summary(report)

Generate a human-readable summary of a comparison report.

Parameters:

Name Type Description Default
report ComparisonReport

ComparisonReport to summarize

required

Returns:

Type Description
str

Human-readable summary string

Example

summary = engine.get_comparison_summary(report) print(summary)

Source code in uaef/experiments/comparison.py
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
def get_comparison_summary(
    self,
    report: ComparisonReport
) -> str:
    """
    Generate a human-readable summary of a comparison report.

    Args:
        report: ComparisonReport to summarize

    Returns:
        Human-readable summary string

    Example:
        >>> summary = engine.get_comparison_summary(report)
        >>> print(summary)
    """
    lines = [
        f"Comparison Report",
        f"=================",
        f"Baseline Run: {report.baseline_run_id}",
        f"Current Run: {report.current_run_id}",
        f"",
        f"Summary:",
        f"  Total Metrics: {report.summary['total_metrics']}",
        f"  Regressions: {report.summary['regression_count']}",
        f"  Improvements: {report.summary['improvement_count']}",
        f"  Unchanged: {report.summary['unchanged_count']}",
        f""
    ]

    if report.regressions:
        lines.append("Regressions:")
        for metric_name in report.regressions:
            comparison = next(
                c for c in report.metric_comparisons
                if c.metric_name == metric_name
            )
            lines.append(
                f"  - {metric_name}: {comparison.baseline_score:.3f} → "
                f"{comparison.current_score:.3f} ({comparison.percent_change:+.1f}%)"
            )
        lines.append("")

    if report.improvements:
        lines.append("Improvements:")
        for metric_name in report.improvements:
            comparison = next(
                c for c in report.metric_comparisons
                if c.metric_name == metric_name
            )
            lines.append(
                f"  - {metric_name}: {comparison.baseline_score:.3f} → "
                f"{comparison.current_score:.3f} ({comparison.percent_change:+.1f}%)"
            )
        lines.append("")

    return "\n".join(lines)

ComparisonReport

Bases: BaseModel

Complete comparison report between two runs.

Attributes:

Name Type Description
baseline_run_id UUID

ID of the baseline run

current_run_id UUID

ID of the current run

metric_comparisons List[MetricComparison]

List of metric-by-metric comparisons

regressions List[str]

List of metrics with regressions

improvements List[str]

List of metrics with improvements

unchanged List[str]

List of metrics with no significant change

summary Dict[str, Any]

Summary statistics

Source code in uaef/experiments/comparison.py
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
class ComparisonReport(BaseModel):
    """
    Complete comparison report between two runs.

    Attributes:
        baseline_run_id: ID of the baseline run
        current_run_id: ID of the current run
        metric_comparisons: List of metric-by-metric comparisons
        regressions: List of metrics with regressions
        improvements: List of metrics with improvements
        unchanged: List of metrics with no significant change
        summary: Summary statistics
    """

    baseline_run_id: UUID = Field(..., description="ID of the baseline run")
    current_run_id: UUID = Field(..., description="ID of the current run")
    metric_comparisons: List[MetricComparison] = Field(
        ...,
        description="List of metric-by-metric comparisons"
    )
    regressions: List[str] = Field(
        ...,
        description="List of metrics with regressions"
    )
    improvements: List[str] = Field(
        ...,
        description="List of metrics with improvements"
    )
    unchanged: List[str] = Field(
        ...,
        description="List of metrics with no significant change"
    )
    summary: Dict[str, Any] = Field(
        ...,
        description="Summary statistics"
    )

    class Config:
        """Pydantic configuration."""
        json_schema_extra = {
            "example": {
                "baseline_run_id": "550e8400-e29b-41d4-a716-446655440000",
                "current_run_id": "660e8400-e29b-41d4-a716-446655440001",
                "metric_comparisons": [],
                "regressions": ["tool_accuracy"],
                "improvements": ["answer_relevance"],
                "unchanged": ["latency_score"],
                "summary": {
                    "total_metrics": 3,
                    "regression_count": 1,
                    "improvement_count": 1,
                    "unchanged_count": 1
                }
            }
        }

Config

Pydantic configuration.

Source code in uaef/experiments/comparison.py
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
class Config:
    """Pydantic configuration."""
    json_schema_extra = {
        "example": {
            "baseline_run_id": "550e8400-e29b-41d4-a716-446655440000",
            "current_run_id": "660e8400-e29b-41d4-a716-446655440001",
            "metric_comparisons": [],
            "regressions": ["tool_accuracy"],
            "improvements": ["answer_relevance"],
            "unchanged": ["latency_score"],
            "summary": {
                "total_metrics": 3,
                "regression_count": 1,
                "improvement_count": 1,
                "unchanged_count": 1
            }
        }
    }

MetricComparison

Bases: BaseModel

Comparison result for a single metric.

Attributes:

Name Type Description
metric_name str

Name of the metric

baseline_score float

Score from baseline run

current_score float

Score from current run

delta float

Absolute difference (current - baseline)

percent_change float

Percentage change ((current - baseline) / baseline * 100)

is_regression bool

Whether this is a regression (negative change > threshold)

is_improvement bool

Whether this is an improvement (positive change > threshold)

Source code in uaef/experiments/comparison.py
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
class MetricComparison(BaseModel):
    """
    Comparison result for a single metric.

    Attributes:
        metric_name: Name of the metric
        baseline_score: Score from baseline run
        current_score: Score from current run
        delta: Absolute difference (current - baseline)
        percent_change: Percentage change ((current - baseline) / baseline * 100)
        is_regression: Whether this is a regression (negative change > threshold)
        is_improvement: Whether this is an improvement (positive change > threshold)
    """

    metric_name: str = Field(..., description="Name of the metric")
    baseline_score: float = Field(..., description="Score from baseline run")
    current_score: float = Field(..., description="Score from current run")
    delta: float = Field(..., description="Absolute difference (current - baseline)")
    percent_change: float = Field(
        ...,
        description="Percentage change ((current - baseline) / baseline * 100)"
    )
    is_regression: bool = Field(
        ...,
        description="Whether this is a regression (negative change > threshold)"
    )
    is_improvement: bool = Field(
        ...,
        description="Whether this is an improvement (positive change > threshold)"
    )

    class Config:
        """Pydantic configuration."""
        json_schema_extra = {
            "example": {
                "metric_name": "tool_accuracy",
                "baseline_score": 0.85,
                "current_score": 0.78,
                "delta": -0.07,
                "percent_change": -8.24,
                "is_regression": True,
                "is_improvement": False
            }
        }

Config

Pydantic configuration.

Source code in uaef/experiments/comparison.py
48
49
50
51
52
53
54
55
56
57
58
59
60
class Config:
    """Pydantic configuration."""
    json_schema_extra = {
        "example": {
            "metric_name": "tool_accuracy",
            "baseline_score": 0.85,
            "current_score": 0.78,
            "delta": -0.07,
            "percent_change": -8.24,
            "is_regression": True,
            "is_improvement": False
        }
    }

uaef.experiments.regression

Regression detector for identifying performance degradations.

Regression

Bases: BaseModel

A detected regression in a metric.

Attributes:

Name Type Description
metric_name str

Name of the metric with regression

baseline_score float

Score from baseline run

current_score float

Score from current run

delta float

Absolute difference (current - baseline)

percent_change float

Percentage change

severity RegressionSeverity

Severity level of the regression

dimension Optional[str]

Dimension the metric belongs to (if known)

Source code in uaef/experiments/regression.py
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
class Regression(BaseModel):
    """
    A detected regression in a metric.

    Attributes:
        metric_name: Name of the metric with regression
        baseline_score: Score from baseline run
        current_score: Score from current run
        delta: Absolute difference (current - baseline)
        percent_change: Percentage change
        severity: Severity level of the regression
        dimension: Dimension the metric belongs to (if known)
    """

    metric_name: str = Field(..., description="Name of the metric with regression")
    baseline_score: float = Field(..., description="Score from baseline run")
    current_score: float = Field(..., description="Score from current run")
    delta: float = Field(..., description="Absolute difference (current - baseline)")
    percent_change: float = Field(..., description="Percentage change")
    severity: RegressionSeverity = Field(..., description="Severity level of the regression")
    dimension: Optional[str] = Field(
        None,
        description="Dimension the metric belongs to (if known)"
    )

    class Config:
        """Pydantic configuration."""
        json_schema_extra = {
            "example": {
                "metric_name": "tool_accuracy",
                "baseline_score": 0.85,
                "current_score": 0.65,
                "delta": -0.20,
                "percent_change": -23.53,
                "severity": "severe",
                "dimension": "tool_calling"
            }
        }

Config

Pydantic configuration.

Source code in uaef/experiments/regression.py
51
52
53
54
55
56
57
58
59
60
61
62
63
class Config:
    """Pydantic configuration."""
    json_schema_extra = {
        "example": {
            "metric_name": "tool_accuracy",
            "baseline_score": 0.85,
            "current_score": 0.65,
            "delta": -0.20,
            "percent_change": -23.53,
            "severity": "severe",
            "dimension": "tool_calling"
        }
    }

RegressionDetector

Detector for identifying performance regressions.

Provides methods for: - Detecting regressions between runs - Calculating severity based on magnitude - Generating regression reports

Source code in uaef/experiments/regression.py
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
class RegressionDetector:
    """
    Detector for identifying performance regressions.

    Provides methods for:
    - Detecting regressions between runs
    - Calculating severity based on magnitude
    - Generating regression reports
    """

    # Default thresholds for severity levels
    DEFAULT_MINOR_THRESHOLD = 5.0  # 5-10% degradation
    DEFAULT_MODERATE_THRESHOLD = 10.0  # 10-20% degradation
    DEFAULT_SEVERE_THRESHOLD = 20.0  # >20% degradation

    # Mapping of metric names to dimensions (can be extended)
    METRIC_DIMENSION_MAP = {
        "tool_accuracy": "tool_calling",
        "tool_selection_accuracy": "tool_calling",
        "tool_sequence_correctness": "tool_calling",
        "parameter_quality": "tool_calling",
        "answer_relevance": "response_quality",
        "completeness": "response_quality",
        "hallucination_score": "response_quality",
        "accuracy": "response_quality",
        "safety_score": "responsible_ai",
        "bias_score": "responsible_ai",
        "toxicity_score": "responsible_ai",
        "latency_score": "performance",
        "token_efficiency": "performance",
        "cost_efficiency": "performance",
        "context_retention": "multi_turn",
        "coherence": "multi_turn",
        "conversation_completeness": "multi_turn",
        "agent_utilization": "multi_agent",
        "delegation_quality": "multi_agent",
        "workflow_completion": "multi_agent",
        "chain_of_thought_coherence": "reasoning",
        "logical_consistency": "reasoning",
    }

    def __init__(
        self,
        thresholds: Optional[Dict[str, float]] = None,
        minor_threshold: float = DEFAULT_MINOR_THRESHOLD,
        moderate_threshold: float = DEFAULT_MODERATE_THRESHOLD,
        severe_threshold: float = DEFAULT_SEVERE_THRESHOLD
    ):
        """
        Initialize the regression detector.

        Args:
            thresholds: Custom thresholds per metric (metric_name -> threshold %)
            minor_threshold: Global threshold for minor regressions (default: 5%)
            moderate_threshold: Global threshold for moderate regressions (default: 10%)
            severe_threshold: Global threshold for severe regressions (default: 20%)
        """
        self.thresholds = thresholds or {}
        self.minor_threshold = minor_threshold
        self.moderate_threshold = moderate_threshold
        self.severe_threshold = severe_threshold

        logger.info(
            f"RegressionDetector initialized with thresholds: "
            f"minor={minor_threshold}%, moderate={moderate_threshold}%, "
            f"severe={severe_threshold}%"
        )

    def detect_regressions(
        self,
        baseline_run: ExperimentRun,
        current_run: ExperimentRun,
        thresholds: Optional[Dict[str, float]] = None
    ) -> RegressionReport:
        """
        Detect regressions between baseline and current runs.

        Args:
            baseline_run: Baseline run for comparison
            current_run: Current run to check for regressions
            thresholds: Custom thresholds per metric (overrides defaults)

        Returns:
            RegressionReport with detected regressions

        Example:
            >>> detector = RegressionDetector()
            >>> report = detector.detect_regressions(baseline_run, current_run)
            >>> print(f"Found {len(report.regressions)} regressions")
            >>> for regression in report.regressions:
            ...     print(f"  {regression.metric_name}: {regression.severity}")
        """
        # Use custom thresholds if provided
        metric_thresholds = thresholds or self.thresholds

        # Get aggregate metrics from both runs
        baseline_metrics = baseline_run.aggregate_metrics
        current_metrics = current_run.aggregate_metrics

        # Find common metrics
        common_metrics = set(baseline_metrics.keys()) & set(current_metrics.keys())

        if not common_metrics:
            logger.warning(
                f"No common metrics found between runs {baseline_run.run_id} "
                f"and {current_run.run_id}"
            )

        # Detect regressions
        regressions: List[Regression] = []

        for metric_name in sorted(common_metrics):
            baseline_score = baseline_metrics[metric_name]
            current_score = current_metrics[metric_name]

            # Get threshold for this metric
            threshold = metric_thresholds.get(metric_name, self.minor_threshold)

            # Check if this is a regression
            regression = self._check_regression(
                metric_name=metric_name,
                baseline_score=baseline_score,
                current_score=current_score,
                threshold=threshold
            )

            if regression:
                regressions.append(regression)

        # Group regressions by severity
        regressions_by_severity = {
            "severe": [],
            "moderate": [],
            "minor": []
        }

        for regression in regressions:
            regressions_by_severity[regression.severity.value].append(
                regression.metric_name
            )

        # Group regressions by dimension
        regressions_by_dimension: Dict[str, List[str]] = {}

        for regression in regressions:
            dimension = regression.dimension or "unknown"
            if dimension not in regressions_by_dimension:
                regressions_by_dimension[dimension] = []
            regressions_by_dimension[dimension].append(regression.metric_name)

        # Generate summary
        summary = {
            "total_regressions": len(regressions),
            "severe_count": len(regressions_by_severity["severe"]),
            "moderate_count": len(regressions_by_severity["moderate"]),
            "minor_count": len(regressions_by_severity["minor"]),
            "dimensions_affected": len(regressions_by_dimension)
        }

        report = RegressionReport(
            baseline_run_id=baseline_run.run_id,
            current_run_id=current_run.run_id,
            regressions=regressions,
            regressions_by_severity=regressions_by_severity,
            regressions_by_dimension=regressions_by_dimension,
            summary=summary
        )

        logger.info(
            f"Detected {len(regressions)} regressions between runs "
            f"{baseline_run.run_id} and {current_run.run_id}: "
            f"{summary['severe_count']} severe, {summary['moderate_count']} moderate, "
            f"{summary['minor_count']} minor"
        )

        return report

    def _check_regression(
        self,
        metric_name: str,
        baseline_score: float,
        current_score: float,
        threshold: float
    ) -> Optional[Regression]:
        """
        Check if a metric has regressed.

        Args:
            metric_name: Name of the metric
            baseline_score: Score from baseline run
            current_score: Score from current run
            threshold: Threshold for regression detection

        Returns:
            Regression instance if regression detected, None otherwise
        """
        # Calculate delta and percent change
        delta = current_score - baseline_score

        # Calculate percent change (handle division by zero)
        if baseline_score == 0:
            if current_score == 0:
                percent_change = 0.0
            else:
                percent_change = 100.0 if current_score > 0 else -100.0
        else:
            percent_change = (delta / baseline_score) * 100.0

        # Check if this is a regression (negative change beyond threshold)
        if percent_change >= -threshold:
            return None  # Not a regression

        # Determine severity based on magnitude
        abs_percent_change = abs(percent_change)

        if abs_percent_change >= self.severe_threshold:
            severity = RegressionSeverity.SEVERE
        elif abs_percent_change >= self.moderate_threshold:
            severity = RegressionSeverity.MODERATE
        else:
            severity = RegressionSeverity.MINOR

        # Get dimension for this metric
        dimension = self.METRIC_DIMENSION_MAP.get(metric_name)

        return Regression(
            metric_name=metric_name,
            baseline_score=baseline_score,
            current_score=current_score,
            delta=delta,
            percent_change=percent_change,
            severity=severity,
            dimension=dimension
        )

    def get_regression_summary(
        self,
        report: RegressionReport
    ) -> str:
        """
        Generate a human-readable summary of a regression report.

        Args:
            report: RegressionReport to summarize

        Returns:
            Human-readable summary string

        Example:
            >>> summary = detector.get_regression_summary(report)
            >>> print(summary)
        """
        lines = [
            f"Regression Detection Report",
            f"===========================",
            f"Baseline Run: {report.baseline_run_id}",
            f"Current Run: {report.current_run_id}",
            f"",
            f"Summary:",
            f"  Total Regressions: {report.summary['total_regressions']}",
            f"  Severe: {report.summary['severe_count']}",
            f"  Moderate: {report.summary['moderate_count']}",
            f"  Minor: {report.summary['minor_count']}",
            f"  Dimensions Affected: {report.summary['dimensions_affected']}",
            f""
        ]

        if not report.regressions:
            lines.append("No regressions detected.")
            return "\n".join(lines)

        # Group by severity
        for severity in ["severe", "moderate", "minor"]:
            metric_names = report.regressions_by_severity.get(severity, [])
            if metric_names:
                lines.append(f"{severity.capitalize()} Regressions:")
                for metric_name in metric_names:
                    regression = next(
                        r for r in report.regressions
                        if r.metric_name == metric_name
                    )
                    lines.append(
                        f"  - {metric_name}: {regression.baseline_score:.3f} → "
                        f"{regression.current_score:.3f} ({regression.percent_change:+.1f}%)"
                    )
                lines.append("")

        # Group by dimension
        if report.regressions_by_dimension:
            lines.append("Regressions by Dimension:")
            for dimension, metric_names in sorted(report.regressions_by_dimension.items()):
                lines.append(f"  {dimension}: {', '.join(metric_names)}")
            lines.append("")

        return "\n".join(lines)

detect_regressions(baseline_run, current_run, thresholds=None)

Detect regressions between baseline and current runs.

Parameters:

Name Type Description Default
baseline_run ExperimentRun

Baseline run for comparison

required
current_run ExperimentRun

Current run to check for regressions

required
thresholds Optional[Dict[str, float]]

Custom thresholds per metric (overrides defaults)

None

Returns:

Type Description
RegressionReport

RegressionReport with detected regressions

Example

detector = RegressionDetector() report = detector.detect_regressions(baseline_run, current_run) print(f"Found {len(report.regressions)} regressions") for regression in report.regressions: ... print(f" {regression.metric_name}: {regression.severity}")

Source code in uaef/experiments/regression.py
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
def detect_regressions(
    self,
    baseline_run: ExperimentRun,
    current_run: ExperimentRun,
    thresholds: Optional[Dict[str, float]] = None
) -> RegressionReport:
    """
    Detect regressions between baseline and current runs.

    Args:
        baseline_run: Baseline run for comparison
        current_run: Current run to check for regressions
        thresholds: Custom thresholds per metric (overrides defaults)

    Returns:
        RegressionReport with detected regressions

    Example:
        >>> detector = RegressionDetector()
        >>> report = detector.detect_regressions(baseline_run, current_run)
        >>> print(f"Found {len(report.regressions)} regressions")
        >>> for regression in report.regressions:
        ...     print(f"  {regression.metric_name}: {regression.severity}")
    """
    # Use custom thresholds if provided
    metric_thresholds = thresholds or self.thresholds

    # Get aggregate metrics from both runs
    baseline_metrics = baseline_run.aggregate_metrics
    current_metrics = current_run.aggregate_metrics

    # Find common metrics
    common_metrics = set(baseline_metrics.keys()) & set(current_metrics.keys())

    if not common_metrics:
        logger.warning(
            f"No common metrics found between runs {baseline_run.run_id} "
            f"and {current_run.run_id}"
        )

    # Detect regressions
    regressions: List[Regression] = []

    for metric_name in sorted(common_metrics):
        baseline_score = baseline_metrics[metric_name]
        current_score = current_metrics[metric_name]

        # Get threshold for this metric
        threshold = metric_thresholds.get(metric_name, self.minor_threshold)

        # Check if this is a regression
        regression = self._check_regression(
            metric_name=metric_name,
            baseline_score=baseline_score,
            current_score=current_score,
            threshold=threshold
        )

        if regression:
            regressions.append(regression)

    # Group regressions by severity
    regressions_by_severity = {
        "severe": [],
        "moderate": [],
        "minor": []
    }

    for regression in regressions:
        regressions_by_severity[regression.severity.value].append(
            regression.metric_name
        )

    # Group regressions by dimension
    regressions_by_dimension: Dict[str, List[str]] = {}

    for regression in regressions:
        dimension = regression.dimension or "unknown"
        if dimension not in regressions_by_dimension:
            regressions_by_dimension[dimension] = []
        regressions_by_dimension[dimension].append(regression.metric_name)

    # Generate summary
    summary = {
        "total_regressions": len(regressions),
        "severe_count": len(regressions_by_severity["severe"]),
        "moderate_count": len(regressions_by_severity["moderate"]),
        "minor_count": len(regressions_by_severity["minor"]),
        "dimensions_affected": len(regressions_by_dimension)
    }

    report = RegressionReport(
        baseline_run_id=baseline_run.run_id,
        current_run_id=current_run.run_id,
        regressions=regressions,
        regressions_by_severity=regressions_by_severity,
        regressions_by_dimension=regressions_by_dimension,
        summary=summary
    )

    logger.info(
        f"Detected {len(regressions)} regressions between runs "
        f"{baseline_run.run_id} and {current_run.run_id}: "
        f"{summary['severe_count']} severe, {summary['moderate_count']} moderate, "
        f"{summary['minor_count']} minor"
    )

    return report

get_regression_summary(report)

Generate a human-readable summary of a regression report.

Parameters:

Name Type Description Default
report RegressionReport

RegressionReport to summarize

required

Returns:

Type Description
str

Human-readable summary string

Example

summary = detector.get_regression_summary(report) print(summary)

Source code in uaef/experiments/regression.py
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
def get_regression_summary(
    self,
    report: RegressionReport
) -> str:
    """
    Generate a human-readable summary of a regression report.

    Args:
        report: RegressionReport to summarize

    Returns:
        Human-readable summary string

    Example:
        >>> summary = detector.get_regression_summary(report)
        >>> print(summary)
    """
    lines = [
        f"Regression Detection Report",
        f"===========================",
        f"Baseline Run: {report.baseline_run_id}",
        f"Current Run: {report.current_run_id}",
        f"",
        f"Summary:",
        f"  Total Regressions: {report.summary['total_regressions']}",
        f"  Severe: {report.summary['severe_count']}",
        f"  Moderate: {report.summary['moderate_count']}",
        f"  Minor: {report.summary['minor_count']}",
        f"  Dimensions Affected: {report.summary['dimensions_affected']}",
        f""
    ]

    if not report.regressions:
        lines.append("No regressions detected.")
        return "\n".join(lines)

    # Group by severity
    for severity in ["severe", "moderate", "minor"]:
        metric_names = report.regressions_by_severity.get(severity, [])
        if metric_names:
            lines.append(f"{severity.capitalize()} Regressions:")
            for metric_name in metric_names:
                regression = next(
                    r for r in report.regressions
                    if r.metric_name == metric_name
                )
                lines.append(
                    f"  - {metric_name}: {regression.baseline_score:.3f} → "
                    f"{regression.current_score:.3f} ({regression.percent_change:+.1f}%)"
                )
            lines.append("")

    # Group by dimension
    if report.regressions_by_dimension:
        lines.append("Regressions by Dimension:")
        for dimension, metric_names in sorted(report.regressions_by_dimension.items()):
            lines.append(f"  {dimension}: {', '.join(metric_names)}")
        lines.append("")

    return "\n".join(lines)

RegressionReport

Bases: BaseModel

Complete regression detection report.

Attributes:

Name Type Description
baseline_run_id UUID

ID of the baseline run

current_run_id UUID

ID of the current run

regressions List[Regression]

List of detected regressions

regressions_by_severity Dict[str, List[str]]

Regressions grouped by severity

regressions_by_dimension Dict[str, List[str]]

Regressions grouped by dimension

summary Dict[str, Any]

Summary statistics

Source code in uaef/experiments/regression.py
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
class RegressionReport(BaseModel):
    """
    Complete regression detection report.

    Attributes:
        baseline_run_id: ID of the baseline run
        current_run_id: ID of the current run
        regressions: List of detected regressions
        regressions_by_severity: Regressions grouped by severity
        regressions_by_dimension: Regressions grouped by dimension
        summary: Summary statistics
    """

    baseline_run_id: UUID = Field(..., description="ID of the baseline run")
    current_run_id: UUID = Field(..., description="ID of the current run")
    regressions: List[Regression] = Field(..., description="List of detected regressions")
    regressions_by_severity: Dict[str, List[str]] = Field(
        ...,
        description="Regressions grouped by severity"
    )
    regressions_by_dimension: Dict[str, List[str]] = Field(
        ...,
        description="Regressions grouped by dimension"
    )
    summary: Dict[str, Any] = Field(..., description="Summary statistics")

    class Config:
        """Pydantic configuration."""
        json_schema_extra = {
            "example": {
                "baseline_run_id": "550e8400-e29b-41d4-a716-446655440000",
                "current_run_id": "660e8400-e29b-41d4-a716-446655440001",
                "regressions": [],
                "regressions_by_severity": {
                    "severe": ["tool_accuracy"],
                    "moderate": ["answer_relevance"],
                    "minor": []
                },
                "regressions_by_dimension": {
                    "tool_calling": ["tool_accuracy"],
                    "response_quality": ["answer_relevance"]
                },
                "summary": {
                    "total_regressions": 2,
                    "severe_count": 1,
                    "moderate_count": 1,
                    "minor_count": 0
                }
            }
        }

Config

Pydantic configuration.

Source code in uaef/experiments/regression.py
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
class Config:
    """Pydantic configuration."""
    json_schema_extra = {
        "example": {
            "baseline_run_id": "550e8400-e29b-41d4-a716-446655440000",
            "current_run_id": "660e8400-e29b-41d4-a716-446655440001",
            "regressions": [],
            "regressions_by_severity": {
                "severe": ["tool_accuracy"],
                "moderate": ["answer_relevance"],
                "minor": []
            },
            "regressions_by_dimension": {
                "tool_calling": ["tool_accuracy"],
                "response_quality": ["answer_relevance"]
            },
            "summary": {
                "total_regressions": 2,
                "severe_count": 1,
                "moderate_count": 1,
                "minor_count": 0
            }
        }
    }

RegressionSeverity

Bases: str, Enum

Severity levels for regressions.

Source code in uaef/experiments/regression.py
18
19
20
21
22
23
class RegressionSeverity(str, Enum):
    """Severity levels for regressions."""

    MINOR = "minor"  # 5-10% degradation
    MODERATE = "moderate"  # 10-20% degradation
    SEVERE = "severe"  # >20% degradation