Evaluation

uaef.api.evaluate is the entry point most callers need. The evaluator classes underneath it are useful when you want direct control over batching or mode.

uaef.api

High-level Python API for UAEF.

This module provides simple, convenient functions for common UAEF workflows: - Simple evaluation with automatic adapter selection - Batch evaluation - Experiment management (create, run, compare) - Simulation (placeholder) - HITL workflows (review queue, feedback, agreement)

The API wraps existing UAEF modules to provide the simplest possible interface for common use cases, handling setup and configuration automatically.

Examples:

Simple evaluation: >>> from uaef.api import evaluate >>> result = evaluate( ... trace=langgraph_output, ... ground_truth=expected_output, ... adapter="langgraph" ... ) >>> print(f"Score: {result.overall_score:.2f}")

Experiment workflow: >>> from uaef.api import create_experiment, create_run, compare_runs >>> experiment = create_experiment( ... name="GPT-4 Agent", ... description="Testing tool calling" ... ) >>> run = create_run( ... experiment_id=experiment.experiment_id, ... run_name="baseline-v1", ... agent_design={"model": "gpt-4", "framework": "langgraph"} ... )

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

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

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
            }
        }
    }

AgentTrace

Bases: BaseModel

Canonical representation of an agent execution trace.

This is the standardized format that all framework-specific traces are transformed into for evaluation.

Attributes:

Name	Type	Description
trace_id	UUID	Unique identifier for this trace
session_id	Optional[str]	Identifier for the conversation session (optional)
messages	List[Message]	List of messages in the conversation
tool_calls	List[ToolCall]	List of all tool calls made during execution
input_tokens	Optional[int]	Number of input tokens consumed (optional)
output_tokens	Optional[int]	Number of output tokens generated (optional)
latency	Optional[float]	Total execution time in seconds (optional)
cost	Optional[float]	Estimated cost in USD (optional)
metadata	Dict[str, Any]	Additional metadata about the trace
framework	Optional[str]	Name of the agent framework used (optional)
timestamp	datetime	When the trace was created

Source code in uaef/models/agent_trace.py

class AgentTrace(BaseModel):
    """
    Canonical representation of an agent execution trace.

    This is the standardized format that all framework-specific traces
    are transformed into for evaluation.

    Attributes:
        trace_id: Unique identifier for this trace
        session_id: Identifier for the conversation session (optional)
        messages: List of messages in the conversation
        tool_calls: List of all tool calls made during execution
        input_tokens: Number of input tokens consumed (optional)
        output_tokens: Number of output tokens generated (optional)
        latency: Total execution time in seconds (optional)
        cost: Estimated cost in USD (optional)
        metadata: Additional metadata about the trace
        framework: Name of the agent framework used (optional)
        timestamp: When the trace was created
    """

    trace_id: UUID = Field(default_factory=uuid4, description="Unique identifier for this trace")
    session_id: Optional[str] = Field(None, description="Identifier for the conversation session")
    messages: List[Message] = Field(
        default_factory=list,
        description="List of messages in the conversation"
    )
    tool_calls: List[ToolCall] = Field(
        default_factory=list,
        description="List of all tool calls made during execution"
    )
    input_tokens: Optional[int] = Field(
        None,
        description="Number of input tokens consumed",
        ge=0
    )
    output_tokens: Optional[int] = Field(
        None,
        description="Number of output tokens generated",
        ge=0
    )
    latency: Optional[float] = Field(
        None,
        description="Total execution time in seconds",
        ge=0.0
    )
    cost: Optional[float] = Field(
        None,
        description="Estimated cost in USD",
        ge=0.0
    )
    metadata: Dict[str, Any] = Field(
        default_factory=dict,
        description="Additional metadata about the trace"
    )
    framework: Optional[str] = Field(None, description="Name of the agent framework used")
    timestamp: datetime = Field(
        default_factory=datetime.utcnow,
        description="When the trace was created"
    )

    @field_validator("trace_id")
    @classmethod
    def validate_trace_id(cls, v: UUID) -> UUID:
        """Validate that trace_id is a valid UUID."""
        if v is None:
            raise ValueError("trace_id cannot be None")
        return v

    @field_validator("timestamp")
    @classmethod
    def validate_timestamp(cls, v: datetime) -> datetime:
        """Validate timestamp format."""
        if v is None:
            raise ValueError("Timestamp cannot be None")
        return v

    @field_validator("input_tokens", "output_tokens")
    @classmethod
    def validate_token_counts(cls, v: Optional[int]) -> Optional[int]:
        """Validate that token counts are non-negative."""
        if v is not None and v < 0:
            raise ValueError("Token counts must be non-negative")
        return v

    @field_validator("latency", "cost")
    @classmethod
    def validate_positive_values(cls, v: Optional[float]) -> Optional[float]:
        """Validate that latency and cost are non-negative."""
        if v is not None and v < 0:
            raise ValueError("Latency and cost must be non-negative")
        return v

    class Config:
        """Pydantic configuration."""
        json_schema_extra = {
            "example": {
                "trace_id": "550e8400-e29b-41d4-a716-446655440000",
                "session_id": "session_123",
                "messages": [
                    {
                        "role": "user",
                        "content": "Find customer orders",
                        "timestamp": "2024-01-15T10:30:00Z"
                    }
                ],
                "tool_calls": [],
                "input_tokens": 150,
                "output_tokens": 75,
                "latency": 1.234,
                "cost": 0.0025,
                "metadata": {"model": "gpt-4"},
                "framework": "langgraph",
                "timestamp": "2024-01-15T10:30:00Z"
            }
        }

Config

Pydantic configuration.

Source code in uaef/models/agent_trace.py

class Config:
    """Pydantic configuration."""
    json_schema_extra = {
        "example": {
            "trace_id": "550e8400-e29b-41d4-a716-446655440000",
            "session_id": "session_123",
            "messages": [
                {
                    "role": "user",
                    "content": "Find customer orders",
                    "timestamp": "2024-01-15T10:30:00Z"
                }
            ],
            "tool_calls": [],
            "input_tokens": 150,
            "output_tokens": 75,
            "latency": 1.234,
            "cost": 0.0025,
            "metadata": {"model": "gpt-4"},
            "framework": "langgraph",
            "timestamp": "2024-01-15T10:30:00Z"
        }
    }

validate_positive_values(v) classmethod

Validate that latency and cost are non-negative.

Source code in uaef/models/agent_trace.py

@field_validator("latency", "cost")
@classmethod
def validate_positive_values(cls, v: Optional[float]) -> Optional[float]:
    """Validate that latency and cost are non-negative."""
    if v is not None and v < 0:
        raise ValueError("Latency and cost must be non-negative")
    return v

validate_timestamp(v) classmethod

Validate timestamp format.

Source code in uaef/models/agent_trace.py

@field_validator("timestamp")
@classmethod
def validate_timestamp(cls, v: datetime) -> datetime:
    """Validate timestamp format."""
    if v is None:
        raise ValueError("Timestamp cannot be None")
    return v

validate_token_counts(v) classmethod

Validate that token counts are non-negative.

Source code in uaef/models/agent_trace.py

@field_validator("input_tokens", "output_tokens")
@classmethod
def validate_token_counts(cls, v: Optional[int]) -> Optional[int]:
    """Validate that token counts are non-negative."""
    if v is not None and v < 0:
        raise ValueError("Token counts must be non-negative")
    return v

validate_trace_id(v) classmethod

Validate that trace_id is a valid UUID.

Source code in uaef/models/agent_trace.py

@field_validator("trace_id")
@classmethod
def validate_trace_id(cls, v: UUID) -> UUID:
    """Validate that trace_id is a valid UUID."""
    if v is None:
        raise ValueError("trace_id cannot be None")
    return v

AgreementMetrics

Bases: BaseModel

Metrics for measuring agreement between LLM judges and human reviewers.

This model tracks agreement statistics to calibrate LLM-based evaluation and identify areas where automated evaluation needs improvement.

Attributes:

Name	Type	Description
metric_name	str	Name of the metric being evaluated
llm_score	Optional[float]	Score from the LLM judge (None if not applicable)
human_scores	List[float]	List of scores from human reviewers
agreement_score	float	Correlation coefficient between LLM and human scores (0-1)
disagreement_details	Dict[str, Any]	Details about disagreements (score deltas, patterns, etc.)

Examples:

>>> # High agreement case
>>> metrics = AgreementMetrics(
...     metric_name="answer_relevance",
...     llm_score=0.85,
...     human_scores=[0.80, 0.85, 0.90],
...     agreement_score=0.95,
...     disagreement_details={
...         "mean_delta": 0.02,
...         "max_delta": 0.05,
...         "inter_rater_agreement": 0.92
...     }
... )

>>> # Low agreement case requiring calibration
>>> metrics = AgreementMetrics(
...     metric_name="safety_score",
...     llm_score=0.90,
...     human_scores=[0.50, 0.55, 0.60],
...     agreement_score=0.45,
...     disagreement_details={
...         "mean_delta": 0.35,
...         "max_delta": 0.40,
...         "pattern": "LLM consistently overestimates safety",
...         "inter_rater_agreement": 0.88
...     }
... )

Source code in uaef/hitl/models.py

class AgreementMetrics(BaseModel):
    """
    Metrics for measuring agreement between LLM judges and human reviewers.

    This model tracks agreement statistics to calibrate LLM-based evaluation
    and identify areas where automated evaluation needs improvement.

    Attributes:
        metric_name: Name of the metric being evaluated
        llm_score: Score from the LLM judge (None if not applicable)
        human_scores: List of scores from human reviewers
        agreement_score: Correlation coefficient between LLM and human scores (0-1)
        disagreement_details: Details about disagreements (score deltas, patterns, etc.)

    Examples:
        >>> # High agreement case
        >>> metrics = AgreementMetrics(
        ...     metric_name="answer_relevance",
        ...     llm_score=0.85,
        ...     human_scores=[0.80, 0.85, 0.90],
        ...     agreement_score=0.95,
        ...     disagreement_details={
        ...         "mean_delta": 0.02,
        ...         "max_delta": 0.05,
        ...         "inter_rater_agreement": 0.92
        ...     }
        ... )

        >>> # Low agreement case requiring calibration
        >>> metrics = AgreementMetrics(
        ...     metric_name="safety_score",
        ...     llm_score=0.90,
        ...     human_scores=[0.50, 0.55, 0.60],
        ...     agreement_score=0.45,
        ...     disagreement_details={
        ...         "mean_delta": 0.35,
        ...         "max_delta": 0.40,
        ...         "pattern": "LLM consistently overestimates safety",
        ...         "inter_rater_agreement": 0.88
        ...     }
        ... )
    """

    metric_name: str = Field(
        ...,
        description="Name of the metric being evaluated",
        min_length=1
    )
    llm_score: Optional[float] = Field(
        None,
        description="Score from the LLM judge",
        ge=0.0,
        le=1.0
    )
    human_scores: List[float] = Field(
        ...,
        description="List of scores from human reviewers"
    )
    agreement_score: float = Field(
        ...,
        description="Correlation coefficient between LLM and human scores",
        ge=0.0,
        le=1.0
    )
    disagreement_details: Dict[str, Any] = Field(
        default_factory=dict,
        description="Details about disagreements (score deltas, patterns, etc.)"
    )

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

    @field_validator("llm_score")
    @classmethod
    def validate_llm_score(cls, v: Optional[float]) -> Optional[float]:
        """Validate that llm_score is between 0 and 1 if provided."""
        if v is not None and not 0.0 <= v <= 1.0:
            raise ValueError(f"llm_score must be between 0 and 1, got {v}")
        return v

    @field_validator("human_scores")
    @classmethod
    def validate_human_scores(cls, v: List[float]) -> List[float]:
        """Validate that all human scores are between 0 and 1."""
        if not v:
            raise ValueError("human_scores cannot be empty")
        for i, score in enumerate(v):
            if not 0.0 <= score <= 1.0:
                raise ValueError(
                    f"Human score at index {i} must be between 0 and 1, got {score}"
                )
        return v

    @field_validator("agreement_score")
    @classmethod
    def validate_agreement_score(cls, v: float) -> float:
        """Validate that agreement_score is between 0 and 1."""
        if not 0.0 <= v <= 1.0:
            raise ValueError(f"agreement_score must be between 0 and 1, got {v}")
        return v

    class Config:
        """Pydantic configuration."""
        json_schema_extra = {
            "example": {
                "metric_name": "answer_relevance",
                "llm_score": 0.85,
                "human_scores": [0.80, 0.85, 0.90],
                "agreement_score": 0.95,
                "disagreement_details": {
                    "mean_delta": 0.02,
                    "max_delta": 0.05,
                    "std_delta": 0.03,
                    "inter_rater_agreement": 0.92,
                    "sample_size": 3,
                    "correlation_type": "pearson"
                }
            }
        }

Config

Pydantic configuration.

Source code in uaef/hitl/models.py

class Config:
    """Pydantic configuration."""
    json_schema_extra = {
        "example": {
            "metric_name": "answer_relevance",
            "llm_score": 0.85,
            "human_scores": [0.80, 0.85, 0.90],
            "agreement_score": 0.95,
            "disagreement_details": {
                "mean_delta": 0.02,
                "max_delta": 0.05,
                "std_delta": 0.03,
                "inter_rater_agreement": 0.92,
                "sample_size": 3,
                "correlation_type": "pearson"
            }
        }
    }

validate_agreement_score(v) classmethod

Validate that agreement_score is between 0 and 1.

Source code in uaef/hitl/models.py

@field_validator("agreement_score")
@classmethod
def validate_agreement_score(cls, v: float) -> float:
    """Validate that agreement_score is between 0 and 1."""
    if not 0.0 <= v <= 1.0:
        raise ValueError(f"agreement_score must be between 0 and 1, got {v}")
    return v

validate_human_scores(v) classmethod

Validate that all human scores are between 0 and 1.

Source code in uaef/hitl/models.py

@field_validator("human_scores")
@classmethod
def validate_human_scores(cls, v: List[float]) -> List[float]:
    """Validate that all human scores are between 0 and 1."""
    if not v:
        raise ValueError("human_scores cannot be empty")
    for i, score in enumerate(v):
        if not 0.0 <= score <= 1.0:
            raise ValueError(
                f"Human score at index {i} must be between 0 and 1, got {score}"
            )
    return v

validate_llm_score(v) classmethod

Validate that llm_score is between 0 and 1 if provided.

Source code in uaef/hitl/models.py

@field_validator("llm_score")
@classmethod
def validate_llm_score(cls, v: Optional[float]) -> Optional[float]:
    """Validate that llm_score is between 0 and 1 if provided."""
    if v is not None and not 0.0 <= v <= 1.0:
        raise ValueError(f"llm_score must be between 0 and 1, got {v}")
    return v

validate_metric_name(v) classmethod

Validate that metric_name is not empty.

Source code in uaef/hitl/models.py

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

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

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

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
            }
        }
    }

EvaluationResult

Bases: BaseModel

Complete evaluation result for an agent trace.

Attributes:

Name	Type	Description
evaluation_id	UUID	Unique identifier for this evaluation
trace_id	UUID	ID of the trace that was evaluated
dimension_results	List[DimensionResult]	List of results for each dimension
overall_score	float	Overall weighted score across all dimensions (0-1)
passed	bool	Whether the evaluation passed all thresholds
failures	List[str]	List of failure messages for threshold violations
warnings	List[str]	List of warning messages
metadata	Dict[str, Any]	Additional metadata about the evaluation
timestamp	datetime	When the evaluation was completed

Source code in uaef/models/evaluation_result.py

class EvaluationResult(BaseModel):
    """
    Complete evaluation result for an agent trace.

    Attributes:
        evaluation_id: Unique identifier for this evaluation
        trace_id: ID of the trace that was evaluated
        dimension_results: List of results for each dimension
        overall_score: Overall weighted score across all dimensions (0-1)
        passed: Whether the evaluation passed all thresholds
        failures: List of failure messages for threshold violations
        warnings: List of warning messages
        metadata: Additional metadata about the evaluation
        timestamp: When the evaluation was completed
    """

    evaluation_id: UUID = Field(
        default_factory=uuid4,
        description="Unique identifier for this evaluation"
    )
    trace_id: UUID = Field(..., description="ID of the trace that was evaluated")
    experiment_id: Optional[UUID] = Field(
        None,
        description="ID of the experiment this evaluation belongs to (set when persist=True)"
    )
    dimension_results: List[DimensionResult] = Field(
        ...,
        description="List of results for each dimension"
    )
    overall_score: float = Field(
        ...,
        description="Overall weighted score across all dimensions",
        ge=0.0,
        le=1.0
    )
    passed: bool = Field(..., description="Whether the evaluation passed all thresholds")
    failures: List[str] = Field(
        default_factory=list,
        description="List of failure messages for threshold violations"
    )
    warnings: List[str] = Field(
        default_factory=list,
        description="List of warning messages"
    )
    metadata: Dict[str, Any] = Field(
        default_factory=dict,
        description="Additional metadata about the evaluation"
    )
    timestamp: datetime = Field(
        default_factory=datetime.utcnow,
        description="When the evaluation was completed"
    )

    @field_validator("evaluation_id", "trace_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("overall_score")
    @classmethod
    def validate_score_bounds(cls, v: float) -> float:
        """Validate that overall score is between 0 and 1."""
        if not 0.0 <= v <= 1.0:
            raise ValueError(f"Overall score must be between 0 and 1, got {v}")
        return v

    class Config:
        """Pydantic configuration."""
        json_schema_extra = {
            "example": {
                "evaluation_id": "550e8400-e29b-41d4-a716-446655440000",
                "trace_id": "660e8400-e29b-41d4-a716-446655440001",
                "dimension_results": [
                    {
                        "dimension_name": "Tool Calling",
                        "metric_scores": [],
                        "aggregate_score": 0.85,
                        "weight": 0.4
                    }
                ],
                "overall_score": 0.82,
                "passed": True,
                "failures": [],
                "warnings": ["Ground truth not provided for some metrics"],
                "metadata": {"evaluation_mode": "offline"},
                "timestamp": "2024-01-15T10:30:00Z"
            }
        }

Config

Pydantic configuration.

Source code in uaef/models/evaluation_result.py

class Config:
    """Pydantic configuration."""
    json_schema_extra = {
        "example": {
            "evaluation_id": "550e8400-e29b-41d4-a716-446655440000",
            "trace_id": "660e8400-e29b-41d4-a716-446655440001",
            "dimension_results": [
                {
                    "dimension_name": "Tool Calling",
                    "metric_scores": [],
                    "aggregate_score": 0.85,
                    "weight": 0.4
                }
            ],
            "overall_score": 0.82,
            "passed": True,
            "failures": [],
            "warnings": ["Ground truth not provided for some metrics"],
            "metadata": {"evaluation_mode": "offline"},
            "timestamp": "2024-01-15T10:30:00Z"
        }
    }

validate_score_bounds(v) classmethod

Validate that overall score is between 0 and 1.

Source code in uaef/models/evaluation_result.py

@field_validator("overall_score")
@classmethod
def validate_score_bounds(cls, v: float) -> float:
    """Validate that overall score is between 0 and 1."""
    if not 0.0 <= v <= 1.0:
        raise ValueError(f"Overall score must be between 0 and 1, got {v}")
    return v

validate_uuid(v) classmethod

Validate that UUIDs are not None.

Source code in uaef/models/evaluation_result.py

@field_validator("evaluation_id", "trace_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

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

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

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

@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

@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

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

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

@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

@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

@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

@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

@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

GroundTruth

Bases: BaseModel

Expected correct outputs for validating agent responses.

Attributes:

Name	Type	Description
expected_output	Optional[str]	Expected text response from the agent (optional)
expected_tool_calls	List[ToolCall]	List of expected tool calls (optional)
expected_arguments	Dict[str, Any]	Dictionary of expected arguments for tools (optional)
context_documents	List[str]	List of context documents that should be used (optional)

Source code in uaef/models/ground_truth.py

class GroundTruth(BaseModel):
    """
    Expected correct outputs for validating agent responses.

    Attributes:
        expected_output: Expected text response from the agent (optional)
        expected_tool_calls: List of expected tool calls (optional)
        expected_arguments: Dictionary of expected arguments for tools (optional)
        context_documents: List of context documents that should be used (optional)
    """

    expected_output: Optional[str] = Field(
        None,
        description="Expected text response from the agent"
    )
    expected_tool_calls: List[ToolCall] = Field(
        default_factory=list,
        description="List of expected tool calls"
    )
    expected_arguments: Dict[str, Any] = Field(
        default_factory=dict,
        description="Dictionary of expected arguments for tools"
    )
    context_documents: List[str] = Field(
        default_factory=list,
        description="List of context documents that should be used"
    )

    class Config:
        """Pydantic configuration."""
        json_schema_extra = {
            "example": {
                "expected_output": "I found 5 customer orders matching your query.",
                "expected_tool_calls": [
                    {
                        "name": "search_database",
                        "arguments": {"query": "customer orders", "limit": 10},
                        "timestamp": "2024-01-15T10:30:00Z"
                    }
                ],
                "expected_arguments": {
                    "search_database": {"query": "customer orders"}
                },
                "context_documents": [
                    "Customer database contains order history...",
                    "Search API documentation..."
                ]
            }
        }

Config

Pydantic configuration.

Source code in uaef/models/ground_truth.py

class Config:
    """Pydantic configuration."""
    json_schema_extra = {
        "example": {
            "expected_output": "I found 5 customer orders matching your query.",
            "expected_tool_calls": [
                {
                    "name": "search_database",
                    "arguments": {"query": "customer orders", "limit": 10},
                    "timestamp": "2024-01-15T10:30:00Z"
                }
            ],
            "expected_arguments": {
                "search_database": {"query": "customer orders"}
            },
            "context_documents": [
                "Customer database contains order history...",
                "Search API documentation..."
            ]
        }
    }

ItemType

Bases: str, Enum

Type of item in the review queue.

Source code in uaef/hitl/models.py

class ItemType(str, Enum):
    """Type of item in the review queue."""

    LLM_EVALUATION = "llm_evaluation"
    AGENT_RESPONSE = "agent_response"
    SIMULATED_CONVERSATION = "simulated_conversation"

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

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

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
            }
        }
    }

ReviewItem

Bases: BaseModel

Item in the review queue for human validation.

This model supports three distinct HITL workflows: 1. LLM judge calibration workflow - validating LLM evaluation scores 2. Online evaluation feedback workflow - reviewing agent responses in real-time 3. Simulated conversation scoring workflow - scoring generated conversations

Attributes:

Name	Type	Description
item_id	UUID	Unique identifier for this review item
item_type	ItemType	Type of item (llm_evaluation, agent_response, simulated_conversation)
content	Dict[str, Any]	The content to be reviewed (question, response, conversation, etc.)
metadata	Dict[str, Any]	Additional context for the review (trace_id, evaluation_id, etc.)
priority	Priority	Priority level (low, medium, high, critical)
confidence_score	float	Confidence score from automated evaluation (0-1)
status	ReviewStatus	Current status (pending, in_review, completed)
created_at	datetime	When the item was added to the queue
reviewed_at	Optional[datetime]	When the review was completed (None if not yet reviewed)
reviewer_id	Optional[str]	ID of the human reviewer (None if not yet assigned)
human_scores	Dict[str, float]	Scores provided by human reviewers (metric_name -> score)
feedback	Optional[str]	Textual feedback from human reviewers

Examples:

>>> # LLM judge calibration item
>>> item = ReviewItem(
...     item_type=ItemType.LLM_EVALUATION,
...     content={
...         "question": "What is the capital of France?",
...         "agent_response": "The capital of France is Paris.",
...         "llm_scores": {"answer_relevance": 0.95}
...     },
...     metadata={"evaluation_id": "550e8400-e29b-41d4-a716-446655440000"},
...     priority=Priority.MEDIUM,
...     confidence_score=0.75
... )

>>> # Online evaluation feedback item
>>> item = ReviewItem(
...     item_type=ItemType.AGENT_RESPONSE,
...     content={
...         "user_query": "Book a flight to NYC",
...         "agent_response": "I've found 3 flights...",
...         "tools_called": ["search_flights", "get_prices"]
...     },
...     priority=Priority.HIGH,
...     confidence_score=0.65
... )

>>> # Simulated conversation scoring item
>>> item = ReviewItem(
...     item_type=ItemType.SIMULATED_CONVERSATION,
...     content={
...         "scenario": "Customer support - refund request",
...         "conversation": [
...             {"role": "user", "content": "I want a refund"},
...             {"role": "agent", "content": "I can help with that..."}
...         ],
...         "goal_achieved": True
...     },
...     priority=Priority.LOW,
...     confidence_score=0.85
... )

Source code in uaef/hitl/models.py

class ReviewItem(BaseModel):
    """
    Item in the review queue for human validation.

    This model supports three distinct HITL workflows:
    1. LLM judge calibration workflow - validating LLM evaluation scores
    2. Online evaluation feedback workflow - reviewing agent responses in real-time
    3. Simulated conversation scoring workflow - scoring generated conversations

    Attributes:
        item_id: Unique identifier for this review item
        item_type: Type of item (llm_evaluation, agent_response, simulated_conversation)
        content: The content to be reviewed (question, response, conversation, etc.)
        metadata: Additional context for the review (trace_id, evaluation_id, etc.)
        priority: Priority level (low, medium, high, critical)
        confidence_score: Confidence score from automated evaluation (0-1)
        status: Current status (pending, in_review, completed)
        created_at: When the item was added to the queue
        reviewed_at: When the review was completed (None if not yet reviewed)
        reviewer_id: ID of the human reviewer (None if not yet assigned)
        human_scores: Scores provided by human reviewers (metric_name -> score)
        feedback: Textual feedback from human reviewers

    Examples:
        >>> # LLM judge calibration item
        >>> item = ReviewItem(
        ...     item_type=ItemType.LLM_EVALUATION,
        ...     content={
        ...         "question": "What is the capital of France?",
        ...         "agent_response": "The capital of France is Paris.",
        ...         "llm_scores": {"answer_relevance": 0.95}
        ...     },
        ...     metadata={"evaluation_id": "550e8400-e29b-41d4-a716-446655440000"},
        ...     priority=Priority.MEDIUM,
        ...     confidence_score=0.75
        ... )

        >>> # Online evaluation feedback item
        >>> item = ReviewItem(
        ...     item_type=ItemType.AGENT_RESPONSE,
        ...     content={
        ...         "user_query": "Book a flight to NYC",
        ...         "agent_response": "I've found 3 flights...",
        ...         "tools_called": ["search_flights", "get_prices"]
        ...     },
        ...     priority=Priority.HIGH,
        ...     confidence_score=0.65
        ... )

        >>> # Simulated conversation scoring item
        >>> item = ReviewItem(
        ...     item_type=ItemType.SIMULATED_CONVERSATION,
        ...     content={
        ...         "scenario": "Customer support - refund request",
        ...         "conversation": [
        ...             {"role": "user", "content": "I want a refund"},
        ...             {"role": "agent", "content": "I can help with that..."}
        ...         ],
        ...         "goal_achieved": True
        ...     },
        ...     priority=Priority.LOW,
        ...     confidence_score=0.85
        ... )
    """

    item_id: UUID = Field(
        default_factory=uuid4,
        description="Unique identifier for this review item"
    )
    item_type: ItemType = Field(
        ...,
        description="Type of item (llm_evaluation, agent_response, simulated_conversation)"
    )
    content: Dict[str, Any] = Field(
        ...,
        description="The content to be reviewed"
    )
    metadata: Dict[str, Any] = Field(
        default_factory=dict,
        description="Additional context for the review"
    )
    priority: Priority = Field(
        ...,
        description="Priority level (low, medium, high, critical)"
    )
    confidence_score: float = Field(
        ...,
        description="Confidence score from automated evaluation",
        ge=0.0,
        le=1.0
    )
    status: ReviewStatus = Field(
        default=ReviewStatus.PENDING,
        description="Current status (pending, in_review, completed)"
    )
    created_at: datetime = Field(
        default_factory=datetime.utcnow,
        description="When the item was added to the queue"
    )
    reviewed_at: Optional[datetime] = Field(
        None,
        description="When the review was completed"
    )
    reviewer_id: Optional[str] = Field(
        None,
        description="ID of the human reviewer"
    )
    human_scores: Dict[str, float] = Field(
        default_factory=dict,
        description="Scores provided by human reviewers (metric_name -> score)"
    )
    feedback: Optional[str] = Field(
        None,
        description="Textual feedback from human reviewers"
    )

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

    @field_validator("content")
    @classmethod
    def validate_content(cls, v: Dict[str, Any]) -> Dict[str, Any]:
        """Validate that content is not empty."""
        if not v:
            raise ValueError("content cannot be empty")
        return v

    @field_validator("confidence_score")
    @classmethod
    def validate_confidence_score(cls, v: float) -> float:
        """Validate that confidence_score is between 0 and 1."""
        if not 0.0 <= v <= 1.0:
            raise ValueError(f"confidence_score must be between 0 and 1, got {v}")
        return v

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

    @field_validator("reviewed_at")
    @classmethod
    def validate_reviewed_at(cls, v: Optional[datetime], info) -> Optional[datetime]:
        """Validate that reviewed_at is set only when status is completed."""
        # Note: info.data contains the other field values
        status = info.data.get("status")
        if status == ReviewStatus.COMPLETED and v is None:
            raise ValueError("reviewed_at must be set when status is completed")
        return v

    @field_validator("reviewer_id")
    @classmethod
    def validate_reviewer_id(cls, v: Optional[str]) -> Optional[str]:
        """Validate that reviewer_id is not empty if provided."""
        if v is not None and not v.strip():
            raise ValueError("reviewer_id cannot be empty string")
        return v.strip() if v else None

    class Config:
        """Pydantic configuration."""
        json_schema_extra = {
            "example": {
                "item_id": "550e8400-e29b-41d4-a716-446655440000",
                "item_type": "llm_evaluation",
                "content": {
                    "question": "What is the capital of France?",
                    "agent_response": "The capital of France is Paris.",
                    "llm_scores": {"answer_relevance": 0.95, "answer_correctness": 0.98}
                },
                "metadata": {
                    "evaluation_id": "660e8400-e29b-41d4-a716-446655440001",
                    "trace_id": "770e8400-e29b-41d4-a716-446655440002"
                },
                "priority": "medium",
                "confidence_score": 0.75,
                "status": "completed",
                "created_at": "2024-01-15T10:00:00Z",
                "reviewed_at": "2024-01-15T10:30:00Z",
                "reviewer_id": "reviewer_123",
                "human_scores": {
                    "answer_relevance": 0.90,
                    "answer_correctness": 1.0
                },
                "feedback": "Response is accurate but could be more detailed"
            }
        }

Config

Pydantic configuration.

Source code in uaef/hitl/models.py

class Config:
    """Pydantic configuration."""
    json_schema_extra = {
        "example": {
            "item_id": "550e8400-e29b-41d4-a716-446655440000",
            "item_type": "llm_evaluation",
            "content": {
                "question": "What is the capital of France?",
                "agent_response": "The capital of France is Paris.",
                "llm_scores": {"answer_relevance": 0.95, "answer_correctness": 0.98}
            },
            "metadata": {
                "evaluation_id": "660e8400-e29b-41d4-a716-446655440001",
                "trace_id": "770e8400-e29b-41d4-a716-446655440002"
            },
            "priority": "medium",
            "confidence_score": 0.75,
            "status": "completed",
            "created_at": "2024-01-15T10:00:00Z",
            "reviewed_at": "2024-01-15T10:30:00Z",
            "reviewer_id": "reviewer_123",
            "human_scores": {
                "answer_relevance": 0.90,
                "answer_correctness": 1.0
            },
            "feedback": "Response is accurate but could be more detailed"
        }
    }

validate_confidence_score(v) classmethod

Validate that confidence_score is between 0 and 1.

Source code in uaef/hitl/models.py

@field_validator("confidence_score")
@classmethod
def validate_confidence_score(cls, v: float) -> float:
    """Validate that confidence_score is between 0 and 1."""
    if not 0.0 <= v <= 1.0:
        raise ValueError(f"confidence_score must be between 0 and 1, got {v}")
    return v

validate_content(v) classmethod

Validate that content is not empty.

Source code in uaef/hitl/models.py

@field_validator("content")
@classmethod
def validate_content(cls, v: Dict[str, Any]) -> Dict[str, Any]:
    """Validate that content is not empty."""
    if not v:
        raise ValueError("content cannot be empty")
    return v

validate_human_scores(v) classmethod

Validate that all human scores are between 0 and 1.

Source code in uaef/hitl/models.py

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

validate_item_id(v) classmethod

Validate that item_id is not None.

Source code in uaef/hitl/models.py

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

validate_reviewed_at(v, info) classmethod

Validate that reviewed_at is set only when status is completed.

Source code in uaef/hitl/models.py

