Domain
domain ¶
Domain layer for gepa-adk evolution engine.
This module exports the core domain models for the GEPA-ADK evolution engine.
| ATTRIBUTE | DESCRIPTION |
|---|---|
EvolutionConfig | Configuration parameters for evolution runs. TYPE: |
EvolutionResult | Outcome of a completed evolution run. TYPE: |
Candidate | Instruction candidate being evolved. TYPE: |
IterationRecord | Metrics for a single evolution iteration. TYPE: |
Score | Type alias for normalized scores. TYPE: |
ComponentName | Type alias for component identifiers. TYPE: |
ModelName | Type alias for model identifiers. TYPE: |
TrajectoryConfig | Configuration for trajectory extraction. TYPE: |
EvolutionError | Base exception for all gepa-adk errors. TYPE: |
ConfigurationError | Raised when configuration validation fails. TYPE: |
EvaluationError | Raised when batch evaluation fails. TYPE: |
AdapterError | Raised when adapter operations fail. TYPE: |
Examples:
Basic usage with configuration and records:
from gepa_adk.domain import EvolutionConfig, IterationRecord
config = EvolutionConfig(max_iterations=20)
record = IterationRecord(
iteration_number=1,
score=0.85,
component_text="Test",
evolved_component="instruction",
accepted=True,
)
See Also
gepa_adk.domain.models: Core dataclass implementations.gepa_adk.domain.types: Type aliases for domain concepts.gepa_adk.domain.exceptions: Exception hierarchy.
Note
This package contains pure domain logic with no external dependencies. All models follow hexagonal architecture principles (ADR-000).
COMPONENT_GENERATE_CONFIG module-attribute ¶
COMPONENT_GENERATE_CONFIG: ComponentName = (
"generate_content_config"
)
Component name for LLM generation configuration (temperature, top_p, etc.).
COMPONENT_INSTRUCTION module-attribute ¶
COMPONENT_INSTRUCTION: ComponentName = 'instruction'
Component name for agent instructions (same as DEFAULT_COMPONENT_NAME).
COMPONENT_OUTPUT_SCHEMA module-attribute ¶
COMPONENT_OUTPUT_SCHEMA: ComponentName = 'output_schema'
Component name for Pydantic output schema definitions.
DEFAULT_COMPONENT_NAME module-attribute ¶
DEFAULT_COMPONENT_NAME: ComponentName = 'instruction'
Default component name for single-component evolution.
This constant provides a single source of truth for the default component name used when evolving a single component (typically an agent's instruction). Use this constant instead of hardcoding 'instruction' throughout the codebase.
AncestorLog module-attribute ¶
Type alias for tracking attempted merges.
Represents a merge attempt triplet that has been tried, preventing duplicate merges.
Type
tuple[int, int, int]: (parent1_idx, parent2_idx, ancestor_idx)
Examples:
from gepa_adk.domain.types import AncestorLog
# Log of attempted merge
log: AncestorLog = (5, 8, 2) # (parent1_idx, parent2_idx, ancestor_idx)
Note
Type alias used by MergeProposer to track which merge combinations have already been attempted, preventing redundant merge operations.
ComponentName module-attribute ¶
Name of a candidate component (e.g., 'instruction', 'output_schema').
MergeAttempt module-attribute ¶
Type alias for merge attempt results.
Represents a successful merge attempt with the merged candidate and parent/ancestor indices, or None if merge was not possible.
Type:
tuple[Candidate, int, int, int] | None: (merged_candidate, parent1_idx,
parent2_idx, ancestor_idx) or None
Examples:
from gepa_adk.domain.types import MergeAttempt
from gepa_adk.domain.models import Candidate
# Successful merge
attempt: MergeAttempt = (
Candidate(components={"instruction": "..."}),
5, # parent1_idx
8, # parent2_idx
2, # ancestor_idx
)
# Failed merge
failed: MergeAttempt = None
Note
Type alias reserved for future merge reporting; currently unused but kept for parity with the merge-proposer design docs.
ModelName module-attribute ¶
Model identifier (e.g., 'gemini-2.5-flash', 'gpt-4o').
MultiAgentCandidate module-attribute ¶
Type alias for multi-agent candidate structure.
Maps qualified component names to their values using dot-separated format: {agent_name}.{component_name} as the key.
See ADR-012 for the addressing scheme rationale.
Examples:
Basic multi-agent candidate:
from gepa_adk.domain.types import MultiAgentCandidate, ComponentSpec
# Using ComponentSpec for type-safe construction
gen_inst = ComponentSpec(agent="generator", component="instruction")
critic_schema = ComponentSpec(agent="critic", component="output_schema")
candidate: MultiAgentCandidate = {
gen_inst.qualified: "Generate Python code...",
critic_schema.qualified: "Review code output schema...",
}
# Equivalent direct construction
candidate: MultiAgentCandidate = {
"generator.instruction": "Generate Python code...",
"critic.output_schema": "Review code output schema...",
}
Note
This type alias is compatible with GEPA's Candidate type (dict[str, str]), enabling seamless integration with existing mutation proposers and evolution engine components.
AdapterError ¶
Bases: EvaluationError
flowchart TD
gepa_adk.domain.AdapterError[AdapterError]
gepa_adk.domain.exceptions.EvaluationError[EvaluationError]
gepa_adk.domain.exceptions.EvolutionError[EvolutionError]
gepa_adk.domain.exceptions.EvaluationError --> gepa_adk.domain.AdapterError
gepa_adk.domain.exceptions.EvolutionError --> gepa_adk.domain.exceptions.EvaluationError
click gepa_adk.domain.AdapterError href "" "gepa_adk.domain.AdapterError"
click gepa_adk.domain.exceptions.EvaluationError href "" "gepa_adk.domain.exceptions.EvaluationError"
click gepa_adk.domain.exceptions.EvolutionError href "" "gepa_adk.domain.exceptions.EvolutionError"
Raised when an adapter operation fails.
This exception is used by adapter implementations (e.g., ADKAdapter) for adapter-specific failures such as session errors or configuration issues.
Examples:
Raising an adapter error:
from gepa_adk.domain.exceptions import AdapterError
if not self._session_service:
raise AdapterError(
"Session service unavailable",
adapter="ADKAdapter",
operation="evaluate",
)
Note
AdapterError is a subclass of EvaluationError, so callers can catch either for different granularity of error handling.
Source code in src/gepa_adk/domain/exceptions.py
ConfigurationError ¶
Bases: EvolutionError
flowchart TD
gepa_adk.domain.ConfigurationError[ConfigurationError]
gepa_adk.domain.exceptions.EvolutionError[EvolutionError]
gepa_adk.domain.exceptions.EvolutionError --> gepa_adk.domain.ConfigurationError
click gepa_adk.domain.ConfigurationError href "" "gepa_adk.domain.ConfigurationError"
click gepa_adk.domain.exceptions.EvolutionError href "" "gepa_adk.domain.exceptions.EvolutionError"
Raised when configuration validation fails.
This exception is raised during EvolutionConfig initialization when a parameter violates its validation constraints.
| ATTRIBUTE | DESCRIPTION |
|---|---|
field | The name of the configuration field that failed validation. TYPE: |
value | The invalid value that was provided. TYPE: |
constraint | Description of the validation constraint that was violated. TYPE: |
Examples:
Creating a configuration error with context:
from gepa_adk.domain.exceptions import ConfigurationError
error = ConfigurationError(
"max_iterations must be non-negative",
field="max_iterations",
value=-5,
constraint=">= 0",
)
print(error.field, error.value, error.constraint)
# Output: max_iterations -5 >= 0
Note
Arises from user-provided invalid settings, not programming errors. Should be caught and reported with clear guidance on valid values.
Source code in src/gepa_adk/domain/exceptions.py
__init__ ¶
__init__(
message: str,
*,
field: str | None = None,
value: object = None,
constraint: str | None = None,
) -> None
Initialize ConfigurationError with context.
| PARAMETER | DESCRIPTION |
|---|---|
message | Human-readable error description. TYPE: |
field | Name of the invalid configuration field. TYPE: |
value | The invalid value provided. TYPE: |
constraint | Description of the validation constraint. TYPE: |
Note
Context fields use keyword-only syntax to ensure explicit labeling and prevent positional argument mistakes.
Source code in src/gepa_adk/domain/exceptions.py
__str__ ¶
Return string representation with context.
| RETURNS | DESCRIPTION |
|---|---|
str | Formatted error message including field and value context. |
Note
Outputs formatted error message with field and value context when available, preserving base message structure.
Source code in src/gepa_adk/domain/exceptions.py
CriticOutputParseError ¶
Bases: ScoringError
flowchart TD
gepa_adk.domain.CriticOutputParseError[CriticOutputParseError]
gepa_adk.domain.exceptions.ScoringError[ScoringError]
gepa_adk.domain.exceptions.EvolutionError[EvolutionError]
gepa_adk.domain.exceptions.ScoringError --> gepa_adk.domain.CriticOutputParseError
gepa_adk.domain.exceptions.EvolutionError --> gepa_adk.domain.exceptions.ScoringError
click gepa_adk.domain.CriticOutputParseError href "" "gepa_adk.domain.CriticOutputParseError"
click gepa_adk.domain.exceptions.ScoringError href "" "gepa_adk.domain.exceptions.ScoringError"
click gepa_adk.domain.exceptions.EvolutionError href "" "gepa_adk.domain.exceptions.EvolutionError"
Raised when critic agent output cannot be parsed as valid JSON.
This exception indicates that the critic agent returned output that could not be parsed as JSON or did not conform to the expected schema.
| ATTRIBUTE | DESCRIPTION |
|---|---|
raw_output | The unparseable output from critic. TYPE: |
parse_error | Description of parsing failure. TYPE: |
Examples:
Handling parse errors:
from gepa_adk.domain.exceptions import CriticOutputParseError
try:
score, metadata = await scorer.async_score(...)
except CriticOutputParseError as e:
print(f"Invalid JSON: {e.raw_output}")
print(f"Error: {e.parse_error}")
Note
Arises when critic agent output cannot be parsed as valid JSON. Typically occurs when LLM output doesn't follow structured format despite output_schema being set.
Source code in src/gepa_adk/domain/exceptions.py
__init__ ¶
__init__(
message: str,
*,
raw_output: str,
parse_error: str,
cause: Exception | None = None,
) -> None
Initialize CriticOutputParseError with context.
| PARAMETER | DESCRIPTION |
|---|---|
message | Human-readable error description. TYPE: |
raw_output | The unparseable output string from critic. TYPE: |
parse_error | Description of the parsing failure. TYPE: |
cause | Original exception that caused this error. TYPE: |
Note
Context fields use keyword-only syntax to ensure explicit labeling and prevent positional argument mistakes.
Source code in src/gepa_adk/domain/exceptions.py
__str__ ¶
Return string with parse error details.
| RETURNS | DESCRIPTION |
|---|---|
str | Formatted message including parse error and raw output preview. |
Note
Outputs formatted error message with parse error and raw output preview (truncated to 100 chars), preserving base message structure.
Source code in src/gepa_adk/domain/exceptions.py
EvaluationError ¶
Bases: EvolutionError
flowchart TD
gepa_adk.domain.EvaluationError[EvaluationError]
gepa_adk.domain.exceptions.EvolutionError[EvolutionError]
gepa_adk.domain.exceptions.EvolutionError --> gepa_adk.domain.EvaluationError
click gepa_adk.domain.EvaluationError href "" "gepa_adk.domain.EvaluationError"
click gepa_adk.domain.exceptions.EvolutionError href "" "gepa_adk.domain.exceptions.EvolutionError"
Raised when batch evaluation fails.
This exception indicates failures during agent evaluation, such as agent execution errors, timeout, or malformed output.
| ATTRIBUTE | DESCRIPTION |
|---|---|
cause | Original exception that caused this error. TYPE: |
context | Additional context for debugging. TYPE: |
Examples:
Wrapping an ADK error:
from gepa_adk.domain.exceptions import EvaluationError
try:
result = await runner.run_async(...)
except ADKError as e:
raise EvaluationError(
"Agent execution failed",
cause=e,
agent_name="my_agent",
) from e
Note
Always preserves the original cause for debugging while providing a consistent interface for error handling.
Source code in src/gepa_adk/domain/exceptions.py
__init__ ¶
Initialize EvaluationError with cause and context.
| PARAMETER | DESCRIPTION |
|---|---|
message | Human-readable error description. TYPE: |
cause | Original exception that caused this error. TYPE: |
**context | Additional context for debugging (agent_name, etc.). TYPE: |
Note
Context is passed via keyword arguments. Positional arguments after message are not allowed.
Source code in src/gepa_adk/domain/exceptions.py
__str__ ¶
Return string with cause chain if present.
| RETURNS | DESCRIPTION |
|---|---|
str | Formatted message with context and cause information. |
Note
Outputs formatted error message with context dict and cause chain when available, preserving base message structure.
Source code in src/gepa_adk/domain/exceptions.py
EvolutionError ¶
Bases: Exception
flowchart TD
gepa_adk.domain.EvolutionError[EvolutionError]
click gepa_adk.domain.EvolutionError href "" "gepa_adk.domain.EvolutionError"
Base exception for all gepa-adk errors.
All custom exceptions in gepa-adk should inherit from this class to allow for unified exception handling.
Examples:
Catching evolution errors:
from gepa_adk.domain.exceptions import EvolutionError
try:
raise EvolutionError("Evolution failed unexpectedly")
except EvolutionError as e:
print(f"Caught: {e}")
# Output: Caught: Evolution failed unexpectedly
Note
Always use this base class or its subclasses for domain errors. Standard Python exceptions should still be raised for programming errors (e.g., TypeError, ValueError for developer mistakes).
Source code in src/gepa_adk/domain/exceptions.py
MissingScoreFieldError ¶
Bases: ScoringError
flowchart TD
gepa_adk.domain.MissingScoreFieldError[MissingScoreFieldError]
gepa_adk.domain.exceptions.ScoringError[ScoringError]
gepa_adk.domain.exceptions.EvolutionError[EvolutionError]
gepa_adk.domain.exceptions.ScoringError --> gepa_adk.domain.MissingScoreFieldError
gepa_adk.domain.exceptions.EvolutionError --> gepa_adk.domain.exceptions.ScoringError
click gepa_adk.domain.MissingScoreFieldError href "" "gepa_adk.domain.MissingScoreFieldError"
click gepa_adk.domain.exceptions.ScoringError href "" "gepa_adk.domain.exceptions.ScoringError"
click gepa_adk.domain.exceptions.EvolutionError href "" "gepa_adk.domain.exceptions.EvolutionError"
Raised when score field is missing or null in parsed output.
This exception indicates that the output was successfully parsed and validated, but the required score field is missing or null.
| ATTRIBUTE | DESCRIPTION |
|---|---|
parsed_output | The parsed output without valid score. TYPE: |
available_fields | Fields that were present. TYPE: |
Examples:
Handling missing score field:
from gepa_adk.domain.exceptions import MissingScoreFieldError
try:
score, metadata = await scorer.async_score(...)
except MissingScoreFieldError as e:
print(f"Missing score. Available fields: {e.available_fields}")
Note
Applies when parsed output lacks a valid score value. The parsed_output may contain other valid fields that will be preserved in metadata if score is found.
Source code in src/gepa_adk/domain/exceptions.py
__init__ ¶
Initialize MissingScoreFieldError with parsed output.
| PARAMETER | DESCRIPTION |
|---|---|
message | Human-readable error description. TYPE: |
parsed_output | The parsed dict without valid score field. TYPE: |
cause | Original exception that caused this error. TYPE: |
Note
Context fields use keyword-only syntax to ensure explicit labeling and prevent positional argument mistakes.
Source code in src/gepa_adk/domain/exceptions.py
__str__ ¶
Return string with available fields.
| RETURNS | DESCRIPTION |
|---|---|
str | Formatted message including list of available fields. |
Note
Outputs formatted error message with list of available fields from parsed output, preserving base message structure.
Source code in src/gepa_adk/domain/exceptions.py
MultiAgentValidationError ¶
Bases: EvolutionError
flowchart TD
gepa_adk.domain.MultiAgentValidationError[MultiAgentValidationError]
gepa_adk.domain.exceptions.EvolutionError[EvolutionError]
gepa_adk.domain.exceptions.EvolutionError --> gepa_adk.domain.MultiAgentValidationError
click gepa_adk.domain.MultiAgentValidationError href "" "gepa_adk.domain.MultiAgentValidationError"
click gepa_adk.domain.exceptions.EvolutionError href "" "gepa_adk.domain.exceptions.EvolutionError"
Raised when multi-agent configuration validation fails.
This exception is raised during MultiAgentAdapter initialization when a parameter violates its validation constraints.
| ATTRIBUTE | DESCRIPTION |
|---|---|
field | The name of the configuration field that failed validation. TYPE: |
value | The invalid value that was provided. TYPE: |
constraint | Description of the validation constraint that was violated. TYPE: |
Examples:
Creating a multi-agent validation error:
from gepa_adk.domain.exceptions import MultiAgentValidationError
error = MultiAgentValidationError(
"agents list cannot be empty",
field="agents",
value=[],
constraint="len >= 1",
)
print(error.field, error.value) # agents []
Note
Arises from user-provided invalid multi-agent settings, not programming errors. Should be caught and reported with clear guidance on valid values.
Source code in src/gepa_adk/domain/exceptions.py
__init__ ¶
Initialize MultiAgentValidationError with context.
| PARAMETER | DESCRIPTION |
|---|---|
message | Human-readable error description. TYPE: |
field | Name of the invalid configuration field. TYPE: |
value | The invalid value provided. TYPE: |
constraint | Description of the validation constraint. TYPE: |
Note
Context fields use keyword-only syntax to ensure explicit labeling and prevent positional argument mistakes.
Source code in src/gepa_adk/domain/exceptions.py
__str__ ¶
Return string representation with context.
| RETURNS | DESCRIPTION |
|---|---|
str | Formatted error message including field and value context. |
Note
Outputs formatted error message with field, value, and constraint context when available, preserving base message structure.
Source code in src/gepa_adk/domain/exceptions.py
NoCandidateAvailableError ¶
Bases: EvolutionError
flowchart TD
gepa_adk.domain.NoCandidateAvailableError[NoCandidateAvailableError]
gepa_adk.domain.exceptions.EvolutionError[EvolutionError]
gepa_adk.domain.exceptions.EvolutionError --> gepa_adk.domain.NoCandidateAvailableError
click gepa_adk.domain.NoCandidateAvailableError href "" "gepa_adk.domain.NoCandidateAvailableError"
click gepa_adk.domain.exceptions.EvolutionError href "" "gepa_adk.domain.exceptions.EvolutionError"
Raised when no candidates are available for selection.
| ATTRIBUTE | DESCRIPTION |
|---|---|
cause | Original exception that caused this error. TYPE: |
context | Extra context (candidate_idx, frontier_type). TYPE: |
Examples:
Note
Arises when candidate selector cannot find any valid candidates from the Pareto frontier, typically due to empty frontier or filtering constraints.
Source code in src/gepa_adk/domain/exceptions.py
__init__ ¶
Initialize NoCandidateAvailableError with context.
| PARAMETER | DESCRIPTION |
|---|---|
message | Human-readable error description. TYPE: |
cause | Original exception that caused this error. TYPE: |
**context | Additional context such as candidate_idx or frontier_type. TYPE: |
Note
Context fields use keyword-only syntax to ensure explicit labeling and prevent positional argument mistakes.
Source code in src/gepa_adk/domain/exceptions.py
__str__ ¶
Return string with context and cause details.
| RETURNS | DESCRIPTION |
|---|---|
str | Formatted error message including context and cause information. |
Note
Outputs formatted error message with context dict and cause chain when available, preserving base message structure.
Source code in src/gepa_adk/domain/exceptions.py
ScoringError ¶
Bases: EvolutionError
flowchart TD
gepa_adk.domain.ScoringError[ScoringError]
gepa_adk.domain.exceptions.EvolutionError[EvolutionError]
gepa_adk.domain.exceptions.EvolutionError --> gepa_adk.domain.ScoringError
click gepa_adk.domain.ScoringError href "" "gepa_adk.domain.ScoringError"
click gepa_adk.domain.exceptions.EvolutionError href "" "gepa_adk.domain.exceptions.EvolutionError"
Base exception for all scoring-related errors.
All scoring exceptions inherit from this class to allow for unified exception handling in scoring operations.
| ATTRIBUTE | DESCRIPTION |
|---|---|
cause | Original exception that caused this error. TYPE: |
Examples:
Catching scoring errors:
from gepa_adk.domain.exceptions import ScoringError
try:
score, metadata = await scorer.async_score(...)
except ScoringError as e:
print(f"Scoring failed: {e}")
Note
All scoring exceptions inherit from this base class. ScoringError extends EvolutionError, following ADR-009 exception hierarchy guidelines.
Source code in src/gepa_adk/domain/exceptions.py
__init__ ¶
Initialize ScoringError with message and optional cause.
| PARAMETER | DESCRIPTION |
|---|---|
message | Human-readable error description. TYPE: |
cause | Original exception that caused this error. TYPE: |
Note
Context fields use keyword-only syntax to ensure explicit labeling and prevent positional argument mistakes.
Source code in src/gepa_adk/domain/exceptions.py
__str__ ¶
Return string with cause chain if present.
| RETURNS | DESCRIPTION |
|---|---|
str | Formatted message with cause information. |
Note
Outputs formatted error message with cause chain when available, preserving base message structure.
Source code in src/gepa_adk/domain/exceptions.py
VideoValidationError ¶
Bases: ConfigurationError
flowchart TD
gepa_adk.domain.VideoValidationError[VideoValidationError]
gepa_adk.domain.exceptions.ConfigurationError[ConfigurationError]
gepa_adk.domain.exceptions.EvolutionError[EvolutionError]
gepa_adk.domain.exceptions.ConfigurationError --> gepa_adk.domain.VideoValidationError
gepa_adk.domain.exceptions.EvolutionError --> gepa_adk.domain.exceptions.ConfigurationError
click gepa_adk.domain.VideoValidationError href "" "gepa_adk.domain.VideoValidationError"
click gepa_adk.domain.exceptions.ConfigurationError href "" "gepa_adk.domain.exceptions.ConfigurationError"
click gepa_adk.domain.exceptions.EvolutionError href "" "gepa_adk.domain.exceptions.EvolutionError"
Raised when video file validation fails.
This exception is raised during video file processing when a video file does not exist, exceeds size limits, or has an invalid MIME type.
| ATTRIBUTE | DESCRIPTION |
|---|---|
video_path | The path to the video file that failed validation. TYPE: |
field | The configuration field name (default "video"). TYPE: |
constraint | Description of the validation constraint that was violated. TYPE: |
Examples:
Raising a video validation error:
from gepa_adk.domain.exceptions import VideoValidationError
raise VideoValidationError(
"Video file not found",
video_path="/path/to/missing.mp4",
constraint="file must exist",
)
Handling video validation errors:
from gepa_adk.domain.exceptions import VideoValidationError
try:
await video_service.prepare_video_parts(["/bad/path.mp4"])
except VideoValidationError as e:
print(f"Invalid video: {e.video_path}")
print(f"Constraint violated: {e.constraint}")
Note
Arises from video file validation failures during multimodal input processing. File existence, size limits (2GB), and MIME type (video/*) are validated before loading video content.
Source code in src/gepa_adk/domain/exceptions.py
751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 | |
__init__ ¶
Initialize VideoValidationError with video file context.
| PARAMETER | DESCRIPTION |
|---|---|
message | Human-readable error description. TYPE: |
video_path | The path to the video file that failed validation. TYPE: |
field | Name of the configuration field (default "video"). TYPE: |
constraint | Description of the validation constraint violated. TYPE: |
Note
Context fields use keyword-only syntax to ensure explicit labeling and prevent positional argument mistakes.
Source code in src/gepa_adk/domain/exceptions.py
__str__ ¶
Return string representation with video path context.
| RETURNS | DESCRIPTION |
|---|---|
str | Formatted error message including video_path and constraint. |
Note
Outputs formatted error message with video_path for easy identification of the problematic file in error logs.
Source code in src/gepa_adk/domain/exceptions.py
Candidate dataclass ¶
Represents an instruction candidate being evolved.
Unlike GEPA's simple dict[str, str] type alias, this class provides richer state tracking for async scenarios including lineage and metadata.
| ATTRIBUTE | DESCRIPTION |
|---|---|
components | Component name to text value mapping. Common keys include 'instruction' (main agent prompt) and 'output_schema'. TYPE: |
generation | Generation number in the evolution lineage (0 = initial). TYPE: |
parent_id | ID of the parent candidate for lineage tracking (legacy field, retained for compatibility). TYPE: |
parent_ids | Multi-parent indices for merge operations. None for seed candidates, [single_idx] for mutations, [idx1, idx2] for merges. TYPE: |
metadata | Extensible metadata dict for async tracking and debugging. TYPE: |
Examples:
Creating a candidate:
from gepa_adk.domain.models import Candidate
candidate = Candidate(
components={"instruction": "Be helpful"},
generation=0,
)
print(candidate.components["instruction"]) # Be helpful
print(candidate.generation) # 0
Note
A mutable candidate representation with richer state tracking than GEPA's simple dict. Components and metadata can be modified during the evolution process. Use generation and parent_id to track lineage.
Source code in src/gepa_adk/domain/models.py
EvolutionConfig dataclass ¶
Configuration parameters for an evolution run.
Defines the parameters that control how evolution proceeds, including iteration limits, concurrency settings, and stopping criteria.
| ATTRIBUTE | DESCRIPTION |
|---|---|
max_iterations | Maximum number of evolution iterations. 0 means just evaluate baseline without evolving. TYPE: |
max_concurrent_evals | Number of concurrent batch evaluations. Must be at least 1. TYPE: |
min_improvement_threshold | Minimum score improvement to accept a new candidate. Set to 0.0 to accept any improvement. TYPE: |
patience | Number of iterations without improvement before stopping early. Set to 0 to disable early stopping. TYPE: |
reflection_model | Model identifier for reflection/mutation operations. TYPE: |
frontier_type | Frontier tracking strategy for Pareto selection (default: INSTANCE). TYPE: |
acceptance_metric | Aggregation method for acceptance decisions on iteration evaluation batches. "sum" uses sum of scores (default, aligns with upstream GEPA). "mean" uses mean of scores (legacy behavior). TYPE: |
use_merge | Enable merge proposals for genetic crossover. Defaults to False. TYPE: |
max_merge_invocations | Maximum number of merge attempts per run. Defaults to 10. Must be non-negative. TYPE: |
reflection_prompt | Custom reflection/mutation prompt template. If provided, this template is used instead of the default when the reflection model proposes improved text. Required placeholders: - {component_text}: The current component text being evolved - {trials}: Trial data with feedback and trajectory for each test case If None or empty string, the default prompt template is used. TYPE: |
stop_callbacks | List of stopper callbacks for custom stop conditions. Each callback receives a StopperState and returns True to signal stop. Defaults to an empty list. TYPE: |
Examples:
Creating a configuration with defaults:
from gepa_adk.domain.models import EvolutionConfig
config = EvolutionConfig(max_iterations=100, patience=10)
print(config.max_iterations) # 100
print(config.reflection_model) # ollama_chat/gpt-oss:20b
Note
All numeric parameters are validated in post_init to ensure they meet their constraints. Invalid values raise ConfigurationError.
Source code in src/gepa_adk/domain/models.py
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 | |
__post_init__ ¶
Validate configuration parameters after initialization.
| RAISES | DESCRIPTION |
|---|---|
ConfigurationError | If any parameter violates its constraints. |
Note
Operates automatically after dataclass init completes. Validates all fields and raises ConfigurationError with context on failure.
Source code in src/gepa_adk/domain/models.py
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 | |
EvolutionResult dataclass ¶
Outcome of a completed evolution run.
Contains the final results after evolution completes, including all evolved component values, performance metrics, and full history.
| ATTRIBUTE | DESCRIPTION |
|---|---|
original_score | Starting performance score (baseline). TYPE: |
final_score | Ending performance score (best achieved). TYPE: |
evolved_components | Dictionary mapping component names to their final evolved text values. Keys include "instruction" and optionally "output_schema" or other components. Access individual components via TYPE: |
iteration_history | Chronological list of iteration records. TYPE: |
total_iterations | Number of iterations performed. TYPE: |
valset_score | Score on validation set used for acceptance decisions. None if no validation set was used. TYPE: |
trainset_score | Score on trainset used for reflection diagnostics. None if not computed. TYPE: |
objective_scores | Optional per-example multi-objective scores from the best candidate's final evaluation. None when no objective scores were tracked. Each dict maps objective name to score value. Index-aligned with evaluation batch examples. TYPE: |
Examples:
Creating and analyzing a result:
from gepa_adk.domain.models import EvolutionResult, IterationRecord
result = EvolutionResult(
original_score=0.60,
final_score=0.85,
evolved_components={"instruction": "Be helpful and concise"},
iteration_history=[],
total_iterations=10,
)
print(result.evolved_components["instruction"]) # "Be helpful and concise"
print(result.improvement) # 0.25
print(result.improved) # True
Note
As a frozen dataclass, EvolutionResult instances cannot be modified.
Source code in src/gepa_adk/domain/models.py
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 | |
improvement property ¶
Calculate the score improvement from original to final.
| RETURNS | DESCRIPTION |
|---|---|
float | The difference between final_score and original_score. |
float | Positive values indicate improvement, negative indicates degradation. |
Note
Override is not needed since frozen dataclasses support properties.
improved property ¶
Check if the final score is better than the original.
| RETURNS | DESCRIPTION |
|---|---|
bool | True if final_score > original_score, False otherwise. |
Note
Only returns True for strict improvement, not equal scores.
IterationRecord dataclass ¶
Captures metrics for a single evolution iteration.
This is an immutable record of what happened during one iteration of the evolution process. Records are created by the engine and stored in EvolutionResult.iteration_history.
| ATTRIBUTE | DESCRIPTION |
|---|---|
iteration_number | 1-indexed iteration number for human readability. TYPE: |
score | Score achieved in this iteration (typically in [0.0, 1.0]). TYPE: |
component_text | The component_text that was evaluated in this iteration (e.g., the instruction text for the "instruction" component). TYPE: |
evolved_component | The name of the component that was evolved in this iteration (e.g., "instruction", "output_schema"). Used for tracking which component changed in round-robin evolution strategies. TYPE: |
accepted | Whether this proposal was accepted as the new best. TYPE: |
objective_scores | Optional per-example multi-objective scores from the valset evaluation. None when adapter does not provide objective scores. Each dict maps objective name to score value. Index-aligned with evaluation batch examples. TYPE: |
Examples:
Creating an iteration record:
from gepa_adk.domain.models import IterationRecord
record = IterationRecord(
iteration_number=1,
score=0.85,
component_text="Be helpful",
evolved_component="instruction",
accepted=True,
)
print(record.score) # 0.85
print(record.evolved_component) # "instruction"
print(record.accepted) # True
Note
An immutable record that captures iteration metrics. Once created, IterationRecord instances cannot be modified, ensuring historical accuracy of the evolution trace.
Source code in src/gepa_adk/domain/models.py
MultiAgentEvolutionResult dataclass ¶
Outcome of a completed multi-agent evolution run.
Contains evolved component_text for all agents in the group, along with performance metrics and evolution history.
| ATTRIBUTE | DESCRIPTION |
|---|---|
evolved_components | Mapping of agent name to evolved component_text. TYPE: |
original_score | Starting performance score (baseline). TYPE: |
final_score | Ending performance score (best achieved). TYPE: |
primary_agent | Name of the agent whose output was used for scoring. TYPE: |
iteration_history | Chronological list of iteration records. TYPE: |
total_iterations | Number of iterations performed. TYPE: |
Examples:
Creating and analyzing a multi-agent result:
from gepa_adk.domain.models import MultiAgentEvolutionResult, IterationRecord
result = MultiAgentEvolutionResult(
evolved_components={
"generator": "Generate high-quality code",
"critic": "Review code thoroughly",
},
original_score=0.60,
final_score=0.85,
primary_agent="generator",
iteration_history=[],
total_iterations=10,
)
print(result.improvement) # 0.25
print(result.improved) # True
print(result.agent_names) # ["critic", "generator"]
Note
An immutable result container for multi-agent evolution. Once created, MultiAgentEvolutionResult instances cannot be modified. Use computed properties like improvement, improved, and agent_names to analyze results without modifying the underlying data.
Source code in src/gepa_adk/domain/models.py
449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 | |
improvement property ¶
Calculate the score improvement from original to final.
| RETURNS | DESCRIPTION |
|---|---|
float | The difference between final_score and original_score. |
float | Positive values indicate improvement, negative indicates degradation. |
Note
Override is not needed since frozen dataclasses support properties.
improved property ¶
Check if the final score is better than the original.
| RETURNS | DESCRIPTION |
|---|---|
bool | True if final_score > original_score, False otherwise. |
Note
Only returns True for strict improvement, not equal scores.
agent_names property ¶
Get sorted list of evolved agent names.
| RETURNS | DESCRIPTION |
|---|---|
list[str] | Sorted list of agent names from evolved_components keys. |
Note
Outputs a new list each time, sorted alphabetically for consistent ordering regardless of insertion order.
ParetoFrontier dataclass ¶
Tracks non-dominated candidates across multiple frontier dimensions.
| ATTRIBUTE | DESCRIPTION |
|---|---|
example_leaders | Instance-level: example_idx → leader candidate indices. TYPE: |
best_scores | Instance-level: example_idx → best score. TYPE: |
objective_leaders | Objective-level: objective_name → leader candidate indices. TYPE: |
objective_best_scores | Objective-level: objective_name → best score. TYPE: |
cartesian_leaders | Cartesian: (example_idx, objective) → leader candidate indices. TYPE: |
cartesian_best_scores | Cartesian: (example_idx, objective) → best score. TYPE: |
Note
A frontier stores the best candidate indices per dimension for sampling. The active dimension depends on frontier_type.
Examples:
Source code in src/gepa_adk/domain/state.py
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 | |
update ¶
update(
candidate_idx: int,
scores: dict[int, float],
*,
logger: FrontierLogger | None = None,
) -> None
Update frontier leadership with a candidate's scores.
| PARAMETER | DESCRIPTION |
|---|---|
candidate_idx | Index of the candidate being added. TYPE: |
scores | Mapping of example index to score. TYPE: |
logger | Optional structured logger for leader updates. TYPE: |
Note
Outputs updated leader sets and best scores for instance-level frontier tracking.
Examples:
Source code in src/gepa_adk/domain/state.py
get_non_dominated ¶
Return candidate indices that lead any example.
Note
Outputs the union of all leader sets across example indices.
Source code in src/gepa_adk/domain/state.py
get_selection_weights ¶
Return selection weights based on leadership frequency.
Note
Outputs weights proportional to how many examples each candidate leads, enabling weighted sampling.
Source code in src/gepa_adk/domain/state.py
update_objective ¶
update_objective(
candidate_idx: int,
objective_scores: dict[str, float],
*,
logger: FrontierLogger | None = None,
) -> None
Update objective-level frontier with a candidate's objective scores.
| PARAMETER | DESCRIPTION |
|---|---|
candidate_idx | Index of the candidate being added. TYPE: |
objective_scores | Mapping of objective name to score. TYPE: |
logger | Optional structured logger for leader updates. TYPE: |
Note
Outputs updated objective leader sets and best scores for objective-level frontier tracking.
Examples:
Source code in src/gepa_adk/domain/state.py
update_cartesian ¶
update_cartesian(
candidate_idx: int,
scores: dict[int, float],
objective_scores: dict[int, dict[str, float]],
*,
logger: FrontierLogger | None = None,
) -> None
Update cartesian frontier per (example, objective) pair.
| PARAMETER | DESCRIPTION |
|---|---|
candidate_idx | Index of the candidate being added. TYPE: |
scores | Mapping of example index to score. TYPE: |
objective_scores | Mapping of example index to objective scores dict. TYPE: |
logger | Optional structured logger for leader updates. TYPE: |
Note
Outputs updated cartesian leader sets and best scores for per (example, objective) pair frontier tracking.
Examples:
Source code in src/gepa_adk/domain/state.py
get_pareto_front_mapping ¶
get_pareto_front_mapping(
frontier_type: FrontierType,
) -> dict[FrontierKey, set[int]]
Return frontier mapping for specified frontier type.
| PARAMETER | DESCRIPTION |
|---|---|
frontier_type | Type of frontier to return mapping for. TYPE: |
| RETURNS | DESCRIPTION |
|---|---|
dict[FrontierKey, set[int]] | dict[FrontierKey, set[int]]: Mapping from frontier key to set of candidate indices. |
| RAISES | DESCRIPTION |
|---|---|
ValueError | If frontier_type is not a supported value. |
Note
Outputs a mapping with keys appropriate for the frontier type (int for INSTANCE, str for OBJECTIVE, tuples for HYBRID/CARTESIAN).
Examples:
mapping = frontier.get_pareto_front_mapping(FrontierType.INSTANCE)
# Returns: {0: {1, 2}, 1: {2, 3}}
Source code in src/gepa_adk/domain/state.py
ParetoState dataclass ¶
Tracks evolution state for Pareto-aware selection.
| ATTRIBUTE | DESCRIPTION |
|---|---|
candidates | Candidates discovered during evolution. TYPE: |
candidate_scores | Per-example scores. TYPE: |
frontier | Current frontier leader sets. TYPE: |
frontier_type | Frontier tracking strategy. TYPE: |
iteration | Current iteration number. TYPE: |
best_average_idx | Index of best-average candidate. TYPE: |
parent_indices | Genealogy map tracking parent relationships. Maps candidate_idx → [parent_idx, ...] or [None] for seeds. TYPE: |
Note
A single state object keeps frontier and candidate metrics aligned.
Examples:
Source code in src/gepa_adk/domain/state.py
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 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 | |
__post_init__ ¶
Validate state configuration and initialize averages.
Note
Checks candidate_scores indices are valid and marks frontier_type as initialized for immutability enforcement.
Source code in src/gepa_adk/domain/state.py
__setattr__ ¶
Enforce frontier_type immutability after initialization (T069).
Note
Only frontier_type is protected because it determines the frontier update routing logic in add_candidate(). Other fields (candidates, frontier, candidate_scores) are intentionally mutable to support evolution state updates. Using frozen=True would prevent all mutations, which is too restrictive for evolution state management.
Source code in src/gepa_adk/domain/state.py
add_candidate ¶
add_candidate(
candidate: Candidate,
scores: Sequence[Score],
*,
score_indices: Sequence[int] | None = None,
objective_scores: dict[str, float] | None = None,
per_example_objective_scores: dict[
int, dict[str, float]
]
| None = None,
parent_indices: list[int] | None = None,
logger: FrontierLogger | None = None,
) -> int
Add a candidate and update frontier tracking.
| PARAMETER | DESCRIPTION |
|---|---|
candidate | Candidate to add. TYPE: |
scores | Per-example scores for the candidate. TYPE: |
score_indices | Optional sequence mapping scores to example indices. If None, scores are assumed to be indexed 0, 1, 2, ... (full valset). If provided, scores[i] corresponds to example index score_indices[i]. TYPE: |
objective_scores | Optional aggregated objective scores (required for OBJECTIVE, HYBRID, CARTESIAN). TYPE: |
per_example_objective_scores | Optional per-example objective scores (required for CARTESIAN). TYPE: |
parent_indices | Optional parent candidate indices for genealogy tracking. If None, uses candidate.parent_ids if available, otherwise [None] for seed. TYPE: |
logger | Optional structured logger for frontier updates. TYPE: |
| RETURNS | DESCRIPTION |
|---|---|
int | Index of the newly added candidate. |
| RAISES | DESCRIPTION |
|---|---|
ConfigurationError | If objective_scores are required but not provided. |
Note
Outputs the new candidate index after routing to the appropriate frontier update method based on frontier_type.
Examples:
candidate_idx = state.add_candidate(candidate, [0.7, 0.8])
candidate_idx = state.add_candidate(
candidate, [0.7, 0.8], objective_scores={"accuracy": 0.9}
)
Source code in src/gepa_adk/domain/state.py
419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 | |
get_average_score ¶
Return average score for a candidate.
| PARAMETER | DESCRIPTION |
|---|---|
candidate_idx | Index of the candidate. TYPE: |
| RETURNS | DESCRIPTION |
|---|---|
float | Mean score across examples. |
| RAISES | DESCRIPTION |
|---|---|
NoCandidateAvailableError | If candidate scores are missing. |
Note
Outputs the arithmetic mean of all scores for the candidate across evaluated examples.
Examples:
Source code in src/gepa_adk/domain/state.py
update_best_average ¶
Update best_average_idx based on current scores.
Note
Outputs the candidate index with the highest mean score, or None if no candidates have scores.
Source code in src/gepa_adk/domain/state.py
StopperState dataclass ¶
Immutable snapshot of evolution state for stopper decisions.
Provides stoppers with read-only access to evolution metrics without exposing internal engine state.
| ATTRIBUTE | DESCRIPTION |
|---|---|
iteration | Current iteration number (0-indexed). TYPE: |
best_score | Best score achieved so far. TYPE: |
stagnation_counter | Number of iterations without improvement. TYPE: |
total_evaluations | Count of all evaluate() calls made. TYPE: |
candidates_count | Number of candidates in the frontier. TYPE: |
elapsed_seconds | Wall-clock time since evolution started. TYPE: |
Examples:
Creating a state snapshot:
from gepa_adk.domain.stopper import StopperState
state = StopperState(
iteration=10,
best_score=0.92,
stagnation_counter=3,
total_evaluations=100,
candidates_count=5,
elapsed_seconds=300.0,
)
print(state.best_score) # 0.92
print(state.stagnation_counter) # 3
Attempting to modify raises an error:
Note
A frozen dataclass, all fields are immutable after creation. Using slots=True for memory efficiency.
Source code in src/gepa_adk/domain/stopper.py
ADKTrajectory dataclass ¶
Execution trace from ADK agent evaluation.
Captures complete execution details from a single agent evaluation run, including all tool calls, state changes, token usage, and final output. This data enables debugging, optimization, and reflection-based learning.
| ATTRIBUTE | DESCRIPTION |
|---|---|
tool_calls | Immutable sequence of tool invocations during execution. TYPE: |
state_deltas | Sequence of state changes (session state updates). TYPE: |
token_usage | Optional token consumption metrics from LLM calls. TYPE: |
final_output | Final text response from the agent. TYPE: |
error | Error message if execution failed, None otherwise. TYPE: |
Examples:
trajectory = ADKTrajectory(
tool_calls=(ToolCallRecord("search", {"query": "AI"}, ["result1"], 0.1),),
state_deltas=({"search_count": 1},),
token_usage=TokenUsage(100, 50, 150),
final_output="Based on the search...",
error=None,
)
Note
All fields use immutable types (tuples, not lists) to prevent accidental modification of captured trace data.
Source code in src/gepa_adk/domain/trajectory.py
MultiAgentTrajectory dataclass ¶
Execution trace from multi-agent pipeline evaluation.
Captures individual agent trajectories and overall pipeline metrics.
| ATTRIBUTE | DESCRIPTION |
|---|---|
agent_trajectories | Mapping of agent name to trajectory. TYPE: |
pipeline_output | Final output from the primary agent. TYPE: |
total_token_usage | Aggregated token usage across all agents. TYPE: |
error | Error message if pipeline execution failed. TYPE: |
Examples:
Creating a multi-agent trajectory:
from gepa_adk.domain.trajectory import (
MultiAgentTrajectory,
ADKTrajectory,
TokenUsage,
)
trajectory = MultiAgentTrajectory(
agent_trajectories={
"generator": ADKTrajectory(...),
"critic": ADKTrajectory(...),
},
pipeline_output="Generated code output",
total_token_usage=TokenUsage(200, 100, 300),
error=None,
)
Note
All fields use immutable types to prevent accidental modification of captured trace data. agent_trajectories maps agent names to their individual ADKTrajectory records.
Source code in src/gepa_adk/domain/trajectory.py
TokenUsage dataclass ¶
Token usage statistics from LLM calls.
Tracks token consumption for monitoring costs and performance of language model interactions during agent execution.
| ATTRIBUTE | DESCRIPTION |
|---|---|
input_tokens | Number of tokens in the prompt/context. TYPE: |
output_tokens | Number of tokens generated in the response. TYPE: |
total_tokens | Sum of input_tokens and output_tokens. TYPE: |
Examples:
Source code in src/gepa_adk/domain/trajectory.py
ToolCallRecord dataclass ¶
Record of a single tool call during agent execution.
Captures the invocation details of a tool/function call made by an agent during evaluation, including arguments, results, and timing information.
| ATTRIBUTE | DESCRIPTION |
|---|---|
name | Tool or function name that was called. TYPE: |
arguments | Dictionary of arguments passed to the tool. TYPE: |
result | Return value from the tool execution. TYPE: |
timestamp | Relative time in seconds from evaluation start. TYPE: |
Examples:
record = ToolCallRecord(
name="get_weather",
arguments={"city": "Paris"},
result={"temp": 22, "condition": "sunny"},
timestamp=0.123,
)
Source code in src/gepa_adk/domain/trajectory.py
FrontierType ¶
Bases: str, Enum
flowchart TD
gepa_adk.domain.FrontierType[FrontierType]
click gepa_adk.domain.FrontierType href "" "gepa_adk.domain.FrontierType"
Supported frontier tracking strategies for Pareto selection.
Note
All four frontier types enable different Pareto dominance tracking strategies for multi-objective optimization.
Examples:
Source code in src/gepa_adk/domain/types.py
SchemaConstraints dataclass ¶
Constraints for output schema evolution.
Controls which fields must be preserved during schema evolution, including field existence and type constraints. Used by the OutputSchemaHandler to validate proposed schema mutations.
| ATTRIBUTE | DESCRIPTION |
|---|---|
required_fields | Field names that must exist in evolved schemas. Mutations removing these fields are rejected. TYPE: |
preserve_types | Mapping of field names to allowed type(s). Mutations changing a field's type to an incompatible type are rejected. TYPE: |
Examples:
Preserve required fields only:
from gepa_adk.domain.types import SchemaConstraints
constraints = SchemaConstraints(
required_fields=("score", "feedback"),
)
Preserve required fields with type constraints:
constraints = SchemaConstraints(
required_fields=("score",),
preserve_types={
"score": (float, int), # Allow numeric types
"order_id": str, # Must stay string
},
)
Note
A frozen dataclass ensures immutability during evolution runs. Configuration is validated at evolution start.
Source code in src/gepa_adk/domain/types.py
TrajectoryConfig dataclass ¶
Configuration for trajectory extraction behavior.
Controls which components are extracted from ADK event streams, whether sensitive data should be redacted, and whether large values should be truncated.
Note
All configuration fields are immutable after instantiation, ensuring consistent extraction behavior throughout evolution.
| ATTRIBUTE | DESCRIPTION |
|---|---|
include_tool_calls | Extract tool/function call records. Defaults to True. TYPE: |
include_state_deltas | Extract session state changes. Defaults to True. TYPE: |
include_token_usage | Extract LLM token consumption metrics. Defaults to True. TYPE: |
redact_sensitive | Apply sensitive data redaction. When True, fields matching sensitive_keys will be replaced with "[REDACTED]". Defaults to True for secure-by-default behavior. TYPE: |
sensitive_keys | Field names to redact via exact match. Case-sensitive. Defaults to ("password", "api_key", "token"). TYPE: |
max_string_length | Truncate strings longer than this with a marker indicating truncation. None disables truncation. Defaults to 10000 characters. TYPE: |
Examples:
Default configuration (secure, with truncation):
Minimal configuration (tool calls only):
config = TrajectoryConfig(
include_tool_calls=True,
include_state_deltas=False,
include_token_usage=False,
redact_sensitive=False,
)
Custom sensitive keys and truncation:
config = TrajectoryConfig(
sensitive_keys=("password", "api_key", "token", "ssn"),
max_string_length=5000, # Truncate DOM/screenshots earlier
)
Disable truncation (keep full values):
Note
Redaction takes precedence over truncation. Sensitive values are replaced with "[REDACTED]" first, then truncation applies to the remaining (non-sensitive) strings.