@field_validator("reviewed_at")
@classmethod
def validate_reviewed_at(cls, v: Optional[datetime], info) -> Optional[datetime]:
    """Validate that reviewed_at is set only when status is completed."""
    # Note: info.data contains the other field values
    status = info.data.get("status")
    if status == ReviewStatus.COMPLETED and v is None:
        raise ValueError("reviewed_at must be set when status is completed")
    return v

validate_reviewer_id(v) classmethod

Validate that reviewer_id is not empty if provided.

Source code in uaef/hitl/models.py

@field_validator("reviewer_id")
@classmethod
def validate_reviewer_id(cls, v: Optional[str]) -> Optional[str]:
    """Validate that reviewer_id is not empty if provided."""
    if v is not None and not v.strip():
        raise ValueError("reviewer_id cannot be empty string")
    return v.strip() if v else None

add_to_review_queue(item_type, content, confidence_score, llm_scores=None, metadata=None)

Add an item to the human review queue.

Parameters:

Name	Type	Description	Default
item_type	Union[ItemType, str]	Type of item ("llm_evaluation", "agent_response", "simulated_conversation")	required
content	Dict[str, Any]	Item content (question, response, context, etc.)	required
confidence_score	float	Confidence score (0-1) for automatic priority assignment	required
llm_scores	Optional[Dict[str, float]]	Optional LLM-generated scores	None
metadata	Optional[Dict[str, Any]]	Optional additional metadata	None

Returns:

Type	Description
ReviewItem	Created ReviewItem with assigned priority

Examples:

Add LLM evaluation for review: >>> item = add_to_review_queue( ... item_type="llm_evaluation", ... content={ ... "question": "What is AI?", ... "response": "AI is artificial intelligence...", ... "context": ["AI definition document"] ... }, ... confidence_score=0.65, ... llm_scores={"answer_relevance": 0.75} ... ) >>> print(f"Priority: {item.priority}")

Add agent response for review: >>> item = add_to_review_queue( ... item_type="agent_response", ... content={"response": "Here's your answer..."}, ... confidence_score=0.45 # Low confidence -> high priority ... )

Source code in uaef/api/__init__.py

def add_to_review_queue(
    item_type: Union[ItemType, str],
    content: Dict[str, Any],
    confidence_score: float,
    llm_scores: Optional[Dict[str, float]] = None,
    metadata: Optional[Dict[str, Any]] = None
) -> ReviewItem:
    """
    Add an item to the human review queue.

    Args:
        item_type: Type of item ("llm_evaluation", "agent_response", "simulated_conversation")
        content: Item content (question, response, context, etc.)
        confidence_score: Confidence score (0-1) for automatic priority assignment
        llm_scores: Optional LLM-generated scores
        metadata: Optional additional metadata

    Returns:
        Created ReviewItem with assigned priority

    Examples:
        Add LLM evaluation for review:
            >>> item = add_to_review_queue(
            ...     item_type="llm_evaluation",
            ...     content={
            ...         "question": "What is AI?",
            ...         "response": "AI is artificial intelligence...",
            ...         "context": ["AI definition document"]
            ...     },
            ...     confidence_score=0.65,
            ...     llm_scores={"answer_relevance": 0.75}
            ... )
            >>> print(f"Priority: {item.priority}")

        Add agent response for review:
            >>> item = add_to_review_queue(
            ...     item_type="agent_response",
            ...     content={"response": "Here's your answer..."},
            ...     confidence_score=0.45  # Low confidence -> high priority
            ... )
    """
    logger.debug(f"Adding item to review queue: type={item_type}")

    # Convert string to ItemType enum if needed
    if isinstance(item_type, str):
        item_type = ItemType(item_type)

    # Create review item
    item = ReviewItem(
        item_type=item_type,
        content=content,
        confidence_score=confidence_score,
        llm_scores=llm_scores or {},
        metadata=metadata or {}
    )

    # Add to queue with automatic priority assignment
    return _review_queue.add_to_queue(item, auto_priority=True)

batch_evaluate(traces, ground_truths=None, adapter=None, metrics=None, metric_set=None, max_workers=4, persist=False, experiment_name=None, experiment_objective=None, created_by=None, **kwargs)

Evaluate multiple agent traces in batch mode with parallel processing.

Parameters:

Name	Type	Description	Default
traces	List[Union[AgentTrace, Dict[str, Any]]]	List of agent traces to evaluate	required
ground_truths	Optional[List[Union[GroundTruth, Dict[str, Any]]]]	List of expected correct outputs (optional)	None
adapter	Optional[str]	Framework adapter name.	None
metrics	Optional[List[str]]	Flat list of metric names. Dimensions resolved automatically.	None
metric_set	Optional[Dict[str, List[str]]]	Dict mapping dimension names to metric name lists.	None
max_workers	int	Maximum number of parallel workers	4
persist	bool	If True, creates a new experiment and persists all results.	False
experiment_name	Optional[str]	Name for the new experiment (defaults to "default").	None
experiment_objective	Optional[str]	Description/objective for the experiment.	None
created_by	Optional[str]	Security review H-02: identity (e.g. Cognito sub) of the caller, recorded on the persisted experiment when persist=True — see evaluate()'s docstring for details.	None
**kwargs		Additional evaluation parameters	{}

Returns:

Type	Description
List[EvaluationResult]	List of EvaluationResult objects (with experiment_id when persist=True)

Examples:

Batch evaluate and persist: >>> results = batch_evaluate( ... traces=[trace1, trace2, trace3], ... ground_truths=[gt1, gt2, gt3], ... persist=True, ... experiment_name="Batch Regression Test", ... )

Source code in uaef/api/__init__.py

def batch_evaluate(
    traces: List[Union[AgentTrace, Dict[str, Any]]],
    ground_truths: Optional[List[Union[GroundTruth, Dict[str, Any]]]] = None,
    adapter: Optional[str] = None,
    metrics: Optional[List[str]] = None,
    metric_set: Optional[Dict[str, List[str]]] = None,
    max_workers: int = 4,
    persist: bool = False,
    experiment_name: Optional[str] = None,
    experiment_objective: Optional[str] = None,
    created_by: Optional[str] = None,
    **kwargs
) -> List[EvaluationResult]:
    """
    Evaluate multiple agent traces in batch mode with parallel processing.

    Args:
        traces: List of agent traces to evaluate
        ground_truths: List of expected correct outputs (optional)
        adapter: Framework adapter name.
        metrics: Flat list of metric names. Dimensions resolved automatically.
        metric_set: Dict mapping dimension names to metric name lists.
        max_workers: Maximum number of parallel workers
        persist: If True, creates a new experiment and persists all results.
        experiment_name: Name for the new experiment (defaults to "default").
        experiment_objective: Description/objective for the experiment.
        created_by: Security review H-02: identity (e.g. Cognito ``sub``) of
            the caller, recorded on the persisted experiment when
            ``persist=True`` — see ``evaluate()``'s docstring for details.
        **kwargs: Additional evaluation parameters

    Returns:
        List of EvaluationResult objects (with experiment_id when persist=True)

    Examples:
        Batch evaluate and persist:
            >>> results = batch_evaluate(
            ...     traces=[trace1, trace2, trace3],
            ...     ground_truths=[gt1, gt2, gt3],
            ...     persist=True,
            ...     experiment_name="Batch Regression Test",
            ... )
    """
    logger.info(f"Starting batch evaluation of {len(traces)} traces")

    # Validate ground_truths length
    if ground_truths is not None and len(ground_truths) != len(traces):
        raise ValueError(
            f"ground_truths length ({len(ground_truths)}) must match "
            f"traces length ({len(traces)})"
        )

    # Transform traces if adapter is specified
    if adapter is not None:
        logger.debug(f"Using adapter: {adapter}")
        adapter_instance = get_adapter(adapter)
        traces = [adapter_instance.transform_to_canonical(t) for t in traces]

    # Transform ground_truths if needed
    if ground_truths is not None:
        ground_truths = [
            gt if isinstance(gt, GroundTruth) else GroundTruth(**gt)
            for gt in ground_truths
        ]

    # If flat metrics list provided, resolve to metric_set (dimension → metrics)
    if metrics is not None and metric_set is None:
        metric_set = _metrics_to_metric_set(metrics)
        logger.debug(f"Resolved {len(metrics)} metrics into {len(metric_set)} dimensions: {metric_set}")

    # Detect multi-turn session dicts in the input list. We route them through
    # SingleAgentEvaluator.evaluate_multi_turn() instead of the OfflineEvaluator.
    def _is_session_dict(t):
        return isinstance(t, dict) and "per_turn_traces" in t and "full_trace" in t

    has_session_dicts = any(_is_session_dict(t) for t in traces)

    if has_session_dicts:
        # Pick the right evaluator per session based on the full trace type so
        # MultiAgentTrace sessions go through MultiAgentEvaluator and AgentTrace
        # sessions go through SingleAgentEvaluator.
        results: List[EvaluationResult] = []
        for i, t in enumerate(traces):
            gt = ground_truths[i] if ground_truths else None
            if _is_session_dict(t):
                evaluator = _select_evaluator(t["full_trace"], metric_set)
                result = evaluator.evaluate_multi_turn(
                    per_turn_traces=t["per_turn_traces"],
                    full_trace=t["full_trace"],
                    ground_truth=gt,
                    metric_set=metric_set,
                    **kwargs,
                )
            else:
                evaluator = _select_evaluator(t, metric_set)
                result = evaluator.evaluate(
                    trace=t,
                    ground_truth=gt,
                    metric_set=metric_set,
                    **kwargs,
                )
            results.append(result)
    else:
        # All-AgentTrace path: use the parallel offline evaluator
        evaluator = OfflineEvaluator(max_workers=max_workers)
        results = evaluator.batch_evaluate(
            traces=traces,
            ground_truths=ground_truths,
            metric_set=metric_set,
            **kwargs
        )

    logger.info(f"Batch evaluation complete: {len(results)} results")

    # Persist if requested — always creates a new experiment
    if persist:
        from uuid import UUID as _UUID

        from uaef.utils.serialization import serialize_results

        exp_id, exp_name = _resolve_experiment(
            experiment_name=experiment_name,
            experiment_objective=experiment_objective,
        )
        for result in results:
            result.experiment_id = _UUID(exp_id)

        store = _get_store()
        result_dicts = serialize_results(results)
        # Security review M-02: opt-in redaction of the persisted copy only
        # (the `results` list returned below to the caller is untouched).
        result_dicts = _redact_result_dicts_for_persistence(result_dicts)

        store.save_experiment(
            experiment_id=exp_id,
            experiment_name=exp_name,
            evaluation_results=result_dicts,
            experiment_objective=experiment_objective,
            created_by=created_by,
        )
        logger.info(
            f"Persisted {len(results)} evaluations to new experiment={exp_id}"
        )

    return results

calculate_agreement(item_ids, include_inter_rater=True, include_llm_human=True)

Calculate agreement metrics for reviewed items.

Parameters:

Name	Type	Description	Default
item_ids	List[UUID]	List of item IDs to calculate agreement for	required
include_inter_rater	bool	Calculate inter-rater agreement between humans	True
include_llm_human	bool	Calculate LLM-human correlation	True

Returns:

Type	Description
AgreementMetrics	AgreementMetrics with correlation scores and statistics

Examples:

Calculate agreement: >>> metrics = calculate_agreement( ... item_ids=[item1.item_id, item2.item_id, item3.item_id], ... include_inter_rater=True, ... include_llm_human=True ... ) >>> print(f"LLM-Human correlation: {metrics.llm_human_correlation:.2f}") >>> print(f"Inter-rater agreement: {metrics.inter_rater_agreement:.2f}")

Source code in uaef/api/__init__.py

def calculate_agreement(
    item_ids: List[UUID],
    include_inter_rater: bool = True,
    include_llm_human: bool = True
) -> AgreementMetrics:
    """
    Calculate agreement metrics for reviewed items.

    Args:
        item_ids: List of item IDs to calculate agreement for
        include_inter_rater: Calculate inter-rater agreement between humans
        include_llm_human: Calculate LLM-human correlation

    Returns:
        AgreementMetrics with correlation scores and statistics

    Examples:
        Calculate agreement:
            >>> metrics = calculate_agreement(
            ...     item_ids=[item1.item_id, item2.item_id, item3.item_id],
            ...     include_inter_rater=True,
            ...     include_llm_human=True
            ... )
            >>> print(f"LLM-Human correlation: {metrics.llm_human_correlation:.2f}")
            >>> print(f"Inter-rater agreement: {metrics.inter_rater_agreement:.2f}")
    """
    logger.info(f"Calculating agreement for {len(item_ids)} items")

    # Get items from queue
    items = [_review_queue.get_item(item_id) for item_id in item_ids]

    # Calculate agreement
    return _agreement_calculator.calculate_agreement(
        items=items,
        include_inter_rater=include_inter_rater,
        include_llm_human=include_llm_human
    )

compare_runs(baseline_run_id, new_run_id, threshold=5.0)

Compare two experiment runs to identify regressions and improvements.

Parameters:

Name	Type	Description	Default
baseline_run_id	UUID	ID of the baseline run	required
new_run_id	UUID	ID of the new run to compare	required
threshold	float	Percentage threshold for detecting changes (default: 5.0%)	5.0

Returns:

Type	Description
ComparisonReport	ComparisonReport with detailed comparison results

Raises:

Type	Description
ValueError	If run IDs don't exist

Examples:

Compare two runs: >>> comparison = compare_runs( ... baseline_run_id=run1.run_id, ... new_run_id=run2.run_id, ... threshold=5.0 ... ) >>> print(f"Regressions: {comparison.regressions}") >>> print(f"Improvements: {comparison.improvements}") >>> >>> # Print detailed comparison >>> for mc in comparison.metric_comparisons: ... print(f"{mc.metric_name}: {mc.baseline_score:.2f} → {mc.current_score:.2f}")

Source code in uaef/api/__init__.py

def compare_runs(
    baseline_run_id: UUID,
    new_run_id: UUID,
    threshold: float = 5.0
) -> ComparisonReport:
    """
    Compare two experiment runs to identify regressions and improvements.

    Args:
        baseline_run_id: ID of the baseline run
        new_run_id: ID of the new run to compare
        threshold: Percentage threshold for detecting changes (default: 5.0%)

    Returns:
        ComparisonReport with detailed comparison results

    Raises:
        ValueError: If run IDs don't exist

    Examples:
        Compare two runs:
            >>> comparison = compare_runs(
            ...     baseline_run_id=run1.run_id,
            ...     new_run_id=run2.run_id,
            ...     threshold=5.0
            ... )
            >>> print(f"Regressions: {comparison.regressions}")
            >>> print(f"Improvements: {comparison.improvements}")
            >>> 
            >>> # Print detailed comparison
            >>> for mc in comparison.metric_comparisons:
            ...     print(f"{mc.metric_name}: {mc.baseline_score:.2f} → {mc.current_score:.2f}")
    """
    logger.info(f"Comparing runs: {baseline_run_id} vs {new_run_id}")

    # Get runs
    baseline_run = _experiment_manager.get_run(baseline_run_id)
    new_run = _experiment_manager.get_run(new_run_id)

    if baseline_run is None:
        raise ValueError(f"Baseline run {baseline_run_id} not found")
    if new_run is None:
        raise ValueError(f"New run {new_run_id} not found")

    # Compare
    return _comparison_engine.compare_runs(
        baseline_run=baseline_run,
        current_run=new_run,
        regression_threshold=threshold,
        improvement_threshold=threshold
    )

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

Create a new experiment for tracking agent evaluations.

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

Examples:

Create a simple experiment: >>> experiment = create_experiment( ... name="Customer Support Agent v2", ... description="Testing improved tool selection" ... ) >>> print(f"Experiment ID: {experiment.experiment_id}")

Create with metadata: >>> experiment = create_experiment( ... name="Production Agent", ... description="Final production version", ... metadata={"owner": "ml-team", "tags": ["production", "v2"]} ... )

Source code in uaef/api/__init__.py

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

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

    Returns:
        Created Experiment instance

    Examples:
        Create a simple experiment:
            >>> experiment = create_experiment(
            ...     name="Customer Support Agent v2",
            ...     description="Testing improved tool selection"
            ... )
            >>> print(f"Experiment ID: {experiment.experiment_id}")

        Create with metadata:
            >>> experiment = create_experiment(
            ...     name="Production Agent",
            ...     description="Final production version",
            ...     metadata={"owner": "ml-team", "tags": ["production", "v2"]}
            ... )
    """
    logger.info(f"Creating experiment: {name}")
    return _experiment_manager.create_experiment(
        name=name,
        description=description,
        metadata=metadata
    )

create_run(experiment_id, run_name, agent_design, config=None, 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	Union[AgentDesign, Dict[str, Any]]	Complete agent design configuration (AgentDesign or dict)	required
config	Optional[Dict[str, Any]]	Full configuration snapshot for reproducibility	None
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

Examples:

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

Create with AgentDesign object: >>> design = AgentDesign( ... framework="bedrock", ... model="us.anthropic.claude-sonnet-4-6", ... tools=["search"] ... ) >>> run = create_run( ... experiment_id=experiment.experiment_id, ... run_name="improved-v2", ... agent_design=design ... )

Source code in uaef/api/__init__.py

def create_run(
    experiment_id: UUID,
    run_name: str,
    agent_design: Union[AgentDesign, Dict[str, Any]],
    config: Optional[Dict[str, Any]] = None,
    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 (AgentDesign or dict)
        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

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

        Create with AgentDesign object:
            >>> design = AgentDesign(
            ...     framework="bedrock",
            ...     model="us.anthropic.claude-sonnet-4-6",
            ...     tools=["search"]
            ... )
            >>> run = create_run(
            ...     experiment_id=experiment.experiment_id,
            ...     run_name="improved-v2",
            ...     agent_design=design
            ... )
    """
    logger.info(f"Creating run: {run_name} for experiment {experiment_id}")

    # Convert dict to AgentDesign if needed
    if not isinstance(agent_design, AgentDesign):
        agent_design = AgentDesign(**agent_design)

    return _experiment_manager.create_run(
        experiment_id=experiment_id,
        run_name=run_name,
        agent_design=agent_design,
        config=config or {},
        git_commit=git_commit,
        metadata=metadata
    )

detect_regressions(baseline_run_id, new_run_id, threshold=5.0)

Detect regressions between two experiment runs.

Parameters:

Name	Type	Description	Default
baseline_run_id	UUID	ID of the baseline run	required
new_run_id	UUID	ID of the new run to check	required
threshold	float	Percentage threshold for regression detection (default: 5.0%)	5.0

Returns:

Type	Description
RegressionReport	RegressionReport with detected regressions and severity

Examples:

Detect regressions: >>> report = detect_regressions( ... baseline_run_id=run1.run_id, ... new_run_id=run2.run_id, ... threshold=5.0 ... ) >>> if report.has_regressions: ... print(f"Found {len(report.regressions)} regressions!") ... for reg in report.regressions: ... print(f" {reg.metric_name}: {reg.severity}")

Source code in uaef/api/__init__.py

def detect_regressions(
    baseline_run_id: UUID,
    new_run_id: UUID,
    threshold: float = 5.0
) -> RegressionReport:
    """
    Detect regressions between two experiment runs.

    Args:
        baseline_run_id: ID of the baseline run
        new_run_id: ID of the new run to check
        threshold: Percentage threshold for regression detection (default: 5.0%)

    Returns:
        RegressionReport with detected regressions and severity

    Examples:
        Detect regressions:
            >>> report = detect_regressions(
            ...     baseline_run_id=run1.run_id,
            ...     new_run_id=run2.run_id,
            ...     threshold=5.0
            ... )
            >>> if report.has_regressions:
            ...     print(f"Found {len(report.regressions)} regressions!")
            ...     for reg in report.regressions:
            ...         print(f"  {reg.metric_name}: {reg.severity}")
    """
    logger.info(f"Detecting regressions: {baseline_run_id} vs {new_run_id}")

    # Get runs
    baseline_run = _experiment_manager.get_run(baseline_run_id)
    new_run = _experiment_manager.get_run(new_run_id)

    if baseline_run is None:
        raise ValueError(f"Baseline run {baseline_run_id} not found")
    if new_run is None:
        raise ValueError(f"New run {new_run_id} not found")

    # Detect regressions
    return _regression_detector.detect_regressions(
        baseline_run=baseline_run,
        current_run=new_run,
        threshold=threshold
    )

evaluate(trace, ground_truth=None, adapter=None, metrics=None, metric_set=None, dimension_weights=None, thresholds=None, context=None, persist=False, experiment_name=None, experiment_objective=None, created_by=None, **kwargs)

Evaluate an agent trace with automatic evaluator selection.

Parameters:

Name	Type	Description	Default
trace	Union[AgentTrace, MultiAgentTrace, Dict[str, Any]]	Agent trace to evaluate (AgentTrace, MultiAgentTrace, or raw output)	required
ground_truth	Optional[Union[GroundTruth, Dict[str, Any]]]	Expected correct outputs (optional)	None
adapter	Optional[str]	Framework adapter name ("langgraph", "bedrock", etc.)	None
metrics	Optional[List[str]]	Flat list of metric names. Dimensions resolved automatically.	None
metric_set	Optional[Dict[str, List[str]]]	Dict mapping dimension names to metric name lists.	None
dimension_weights	Optional[Dict[str, float]]	Weights for each dimension (uses defaults if None)	None
thresholds	Optional[Dict[str, float]]	Threshold values for pass/fail (uses defaults if None)	None
context	Optional[List[str]]	List of context documents (optional)	None
persist	bool	If True, creates a new experiment and persists to DynamoDB + S3.	False
experiment_name	Optional[str]	Name for the new experiment (defaults to "default").	None
experiment_objective	Optional[str]	Description/objective for the experiment.	None
created_by	Optional[str]	Security review H-02: identity (e.g. Cognito sub) of the caller, recorded on the persisted experiment when persist=True so a service in front of the library (see uaef-service) can scope experiment listing/access to the owner. Ignored when persist=False. Optional and unused by direct library callers with no per-caller identity concept.	None
**kwargs		Additional evaluation parameters	{}

Returns:

Type	Description
EvaluationResult	EvaluationResult with experiment_id populated when persist=True.

Examples:

Ephemeral (no persistence): >>> result = evaluate(trace=trace, ground_truth=gt)

Persisted: >>> result = evaluate( ... trace=trace, ... persist=True, ... experiment_name="Customer Support Agent v2", ... experiment_objective="Measure quality after prompt changes", ... )

Source code in uaef/api/__init__.py

def evaluate(
    trace: Union[AgentTrace, MultiAgentTrace, Dict[str, Any]],
    ground_truth: Optional[Union[GroundTruth, Dict[str, Any]]] = None,
    adapter: Optional[str] = None,
    metrics: Optional[List[str]] = None,
    metric_set: Optional[Dict[str, List[str]]] = None,
    dimension_weights: Optional[Dict[str, float]] = None,
    thresholds: Optional[Dict[str, float]] = None,
    context: Optional[List[str]] = None,
    persist: bool = False,
    experiment_name: Optional[str] = None,
    experiment_objective: Optional[str] = None,
    created_by: Optional[str] = None,
    **kwargs
) -> EvaluationResult:
    """
    Evaluate an agent trace with automatic evaluator selection.

    Args:
        trace: Agent trace to evaluate (AgentTrace, MultiAgentTrace, or raw output)
        ground_truth: Expected correct outputs (optional)
        adapter: Framework adapter name ("langgraph", "bedrock", etc.)
        metrics: Flat list of metric names. Dimensions resolved automatically.
        metric_set: Dict mapping dimension names to metric name lists.
        dimension_weights: Weights for each dimension (uses defaults if None)
        thresholds: Threshold values for pass/fail (uses defaults if None)
        context: List of context documents (optional)
        persist: If True, creates a new experiment and persists to DynamoDB + S3.
        experiment_name: Name for the new experiment (defaults to "default").
        experiment_objective: Description/objective for the experiment.
        created_by: Security review H-02: identity (e.g. Cognito ``sub``) of
            the caller, recorded on the persisted experiment when
            ``persist=True`` so a service in front of the library (see
            ``uaef-service``) can scope experiment listing/access to the
            owner. Ignored when ``persist=False``. Optional and unused by
            direct library callers with no per-caller identity concept.
        **kwargs: Additional evaluation parameters

    Returns:
        EvaluationResult with experiment_id populated when persist=True.

    Examples:
        Ephemeral (no persistence):
            >>> result = evaluate(trace=trace, ground_truth=gt)

        Persisted:
            >>> result = evaluate(
            ...     trace=trace,
            ...     persist=True,
            ...     experiment_name="Customer Support Agent v2",
            ...     experiment_objective="Measure quality after prompt changes",
            ... )
    """
    logger.info("Starting evaluation")

    # Transform trace if adapter is specified
    if adapter is not None:
        logger.debug(f"Using adapter: {adapter}")
        adapter_instance = get_adapter(adapter)
        trace = adapter_instance.transform_to_canonical(trace)

    # Validate input type. evaluate() handles a single trace or a single multi-turn
    # session dict — multi-session lists must be passed to batch_evaluate() explicitly.
    if isinstance(trace, list):
        raise ValueError(
            "evaluate() does not accept a list of sessions. "
            "Use batch_evaluate() for multi-session input."
        )
    if not isinstance(trace, (AgentTrace, MultiAgentTrace, dict)):
        raise ValueError(
            "trace must be an AgentTrace, MultiAgentTrace, or session dict "
            "({'per_turn_traces': [...], 'full_trace': ...})"
        )

    # Transform ground_truth if it's a dict
    if ground_truth is not None and not isinstance(ground_truth, GroundTruth):
        ground_truth = GroundTruth(**ground_truth)

    # If flat metrics list provided, resolve to metric_set (dimension → metrics)
    if metrics is not None:
        metric_set = _metrics_to_metric_set(metrics)
        logger.debug(f"Resolved {len(metrics)} metrics into {len(metric_set)} dimensions: {metric_set}")

    # Multi-turn session dict: dispatch to the per-turn evaluator's evaluate_multi_turn
    if isinstance(trace, dict):
        if "per_turn_traces" not in trace or "full_trace" not in trace:
            raise ValueError(
                "Session dict must contain both 'per_turn_traces' and 'full_trace' keys"
            )
        evaluator = _select_evaluator(trace["full_trace"], metric_set)
        result = evaluator.evaluate_multi_turn(
            per_turn_traces=trace["per_turn_traces"],
            full_trace=trace["full_trace"],
            ground_truth=ground_truth,
            metric_set=metric_set,
            dimension_weights=dimension_weights,
            thresholds=thresholds,
            context=context,
            **kwargs,
        )
    else:
        # Single AgentTrace / MultiAgentTrace path
        evaluator = _select_evaluator(trace, metric_set)
        result = evaluator.evaluate(
            trace=trace,
            ground_truth=ground_truth,
            metric_set=metric_set,
            dimension_weights=dimension_weights,
            thresholds=thresholds,
            context=context,
            **kwargs
        )

    logger.info(f"Evaluation complete: score={result.overall_score:.3f}")

    # Persist if requested — always creates a new experiment
    if persist:
        from uuid import UUID as _UUID

        from uaef.utils.serialization import serialize_results

        exp_id, exp_name = _resolve_experiment(
            experiment_name=experiment_name,
            experiment_objective=experiment_objective,
        )
        result.experiment_id = _UUID(exp_id)

        store = _get_store()
        result_dicts = serialize_results([result])
        # Security review M-02: opt-in redaction of the persisted copy only
        # (the `result` object returned below to the caller is untouched).
        result_dicts = _redact_result_dicts_for_persistence(result_dicts)

        store.save_experiment(
            experiment_id=exp_id,
            experiment_name=exp_name,
            evaluation_results=result_dicts,
            experiment_objective=experiment_objective,
            created_by=created_by,
        )
        logger.info(f"Persisted evaluation to new experiment={exp_id}")

    return result

generate_conversation(scenario_config, user_model, agent_config, max_turns=10)

Generate a simulated conversation (PLACEHOLDER).

NOTE: This is a placeholder implementation. Full conversation simulation will be implemented in future work.

Parameters:

Name	Type	Description	Default
scenario_config	Dict[str, Any]	Scenario configuration (domain, goal, success criteria)	required
user_model	Dict[str, Any]	User behavior model (personality, patience, knowledge)	required
agent_config	Dict[str, Any]	Agent configuration (framework, model, tools)	required
max_turns	int	Maximum number of conversation turns	10

Returns:

Type	Description
Dict[str, Any]	Generated conversation with messages and metadata

Raises:

Type	Description
NotImplementedError	This feature is not yet implemented

Examples:

Generate conversation (future): >>> conversation = generate_conversation( ... scenario_config={ ... "domain": "customer_support", ... "user_goal": "Get refund", ... "success_criteria": ["Refund approved"] ... }, ... user_model={ ... "personality": "frustrated", ... "patience_level": 3 ... }, ... agent_config={ ... "framework": "langgraph", ... "model": "claude-3-sonnet" ... }, ... max_turns=10 ... )

Source code in uaef/api/__init__.py

def generate_conversation(
    scenario_config: Dict[str, Any],
    user_model: Dict[str, Any],
    agent_config: Dict[str, Any],
    max_turns: int = 10
) -> Dict[str, Any]:
    """
    Generate a simulated conversation (PLACEHOLDER).

    NOTE: This is a placeholder implementation. Full conversation simulation
    will be implemented in future work.

    Args:
        scenario_config: Scenario configuration (domain, goal, success criteria)
        user_model: User behavior model (personality, patience, knowledge)
        agent_config: Agent configuration (framework, model, tools)
        max_turns: Maximum number of conversation turns

    Returns:
        Generated conversation with messages and metadata

    Raises:
        NotImplementedError: This feature is not yet implemented

    Examples:
        Generate conversation (future):
            >>> conversation = generate_conversation(
            ...     scenario_config={
            ...         "domain": "customer_support",
            ...         "user_goal": "Get refund",
            ...         "success_criteria": ["Refund approved"]
            ...     },
            ...     user_model={
            ...         "personality": "frustrated",
            ...         "patience_level": 3
            ...     },
            ...     agent_config={
            ...         "framework": "langgraph",
            ...         "model": "claude-3-sonnet"
            ...     },
            ...     max_turns=10
            ... )
    """
    raise NotImplementedError(
        "Conversation simulation is not yet implemented. "
        "This is a placeholder for future functionality."
    )

get_full_metric_catalog()

Return every available metric name grouped by dimension/integration.

This is the single source of truth for the metric catalog presented by the demo backend and the UAEF service. It merges two registries:

• The built-in metric registry, grouped by dimension (e.g. "Response Quality", "Tool Calling"), via :func:list_metrics_by_dimension.
• The external integrations registry, grouped by integration name (e.g. "ragas", "deepeval"), via the integrations registry's list_all_metrics().

Integration groups are included only when their optional extra is installed. If an integration extra is not installed, that integration's group is omitted from the returned catalog without raising (Requirement 12.8). Callers must not hardcode metric-name lists (Requirement 12.6).

Returns:

Type	Description
Dict[str, List[str]]	Dict mapping each dimension or integration group name to a sorted
Dict[str, List[str]]	list of the metric names available in that group.

Source code in uaef/metrics/__init__.py

def get_full_metric_catalog() -> Dict[str, List[str]]:
    """Return every available metric name grouped by dimension/integration.

    This is the single source of truth for the metric catalog presented by
    the demo backend and the UAEF service. It merges two registries:

    - The built-in metric registry, grouped by dimension (e.g. "Response
      Quality", "Tool Calling"), via :func:`list_metrics_by_dimension`.
    - The external integrations registry, grouped by integration name (e.g.
      "ragas", "deepeval"), via the integrations registry's
      ``list_all_metrics()``.

    Integration groups are included only when their optional extra is
    installed. If an integration extra is not installed, that integration's
    group is omitted from the returned catalog without raising
    (Requirement 12.8). Callers must not hardcode metric-name lists
    (Requirement 12.6).

    Returns:
        Dict mapping each dimension or integration group name to a sorted
        list of the metric names available in that group.
    """
    catalog: Dict[str, List[str]] = {}

    # Built-in metrics grouped by dimension.
    for dimension, metric_names in list_metrics_by_dimension().items():
        catalog[dimension] = sorted(metric_names)

    # Integration metrics grouped by integration name. The integrations
    # registry only reports connectors whose backing library is installed, so
    # groups for uninstalled extras are naturally omitted. We additionally
    # guard the import so a missing integrations subpackage does not raise.
    #
    # De-duplication: the built-in metric registry is the authoritative source.

    try:
        from uaef.integrations.registry import get_registry as _get_integrations_registry

        _existing_upper = {k.upper() for k in catalog}
        integrations_registry = _get_integrations_registry()
        for integration_name, metric_names in integrations_registry.list_all_metrics().items():
            # Skip if a built-in dimension already covers this integration.
            if integration_name.upper() in _existing_upper:
                continue
            catalog[integration_name] = sorted(metric_names)
    except ImportError:
        # Integrations subpackage or its extras are unavailable; omit
        # integration groups rather than failing the catalog.
        pass

    return catalog

get_run_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

Examples:

Get evaluations for a run: >>> evaluations = get_run_evaluations(run.run_id) >>> print(f"Found {len(evaluations)} evaluations")

Source code in uaef/api/__init__.py

def get_run_evaluations(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

    Examples:
        Get evaluations for a run:
            >>> evaluations = get_run_evaluations(run.run_id)
            >>> print(f"Found {len(evaluations)} evaluations")
    """
    logger.debug(f"Getting evaluations for run {run_id}")
    return _experiment_manager.get_evaluations(run_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

Examples:

Record evaluation result: >>> result = evaluate(trace, ground_truth, adapter="langgraph") >>> record_evaluation(run.run_id, result)

Source code in uaef/api/__init__.py

def record_evaluation(
    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

    Examples:
        Record evaluation result:
            >>> result = evaluate(trace, ground_truth, adapter="langgraph")
            >>> record_evaluation(run.run_id, result)
    """
    logger.debug(f"Recording evaluation for run {run_id}")
    _experiment_manager.record_evaluation(run_id, evaluation_result)

run_scenario(scenario_id, agent_config, parameters=None)

Run a predefined scenario (PLACEHOLDER).

NOTE: This is a placeholder implementation. Full scenario execution will be implemented in future work.

Parameters:

Name	Type	Description	Default
scenario_id	str	ID of the predefined scenario	required
agent_config	Dict[str, Any]	Agent configuration	required
parameters	Optional[Dict[str, Any]]	Optional scenario parameters	None

Returns:

Type	Description
Dict[str, Any]	Scenario execution results

Raises:

Type	Description
NotImplementedError	This feature is not yet implemented

Examples:

Run scenario (future): >>> result = run_scenario( ... scenario_id="customer_support_refund", ... agent_config={"framework": "langgraph", "model": "gpt-4"} ... )

Source code in uaef/api/__init__.py

def run_scenario(
    scenario_id: str,
    agent_config: Dict[str, Any],
    parameters: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
    """
    Run a predefined scenario (PLACEHOLDER).

    NOTE: This is a placeholder implementation. Full scenario execution
    will be implemented in future work.

    Args:
        scenario_id: ID of the predefined scenario
        agent_config: Agent configuration
        parameters: Optional scenario parameters

    Returns:
        Scenario execution results

    Raises:
        NotImplementedError: This feature is not yet implemented

    Examples:
        Run scenario (future):
            >>> result = run_scenario(
            ...     scenario_id="customer_support_refund",
            ...     agent_config={"framework": "langgraph", "model": "gpt-4"}
            ... )
    """
    raise NotImplementedError(
        "Scenario execution is not yet implemented. "
        "This is a placeholder for future functionality."
    )

submit_review(item_id, human_scores, reviewer_id, feedback=None)

Submit a human review for an item in the queue.

Parameters:

Name	Type	Description	Default
item_id	UUID	UUID of the item to review	required
human_scores	Dict[str, float]	Dictionary of metric_name -> score (0-1)	required
reviewer_id	str	ID of the human reviewer	required
feedback	Optional[str]	Optional textual feedback	None

Returns:

Type	Description
ReviewItem	Updated ReviewItem with completed status

Raises:

Type	Description
ValueError	If item not found or already completed

Examples:

Submit review: >>> reviewed = submit_review( ... item_id=item.item_id, ... human_scores={ ... "answer_relevance": 0.85, ... "answer_correctness": 0.90 ... }, ... reviewer_id="reviewer_123", ... feedback="Good response but could be more detailed" ... ) >>> print(f"Status: {reviewed.status}")

Source code in uaef/api/__init__.py

def submit_review(
    item_id: UUID,
    human_scores: Dict[str, float],
    reviewer_id: str,
    feedback: Optional[str] = None
) -> ReviewItem:
    """
    Submit a human review for an item in the queue.

    Args:
        item_id: UUID of the item to review
        human_scores: Dictionary of metric_name -> score (0-1)
        reviewer_id: ID of the human reviewer
        feedback: Optional textual feedback

    Returns:
        Updated ReviewItem with completed status

    Raises:
        ValueError: If item not found or already completed

    Examples:
        Submit review:
            >>> reviewed = submit_review(
            ...     item_id=item.item_id,
            ...     human_scores={
            ...         "answer_relevance": 0.85,
            ...         "answer_correctness": 0.90
            ...     },
            ...     reviewer_id="reviewer_123",
            ...     feedback="Good response but could be more detailed"
            ... )
            >>> print(f"Status: {reviewed.status}")
    """
    logger.debug(f"Submitting review for item {item_id}")
    return _review_queue.submit_review(
        item_id=item_id,
        human_scores=human_scores,
        reviewer_id=reviewer_id,
        feedback=feedback
    )

uaef.evaluation.base_evaluator

Base evaluator interface for UAEF.

This module provides the abstract base class for all evaluators, defining the core evaluation interface and common functionality.

BaseEvaluator

Bases: ABC

Abstract base class for all evaluators.

Provides common functionality for: - Single evaluation - Batch evaluation - Parallel metric calculation - Async support for LLM-based metrics

Source code in uaef/evaluation/base_evaluator.py

class BaseEvaluator(ABC):
    """
    Abstract base class for all evaluators.

    Provides common functionality for:
    - Single evaluation
    - Batch evaluation
    - Parallel metric calculation
    - Async support for LLM-based metrics
    """

    def __init__(
        self,
        metric_registry: Optional[any] = None,
        max_workers: int = 4
    ):
        """
        Initialize base evaluator.

        Args:
            metric_registry: MetricRegistry instance (uses global if None)
            max_workers: Maximum number of parallel workers for metric calculation
        """
        self.metric_registry = metric_registry or get_registry()
        self.max_workers = max_workers
        self._warned_dimension_keys: set = set()

        logger.info(
            f"Initialized {self.__class__.__name__} with max_workers={max_workers}"
        )

    @abstractmethod
    def evaluate(
        self,
        trace: any,
        ground_truth: Optional[GroundTruth] = None,
        metric_set: Optional[Dict[str, List[str]]] = None,
        **kwargs
    ) -> EvaluationResult:
        """
        Evaluate a single trace.

        Args:
            trace: Agent trace to evaluate
            ground_truth: Expected correct outputs (optional)
            metric_set: Dictionary mapping dimension names to lists of metric names
            **kwargs: Additional evaluation parameters

        Returns:
            EvaluationResult with scores and metadata

        Raises:
            ValueError: If inputs are invalid
            RuntimeError: If evaluation fails
        """
        pass

    def batch_evaluate(
        self,
        traces: List[any],
        ground_truths: Optional[List[Optional[GroundTruth]]] = None,
        metric_set: Optional[Dict[str, List[str]]] = None,
        **kwargs
    ) -> List[EvaluationResult]:
        """
        Evaluate multiple traces in batch.

        Args:
            traces: List of agent traces to evaluate
            ground_truths: List of ground truths (optional, can be None for each)
            metric_set: Dictionary mapping dimension names to lists of metric names
            **kwargs: Additional evaluation parameters

        Returns:
            List of EvaluationResult objects

        Raises:
            ValueError: If inputs are invalid
        """
        logger.info(f"Starting batch evaluation for {len(traces)} traces")

        # Validate inputs
        if not traces:
            raise ValueError("traces cannot be empty")

        if ground_truths is None:
            ground_truths = [None] * len(traces)

        if len(traces) != len(ground_truths):
            raise ValueError(
                f"Length mismatch: {len(traces)} traces but {len(ground_truths)} ground truths"
            )

        # Evaluate each trace
        results = []
        for i, (trace, gt) in enumerate(zip(traces, ground_truths)):
            try:
                logger.debug(f"Evaluating trace {i+1}/{len(traces)}")
                result = self.evaluate(
                    trace=trace,
                    ground_truth=gt,
                    metric_set=metric_set,
                    **kwargs
                )
                results.append(result)

            except Exception as e:
                logger.error(f"Failed to evaluate trace {i+1}: {str(e)}")
                # Create a failed evaluation result
                results.append(self._create_failed_result(trace, str(e)))

        success_count = sum(1 for r in results if r.passed)
        logger.info(
            f"Batch evaluation complete: {success_count}/{len(results)} passed"
        )

        return results

    def calculate_metrics_with_dependencies(
        self,
        metrics: List[BaseMetric],
        evaluation_input: EvaluationInput
    ) -> List[MetricScore]:
        """
        Calculate metrics respecting dependencies.

        Simple two-phase approach:
        - Phase 1: Calculate metrics with no dependencies
        - Phase 2: Calculate metrics with dependencies (which can now read metadata)

        Args:
            metrics: List of metric instances to calculate
            evaluation_input: Input data for evaluation

        Returns:
            List of MetricScore objects for all metrics
        """
        logger.debug(f"Calculating {len(metrics)} metrics with dependency resolution")

        # Separate metrics: independent vs dependent
        independent_metrics = []
        dependent_metrics = []

        for metric in metrics:
            deps = metric.get_dependencies()
            has_deps = bool(deps.get("requires_metrics") or deps.get("requires_metadata"))

            if has_deps:
                dependent_metrics.append(metric)
            else:
                independent_metrics.append(metric)

        all_scores = []

        # If no dependent metrics, calculate everything in parallel
        if not dependent_metrics:
            logger.debug("No dependent metrics, calculating all metrics in parallel")
            return self.calculate_metrics_parallel(metrics, evaluation_input)

        # Check which dependencies are already satisfied
        required_metadata = set()
        for metric in dependent_metrics:
            deps = metric.get_dependencies()
            required_metadata.update(deps.get("requires_metadata", []))

        # Filter out independent metrics if their metadata is already present
        missing_metadata = [key for key in required_metadata if key not in evaluation_input.trace.metadata]

        if not missing_metadata:
            # All dependencies already satisfied, calculate everything in parallel
            logger.debug("All dependencies already satisfied in trace.metadata, calculating all metrics in parallel")
            return self.calculate_metrics_parallel(metrics, evaluation_input)

        logger.debug(f"Missing metadata: {missing_metadata}, will calculate provider metrics first")

        # Phase 1: Calculate independent metrics to provide missing metadata
        logger.debug(f"Phase 1: Calculating {len(independent_metrics)} independent metrics")
        phase1_scores = self.calculate_metrics_parallel(independent_metrics, evaluation_input)
        all_scores.extend(phase1_scores)

        # Store metadata from phase 1 for phase 2
        self._store_metadata_from_scores(phase1_scores, evaluation_input)

        # Phase 2: Calculate dependent metrics
        logger.debug(f"Phase 2: Calculating {len(dependent_metrics)} dependent metrics")
        phase2_scores = self.calculate_metrics_parallel(dependent_metrics, evaluation_input)
        all_scores.extend(phase2_scores)

        logger.debug(f"Calculated {len(all_scores)} total metric scores")

        return all_scores

    def _store_metadata_from_scores(
        self,
        scores: List[MetricScore],
        evaluation_input: EvaluationInput
    ) -> None:
        """
        Store metadata in trace from calculated metric scores.

        Automatically extracts and stores:
        - quality_score: Average of quality metrics
        - completeness_score: From conversation_completeness

        Args:
            scores: List of calculated metric scores
            evaluation_input: Evaluation input containing trace
        """
        # Collect quality scores
        quality_metric_names = ["answer_relevance", "completeness", "accuracy", "hallucination_score"]
        quality_scores = []

        for score in scores:
            if score.score is None:
                continue

            # Quality metrics
            if score.metric_name in quality_metric_names:
                if score.metric_name == "hallucination_score":
                    quality_scores.append(1.0 - score.score)  # Invert
                else:
                    quality_scores.append(score.score)

            # Completeness score
            elif score.metric_name == "conversation_completeness":
                evaluation_input.trace.metadata["completeness_score"] = score.score
                logger.debug(f"Stored completeness_score={score.score:.3f}")

        # Store average quality score
        if quality_scores:
            avg_quality = sum(quality_scores) / len(quality_scores)
            evaluation_input.trace.metadata["quality_score"] = avg_quality
            logger.debug(f"Stored quality_score={avg_quality:.3f}")

    def calculate_metrics_parallel(
        self,
        metrics: List[BaseMetric],
        evaluation_input: EvaluationInput
    ) -> List[MetricScore]:
        """
        Calculate multiple metrics in parallel.

        Separates deterministic and LLM-based metrics, running deterministic
        metrics in parallel threads and LLM-based metrics asynchronously.

        Args:
            metrics: List of metric instances to calculate
            evaluation_input: Input data for evaluation

        Returns:
            List of MetricScore objects
        """
        logger.debug(f"Calculating {len(metrics)} metrics in parallel")

        # Separate deterministic and LLM-based metrics
        deterministic_metrics = []
        llm_metrics = []

        for metric in metrics:
            if metric.requires_llm_judge():
                llm_metrics.append(metric)
            else:
                deterministic_metrics.append(metric)

        logger.debug(
            f"Split metrics: {len(deterministic_metrics)} deterministic, "
            f"{len(llm_metrics)} LLM-based"
        )

        # Calculate deterministic metrics in parallel
        deterministic_scores = self._calculate_deterministic_parallel(
            deterministic_metrics,
            evaluation_input
        )

        # Calculate LLM-based metrics asynchronously
        llm_scores = self._calculate_llm_async(
            llm_metrics,
            evaluation_input
        )

        # Combine results
        all_scores = deterministic_scores + llm_scores

        logger.debug(f"Calculated {len(all_scores)} metric scores")

        return all_scores

    def _calculate_deterministic_parallel(
        self,
        metrics: List[BaseMetric],
        evaluation_input: EvaluationInput
    ) -> List[MetricScore]:
        """
        Calculate deterministic metrics in parallel using ThreadPoolExecutor.

        Args:
            metrics: List of deterministic metric instances
            evaluation_input: Input data for evaluation

        Returns:
            List of MetricScore objects
        """
        if not metrics:
            return []

        scores = []

        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            # Submit all metric calculations
            future_to_metric = {
                executor.submit(
                    self._safe_calculate_metric,
                    metric,
                    evaluation_input
                ): metric
                for metric in metrics
            }

            # Collect results as they complete
            for future in as_completed(future_to_metric):
                metric = future_to_metric[future]
                try:
                    score = future.result()
                    if score is not None:
                        scores.append(score)
                except Exception as e:
                    logger.error(
                        f"Failed to calculate metric {metric.get_name()}: {str(e)}"
                    )

        return scores

    def _calculate_llm_async(
        self,
        metrics: List[BaseMetric],
        evaluation_input: EvaluationInput
    ) -> List[MetricScore]:
        """
        Calculate LLM-based metrics asynchronously.

        Args:
            metrics: List of LLM-based metric instances
            evaluation_input: Input data for evaluation

        Returns:
            List of MetricScore objects
        """
        if not metrics:
            return []

        # Run async metrics
        # Handle both Jupyter (with running loop) and regular Python environments
        try:
            # Try to get the running loop (Jupyter/IPython)
            loop = asyncio.get_running_loop()
            # If we're here, there's a running loop - use nest_asyncio
            try:
                import nest_asyncio
                nest_asyncio.apply()
                scores = loop.run_until_complete(
                    self._calculate_llm_async_impl(metrics, evaluation_input)
                )
            except ImportError:
                # nest_asyncio not available, create task in existing loop
                import concurrent.futures
                with concurrent.futures.ThreadPoolExecutor() as executor:
                    future = executor.submit(
                        asyncio.run,
                        self._calculate_llm_async_impl(metrics, evaluation_input)
                    )
                    scores = future.result()
        except RuntimeError:
            # No running loop, create a new one (regular Python)
            scores = asyncio.run(
                self._calculate_llm_async_impl(metrics, evaluation_input)
            )

        return scores

    async def _calculate_llm_async_impl(
        self,
        metrics: List[BaseMetric],
        evaluation_input: EvaluationInput
    ) -> List[MetricScore]:
        """
        Async implementation of LLM metric calculation.

        Groups metrics by their ``batch_group()`` key. Metrics that share a
        batch group are evaluated together via ``batch_calculate()`` (which can
        make concurrent external calls). Metrics without a batch group are run
        sequentially to avoid event-loop conflicts with libraries that manage
        their own async internals (e.g., DeepEval).

        Args:
            metrics: List of LLM-based metric instances
            evaluation_input: Input data for evaluation

        Returns:
            List of MetricScore objects
        """
        # Separate batchable metrics from unbatchable ones
        batches: Dict[str, List[BaseMetric]] = {}
        sequential_metrics: List[BaseMetric] = []

        for metric in metrics:
            group = metric.batch_group()
            if group is not None:
                batches.setdefault(group, []).append(metric)
            else:
                sequential_metrics.append(metric)

        scores: List[MetricScore] = []

        # Evaluate each batch group via batch_calculate
        for group_key, group_metrics in batches.items():
            try:
                # Use the first metric's class to call the classmethod
                batch_cls = type(group_metrics[0])
                loop = asyncio.get_event_loop()
                batch_scores = await loop.run_in_executor(
                    None,
                    batch_cls.batch_calculate,
                    group_metrics,
                    evaluation_input,
                )
                scores.extend(batch_scores)
            except Exception as e:
                logger.error(f"Batch evaluation failed for group '{group_key}': {e}")
                # Fall back to sequential for this group
                for metric in group_metrics:
                    result = await self._safe_calculate_metric_async(metric, evaluation_input)
                    if result is not None:
                        scores.append(result)

        # Run remaining LLM metrics sequentially
        for metric in sequential_metrics:
            result = await self._safe_calculate_metric_async(metric, evaluation_input)
            if result is not None:
                scores.append(result)

        return scores

    def _safe_calculate_metric(
        self,
        metric: BaseMetric,
        evaluation_input: EvaluationInput
    ) -> Optional[MetricScore]:
        """
        Safely calculate a metric with error handling.

        Args:
            metric: Metric instance to calculate
            evaluation_input: Input data for evaluation

        Returns:
            MetricScore, a failed placeholder MetricScore (score=None,
            metadata["error"] set) if calculation raises, or None if the
            metric was skipped for missing ground truth
        """
        try:
            # Check if ground truth is required but missing
            if metric.requires_ground_truth() and not evaluation_input.ground_truth:
                logger.warning(
                    f"Skipping metric {metric.get_name()}: requires ground truth but none provided"
                )
                return None

            # Calculate metric
            score = metric.calculate(evaluation_input)
            return score

        except Exception as e:
            logger.error(f"Error calculating metric {metric.get_name()}: {str(e)}")
            return self._failed_metric_score(metric, e)

    @staticmethod
    def _failed_metric_score(metric: BaseMetric, error: Exception) -> MetricScore:
        """
        Build a placeholder score for a metric whose calculation raised.

        Keeps the failure visible in dimension_results (distinct from a metric
        that legitimately returned score=None as not applicable).
        """
        return MetricScore(
            metric_name=metric.get_name(),
            score=None,
            reasoning=f"Evaluation failed: {error}",
            metadata={"error": str(error), "error_type": type(error).__name__}
        )

    @staticmethod
    def _metric_error_warnings(metric_scores: List[MetricScore]) -> List[str]:
        """
        Build one warning per metric whose calculation raised.

        A dimension whose metrics all errored aggregates to None (N/A) and is
        skipped by threshold checks, so without these warnings an errored
        metric would be indistinguishable from a non-applicable one.
        """
        return [
            f"Metric {s.metric_name} failed "
            f"({s.metadata.get('error_type', 'Error')}): {s.metadata['error']}"
            for s in metric_scores
            if s.metadata and s.metadata.get("error")
        ]

    async def _safe_calculate_metric_async(
        self,
        metric: BaseMetric,
        evaluation_input: EvaluationInput
    ) -> Optional[MetricScore]:
        """
        Safely calculate a metric asynchronously with error handling.

        Runs the metric's synchronous calculate() in a thread executor to
        avoid event loop conflicts with RAGAS/DeepEval which manage their
        own async loops internally.

        Args:
            metric: Metric instance to calculate
            evaluation_input: Input data for evaluation

        Returns:
            MetricScore, a failed placeholder MetricScore (score=None,
            metadata["error"] set) if calculation raises, or None if the
            metric was skipped for missing ground truth
        """
        try:
            # Check if ground truth is required but missing
            if metric.requires_ground_truth() and not evaluation_input.ground_truth:
                logger.warning(
                    f"Skipping metric {metric.get_name()}: requires ground truth but none provided"
                )
                return None

            # Run synchronous calculate() in a thread to avoid nested event loop issues
            loop = asyncio.get_event_loop()
            score = await loop.run_in_executor(None, metric.calculate, evaluation_input)
            return score

        except Exception as e:
            logger.error(f"Error calculating metric {metric.get_name()}: {str(e)}")
            return self._failed_metric_score(metric, e)

    def _run_multi_turn(
        self,
        per_turn_traces: List[any],
        full_trace: any,
        ground_truth: Optional[GroundTruth] = None,
        metric_set: Optional[Dict[str, List[str]]] = None,
        dimension_weights: Optional[Dict[str, float]] = None,
        thresholds: Optional[Dict[str, float]] = None,
        context: Optional[List[str]] = None,
        **kwargs,
    ) -> EvaluationResult:
        """
        Type-agnostic multi-turn evaluation loop.

        Subclasses (SingleAgentEvaluator, MultiAgentEvaluator) expose their own
        public ``evaluate_multi_turn`` that validates the trace type and then
        delegates here. The body works for both AgentTrace and MultiAgentTrace
        because:
        - per-turn metrics use ``self._get_metrics_to_calculate`` which the
          subclass picks
        - multi-agent metrics guard on ``isinstance(trace, MultiAgentTrace)``
          and skip on plain AgentTrace
        - multi-turn metrics read ``trace.messages`` which exists on both
          (a computed property on MultiAgentTrace)

        Args:
            per_turn_traces: Per-turn traces (one per user turn)
            full_trace: Trace covering the entire conversation
            ground_truth: Expected outputs (optional). May contain
                          per_turn_expected/per_turn_tools in expected_arguments.
            metric_set: Optional dimension → metric-name mapping
            dimension_weights: Per-dimension weights (defaults to subclass defaults)
            thresholds: Per-dimension thresholds (defaults to subclass defaults)
            context: Optional list of context documents
            **kwargs: Forwarded to underlying evaluation calls

        Returns:
            EvaluationResult with per-turn breakdown in metadata
        """
        # Avoid circular import: imported lazily so base_evaluator stays
        # decoupled from the trace types.
        from uaef.models.multi_agent_trace import MultiAgentTrace

        is_multi_agent = isinstance(full_trace, MultiAgentTrace)

        logger.info(
            f"Evaluating multi-turn conversation: trace={full_trace.trace_id}, "
            f"turns={len(per_turn_traces)}"
        )

        dimension_weights = dimension_weights or self.default_dimension_weights
        thresholds = thresholds or self.default_thresholds
        ctx = context or (ground_truth.context_documents if ground_truth else [])

        metrics_to_calculate = self._get_metrics_to_calculate(metric_set)

        # Metrics that only make sense on the full conversation trace.
        # Today every Multi-Turn-dimension metric is full-trace; per-turn
        # relevance scoring is provided by `answer_relevance` (Response Quality
        # dimension), which the per-turn loop already runs and averages.
        FULL_TRACE_ONLY = {
            "context_retention",
            "coherence",
            "conversation_completeness",
            "turn_efficiency",
            "role_adherence",
            "holistic_llm_judge",
            "user_satisfaction",
            "sentiment",
            "agent_tone",
            "naturalness",
            "per_turn_sentiment",
            "per_turn_agent_tone",
            "per_turn_naturalness",
            "deepeval_role_adherence",
            "containment",
            "resolution",
            "instruction_compliance",
            "optimum_turns",
        }

        per_turn_metrics = [
            m for m in metrics_to_calculate
            if m.get_name() not in FULL_TRACE_ONLY
        ]
        full_trace_metrics = [
            m for m in metrics_to_calculate
            if m.get_name() in FULL_TRACE_ONLY
        ]

        # ------------------------------------------------------------------
        # Per-turn evaluation
        # ------------------------------------------------------------------
        per_turn_expected = (
            ground_truth.expected_arguments.get("per_turn_expected", [])
            if ground_truth else []
        )
        per_turn_tools = (
            ground_truth.expected_arguments.get("per_turn_tools", [])
            if ground_truth else []
        )

        per_turn_results: List[Dict] = []
        score_buckets: Dict[str, List[float]] = {}
        # Per-turn scores are re-aggregated below without their error
        # metadata, so record failures here, per turn.
        per_turn_warnings: List[str] = []

        for turn_idx, turn_trace in enumerate(per_turn_traces):
            logger.info(f"  Per-turn evaluation: turn {turn_idx + 1}/{len(per_turn_traces)}")

            turn_gt = None
            if ground_truth:
                turn_expected = (
                    per_turn_expected[turn_idx]
                    if turn_idx < len(per_turn_expected)
                    else ground_truth.expected_output
                )
                turn_tool_calls = (
                    per_turn_tools[turn_idx]
                    if turn_idx < len(per_turn_tools)
                    else ground_truth.expected_tool_calls
                )
                turn_gt = GroundTruth(
                    expected_output=turn_expected or None,
                    expected_tool_calls=turn_tool_calls,
                    context_documents=ground_truth.context_documents,
                )

            eval_input = EvaluationInput(
                trace=turn_trace,
                ground_truth=turn_gt,
                context=ctx,
            )
            scores = self.calculate_metrics_with_dependencies(
                metrics=per_turn_metrics,
                evaluation_input=eval_input,
            )
            per_turn_warnings.extend(
                f"Turn {turn_idx + 1}: {w}" for w in self._metric_error_warnings(scores)
            )
            turn_scores = {s.metric_name: s.score for s in scores}
            turn_reasoning = {s.metric_name: s.reasoning for s in scores}
            per_turn_results.append({
                "turn": turn_idx + 1,
                "metric_scores": turn_scores,
                "reasoning": turn_reasoning,
            })

            for metric_name, score in turn_scores.items():
                if score is not None:
                    score_buckets.setdefault(metric_name, []).append(score)

        per_turn_averages = {
            metric_name: sum(vals) / len(vals)
            for metric_name, vals in score_buckets.items()
        }

        # Build per-metric reasoning bundles so the aggregated MetricScore
        # carries each turn's reasoning instead of a generic "Average across
        # N turns" string. The structured form is in metadata["per_turn"]
        # for programmatic access; the formatted form lands in `reasoning`
        # so notebook-style printing surfaces it without extra plumbing.
        reasoning_buckets: Dict[str, List[Dict[str, any]]] = {}
        for turn_entry in per_turn_results:
            turn_no = turn_entry["turn"]
            for mname, sc in turn_entry["metric_scores"].items():
                reasoning_buckets.setdefault(mname, []).append({
                    "turn": turn_no,
                    "score": sc,
                    "reasoning": turn_entry["reasoning"].get(mname, ""),
                })

        def _format_per_turn_reasoning(name: str) -> str:
            entries = reasoning_buckets.get(name, [])
            if not entries:
                return f"Average across {len(per_turn_traces)} turns"
            lines = [f"Average across {len(per_turn_traces)} turns:"]
            for e in entries:
                sc_str = f"{e['score']:.2f}" if e["score"] is not None else "N/A"
                reason = (e["reasoning"] or "").strip()
                if reason:
                    lines.append(f"  Turn {e['turn']} ({sc_str}): {reason}")
                else:
                    lines.append(f"  Turn {e['turn']} ({sc_str})")
            return "\n".join(lines)

        # Build metric scores from per-turn averages
        metric_scores: List[MetricScore] = []
        for m in per_turn_metrics:
            name = m.get_name()
            if name in per_turn_averages:
                metric_scores.append(MetricScore(
                    metric_name=name,
                    score=per_turn_averages[name],
                    reasoning=_format_per_turn_reasoning(name),
                    metadata={"per_turn": reasoning_buckets.get(name, [])},
                ))

        # ------------------------------------------------------------------
        # Full-trace evaluation
        # ------------------------------------------------------------------
        if full_trace_metrics:
            # Make quality_score available on the full trace so any quality-dependent
            # full-trace metric can resolve it.
            quality_metric_names = ["answer_relevance", "completeness", "accuracy"]
            quality_vals = [per_turn_averages[n] for n in quality_metric_names if n in per_turn_averages]
            if "hallucination_score" in per_turn_averages:
                quality_vals.append(1.0 - per_turn_averages["hallucination_score"])
            if quality_vals:
                full_trace.metadata["quality_score"] = sum(quality_vals) / len(quality_vals)

            full_eval_input = EvaluationInput(
                trace=full_trace,
                ground_truth=ground_truth,
                context=ctx,
            )
            full_trace_scores = self.calculate_metrics_with_dependencies(
                metrics=full_trace_metrics,
                evaluation_input=full_eval_input,
            )
            metric_scores.extend(full_trace_scores)

        # ------------------------------------------------------------------
        # Aggregate, threshold, return
        # ------------------------------------------------------------------
        dimension_results = self.aggregate_dimension_scores(
            metric_scores=metric_scores,
            dimension_weights=dimension_weights,
        )
        overall_score = self.calculate_overall_score(dimension_results)
        passed, failures = self.check_thresholds(
            dimension_results=dimension_results,
            thresholds=thresholds,
        )

        return EvaluationResult(
            trace_id=full_trace.trace_id,
            dimension_results=dimension_results,
            overall_score=overall_score,
            passed=passed,
            failures=failures,
            warnings=per_turn_warnings + self._metric_error_warnings(metric_scores),
            metadata={
                "num_metrics": len(metric_scores),
                "num_dimensions": len(dimension_results),
                "num_turns": len(per_turn_traces),
                "is_multi_agent": is_multi_agent,
                "dimension_weights": dimension_weights,
                "thresholds": thresholds,
                "per_turn": per_turn_results,
                "per_turn_averages": per_turn_averages,
            },
        )

    def _warn_unknown_dimension_keys(
        self,
        config: Optional[Dict[str, float]],
        label: str
    ) -> None:
        """
        Warn (once per key) when a dimension_weights/thresholds key doesn't
        match any dimension name the metric registry produces.

        Such keys are silently ignored by aggregation and threshold checks,
        so a typo or naming mismatch would otherwise go unnoticed.
        """
        if not config:
            return

        try:
            known = set(self.metric_registry.list_metrics_by_dimension().keys())
        except Exception as e:
            logger.debug(f"Could not list registry dimensions: {e}")
            return

        for key in config:
            if key in known or (label, key) in self._warned_dimension_keys:
                continue
            self._warned_dimension_keys.add((label, key))
            logger.warning(
                f"{label} key '{key}' does not match any registered dimension "
                f"{sorted(known)}; it will be ignored"
            )

    def aggregate_dimension_scores(
        self,
        metric_scores: List[MetricScore],
        dimension_weights: Optional[Dict[str, float]] = None
    ) -> List[DimensionResult]:
        """
        Aggregate metric scores by dimension.

        Args:
            metric_scores: List of metric scores to aggregate
            dimension_weights: Optional weights for each dimension

        Returns:
            List of DimensionResult objects
        """
        self._warn_unknown_dimension_keys(dimension_weights, "dimension_weights")

        # Group scores by dimension
        dimension_scores: Dict[str, List[MetricScore]] = {}

        for score in metric_scores:
            # Get metric info to determine dimension
            try:
                metric_info = self.metric_registry.get_metric_info(score.metric_name)
                dimension = metric_info.get("dimension", "unknown")

                if dimension not in dimension_scores:
                    dimension_scores[dimension] = []

                dimension_scores[dimension].append(score)

            except KeyError:
                logger.warning(
                    f"Metric {score.metric_name} not found in registry, "
                    f"assigning to 'unknown' dimension"
                )
                if "unknown" not in dimension_scores:
                    dimension_scores["unknown"] = []
                dimension_scores["unknown"].append(score)

        # Calculate aggregate score for each dimension
        dimension_results = []

        for dimension_name, scores in dimension_scores.items():
            # Filter out None scores (non-applicable metrics)
            valid_scores = [s for s in scores if s.score is not None]

            # Calculate weighted average from valid scores only. A dimension
            # with no applicable metrics is not applicable (None), not 0.0.
            if valid_scores:
                aggregate_score = sum(s.score for s in valid_scores) / len(valid_scores)
            else:
                aggregate_score = None

            # Get dimension weight
            weight = 1.0
            if dimension_weights and dimension_name in dimension_weights:
                weight = dimension_weights[dimension_name]

            dimension_result = DimensionResult(
                dimension_name=dimension_name,
                metric_scores=scores,  # Include all scores (even None) for transparency
                aggregate_score=aggregate_score,
                weight=weight
            )

            dimension_results.append(dimension_result)

        return dimension_results

    def calculate_overall_score(
        self,
        dimension_results: List[DimensionResult]
    ) -> float:
        """
        Calculate overall score from dimension results.

        Args:
            dimension_results: List of dimension results

        Returns:
            Overall weighted score (0-1)
        """
        # Skip dimensions with no applicable metrics
        dimension_results = [
            d for d in dimension_results if d.aggregate_score is not None
        ]

        if not dimension_results:
            return 0.0

        # Calculate weighted average
        total_weight = sum(d.weight for d in dimension_results)

        if total_weight == 0:
            # Equal weights if all weights are 0
            return sum(d.aggregate_score for d in dimension_results) / len(dimension_results)

        weighted_sum = sum(d.aggregate_score * d.weight for d in dimension_results)
        overall_score = weighted_sum / total_weight

        return overall_score

    def check_thresholds(
        self,
        dimension_results: List[DimensionResult],
        thresholds: Optional[Dict[str, float]] = None
    ) -> Tuple[bool, List[str]]:
        """
        Check if dimension scores meet thresholds.

        Args:
            dimension_results: List of dimension results
            thresholds: Dictionary mapping dimension names to threshold values

        Returns:
            Tuple of (passed, failures) where passed is True if all thresholds met,
            and failures is a list of failure messages
        """
        if not thresholds:
            return True, []

        self._warn_unknown_dimension_keys(thresholds, "thresholds")

        failures = []

        for dimension_result in dimension_results:
            dimension_name = dimension_result.dimension_name

            # Dimensions with no applicable metrics can't fail a threshold
            if dimension_result.aggregate_score is None:
                continue

            if dimension_name in thresholds:
                threshold = thresholds[dimension_name]

                if dimension_result.aggregate_score < threshold:
                    failures.append(
                        f"{dimension_name}: score {dimension_result.aggregate_score:.3f} "
                        f"below threshold {threshold:.3f}"
                    )

        passed = len(failures) == 0

        return passed, failures

    def _create_failed_result(
        self,
        trace: any,
        error_message: str
    ) -> EvaluationResult:
        """
        Create a failed evaluation result.

        Args:
            trace: The trace that failed evaluation
            error_message: Error message describing the failure

        Returns:
            EvaluationResult with failed status
        """
        # Extract trace_id if possible
        trace_id = getattr(trace, "trace_id", None)

        return EvaluationResult(
            trace_id=trace_id,
            dimension_results=[],
            overall_score=0.0,
            passed=False,
            failures=[f"Evaluation failed: {error_message}"],
            warnings=[],
            metadata={"error": error_message}
        )

aggregate_dimension_scores(metric_scores, dimension_weights=None)

Aggregate metric scores by dimension.

Parameters:

Name	Type	Description	Default
metric_scores	List[MetricScore]	List of metric scores to aggregate	required
dimension_weights	Optional[Dict[str, float]]	Optional weights for each dimension	None

Returns:

Type	Description
List[DimensionResult]	List of DimensionResult objects

Source code in uaef/evaluation/base_evaluator.py

def aggregate_dimension_scores(
    self,
    metric_scores: List[MetricScore],
    dimension_weights: Optional[Dict[str, float]] = None
) -> List[DimensionResult]:
    """
    Aggregate metric scores by dimension.

    Args:
        metric_scores: List of metric scores to aggregate
        dimension_weights: Optional weights for each dimension

    Returns:
        List of DimensionResult objects
    """
    self._warn_unknown_dimension_keys(dimension_weights, "dimension_weights")

    # Group scores by dimension
    dimension_scores: Dict[str, List[MetricScore]] = {}

    for score in metric_scores:
        # Get metric info to determine dimension
        try:
            metric_info = self.metric_registry.get_metric_info(score.metric_name)
            dimension = metric_info.get("dimension", "unknown")

            if dimension not in dimension_scores:
                dimension_scores[dimension] = []

            dimension_scores[dimension].append(score)

        except KeyError:
            logger.warning(
                f"Metric {score.metric_name} not found in registry, "
                f"assigning to 'unknown' dimension"
            )
            if "unknown" not in dimension_scores:
                dimension_scores["unknown"] = []
            dimension_scores["unknown"].append(score)

    # Calculate aggregate score for each dimension
    dimension_results = []

    for dimension_name, scores in dimension_scores.items():
        # Filter out None scores (non-applicable metrics)
        valid_scores = [s for s in scores if s.score is not None]

        # Calculate weighted average from valid scores only. A dimension
        # with no applicable metrics is not applicable (None), not 0.0.
        if valid_scores:
            aggregate_score = sum(s.score for s in valid_scores) / len(valid_scores)
        else:
            aggregate_score = None

        # Get dimension weight
        weight = 1.0
        if dimension_weights and dimension_name in dimension_weights:
            weight = dimension_weights[dimension_name]

        dimension_result = DimensionResult(
            dimension_name=dimension_name,
            metric_scores=scores,  # Include all scores (even None) for transparency
            aggregate_score=aggregate_score,
            weight=weight
        )

        dimension_results.append(dimension_result)

    return dimension_results

batch_evaluate(traces, ground_truths=None, metric_set=None, **kwargs)

Evaluate multiple traces in batch.

Parameters:

Name	Type	Description	Default
traces	List[any]	List of agent traces to evaluate	required
ground_truths	Optional[List[Optional[GroundTruth]]]	List of ground truths (optional, can be None for each)	None
metric_set	Optional[Dict[str, List[str]]]	Dictionary mapping dimension names to lists of metric names	None
**kwargs		Additional evaluation parameters	{}

Returns:

Type	Description
List[EvaluationResult]	List of EvaluationResult objects

Raises:

Type	Description
ValueError	If inputs are invalid

Source code in uaef/evaluation/base_evaluator.py

def batch_evaluate(
    self,
    traces: List[any],
    ground_truths: Optional[List[Optional[GroundTruth]]] = None,
    metric_set: Optional[Dict[str, List[str]]] = None,
    **kwargs
) -> List[EvaluationResult]:
    """
    Evaluate multiple traces in batch.

    Args:
        traces: List of agent traces to evaluate
        ground_truths: List of ground truths (optional, can be None for each)
        metric_set: Dictionary mapping dimension names to lists of metric names
        **kwargs: Additional evaluation parameters

    Returns:
        List of EvaluationResult objects

    Raises:
        ValueError: If inputs are invalid
    """
    logger.info(f"Starting batch evaluation for {len(traces)} traces")

    # Validate inputs
    if not traces:
        raise ValueError("traces cannot be empty")

    if ground_truths is None:
        ground_truths = [None] * len(traces)

    if len(traces) != len(ground_truths):
        raise ValueError(
            f"Length mismatch: {len(traces)} traces but {len(ground_truths)} ground truths"
        )

    # Evaluate each trace
    results = []
    for i, (trace, gt) in enumerate(zip(traces, ground_truths)):
        try:
            logger.debug(f"Evaluating trace {i+1}/{len(traces)}")
            result = self.evaluate(
                trace=trace,
                ground_truth=gt,
                metric_set=metric_set,
                **kwargs
            )
            results.append(result)

        except Exception as e:
            logger.error(f"Failed to evaluate trace {i+1}: {str(e)}")
            # Create a failed evaluation result
            results.append(self._create_failed_result(trace, str(e)))

    success_count = sum(1 for r in results if r.passed)
    logger.info(
        f"Batch evaluation complete: {success_count}/{len(results)} passed"
    )

    return results

calculate_metrics_parallel(metrics, evaluation_input)

Calculate multiple metrics in parallel.

Separates deterministic and LLM-based metrics, running deterministic metrics in parallel threads and LLM-based metrics asynchronously.

Parameters:

Name	Type	Description	Default
metrics	List[BaseMetric]	List of metric instances to calculate	required
evaluation_input	EvaluationInput	Input data for evaluation	required

Returns:

Type	Description
List[MetricScore]	List of MetricScore objects

Source code in uaef/evaluation/base_evaluator.py

def calculate_metrics_parallel(
    self,
    metrics: List[BaseMetric],
    evaluation_input: EvaluationInput
) -> List[MetricScore]:
    """
    Calculate multiple metrics in parallel.

    Separates deterministic and LLM-based metrics, running deterministic
    metrics in parallel threads and LLM-based metrics asynchronously.

    Args:
        metrics: List of metric instances to calculate
        evaluation_input: Input data for evaluation

    Returns:
        List of MetricScore objects
    """
    logger.debug(f"Calculating {len(metrics)} metrics in parallel")

    # Separate deterministic and LLM-based metrics
    deterministic_metrics = []
    llm_metrics = []

    for metric in metrics:
        if metric.requires_llm_judge():
            llm_metrics.append(metric)
        else:
            deterministic_metrics.append(metric)

    logger.debug(
        f"Split metrics: {len(deterministic_metrics)} deterministic, "
        f"{len(llm_metrics)} LLM-based"
    )

    # Calculate deterministic metrics in parallel
    deterministic_scores = self._calculate_deterministic_parallel(
        deterministic_metrics,
        evaluation_input
    )

    # Calculate LLM-based metrics asynchronously
    llm_scores = self._calculate_llm_async(
        llm_metrics,
        evaluation_input
    )

    # Combine results
    all_scores = deterministic_scores + llm_scores

    logger.debug(f"Calculated {len(all_scores)} metric scores")

    return all_scores

calculate_metrics_with_dependencies(metrics, evaluation_input)

Calculate metrics respecting dependencies.

Simple two-phase approach: - Phase 1: Calculate metrics with no dependencies - Phase 2: Calculate metrics with dependencies (which can now read metadata)

Parameters:

Name	Type	Description	Default
metrics	List[BaseMetric]	List of metric instances to calculate	required
evaluation_input	EvaluationInput	Input data for evaluation	required

Returns:

Type	Description
List[MetricScore]	List of MetricScore objects for all metrics

Source code in uaef/evaluation/base_evaluator.py

def calculate_metrics_with_dependencies(
    self,
    metrics: List[BaseMetric],
    evaluation_input: EvaluationInput
) -> List[MetricScore]:
    """
    Calculate metrics respecting dependencies.

    Simple two-phase approach:
    - Phase 1: Calculate metrics with no dependencies
    - Phase 2: Calculate metrics with dependencies (which can now read metadata)

    Args:
        metrics: List of metric instances to calculate
        evaluation_input: Input data for evaluation

    Returns:
        List of MetricScore objects for all metrics
    """
    logger.debug(f"Calculating {len(metrics)} metrics with dependency resolution")

    # Separate metrics: independent vs dependent
    independent_metrics = []
    dependent_metrics = []

    for metric in metrics:
        deps = metric.get_dependencies()
        has_deps = bool(deps.get("requires_metrics") or deps.get("requires_metadata"))

        if has_deps:
            dependent_metrics.append(metric)
        else:
            independent_metrics.append(metric)

    all_scores = []

    # If no dependent metrics, calculate everything in parallel
    if not dependent_metrics:
        logger.debug("No dependent metrics, calculating all metrics in parallel")
        return self.calculate_metrics_parallel(metrics, evaluation_input)

    # Check which dependencies are already satisfied
    required_metadata = set()
    for metric in dependent_metrics:
        deps = metric.get_dependencies()
        required_metadata.update(deps.get("requires_metadata", []))

    # Filter out independent metrics if their metadata is already present
    missing_metadata = [key for key in required_metadata if key not in evaluation_input.trace.metadata]

    if not missing_metadata:
        # All dependencies already satisfied, calculate everything in parallel
        logger.debug("All dependencies already satisfied in trace.metadata, calculating all metrics in parallel")
        return self.calculate_metrics_parallel(metrics, evaluation_input)

    logger.debug(f"Missing metadata: {missing_metadata}, will calculate provider metrics first")

    # Phase 1: Calculate independent metrics to provide missing metadata
    logger.debug(f"Phase 1: Calculating {len(independent_metrics)} independent metrics")
    phase1_scores = self.calculate_metrics_parallel(independent_metrics, evaluation_input)
    all_scores.extend(phase1_scores)

    # Store metadata from phase 1 for phase 2
    self._store_metadata_from_scores(phase1_scores, evaluation_input)

    # Phase 2: Calculate dependent metrics
    logger.debug(f"Phase 2: Calculating {len(dependent_metrics)} dependent metrics")
    phase2_scores = self.calculate_metrics_parallel(dependent_metrics, evaluation_input)
    all_scores.extend(phase2_scores)

    logger.debug(f"Calculated {len(all_scores)} total metric scores")

    return all_scores

calculate_overall_score(dimension_results)

Calculate overall score from dimension results.

Parameters:

Name	Type	Description	Default
dimension_results	List[DimensionResult]	List of dimension results	required

Returns:

Type	Description
float	Overall weighted score (0-1)

Source code in uaef/evaluation/base_evaluator.py

def calculate_overall_score(
    self,
    dimension_results: List[DimensionResult]
) -> float:
    """
    Calculate overall score from dimension results.

    Args:
        dimension_results: List of dimension results

    Returns:
        Overall weighted score (0-1)
    """
    # Skip dimensions with no applicable metrics
    dimension_results = [
        d for d in dimension_results if d.aggregate_score is not None
    ]

    if not dimension_results:
        return 0.0

    # Calculate weighted average
    total_weight = sum(d.weight for d in dimension_results)

    if total_weight == 0:
        # Equal weights if all weights are 0
        return sum(d.aggregate_score for d in dimension_results) / len(dimension_results)

    weighted_sum = sum(d.aggregate_score * d.weight for d in dimension_results)
    overall_score = weighted_sum / total_weight

    return overall_score

check_thresholds(dimension_results, thresholds=None)

Check if dimension scores meet thresholds.

Parameters:

Name	Type	Description	Default
dimension_results	List[DimensionResult]	List of dimension results	required
thresholds	Optional[Dict[str, float]]	Dictionary mapping dimension names to threshold values	None

Returns:

Type	Description
bool	Tuple of (passed, failures) where passed is True if all thresholds met,
List[str]	and failures is a list of failure messages

Source code in uaef/evaluation/base_evaluator.py

def check_thresholds(
    self,
    dimension_results: List[DimensionResult],
    thresholds: Optional[Dict[str, float]] = None
) -> Tuple[bool, List[str]]:
    """
    Check if dimension scores meet thresholds.

    Args:
        dimension_results: List of dimension results
        thresholds: Dictionary mapping dimension names to threshold values

    Returns:
        Tuple of (passed, failures) where passed is True if all thresholds met,
        and failures is a list of failure messages
    """
    if not thresholds:
        return True, []

    self._warn_unknown_dimension_keys(thresholds, "thresholds")

    failures = []

    for dimension_result in dimension_results:
        dimension_name = dimension_result.dimension_name

        # Dimensions with no applicable metrics can't fail a threshold
        if dimension_result.aggregate_score is None:
            continue

        if dimension_name in thresholds:
            threshold = thresholds[dimension_name]

            if dimension_result.aggregate_score < threshold:
                failures.append(
                    f"{dimension_name}: score {dimension_result.aggregate_score:.3f} "
                    f"below threshold {threshold:.3f}"
                )

    passed = len(failures) == 0

    return passed, failures

evaluate(trace, ground_truth=None, metric_set=None, **kwargs) abstractmethod

Evaluate a single trace.

Parameters:

Name	Type	Description	Default
trace	any	Agent trace to evaluate	required
ground_truth	Optional[GroundTruth]	Expected correct outputs (optional)	None
metric_set	Optional[Dict[str, List[str]]]	Dictionary mapping dimension names to lists of metric names	None
**kwargs		Additional evaluation parameters	{}

Returns:

Type	Description
EvaluationResult	EvaluationResult with scores and metadata

Raises:

Type	Description
ValueError	If inputs are invalid
RuntimeError	If evaluation fails

Source code in uaef/evaluation/base_evaluator.py

@abstractmethod
def evaluate(
    self,
    trace: any,
    ground_truth: Optional[GroundTruth] = None,
    metric_set: Optional[Dict[str, List[str]]] = None,
    **kwargs
) -> EvaluationResult:
    """
    Evaluate a single trace.

    Args:
        trace: Agent trace to evaluate
        ground_truth: Expected correct outputs (optional)
        metric_set: Dictionary mapping dimension names to lists of metric names
        **kwargs: Additional evaluation parameters

    Returns:
        EvaluationResult with scores and metadata

    Raises:
        ValueError: If inputs are invalid
        RuntimeError: If evaluation fails
    """
    pass

uaef.evaluation.single_agent

Single-agent evaluator for UAEF.

This module provides evaluation functionality for single-agent traces, coordinating the adapter → metrics → LLM judge flow.

SingleAgentEvaluator

Bases: BaseEvaluator

Evaluator for single-agent traces.

Coordinates the complete evaluation flow: 1. Create EvaluationInput from trace and ground truth 2. Calculate metrics for each dimension 3. Aggregate dimension scores with weighted averages 4. Calculate overall score 5. Check thresholds and identify failures

Source code in uaef/evaluation/single_agent.py

class SingleAgentEvaluator(BaseEvaluator):
    """
    Evaluator for single-agent traces.

    Coordinates the complete evaluation flow:
    1. Create EvaluationInput from trace and ground truth
    2. Calculate metrics for each dimension
    3. Aggregate dimension scores with weighted averages
    4. Calculate overall score
    5. Check thresholds and identify failures
    """

    def __init__(
        self,
        metric_registry: Optional[any] = None,
        max_workers: int = 4,
        default_dimension_weights: Optional[Dict[str, float]] = None,
        default_thresholds: Optional[Dict[str, float]] = None
    ):
        """
        Initialize single-agent evaluator.

        Args:
            metric_registry: MetricRegistry instance (uses global if None)
            max_workers: Maximum number of parallel workers for metric calculation
            default_dimension_weights: Default weights for dimensions
            default_thresholds: Default threshold values for dimensions
        """
        super().__init__(metric_registry=metric_registry, max_workers=max_workers)

        # Set default dimension weights if not provided
        self.default_dimension_weights = default_dimension_weights or {
            "Tool Calling": 0.25,
            "Response Quality": 0.25,
            "Responsible AI": 0.20,
            "Performance": 0.10,
            "Multi-Turn": 0.10,
            "Reasoning": 0.10
        }

        # Set default thresholds if not provided
        self.default_thresholds = default_thresholds or {
            "Tool Calling": 0.7,
            "Response Quality": 0.7,
            "Responsible AI": 0.8,
            "Performance": 0.6,
            "Multi-Turn": 0.6,
            "Reasoning": 0.6
        }

        logger.info(
            f"Initialized SingleAgentEvaluator with "
            f"{len(self.default_dimension_weights)} dimensions"
        )

    def evaluate(
        self,
        trace: AgentTrace,
        ground_truth: Optional[GroundTruth] = None,
        metric_set: Optional[Dict[str, List[str]]] = None,
        dimension_weights: Optional[Dict[str, float]] = None,
        thresholds: Optional[Dict[str, float]] = None,
        context: Optional[List[str]] = None,
        **kwargs
    ) -> EvaluationResult:
        """
        Evaluate a single agent trace.

        Args:
            trace: Agent trace to evaluate
            ground_truth: Expected correct outputs (optional)
            metric_set: Dictionary mapping dimension names to lists of metric names.
                       If None, uses all available metrics.
            dimension_weights: Weights for each dimension (uses defaults if None)
            thresholds: Threshold values for pass/fail (uses defaults if None)
            context: List of context documents (optional)
            **kwargs: Additional evaluation parameters

        Returns:
            EvaluationResult with scores and metadata

        Raises:
            ValueError: If inputs are invalid
            RuntimeError: If evaluation fails
        """
        logger.info(f"Evaluating trace {trace.trace_id}")

        # Validate inputs
        if not isinstance(trace, AgentTrace):
            raise ValueError("trace must be an AgentTrace instance")

        # Use default weights and thresholds if not provided
        dimension_weights = dimension_weights or self.default_dimension_weights
        thresholds = thresholds or self.default_thresholds

        # Create evaluation input
        evaluation_input = EvaluationInput(
            trace=trace,
            ground_truth=ground_truth,
            context=context or (ground_truth.context_documents if ground_truth else [])
        )

        # Determine which metrics to calculate
        metrics_to_calculate = self._get_metrics_to_calculate(metric_set)

        logger.debug(f"Calculating {len(metrics_to_calculate)} metrics")

        # Calculate metrics with automatic dependency resolution
        metric_scores = self.calculate_metrics_with_dependencies(
            metrics=metrics_to_calculate,
            evaluation_input=evaluation_input
        )

        logger.debug(f"Calculated {len(metric_scores)} metric scores")

        # Aggregate scores by dimension
        dimension_results = self.aggregate_dimension_scores(
            metric_scores=metric_scores,
            dimension_weights=dimension_weights
        )

        logger.debug(f"Aggregated scores for {len(dimension_results)} dimensions")

        # Calculate overall score
        overall_score = self.calculate_overall_score(dimension_results)

        logger.debug(f"Overall score: {overall_score:.3f}")

        # Check thresholds
        passed, failures = self.check_thresholds(
            dimension_results=dimension_results,
            thresholds=thresholds
        )

        # Collect warnings
        warnings = []
        if ground_truth is None:
            gt_metrics = self.metric_registry.get_metrics_requiring_ground_truth()
            if any(m.metric_name in gt_metrics for m in metric_scores):
                warnings.append(
                    "Ground truth not provided, some metrics may be skipped"
                )

        warnings.extend(self._metric_error_warnings(metric_scores))

        # Create evaluation result
        result = EvaluationResult(
            trace_id=trace.trace_id,
            dimension_results=dimension_results,
            overall_score=overall_score,
            passed=passed,
            failures=failures,
            warnings=warnings,
            metadata={
                "num_metrics": len(metric_scores),
                "num_dimensions": len(dimension_results),
                "dimension_weights": dimension_weights,
                "thresholds": thresholds
            }
        )

        logger.info(
            f"Evaluation complete for trace {trace.trace_id}: "
            f"score={overall_score:.3f}, passed={passed}"
        )

        return result

    def evaluate_trace(
        self,
        trace: AgentTrace,
        ground_truth: Optional[GroundTruth] = None,
        metric_set: Optional[Dict[str, List[str]]] = None,
        **kwargs
    ) -> EvaluationResult:
        """
        Alias for evaluate() method for backward compatibility.

        Args:
            trace: Agent trace to evaluate
            ground_truth: Expected correct outputs (optional)
            metric_set: Dictionary mapping dimension names to lists of metric names
            **kwargs: Additional evaluation parameters

        Returns:
            EvaluationResult with scores and metadata
        """
        return self.evaluate(
            trace=trace,
            ground_truth=ground_truth,
            metric_set=metric_set,
            **kwargs
        )

    def evaluate_multi_turn(
        self,
        per_turn_traces: List[AgentTrace],
        full_trace: AgentTrace,
        ground_truth: Optional[GroundTruth] = None,
        metric_set: Optional[Dict[str, List[str]]] = None,
        dimension_weights: Optional[Dict[str, float]] = None,
        thresholds: Optional[Dict[str, float]] = None,
        context: Optional[List[str]] = None,
        **kwargs
    ) -> EvaluationResult:
        """
        Evaluate a single-agent multi-turn conversation from pre-split per-turn traces.

        Both per_turn_traces and full_trace must be AgentTrace instances.
        Multi-agent traces are handled by MultiAgentEvaluator.evaluate_multi_turn().

        Flow:
        1. Iterate per_turn_traces, run per-turn-eligible metrics on each
           (using `ground_truth.expected_arguments["per_turn_expected"][i]` and
           `["per_turn_tools"][i]` for that turn's ground truth)
        2. Average per-turn scores across turns
        3. Run full-trace-only metrics (multi-turn + throughput) on full_trace
        4. Combine and return EvaluationResult with per-turn breakdown in metadata

        Args:
            per_turn_traces: List of per-turn AgentTraces (one per user turn)
            full_trace: AgentTrace covering the entire conversation
            ground_truth: Expected correct outputs (optional). May contain
                         per_turn_expected and per_turn_tools arrays in expected_arguments.
            metric_set: Dictionary mapping dimension names to lists of metric names
            dimension_weights: Weights for each dimension (uses defaults if None)
            thresholds: Threshold values for pass/fail (uses defaults if None)
            context: List of context documents (optional)
            **kwargs: Additional evaluation parameters

        Returns:
            EvaluationResult with scores and metadata (including per_turn breakdown)
        """
        if not per_turn_traces:
            raise ValueError("per_turn_traces must contain at least one trace")
        if not isinstance(full_trace, AgentTrace) or isinstance(full_trace, MultiAgentTrace):
            raise ValueError(
                "full_trace must be an AgentTrace instance "
                "(use MultiAgentEvaluator.evaluate_multi_turn for MultiAgentTrace)"
            )
        for i, t in enumerate(per_turn_traces):
            if not isinstance(t, AgentTrace) or isinstance(t, MultiAgentTrace):
                raise ValueError(
                    f"per_turn_traces[{i}] must be an AgentTrace instance "
                    "(use MultiAgentEvaluator.evaluate_multi_turn for MultiAgentTrace)"
                )

        return self._run_multi_turn(
            per_turn_traces=per_turn_traces,
            full_trace=full_trace,
            ground_truth=ground_truth,
            metric_set=metric_set,
            dimension_weights=dimension_weights,
            thresholds=thresholds,
            context=context,
            **kwargs,
        )

    def _get_metrics_to_calculate(
        self,
        metric_set: Optional[Dict[str, List[str]]] = None
    ) -> List[any]:
        """
        Get list of metric instances to calculate.

        Args:
            metric_set: Dictionary mapping dimension names to lists of metric names.
                       If None, uses all available metrics.

        Returns:
            List of metric instances
        """
        metrics = []

        if metric_set is None:
            # Use all available metrics
            all_metric_names = self.metric_registry.list_metrics()
            for metric_name in all_metric_names:
                try:
                    metric = self.metric_registry.get_metric(metric_name)
                    metrics.append(metric)
                except Exception as e:
                    logger.warning(
                        f"Failed to create metric {metric_name}: {str(e)}"
                    )
        else:
            # Use specified metrics
            for dimension_name, metric_names in metric_set.items():
                for metric_name in metric_names:
                    try:
                        metric = self.metric_registry.get_metric(metric_name)
                        metrics.append(metric)
                    except Exception as e:
                        logger.warning(
                            f"Failed to create metric {metric_name}: {str(e)}"
                        )

        return metrics

    def evaluate_with_custom_metrics(
        self,
        trace: AgentTrace,
        metrics: List[str],
        ground_truth: Optional[GroundTruth] = None,
        dimension_weights: Optional[Dict[str, float]] = None,
        thresholds: Optional[Dict[str, float]] = None,
        **kwargs
    ) -> EvaluationResult:
        """
        Evaluate a trace with a custom list of metrics.

        Args:
            trace: Agent trace to evaluate
            metrics: List of metric names to calculate
            ground_truth: Expected correct outputs (optional)
            dimension_weights: Weights for each dimension (uses defaults if None)
            thresholds: Threshold values for pass/fail (uses defaults if None)
            **kwargs: Additional evaluation parameters

        Returns:
            EvaluationResult with scores and metadata
        """
        # Group metrics by dimension
        metric_set = {}

        for metric_name in metrics:
            try:
                metric_info = self.metric_registry.get_metric_info(metric_name)
                dimension = metric_info.get("dimension", "unknown")

                if dimension not in metric_set:
                    metric_set[dimension] = []

                metric_set[dimension].append(metric_name)

            except KeyError:
                logger.warning(f"Metric {metric_name} not found in registry")

        return self.evaluate(
            trace=trace,
            ground_truth=ground_truth,
            metric_set=metric_set,
            dimension_weights=dimension_weights,
            thresholds=thresholds,
            **kwargs
        )

evaluate(trace, ground_truth=None, metric_set=None, dimension_weights=None, thresholds=None, context=None, **kwargs)

Evaluate a single agent trace.

Parameters:

Name	Type	Description	Default
trace	AgentTrace	Agent trace to evaluate	required
ground_truth	Optional[GroundTruth]	Expected correct outputs (optional)	None
metric_set	Optional[Dict[str, List[str]]]	Dictionary mapping dimension names to lists of metric names. If None, uses all available metrics.	None
dimension_weights	Optional[Dict[str, float]]	Weights for each dimension (uses defaults if None)	None
thresholds	Optional[Dict[str, float]]	Threshold values for pass/fail (uses defaults if None)	None
context	Optional[List[str]]	List of context documents (optional)	None
**kwargs		Additional evaluation parameters	{}

Returns:

Type	Description
EvaluationResult	EvaluationResult with scores and metadata

Raises:

Type	Description
ValueError	If inputs are invalid
RuntimeError	If evaluation fails

Source code in uaef/evaluation/single_agent.py

def evaluate(
    self,
    trace: AgentTrace,
    ground_truth: Optional[GroundTruth] = None,
    metric_set: Optional[Dict[str, List[str]]] = None,
    dimension_weights: Optional[Dict[str, float]] = None,
    thresholds: Optional[Dict[str, float]] = None,
    context: Optional[List[str]] = None,
    **kwargs
) -> EvaluationResult:
    """
    Evaluate a single agent trace.

    Args:
        trace: Agent trace to evaluate
        ground_truth: Expected correct outputs (optional)
        metric_set: Dictionary mapping dimension names to lists of metric names.
                   If None, uses all available metrics.
        dimension_weights: Weights for each dimension (uses defaults if None)
        thresholds: Threshold values for pass/fail (uses defaults if None)
        context: List of context documents (optional)
        **kwargs: Additional evaluation parameters

    Returns:
        EvaluationResult with scores and metadata

    Raises:
        ValueError: If inputs are invalid
        RuntimeError: If evaluation fails
    """
    logger.info(f"Evaluating trace {trace.trace_id}")

    # Validate inputs
    if not isinstance(trace, AgentTrace):
        raise ValueError("trace must be an AgentTrace instance")

    # Use default weights and thresholds if not provided
    dimension_weights = dimension_weights or self.default_dimension_weights
    thresholds = thresholds or self.default_thresholds

    # Create evaluation input
    evaluation_input = EvaluationInput(
        trace=trace,
        ground_truth=ground_truth,
        context=context or (ground_truth.context_documents if ground_truth else [])
    )

    # Determine which metrics to calculate
    metrics_to_calculate = self._get_metrics_to_calculate(metric_set)

    logger.debug(f"Calculating {len(metrics_to_calculate)} metrics")

    # Calculate metrics with automatic dependency resolution
    metric_scores = self.calculate_metrics_with_dependencies(
        metrics=metrics_to_calculate,
        evaluation_input=evaluation_input
    )

    logger.debug(f"Calculated {len(metric_scores)} metric scores")

    # Aggregate scores by dimension
    dimension_results = self.aggregate_dimension_scores(
        metric_scores=metric_scores,
        dimension_weights=dimension_weights
    )

    logger.debug(f"Aggregated scores for {len(dimension_results)} dimensions")

    # Calculate overall score
    overall_score = self.calculate_overall_score(dimension_results)

    logger.debug(f"Overall score: {overall_score:.3f}")

    # Check thresholds
    passed, failures = self.check_thresholds(
        dimension_results=dimension_results,
        thresholds=thresholds
    )

    # Collect warnings
    warnings = []
    if ground_truth is None:
        gt_metrics = self.metric_registry.get_metrics_requiring_ground_truth()
        if any(m.metric_name in gt_metrics for m in metric_scores):
            warnings.append(
                "Ground truth not provided, some metrics may be skipped"
            )

    warnings.extend(self._metric_error_warnings(metric_scores))

    # Create evaluation result
    result = EvaluationResult(
        trace_id=trace.trace_id,
        dimension_results=dimension_results,
        overall_score=overall_score,
        passed=passed,
        failures=failures,
        warnings=warnings,
        metadata={
            "num_metrics": len(metric_scores),
            "num_dimensions": len(dimension_results),
            "dimension_weights": dimension_weights,
            "thresholds": thresholds
        }
    )

    logger.info(
        f"Evaluation complete for trace {trace.trace_id}: "
        f"score={overall_score:.3f}, passed={passed}"
    )

    return result

evaluate_multi_turn(per_turn_traces, full_trace, ground_truth=None, metric_set=None, dimension_weights=None, thresholds=None, context=None, **kwargs)

Evaluate a single-agent multi-turn conversation from pre-split per-turn traces.

Both per_turn_traces and full_trace must be AgentTrace instances. Multi-agent traces are handled by MultiAgentEvaluator.evaluate_multi_turn().

Flow: 1. Iterate per_turn_traces, run per-turn-eligible metrics on each (using ground_truth.expected_arguments["per_turn_expected"][i] and ["per_turn_tools"][i] for that turn's ground truth) 2. Average per-turn scores across turns 3. Run full-trace-only metrics (multi-turn + throughput) on full_trace 4. Combine and return EvaluationResult with per-turn breakdown in metadata

Parameters:

Name	Type	Description	Default
per_turn_traces	List[AgentTrace]	List of per-turn AgentTraces (one per user turn)	required
full_trace	AgentTrace	AgentTrace covering the entire conversation	required
ground_truth	Optional[GroundTruth]	Expected correct outputs (optional). May contain per_turn_expected and per_turn_tools arrays in expected_arguments.	None
metric_set	Optional[Dict[str, List[str]]]	Dictionary mapping dimension names to lists of metric names	None
dimension_weights	Optional[Dict[str, float]]	Weights for each dimension (uses defaults if None)	None
thresholds	Optional[Dict[str, float]]	Threshold values for pass/fail (uses defaults if None)	None
context	Optional[List[str]]	List of context documents (optional)	None
**kwargs		Additional evaluation parameters	{}

Returns:

Type	Description
EvaluationResult	EvaluationResult with scores and metadata (including per_turn breakdown)

Source code in uaef/evaluation/single_agent.py

def evaluate_multi_turn(
    self,
    per_turn_traces: List[AgentTrace],
    full_trace: AgentTrace,
    ground_truth: Optional[GroundTruth] = None,
    metric_set: Optional[Dict[str, List[str]]] = None,
    dimension_weights: Optional[Dict[str, float]] = None,
    thresholds: Optional[Dict[str, float]] = None,
    context: Optional[List[str]] = None,
    **kwargs
) -> EvaluationResult:
    """
    Evaluate a single-agent multi-turn conversation from pre-split per-turn traces.

    Both per_turn_traces and full_trace must be AgentTrace instances.
    Multi-agent traces are handled by MultiAgentEvaluator.evaluate_multi_turn().

    Flow:
    1. Iterate per_turn_traces, run per-turn-eligible metrics on each
       (using `ground_truth.expected_arguments["per_turn_expected"][i]` and
       `["per_turn_tools"][i]` for that turn's ground truth)
    2. Average per-turn scores across turns
    3. Run full-trace-only metrics (multi-turn + throughput) on full_trace
    4. Combine and return EvaluationResult with per-turn breakdown in metadata

    Args:
        per_turn_traces: List of per-turn AgentTraces (one per user turn)
        full_trace: AgentTrace covering the entire conversation
        ground_truth: Expected correct outputs (optional). May contain
                     per_turn_expected and per_turn_tools arrays in expected_arguments.
        metric_set: Dictionary mapping dimension names to lists of metric names
        dimension_weights: Weights for each dimension (uses defaults if None)
        thresholds: Threshold values for pass/fail (uses defaults if None)
        context: List of context documents (optional)
        **kwargs: Additional evaluation parameters

    Returns:
        EvaluationResult with scores and metadata (including per_turn breakdown)
    """
    if not per_turn_traces:
        raise ValueError("per_turn_traces must contain at least one trace")
    if not isinstance(full_trace, AgentTrace) or isinstance(full_trace, MultiAgentTrace):
        raise ValueError(
            "full_trace must be an AgentTrace instance "
            "(use MultiAgentEvaluator.evaluate_multi_turn for MultiAgentTrace)"
        )
    for i, t in enumerate(per_turn_traces):
        if not isinstance(t, AgentTrace) or isinstance(t, MultiAgentTrace):
            raise ValueError(
                f"per_turn_traces[{i}] must be an AgentTrace instance "
                "(use MultiAgentEvaluator.evaluate_multi_turn for MultiAgentTrace)"
            )

    return self._run_multi_turn(
        per_turn_traces=per_turn_traces,
        full_trace=full_trace,
        ground_truth=ground_truth,
        metric_set=metric_set,
        dimension_weights=dimension_weights,
        thresholds=thresholds,
        context=context,
        **kwargs,
    )

evaluate_trace(trace, ground_truth=None, metric_set=None, **kwargs)

Alias for evaluate() method for backward compatibility.

Parameters:

Name	Type	Description	Default
trace	AgentTrace	Agent trace to evaluate	required
ground_truth	Optional[GroundTruth]	Expected correct outputs (optional)	None
metric_set	Optional[Dict[str, List[str]]]	Dictionary mapping dimension names to lists of metric names	None
**kwargs		Additional evaluation parameters	{}

Returns:

Type	Description
EvaluationResult	EvaluationResult with scores and metadata

Source code in uaef/evaluation/single_agent.py

def evaluate_trace(
    self,
    trace: AgentTrace,
    ground_truth: Optional[GroundTruth] = None,
    metric_set: Optional[Dict[str, List[str]]] = None,
    **kwargs
) -> EvaluationResult:
    """
    Alias for evaluate() method for backward compatibility.

    Args:
        trace: Agent trace to evaluate
        ground_truth: Expected correct outputs (optional)
        metric_set: Dictionary mapping dimension names to lists of metric names
        **kwargs: Additional evaluation parameters

    Returns:
        EvaluationResult with scores and metadata
    """
    return self.evaluate(
        trace=trace,
        ground_truth=ground_truth,
        metric_set=metric_set,
        **kwargs
    )

evaluate_with_custom_metrics(trace, metrics, ground_truth=None, dimension_weights=None, thresholds=None, **kwargs)

Evaluate a trace with a custom list of metrics.

Parameters:

Name	Type	Description	Default
trace	AgentTrace	Agent trace to evaluate	required
metrics	List[str]	List of metric names to calculate	required
ground_truth	Optional[GroundTruth]	Expected correct outputs (optional)	None
dimension_weights	Optional[Dict[str, float]]	Weights for each dimension (uses defaults if None)	None
thresholds	Optional[Dict[str, float]]	Threshold values for pass/fail (uses defaults if None)	None
**kwargs		Additional evaluation parameters	{}

Returns:

Type	Description
EvaluationResult	EvaluationResult with scores and metadata

Source code in uaef/evaluation/single_agent.py

def evaluate_with_custom_metrics(
    self,
    trace: AgentTrace,
    metrics: List[str],
    ground_truth: Optional[GroundTruth] = None,
    dimension_weights: Optional[Dict[str, float]] = None,
    thresholds: Optional[Dict[str, float]] = None,
    **kwargs
) -> EvaluationResult:
    """
    Evaluate a trace with a custom list of metrics.

    Args:
        trace: Agent trace to evaluate
        metrics: List of metric names to calculate
        ground_truth: Expected correct outputs (optional)
        dimension_weights: Weights for each dimension (uses defaults if None)
        thresholds: Threshold values for pass/fail (uses defaults if None)
        **kwargs: Additional evaluation parameters

    Returns:
        EvaluationResult with scores and metadata
    """
    # Group metrics by dimension
    metric_set = {}

    for metric_name in metrics:
        try:
            metric_info = self.metric_registry.get_metric_info(metric_name)
            dimension = metric_info.get("dimension", "unknown")

            if dimension not in metric_set:
                metric_set[dimension] = []

            metric_set[dimension].append(metric_name)

        except KeyError:
            logger.warning(f"Metric {metric_name} not found in registry")

    return self.evaluate(
        trace=trace,
        ground_truth=ground_truth,
        metric_set=metric_set,
        dimension_weights=dimension_weights,
        thresholds=thresholds,
        **kwargs
    )

uaef.evaluation.multi_turn

Multi-turn conversation evaluator for UAEF.

This module provides evaluation functionality for multi-turn conversations, assessing context retention, coherence, and goal achievement across turns.

MultiTurnEvaluator

Bases: BaseEvaluator

Evaluator for multi-turn conversations.

Specializes in evaluating: - Context retention across turns - Coherence and logical flow - Conversation completeness and goal achievement - Turn efficiency

Requirements: 10.1-10.6

Source code in uaef/evaluation/multi_turn.py

class MultiTurnEvaluator(BaseEvaluator):
    """
    Evaluator for multi-turn conversations.

    Specializes in evaluating:
    - Context retention across turns
    - Coherence and logical flow
    - Conversation completeness and goal achievement
    - Turn efficiency

    Requirements: 10.1-10.6
    """

    def __init__(
        self,
        metric_registry: Optional[any] = None,
        max_workers: int = 4,
        default_dimension_weights: Optional[Dict[str, float]] = None,
        default_thresholds: Optional[Dict[str, float]] = None
    ):
        """
        Initialize multi-turn evaluator.

        Args:
            metric_registry: MetricRegistry instance (uses global if None)
            max_workers: Maximum number of parallel workers for metric calculation
            default_dimension_weights: Default weights for dimensions
            default_thresholds: Default threshold values for dimensions
        """
        super().__init__(metric_registry=metric_registry, max_workers=max_workers)

        # Set default dimension weights focused on multi-turn metrics
        self.default_dimension_weights = default_dimension_weights or {
            "Multi-Turn": 1.0  # Focus on multi-turn dimension
        }

        # Set default thresholds
        self.default_thresholds = default_thresholds or {
            "Multi-Turn": 0.7
        }

        logger.info("Initialized MultiTurnEvaluator")

    def evaluate_conversation(
        self,
        messages: List[Message],
        goal: Optional[str] = None,
        ground_truth: Optional[GroundTruth] = None,
        metric_set: Optional[Dict[str, List[str]]] = None,
        dimension_weights: Optional[Dict[str, float]] = None,
        thresholds: Optional[Dict[str, float]] = None,
        context: Optional[List[str]] = None,
        **kwargs
    ) -> EvaluationResult:
        """
        Evaluate a multi-turn conversation.

        Requirements:
        - 10.1: Assess context retention across turns
        - 10.2: Evaluate coherence by checking logical flow
        - 10.3: Evaluate conversation completeness
        - 10.4: Verify information from earlier turns is used appropriately
        - 10.5: Support goal-based evaluation
        - 10.6: Identify which success criteria were not met

        Args:
            messages: List of conversation messages
            goal: User goal for the conversation (optional)
            ground_truth: Expected correct outputs (optional)
            metric_set: Dictionary mapping dimension names to lists of metric names.
                       If None, uses all multi-turn metrics.
            dimension_weights: Weights for each dimension (uses defaults if None)
            thresholds: Threshold values for pass/fail (uses defaults if None)
            context: List of context documents (optional)
            **kwargs: Additional evaluation parameters

        Returns:
            EvaluationResult with scores and metadata

        Raises:
            ValueError: If inputs are invalid
            RuntimeError: If evaluation fails
        """
        logger.info(f"Evaluating conversation with {len(messages)} messages")

        # Validate inputs
        if not messages:
            raise ValueError("messages list cannot be empty")

        if not all(isinstance(m, Message) for m in messages):
            raise ValueError("All items in messages must be Message instances")

        # Use default weights and thresholds if not provided
        dimension_weights = dimension_weights or self.default_dimension_weights
        thresholds = thresholds or self.default_thresholds

        # Create a trace from messages for evaluation
        import uuid
        trace = AgentTrace(
            trace_id=str(uuid.uuid4()),
            messages=messages,
            tool_calls=[],  # Tool calls are embedded in messages
            metadata={
                "conversation_goal": goal,
                "num_turns": len(messages)
            }
        )

        # Create evaluation input
        evaluation_input = EvaluationInput(
            trace=trace,
            ground_truth=ground_truth,
            context=context or []
        )

        # Determine which metrics to calculate
        # Default to multi-turn metrics if not specified
        if metric_set is None:
            metric_set = {"multi_turn": self._get_default_multi_turn_metrics()}

        metrics_to_calculate = self._get_metrics_to_calculate(metric_set)

        logger.debug(f"Calculating {len(metrics_to_calculate)} metrics")

        # Calculate all metrics in parallel
        metric_scores = self.calculate_metrics_parallel(
            metrics=metrics_to_calculate,
            evaluation_input=evaluation_input
        )

        logger.debug(f"Calculated {len(metric_scores)} metric scores")

        # Aggregate scores by dimension
        dimension_results = self.aggregate_dimension_scores(
            metric_scores=metric_scores,
            dimension_weights=dimension_weights
        )

        logger.debug(f"Aggregated scores for {len(dimension_results)} dimensions")

        # Calculate overall score
        overall_score = self.calculate_overall_score(dimension_results)

        logger.debug(f"Overall score: {overall_score:.3f}")

        # Check thresholds
        passed, failures = self.check_thresholds(
            dimension_results=dimension_results,
            thresholds=thresholds
        )

        # Collect warnings
        warnings = []
        if goal is None:
            warnings.append(
                "Conversation goal not provided, completeness evaluation may be limited"
            )

        if ground_truth is None:
            warnings.append(
                "Ground truth not provided, some metrics may be skipped"
            )

        warnings.extend(self._metric_error_warnings(metric_scores))

        # Identify unmet success criteria (Requirement 10.6)
        unmet_criteria = self._identify_unmet_criteria(
            metric_scores=metric_scores,
            goal=goal,
            passed=passed
        )

        if unmet_criteria:
            failures.extend(unmet_criteria)

        # Create evaluation result
        result = EvaluationResult(
            trace_id=trace.trace_id,
            dimension_results=dimension_results,
            overall_score=overall_score,
            passed=passed,
            failures=failures,
            warnings=warnings,
            metadata={
                "num_metrics": len(metric_scores),
                "num_dimensions": len(dimension_results),
                "num_turns": len(messages),
                "conversation_goal": goal,
                "dimension_weights": dimension_weights,
                "thresholds": thresholds
            }
        )

        logger.info(
            f"Conversation evaluation complete: "
            f"score={overall_score:.3f}, passed={passed}, turns={len(messages)}"
        )

        return result

    def evaluate(
        self,
        trace: AgentTrace,
        ground_truth: Optional[GroundTruth] = None,
        metric_set: Optional[Dict[str, List[str]]] = None,
        dimension_weights: Optional[Dict[str, float]] = None,
        thresholds: Optional[Dict[str, float]] = None,
        context: Optional[List[str]] = None,
        **kwargs
    ) -> EvaluationResult:
        """
        Evaluate a multi-turn conversation from an AgentTrace.

        This is an alternative interface that accepts an AgentTrace directly.

        Args:
            trace: Agent trace containing conversation messages
            ground_truth: Expected correct outputs (optional)
            metric_set: Dictionary mapping dimension names to lists of metric names
            dimension_weights: Weights for each dimension (uses defaults if None)
            thresholds: Threshold values for pass/fail (uses defaults if None)
            context: List of context documents (optional)
            **kwargs: Additional evaluation parameters (can include 'goal')

        Returns:
            EvaluationResult with scores and metadata
        """
        # Extract goal from kwargs or trace metadata
        goal = kwargs.get("goal")
        if goal is None and trace.metadata:
            goal = trace.metadata.get("conversation_goal")

        return self.evaluate_conversation(
            messages=trace.messages,
            goal=goal,
            ground_truth=ground_truth,
            metric_set=metric_set,
            dimension_weights=dimension_weights,
            thresholds=thresholds,
            context=context,
            **kwargs
        )

    def evaluate_context_retention(
        self,
        messages: List[Message],
        ground_truth: Optional[GroundTruth] = None
    ) -> MetricScore:
        """
        Evaluate context retention across conversation turns.

        Requirement 10.1: Assess context retention across turns
        Requirement 10.4: Verify information from earlier turns is used appropriately

        Args:
            messages: List of conversation messages
            ground_truth: Expected correct outputs (optional)

        Returns:
            MetricScore for context retention
        """
        logger.debug("Evaluating context retention")

        # Create trace and evaluation input
        import uuid
        trace = AgentTrace(
            trace_id=str(uuid.uuid4()),
            messages=messages,
            tool_calls=[],
            metadata={"num_turns": len(messages)}
        )

        evaluation_input = EvaluationInput(
            trace=trace,
            ground_truth=ground_truth,
            context=[]
        )

        # Get context retention metric
        try:
            metric = self.metric_registry.get_metric("context_retention")
            score = metric.calculate(evaluation_input)
            logger.debug(f"Context retention score: {score.score:.3f}")
            return score
        except Exception as e:
            logger.error(f"Failed to calculate context retention: {str(e)}")
            return MetricScore(
                metric_name="context_retention",
                score=0.0,
                reasoning=f"Error calculating context retention: {str(e)}"
            )

    def evaluate_coherence(
        self,
        messages: List[Message]
    ) -> MetricScore:
        """
        Evaluate coherence and logical flow of conversation.

        Requirement 10.2: Evaluate coherence by checking logical flow and consistency

        Args:
            messages: List of conversation messages

        Returns:
            MetricScore for coherence
        """
        logger.debug("Evaluating coherence")

        # Create trace and evaluation input
        import uuid
        trace = AgentTrace(
            trace_id=str(uuid.uuid4()),
            messages=messages,
            tool_calls=[],
            metadata={"num_turns": len(messages)}
        )

        evaluation_input = EvaluationInput(
            trace=trace,
            ground_truth=None,
            context=[]
        )

        # Get coherence metric
        try:
            metric = self.metric_registry.get_metric("coherence")
            score = metric.calculate(evaluation_input)
            logger.debug(f"Coherence score: {score.score:.3f}")
            return score
        except Exception as e:
            logger.error(f"Failed to calculate coherence: {str(e)}")
            return MetricScore(
                metric_name="coherence",
                score=0.0,
                reasoning=f"Error calculating coherence: {str(e)}"
            )

    def _get_default_multi_turn_metrics(self) -> List[str]:
        """
        Get default list of multi-turn metric names.

        Returns:
            List of metric names for multi-turn evaluation
        """
        return [
            "context_retention",
            "coherence",
            "conversation_completeness",
            "turn_efficiency"
        ]

    def _get_metrics_to_calculate(
        self,
        metric_set: Optional[Dict[str, List[str]]] = None
    ) -> List[any]:
        """
        Get list of metric instances to calculate.

        Args:
            metric_set: Dictionary mapping dimension names to lists of metric names.
                       If None, uses default multi-turn metrics.

        Returns:
            List of metric instances
        """
        metrics = []

        if metric_set is None:
            # Use default multi-turn metrics
            metric_names = self._get_default_multi_turn_metrics()
            for metric_name in metric_names:
                try:
                    metric = self.metric_registry.get_metric(metric_name)
                    metrics.append(metric)
                except Exception as e:
                    logger.warning(
                        f"Failed to create metric {metric_name}: {str(e)}"
                    )
        else:
            # Use specified metrics
            for dimension_name, metric_names in metric_set.items():
                for metric_name in metric_names:
                    try:
                        metric = self.metric_registry.get_metric(metric_name)
                        metrics.append(metric)
                    except Exception as e:
                        logger.warning(
                            f"Failed to create metric {metric_name}: {str(e)}"
                        )

        return metrics

    def _identify_unmet_criteria(
        self,
        metric_scores: List[MetricScore],
        goal: Optional[str],
        passed: bool
    ) -> List[str]:
        """
        Identify which success criteria were not met.

        Requirement 10.6: Identify which success criteria were not met

        Args:
            metric_scores: List of calculated metric scores
            goal: Conversation goal (optional)
            passed: Whether overall evaluation passed

        Returns:
            List of unmet criteria descriptions
        """
        unmet_criteria = []

        # Check if conversation was incomplete
        completeness_scores = [
            s for s in metric_scores 
            if s.metric_name == "conversation_completeness" and s.score is not None
        ]

        if completeness_scores:
            completeness_score = completeness_scores[0]
            if completeness_score.score < 0.7:
                unmet_criteria.append(
                    f"Conversation completeness below threshold: "
                    f"{completeness_score.score:.3f} < 0.7"
                )
                if goal:
                    unmet_criteria.append(
                        f"Goal not achieved: {goal}"
                    )

        # Check context retention
        context_scores = [
            s for s in metric_scores 
            if s.metric_name == "context_retention" and s.score is not None
        ]

        if context_scores:
            context_score = context_scores[0]
            if context_score.score < 0.6:
                unmet_criteria.append(
                    f"Context retention insufficient: "
                    f"{context_score.score:.3f} < 0.6"
                )

        # Check coherence
        coherence_scores = [
            s for s in metric_scores 
            if s.metric_name == "coherence" and s.score is not None
        ]

        if coherence_scores:
            coherence_score = coherence_scores[0]
            if coherence_score.score < 0.6:
                unmet_criteria.append(
                    f"Coherence insufficient: "
                    f"{coherence_score.score:.3f} < 0.6"
                )

        return unmet_criteria

evaluate(trace, ground_truth=None, metric_set=None, dimension_weights=None, thresholds=None, context=None, **kwargs)

Evaluate a multi-turn conversation from an AgentTrace.

This is an alternative interface that accepts an AgentTrace directly.

Parameters:

Name	Type	Description	Default
trace	AgentTrace	Agent trace containing conversation messages	required
ground_truth	Optional[GroundTruth]	Expected correct outputs (optional)	None
metric_set	Optional[Dict[str, List[str]]]	Dictionary mapping dimension names to lists of metric names	None
dimension_weights	Optional[Dict[str, float]]	Weights for each dimension (uses defaults if None)	None
thresholds	Optional[Dict[str, float]]	Threshold values for pass/fail (uses defaults if None)	None
context	Optional[List[str]]	List of context documents (optional)	None
**kwargs		Additional evaluation parameters (can include 'goal')	{}

Returns:

Type	Description
EvaluationResult	EvaluationResult with scores and metadata

Source code in uaef/evaluation/multi_turn.py

def evaluate(
    self,
    trace: AgentTrace,
    ground_truth: Optional[GroundTruth] = None,
    metric_set: Optional[Dict[str, List[str]]] = None,
    dimension_weights: Optional[Dict[str, float]] = None,
    thresholds: Optional[Dict[str, float]] = None,
    context: Optional[List[str]] = None,
    **kwargs
) -> EvaluationResult:
    """
    Evaluate a multi-turn conversation from an AgentTrace.

    This is an alternative interface that accepts an AgentTrace directly.

    Args:
        trace: Agent trace containing conversation messages
        ground_truth: Expected correct outputs (optional)
        metric_set: Dictionary mapping dimension names to lists of metric names
        dimension_weights: Weights for each dimension (uses defaults if None)
        thresholds: Threshold values for pass/fail (uses defaults if None)
        context: List of context documents (optional)
        **kwargs: Additional evaluation parameters (can include 'goal')

    Returns:
        EvaluationResult with scores and metadata
    """
    # Extract goal from kwargs or trace metadata
    goal = kwargs.get("goal")
    if goal is None and trace.metadata:
        goal = trace.metadata.get("conversation_goal")

    return self.evaluate_conversation(
        messages=trace.messages,
        goal=goal,
        ground_truth=ground_truth,
        metric_set=metric_set,
        dimension_weights=dimension_weights,
        thresholds=thresholds,
        context=context,
        **kwargs
    )

evaluate_coherence(messages)

Evaluate coherence and logical flow of conversation.

Requirement 10.2: Evaluate coherence by checking logical flow and consistency

Parameters:

Name	Type	Description	Default
messages	List[Message]	List of conversation messages	required

Returns:

Type	Description
MetricScore	MetricScore for coherence

Source code in uaef/evaluation/multi_turn.py

def evaluate_coherence(
    self,
    messages: List[Message]
) -> MetricScore:
    """
    Evaluate coherence and logical flow of conversation.

    Requirement 10.2: Evaluate coherence by checking logical flow and consistency

    Args:
        messages: List of conversation messages

    Returns:
        MetricScore for coherence
    """
    logger.debug("Evaluating coherence")

    # Create trace and evaluation input
    import uuid
    trace = AgentTrace(
        trace_id=str(uuid.uuid4()),
        messages=messages,
        tool_calls=[],
        metadata={"num_turns": len(messages)}
    )

    evaluation_input = EvaluationInput(
        trace=trace,
        ground_truth=None,
        context=[]
    )

    # Get coherence metric
    try:
        metric = self.metric_registry.get_metric("coherence")
        score = metric.calculate(evaluation_input)
        logger.debug(f"Coherence score: {score.score:.3f}")
        return score
    except Exception as e:
        logger.error(f"Failed to calculate coherence: {str(e)}")
        return MetricScore(
            metric_name="coherence",
            score=0.0,
            reasoning=f"Error calculating coherence: {str(e)}"
        )

evaluate_context_retention(messages, ground_truth=None)

Evaluate context retention across conversation turns.

Requirement 10.1: Assess context retention across turns Requirement 10.4: Verify information from earlier turns is used appropriately

Parameters:

Name	Type	Description	Default
messages	List[Message]	List of conversation messages	required
ground_truth	Optional[GroundTruth]	Expected correct outputs (optional)	None

Returns:

Type	Description
MetricScore	MetricScore for context retention

Source code in uaef/evaluation/multi_turn.py

def evaluate_context_retention(
    self,
    messages: List[Message],
    ground_truth: Optional[GroundTruth] = None
) -> MetricScore:
    """
    Evaluate context retention across conversation turns.

    Requirement 10.1: Assess context retention across turns
    Requirement 10.4: Verify information from earlier turns is used appropriately

    Args:
        messages: List of conversation messages
        ground_truth: Expected correct outputs (optional)

    Returns:
        MetricScore for context retention
    """
    logger.debug("Evaluating context retention")

    # Create trace and evaluation input
    import uuid
    trace = AgentTrace(
        trace_id=str(uuid.uuid4()),
        messages=messages,
        tool_calls=[],
        metadata={"num_turns": len(messages)}
    )

    evaluation_input = EvaluationInput(
        trace=trace,
        ground_truth=ground_truth,
        context=[]
    )

    # Get context retention metric
    try:
        metric = self.metric_registry.get_metric("context_retention")
        score = metric.calculate(evaluation_input)
        logger.debug(f"Context retention score: {score.score:.3f}")
        return score
    except Exception as e:
        logger.error(f"Failed to calculate context retention: {str(e)}")
        return MetricScore(
            metric_name="context_retention",
            score=0.0,
            reasoning=f"Error calculating context retention: {str(e)}"
        )

evaluate_conversation(messages, goal=None, ground_truth=None, metric_set=None, dimension_weights=None, thresholds=None, context=None, **kwargs)

Evaluate a multi-turn conversation.

Requirements: - 10.1: Assess context retention across turns - 10.2: Evaluate coherence by checking logical flow - 10.3: Evaluate conversation completeness - 10.4: Verify information from earlier turns is used appropriately - 10.5: Support goal-based evaluation - 10.6: Identify which success criteria were not met

Parameters:

Name	Type	Description	Default
messages	List[Message]	List of conversation messages	required
goal	Optional[str]	User goal for the conversation (optional)	None
ground_truth	Optional[GroundTruth]	Expected correct outputs (optional)	None
metric_set	Optional[Dict[str, List[str]]]	Dictionary mapping dimension names to lists of metric names. If None, uses all multi-turn metrics.	None
dimension_weights	Optional[Dict[str, float]]	Weights for each dimension (uses defaults if None)	None
thresholds	Optional[Dict[str, float]]	Threshold values for pass/fail (uses defaults if None)	None
context	Optional[List[str]]	List of context documents (optional)	None
**kwargs		Additional evaluation parameters	{}

Returns:

Type	Description
EvaluationResult	EvaluationResult with scores and metadata

Raises:

Type	Description
ValueError	If inputs are invalid
RuntimeError	If evaluation fails

Source code in uaef/evaluation/multi_turn.py

def evaluate_conversation(
    self,
    messages: List[Message],
    goal: Optional[str] = None,
    ground_truth: Optional[GroundTruth] = None,
    metric_set: Optional[Dict[str, List[str]]] = None,
    dimension_weights: Optional[Dict[str, float]] = None,
    thresholds: Optional[Dict[str, float]] = None,
    context: Optional[List[str]] = None,
    **kwargs
) -> EvaluationResult:
    """
    Evaluate a multi-turn conversation.

    Requirements:
    - 10.1: Assess context retention across turns
    - 10.2: Evaluate coherence by checking logical flow
    - 10.3: Evaluate conversation completeness
    - 10.4: Verify information from earlier turns is used appropriately
    - 10.5: Support goal-based evaluation
    - 10.6: Identify which success criteria were not met

    Args:
        messages: List of conversation messages
        goal: User goal for the conversation (optional)
        ground_truth: Expected correct outputs (optional)
        metric_set: Dictionary mapping dimension names to lists of metric names.
                   If None, uses all multi-turn metrics.
        dimension_weights: Weights for each dimension (uses defaults if None)
        thresholds: Threshold values for pass/fail (uses defaults if None)
        context: List of context documents (optional)
        **kwargs: Additional evaluation parameters

    Returns:
        EvaluationResult with scores and metadata

    Raises:
        ValueError: If inputs are invalid
        RuntimeError: If evaluation fails
    """
    logger.info(f"Evaluating conversation with {len(messages)} messages")

    # Validate inputs
    if not messages:
        raise ValueError("messages list cannot be empty")

    if not all(isinstance(m, Message) for m in messages):
        raise ValueError("All items in messages must be Message instances")

    # Use default weights and thresholds if not provided
    dimension_weights = dimension_weights or self.default_dimension_weights
    thresholds = thresholds or self.default_thresholds

    # Create a trace from messages for evaluation
    import uuid
    trace = AgentTrace(
        trace_id=str(uuid.uuid4()),
        messages=messages,
        tool_calls=[],  # Tool calls are embedded in messages
        metadata={
            "conversation_goal": goal,
            "num_turns": len(messages)
        }
    )

    # Create evaluation input
    evaluation_input = EvaluationInput(
        trace=trace,
        ground_truth=ground_truth,
        context=context or []
    )

    # Determine which metrics to calculate
    # Default to multi-turn metrics if not specified
    if metric_set is None:
        metric_set = {"multi_turn": self._get_default_multi_turn_metrics()}

    metrics_to_calculate = self._get_metrics_to_calculate(metric_set)

    logger.debug(f"Calculating {len(metrics_to_calculate)} metrics")

    # Calculate all metrics in parallel
    metric_scores = self.calculate_metrics_parallel(
        metrics=metrics_to_calculate,
        evaluation_input=evaluation_input
    )

    logger.debug(f"Calculated {len(metric_scores)} metric scores")

    # Aggregate scores by dimension
    dimension_results = self.aggregate_dimension_scores(
        metric_scores=metric_scores,
        dimension_weights=dimension_weights
    )

    logger.debug(f"Aggregated scores for {len(dimension_results)} dimensions")

    # Calculate overall score
    overall_score = self.calculate_overall_score(dimension_results)

    logger.debug(f"Overall score: {overall_score:.3f}")

    # Check thresholds
    passed, failures = self.check_thresholds(
        dimension_results=dimension_results,
        thresholds=thresholds
    )

    # Collect warnings
    warnings = []
    if goal is None:
        warnings.append(
            "Conversation goal not provided, completeness evaluation may be limited"
        )

    if ground_truth is None:
        warnings.append(
            "Ground truth not provided, some metrics may be skipped"
        )

    warnings.extend(self._metric_error_warnings(metric_scores))

    # Identify unmet success criteria (Requirement 10.6)
    unmet_criteria = self._identify_unmet_criteria(
        metric_scores=metric_scores,
        goal=goal,
        passed=passed
    )

    if unmet_criteria:
        failures.extend(unmet_criteria)

    # Create evaluation result
    result = EvaluationResult(
        trace_id=trace.trace_id,
        dimension_results=dimension_results,
        overall_score=overall_score,
        passed=passed,
        failures=failures,
        warnings=warnings,
        metadata={
            "num_metrics": len(metric_scores),
            "num_dimensions": len(dimension_results),
            "num_turns": len(messages),
            "conversation_goal": goal,
            "dimension_weights": dimension_weights,
            "thresholds": thresholds
        }
    )

    logger.info(
        f"Conversation evaluation complete: "
        f"score={overall_score:.3f}, passed={passed}, turns={len(messages)}"
    )

    return result

uaef.evaluation.multi_agent

Multi-agent evaluator for UAEF.

This module provides evaluation functionality for multi-agent traces, coordinating metric calculation for multi-agent coordination scenarios.

MultiAgentEvaluator

Bases: BaseEvaluator

Evaluator for multi-agent traces.

Coordinates the complete evaluation flow for multi-agent coordination: 1. Create EvaluationInput from multi-agent trace and ground truth 2. Calculate metrics for each dimension (including multi-agent specific) 3. Aggregate dimension scores with weighted averages 4. Calculate overall score 5. Check thresholds and identify failures

Source code in uaef/evaluation/multi_agent.py

class MultiAgentEvaluator(BaseEvaluator):
    """
    Evaluator for multi-agent traces.

    Coordinates the complete evaluation flow for multi-agent coordination:
    1. Create EvaluationInput from multi-agent trace and ground truth
    2. Calculate metrics for each dimension (including multi-agent specific)
    3. Aggregate dimension scores with weighted averages
    4. Calculate overall score
    5. Check thresholds and identify failures
    """

    def __init__(
        self,
        metric_registry: Optional[any] = None,
        max_workers: int = 4,
        default_dimension_weights: Optional[Dict[str, float]] = None,
        default_thresholds: Optional[Dict[str, float]] = None
    ):
        """
        Initialize multi-agent evaluator.

        Args:
            metric_registry: MetricRegistry instance (uses global if None)
            max_workers: Maximum number of parallel workers for metric calculation
            default_dimension_weights: Default weights for dimensions
            default_thresholds: Default threshold values for dimensions
        """
        super().__init__(metric_registry=metric_registry, max_workers=max_workers)

        # Set default dimension weights if not provided
        self.default_dimension_weights = default_dimension_weights or {
            "Tool Calling": 0.20,
            "Response Quality": 0.20,
            "Responsible AI": 0.15,
            "Performance": 0.10,
            "Multi-Turn": 0.10,
            "Multi-Agent": 0.20,  # Higher weight for multi-agent scenarios
            "Reasoning": 0.05
        }

        # Set default thresholds if not provided
        self.default_thresholds = default_thresholds or {
            "Tool Calling": 0.7,
            "Response Quality": 0.7,
            "Responsible AI": 0.8,
            "Performance": 0.6,
            "Multi-Turn": 0.6,
            "Multi-Agent": 0.7,
            "Reasoning": 0.6
        }

        logger.info(
            f"Initialized MultiAgentEvaluator with "
            f"{len(self.default_dimension_weights)} dimensions"
        )

    def evaluate(
        self,
        trace: MultiAgentTrace,
        ground_truth: Optional[GroundTruth] = None,
        metric_set: Optional[Dict[str, List[str]]] = None,
        dimension_weights: Optional[Dict[str, float]] = None,
        thresholds: Optional[Dict[str, float]] = None,
        context: Optional[List[str]] = None,
        **kwargs
    ) -> EvaluationResult:
        """
        Evaluate a multi-agent trace.

        Args:
            trace: Multi-agent trace to evaluate
            ground_truth: Expected correct outputs (optional)
            metric_set: Dictionary mapping dimension names to lists of metric names.
                       If None, uses all available metrics.
            dimension_weights: Weights for each dimension (uses defaults if None)
            thresholds: Threshold values for pass/fail (uses defaults if None)
            context: List of context documents (optional)
            **kwargs: Additional evaluation parameters

        Returns:
            EvaluationResult with scores and metadata

        Raises:
            ValueError: If inputs are invalid
            RuntimeError: If evaluation fails
        """
        logger.info(f"Evaluating multi-agent trace {trace.trace_id}")

        # Validate inputs
        if not isinstance(trace, MultiAgentTrace):
            raise ValueError("trace must be a MultiAgentTrace instance")

        # Use default weights and thresholds if not provided
        dimension_weights = dimension_weights or self.default_dimension_weights
        thresholds = thresholds or self.default_thresholds

        # Create evaluation input
        evaluation_input = EvaluationInput(
            trace=trace,
            ground_truth=ground_truth,
            context=context or (ground_truth.context_documents if ground_truth else [])
        )

        # Determine which metrics to calculate
        metrics_to_calculate = self._get_metrics_to_calculate(metric_set)

        logger.debug(f"Calculating {len(metrics_to_calculate)} metrics")

        # Calculate metrics with automatic dependency resolution
        metric_scores = self.calculate_metrics_with_dependencies(
            metrics=metrics_to_calculate,
            evaluation_input=evaluation_input
        )

        logger.debug(f"Calculated {len(metric_scores)} metric scores")

        # Aggregate scores by dimension
        dimension_results = self.aggregate_dimension_scores(
            metric_scores=metric_scores,
            dimension_weights=dimension_weights
        )

        logger.debug(f"Aggregated scores for {len(dimension_results)} dimensions")

        # Calculate overall score
        overall_score = self.calculate_overall_score(dimension_results)

        logger.debug(f"Overall score: {overall_score:.3f}")

        # Check thresholds
        passed, failures = self.check_thresholds(
            dimension_results=dimension_results,
            thresholds=thresholds
        )

        # Collect warnings
        warnings = []
        if ground_truth is None:
            gt_metrics = self.metric_registry.get_metrics_requiring_ground_truth()
            if any(m.metric_name in gt_metrics for m in metric_scores):
                warnings.append(
                    "Ground truth not provided, some metrics may be skipped"
                )

        warnings.extend(self._metric_error_warnings(metric_scores))

        # Create evaluation result
        result = EvaluationResult(
            trace_id=trace.trace_id,
            dimension_results=dimension_results,
            overall_score=overall_score,
            passed=passed,
            failures=failures,
            warnings=warnings,
            metadata={
                "num_metrics": len(metric_scores),
                "num_dimensions": len(dimension_results),
                "num_agents": len(trace.agent_traces),
                "workflow_status": trace.workflow_status.value,
                "dimension_weights": dimension_weights,
                "thresholds": thresholds
            }
        )

        logger.info(
            f"Evaluation complete for trace {trace.trace_id}: "
            f"score={overall_score:.3f}, passed={passed}"
        )

        return result

    def evaluate_multi_turn(
        self,
        per_turn_traces: List[MultiAgentTrace],
        full_trace: MultiAgentTrace,
        ground_truth: Optional[GroundTruth] = None,
        metric_set: Optional[Dict[str, List[str]]] = None,
        dimension_weights: Optional[Dict[str, float]] = None,
        thresholds: Optional[Dict[str, float]] = None,
        context: Optional[List[str]] = None,
        **kwargs,
    ) -> EvaluationResult:
        """
        Evaluate a multi-agent multi-turn conversation from pre-split per-turn traces.

        Both per_turn_traces and full_trace must be MultiAgentTrace instances.
        Single-agent (AgentTrace) inputs are handled by
        SingleAgentEvaluator.evaluate_multi_turn().

        Per-turn metrics (tool calling, response quality, multi-agent, ...)
        run on each MultiAgentTrace turn and are averaged. Full-trace metrics
        (context_retention, coherence, ...) run on the full MultiAgentTrace,
        whose `messages` property aggregates sub-agent messages in order.

        Args:
            per_turn_traces: List of per-turn MultiAgentTraces (one per user turn)
            full_trace: MultiAgentTrace covering the entire conversation
            ground_truth: Expected correct outputs (optional). May contain
                         per_turn_expected and per_turn_tools arrays in expected_arguments.
            metric_set: Dictionary mapping dimension names to lists of metric names
            dimension_weights: Weights for each dimension (uses defaults if None)
            thresholds: Threshold values for pass/fail (uses defaults if None)
            context: List of context documents (optional)
            **kwargs: Additional evaluation parameters

        Returns:
            EvaluationResult with per-turn breakdown in metadata
        """
        if not per_turn_traces:
            raise ValueError("per_turn_traces must contain at least one trace")
        if not isinstance(full_trace, MultiAgentTrace):
            raise ValueError(
                "full_trace must be a MultiAgentTrace instance "
                "(use SingleAgentEvaluator.evaluate_multi_turn for AgentTrace)"
            )
        for i, t in enumerate(per_turn_traces):
            if not isinstance(t, MultiAgentTrace):
                raise ValueError(
                    f"per_turn_traces[{i}] must be a MultiAgentTrace instance "
                    "(use SingleAgentEvaluator.evaluate_multi_turn for AgentTrace)"
                )

        # Reuse the shared multi-turn loop on BaseEvaluator. Because
        # self._get_metrics_to_calculate / self.default_dimension_weights /
        # self.default_thresholds are all overridden on this class, the loop
        # picks up the multi-agent metric registry and weights automatically.
        return self._run_multi_turn(
            per_turn_traces=per_turn_traces,
            full_trace=full_trace,
            ground_truth=ground_truth,
            metric_set=metric_set,
            dimension_weights=dimension_weights,
            thresholds=thresholds,
            context=context,
            **kwargs,
        )

    def _get_metrics_to_calculate(
        self,
        metric_set: Optional[Dict[str, List[str]]] = None
    ) -> List[any]:
        """
        Get list of metric instances to calculate.

        Args:
            metric_set: Dictionary mapping dimension names to lists of metric names.
                       If None, uses all available metrics.

        Returns:
            List of metric instances
        """
        metrics = []

        if metric_set is None:
            # Use all available metrics
            all_metric_names = self.metric_registry.list_metrics()
            for metric_name in all_metric_names:
                try:
                    metric = self.metric_registry.get_metric(metric_name)
                    metrics.append(metric)
                except Exception as e:
                    logger.warning(
                        f"Failed to create metric {metric_name}: {str(e)}"
                    )
        else:
            # Use specified metrics
            for dimension_name, metric_names in metric_set.items():
                for metric_name in metric_names:
                    try:
                        metric = self.metric_registry.get_metric(metric_name)
                        metrics.append(metric)
                    except Exception as e:
                        logger.warning(
                            f"Failed to create metric {metric_name}: {str(e)}"
                        )

        return metrics

    def batch_evaluate(
        self,
        traces: List[MultiAgentTrace],
        ground_truths: Optional[List[Optional[GroundTruth]]] = None,
        metric_set: Optional[Dict[str, List[str]]] = None,
        **kwargs
    ) -> List[EvaluationResult]:
        """
        Evaluate multiple multi-agent traces in batch.

        Automatically calculates batch-level metrics for metrics that support it
        (e.g., agent_utilization_batch).

        Args:
            traces: List of multi-agent traces to evaluate
            ground_truths: List of ground truths (optional, can be None for each)
            metric_set: Dictionary mapping dimension names to lists of metric names
            **kwargs: Additional evaluation parameters

        Returns:
            List of EvaluationResult objects, with batch metrics in results[0].metadata

        Raises:
            ValueError: If inputs are invalid
        """
        # Call parent batch_evaluate
        results = super().batch_evaluate(traces, ground_truths, metric_set, **kwargs)

        # Calculate batch-level metrics
        self._calculate_batch_metrics(traces, ground_truths, results, metric_set)

        return results

    def _calculate_batch_metrics(
        self,
        traces: List[MultiAgentTrace],
        ground_truths: List[Optional[GroundTruth]],
        results: List[EvaluationResult],
        metric_set: Optional[Dict[str, List[str]]]
    ) -> None:
        """
        Calculate batch-level metrics for multi-agent scenarios.

        Automatically detects which metrics have calculate_batch() static method
        and calls them to generate aggregate statistics across the batch.

        Args:
            traces: List of multi-agent traces
            ground_truths: List of ground truths
            results: List of evaluation results (will be updated with batch metrics)
            metric_set: Metric set configuration
        """
        from uaef.models.evaluation_input import EvaluationInput

        # Get metrics to calculate
        metrics_to_calculate = self._get_metrics_to_calculate(metric_set)

        # Find metrics that support batch calculation
        batch_metrics = []
        for metric in metrics_to_calculate:
            if hasattr(metric.__class__, 'calculate_batch') and callable(getattr(metric.__class__, 'calculate_batch')):
                batch_metrics.append(metric)

        if not batch_metrics:
            logger.debug("No metrics support batch calculation")
            return

        logger.debug(f"Found {len(batch_metrics)} metrics supporting batch calculation")

        # Build evaluation inputs
        if ground_truths is None:
            ground_truths = [None] * len(traces)

        evaluation_inputs = []
        for trace, gt in zip(traces, ground_truths):
            eval_input = EvaluationInput(trace=trace, ground_truth=gt, context=[])
            evaluation_inputs.append(eval_input)

        # Calculate batch metrics
        batch_scores = []
        for metric in batch_metrics:
            try:
                # Call the static calculate_batch method
                batch_score = metric.__class__.calculate_batch(evaluation_inputs)
                if batch_score.score is not None:
                    batch_scores.append(batch_score)
                    logger.info(f"Calculated batch metric: {batch_score.metric_name} = {batch_score.score:.3f}")
            except Exception as e:
                logger.error(f"Failed to calculate batch metric {metric.get_name()}: {str(e)}")

        # Store batch metrics in first result's metadata
        if results and batch_scores:
            if "batch_metrics" not in results[0].metadata:
                results[0].metadata["batch_metrics"] = {}

            for batch_score in batch_scores:
                results[0].metadata["batch_metrics"][batch_score.metric_name] = {
                    "score": batch_score.score,
                    "reasoning": batch_score.reasoning,
                    "metadata": batch_score.metadata
                }

            logger.info(f"Stored {len(batch_scores)} batch-level metrics in results")

batch_evaluate(traces, ground_truths=None, metric_set=None, **kwargs)

Evaluate multiple multi-agent traces in batch.

Automatically calculates batch-level metrics for metrics that support it (e.g., agent_utilization_batch).

Parameters:

Name	Type	Description	Default
traces	List[MultiAgentTrace]	List of multi-agent traces to evaluate	required
ground_truths	Optional[List[Optional[GroundTruth]]]	List of ground truths (optional, can be None for each)	None
metric_set	Optional[Dict[str, List[str]]]	Dictionary mapping dimension names to lists of metric names	None
**kwargs		Additional evaluation parameters	{}

Returns:

Type	Description
List[EvaluationResult]	List of EvaluationResult objects, with batch metrics in results[0].metadata

Raises:

Type	Description
ValueError	If inputs are invalid

Source code in uaef/evaluation/multi_agent.py

def batch_evaluate(
    self,
    traces: List[MultiAgentTrace],
    ground_truths: Optional[List[Optional[GroundTruth]]] = None,
    metric_set: Optional[Dict[str, List[str]]] = None,
    **kwargs
) -> List[EvaluationResult]:
    """
    Evaluate multiple multi-agent traces in batch.

    Automatically calculates batch-level metrics for metrics that support it
    (e.g., agent_utilization_batch).

    Args:
        traces: List of multi-agent traces to evaluate
        ground_truths: List of ground truths (optional, can be None for each)
        metric_set: Dictionary mapping dimension names to lists of metric names
        **kwargs: Additional evaluation parameters

    Returns:
        List of EvaluationResult objects, with batch metrics in results[0].metadata

    Raises:
        ValueError: If inputs are invalid
    """
    # Call parent batch_evaluate
    results = super().batch_evaluate(traces, ground_truths, metric_set, **kwargs)

    # Calculate batch-level metrics
    self._calculate_batch_metrics(traces, ground_truths, results, metric_set)

    return results

evaluate(trace, ground_truth=None, metric_set=None, dimension_weights=None, thresholds=None, context=None, **kwargs)

Evaluate a multi-agent trace.

Parameters:

Name	Type	Description	Default
trace	MultiAgentTrace	Multi-agent trace to evaluate	required
ground_truth	Optional[GroundTruth]	Expected correct outputs (optional)	None
metric_set	Optional[Dict[str, List[str]]]	Dictionary mapping dimension names to lists of metric names. If None, uses all available metrics.	None
dimension_weights	Optional[Dict[str, float]]	Weights for each dimension (uses defaults if None)	None
thresholds	Optional[Dict[str, float]]	Threshold values for pass/fail (uses defaults if None)	None
context	Optional[List[str]]	List of context documents (optional)	None
**kwargs		Additional evaluation parameters	{}

Returns:

Type	Description
EvaluationResult	EvaluationResult with scores and metadata

Raises:

Type	Description
ValueError	If inputs are invalid
RuntimeError	If evaluation fails

Source code in uaef/evaluation/multi_agent.py

def evaluate(
    self,
    trace: MultiAgentTrace,
    ground_truth: Optional[GroundTruth] = None,
    metric_set: Optional[Dict[str, List[str]]] = None,
    dimension_weights: Optional[Dict[str, float]] = None,
    thresholds: Optional[Dict[str, float]] = None,
    context: Optional[List[str]] = None,
    **kwargs
) -> EvaluationResult:
    """
    Evaluate a multi-agent trace.

    Args:
        trace: Multi-agent trace to evaluate
        ground_truth: Expected correct outputs (optional)
        metric_set: Dictionary mapping dimension names to lists of metric names.
                   If None, uses all available metrics.
        dimension_weights: Weights for each dimension (uses defaults if None)
        thresholds: Threshold values for pass/fail (uses defaults if None)
        context: List of context documents (optional)
        **kwargs: Additional evaluation parameters

    Returns:
        EvaluationResult with scores and metadata

    Raises:
        ValueError: If inputs are invalid
        RuntimeError: If evaluation fails
    """
    logger.info(f"Evaluating multi-agent trace {trace.trace_id}")

    # Validate inputs
    if not isinstance(trace, MultiAgentTrace):
        raise ValueError("trace must be a MultiAgentTrace instance")

    # Use default weights and thresholds if not provided
    dimension_weights = dimension_weights or self.default_dimension_weights
    thresholds = thresholds or self.default_thresholds

    # Create evaluation input
    evaluation_input = EvaluationInput(
        trace=trace,
        ground_truth=ground_truth,
        context=context or (ground_truth.context_documents if ground_truth else [])
    )

    # Determine which metrics to calculate
    metrics_to_calculate = self._get_metrics_to_calculate(metric_set)

    logger.debug(f"Calculating {len(metrics_to_calculate)} metrics")

    # Calculate metrics with automatic dependency resolution
    metric_scores = self.calculate_metrics_with_dependencies(
        metrics=metrics_to_calculate,
        evaluation_input=evaluation_input
    )

    logger.debug(f"Calculated {len(metric_scores)} metric scores")

    # Aggregate scores by dimension
    dimension_results = self.aggregate_dimension_scores(
        metric_scores=metric_scores,
        dimension_weights=dimension_weights
    )

    logger.debug(f"Aggregated scores for {len(dimension_results)} dimensions")

    # Calculate overall score
    overall_score = self.calculate_overall_score(dimension_results)

    logger.debug(f"Overall score: {overall_score:.3f}")

    # Check thresholds
    passed, failures = self.check_thresholds(
        dimension_results=dimension_results,
        thresholds=thresholds
    )

    # Collect warnings
    warnings = []
    if ground_truth is None:
        gt_metrics = self.metric_registry.get_metrics_requiring_ground_truth()
        if any(m.metric_name in gt_metrics for m in metric_scores):
            warnings.append(
                "Ground truth not provided, some metrics may be skipped"
            )

    warnings.extend(self._metric_error_warnings(metric_scores))

    # Create evaluation result
    result = EvaluationResult(
        trace_id=trace.trace_id,
        dimension_results=dimension_results,
        overall_score=overall_score,
        passed=passed,
        failures=failures,
        warnings=warnings,
        metadata={
            "num_metrics": len(metric_scores),
            "num_dimensions": len(dimension_results),
            "num_agents": len(trace.agent_traces),
            "workflow_status": trace.workflow_status.value,
            "dimension_weights": dimension_weights,
            "thresholds": thresholds
        }
    )

    logger.info(
        f"Evaluation complete for trace {trace.trace_id}: "
        f"score={overall_score:.3f}, passed={passed}"
    )

    return result

evaluate_multi_turn(per_turn_traces, full_trace, ground_truth=None, metric_set=None, dimension_weights=None, thresholds=None, context=None, **kwargs)

Evaluate a multi-agent multi-turn conversation from pre-split per-turn traces.

Both per_turn_traces and full_trace must be MultiAgentTrace instances. Single-agent (AgentTrace) inputs are handled by SingleAgentEvaluator.evaluate_multi_turn().

Per-turn metrics (tool calling, response quality, multi-agent, ...) run on each MultiAgentTrace turn and are averaged. Full-trace metrics (context_retention, coherence, ...) run on the full MultiAgentTrace, whose messages property aggregates sub-agent messages in order.

Parameters:

Name	Type	Description	Default
per_turn_traces	List[MultiAgentTrace]	List of per-turn MultiAgentTraces (one per user turn)	required
full_trace	MultiAgentTrace	MultiAgentTrace covering the entire conversation	required
ground_truth	Optional[GroundTruth]	Expected correct outputs (optional). May contain per_turn_expected and per_turn_tools arrays in expected_arguments.	None
metric_set	Optional[Dict[str, List[str]]]	Dictionary mapping dimension names to lists of metric names	None
dimension_weights	Optional[Dict[str, float]]	Weights for each dimension (uses defaults if None)	None
thresholds	Optional[Dict[str, float]]	Threshold values for pass/fail (uses defaults if None)	None
context	Optional[List[str]]	List of context documents (optional)	None
**kwargs		Additional evaluation parameters	{}

Returns:

Type	Description
EvaluationResult	EvaluationResult with per-turn breakdown in metadata

Source code in uaef/evaluation/multi_agent.py

def evaluate_multi_turn(
    self,
    per_turn_traces: List[MultiAgentTrace],
    full_trace: MultiAgentTrace,
    ground_truth: Optional[GroundTruth] = None,
    metric_set: Optional[Dict[str, List[str]]] = None,
    dimension_weights: Optional[Dict[str, float]] = None,
    thresholds: Optional[Dict[str, float]] = None,
    context: Optional[List[str]] = None,
    **kwargs,
) -> EvaluationResult:
    """
    Evaluate a multi-agent multi-turn conversation from pre-split per-turn traces.

    Both per_turn_traces and full_trace must be MultiAgentTrace instances.
    Single-agent (AgentTrace) inputs are handled by
    SingleAgentEvaluator.evaluate_multi_turn().

    Per-turn metrics (tool calling, response quality, multi-agent, ...)
    run on each MultiAgentTrace turn and are averaged. Full-trace metrics
    (context_retention, coherence, ...) run on the full MultiAgentTrace,
    whose `messages` property aggregates sub-agent messages in order.

    Args:
        per_turn_traces: List of per-turn MultiAgentTraces (one per user turn)
        full_trace: MultiAgentTrace covering the entire conversation
        ground_truth: Expected correct outputs (optional). May contain
                     per_turn_expected and per_turn_tools arrays in expected_arguments.
        metric_set: Dictionary mapping dimension names to lists of metric names
        dimension_weights: Weights for each dimension (uses defaults if None)
        thresholds: Threshold values for pass/fail (uses defaults if None)
        context: List of context documents (optional)
        **kwargs: Additional evaluation parameters

    Returns:
        EvaluationResult with per-turn breakdown in metadata
    """
    if not per_turn_traces:
        raise ValueError("per_turn_traces must contain at least one trace")
    if not isinstance(full_trace, MultiAgentTrace):
        raise ValueError(
            "full_trace must be a MultiAgentTrace instance "
            "(use SingleAgentEvaluator.evaluate_multi_turn for AgentTrace)"
        )
    for i, t in enumerate(per_turn_traces):
        if not isinstance(t, MultiAgentTrace):
            raise ValueError(
                f"per_turn_traces[{i}] must be a MultiAgentTrace instance "
                "(use SingleAgentEvaluator.evaluate_multi_turn for AgentTrace)"
            )

    # Reuse the shared multi-turn loop on BaseEvaluator. Because
    # self._get_metrics_to_calculate / self.default_dimension_weights /
    # self.default_thresholds are all overridden on this class, the loop
    # picks up the multi-agent metric registry and weights automatically.
    return self._run_multi_turn(
        per_turn_traces=per_turn_traces,
        full_trace=full_trace,
        ground_truth=ground_truth,
        metric_set=metric_set,
        dimension_weights=dimension_weights,
        thresholds=thresholds,
        context=context,
        **kwargs,
    )

uaef.evaluation.online

Online evaluation mode for UAEF.

This module provides real-time evaluation functionality with: - Fast deterministic metrics (< 500ms) - Async LLM-based metrics (non-blocking) - Partial results for immediate decisions - Background storage of complete results

OnlineEvaluator

Bases: BaseEvaluator

Evaluator for online/real-time evaluation.

Optimized for low-latency evaluation with: - Synchronous calculation of fast deterministic metrics - Asynchronous execution of LLM-based metrics - Immediate return of partial results - Background completion and storage - Threshold-based pass/fail decisions

Source code in uaef/evaluation/online.py

class OnlineEvaluator(BaseEvaluator):
    """
    Evaluator for online/real-time evaluation.

    Optimized for low-latency evaluation with:
    - Synchronous calculation of fast deterministic metrics
    - Asynchronous execution of LLM-based metrics
    - Immediate return of partial results
    - Background completion and storage
    - Threshold-based pass/fail decisions
    """

    def __init__(
        self,
        metric_registry: Optional[any] = None,
        max_workers: int = 4,
        fast_metrics_only: bool = False,
        latency_threshold_ms: float = 500.0,
        default_thresholds: Optional[Dict[str, float]] = None
    ):
        """
        Initialize online evaluator.

        Args:
            metric_registry: MetricRegistry instance (uses global if None)
            max_workers: Maximum number of parallel workers
            fast_metrics_only: If True, only calculate deterministic metrics
            latency_threshold_ms: Maximum latency for synchronous metrics (ms)
            default_thresholds: Default threshold values for dimensions
        """
        super().__init__(metric_registry=metric_registry, max_workers=max_workers)

        self.fast_metrics_only = fast_metrics_only
        self.latency_threshold_ms = latency_threshold_ms

        # Set default thresholds for online evaluation
        self.default_thresholds = default_thresholds or {
            "Tool Calling": 0.7,
            "Response Quality": 0.6,
            "Responsible AI": 0.8,
            "Performance": 0.5
        }

        # Storage for async results (in production, use database)
        self._async_results: Dict[str, EvaluationResult] = {}

        logger.info(
            f"Initialized OnlineEvaluator with fast_metrics_only={fast_metrics_only}, "
            f"latency_threshold={latency_threshold_ms}ms"
        )

    def evaluate(
        self,
        trace: AgentTrace,
        ground_truth: Optional[GroundTruth] = None,
        metric_set: Optional[Dict[str, List[str]]] = None,
        thresholds: Optional[Dict[str, float]] = None,
        context: Optional[List[str]] = None,
        **kwargs
    ) -> EvaluationResult:
        """
        Evaluate a trace in online mode (blocking until complete).

        This method blocks until all metrics are calculated. For non-blocking
        evaluation, use evaluate_response() instead.

        Args:
            trace: Agent trace to evaluate
            ground_truth: Expected correct outputs (optional)
            metric_set: Dictionary mapping dimension names to lists of metric names
            thresholds: Threshold values for pass/fail (uses defaults if None)
            context: List of context documents (optional)
            **kwargs: Additional evaluation parameters

        Returns:
            Complete EvaluationResult
        """
        logger.info(f"Online evaluation (blocking) for trace {trace.trace_id}")

        # Use evaluate_response and wait for completion
        partial_result, complete_future = self.evaluate_response(
            trace=trace,
            ground_truth=ground_truth,
            metric_set=metric_set,
            thresholds=thresholds,
            context=context,
            **kwargs
        )

        # Wait for completion
        if complete_future:
            try:
                loop = asyncio.get_event_loop()
            except RuntimeError:
                loop = asyncio.new_event_loop()
                asyncio.set_event_loop(loop)

            complete_result = loop.run_until_complete(complete_future)
            return complete_result
        else:
            # No async metrics, convert partial to complete
            return self._partial_to_complete(partial_result, [])

    def evaluate_response(
        self,
        trace: AgentTrace,
        ground_truth: Optional[GroundTruth] = None,
        metric_set: Optional[Dict[str, List[str]]] = None,
        thresholds: Optional[Dict[str, float]] = None,
        context: Optional[List[str]] = None,
        store_async: bool = True,
        **kwargs
    ) -> Tuple[PartialEvaluationResult, Optional[asyncio.Task]]:
        """
        Evaluate a response in online mode (non-blocking).

        Returns partial results immediately with deterministic metrics,
        and starts async calculation of LLM-based metrics in background.

        Args:
            trace: Agent trace to evaluate
            ground_truth: Expected correct outputs (optional)
            metric_set: Dictionary mapping dimension names to lists of metric names.
                       If None, uses subset of metrics suitable for online evaluation.
            thresholds: Threshold values for pass/fail (uses defaults if None)
            context: List of context documents (optional)
            store_async: If True, store complete results in background
            **kwargs: Additional evaluation parameters

        Returns:
            Tuple of (PartialEvaluationResult, Optional[asyncio.Task])
            The task completes when all LLM metrics are calculated.
        """
        start_time = time.time()

        logger.info(f"Online evaluation (non-blocking) for trace {trace.trace_id}")

        # Validate inputs
        if not isinstance(trace, AgentTrace):
            raise ValueError("trace must be an AgentTrace instance")

        # Use default thresholds if not provided
        thresholds = thresholds or self.default_thresholds

        # Create evaluation input
        evaluation_input = EvaluationInput(
            trace=trace,
            ground_truth=ground_truth,
            context=context or []
        )

        # Determine which metrics to calculate
        metrics_to_calculate = self._get_online_metrics(metric_set)

        # Separate deterministic and LLM-based metrics
        deterministic_metrics = []
        llm_metrics = []

        for metric in metrics_to_calculate:
            if metric.requires_llm_judge() and not self.fast_metrics_only:
                llm_metrics.append(metric)
            else:
                deterministic_metrics.append(metric)

        logger.debug(
            f"Split metrics: {len(deterministic_metrics)} deterministic, "
            f"{len(llm_metrics)} LLM-based"
        )

        # Calculate deterministic metrics synchronously
        deterministic_scores = self._calculate_deterministic_parallel(
            deterministic_metrics,
            evaluation_input
        )

        deterministic_time = (time.time() - start_time) * 1000
        logger.debug(f"Deterministic metrics calculated in {deterministic_time:.1f}ms")

        # Check if we exceeded latency threshold
        warnings = []
        if deterministic_time > self.latency_threshold_ms:
            warnings.append(
                f"Deterministic metrics took {deterministic_time:.1f}ms, "
                f"exceeding threshold of {self.latency_threshold_ms}ms"
            )

        warnings.extend(self._metric_error_warnings(deterministic_scores))

        # Calculate partial scores and pass/fail
        partial_dimension_results = self.aggregate_dimension_scores(
            metric_scores=deterministic_scores,
            dimension_weights=None  # Use equal weights for partial results
        )

        partial_overall_score = self.calculate_overall_score(partial_dimension_results)

        passed, failures = self.check_thresholds(
            dimension_results=partial_dimension_results,
            thresholds=thresholds
        )

        # Create partial result
        partial_result = PartialEvaluationResult(
            trace_id=trace.trace_id,
            deterministic_scores=deterministic_scores,
            llm_scores_pending=len(llm_metrics),
            overall_score=partial_overall_score,
            passed=passed,
            failures=failures,
            warnings=warnings,
            metadata={
                "num_deterministic_metrics": len(deterministic_scores),
                "num_llm_metrics_pending": len(llm_metrics),
                "deterministic_latency_ms": deterministic_time,
                "fast_metrics_only": self.fast_metrics_only
            }
        )

        logger.info(
            f"Partial result ready in {deterministic_time:.1f}ms: "
            f"score={partial_overall_score:.3f}, passed={passed}"
        )

        # Start async calculation of LLM metrics if needed
        complete_task = None
        if llm_metrics and not self.fast_metrics_only:
            try:
                loop = asyncio.get_event_loop()
            except RuntimeError:
                loop = asyncio.new_event_loop()
                asyncio.set_event_loop(loop)

            complete_task = loop.create_task(
                self._complete_evaluation_async(
                    partial_result=partial_result,
                    llm_metrics=llm_metrics,
                    evaluation_input=evaluation_input,
                    thresholds=thresholds,
                    store_async=store_async
                )
            )

            logger.debug(f"Started async calculation of {len(llm_metrics)} LLM metrics")

        return partial_result, complete_task

    async def _complete_evaluation_async(
        self,
        partial_result: PartialEvaluationResult,
        llm_metrics: List[any],
        evaluation_input: EvaluationInput,
        thresholds: Dict[str, float],
        store_async: bool
    ) -> EvaluationResult:
        """
        Complete evaluation asynchronously by calculating LLM metrics.

        Args:
            partial_result: Partial result with deterministic metrics
            llm_metrics: List of LLM-based metrics to calculate
            evaluation_input: Input data for evaluation
            thresholds: Threshold values for pass/fail
            store_async: If True, store complete results

        Returns:
            Complete EvaluationResult
        """
        start_time = time.time()

        logger.debug(f"Starting async LLM metric calculation for {partial_result.trace_id}")

        # Calculate LLM metrics asynchronously
        llm_scores = await self._calculate_llm_async_impl(
            llm_metrics,
            evaluation_input
        )

        llm_time = (time.time() - start_time) * 1000
        logger.debug(f"LLM metrics calculated in {llm_time:.1f}ms")

        # Combine with deterministic scores
        all_scores = partial_result.deterministic_scores + llm_scores

        # Create complete result
        complete_result = self._partial_to_complete(
            partial_result,
            llm_scores,
            thresholds
        )

        # Store result if requested
        if store_async:
            self._store_result_async(complete_result)

        logger.info(
            f"Complete evaluation ready for {partial_result.trace_id}: "
            f"score={complete_result.overall_score:.3f}, "
            f"total_time={llm_time:.1f}ms"
        )

        return complete_result

    def _get_online_metrics(
        self,
        metric_set: Optional[Dict[str, List[str]]] = None
    ) -> List[any]:
        """
        Get list of metrics suitable for online evaluation.

        If metric_set is None, returns a subset of metrics that:
        - Don't require ground truth
        - Are fast to calculate
        - Are relevant for online decisions

        Args:
            metric_set: Dictionary mapping dimension names to lists of metric names.
                       If None, uses default online metric set.

        Returns:
            List of metric instances
        """
        if metric_set is not None:
            # Use specified metrics
            metrics = []
            for dimension_name, metric_names in metric_set.items():
                for metric_name in metric_names:
                    try:
                        metric = self.metric_registry.get_metric(metric_name)
                        metrics.append(metric)
                    except Exception as e:
                        logger.warning(
                            f"Failed to create metric {metric_name}: {str(e)}"
                        )
            return metrics

        # Use default online metrics (no ground truth required)
        online_metric_names = [
            # Performance metrics (fast, deterministic)
            "latency_score",
            "token_efficiency",
            "throughput",

            # Responsible AI metrics (LLM-based but important for online)
            "safety_score",
            "toxicity_score",
            "prompt_injection_detection",

            # Response quality metrics (LLM-based, no GT required)
            "answer_relevance",
        ]

        metrics = []
        for metric_name in online_metric_names:
            try:
                metric = self.metric_registry.get_metric(metric_name)
                # Skip if requires ground truth
                if not metric.requires_ground_truth():
                    metrics.append(metric)
            except KeyError:
                logger.debug(f"Metric {metric_name} not available in registry")
            except Exception as e:
                logger.warning(f"Failed to create metric {metric_name}: {str(e)}")

        return metrics

    def _partial_to_complete(
        self,
        partial_result: PartialEvaluationResult,
        llm_scores: List[MetricScore],
        thresholds: Optional[Dict[str, float]] = None
    ) -> EvaluationResult:
        """
        Convert partial result to complete EvaluationResult.

        Args:
            partial_result: Partial result with deterministic metrics
            llm_scores: LLM-based metric scores
            thresholds: Threshold values for pass/fail

        Returns:
            Complete EvaluationResult
        """
        # Combine all scores
        all_scores = partial_result.deterministic_scores + llm_scores

        # Aggregate by dimension
        dimension_results = self.aggregate_dimension_scores(
            metric_scores=all_scores,
            dimension_weights=None  # Use equal weights
        )

        # Calculate overall score
        overall_score = self.calculate_overall_score(dimension_results)

        # Check thresholds
        thresholds = thresholds or self.default_thresholds
        passed, failures = self.check_thresholds(
            dimension_results=dimension_results,
            thresholds=thresholds
        )

        # Create complete result
        metadata = partial_result.metadata.copy()
        metadata.update({
            "num_total_metrics": len(all_scores),
            "num_llm_metrics": len(llm_scores)
        })

        return EvaluationResult(
            trace_id=partial_result.trace_id,
            dimension_results=dimension_results,
            overall_score=overall_score,
            passed=passed,
            failures=failures,
            warnings=partial_result.warnings + self._metric_error_warnings(llm_scores),
            metadata=metadata
        )

    def _store_result_async(self, result: EvaluationResult) -> None:
        """
        Store complete result in background.

        In production, this would write to a database. For now, stores in memory.

        Args:
            result: Complete evaluation result to store
        """
        self._async_results[str(result.trace_id)] = result
        logger.debug(f"Stored complete result for {result.trace_id}")

    def get_stored_result(self, trace_id: str) -> Optional[EvaluationResult]:
        """
        Retrieve a stored result by trace ID.

        Args:
            trace_id: Trace ID to look up

        Returns:
            EvaluationResult if found, None otherwise
        """
        return self._async_results.get(trace_id)

    def aggregate_session_metrics(
        self,
        turn_results: List[EvaluationResult],
        session_id: Optional[str] = None
    ) -> EvaluationResult:
        """
        Aggregate metrics across all turns in a conversation session.

        This method combines per-turn evaluation results into session-level metrics,
        providing an overall assessment of the entire conversation including:
        - Average scores across all turns
        - Context retention across the conversation
        - Overall coherence and goal achievement
        - Aggregated performance metrics

        Args:
            turn_results: List of EvaluationResult objects, one per turn
            session_id: Optional session identifier

        Returns:
            EvaluationResult with aggregated session-level metrics

        Example:
            >>> evaluator = OnlineEvaluator()
            >>> turn_results = []
            >>> 
            >>> # Evaluate each turn
            >>> for turn_trace in conversation_turns:
            >>>     result = evaluator.evaluate(turn_trace)
            >>>     turn_results.append(result)
            >>> 
            >>> # Get session-level aggregation
            >>> session_result = evaluator.aggregate_session_metrics(turn_results)
            >>> print(f"Overall conversation score: {session_result.overall_score:.3f}")
        """
        if not turn_results:
            raise ValueError("turn_results cannot be empty")

        logger.info(f"Aggregating metrics for session with {len(turn_results)} turns")

        # Collect all dimension results across turns
        all_dimension_results: Dict[str, List[DimensionResult]] = {}

        for turn_result in turn_results:
            for dim_result in turn_result.dimension_results:
                all_dimension_results.setdefault(dim_result.dimension_name, []).append(dim_result)

        # Average each dimension over the turns where it was applicable
        aggregated_dimensions = []

        for dim_name, dim_results in all_dimension_results.items():
            applicable = [dr.aggregate_score for dr in dim_results if dr.aggregate_score is not None]
            avg_score = sum(applicable) / len(applicable) if applicable else None

            # Combine all metric scores from this dimension
            all_metric_scores = []
            for dr in dim_results:
                all_metric_scores.extend(dr.metric_scores)

            aggregated_dimensions.append(DimensionResult(
                dimension_name=dim_name,
                aggregate_score=avg_score,
                metric_scores=all_metric_scores,
                weight=dim_results[0].weight,
            ))

        overall_score = self.calculate_overall_score(aggregated_dimensions)

        # Aggregate failures and warnings, preserving order
        all_failures = list(dict.fromkeys(f for tr in turn_results for f in tr.failures))
        all_warnings = list(dict.fromkeys(w for tr in turn_results for w in tr.warnings))

        # Determine if session passed (all turns must pass)
        passed = all(tr.passed for tr in turn_results)

        # trace_id is a UUID distinct from every turn's; derive it stably from
        # the session id, or from the turns' trace ids when none is given
        session_key = session_id or ",".join(str(tr.trace_id) for tr in turn_results)
        trace_id = uuid5(NAMESPACE_URL, f"uaef-session/{session_key}")

        # Create session result
        session_result = EvaluationResult(
            trace_id=trace_id,
            dimension_results=aggregated_dimensions,
            overall_score=overall_score,
            passed=passed,
            failures=all_failures,
            warnings=all_warnings,
            metadata={
                "num_turns": len(turn_results),
                "session_id": session_id,
                "turn_scores": [tr.overall_score for tr in turn_results],
                "aggregation_type": "session_level"
            }
        )

        logger.info(
            f"Session aggregation complete: {len(turn_results)} turns, "
            f"overall_score={overall_score:.3f}, passed={passed}"
        )

        return session_result

aggregate_session_metrics(turn_results, session_id=None)

Aggregate metrics across all turns in a conversation session.

This method combines per-turn evaluation results into session-level metrics, providing an overall assessment of the entire conversation including: - Average scores across all turns - Context retention across the conversation - Overall coherence and goal achievement - Aggregated performance metrics

Parameters:

Name	Type	Description	Default
turn_results	List[EvaluationResult]	List of EvaluationResult objects, one per turn	required
session_id	Optional[str]	Optional session identifier	None

Returns:

Type	Description
EvaluationResult	EvaluationResult with aggregated session-level metrics

Example

evaluator = OnlineEvaluator() turn_results = []

Evaluate each turn

for turn_trace in conversation_turns: result = evaluator.evaluate(turn_trace) turn_results.append(result)

Get session-level aggregation

session_result = evaluator.aggregate_session_metrics(turn_results) print(f"Overall conversation score: {session_result.overall_score:.3f}")

Source code in uaef/evaluation/online.py

def aggregate_session_metrics(
    self,
    turn_results: List[EvaluationResult],
    session_id: Optional[str] = None
) -> EvaluationResult:
    """
    Aggregate metrics across all turns in a conversation session.

    This method combines per-turn evaluation results into session-level metrics,
    providing an overall assessment of the entire conversation including:
    - Average scores across all turns
    - Context retention across the conversation
    - Overall coherence and goal achievement
    - Aggregated performance metrics

    Args:
        turn_results: List of EvaluationResult objects, one per turn
        session_id: Optional session identifier

    Returns:
        EvaluationResult with aggregated session-level metrics

    Example:
        >>> evaluator = OnlineEvaluator()
        >>> turn_results = []
        >>> 
        >>> # Evaluate each turn
        >>> for turn_trace in conversation_turns:
        >>>     result = evaluator.evaluate(turn_trace)
        >>>     turn_results.append(result)
        >>> 
        >>> # Get session-level aggregation
        >>> session_result = evaluator.aggregate_session_metrics(turn_results)
        >>> print(f"Overall conversation score: {session_result.overall_score:.3f}")
    """
    if not turn_results:
        raise ValueError("turn_results cannot be empty")

    logger.info(f"Aggregating metrics for session with {len(turn_results)} turns")

    # Collect all dimension results across turns
    all_dimension_results: Dict[str, List[DimensionResult]] = {}

    for turn_result in turn_results:
        for dim_result in turn_result.dimension_results:
            all_dimension_results.setdefault(dim_result.dimension_name, []).append(dim_result)

    # Average each dimension over the turns where it was applicable
    aggregated_dimensions = []

    for dim_name, dim_results in all_dimension_results.items():
        applicable = [dr.aggregate_score for dr in dim_results if dr.aggregate_score is not None]
        avg_score = sum(applicable) / len(applicable) if applicable else None

        # Combine all metric scores from this dimension
        all_metric_scores = []
        for dr in dim_results:
            all_metric_scores.extend(dr.metric_scores)

        aggregated_dimensions.append(DimensionResult(
            dimension_name=dim_name,
            aggregate_score=avg_score,
            metric_scores=all_metric_scores,
            weight=dim_results[0].weight,
        ))

    overall_score = self.calculate_overall_score(aggregated_dimensions)

    # Aggregate failures and warnings, preserving order
    all_failures = list(dict.fromkeys(f for tr in turn_results for f in tr.failures))
    all_warnings = list(dict.fromkeys(w for tr in turn_results for w in tr.warnings))

    # Determine if session passed (all turns must pass)
    passed = all(tr.passed for tr in turn_results)

    # trace_id is a UUID distinct from every turn's; derive it stably from
    # the session id, or from the turns' trace ids when none is given
    session_key = session_id or ",".join(str(tr.trace_id) for tr in turn_results)
    trace_id = uuid5(NAMESPACE_URL, f"uaef-session/{session_key}")

    # Create session result
    session_result = EvaluationResult(
        trace_id=trace_id,
        dimension_results=aggregated_dimensions,
        overall_score=overall_score,
        passed=passed,
        failures=all_failures,
        warnings=all_warnings,
        metadata={
            "num_turns": len(turn_results),
            "session_id": session_id,
            "turn_scores": [tr.overall_score for tr in turn_results],
            "aggregation_type": "session_level"
        }
    )

    logger.info(
        f"Session aggregation complete: {len(turn_results)} turns, "
        f"overall_score={overall_score:.3f}, passed={passed}"
    )

    return session_result

evaluate(trace, ground_truth=None, metric_set=None, thresholds=None, context=None, **kwargs)

Evaluate a trace in online mode (blocking until complete).

This method blocks until all metrics are calculated. For non-blocking evaluation, use evaluate_response() instead.

Parameters:

Name	Type	Description	Default
trace	AgentTrace	Agent trace to evaluate	required
ground_truth	Optional[GroundTruth]	Expected correct outputs (optional)	None
metric_set	Optional[Dict[str, List[str]]]	Dictionary mapping dimension names to lists of metric names	None
thresholds	Optional[Dict[str, float]]	Threshold values for pass/fail (uses defaults if None)	None
context	Optional[List[str]]	List of context documents (optional)	None
**kwargs		Additional evaluation parameters	{}

Returns:

Type	Description
EvaluationResult	Complete EvaluationResult

Source code in uaef/evaluation/online.py

def evaluate(
    self,
    trace: AgentTrace,
    ground_truth: Optional[GroundTruth] = None,
    metric_set: Optional[Dict[str, List[str]]] = None,
    thresholds: Optional[Dict[str, float]] = None,
    context: Optional[List[str]] = None,
    **kwargs
) -> EvaluationResult:
    """
    Evaluate a trace in online mode (blocking until complete).

    This method blocks until all metrics are calculated. For non-blocking
    evaluation, use evaluate_response() instead.

    Args:
        trace: Agent trace to evaluate
        ground_truth: Expected correct outputs (optional)
        metric_set: Dictionary mapping dimension names to lists of metric names
        thresholds: Threshold values for pass/fail (uses defaults if None)
        context: List of context documents (optional)
        **kwargs: Additional evaluation parameters

    Returns:
        Complete EvaluationResult
    """
    logger.info(f"Online evaluation (blocking) for trace {trace.trace_id}")

    # Use evaluate_response and wait for completion
    partial_result, complete_future = self.evaluate_response(
        trace=trace,
        ground_truth=ground_truth,
        metric_set=metric_set,
        thresholds=thresholds,
        context=context,
        **kwargs
    )

    # Wait for completion
    if complete_future:
        try:
            loop = asyncio.get_event_loop()
        except RuntimeError:
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)

        complete_result = loop.run_until_complete(complete_future)
        return complete_result
    else:
        # No async metrics, convert partial to complete
        return self._partial_to_complete(partial_result, [])

evaluate_response(trace, ground_truth=None, metric_set=None, thresholds=None, context=None, store_async=True, **kwargs)

Evaluate a response in online mode (non-blocking).

Returns partial results immediately with deterministic metrics, and starts async calculation of LLM-based metrics in background.

Parameters:

Name	Type	Description	Default
trace	AgentTrace	Agent trace to evaluate	required
ground_truth	Optional[GroundTruth]	Expected correct outputs (optional)	None
metric_set	Optional[Dict[str, List[str]]]	Dictionary mapping dimension names to lists of metric names. If None, uses subset of metrics suitable for online evaluation.	None
thresholds	Optional[Dict[str, float]]	Threshold values for pass/fail (uses defaults if None)	None
context	Optional[List[str]]	List of context documents (optional)	None
store_async	bool	If True, store complete results in background	True
**kwargs		Additional evaluation parameters	{}

Returns:

Type	Description
PartialEvaluationResult	Tuple of (PartialEvaluationResult, Optional[asyncio.Task])
Optional[Task]	The task completes when all LLM metrics are calculated.

Source code in uaef/evaluation/online.py

def evaluate_response(
    self,
    trace: AgentTrace,
    ground_truth: Optional[GroundTruth] = None,
    metric_set: Optional[Dict[str, List[str]]] = None,
    thresholds: Optional[Dict[str, float]] = None,
    context: Optional[List[str]] = None,
    store_async: bool = True,
    **kwargs
) -> Tuple[PartialEvaluationResult, Optional[asyncio.Task]]:
    """
    Evaluate a response in online mode (non-blocking).

    Returns partial results immediately with deterministic metrics,
    and starts async calculation of LLM-based metrics in background.

    Args:
        trace: Agent trace to evaluate
        ground_truth: Expected correct outputs (optional)
        metric_set: Dictionary mapping dimension names to lists of metric names.
                   If None, uses subset of metrics suitable for online evaluation.
        thresholds: Threshold values for pass/fail (uses defaults if None)
        context: List of context documents (optional)
        store_async: If True, store complete results in background
        **kwargs: Additional evaluation parameters

    Returns:
        Tuple of (PartialEvaluationResult, Optional[asyncio.Task])
        The task completes when all LLM metrics are calculated.
    """
    start_time = time.time()

    logger.info(f"Online evaluation (non-blocking) for trace {trace.trace_id}")

    # Validate inputs
    if not isinstance(trace, AgentTrace):
        raise ValueError("trace must be an AgentTrace instance")

    # Use default thresholds if not provided
    thresholds = thresholds or self.default_thresholds

    # Create evaluation input
    evaluation_input = EvaluationInput(
        trace=trace,
        ground_truth=ground_truth,
        context=context or []
    )

    # Determine which metrics to calculate
    metrics_to_calculate = self._get_online_metrics(metric_set)

    # Separate deterministic and LLM-based metrics
    deterministic_metrics = []
    llm_metrics = []

    for metric in metrics_to_calculate:
        if metric.requires_llm_judge() and not self.fast_metrics_only:
            llm_metrics.append(metric)
        else:
            deterministic_metrics.append(metric)

    logger.debug(
        f"Split metrics: {len(deterministic_metrics)} deterministic, "
        f"{len(llm_metrics)} LLM-based"
    )

    # Calculate deterministic metrics synchronously
    deterministic_scores = self._calculate_deterministic_parallel(
        deterministic_metrics,
        evaluation_input
    )

    deterministic_time = (time.time() - start_time) * 1000
    logger.debug(f"Deterministic metrics calculated in {deterministic_time:.1f}ms")

    # Check if we exceeded latency threshold
    warnings = []
    if deterministic_time > self.latency_threshold_ms:
        warnings.append(
            f"Deterministic metrics took {deterministic_time:.1f}ms, "
            f"exceeding threshold of {self.latency_threshold_ms}ms"
        )

    warnings.extend(self._metric_error_warnings(deterministic_scores))

    # Calculate partial scores and pass/fail
    partial_dimension_results = self.aggregate_dimension_scores(
        metric_scores=deterministic_scores,
        dimension_weights=None  # Use equal weights for partial results
    )

    partial_overall_score = self.calculate_overall_score(partial_dimension_results)

    passed, failures = self.check_thresholds(
        dimension_results=partial_dimension_results,
        thresholds=thresholds
    )

    # Create partial result
    partial_result = PartialEvaluationResult(
        trace_id=trace.trace_id,
        deterministic_scores=deterministic_scores,
        llm_scores_pending=len(llm_metrics),
        overall_score=partial_overall_score,
        passed=passed,
        failures=failures,
        warnings=warnings,
        metadata={
            "num_deterministic_metrics": len(deterministic_scores),
            "num_llm_metrics_pending": len(llm_metrics),
            "deterministic_latency_ms": deterministic_time,
            "fast_metrics_only": self.fast_metrics_only
        }
    )

    logger.info(
        f"Partial result ready in {deterministic_time:.1f}ms: "
        f"score={partial_overall_score:.3f}, passed={passed}"
    )

    # Start async calculation of LLM metrics if needed
    complete_task = None
    if llm_metrics and not self.fast_metrics_only:
        try:
            loop = asyncio.get_event_loop()
        except RuntimeError:
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)

        complete_task = loop.create_task(
            self._complete_evaluation_async(
                partial_result=partial_result,
                llm_metrics=llm_metrics,
                evaluation_input=evaluation_input,
                thresholds=thresholds,
                store_async=store_async
            )
        )

        logger.debug(f"Started async calculation of {len(llm_metrics)} LLM metrics")

    return partial_result, complete_task

get_stored_result(trace_id)

Retrieve a stored result by trace ID.

Parameters:

Name	Type	Description	Default
trace_id	str	Trace ID to look up	required

Returns:

Type	Description
Optional[EvaluationResult]	EvaluationResult if found, None otherwise

Source code in uaef/evaluation/online.py

def get_stored_result(self, trace_id: str) -> Optional[EvaluationResult]:
    """
    Retrieve a stored result by trace ID.

    Args:
        trace_id: Trace ID to look up

    Returns:
        EvaluationResult if found, None otherwise
    """
    return self._async_results.get(trace_id)

PartialEvaluationResult

Partial evaluation result for online mode.

Contains immediately available deterministic metrics and placeholders for async LLM-based metrics that are still being calculated.

Source code in uaef/evaluation/online.py

class PartialEvaluationResult:
    """
    Partial evaluation result for online mode.

    Contains immediately available deterministic metrics and placeholders
    for async LLM-based metrics that are still being calculated.
    """

    def __init__(
        self,
        trace_id: any,
        deterministic_scores: List[MetricScore],
        llm_scores_pending: int,
        overall_score: float,
        passed: bool,
        failures: List[str],
        warnings: List[str],
        metadata: Dict[str, any]
    ):
        """Initialize partial evaluation result."""
        self.trace_id = trace_id
        self.deterministic_scores = deterministic_scores
        self.llm_scores_pending = llm_scores_pending
        self.overall_score = overall_score
        self.passed = passed
        self.failures = failures
        self.warnings = warnings
        self.metadata = metadata
        self.is_complete = llm_scores_pending == 0

    def to_dict(self) -> Dict[str, any]:
        """Convert to dictionary representation."""
        return {
            "trace_id": str(self.trace_id),
            "deterministic_scores": [
                {
                    "metric_name": s.metric_name,
                    "score": s.score,
                    "reasoning": s.reasoning
                }
                for s in self.deterministic_scores
            ],
            "llm_scores_pending": self.llm_scores_pending,
            "overall_score": self.overall_score,
            "passed": self.passed,
            "failures": self.failures,
            "warnings": self.warnings,
            "is_complete": self.is_complete,
            "metadata": self.metadata
        }

to_dict()

Convert to dictionary representation.

Source code in uaef/evaluation/online.py

def to_dict(self) -> Dict[str, any]:
    """Convert to dictionary representation."""
    return {
        "trace_id": str(self.trace_id),
        "deterministic_scores": [
            {
                "metric_name": s.metric_name,
                "score": s.score,
                "reasoning": s.reasoning
            }
            for s in self.deterministic_scores
        ],
        "llm_scores_pending": self.llm_scores_pending,
        "overall_score": self.overall_score,
        "passed": self.passed,
        "failures": self.failures,
        "warnings": self.warnings,
        "is_complete": self.is_complete,
        "metadata": self.metadata
    }

uaef.evaluation.offline

Offline batch evaluation mode for UAEF.

This module provides batch evaluation functionality with: - Parallel processing of multiple traces - Batch LLM API calls for efficiency - Aggregate statistics across batch - Configurable parallelism - Graceful error handling - Export to various formats

BatchStatistics

Aggregate statistics for a batch evaluation.

Source code in uaef/evaluation/offline.py

class BatchStatistics:
    """
    Aggregate statistics for a batch evaluation.
    """

    def __init__(self, results: List[EvaluationResult]):
        """
        Initialize batch statistics from evaluation results.

        Args:
            results: List of evaluation results
        """
        self.total_count = len(results)
        self.passed_count = sum(1 for r in results if r.passed)
        self.failed_count = self.total_count - self.passed_count
        self.pass_rate = self.passed_count / self.total_count if self.total_count > 0 else 0.0

        # Calculate score statistics
        scores = [r.overall_score for r in results]
        self.mean_score = sum(scores) / len(scores) if scores else 0.0
        self.min_score = min(scores) if scores else 0.0
        self.max_score = max(scores) if scores else 0.0

        # Calculate dimension statistics
        self.dimension_stats = self._calculate_dimension_stats(results)

        # Calculate metric statistics
        self.metric_stats = self._calculate_metric_stats(results)

    def _calculate_dimension_stats(
        self,
        results: List[EvaluationResult]
    ) -> Dict[str, Dict[str, float]]:
        """Calculate statistics per dimension."""
        dimension_scores: Dict[str, List[float]] = {}

        for result in results:
            for dim_result in result.dimension_results:
                dim_name = dim_result.dimension_name
                if dim_result.aggregate_score is None:
                    continue
                if dim_name not in dimension_scores:
                    dimension_scores[dim_name] = []
                dimension_scores[dim_name].append(dim_result.aggregate_score)

        stats = {}
        for dim_name, scores in dimension_scores.items():
            stats[dim_name] = {
                "mean": sum(scores) / len(scores) if scores else 0.0,
                "min": min(scores) if scores else 0.0,
                "max": max(scores) if scores else 0.0,
                "count": len(scores)
            }

        return stats

    def _calculate_metric_stats(
        self,
        results: List[EvaluationResult]
    ) -> Dict[str, Dict[str, float]]:
        """Calculate statistics per metric."""
        metric_scores: Dict[str, List[float]] = {}

        for result in results:
            for dim_result in result.dimension_results:
                for metric_score in dim_result.metric_scores:
                    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)

        stats = {}
        for metric_name, scores in metric_scores.items():
            stats[metric_name] = {
                "mean": sum(scores) / len(scores) if scores else 0.0,
                "min": min(scores) if scores else 0.0,
                "max": max(scores) if scores else 0.0,
                "count": len(scores)
            }

        return stats

    def to_dict(self) -> Dict[str, Any]:
        """Convert to dictionary representation."""
        return {
            "total_count": self.total_count,
            "passed_count": self.passed_count,
            "failed_count": self.failed_count,
            "pass_rate": self.pass_rate,
            "mean_score": self.mean_score,
            "min_score": self.min_score,
            "max_score": self.max_score,
            "dimension_stats": self.dimension_stats,
            "metric_stats": self.metric_stats
        }

to_dict()

Convert to dictionary representation.

Source code in uaef/evaluation/offline.py

def to_dict(self) -> Dict[str, Any]:
    """Convert to dictionary representation."""
    return {
        "total_count": self.total_count,
        "passed_count": self.passed_count,
        "failed_count": self.failed_count,
        "pass_rate": self.pass_rate,
        "mean_score": self.mean_score,
        "min_score": self.min_score,
        "max_score": self.max_score,
        "dimension_stats": self.dimension_stats,
        "metric_stats": self.metric_stats
    }

OfflineEvaluator

Bases: BaseEvaluator

Evaluator for offline/batch evaluation.

Optimized for throughput and efficiency with: - Parallel processing of multiple traces - Batch LLM API calls - Aggregate statistics calculation - Configurable parallelism - Graceful error handling - Export to various formats

Source code in uaef/evaluation/offline.py

class OfflineEvaluator(BaseEvaluator):
    """
    Evaluator for offline/batch evaluation.

    Optimized for throughput and efficiency with:
    - Parallel processing of multiple traces
    - Batch LLM API calls
    - Aggregate statistics calculation
    - Configurable parallelism
    - Graceful error handling
    - Export to various formats
    """

    def __init__(
        self,
        metric_registry: Optional[any] = None,
        max_workers: int = 8,
        batch_size: int = 10,
        default_dimension_weights: Optional[Dict[str, float]] = None,
        default_thresholds: Optional[Dict[str, float]] = None
    ):
        """
        Initialize offline evaluator.

        Args:
            metric_registry: MetricRegistry instance (uses global if None)
            max_workers: Maximum number of parallel workers for trace processing
            batch_size: Number of traces to process in each batch
            default_dimension_weights: Default weights for dimensions
            default_thresholds: Default threshold values for dimensions
        """
        super().__init__(metric_registry=metric_registry, max_workers=max_workers)

        self.batch_size = batch_size

        # Set default dimension weights
        self.default_dimension_weights = default_dimension_weights or {
            "Tool Calling": 0.25,
            "Response Quality": 0.25,
            "Responsible AI": 0.20,
            "Performance": 0.10,
            "Multi-Turn": 0.10,
            "Multi-Agent": 0.05,
            "Reasoning": 0.05
        }

        # Set default thresholds
        self.default_thresholds = default_thresholds or {
            "Tool Calling": 0.7,
            "Response Quality": 0.7,
            "Responsible AI": 0.8,
            "Performance": 0.6,
            "Multi-Turn": 0.6,
            "Multi-Agent": 0.6,
            "Reasoning": 0.6
        }

        logger.info(
            f"Initialized OfflineEvaluator with max_workers={max_workers}, "
            f"batch_size={batch_size}"
        )

    def evaluate(
        self,
        trace: AgentTrace,
        ground_truth: Optional[GroundTruth] = None,
        metric_set: Optional[Dict[str, List[str]]] = None,
        dimension_weights: Optional[Dict[str, float]] = None,
        thresholds: Optional[Dict[str, float]] = None,
        context: Optional[List[str]] = None,
        **kwargs
    ) -> EvaluationResult:
        """
        Evaluate a single trace (delegates to batch_evaluate).

        Args:
            trace: Agent trace to evaluate
            ground_truth: Expected correct outputs (optional)
            metric_set: Dictionary mapping dimension names to lists of metric names
            dimension_weights: Weights for each dimension (uses defaults if None)
            thresholds: Threshold values for pass/fail (uses defaults if None)
            context: List of context documents (optional)
            **kwargs: Additional evaluation parameters

        Returns:
            EvaluationResult with scores and metadata
        """
        results = self.batch_evaluate(
            traces=[trace],
            ground_truths=[ground_truth] if ground_truth else None,
            metric_set=metric_set,
            dimension_weights=dimension_weights,
            thresholds=thresholds,
            contexts=[context] if context else None,
            **kwargs
        )

        return results[0]

    def evaluate_dataset(
        self,
        dataset: List[Tuple[AgentTrace, Optional[GroundTruth]]],
        metric_set: Optional[Dict[str, List[str]]] = None,
        dimension_weights: Optional[Dict[str, float]] = None,
        thresholds: Optional[Dict[str, float]] = None,
        contexts: Optional[List[Optional[List[str]]]] = None,
        **kwargs
    ) -> Tuple[List[EvaluationResult], BatchStatistics]:
        """
        Evaluate a complete dataset of traces.

        Args:
            dataset: List of (trace, ground_truth) tuples
            metric_set: Dictionary mapping dimension names to lists of metric names
            dimension_weights: Weights for each dimension (uses defaults if None)
            thresholds: Threshold values for pass/fail (uses defaults if None)
            contexts: List of context documents for each trace (optional)
            **kwargs: Additional evaluation parameters

        Returns:
            Tuple of (results, statistics)
        """
        logger.info(f"Evaluating dataset with {len(dataset)} traces")

        start_time = time.time()

        # Separate traces and ground truths
        traces = [item[0] for item in dataset]
        ground_truths = [item[1] for item in dataset]

        # Evaluate batch
        results = self.batch_evaluate(
            traces=traces,
            ground_truths=ground_truths,
            metric_set=metric_set,
            dimension_weights=dimension_weights,
            thresholds=thresholds,
            contexts=contexts,
            **kwargs
        )

        # Calculate statistics
        statistics = BatchStatistics(results)

        elapsed_time = time.time() - start_time

        logger.info(
            f"Dataset evaluation complete: {len(results)} traces in {elapsed_time:.1f}s "
            f"({len(results)/elapsed_time:.1f} traces/sec), "
            f"pass_rate={statistics.pass_rate:.2%}"
        )

        return results, statistics

    def batch_evaluate(
        self,
        traces: List[AgentTrace],
        ground_truths: Optional[List[Optional[GroundTruth]]] = None,
        metric_set: Optional[Dict[str, List[str]]] = None,
        dimension_weights: Optional[Dict[str, float]] = None,
        thresholds: Optional[Dict[str, float]] = None,
        contexts: Optional[List[Optional[List[str]]]] = None,
        **kwargs
    ) -> List[EvaluationResult]:
        """
        Evaluate multiple traces in parallel with batch optimization.

        Args:
            traces: List of agent traces to evaluate
            ground_truths: List of ground truths (optional)
            metric_set: Dictionary mapping dimension names to lists of metric names
            dimension_weights: Weights for each dimension (uses defaults if None)
            thresholds: Threshold values for pass/fail (uses defaults if None)
            contexts: List of context documents for each trace (optional)
            **kwargs: Additional evaluation parameters

        Returns:
            List of EvaluationResult objects
        """
        logger.info(f"Starting batch evaluation for {len(traces)} traces")

        start_time = time.time()

        # Validate inputs
        if not traces:
            raise ValueError("traces cannot be empty")

        if ground_truths is None:
            ground_truths = [None] * len(traces)

        if contexts is None:
            contexts = [None] * len(traces)

        if len(traces) != len(ground_truths):
            raise ValueError(
                f"Length mismatch: {len(traces)} traces but {len(ground_truths)} ground truths"
            )

        if len(traces) != len(contexts):
            raise ValueError(
                f"Length mismatch: {len(traces)} traces but {len(contexts)} contexts"
            )

        # Use default weights and thresholds if not provided
        dimension_weights = dimension_weights or self.default_dimension_weights
        thresholds = thresholds or self.default_thresholds

        # Process traces in parallel
        results = []

        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            # Submit all trace evaluations
            future_to_index = {}

            for i, (trace, gt, ctx) in enumerate(zip(traces, ground_truths, contexts)):
                future = executor.submit(
                    self._evaluate_single_trace,
                    trace=trace,
                    ground_truth=gt,
                    metric_set=metric_set,
                    dimension_weights=dimension_weights,
                    thresholds=thresholds,
                    context=ctx,
                    **kwargs
                )
                future_to_index[future] = i

            # Collect results in order
            results_dict = {}

            for future in as_completed(future_to_index):
                index = future_to_index[future]
                try:
                    result = future.result()
                    results_dict[index] = result

                    if (index + 1) % 10 == 0:
                        logger.debug(f"Completed {index + 1}/{len(traces)} evaluations")

                except Exception as e:
                    logger.error(f"Failed to evaluate trace {index}: {str(e)}")
                    # Create failed result
                    results_dict[index] = self._create_failed_result(
                        traces[index],
                        str(e)
                    )

            # Sort results by index
            results = [results_dict[i] for i in range(len(traces))]

        elapsed_time = time.time() - start_time
        success_count = sum(1 for r in results if r.passed)

        logger.info(
            f"Batch evaluation complete: {success_count}/{len(results)} passed "
            f"in {elapsed_time:.1f}s ({len(results)/elapsed_time:.1f} traces/sec)"
        )

        return results

    def _evaluate_single_trace(
        self,
        trace: AgentTrace,
        ground_truth: Optional[GroundTruth],
        metric_set: Optional[Dict[str, List[str]]],
        dimension_weights: Dict[str, float],
        thresholds: Dict[str, float],
        context: Optional[List[str]],
        **kwargs
    ) -> EvaluationResult:
        """
        Evaluate a single trace (used by parallel workers).

        This is a wrapper around the evaluation logic that can be called
        by ThreadPoolExecutor workers.
        """
        from uaef.evaluation.multi_agent import MultiAgentEvaluator
        from uaef.evaluation.single_agent import SingleAgentEvaluator
        from uaef.models.multi_agent_trace import MultiAgentTrace

        if isinstance(trace, MultiAgentTrace):
            evaluator = MultiAgentEvaluator(
                metric_registry=self.metric_registry,
                max_workers=4,
                default_dimension_weights=dimension_weights,
                default_thresholds=thresholds
            )
        else:
            evaluator = SingleAgentEvaluator(
                metric_registry=self.metric_registry,
                max_workers=4,
                default_dimension_weights=dimension_weights,
                default_thresholds=thresholds
            )

        return evaluator.evaluate(
            trace=trace,
            ground_truth=ground_truth,
            metric_set=metric_set,
            dimension_weights=dimension_weights,
            thresholds=thresholds,
            context=context,
            **kwargs
        )

    def export_results(
        self,
        results: List[EvaluationResult],
        statistics: Optional[BatchStatistics],
        output_path: str,
        format: str = "json"
    ) -> None:
        """
        Export evaluation results to file.

        Args:
            results: List of evaluation results
            statistics: Batch statistics (optional)
            output_path: Path to output file
            format: Export format ("json", "csv", "html")

        Raises:
            ValueError: If format is not supported
        """
        logger.info(f"Exporting {len(results)} results to {output_path} ({format})")

        output_file = Path(output_path)
        output_file.parent.mkdir(parents=True, exist_ok=True)

        if format == "json":
            self._export_json(results, statistics, output_file)
        elif format == "csv":
            self._export_csv(results, statistics, output_file)
        elif format == "html":
            self._export_html(results, statistics, output_file)
        else:
            raise ValueError(f"Unsupported export format: {format}")

        logger.info(f"Results exported to {output_path}")

    def _export_json(
        self,
        results: List[EvaluationResult],
        statistics: Optional[BatchStatistics],
        output_file: Path
    ) -> None:
        """Export results to JSON format."""
        data = {
            "results": [
                {
                    "trace_id": str(r.trace_id),
                    "overall_score": r.overall_score,
                    "passed": r.passed,
                    "failures": r.failures,
                    "warnings": r.warnings,
                    "dimensions": [
                        {
                            "name": d.dimension_name,
                            "score": d.aggregate_score,
                            "weight": d.weight,
                            "metrics": [
                                {
                                    "name": m.metric_name,
                                    "score": m.score,
                                    "reasoning": m.reasoning
                                }
                                for m in d.metric_scores
                            ]
                        }
                        for d in r.dimension_results
                    ],
                    "metadata": r.metadata,
                    "timestamp": r.timestamp.isoformat()
                }
                for r in results
            ]
        }

        if statistics:
            data["statistics"] = statistics.to_dict()

        with open(output_file, "w") as f:
            json.dump(data, f, indent=2)

    def _export_csv(
        self,
        results: List[EvaluationResult],
        statistics: Optional[BatchStatistics],
        output_file: Path
    ) -> None:
        """Export results to CSV format."""
        import csv

        with open(output_file, "w", newline="") as f:
            writer = csv.writer(f)

            # Write header
            header = ["trace_id", "overall_score", "passed"]

            # Add dimension columns
            if results:
                for dim in results[0].dimension_results:
                    header.append(f"{dim.dimension_name}_score")

            writer.writerow(header)

            # Write data rows
            for result in results:
                row = [
                    str(result.trace_id),
                    f"{result.overall_score:.3f}",
                    str(result.passed)
                ]

                for dim in result.dimension_results:
                    row.append(
                        "N/A" if dim.aggregate_score is None
                        else f"{dim.aggregate_score:.3f}"
                    )

                writer.writerow(row)

            # Write statistics if available
            if statistics:
                writer.writerow([])
                writer.writerow(["Statistics"])
                writer.writerow(["Total Count", statistics.total_count])
                writer.writerow(["Passed Count", statistics.passed_count])
                writer.writerow(["Failed Count", statistics.failed_count])
                writer.writerow(["Pass Rate", f"{statistics.pass_rate:.2%}"])
                writer.writerow(["Mean Score", f"{statistics.mean_score:.3f}"])
                writer.writerow(["Min Score", f"{statistics.min_score:.3f}"])
                writer.writerow(["Max Score", f"{statistics.max_score:.3f}"])

    def _export_html(
        self,
        results: List[EvaluationResult],
        statistics: Optional[BatchStatistics],
        output_file: Path
    ) -> None:
        """Export results to HTML format."""
        html = """<!DOCTYPE html>
<html>
<head>
    <title>UAEF Evaluation Results</title>
    <style>
        body { font-family: Arial, sans-serif; margin: 20px; }
        h1 { color: #333; }
        .stats { background: #f5f5f5; padding: 15px; margin: 20px 0; border-radius: 5px; }
        table { border-collapse: collapse; width: 100%; margin: 20px 0; }
        th, td { border: 1px solid #ddd; padding: 8px; text-align: left; }
        th { background-color: #4CAF50; color: white; }
        tr:nth-child(even) { background-color: #f2f2f2; }
        .passed { color: green; font-weight: bold; }
        .failed { color: red; font-weight: bold; }
    </style>
</head>
<body>
    <h1>UAEF Evaluation Results</h1>
"""

        # Add statistics
        if statistics:
            html += f"""
    <div class="stats">
        <h2>Statistics</h2>
        <p><strong>Total Count:</strong> {statistics.total_count}</p>
        <p><strong>Passed:</strong> {statistics.passed_count} ({statistics.pass_rate:.1%})</p>
        <p><strong>Failed:</strong> {statistics.failed_count}</p>
        <p><strong>Mean Score:</strong> {statistics.mean_score:.3f}</p>
        <p><strong>Score Range:</strong> {statistics.min_score:.3f} - {statistics.max_score:.3f}</p>
    </div>
"""

        # Add results table
        html += """
    <h2>Results</h2>
    <table>
        <tr>
            <th>Trace ID</th>
            <th>Overall Score</th>
            <th>Status</th>
            <th>Dimensions</th>
        </tr>
"""

        for result in results:
            status_class = "passed" if result.passed else "failed"
            status_text = "PASSED" if result.passed else "FAILED"

            dimensions_html = "<br>".join([
                f"{d.dimension_name}: "
                f"{'N/A' if d.aggregate_score is None else f'{d.aggregate_score:.3f}'}"
                for d in result.dimension_results
            ])

            html += f"""
        <tr>
            <td>{result.trace_id}</td>
            <td>{result.overall_score:.3f}</td>
            <td class="{status_class}">{status_text}</td>
            <td>{dimensions_html}</td>
        </tr>
"""

        html += """
    </table>
</body>
</html>
"""

        with open(output_file, "w") as f:
            f.write(html)

batch_evaluate(traces, ground_truths=None, metric_set=None, dimension_weights=None, thresholds=None, contexts=None, **kwargs)

Evaluate multiple traces in parallel with batch optimization.

Parameters:

Name	Type	Description	Default
traces	List[AgentTrace]	List of agent traces to evaluate	required
ground_truths	Optional[List[Optional[GroundTruth]]]	List of ground truths (optional)	None
metric_set	Optional[Dict[str, List[str]]]	Dictionary mapping dimension names to lists of metric names	None
dimension_weights	Optional[Dict[str, float]]	Weights for each dimension (uses defaults if None)	None
thresholds	Optional[Dict[str, float]]	Threshold values for pass/fail (uses defaults if None)	None
contexts	Optional[List[Optional[List[str]]]]	List of context documents for each trace (optional)	None
**kwargs		Additional evaluation parameters	{}

Returns:

Type	Description
List[EvaluationResult]	List of EvaluationResult objects

Source code in uaef/evaluation/offline.py

def batch_evaluate(
    self,
    traces: List[AgentTrace],
    ground_truths: Optional[List[Optional[GroundTruth]]] = None,
    metric_set: Optional[Dict[str, List[str]]] = None,
    dimension_weights: Optional[Dict[str, float]] = None,
    thresholds: Optional[Dict[str, float]] = None,
    contexts: Optional[List[Optional[List[str]]]] = None,
    **kwargs
) -> List[EvaluationResult]:
    """
    Evaluate multiple traces in parallel with batch optimization.

    Args:
        traces: List of agent traces to evaluate
        ground_truths: List of ground truths (optional)
        metric_set: Dictionary mapping dimension names to lists of metric names
        dimension_weights: Weights for each dimension (uses defaults if None)
        thresholds: Threshold values for pass/fail (uses defaults if None)
        contexts: List of context documents for each trace (optional)
        **kwargs: Additional evaluation parameters

    Returns:
        List of EvaluationResult objects
    """
    logger.info(f"Starting batch evaluation for {len(traces)} traces")

    start_time = time.time()

    # Validate inputs
    if not traces:
        raise ValueError("traces cannot be empty")

    if ground_truths is None:
        ground_truths = [None] * len(traces)

    if contexts is None:
        contexts = [None] * len(traces)

    if len(traces) != len(ground_truths):
        raise ValueError(
            f"Length mismatch: {len(traces)} traces but {len(ground_truths)} ground truths"
        )

    if len(traces) != len(contexts):
        raise ValueError(
            f"Length mismatch: {len(traces)} traces but {len(contexts)} contexts"
        )

    # Use default weights and thresholds if not provided
    dimension_weights = dimension_weights or self.default_dimension_weights
    thresholds = thresholds or self.default_thresholds

    # Process traces in parallel
    results = []

    with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
        # Submit all trace evaluations
        future_to_index = {}

        for i, (trace, gt, ctx) in enumerate(zip(traces, ground_truths, contexts)):
            future = executor.submit(
                self._evaluate_single_trace,
                trace=trace,
                ground_truth=gt,
                metric_set=metric_set,
                dimension_weights=dimension_weights,
                thresholds=thresholds,
                context=ctx,
                **kwargs
            )
            future_to_index[future] = i

        # Collect results in order
        results_dict = {}

        for future in as_completed(future_to_index):
            index = future_to_index[future]
            try:
                result = future.result()
                results_dict[index] = result

                if (index + 1) % 10 == 0:
                    logger.debug(f"Completed {index + 1}/{len(traces)} evaluations")

            except Exception as e:
                logger.error(f"Failed to evaluate trace {index}: {str(e)}")
                # Create failed result
                results_dict[index] = self._create_failed_result(
                    traces[index],
                    str(e)
                )

        # Sort results by index
        results = [results_dict[i] for i in range(len(traces))]

    elapsed_time = time.time() - start_time
    success_count = sum(1 for r in results if r.passed)

    logger.info(
        f"Batch evaluation complete: {success_count}/{len(results)} passed "
        f"in {elapsed_time:.1f}s ({len(results)/elapsed_time:.1f} traces/sec)"
    )

    return results

evaluate(trace, ground_truth=None, metric_set=None, dimension_weights=None, thresholds=None, context=None, **kwargs)

Evaluate a single trace (delegates to batch_evaluate).

Parameters:

Name	Type	Description	Default
trace	AgentTrace	Agent trace to evaluate	required
ground_truth	Optional[GroundTruth]	Expected correct outputs (optional)	None
metric_set	Optional[Dict[str, List[str]]]	Dictionary mapping dimension names to lists of metric names	None
dimension_weights	Optional[Dict[str, float]]	Weights for each dimension (uses defaults if None)	None
thresholds	Optional[Dict[str, float]]	Threshold values for pass/fail (uses defaults if None)	None
context	Optional[List[str]]	List of context documents (optional)	None
**kwargs		Additional evaluation parameters	{}

Returns:

Type	Description
EvaluationResult	EvaluationResult with scores and metadata

Source code in uaef/evaluation/offline.py

def evaluate(
    self,
    trace: AgentTrace,
    ground_truth: Optional[GroundTruth] = None,
    metric_set: Optional[Dict[str, List[str]]] = None,
    dimension_weights: Optional[Dict[str, float]] = None,
    thresholds: Optional[Dict[str, float]] = None,
    context: Optional[List[str]] = None,
    **kwargs
) -> EvaluationResult:
    """
    Evaluate a single trace (delegates to batch_evaluate).

    Args:
        trace: Agent trace to evaluate
        ground_truth: Expected correct outputs (optional)
        metric_set: Dictionary mapping dimension names to lists of metric names
        dimension_weights: Weights for each dimension (uses defaults if None)
        thresholds: Threshold values for pass/fail (uses defaults if None)
        context: List of context documents (optional)
        **kwargs: Additional evaluation parameters

    Returns:
        EvaluationResult with scores and metadata
    """
    results = self.batch_evaluate(
        traces=[trace],
        ground_truths=[ground_truth] if ground_truth else None,
        metric_set=metric_set,
        dimension_weights=dimension_weights,
        thresholds=thresholds,
        contexts=[context] if context else None,
        **kwargs
    )

    return results[0]

evaluate_dataset(dataset, metric_set=None, dimension_weights=None, thresholds=None, contexts=None, **kwargs)

Evaluate a complete dataset of traces.

Parameters:

Name	Type	Description	Default
dataset	List[Tuple[AgentTrace, Optional[GroundTruth]]]	List of (trace, ground_truth) tuples	required
metric_set	Optional[Dict[str, List[str]]]	Dictionary mapping dimension names to lists of metric names	None
dimension_weights	Optional[Dict[str, float]]	Weights for each dimension (uses defaults if None)	None
thresholds	Optional[Dict[str, float]]	Threshold values for pass/fail (uses defaults if None)	None
contexts	Optional[List[Optional[List[str]]]]	List of context documents for each trace (optional)	None
**kwargs		Additional evaluation parameters	{}

Returns:

Type	Description
Tuple[List[EvaluationResult], BatchStatistics]	Tuple of (results, statistics)

Source code in uaef/evaluation/offline.py

def evaluate_dataset(
    self,
    dataset: List[Tuple[AgentTrace, Optional[GroundTruth]]],
    metric_set: Optional[Dict[str, List[str]]] = None,
    dimension_weights: Optional[Dict[str, float]] = None,
    thresholds: Optional[Dict[str, float]] = None,
    contexts: Optional[List[Optional[List[str]]]] = None,
    **kwargs
) -> Tuple[List[EvaluationResult], BatchStatistics]:
    """
    Evaluate a complete dataset of traces.

    Args:
        dataset: List of (trace, ground_truth) tuples
        metric_set: Dictionary mapping dimension names to lists of metric names
        dimension_weights: Weights for each dimension (uses defaults if None)
        thresholds: Threshold values for pass/fail (uses defaults if None)
        contexts: List of context documents for each trace (optional)
        **kwargs: Additional evaluation parameters

    Returns:
        Tuple of (results, statistics)
    """
    logger.info(f"Evaluating dataset with {len(dataset)} traces")

    start_time = time.time()

    # Separate traces and ground truths
    traces = [item[0] for item in dataset]
    ground_truths = [item[1] for item in dataset]

    # Evaluate batch
    results = self.batch_evaluate(
        traces=traces,
        ground_truths=ground_truths,
        metric_set=metric_set,
        dimension_weights=dimension_weights,
        thresholds=thresholds,
        contexts=contexts,
        **kwargs
    )

    # Calculate statistics
    statistics = BatchStatistics(results)

    elapsed_time = time.time() - start_time

    logger.info(
        f"Dataset evaluation complete: {len(results)} traces in {elapsed_time:.1f}s "
        f"({len(results)/elapsed_time:.1f} traces/sec), "
        f"pass_rate={statistics.pass_rate:.2%}"
    )

    return results, statistics

export_results(results, statistics, output_path, format='json')

Export evaluation results to file.

Parameters:

Name	Type	Description	Default
results	List[EvaluationResult]	List of evaluation results	required
statistics	Optional[BatchStatistics]	Batch statistics (optional)	required
output_path	str	Path to output file	required
format	str	Export format ("json", "csv", "html")	'json'

Raises:

Type	Description
ValueError	If format is not supported

Source code in uaef/evaluation/offline.py

def export_results(
    self,
    results: List[EvaluationResult],
    statistics: Optional[BatchStatistics],
    output_path: str,
    format: str = "json"
) -> None:
    """
    Export evaluation results to file.

    Args:
        results: List of evaluation results
        statistics: Batch statistics (optional)
        output_path: Path to output file
        format: Export format ("json", "csv", "html")

    Raises:
        ValueError: If format is not supported
    """
    logger.info(f"Exporting {len(results)} results to {output_path} ({format})")

    output_file = Path(output_path)
    output_file.parent.mkdir(parents=True, exist_ok=True)

    if format == "json":
        self._export_json(results, statistics, output_file)
    elif format == "csv":
        self._export_csv(results, statistics, output_file)
    elif format == "html":
        self._export_html(results, statistics, output_file)
    else:
        raise ValueError(f"Unsupported export format: {format}")

    logger.info(f"Results exported to {output_path}")

uaef.config

Configuration management for UAEF.

AWSConfig

Bases: BaseModel

AWS configuration for Bedrock LLM Judge.

Source code in uaef/config.py

class AWSConfig(BaseModel):
    """AWS configuration for Bedrock LLM Judge."""

    region: str = Field(
        default="us-east-1",
        description="AWS region for Bedrock"
    )
    access_key_id: Optional[str] = Field(
        None,
        description="AWS access key ID (optional, can use IAM role)"
    )
    secret_access_key: Optional[str] = Field(
        None,
        description="AWS secret access key (optional, can use IAM role)"
    )
    session_token: Optional[str] = Field(
        None,
        description="AWS session token (optional)"
    )

    @classmethod
    def from_env(cls) -> "AWSConfig":
        """Load AWS configuration from environment variables."""
        return cls(
            region=os.getenv("AWS_REGION", "us-east-1"),
            access_key_id=os.getenv("AWS_ACCESS_KEY_ID"),
            secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY"),
            session_token=os.getenv("AWS_SESSION_TOKEN"),
        )

from_env() classmethod

Load AWS configuration from environment variables.

Source code in uaef/config.py

@classmethod
def from_env(cls) -> "AWSConfig":
    """Load AWS configuration from environment variables."""
    return cls(
        region=os.getenv("AWS_REGION", "us-east-1"),
        access_key_id=os.getenv("AWS_ACCESS_KEY_ID"),
        secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY"),
        session_token=os.getenv("AWS_SESSION_TOKEN"),
    )

InsightConfig

Bases: BaseModel

Configuration for LLM-powered insight generation.

Defaults match LLMJudgeConfig but with slightly higher temperature for more natural prose output.

Source code in uaef/config.py

class InsightConfig(BaseModel):
    """Configuration for LLM-powered insight generation.

    Defaults match LLMJudgeConfig but with slightly higher temperature
    for more natural prose output.
    """

    model_id: str = Field(
        default="us.anthropic.claude-sonnet-4-6",
        description="Bedrock model ID for insight generation"
    )
    temperature: float = Field(
        default=0.3,
        description="Temperature for insight generation (slightly creative)",
        ge=0.0,
        le=1.0
    )
    max_tokens: int = Field(
        default=4096,
        description="Maximum tokens for insight response",
        gt=0
    )
    retry_attempts: int = Field(
        default=3,
        description="Number of retry attempts for failed API calls",
        ge=1
    )
    retry_delay: float = Field(
        default=1.0,
        description="Initial delay between retries in seconds",
        gt=0.0
    )

LLMJudgeConfig

Bases: BaseModel

Configuration for LLM Judge.

Source code in uaef/config.py

class LLMJudgeConfig(BaseModel):
    """Configuration for LLM Judge."""

    model_id: str = Field(
        default="us.anthropic.claude-sonnet-4-6",
        description="Bedrock model ID for LLM judge"
    )
    temperature: float = Field(
        default=0.0,
        description="Temperature for LLM generation",
        ge=0.0,
        le=1.0
    )
    max_tokens: int = Field(
        default=2048,
        description="Maximum tokens for LLM response",
        gt=0
    )
    retry_attempts: int = Field(
        default=3,
        description="Number of retry attempts for failed API calls",
        ge=1
    )
    retry_delay: float = Field(
        default=1.0,
        description="Initial delay between retries in seconds",
        gt=0.0
    )

    @field_validator("temperature")
    @classmethod
    def validate_temperature(cls, v: float) -> float:
        """Validate temperature is between 0 and 1."""
        if not 0.0 <= v <= 1.0:
            raise ValueError(f"Temperature must be between 0 and 1, got {v}")
        return v

    @field_validator("model_id")
    @classmethod
    def validate_model_id(cls, v: str) -> str:
        """Reject a judge model_id outside the vetted Anthropic Claude family."""
        return _validate_judge_model_id(v)

validate_model_id(v) classmethod

Reject a judge model_id outside the vetted Anthropic Claude family.

Source code in uaef/config.py

@field_validator("model_id")
@classmethod
def validate_model_id(cls, v: str) -> str:
    """Reject a judge model_id outside the vetted Anthropic Claude family."""
    return _validate_judge_model_id(v)

validate_temperature(v) classmethod

Validate temperature is between 0 and 1.

Source code in uaef/config.py

@field_validator("temperature")
@classmethod
def validate_temperature(cls, v: float) -> float:
    """Validate temperature is between 0 and 1."""
    if not 0.0 <= v <= 1.0:
        raise ValueError(f"Temperature must be between 0 and 1, got {v}")
    return v

StorageConfig

Bases: BaseModel

Configuration for DynamoDB + S3 persistence.

Source code in uaef/config.py

class StorageConfig(BaseModel):
    """Configuration for DynamoDB + S3 persistence."""

    dynamodb_table_name: str = Field(
        default="uaef-experiments",
        description="DynamoDB table name for experiment metadata"
    )
    s3_bucket: Optional[str] = Field(
        default=None,
        description=(
            "S3 bucket for full evaluation results. Defaults to "
            "uaef-results-<account-id>-<region>: bucket names are global, so a "
            "fixed default could resolve to a bucket owned by another account."
        )
    )
    s3_prefix: str = Field(
        default="evaluations/",
        description="S3 key prefix for result files"
    )
    region: Optional[str] = Field(
        None,
        description="AWS region override (defaults to aws.region if not set)"
    )

    @classmethod
    def from_env(cls) -> "StorageConfig":
        """Load storage configuration from environment variables."""
        return cls(
            dynamodb_table_name=os.getenv("UAEF_DYNAMODB_TABLE", "uaef-experiments"),
            s3_bucket=os.getenv("UAEF_S3_BUCKET") or None,
            s3_prefix=os.getenv("UAEF_S3_PREFIX", "evaluations/"),
            region=os.getenv("UAEF_STORAGE_REGION"),
        )

from_env() classmethod

Load storage configuration from environment variables.

Source code in uaef/config.py

@classmethod
def from_env(cls) -> "StorageConfig":
    """Load storage configuration from environment variables."""
    return cls(
        dynamodb_table_name=os.getenv("UAEF_DYNAMODB_TABLE", "uaef-experiments"),
        s3_bucket=os.getenv("UAEF_S3_BUCKET") or None,
        s3_prefix=os.getenv("UAEF_S3_PREFIX", "evaluations/"),
        region=os.getenv("UAEF_STORAGE_REGION"),
    )

UAEFConfig

Bases: BaseModel

Main UAEF configuration.

Source code in uaef/config.py

class UAEFConfig(BaseModel):
    """Main UAEF configuration."""

    aws: AWSConfig = Field(
        default_factory=AWSConfig,
        description="AWS configuration"
    )
    llm_judge: LLMJudgeConfig = Field(
        default_factory=LLMJudgeConfig,
        description="LLM Judge configuration"
    )
    insight: InsightConfig = Field(
        default_factory=InsightConfig,
        description="LLM-powered insight generation configuration"
    )
    storage: StorageConfig = Field(
        default_factory=StorageConfig,
        description="DynamoDB + S3 storage configuration"
    )
    log_level: str = Field(
        default="INFO",
        description="Logging level"
    )
    enable_caching: bool = Field(
        default=True,
        description="Enable caching for LLM responses"
    )
    cache_ttl_seconds: int = Field(
        default=3600,
        description="Cache TTL in seconds",
        gt=0
    )

    @field_validator("log_level")
    @classmethod
    def validate_log_level(cls, v: str) -> str:
        """Validate log level is valid."""
        valid_levels = ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
        v_upper = v.upper()
        if v_upper not in valid_levels:
            raise ValueError(f"Invalid log level: {v}. Must be one of {valid_levels}")
        return v_upper

    @classmethod
    def from_env(cls) -> "UAEFConfig":
        """Load configuration from environment variables."""
        return cls(
            aws=AWSConfig.from_env(),
            llm_judge=LLMJudgeConfig(
                model_id=os.getenv(
                    "UAEF_LLM_MODEL_ID",
                    "us.anthropic.claude-sonnet-4-6"
                ),
                temperature=float(os.getenv("UAEF_LLM_TEMPERATURE", "0.0")),
                max_tokens=int(os.getenv("UAEF_LLM_MAX_TOKENS", "2048")),
            ),
            insight=InsightConfig(
                model_id=os.getenv(
                    "UAEF_INSIGHT_MODEL_ID",
                    os.getenv("UAEF_LLM_MODEL_ID", "us.anthropic.claude-sonnet-4-6")
                ),
                temperature=float(os.getenv("UAEF_INSIGHT_TEMPERATURE", "0.3")),
                max_tokens=int(os.getenv("UAEF_INSIGHT_MAX_TOKENS", "4096")),
            ),
            storage=StorageConfig.from_env(),
            log_level=os.getenv("UAEF_LOG_LEVEL", "INFO"),
            enable_caching=os.getenv("UAEF_ENABLE_CACHING", "true").lower() == "true",
            cache_ttl_seconds=int(os.getenv("UAEF_CACHE_TTL_SECONDS", "3600")),
        )

    @classmethod
    def from_file(cls, config_path: Union[str, Path]) -> "UAEFConfig":
        """
        Load configuration from a YAML or JSON file.

        Args:
            config_path: Path to configuration file

        Returns:
            UAEFConfig instance

        Raises:
            FileNotFoundError: If config file doesn't exist
            ValueError: If config file format is invalid
        """
        config_path = Path(config_path)
        if not config_path.exists():
            raise FileNotFoundError(f"Configuration file not found: {config_path}")

        with open(config_path, "r") as f:
            if config_path.suffix in [".yaml", ".yml"]:
                config_dict = yaml.safe_load(f) or {}
            elif config_path.suffix == ".json":
                import json
                config_dict = json.load(f)
            else:
                raise ValueError(
                    f"Unsupported config file format: {config_path.suffix}. "
                    "Use .yaml, .yml, or .json"
                )

        return cls(**config_dict)

    def to_dict(self) -> Dict[str, Any]:
        """Convert configuration to dictionary."""
        return self.model_dump()

    def save_to_file(self, config_path: Path) -> None:
        """
        Save configuration to a YAML or JSON file.

        Args:
            config_path: Path to save configuration file
        """
        config_dict = self.to_dict()

        with open(config_path, "w") as f:
            if config_path.suffix in [".yaml", ".yml"]:
                yaml.safe_dump(config_dict, f, default_flow_style=False)
            elif config_path.suffix == ".json":
                import json
                json.dump(config_dict, f, indent=2)
            else:
                raise ValueError(
                    f"Unsupported config file format: {config_path.suffix}. "
                    "Use .yaml, .yml, or .json"
                )

from_env() classmethod

Load configuration from environment variables.

Source code in uaef/config.py

@classmethod
def from_env(cls) -> "UAEFConfig":
    """Load configuration from environment variables."""
    return cls(
        aws=AWSConfig.from_env(),
        llm_judge=LLMJudgeConfig(
            model_id=os.getenv(
                "UAEF_LLM_MODEL_ID",
                "us.anthropic.claude-sonnet-4-6"
            ),
            temperature=float(os.getenv("UAEF_LLM_TEMPERATURE", "0.0")),
            max_tokens=int(os.getenv("UAEF_LLM_MAX_TOKENS", "2048")),
        ),
        insight=InsightConfig(
            model_id=os.getenv(
                "UAEF_INSIGHT_MODEL_ID",
                os.getenv("UAEF_LLM_MODEL_ID", "us.anthropic.claude-sonnet-4-6")
            ),
            temperature=float(os.getenv("UAEF_INSIGHT_TEMPERATURE", "0.3")),
            max_tokens=int(os.getenv("UAEF_INSIGHT_MAX_TOKENS", "4096")),
        ),
        storage=StorageConfig.from_env(),
        log_level=os.getenv("UAEF_LOG_LEVEL", "INFO"),
        enable_caching=os.getenv("UAEF_ENABLE_CACHING", "true").lower() == "true",
        cache_ttl_seconds=int(os.getenv("UAEF_CACHE_TTL_SECONDS", "3600")),
    )

from_file(config_path) classmethod

Load configuration from a YAML or JSON file.

Parameters:

Name	Type	Description	Default
config_path	Union[str, Path]	Path to configuration file	required

Returns:

Type	Description
UAEFConfig	UAEFConfig instance

Raises:

Type	Description
FileNotFoundError	If config file doesn't exist
ValueError	If config file format is invalid

Source code in uaef/config.py

@classmethod
def from_file(cls, config_path: Union[str, Path]) -> "UAEFConfig":
    """
    Load configuration from a YAML or JSON file.

    Args:
        config_path: Path to configuration file

    Returns:
        UAEFConfig instance

    Raises:
        FileNotFoundError: If config file doesn't exist
        ValueError: If config file format is invalid
    """
    config_path = Path(config_path)
    if not config_path.exists():
        raise FileNotFoundError(f"Configuration file not found: {config_path}")

    with open(config_path, "r") as f:
        if config_path.suffix in [".yaml", ".yml"]:
            config_dict = yaml.safe_load(f) or {}
        elif config_path.suffix == ".json":
            import json
            config_dict = json.load(f)
        else:
            raise ValueError(
                f"Unsupported config file format: {config_path.suffix}. "
                "Use .yaml, .yml, or .json"
            )

    return cls(**config_dict)

save_to_file(config_path)

Save configuration to a YAML or JSON file.

Parameters:

Name	Type	Description	Default
config_path	Path	Path to save configuration file	required

Source code in uaef/config.py

def save_to_file(self, config_path: Path) -> None:
    """
    Save configuration to a YAML or JSON file.

    Args:
        config_path: Path to save configuration file
    """
    config_dict = self.to_dict()

    with open(config_path, "w") as f:
        if config_path.suffix in [".yaml", ".yml"]:
            yaml.safe_dump(config_dict, f, default_flow_style=False)
        elif config_path.suffix == ".json":
            import json
            json.dump(config_dict, f, indent=2)
        else:
            raise ValueError(
                f"Unsupported config file format: {config_path.suffix}. "
                "Use .yaml, .yml, or .json"
            )

to_dict()

Convert configuration to dictionary.

Source code in uaef/config.py

def to_dict(self) -> Dict[str, Any]:
    """Convert configuration to dictionary."""
    return self.model_dump()

validate_log_level(v) classmethod

Validate log level is valid.

Source code in uaef/config.py

@field_validator("log_level")
@classmethod
def validate_log_level(cls, v: str) -> str:
    """Validate log level is valid."""
    valid_levels = ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
    v_upper = v.upper()
    if v_upper not in valid_levels:
        raise ValueError(f"Invalid log level: {v}. Must be one of {valid_levels}")
    return v_upper

get_config()

Get the global UAEF configuration.

Loads from environment variables on first call.

Returns:

Type	Description
UAEFConfig	UAEFConfig instance

Source code in uaef/config.py

def get_config() -> UAEFConfig:
    """
    Get the global UAEF configuration.

    Loads from environment variables on first call.

    Returns:
        UAEFConfig instance
    """
    global _config
    if _config is None:
        _config = UAEFConfig.from_env()
    return _config

reset_config()

Reset the global configuration to None.

Source code in uaef/config.py

def reset_config() -> None:
    """Reset the global configuration to None."""
    global _config
    _config = None

set_config(config)

Set the global UAEF configuration.

Parameters:

Name	Type	Description	Default
config	UAEFConfig	UAEFConfig instance to set as global	required

Source code in uaef/config.py

def set_config(config: UAEFConfig) -> None:
    """
    Set the global UAEF configuration.

    Args:
        config: UAEFConfig instance to set as global
    """
    global _config
    _config = config