Analysis

Turns evaluation results into diagnoses: why traces failed, which failures share a cause, and what to change.

uaef.analysis.root_cause

Root cause analyzer for evaluation failures.

FailureCategory

Bases: BaseModel

Category of failure with associated metrics.

Attributes:

Name	Type	Description
category	str	Type of failure (tool_selection, reasoning, context, response)
failed_metrics	List[str]	List of metric names that failed in this category
severity	str	Severity level (low, medium, high, critical)
description	str	Human-readable description of the failure

Source code in uaef/analysis/root_cause.py

class FailureCategory(BaseModel):
    """
    Category of failure with associated metrics.

    Attributes:
        category: Type of failure (tool_selection, reasoning, context, response)
        failed_metrics: List of metric names that failed in this category
        severity: Severity level (low, medium, high, critical)
        description: Human-readable description of the failure
    """

    category: str = Field(..., description="Type of failure")
    failed_metrics: List[str] = Field(
        default_factory=list,
        description="List of metric names that failed in this category"
    )
    severity: str = Field(..., description="Severity level")
    description: str = Field(..., description="Human-readable description of the failure")

    class Config:
        """Pydantic configuration."""
        json_schema_extra = {
            "example": {
                "category": "tool_selection",
                "failed_metrics": ["tool_accuracy", "tool_sequence_accuracy"],
                "severity": "high",
                "description": "Agent selected incorrect tools or used them in wrong order"
            }
        }

Config

Pydantic configuration.

Source code in uaef/analysis/root_cause.py

class Config:
    """Pydantic configuration."""
    json_schema_extra = {
        "example": {
            "category": "tool_selection",
            "failed_metrics": ["tool_accuracy", "tool_sequence_accuracy"],
            "severity": "high",
            "description": "Agent selected incorrect tools or used them in wrong order"
        }
    }

FailurePattern

Bases: BaseModel

Pattern identified across multiple evaluation failures.

Attributes:

Name	Type	Description
pattern_type	str	Type of pattern (e.g., "consistent_tool_error", "context_loss")
frequency	int	Number of evaluations exhibiting this pattern
affected_metrics	List[str]	List of metrics commonly affected
description	str	Description of the pattern
severity	str	Average severity across occurrences
recommendations	List[str]	Recommendations to address this pattern

Source code in uaef/analysis/root_cause.py

class FailurePattern(BaseModel):
    """
    Pattern identified across multiple evaluation failures.

    Attributes:
        pattern_type: Type of pattern (e.g., "consistent_tool_error", "context_loss")
        frequency: Number of evaluations exhibiting this pattern
        affected_metrics: List of metrics commonly affected
        description: Description of the pattern
        severity: Average severity across occurrences
        recommendations: Recommendations to address this pattern
    """

    pattern_type: str = Field(..., description="Type of pattern")
    frequency: int = Field(..., description="Number of evaluations exhibiting this pattern")
    affected_metrics: List[str] = Field(
        default_factory=list,
        description="List of metrics commonly affected"
    )
    description: str = Field(..., description="Description of the pattern")
    severity: str = Field(..., description="Average severity across occurrences")
    recommendations: List[str] = Field(
        default_factory=list,
        description="Recommendations to address this pattern"
    )

    class Config:
        """Pydantic configuration."""
        json_schema_extra = {
            "example": {
                "pattern_type": "consistent_tool_error",
                "frequency": 15,
                "affected_metrics": ["tool_accuracy", "tool_sequence_accuracy"],
                "description": "Agent consistently selects wrong tools for data retrieval tasks",
                "severity": "high",
                "recommendations": [
                    "Add more tool selection examples to system prompt",
                    "Consider simplifying tool descriptions"
                ]
            }
        }

Config

Pydantic configuration.

Source code in uaef/analysis/root_cause.py

class Config:
    """Pydantic configuration."""
    json_schema_extra = {
        "example": {
            "pattern_type": "consistent_tool_error",
            "frequency": 15,
            "affected_metrics": ["tool_accuracy", "tool_sequence_accuracy"],
            "description": "Agent consistently selects wrong tools for data retrieval tasks",
            "severity": "high",
            "recommendations": [
                "Add more tool selection examples to system prompt",
                "Consider simplifying tool descriptions"
            ]
        }
    }

RootCauseAnalyzer

Analyzer for identifying root causes of evaluation failures.

Provides methods for: - Analyzing individual evaluation failures - Identifying patterns across multiple failures - Categorizing failures by type - Generating actionable recommendations

Source code in uaef/analysis/root_cause.py

class RootCauseAnalyzer:
    """
    Analyzer for identifying root causes of evaluation failures.

    Provides methods for:
    - Analyzing individual evaluation failures
    - Identifying patterns across multiple failures
    - Categorizing failures by type
    - Generating actionable recommendations
    """

    # Metric to category mapping
    METRIC_CATEGORIES = {
        # Tool calling metrics
        "tool_accuracy": "tool_selection",
        "tool_call_accuracy": "tool_selection",
        "tool_sequence_accuracy": "tool_selection",
        "parameter_quality": "tool_selection",
        "incorrect_tool_rate": "tool_selection",
        "missed_tool_rate": "tool_selection",

        # Response quality metrics
        "answer_relevance": "response",
        "answer_correctness": "response",
        "completeness": "response",
        "hallucination_score": "response",
        "faithfulness": "response",

        # Context metrics
        "context_retention": "context",
        "context_relevance": "context",
        "context_precision": "context",
        "context_recall": "context",

        # Reasoning metrics
        "reasoning_coherence": "reasoning",
        "logical_consistency": "reasoning",
        "reasoning_completeness": "reasoning",
        "fallacy_detection": "reasoning",
        "chain_of_thought_quality": "reasoning",
    }

    def __init__(self, severity_thresholds: Optional[Dict[str, float]] = None):
        """
        Initialize the root cause analyzer.

        Args:
            severity_thresholds: Custom thresholds for severity classification
                                Format: {"critical": 0.3, "high": 0.5, "medium": 0.7}
        """
        self.severity_thresholds = severity_thresholds or {
            "critical": 0.3,
            "high": 0.5,
            "medium": 0.7,
            "low": 1.0
        }
        logger.info("RootCauseAnalyzer initialized")

    def analyze_failure(
        self,
        evaluation_result: EvaluationResult
    ) -> RootCauseReport:
        """
        Analyze a single failed evaluation to identify root causes.

        Args:
            evaluation_result: EvaluationResult with failures

        Returns:
            RootCauseReport with identified root causes and recommendations

        Raises:
            ValueError: If evaluation_result passed (no failures to analyze)

        Example:
            >>> analyzer = RootCauseAnalyzer()
            >>> report = analyzer.analyze_failure(failed_evaluation)
            >>> print(f"Primary root cause: {report.primary_root_cause}")
            >>> print(f"Recommendations: {report.recommendations}")
        """
        if evaluation_result.passed:
            raise ValueError(
                f"Cannot analyze failure for passed evaluation {evaluation_result.evaluation_id}"
            )

        logger.info(
            f"Analyzing failure for evaluation {evaluation_result.evaluation_id}, "
            f"trace {evaluation_result.trace_id}"
        )

        # Collect failed metrics by dimension
        failed_metrics_by_dimension: Dict[str, List[str]] = defaultdict(list)
        failed_metrics_by_category: Dict[str, List[str]] = defaultdict(list)

        for dimension_result in evaluation_result.dimension_results:
            for metric_score in dimension_result.scored_metrics():
                # Consider a metric failed if score is below 0.7 (typical threshold)
                if metric_score.score < 0.7:
                    failed_metrics_by_dimension[dimension_result.dimension_name].append(
                        metric_score.metric_name
                    )

                    # Map to category
                    category = self._get_metric_category(metric_score.metric_name)
                    failed_metrics_by_category[category].append(metric_score.metric_name)

        # Create failure categories
        failure_categories = []
        for category, metrics in failed_metrics_by_category.items():
            if not metrics:
                continue

            # Calculate severity based on number of failed metrics and their scores
            severity = self._calculate_severity(evaluation_result, metrics)

            # Generate description
            description = self._generate_category_description(category, metrics)

            failure_categories.append(FailureCategory(
                category=category,
                failed_metrics=list(set(metrics)),  # Remove duplicates
                severity=severity,
                description=description
            ))

        # Determine primary root cause (category with most failures or highest severity)
        primary_root_cause = self._determine_primary_root_cause(failure_categories)

        # Generate recommendations
        recommendations = self._generate_recommendations(
            failure_categories,
            primary_root_cause,
            evaluation_result
        )

        # Collect details
        details = {
            "overall_score": evaluation_result.overall_score,
            "failed_dimensions": list(failed_metrics_by_dimension.keys()),
            "total_failed_metrics": sum(len(m) for m in failed_metrics_by_category.values()),
            "failures": evaluation_result.failures,
            "warnings": evaluation_result.warnings
        }

        report = RootCauseReport(
            evaluation_id=str(evaluation_result.evaluation_id),
            trace_id=str(evaluation_result.trace_id),
            failure_categories=failure_categories,
            primary_root_cause=primary_root_cause,
            recommendations=recommendations,
            details=details
        )

        logger.info(
            f"Root cause analysis complete for {evaluation_result.evaluation_id}: "
            f"primary cause = {primary_root_cause}, "
            f"{len(failure_categories)} categories identified"
        )

        return report

    def identify_failure_patterns(
        self,
        evaluation_results: List[EvaluationResult]
    ) -> List[FailurePattern]:
        """
        Identify patterns across multiple evaluation failures.

        Args:
            evaluation_results: List of EvaluationResult instances (passed and failed)

        Returns:
            List of FailurePattern instances, sorted by frequency (descending)

        Example:
            >>> analyzer = RootCauseAnalyzer()
            >>> patterns = analyzer.identify_failure_patterns(evaluation_results)
            >>> for pattern in patterns:
            ...     print(f"{pattern.pattern_type}: {pattern.frequency} occurrences")
        """
        if not evaluation_results:
            logger.warning("No evaluation results provided for pattern identification")
            return []

        # Filter to failed evaluations
        failed_evaluations = [r for r in evaluation_results if not r.passed]

        if not failed_evaluations:
            logger.info("No failed evaluations found in provided results")
            return []

        logger.info(
            f"Identifying failure patterns across {len(failed_evaluations)} failed evaluations "
            f"(out of {len(evaluation_results)} total)"
        )

        # Track metrics that fail together
        category_failures: Dict[str, int] = Counter()
        metric_failures: Dict[str, int] = Counter()
        category_to_metrics: Dict[str, List[str]] = defaultdict(list)

        for result in failed_evaluations:
            failed_metrics_in_eval = set()

            for dimension_result in result.dimension_results:
                for metric_score in dimension_result.scored_metrics():
                    if metric_score.score < 0.7:
                        metric_name = metric_score.metric_name
                        failed_metrics_in_eval.add(metric_name)
                        metric_failures[metric_name] += 1

                        category = self._get_metric_category(metric_name)
                        category_to_metrics[category].append(metric_name)

            # Count category failures
            categories_in_eval = set(
                self._get_metric_category(m) for m in failed_metrics_in_eval
            )
            for category in categories_in_eval:
                category_failures[category] += 1

        # Generate patterns from categories
        patterns = []
        for category, frequency in category_failures.most_common():
            # Get most common metrics in this category
            metrics_in_category = category_to_metrics[category]
            metric_counter = Counter(metrics_in_category)
            affected_metrics = [m for m, _ in metric_counter.most_common(5)]

            # Determine severity based on frequency
            severity = self._calculate_pattern_severity(frequency, len(failed_evaluations))

            # Generate pattern description
            description = self._generate_pattern_description(
                category,
                frequency,
                len(failed_evaluations),
                affected_metrics
            )

            # Generate recommendations
            recommendations = self._generate_pattern_recommendations(
                category,
                affected_metrics,
                frequency
            )

            pattern = FailurePattern(
                pattern_type=f"{category}_failures",
                frequency=frequency,
                affected_metrics=affected_metrics,
                description=description,
                severity=severity,
                recommendations=recommendations
            )
            patterns.append(pattern)

        logger.info(f"Identified {len(patterns)} failure patterns")

        return patterns

    def _get_metric_category(self, metric_name: str) -> str:
        """
        Get the category for a metric name.

        Args:
            metric_name: Name of the metric

        Returns:
            Category name (tool_selection, reasoning, context, response, or unknown)
        """
        return self.METRIC_CATEGORIES.get(metric_name, "unknown")

    def _calculate_severity(
        self,
        evaluation_result: EvaluationResult,
        failed_metrics: List[str]
    ) -> str:
        """
        Calculate severity level based on scores and number of failures.

        Args:
            evaluation_result: The evaluation result
            failed_metrics: List of failed metric names

        Returns:
            Severity level: "critical", "high", "medium", or "low"
        """
        # Get average score of failed metrics
        scores = []
        for dimension_result in evaluation_result.dimension_results:
            for metric_score in dimension_result.scored_metrics():
                if metric_score.metric_name in failed_metrics:
                    scores.append(metric_score.score)

        if not scores:
            return "low"

        avg_score = sum(scores) / len(scores)

        # Determine severity based on thresholds
        for severity, threshold in sorted(
            self.severity_thresholds.items(),
            key=lambda x: x[1]
        ):
            if avg_score <= threshold:
                return severity

        return "low"

    def _generate_category_description(
        self,
        category: str,
        failed_metrics: List[str]
    ) -> str:
        """
        Generate a human-readable description for a failure category.

        Args:
            category: Category name
            failed_metrics: List of failed metrics in this category

        Returns:
            Human-readable description
        """
        # Get unique metrics and limit to first 3
        unique_metrics = list(set(failed_metrics))[:3]
        metrics_str = ', '.join(unique_metrics)

        descriptions = {
            "tool_selection": (
                f"Agent had issues with tool selection and usage. "
                f"Failed metrics: {metrics_str}"
            ),
            "reasoning": (
                f"Agent's reasoning process had logical issues or inconsistencies. "
                f"Failed metrics: {metrics_str}"
            ),
            "context": (
                f"Agent failed to properly retain or use context information. "
                f"Failed metrics: {metrics_str}"
            ),
            "response": (
                f"Agent's response quality was insufficient. "
                f"Failed metrics: {metrics_str}"
            ),
            "unknown": (
                f"Unclassified failure. "
                f"Failed metrics: {metrics_str}"
            )
        }

        return descriptions.get(category, descriptions["unknown"])

    def _determine_primary_root_cause(
        self,
        failure_categories: List[FailureCategory]
    ) -> str:
        """
        Determine the primary root cause from failure categories.

        Args:
            failure_categories: List of failure categories

        Returns:
            Primary root cause category name
        """
        if not failure_categories:
            return "unknown"

        # Severity weights
        severity_weights = {
            "critical": 4,
            "high": 3,
            "medium": 2,
            "low": 1
        }

        # Calculate weighted score for each category
        category_scores = []
        for fc in failure_categories:
            weight = severity_weights.get(fc.severity, 1)
            score = len(fc.failed_metrics) * weight
            category_scores.append((fc.category, score))

        # Return category with highest score
        primary = max(category_scores, key=lambda x: x[1])
        return primary[0]

    def _generate_recommendations(
        self,
        failure_categories: List[FailureCategory],
        primary_root_cause: str,
        evaluation_result: EvaluationResult
    ) -> List[str]:
        """
        Generate actionable recommendations based on failure analysis.

        Args:
            failure_categories: List of failure categories
            primary_root_cause: Primary root cause
            evaluation_result: The evaluation result

        Returns:
            List of actionable recommendations
        """
        recommendations = []

        # Category-specific recommendations
        category_recommendations = {
            "tool_selection": [
                "Review and improve tool selection logic in agent prompt",
                "Add more examples of correct tool usage to system prompt",
                "Verify tool descriptions are clear and unambiguous",
                "Consider simplifying the tool set or combining similar tools"
            ],
            "reasoning": [
                "Enhance chain-of-thought prompting in system instructions",
                "Add explicit reasoning validation steps",
                "Include examples of correct reasoning patterns",
                "Consider using a reasoning-optimized model"
            ],
            "context": [
                "Improve context retention mechanisms",
                "Add explicit context summarization steps",
                "Reduce context window usage to prevent information loss",
                "Implement better context retrieval strategies"
            ],
            "response": [
                "Improve response generation prompts",
                "Add quality checks before returning responses",
                "Include more examples of high-quality responses",
                "Consider post-processing to improve response quality"
            ]
        }

        # Add recommendations for primary root cause
        if primary_root_cause in category_recommendations:
            recommendations.extend(category_recommendations[primary_root_cause][:2])

        # Add recommendations for other significant categories
        for fc in failure_categories:
            if fc.category != primary_root_cause and fc.severity in ["critical", "high"]:
                if fc.category in category_recommendations:
                    recommendations.append(category_recommendations[fc.category][0])

        # Add general recommendation based on overall score
        if evaluation_result.overall_score < 0.3:
            recommendations.insert(
                0,
                "Overall score is critically low - consider comprehensive agent redesign"
            )

        return recommendations[:5]  # Limit to top 5 recommendations

    def _calculate_pattern_severity(
        self,
        frequency: int,
        total_failures: int
    ) -> str:
        """
        Calculate severity for a failure pattern based on frequency.

        Args:
            frequency: Number of times pattern occurred
            total_failures: Total number of failed evaluations

        Returns:
            Severity level: "critical", "high", "medium", or "low"
        """
        percentage = (frequency / total_failures) * 100 if total_failures > 0 else 0

        if percentage >= 75:
            return "critical"
        elif percentage >= 50:
            return "high"
        elif percentage >= 25:
            return "medium"
        else:
            return "low"

    def _generate_pattern_description(
        self,
        category: str,
        frequency: int,
        total_failures: int,
        affected_metrics: List[str]
    ) -> str:
        """
        Generate description for a failure pattern.

        Args:
            category: Category name
            frequency: Number of occurrences
            total_failures: Total number of failures
            affected_metrics: List of commonly affected metrics

        Returns:
            Human-readable description
        """
        percentage = (frequency / total_failures) * 100 if total_failures > 0 else 0

        category_descriptions = {
            "tool_selection": "tool selection and usage",
            "reasoning": "reasoning and logical consistency",
            "context": "context retention and usage",
            "response": "response quality and accuracy",
            "unknown": "unclassified issues"
        }

        category_desc = category_descriptions.get(category, category)

        return (
            f"Agent consistently fails on {category_desc} "
            f"({frequency}/{total_failures} failures, {percentage:.1f}%). "
            f"Most affected metrics: {', '.join(affected_metrics[:3])}"
        )

    def _generate_pattern_recommendations(
        self,
        category: str,
        affected_metrics: List[str],
        frequency: int
    ) -> List[str]:
        """
        Generate recommendations for addressing a failure pattern.

        Args:
            category: Category name
            affected_metrics: List of affected metrics
            frequency: Number of occurrences

        Returns:
            List of recommendations
        """
        base_recommendations = {
            "tool_selection": [
                "Systematically review and improve tool selection prompts",
                "Add comprehensive tool usage examples to training data",
                "Consider implementing tool selection validation logic",
                "Simplify tool descriptions and reduce tool set complexity"
            ],
            "reasoning": [
                "Implement structured reasoning frameworks (e.g., ReAct, Chain-of-Thought)",
                "Add reasoning validation checkpoints in agent workflow",
                "Use reasoning-optimized models or fine-tuning",
                "Include diverse reasoning examples in system prompt"
            ],
            "context": [
                "Implement robust context management strategies",
                "Add context summarization between conversation turns",
                "Use retrieval-augmented generation (RAG) for long contexts",
                "Optimize context window usage and prioritization"
            ],
            "response": [
                "Enhance response generation with quality templates",
                "Implement response validation before returning to user",
                "Add post-processing steps for response refinement",
                "Use higher-quality models for response generation"
            ]
        }

        recommendations = base_recommendations.get(category, [
            "Investigate root cause through detailed trace analysis",
            "Consider agent architecture improvements",
            "Add monitoring and alerting for this failure type"
        ])

        # Add urgency note for high-frequency patterns
        if frequency > 10:
            recommendations.insert(
                0,
                f"HIGH PRIORITY: This pattern occurs frequently ({frequency} times) - "
                "immediate action recommended"
            )

        return recommendations[:4]  # Limit to top 4 recommendations

analyze_failure(evaluation_result)

Analyze a single failed evaluation to identify root causes.

Parameters:

Name	Type	Description	Default
evaluation_result	EvaluationResult	EvaluationResult with failures	required

Returns:

Type	Description
RootCauseReport	RootCauseReport with identified root causes and recommendations

Raises:

Type	Description
ValueError	If evaluation_result passed (no failures to analyze)

Example

analyzer = RootCauseAnalyzer() report = analyzer.analyze_failure(failed_evaluation) print(f"Primary root cause: {report.primary_root_cause}") print(f"Recommendations: {report.recommendations}")

Source code in uaef/analysis/root_cause.py

def analyze_failure(
    self,
    evaluation_result: EvaluationResult
) -> RootCauseReport:
    """
    Analyze a single failed evaluation to identify root causes.

    Args:
        evaluation_result: EvaluationResult with failures

    Returns:
        RootCauseReport with identified root causes and recommendations

    Raises:
        ValueError: If evaluation_result passed (no failures to analyze)

    Example:
        >>> analyzer = RootCauseAnalyzer()
        >>> report = analyzer.analyze_failure(failed_evaluation)
        >>> print(f"Primary root cause: {report.primary_root_cause}")
        >>> print(f"Recommendations: {report.recommendations}")
    """
    if evaluation_result.passed:
        raise ValueError(
            f"Cannot analyze failure for passed evaluation {evaluation_result.evaluation_id}"
        )

    logger.info(
        f"Analyzing failure for evaluation {evaluation_result.evaluation_id}, "
        f"trace {evaluation_result.trace_id}"
    )

    # Collect failed metrics by dimension
    failed_metrics_by_dimension: Dict[str, List[str]] = defaultdict(list)
    failed_metrics_by_category: Dict[str, List[str]] = defaultdict(list)

    for dimension_result in evaluation_result.dimension_results:
        for metric_score in dimension_result.scored_metrics():
            # Consider a metric failed if score is below 0.7 (typical threshold)
            if metric_score.score < 0.7:
                failed_metrics_by_dimension[dimension_result.dimension_name].append(
                    metric_score.metric_name
                )

                # Map to category
                category = self._get_metric_category(metric_score.metric_name)
                failed_metrics_by_category[category].append(metric_score.metric_name)

    # Create failure categories
    failure_categories = []
    for category, metrics in failed_metrics_by_category.items():
        if not metrics:
            continue

        # Calculate severity based on number of failed metrics and their scores
        severity = self._calculate_severity(evaluation_result, metrics)

        # Generate description
        description = self._generate_category_description(category, metrics)

        failure_categories.append(FailureCategory(
            category=category,
            failed_metrics=list(set(metrics)),  # Remove duplicates
            severity=severity,
            description=description
        ))

    # Determine primary root cause (category with most failures or highest severity)
    primary_root_cause = self._determine_primary_root_cause(failure_categories)

    # Generate recommendations
    recommendations = self._generate_recommendations(
        failure_categories,
        primary_root_cause,
        evaluation_result
    )

    # Collect details
    details = {
        "overall_score": evaluation_result.overall_score,
        "failed_dimensions": list(failed_metrics_by_dimension.keys()),
        "total_failed_metrics": sum(len(m) for m in failed_metrics_by_category.values()),
        "failures": evaluation_result.failures,
        "warnings": evaluation_result.warnings
    }

    report = RootCauseReport(
        evaluation_id=str(evaluation_result.evaluation_id),
        trace_id=str(evaluation_result.trace_id),
        failure_categories=failure_categories,
        primary_root_cause=primary_root_cause,
        recommendations=recommendations,
        details=details
    )

    logger.info(
        f"Root cause analysis complete for {evaluation_result.evaluation_id}: "
        f"primary cause = {primary_root_cause}, "
        f"{len(failure_categories)} categories identified"
    )

    return report

identify_failure_patterns(evaluation_results)

Identify patterns across multiple evaluation failures.

Parameters:

Name	Type	Description	Default
evaluation_results	List[EvaluationResult]	List of EvaluationResult instances (passed and failed)	required

Returns:

Type	Description
List[FailurePattern]	List of FailurePattern instances, sorted by frequency (descending)

Example

analyzer = RootCauseAnalyzer() patterns = analyzer.identify_failure_patterns(evaluation_results) for pattern in patterns: ... print(f"{pattern.pattern_type}: {pattern.frequency} occurrences")

Source code in uaef/analysis/root_cause.py

def identify_failure_patterns(
    self,
    evaluation_results: List[EvaluationResult]
) -> List[FailurePattern]:
    """
    Identify patterns across multiple evaluation failures.

    Args:
        evaluation_results: List of EvaluationResult instances (passed and failed)

    Returns:
        List of FailurePattern instances, sorted by frequency (descending)

    Example:
        >>> analyzer = RootCauseAnalyzer()
        >>> patterns = analyzer.identify_failure_patterns(evaluation_results)
        >>> for pattern in patterns:
        ...     print(f"{pattern.pattern_type}: {pattern.frequency} occurrences")
    """
    if not evaluation_results:
        logger.warning("No evaluation results provided for pattern identification")
        return []

    # Filter to failed evaluations
    failed_evaluations = [r for r in evaluation_results if not r.passed]

    if not failed_evaluations:
        logger.info("No failed evaluations found in provided results")
        return []

    logger.info(
        f"Identifying failure patterns across {len(failed_evaluations)} failed evaluations "
        f"(out of {len(evaluation_results)} total)"
    )

    # Track metrics that fail together
    category_failures: Dict[str, int] = Counter()
    metric_failures: Dict[str, int] = Counter()
    category_to_metrics: Dict[str, List[str]] = defaultdict(list)

    for result in failed_evaluations:
        failed_metrics_in_eval = set()

        for dimension_result in result.dimension_results:
            for metric_score in dimension_result.scored_metrics():
                if metric_score.score < 0.7:
                    metric_name = metric_score.metric_name
                    failed_metrics_in_eval.add(metric_name)
                    metric_failures[metric_name] += 1

                    category = self._get_metric_category(metric_name)
                    category_to_metrics[category].append(metric_name)

        # Count category failures
        categories_in_eval = set(
            self._get_metric_category(m) for m in failed_metrics_in_eval
        )
        for category in categories_in_eval:
            category_failures[category] += 1

    # Generate patterns from categories
    patterns = []
    for category, frequency in category_failures.most_common():
        # Get most common metrics in this category
        metrics_in_category = category_to_metrics[category]
        metric_counter = Counter(metrics_in_category)
        affected_metrics = [m for m, _ in metric_counter.most_common(5)]

        # Determine severity based on frequency
        severity = self._calculate_pattern_severity(frequency, len(failed_evaluations))

        # Generate pattern description
        description = self._generate_pattern_description(
            category,
            frequency,
            len(failed_evaluations),
            affected_metrics
        )

        # Generate recommendations
        recommendations = self._generate_pattern_recommendations(
            category,
            affected_metrics,
            frequency
        )

        pattern = FailurePattern(
            pattern_type=f"{category}_failures",
            frequency=frequency,
            affected_metrics=affected_metrics,
            description=description,
            severity=severity,
            recommendations=recommendations
        )
        patterns.append(pattern)

    logger.info(f"Identified {len(patterns)} failure patterns")

    return patterns

RootCauseReport

Bases: BaseModel

Root cause analysis report for a failed evaluation.

Attributes:

Name	Type	Description
evaluation_id	str	ID of the evaluated trace
trace_id	str	ID of the agent trace
failure_categories	List[FailureCategory]	List of identified failure categories
primary_root_cause	str	Most likely root cause
recommendations	List[str]	List of actionable recommendations
details	Dict[str, Any]	Additional details about the analysis

Source code in uaef/analysis/root_cause.py

class RootCauseReport(BaseModel):
    """
    Root cause analysis report for a failed evaluation.

    Attributes:
        evaluation_id: ID of the evaluated trace
        trace_id: ID of the agent trace
        failure_categories: List of identified failure categories
        primary_root_cause: Most likely root cause
        recommendations: List of actionable recommendations
        details: Additional details about the analysis
    """

    evaluation_id: str = Field(..., description="ID of the evaluated trace")
    trace_id: str = Field(..., description="ID of the agent trace")
    failure_categories: List[FailureCategory] = Field(
        default_factory=list,
        description="List of identified failure categories"
    )
    primary_root_cause: str = Field(..., description="Most likely root cause")
    recommendations: List[str] = Field(
        default_factory=list,
        description="List of actionable recommendations"
    )
    details: Dict[str, Any] = Field(
        default_factory=dict,
        description="Additional details about the analysis"
    )

    class Config:
        """Pydantic configuration."""
        json_schema_extra = {
            "example": {
                "evaluation_id": "550e8400-e29b-41d4-a716-446655440000",
                "trace_id": "660e8400-e29b-41d4-a716-446655440001",
                "failure_categories": [],
                "primary_root_cause": "tool_selection",
                "recommendations": [
                    "Review tool selection logic in agent prompt",
                    "Add examples of correct tool usage"
                ],
                "details": {"overall_score": 0.45, "failed_dimensions": ["tool_calling"]}
            }
        }

Config

Pydantic configuration.

Source code in uaef/analysis/root_cause.py

class Config:
    """Pydantic configuration."""
    json_schema_extra = {
        "example": {
            "evaluation_id": "550e8400-e29b-41d4-a716-446655440000",
            "trace_id": "660e8400-e29b-41d4-a716-446655440001",
            "failure_categories": [],
            "primary_root_cause": "tool_selection",
            "recommendations": [
                "Review tool selection logic in agent prompt",
                "Add examples of correct tool usage"
            ],
            "details": {"overall_score": 0.45, "failed_dimensions": ["tool_calling"]}
        }
    }

uaef.analysis.clustering

Failure clustering for identifying common failure modes.

FailureCluster

Bases: BaseModel

A cluster of similar failures.

Attributes:

Name	Type	Description
cluster_id	int	Unique identifier for this cluster
size	int	Number of failures in this cluster
evaluation_ids	List[str]	List of evaluation IDs in this cluster
common_failures	List[str]	List of failure messages common to this cluster
common_metrics	List[str]	Metrics that commonly failed in this cluster
severity	str	Average severity of failures in this cluster
summary	str	Human-readable summary of the cluster
representative_failure	Optional[str]	Most representative failure in the cluster

Source code in uaef/analysis/clustering.py

class FailureCluster(BaseModel):
    """
    A cluster of similar failures.

    Attributes:
        cluster_id: Unique identifier for this cluster
        size: Number of failures in this cluster
        evaluation_ids: List of evaluation IDs in this cluster
        common_failures: List of failure messages common to this cluster
        common_metrics: Metrics that commonly failed in this cluster
        severity: Average severity of failures in this cluster
        summary: Human-readable summary of the cluster
        representative_failure: Most representative failure in the cluster
    """

    cluster_id: int = Field(..., description="Unique identifier for this cluster")
    size: int = Field(..., description="Number of failures in this cluster")
    evaluation_ids: List[str] = Field(
        default_factory=list,
        description="List of evaluation IDs in this cluster"
    )
    common_failures: List[str] = Field(
        default_factory=list,
        description="List of failure messages common to this cluster"
    )
    common_metrics: List[str] = Field(
        default_factory=list,
        description="Metrics that commonly failed in this cluster"
    )
    severity: str = Field(..., description="Average severity of failures in this cluster")
    summary: str = Field(..., description="Human-readable summary of the cluster")
    representative_failure: Optional[str] = Field(
        None,
        description="Most representative failure in the cluster"
    )

    class Config:
        """Pydantic configuration."""
        json_schema_extra = {
            "example": {
                "cluster_id": 0,
                "size": 15,
                "evaluation_ids": ["eval-1", "eval-2", "eval-3"],
                "common_failures": [
                    "tool_calling dimension score 0.45 below threshold 0.70"
                ],
                "common_metrics": ["tool_accuracy", "tool_sequence_accuracy"],
                "severity": "high",
                "summary": "Tool selection errors: Agent consistently selects wrong tools",
                "representative_failure": "tool_calling dimension score 0.45 below threshold 0.70"
            }
        }

Config

Pydantic configuration.

Source code in uaef/analysis/clustering.py

class Config:
    """Pydantic configuration."""
    json_schema_extra = {
        "example": {
            "cluster_id": 0,
            "size": 15,
            "evaluation_ids": ["eval-1", "eval-2", "eval-3"],
            "common_failures": [
                "tool_calling dimension score 0.45 below threshold 0.70"
            ],
            "common_metrics": ["tool_accuracy", "tool_sequence_accuracy"],
            "severity": "high",
            "summary": "Tool selection errors: Agent consistently selects wrong tools",
            "representative_failure": "tool_calling dimension score 0.45 below threshold 0.70"
        }
    }

FailureClusterer

Clusterer for identifying common failure modes across evaluations.

Provides methods for: - Clustering failures based on similarity - Supporting multiple clustering algorithms (k-means, hierarchical) - Generating cluster summaries - Identifying common failure patterns

Source code in uaef/analysis/clustering.py

class FailureClusterer:
    """
    Clusterer for identifying common failure modes across evaluations.

    Provides methods for:
    - Clustering failures based on similarity
    - Supporting multiple clustering algorithms (k-means, hierarchical)
    - Generating cluster summaries
    - Identifying common failure patterns
    """

    def __init__(
        self,
        min_cluster_size: int = 2,
        similarity_threshold: float = 0.5
    ):
        """
        Initialize the failure clusterer.

        Args:
            min_cluster_size: Minimum number of failures to form a cluster
            similarity_threshold: Threshold for hierarchical clustering (0-1)

        Raises:
            ValueError: If parameters are invalid
            ImportError: If scikit-learn is not installed
        """
        if not SKLEARN_AVAILABLE:
            raise ImportError(
                "scikit-learn is required for failure clustering. "
                "Install it with: pip install scikit-learn>=1.3.0"
            )

        if min_cluster_size < 1:
            raise ValueError("min_cluster_size must be at least 1")
        if not 0.0 <= similarity_threshold <= 1.0:
            raise ValueError("similarity_threshold must be between 0 and 1")

        self.min_cluster_size = min_cluster_size
        self.similarity_threshold = similarity_threshold
        logger.info(
            f"Initialized FailureClusterer with min_cluster_size={min_cluster_size}, "
            f"similarity_threshold={similarity_threshold}"
        )

    def cluster_failures(
        self,
        evaluation_results: List[EvaluationResult],
        method: Literal["kmeans", "hierarchical"] = "hierarchical",
        n_clusters: Optional[int] = None
    ) -> List[FailureCluster]:
        """
        Cluster failures based on similarity.

        Args:
            evaluation_results: List of evaluation results to cluster
            method: Clustering method ("kmeans" or "hierarchical")
            n_clusters: Number of clusters for k-means (auto-detected if None)

        Returns:
            List of failure clusters

        Raises:
            ValueError: If inputs are invalid or clustering fails
        """
        try:
            # Filter to only failed evaluations
            failed_results = [r for r in evaluation_results if not r.passed]

            if not failed_results:
                logger.info("No failed evaluations to cluster")
                return []

            if len(failed_results) < self.min_cluster_size:
                logger.info(
                    f"Only {len(failed_results)} failures, less than min_cluster_size "
                    f"{self.min_cluster_size}. Creating single cluster."
                )
                return [self._create_single_cluster(failed_results)]

            logger.info(
                f"Clustering {len(failed_results)} failures using {method} method"
            )

            # Create feature vectors for clustering
            feature_vectors, feature_names = self._create_feature_vectors(failed_results)

            # Perform clustering
            if method == "kmeans":
                labels = self._cluster_kmeans(feature_vectors, n_clusters)
            elif method == "hierarchical":
                labels = self._cluster_hierarchical(feature_vectors)
            else:
                raise ValueError(f"Unknown clustering method: {method}")

            # Generate cluster summaries
            clusters = self._generate_clusters(
                failed_results,
                labels,
                feature_names
            )

            logger.info(f"Created {len(clusters)} failure clusters")
            return clusters

        except Exception as e:
            logger.error(f"Error clustering failures: {e}", exc_info=True)
            raise ValueError(f"Failed to cluster failures: {e}") from e

    def _create_feature_vectors(
        self,
        results: List[EvaluationResult]
    ) -> tuple[np.ndarray, List[str]]:
        """
        Create feature vectors for clustering.

        Features include:
        - Which dimensions failed
        - Which specific metrics failed
        - Severity of failures (based on scores)

        Args:
            results: List of evaluation results

        Returns:
            Tuple of (feature_vectors, feature_names)
        """
        # Collect all unique dimension names and metric names
        all_dimensions = set()
        all_metrics = set()

        for result in results:
            for dim_result in result.dimension_results:
                all_dimensions.add(dim_result.dimension_name)
                for metric_score in dim_result.metric_scores:
                    all_metrics.add(metric_score.metric_name)

        dimension_list = sorted(all_dimensions)
        metric_list = sorted(all_metrics)

        # Create feature names
        feature_names = (
            [f"dim_{dim}" for dim in dimension_list] +
            [f"metric_{metric}" for metric in metric_list] +
            ["overall_score", "num_failures"]
        )

        # Create feature vectors
        vectors = []
        for result in results:
            vector = []

            # Dimension features (1 if dimension failed, 0 otherwise)
            dim_scores = {
                dr.dimension_name: dr.aggregate_score
                for dr in result.dimension_results
                if dr.aggregate_score is not None
            }
            for dim in dimension_list:
                # Consider failed if score < 0.7 (common threshold)
                score = dim_scores.get(dim, 1.0)
                vector.append(1.0 if score < 0.7 else 0.0)

            # Metric features (1 if metric failed, 0 otherwise). A metric whose
            # calculation raised counts as failed; a not-applicable one (score
            # None, no error) is treated like an absent metric.
            metric_scores = {}
            for dim_result in result.dimension_results:
                for metric_score in dim_result.metric_scores:
                    if metric_score.score is not None:
                        metric_scores[metric_score.metric_name] = metric_score.score
                    elif (metric_score.metadata or {}).get("error"):
                        metric_scores[metric_score.metric_name] = 0.0

            for metric in metric_list:
                score = metric_scores.get(metric, 1.0)
                vector.append(1.0 if score < 0.7 else 0.0)

            # Overall score (inverted so lower scores = higher feature value)
            vector.append(1.0 - result.overall_score)

            # Number of failures (normalized)
            vector.append(min(len(result.failures) / 10.0, 1.0))

            vectors.append(vector)

        return np.array(vectors), feature_names

    def _cluster_kmeans(
        self,
        feature_vectors: np.ndarray,
        n_clusters: Optional[int] = None
    ) -> np.ndarray:
        """
        Perform k-means clustering.

        Args:
            feature_vectors: Feature vectors for clustering
            n_clusters: Number of clusters (auto-detected if None)

        Returns:
            Cluster labels for each sample
        """
        n_samples = len(feature_vectors)

        # Auto-detect number of clusters if not specified
        if n_clusters is None:
            # Use elbow method heuristic: sqrt(n/2)
            n_clusters = max(2, min(int(np.sqrt(n_samples / 2)), n_samples // 2))
            logger.info(f"Auto-detected n_clusters={n_clusters} for {n_samples} samples")

        # Ensure n_clusters is valid
        n_clusters = max(1, min(n_clusters, n_samples))

        if n_clusters == 1:
            return np.zeros(n_samples, dtype=int)

        # Perform k-means clustering
        kmeans = KMeans(
            n_clusters=n_clusters,
            random_state=42,
            n_init=10
        )
        labels = kmeans.fit_predict(feature_vectors)

        return labels

    def _cluster_hierarchical(
        self,
        feature_vectors: np.ndarray
    ) -> np.ndarray:
        """
        Perform hierarchical clustering.

        Args:
            feature_vectors: Feature vectors for clustering

        Returns:
            Cluster labels for each sample
        """
        n_samples = len(feature_vectors)

        if n_samples == 1:
            return np.zeros(1, dtype=int)

        # Use distance threshold based on similarity threshold
        # distance = 1 - similarity, so distance_threshold = 1 - similarity_threshold
        distance_threshold = 1.0 - self.similarity_threshold

        # Perform hierarchical clustering
        clustering = AgglomerativeClustering(
            n_clusters=None,
            distance_threshold=distance_threshold,
            linkage="average"
        )
        labels = clustering.fit_predict(feature_vectors)

        return labels

    def _generate_clusters(
        self,
        results: List[EvaluationResult],
        labels: np.ndarray,
        feature_names: List[str]
    ) -> List[FailureCluster]:
        """
        Generate cluster summaries from clustering results.

        Args:
            results: List of evaluation results
            labels: Cluster labels for each result
            feature_names: Names of features used for clustering

        Returns:
            List of failure clusters
        """
        clusters = []
        unique_labels = np.unique(labels)

        for cluster_id in unique_labels:
            # Get results in this cluster
            cluster_mask = labels == cluster_id
            cluster_results = [r for i, r in enumerate(results) if cluster_mask[i]]

            # Skip if cluster is too small
            if len(cluster_results) < self.min_cluster_size:
                logger.debug(
                    f"Skipping cluster {cluster_id} with only "
                    f"{len(cluster_results)} members"
                )
                continue

            # Generate cluster summary
            cluster = self._summarize_cluster(cluster_id, cluster_results)
            clusters.append(cluster)

        return clusters

    def _summarize_cluster(
        self,
        cluster_id: int,
        results: List[EvaluationResult]
    ) -> FailureCluster:
        """
        Generate a summary for a cluster of failures.

        Args:
            cluster_id: ID of the cluster
            results: Evaluation results in this cluster

        Returns:
            FailureCluster summary
        """
        # Collect common failures
        failure_counts = defaultdict(int)
        for result in results:
            for failure in result.failures:
                failure_counts[failure] += 1

        # Get failures that appear in at least 50% of cluster members
        threshold = len(results) * 0.5
        common_failures = [
            failure for failure, count in failure_counts.items()
            if count >= threshold
        ]

        # Collect common metrics
        metric_counts = defaultdict(int)
        for result in results:
            for dim_result in result.dimension_results:
                for metric_score in dim_result.scored_metrics():
                    if metric_score.score < 0.7:  # Failed metric
                        metric_counts[metric_score.metric_name] += 1

        # Get metrics that failed in at least 50% of cluster members
        common_metrics = [
            metric for metric, count in metric_counts.items()
            if count >= threshold
        ]

        # Calculate average severity
        avg_score = np.mean([r.overall_score for r in results])
        if avg_score < 0.4:
            severity = "critical"
        elif avg_score < 0.6:
            severity = "high"
        elif avg_score < 0.8:
            severity = "medium"
        else:
            severity = "low"

        # Generate summary
        summary = self._generate_cluster_summary(
            common_failures,
            common_metrics,
            severity
        )

        # Get representative failure (most common)
        representative_failure = None
        if common_failures:
            representative_failure = max(
                failure_counts.items(),
                key=lambda x: x[1]
            )[0]

        return FailureCluster(
            cluster_id=cluster_id,
            size=len(results),
            evaluation_ids=[str(r.evaluation_id) for r in results],
            common_failures=common_failures,
            common_metrics=common_metrics,
            severity=severity,
            summary=summary,
            representative_failure=representative_failure
        )

    def _generate_cluster_summary(
        self,
        common_failures: List[str],
        common_metrics: List[str],
        severity: str
    ) -> str:
        """
        Generate a human-readable summary for a cluster.

        Args:
            common_failures: Common failure messages
            common_metrics: Common failed metrics
            severity: Severity level

        Returns:
            Human-readable summary
        """
        if not common_metrics and not common_failures:
            return f"Cluster of {severity} severity failures with varied issues"

        # Categorize metrics
        tool_metrics = [m for m in common_metrics if "tool" in m.lower()]
        response_metrics = [m for m in common_metrics if any(
            x in m.lower() for x in ["relevance", "completeness", "hallucination"]
        )]
        safety_metrics = [m for m in common_metrics if any(
            x in m.lower() for x in ["safety", "bias", "toxicity"]
        )]
        performance_metrics = [m for m in common_metrics if any(
            x in m.lower() for x in ["latency", "token", "cost"]
        )]

        # Generate summary based on dominant category
        if tool_metrics:
            return f"Tool calling issues: Failures in {', '.join(tool_metrics[:3])}"
        elif response_metrics:
            return f"Response quality issues: Failures in {', '.join(response_metrics[:3])}"
        elif safety_metrics:
            return f"Safety concerns: Failures in {', '.join(safety_metrics[:3])}"
        elif performance_metrics:
            return f"Performance issues: Failures in {', '.join(performance_metrics[:3])}"
        else:
            return f"Mixed failures: {', '.join(common_metrics[:3])}"

    def _create_single_cluster(
        self,
        results: List[EvaluationResult]
    ) -> FailureCluster:
        """
        Create a single cluster containing all results.

        Args:
            results: All evaluation results

        Returns:
            Single failure cluster
        """
        return self._summarize_cluster(0, results)

cluster_failures(evaluation_results, method='hierarchical', n_clusters=None)

Cluster failures based on similarity.

Parameters:

Name	Type	Description	Default
evaluation_results	List[EvaluationResult]	List of evaluation results to cluster	required
method	Literal['kmeans', 'hierarchical']	Clustering method ("kmeans" or "hierarchical")	'hierarchical'
n_clusters	Optional[int]	Number of clusters for k-means (auto-detected if None)	None

Returns:

Type	Description
List[FailureCluster]	List of failure clusters

Raises:

Type	Description
ValueError	If inputs are invalid or clustering fails

Source code in uaef/analysis/clustering.py

def cluster_failures(
    self,
    evaluation_results: List[EvaluationResult],
    method: Literal["kmeans", "hierarchical"] = "hierarchical",
    n_clusters: Optional[int] = None
) -> List[FailureCluster]:
    """
    Cluster failures based on similarity.

    Args:
        evaluation_results: List of evaluation results to cluster
        method: Clustering method ("kmeans" or "hierarchical")
        n_clusters: Number of clusters for k-means (auto-detected if None)

    Returns:
        List of failure clusters

    Raises:
        ValueError: If inputs are invalid or clustering fails
    """
    try:
        # Filter to only failed evaluations
        failed_results = [r for r in evaluation_results if not r.passed]

        if not failed_results:
            logger.info("No failed evaluations to cluster")
            return []

        if len(failed_results) < self.min_cluster_size:
            logger.info(
                f"Only {len(failed_results)} failures, less than min_cluster_size "
                f"{self.min_cluster_size}. Creating single cluster."
            )
            return [self._create_single_cluster(failed_results)]

        logger.info(
            f"Clustering {len(failed_results)} failures using {method} method"
        )

        # Create feature vectors for clustering
        feature_vectors, feature_names = self._create_feature_vectors(failed_results)

        # Perform clustering
        if method == "kmeans":
            labels = self._cluster_kmeans(feature_vectors, n_clusters)
        elif method == "hierarchical":
            labels = self._cluster_hierarchical(feature_vectors)
        else:
            raise ValueError(f"Unknown clustering method: {method}")

        # Generate cluster summaries
        clusters = self._generate_clusters(
            failed_results,
            labels,
            feature_names
        )

        logger.info(f"Created {len(clusters)} failure clusters")
        return clusters

    except Exception as e:
        logger.error(f"Error clustering failures: {e}", exc_info=True)
        raise ValueError(f"Failed to cluster failures: {e}") from e

uaef.analysis.trends

Trend analyzer for tracking metric changes over time.

Anomaly

Bases: BaseModel

Detected anomaly in metric values.

Attributes:

Name	Type	Description
metric_name	str	Name of the metric with anomaly
value	float	Anomalous value
timestamp	datetime	When the anomaly occurred
expected_range	Tuple[float, float]	Expected range (min, max) based on historical data
z_score	float	Z-score of the anomalous value
severity	Literal['low', 'medium', 'high', 'critical']	Severity level (low, medium, high, critical)
description	str	Human-readable description of the anomaly

Source code in uaef/analysis/trends.py

class Anomaly(BaseModel):
    """
    Detected anomaly in metric values.

    Attributes:
        metric_name: Name of the metric with anomaly
        value: Anomalous value
        timestamp: When the anomaly occurred
        expected_range: Expected range (min, max) based on historical data
        z_score: Z-score of the anomalous value
        severity: Severity level (low, medium, high, critical)
        description: Human-readable description of the anomaly
    """

    metric_name: str = Field(..., description="Name of the metric with anomaly")
    value: float = Field(..., description="Anomalous value")
    timestamp: datetime = Field(..., description="When the anomaly occurred")
    expected_range: Tuple[float, float] = Field(
        ...,
        description="Expected range (min, max) based on historical data"
    )
    z_score: float = Field(..., description="Z-score of the anomalous value")
    severity: Literal["low", "medium", "high", "critical"] = Field(
        ...,
        description="Severity level"
    )
    description: str = Field(
        ...,
        description="Human-readable description of the anomaly"
    )

    class Config:
        """Pydantic configuration."""
        json_schema_extra = {
            "example": {
                "metric_name": "tool_accuracy",
                "value": 0.45,
                "timestamp": "2024-01-15T10:00:00Z",
                "expected_range": (0.75, 0.90),
                "z_score": -3.5,
                "severity": "critical",
                "description": "tool_accuracy dropped to 0.45, significantly below expected range [0.75, 0.90]"
            }
        }

Config

Pydantic configuration.

Source code in uaef/analysis/trends.py

class Config:
    """Pydantic configuration."""
    json_schema_extra = {
        "example": {
            "metric_name": "tool_accuracy",
            "value": 0.45,
            "timestamp": "2024-01-15T10:00:00Z",
            "expected_range": (0.75, 0.90),
            "z_score": -3.5,
            "severity": "critical",
            "description": "tool_accuracy dropped to 0.45, significantly below expected range [0.75, 0.90]"
        }
    }

MetricTrend

Bases: BaseModel

Trend information for a specific metric.

Attributes:

Name	Type	Description
metric_name	str	Name of the metric
values	List[float]	List of metric values over time
timestamps	List[datetime]	List of timestamps corresponding to values
trend_direction	Literal['improving', 'declining', 'stable']	Overall trend direction (improving, declining, stable)
slope	float	Linear regression slope (positive = improving)
mean	float	Mean value across all data points
std_dev	float	Standard deviation
min_value	float	Minimum value
max_value	float	Maximum value
change_percentage	float	Percentage change from first to last value

Source code in uaef/analysis/trends.py

class MetricTrend(BaseModel):
    """
    Trend information for a specific metric.

    Attributes:
        metric_name: Name of the metric
        values: List of metric values over time
        timestamps: List of timestamps corresponding to values
        trend_direction: Overall trend direction (improving, declining, stable)
        slope: Linear regression slope (positive = improving)
        mean: Mean value across all data points
        std_dev: Standard deviation
        min_value: Minimum value
        max_value: Maximum value
        change_percentage: Percentage change from first to last value
    """

    metric_name: str = Field(..., description="Name of the metric")
    values: List[float] = Field(
        default_factory=list,
        description="List of metric values over time"
    )
    timestamps: List[datetime] = Field(
        default_factory=list,
        description="List of timestamps corresponding to values"
    )
    trend_direction: Literal["improving", "declining", "stable"] = Field(
        ...,
        description="Overall trend direction"
    )
    slope: float = Field(..., description="Linear regression slope")
    mean: float = Field(..., description="Mean value across all data points")
    std_dev: float = Field(..., description="Standard deviation")
    min_value: float = Field(..., description="Minimum value")
    max_value: float = Field(..., description="Maximum value")
    change_percentage: float = Field(
        ...,
        description="Percentage change from first to last value"
    )

    class Config:
        """Pydantic configuration."""
        json_schema_extra = {
            "example": {
                "metric_name": "tool_accuracy",
                "values": [0.75, 0.78, 0.82, 0.85],
                "timestamps": ["2024-01-01T10:00:00Z", "2024-01-02T10:00:00Z"],
                "trend_direction": "improving",
                "slope": 0.033,
                "mean": 0.80,
                "std_dev": 0.042,
                "min_value": 0.75,
                "max_value": 0.85,
                "change_percentage": 13.33
            }
        }

Config

Pydantic configuration.

Source code in uaef/analysis/trends.py

class Config:
    """Pydantic configuration."""
    json_schema_extra = {
        "example": {
            "metric_name": "tool_accuracy",
            "values": [0.75, 0.78, 0.82, 0.85],
            "timestamps": ["2024-01-01T10:00:00Z", "2024-01-02T10:00:00Z"],
            "trend_direction": "improving",
            "slope": 0.033,
            "mean": 0.80,
            "std_dev": 0.042,
            "min_value": 0.75,
            "max_value": 0.85,
            "change_percentage": 13.33
        }
    }

TrendAnalyzer

Analyzer for tracking metric trends over time.

Provides methods for: - Analyzing metric trends across experiment runs - Detecting anomalies using statistical methods - Tracking metric changes over time - Generating trend reports with visualizations

Source code in uaef/analysis/trends.py

class TrendAnalyzer:
    """
    Analyzer for tracking metric trends over time.

    Provides methods for:
    - Analyzing metric trends across experiment runs
    - Detecting anomalies using statistical methods
    - Tracking metric changes over time
    - Generating trend reports with visualizations
    """

    def __init__(
        self,
        min_data_points: int = 3,
        anomaly_threshold: float = 2.5,
        stable_threshold: float = 0.01
    ):
        """
        Initialize the trend analyzer.

        Args:
            min_data_points: Minimum number of data points required for trend analysis
            anomaly_threshold: Z-score threshold for anomaly detection (default: 2.5)
            stable_threshold: Threshold for considering trend stable (default: 0.01)

        Raises:
            ValueError: If parameters are invalid
        """
        if min_data_points < 2:
            raise ValueError("min_data_points must be at least 2")
        if anomaly_threshold <= 0:
            raise ValueError("anomaly_threshold must be positive")
        if stable_threshold <= 0:
            raise ValueError("stable_threshold must be positive")

        self.min_data_points = min_data_points
        self.anomaly_threshold = anomaly_threshold
        self.stable_threshold = stable_threshold

        logger.info(
            f"TrendAnalyzer initialized with min_data_points={min_data_points}, "
            f"anomaly_threshold={anomaly_threshold}, stable_threshold={stable_threshold}"
        )

    def analyze_metric_trends(
        self,
        evaluation_results: List[EvaluationResult],
        metric_name: Optional[str] = None
    ) -> TrendReport:
        """
        Analyze metric trends over time across evaluation results.

        Args:
            evaluation_results: List of evaluation results ordered by time
            metric_name: Specific metric to analyze (None = analyze all metrics)

        Returns:
            TrendReport with trend analysis for requested metrics

        Raises:
            ValueError: If insufficient data or invalid inputs

        Example:
            >>> analyzer = TrendAnalyzer()
            >>> report = analyzer.analyze_metric_trends(results, "tool_accuracy")
            >>> print(f"Trend: {report.metric_trends['tool_accuracy'].trend_direction}")
            >>> print(f"Change: {report.metric_trends['tool_accuracy'].change_percentage:.1f}%")
        """
        if not evaluation_results:
            raise ValueError("No evaluation results provided")

        if len(evaluation_results) < self.min_data_points:
            raise ValueError(
                f"Insufficient data points: {len(evaluation_results)} provided, "
                f"{self.min_data_points} required"
            )

        logger.info(
            f"Analyzing trends for {len(evaluation_results)} evaluation results"
            + (f" (metric: {metric_name})" if metric_name else " (all metrics)")
        )

        # Sort by timestamp
        sorted_results = sorted(evaluation_results, key=lambda r: r.timestamp)

        # Extract metric values over time
        metric_data = self._extract_metric_data(sorted_results, metric_name)

        if not metric_data:
            raise ValueError(
                f"No data found for metric: {metric_name}" if metric_name
                else "No metric data found in evaluation results"
            )

        # Analyze trends for each metric
        metric_trends = {}
        for metric, data in metric_data.items():
            if len(data["values"]) >= self.min_data_points:
                trend = self._analyze_single_metric_trend(
                    metric,
                    data["values"],
                    data["timestamps"]
                )
                metric_trends[metric] = trend
            else:
                logger.warning(
                    f"Skipping metric {metric}: only {len(data['values'])} data points"
                )

        # Detect anomalies
        anomalies = self._detect_all_anomalies(metric_data)

        # Generate summary
        summary = self._generate_trend_summary(metric_trends, anomalies)

        # Determine time range
        all_timestamps = [r.timestamp for r in sorted_results]
        time_range = (min(all_timestamps), max(all_timestamps))

        report = TrendReport(
            metric_trends=metric_trends,
            anomalies=anomalies,
            summary=summary,
            time_range=time_range,
            total_data_points=len(sorted_results),
            metrics_analyzed=len(metric_trends)
        )

        logger.info(
            f"Trend analysis complete: {len(metric_trends)} metrics analyzed, "
            f"{len(anomalies)} anomalies detected"
        )

        return report

    def detect_anomalies(
        self,
        evaluation_results: List[EvaluationResult],
        metric_name: Optional[str] = None,
        method: Literal["zscore", "iqr"] = "zscore"
    ) -> List[Anomaly]:
        """
        Detect statistical anomalies in metric values.

        Args:
            evaluation_results: List of evaluation results
            metric_name: Specific metric to check (None = check all metrics)
            method: Anomaly detection method ("zscore" or "iqr")

        Returns:
            List of detected anomalies

        Raises:
            ValueError: If insufficient data or invalid inputs

        Example:
            >>> analyzer = TrendAnalyzer()
            >>> anomalies = analyzer.detect_anomalies(results, method="zscore")
            >>> for anomaly in anomalies:
            ...     print(f"{anomaly.metric_name}: {anomaly.description}")
        """
        if not evaluation_results:
            raise ValueError("No evaluation results provided")

        if len(evaluation_results) < self.min_data_points:
            raise ValueError(
                f"Insufficient data points: {len(evaluation_results)} provided, "
                f"{self.min_data_points} required"
            )

        logger.info(
            f"Detecting anomalies in {len(evaluation_results)} evaluation results "
            f"using {method} method"
            + (f" (metric: {metric_name})" if metric_name else " (all metrics)")
        )

        # Extract metric data
        metric_data = self._extract_metric_data(evaluation_results, metric_name)

        if not metric_data:
            raise ValueError(
                f"No data found for metric: {metric_name}" if metric_name
                else "No metric data found in evaluation results"
            )

        # Detect anomalies based on method
        if method == "zscore":
            anomalies = self._detect_anomalies_zscore(metric_data)
        elif method == "iqr":
            anomalies = self._detect_anomalies_iqr(metric_data)
        else:
            raise ValueError(f"Unknown anomaly detection method: {method}")

        logger.info(f"Detected {len(anomalies)} anomalies using {method} method")

        return anomalies

    def _extract_metric_data(
        self,
        evaluation_results: List[EvaluationResult],
        metric_name: Optional[str] = None
    ) -> Dict[str, Dict[str, List]]:
        """
        Extract metric values and timestamps from evaluation results.

        Args:
            evaluation_results: List of evaluation results
            metric_name: Specific metric to extract (None = extract all)

        Returns:
            Dictionary mapping metric names to {"values": [...], "timestamps": [...]}
        """
        metric_data: Dict[str, Dict[str, List]] = defaultdict(
            lambda: {"values": [], "timestamps": []}
        )

        for result in evaluation_results:
            for dim_result in result.dimension_results:
                for metric_score in dim_result.scored_metrics():
                    # Filter by metric name if specified
                    if metric_name and metric_score.metric_name != metric_name:
                        continue

                    metric_data[metric_score.metric_name]["values"].append(metric_score.score)
                    metric_data[metric_score.metric_name]["timestamps"].append(result.timestamp)

        return dict(metric_data)

    def _analyze_single_metric_trend(
        self,
        metric_name: str,
        values: List[float],
        timestamps: List[datetime]
    ) -> MetricTrend:
        """
        Analyze trend for a single metric.

        Args:
            metric_name: Name of the metric
            values: List of metric values
            timestamps: List of timestamps

        Returns:
            MetricTrend with analysis results
        """
        values_array = np.array(values)

        # Calculate statistics
        mean = float(np.mean(values_array))
        std_dev = float(np.std(values_array))
        min_value = float(np.min(values_array))
        max_value = float(np.max(values_array))

        # Calculate linear regression slope
        x = np.arange(len(values))
        slope = float(np.polyfit(x, values_array, 1)[0])

        # Determine trend direction
        if abs(slope) < self.stable_threshold:
            trend_direction = "stable"
        elif slope > 0:
            trend_direction = "improving"
        else:
            trend_direction = "declining"

        # Calculate percentage change
        if values[0] != 0:
            change_percentage = ((values[-1] - values[0]) / values[0]) * 100
        else:
            change_percentage = 0.0

        return MetricTrend(
            metric_name=metric_name,
            values=values,
            timestamps=timestamps,
            trend_direction=trend_direction,
            slope=slope,
            mean=mean,
            std_dev=std_dev,
            min_value=min_value,
            max_value=max_value,
            change_percentage=change_percentage
        )

    def _detect_all_anomalies(
        self,
        metric_data: Dict[str, Dict[str, List]]
    ) -> List[Anomaly]:
        """
        Detect anomalies across all metrics using z-score method.

        Args:
            metric_data: Dictionary of metric data

        Returns:
            List of detected anomalies
        """
        return self._detect_anomalies_zscore(metric_data)

    def _detect_anomalies_zscore(
        self,
        metric_data: Dict[str, Dict[str, List]]
    ) -> List[Anomaly]:
        """
        Detect anomalies using z-score method.

        Args:
            metric_data: Dictionary of metric data

        Returns:
            List of detected anomalies
        """
        anomalies = []

        for metric_name, data in metric_data.items():
            values = np.array(data["values"])
            timestamps = data["timestamps"]

            if len(values) < self.min_data_points:
                continue

            # Calculate mean and standard deviation
            mean = np.mean(values)
            std_dev = np.std(values)

            # Skip if std_dev is too small (no variation)
            if std_dev < 1e-6:
                continue

            # Calculate z-scores
            z_scores = (values - mean) / std_dev

            # Detect anomalies
            for i, (value, timestamp, z_score) in enumerate(
                zip(values, timestamps, z_scores)
            ):
                if abs(z_score) > self.anomaly_threshold:
                    # Calculate expected range (mean ± 2 std_dev)
                    expected_min = max(0.0, mean - 2 * std_dev)
                    expected_max = min(1.0, mean + 2 * std_dev)

                    # Determine severity based on z-score magnitude
                    severity = self._calculate_anomaly_severity(abs(z_score))

                    # Generate description
                    description = self._generate_anomaly_description(
                        metric_name,
                        value,
                        (expected_min, expected_max),
                        z_score
                    )

                    anomaly = Anomaly(
                        metric_name=metric_name,
                        value=float(value),
                        timestamp=timestamp,
                        expected_range=(float(expected_min), float(expected_max)),
                        z_score=float(z_score),
                        severity=severity,
                        description=description
                    )
                    anomalies.append(anomaly)

        # Sort by severity and z-score magnitude
        severity_order = {"critical": 0, "high": 1, "medium": 2, "low": 3}
        anomalies.sort(
            key=lambda a: (severity_order[a.severity], -abs(a.z_score))
        )

        return anomalies

    def _detect_anomalies_iqr(
        self,
        metric_data: Dict[str, Dict[str, List]]
    ) -> List[Anomaly]:
        """
        Detect anomalies using Interquartile Range (IQR) method.

        Args:
            metric_data: Dictionary of metric data

        Returns:
            List of detected anomalies
        """
        anomalies = []

        for metric_name, data in metric_data.items():
            values = np.array(data["values"])
            timestamps = data["timestamps"]

            if len(values) < self.min_data_points:
                continue

            # Calculate quartiles and IQR
            q1 = np.percentile(values, 25)
            q3 = np.percentile(values, 75)
            iqr = q3 - q1

            # Skip if IQR is too small (no variation)
            if iqr < 1e-6:
                continue

            # Calculate bounds (1.5 * IQR is standard for outliers)
            lower_bound = max(0.0, q1 - 1.5 * iqr)
            upper_bound = min(1.0, q3 + 1.5 * iqr)

            # Detect anomalies
            for value, timestamp in zip(values, timestamps):
                if value < lower_bound or value > upper_bound:
                    # Calculate pseudo z-score for severity
                    mean = np.mean(values)
                    std_dev = np.std(values)
                    z_score = (value - mean) / std_dev if std_dev > 1e-6 else 0.0

                    # Determine severity
                    severity = self._calculate_anomaly_severity(abs(z_score))

                    # Generate description
                    description = self._generate_anomaly_description(
                        metric_name,
                        value,
                        (lower_bound, upper_bound),
                        z_score
                    )

                    anomaly = Anomaly(
                        metric_name=metric_name,
                        value=float(value),
                        timestamp=timestamp,
                        expected_range=(float(lower_bound), float(upper_bound)),
                        z_score=float(z_score),
                        severity=severity,
                        description=description
                    )
                    anomalies.append(anomaly)

        # Sort by severity
        severity_order = {"critical": 0, "high": 1, "medium": 2, "low": 3}
        anomalies.sort(
            key=lambda a: (severity_order[a.severity], -abs(a.z_score))
        )

        return anomalies

    def _calculate_anomaly_severity(self, abs_z_score: float) -> str:
        """
        Calculate anomaly severity based on z-score magnitude.

        Args:
            abs_z_score: Absolute value of z-score

        Returns:
            Severity level: "low", "medium", "high", or "critical"
        """
        if abs_z_score >= 4.0:
            return "critical"
        elif abs_z_score >= 3.5:
            return "high"
        elif abs_z_score >= 3.0:
            return "medium"
        else:
            return "low"

    def _generate_anomaly_description(
        self,
        metric_name: str,
        value: float,
        expected_range: Tuple[float, float],
        z_score: float
    ) -> str:
        """
        Generate human-readable description for an anomaly.

        Args:
            metric_name: Name of the metric
            value: Anomalous value
            expected_range: Expected range (min, max)
            z_score: Z-score of the value

        Returns:
            Human-readable description
        """
        direction = "above" if z_score > 0 else "below"

        return (
            f"{metric_name} value {value:.3f} is significantly {direction} "
            f"expected range [{expected_range[0]:.3f}, {expected_range[1]:.3f}] "
            f"(z-score: {z_score:.2f})"
        )

    def _generate_trend_summary(
        self,
        metric_trends: Dict[str, MetricTrend],
        anomalies: List[Anomaly]
    ) -> str:
        """
        Generate high-level summary of trends.

        Args:
            metric_trends: Dictionary of metric trends
            anomalies: List of detected anomalies

        Returns:
            Human-readable summary
        """
        if not metric_trends:
            return "No trends to analyze"

        # Count trend directions
        improving = sum(1 for t in metric_trends.values() if t.trend_direction == "improving")
        declining = sum(1 for t in metric_trends.values() if t.trend_direction == "declining")
        stable = sum(1 for t in metric_trends.values() if t.trend_direction == "stable")

        # Count anomaly severities
        critical_anomalies = sum(1 for a in anomalies if a.severity == "critical")
        high_anomalies = sum(1 for a in anomalies if a.severity == "high")

        # Build summary
        parts = []

        if improving > declining:
            parts.append(f"Overall improving trend across {improving} metrics")
        elif declining > improving:
            parts.append(f"Overall declining trend across {declining} metrics")
        else:
            parts.append(f"Mixed trends: {improving} improving, {declining} declining, {stable} stable")

        if critical_anomalies > 0:
            parts.append(f"{critical_anomalies} critical anomalies detected")
        elif high_anomalies > 0:
            parts.append(f"{high_anomalies} high-severity anomalies detected")
        elif len(anomalies) > 0:
            parts.append(f"{len(anomalies)} anomalies detected")
        else:
            parts.append("no anomalies detected")

        return "; ".join(parts)

analyze_metric_trends(evaluation_results, metric_name=None)

Analyze metric trends over time across evaluation results.

Parameters:

Name	Type	Description	Default
evaluation_results	List[EvaluationResult]	List of evaluation results ordered by time	required
metric_name	Optional[str]	Specific metric to analyze (None = analyze all metrics)	None

Returns:

Type	Description
TrendReport	TrendReport with trend analysis for requested metrics

Raises:

Type	Description
ValueError	If insufficient data or invalid inputs

Example

analyzer = TrendAnalyzer() report = analyzer.analyze_metric_trends(results, "tool_accuracy") print(f"Trend: {report.metric_trends['tool_accuracy'].trend_direction}") print(f"Change: {report.metric_trends['tool_accuracy'].change_percentage:.1f}%")

Source code in uaef/analysis/trends.py

def analyze_metric_trends(
    self,
    evaluation_results: List[EvaluationResult],
    metric_name: Optional[str] = None
) -> TrendReport:
    """
    Analyze metric trends over time across evaluation results.

    Args:
        evaluation_results: List of evaluation results ordered by time
        metric_name: Specific metric to analyze (None = analyze all metrics)

    Returns:
        TrendReport with trend analysis for requested metrics

    Raises:
        ValueError: If insufficient data or invalid inputs

    Example:
        >>> analyzer = TrendAnalyzer()
        >>> report = analyzer.analyze_metric_trends(results, "tool_accuracy")
        >>> print(f"Trend: {report.metric_trends['tool_accuracy'].trend_direction}")
        >>> print(f"Change: {report.metric_trends['tool_accuracy'].change_percentage:.1f}%")
    """
    if not evaluation_results:
        raise ValueError("No evaluation results provided")

    if len(evaluation_results) < self.min_data_points:
        raise ValueError(
            f"Insufficient data points: {len(evaluation_results)} provided, "
            f"{self.min_data_points} required"
        )

    logger.info(
        f"Analyzing trends for {len(evaluation_results)} evaluation results"
        + (f" (metric: {metric_name})" if metric_name else " (all metrics)")
    )

    # Sort by timestamp
    sorted_results = sorted(evaluation_results, key=lambda r: r.timestamp)

    # Extract metric values over time
    metric_data = self._extract_metric_data(sorted_results, metric_name)

    if not metric_data:
        raise ValueError(
            f"No data found for metric: {metric_name}" if metric_name
            else "No metric data found in evaluation results"
        )

    # Analyze trends for each metric
    metric_trends = {}
    for metric, data in metric_data.items():
        if len(data["values"]) >= self.min_data_points:
            trend = self._analyze_single_metric_trend(
                metric,
                data["values"],
                data["timestamps"]
            )
            metric_trends[metric] = trend
        else:
            logger.warning(
                f"Skipping metric {metric}: only {len(data['values'])} data points"
            )

    # Detect anomalies
    anomalies = self._detect_all_anomalies(metric_data)

    # Generate summary
    summary = self._generate_trend_summary(metric_trends, anomalies)

    # Determine time range
    all_timestamps = [r.timestamp for r in sorted_results]
    time_range = (min(all_timestamps), max(all_timestamps))

    report = TrendReport(
        metric_trends=metric_trends,
        anomalies=anomalies,
        summary=summary,
        time_range=time_range,
        total_data_points=len(sorted_results),
        metrics_analyzed=len(metric_trends)
    )

    logger.info(
        f"Trend analysis complete: {len(metric_trends)} metrics analyzed, "
        f"{len(anomalies)} anomalies detected"
    )

    return report

detect_anomalies(evaluation_results, metric_name=None, method='zscore')

Detect statistical anomalies in metric values.

Parameters:

Name	Type	Description	Default
evaluation_results	List[EvaluationResult]	List of evaluation results	required
metric_name	Optional[str]	Specific metric to check (None = check all metrics)	None
method	Literal['zscore', 'iqr']	Anomaly detection method ("zscore" or "iqr")	'zscore'

Returns:

Type	Description
List[Anomaly]	List of detected anomalies

Raises:

Type	Description
ValueError	If insufficient data or invalid inputs

Example

analyzer = TrendAnalyzer() anomalies = analyzer.detect_anomalies(results, method="zscore") for anomaly in anomalies: ... print(f"{anomaly.metric_name}: {anomaly.description}")

Source code in uaef/analysis/trends.py

def detect_anomalies(
    self,
    evaluation_results: List[EvaluationResult],
    metric_name: Optional[str] = None,
    method: Literal["zscore", "iqr"] = "zscore"
) -> List[Anomaly]:
    """
    Detect statistical anomalies in metric values.

    Args:
        evaluation_results: List of evaluation results
        metric_name: Specific metric to check (None = check all metrics)
        method: Anomaly detection method ("zscore" or "iqr")

    Returns:
        List of detected anomalies

    Raises:
        ValueError: If insufficient data or invalid inputs

    Example:
        >>> analyzer = TrendAnalyzer()
        >>> anomalies = analyzer.detect_anomalies(results, method="zscore")
        >>> for anomaly in anomalies:
        ...     print(f"{anomaly.metric_name}: {anomaly.description}")
    """
    if not evaluation_results:
        raise ValueError("No evaluation results provided")

    if len(evaluation_results) < self.min_data_points:
        raise ValueError(
            f"Insufficient data points: {len(evaluation_results)} provided, "
            f"{self.min_data_points} required"
        )

    logger.info(
        f"Detecting anomalies in {len(evaluation_results)} evaluation results "
        f"using {method} method"
        + (f" (metric: {metric_name})" if metric_name else " (all metrics)")
    )

    # Extract metric data
    metric_data = self._extract_metric_data(evaluation_results, metric_name)

    if not metric_data:
        raise ValueError(
            f"No data found for metric: {metric_name}" if metric_name
            else "No metric data found in evaluation results"
        )

    # Detect anomalies based on method
    if method == "zscore":
        anomalies = self._detect_anomalies_zscore(metric_data)
    elif method == "iqr":
        anomalies = self._detect_anomalies_iqr(metric_data)
    else:
        raise ValueError(f"Unknown anomaly detection method: {method}")

    logger.info(f"Detected {len(anomalies)} anomalies using {method} method")

    return anomalies

TrendReport

Bases: BaseModel

Comprehensive trend analysis report.

Attributes:

Name	Type	Description
metric_trends	Dict[str, MetricTrend]	Dictionary mapping metric names to their trends
anomalies	List[Anomaly]	List of detected anomalies
summary	str	High-level summary of trends
time_range	Tuple[datetime, datetime]	Time range covered by the analysis (start, end)
total_data_points	int	Total number of data points analyzed
metrics_analyzed	int	Number of unique metrics analyzed

Source code in uaef/analysis/trends.py

class TrendReport(BaseModel):
    """
    Comprehensive trend analysis report.

    Attributes:
        metric_trends: Dictionary mapping metric names to their trends
        anomalies: List of detected anomalies
        summary: High-level summary of trends
        time_range: Time range covered by the analysis (start, end)
        total_data_points: Total number of data points analyzed
        metrics_analyzed: Number of unique metrics analyzed
    """

    metric_trends: Dict[str, MetricTrend] = Field(
        default_factory=dict,
        description="Dictionary mapping metric names to their trends"
    )
    anomalies: List[Anomaly] = Field(
        default_factory=list,
        description="List of detected anomalies"
    )
    summary: str = Field(..., description="High-level summary of trends")
    time_range: Tuple[datetime, datetime] = Field(
        ...,
        description="Time range covered by the analysis"
    )
    total_data_points: int = Field(
        ...,
        description="Total number of data points analyzed"
    )
    metrics_analyzed: int = Field(
        ...,
        description="Number of unique metrics analyzed"
    )

    class Config:
        """Pydantic configuration."""
        json_schema_extra = {
            "example": {
                "metric_trends": {},
                "anomalies": [],
                "summary": "Overall improving trend across 5 metrics with 2 anomalies detected",
                "time_range": ("2024-01-01T00:00:00Z", "2024-01-15T00:00:00Z"),
                "total_data_points": 100,
                "metrics_analyzed": 5
            }
        }

Config

Pydantic configuration.

Source code in uaef/analysis/trends.py

class Config:
    """Pydantic configuration."""
    json_schema_extra = {
        "example": {
            "metric_trends": {},
            "anomalies": [],
            "summary": "Overall improving trend across 5 metrics with 2 anomalies detected",
            "time_range": ("2024-01-01T00:00:00Z", "2024-01-15T00:00:00Z"),
            "total_data_points": 100,
            "metrics_analyzed": 5
        }
    }

uaef.analysis.improvement_tracker

Improvement tracker for monitoring recommendation implementation and impact.

ImprovementDelta

Bases: BaseModel

Calculated improvement delta for a metric.

Attributes:

Name	Type	Description
metric_name	str	Name of the metric
baseline_value	float	Value before implementation
post_implementation_value	float	Value after implementation
absolute_change	float	Absolute change (post - baseline)
percentage_change	float	Percentage change ((post - baseline) / baseline * 100)
improved	bool	Whether the metric improved (positive change)

Source code in uaef/analysis/improvement_tracker.py

class ImprovementDelta(BaseModel):
    """
    Calculated improvement delta for a metric.

    Attributes:
        metric_name: Name of the metric
        baseline_value: Value before implementation
        post_implementation_value: Value after implementation
        absolute_change: Absolute change (post - baseline)
        percentage_change: Percentage change ((post - baseline) / baseline * 100)
        improved: Whether the metric improved (positive change)
    """

    metric_name: str = Field(..., description="Name of the metric")
    baseline_value: float = Field(..., description="Value before implementation")
    post_implementation_value: float = Field(..., description="Value after implementation")
    absolute_change: float = Field(..., description="Absolute change")
    percentage_change: float = Field(..., description="Percentage change")
    improved: bool = Field(..., description="Whether the metric improved")

    class Config:
        """Pydantic configuration."""
        json_schema_extra = {
            "example": {
                "metric_name": "tool_accuracy",
                "baseline_value": 0.65,
                "post_implementation_value": 0.82,
                "absolute_change": 0.17,
                "percentage_change": 26.15,
                "improved": True
            }
        }

Config

Pydantic configuration.

Source code in uaef/analysis/improvement_tracker.py

class Config:
    """Pydantic configuration."""
    json_schema_extra = {
        "example": {
            "metric_name": "tool_accuracy",
            "baseline_value": 0.65,
            "post_implementation_value": 0.82,
            "absolute_change": 0.17,
            "percentage_change": 26.15,
            "improved": True
        }
    }

ImprovementReport

Bases: BaseModel

Report showing before/after comparison for a recommendation.

Attributes:

Name	Type	Description
recommendation_id	UUID	ID of the tracked recommendation
recommendation_title	str	Title of the recommendation
recommendation_category	str	Category of the recommendation
status	Literal['pending', 'in_progress', 'completed']	Current implementation status
metric_deltas	List[ImprovementDelta]	List of improvement deltas for affected metrics
overall_improvement	str	Overall improvement summary
implementation_duration_days	Optional[float]	Days from start to completion (if completed)
created_at	datetime	When tracking started
completed_at	Optional[datetime]	When implementation completed (if completed)
notes	str	Implementation notes

Source code in uaef/analysis/improvement_tracker.py

class ImprovementReport(BaseModel):
    """
    Report showing before/after comparison for a recommendation.

    Attributes:
        recommendation_id: ID of the tracked recommendation
        recommendation_title: Title of the recommendation
        recommendation_category: Category of the recommendation
        status: Current implementation status
        metric_deltas: List of improvement deltas for affected metrics
        overall_improvement: Overall improvement summary
        implementation_duration_days: Days from start to completion (if completed)
        created_at: When tracking started
        completed_at: When implementation completed (if completed)
        notes: Implementation notes
    """

    recommendation_id: UUID = Field(
        ...,
        description="ID of the tracked recommendation"
    )
    recommendation_title: str = Field(..., description="Title of the recommendation")
    recommendation_category: str = Field(..., description="Category of the recommendation")
    status: Literal["pending", "in_progress", "completed"] = Field(
        ...,
        description="Current implementation status"
    )
    metric_deltas: List[ImprovementDelta] = Field(
        default_factory=list,
        description="List of improvement deltas for affected metrics"
    )
    overall_improvement: str = Field(
        ...,
        description="Overall improvement summary"
    )
    implementation_duration_days: Optional[float] = Field(
        default=None,
        description="Days from start to completion"
    )
    created_at: datetime = Field(..., description="When tracking started")
    completed_at: Optional[datetime] = Field(
        default=None,
        description="When implementation completed"
    )
    notes: str = Field(default="", description="Implementation notes")

    class Config:
        """Pydantic configuration."""
        json_schema_extra = {
            "example": {
                "recommendation_id": "550e8400-e29b-41d4-a716-446655440000",
                "recommendation_title": "Add data validation tool",
                "recommendation_category": "tool",
                "status": "completed",
                "metric_deltas": [
                    {
                        "metric_name": "tool_accuracy",
                        "baseline_value": 0.65,
                        "post_implementation_value": 0.82,
                        "absolute_change": 0.17,
                        "percentage_change": 26.15,
                        "improved": True
                    }
                ],
                "overall_improvement": "Improved 1/1 metrics (100% success rate)",
                "implementation_duration_days": 3.0,
                "created_at": "2024-01-01T10:00:00Z",
                "completed_at": "2024-01-05T10:00:00Z",
                "notes": "Added validate_data_format tool"
            }
        }

Config

Pydantic configuration.

Source code in uaef/analysis/improvement_tracker.py

class Config:
    """Pydantic configuration."""
    json_schema_extra = {
        "example": {
            "recommendation_id": "550e8400-e29b-41d4-a716-446655440000",
            "recommendation_title": "Add data validation tool",
            "recommendation_category": "tool",
            "status": "completed",
            "metric_deltas": [
                {
                    "metric_name": "tool_accuracy",
                    "baseline_value": 0.65,
                    "post_implementation_value": 0.82,
                    "absolute_change": 0.17,
                    "percentage_change": 26.15,
                    "improved": True
                }
            ],
            "overall_improvement": "Improved 1/1 metrics (100% success rate)",
            "implementation_duration_days": 3.0,
            "created_at": "2024-01-01T10:00:00Z",
            "completed_at": "2024-01-05T10:00:00Z",
            "notes": "Added validate_data_format tool"
        }
    }

ImprovementTracker

Tracker for monitoring recommendation implementation and measuring impact.

Tracks: - Recommendation implementation status (pending, in_progress, completed) - Baseline metrics before implementation - Post-implementation metrics - Improvement deltas and percentage changes - Multiple recommendations simultaneously

Provides: - Before/after comparison reports - Impact measurement for completed recommendations - Status tracking across multiple recommendations

Source code in uaef/analysis/improvement_tracker.py

class ImprovementTracker:
    """
    Tracker for monitoring recommendation implementation and measuring impact.

    Tracks:
    - Recommendation implementation status (pending, in_progress, completed)
    - Baseline metrics before implementation
    - Post-implementation metrics
    - Improvement deltas and percentage changes
    - Multiple recommendations simultaneously

    Provides:
    - Before/after comparison reports
    - Impact measurement for completed recommendations
    - Status tracking across multiple recommendations
    """

    def __init__(self):
        """Initialize the improvement tracker."""
        self._tracked_recommendations: Dict[UUID, RecommendationStatus] = {}
        logger.info("ImprovementTracker initialized")

    def start_tracking(
        self,
        recommendation: Recommendation,
        baseline_evaluation_results: List[EvaluationResult]
    ) -> UUID:
        """
        Start tracking a recommendation implementation.

        Records baseline metrics from evaluation results before implementation begins.

        Args:
            recommendation: The recommendation to track
            baseline_evaluation_results: Evaluation results before implementation

        Returns:
            UUID: Recommendation tracking ID

        Raises:
            ValueError: If no baseline evaluation results provided

        Example:
            >>> tracker = ImprovementTracker()
            >>> rec_id = tracker.start_tracking(recommendation, baseline_results)
            >>> print(f"Tracking recommendation: {rec_id}")
        """
        if not baseline_evaluation_results:
            raise ValueError("No baseline evaluation results provided")

        # Calculate baseline metrics
        baseline_metrics = self._calculate_average_metrics(baseline_evaluation_results)

        # Create tracking status
        status = RecommendationStatus(
            recommendation=recommendation,
            status="pending",
            baseline_metrics=baseline_metrics
        )

        self._tracked_recommendations[status.recommendation_id] = status

        logger.info(
            f"Started tracking recommendation '{recommendation.title}' "
            f"(ID: {status.recommendation_id})"
        )

        return status.recommendation_id

    def mark_in_progress(
        self,
        recommendation_id: UUID,
        notes: str = ""
    ) -> None:
        """
        Mark a recommendation as in progress.

        Args:
            recommendation_id: ID of the tracked recommendation
            notes: Optional notes about the implementation

        Raises:
            ValueError: If recommendation ID not found or already completed

        Example:
            >>> tracker.mark_in_progress(rec_id, "Started implementing tool")
        """
        status = self._get_status(recommendation_id)

        if status.status == "completed":
            raise ValueError(
                f"Recommendation {recommendation_id} is already completed"
            )

        status.status = "in_progress"
        status.started_at = datetime.utcnow()
        if notes:
            status.notes = notes

        logger.info(f"Marked recommendation {recommendation_id} as in_progress")

    def mark_completed(
        self,
        recommendation_id: UUID,
        post_implementation_evaluation_results: List[EvaluationResult],
        notes: str = ""
    ) -> ImprovementReport:
        """
        Mark a recommendation as completed and record post-implementation metrics.

        Calculates improvement deltas and generates a before/after report.

        Args:
            recommendation_id: ID of the tracked recommendation
            post_implementation_evaluation_results: Evaluation results after implementation
            notes: Optional notes about the implementation

        Returns:
            ImprovementReport: Report showing before/after comparison

        Raises:
            ValueError: If recommendation ID not found or no post-implementation results

        Example:
            >>> report = tracker.mark_completed(rec_id, post_results, "Tool added successfully")
            >>> print(f"Overall improvement: {report.overall_improvement}")
        """
        status = self._get_status(recommendation_id)

        if not post_implementation_evaluation_results:
            raise ValueError("No post-implementation evaluation results provided")

        # Calculate post-implementation metrics
        post_metrics = self._calculate_average_metrics(
            post_implementation_evaluation_results
        )

        # Update status
        status.status = "completed"
        status.completed_at = datetime.utcnow()
        status.post_implementation_metrics = post_metrics
        if notes:
            status.notes = notes

        # If not already marked in_progress, set started_at to completed_at
        if status.started_at is None:
            status.started_at = status.completed_at

        logger.info(f"Marked recommendation {recommendation_id} as completed")

        # Generate improvement report
        return self.generate_improvement_report(recommendation_id)

    def generate_improvement_report(
        self,
        recommendation_id: UUID
    ) -> ImprovementReport:
        """
        Generate an improvement report for a tracked recommendation.

        Shows before/after comparison with improvement deltas.

        Args:
            recommendation_id: ID of the tracked recommendation

        Returns:
            ImprovementReport: Report with before/after comparison

        Raises:
            ValueError: If recommendation ID not found or not completed

        Example:
            >>> report = tracker.generate_improvement_report(rec_id)
            >>> for delta in report.metric_deltas:
            ...     print(f"{delta.metric_name}: {delta.percentage_change:.1f}%")
        """
        status = self._get_status(recommendation_id)

        if status.status != "completed":
            raise ValueError(
                f"Recommendation {recommendation_id} is not completed yet "
                f"(status: {status.status})"
            )

        # Calculate improvement deltas
        metric_deltas = self._calculate_improvement_deltas(status)

        # Calculate implementation duration
        duration_days = None
        if status.started_at and status.completed_at:
            duration = status.completed_at - status.started_at
            duration_days = duration.total_seconds() / 86400  # Convert to days

        # Generate overall improvement summary
        improved_count = sum(1 for delta in metric_deltas if delta.improved)
        total_count = len(metric_deltas)

        if total_count == 0:
            overall_improvement = "No metrics tracked"
        else:
            success_rate = (improved_count / total_count) * 100
            overall_improvement = (
                f"Improved {improved_count}/{total_count} metrics "
                f"({success_rate:.1f}% success rate)"
            )

        report = ImprovementReport(
            recommendation_id=status.recommendation_id,
            recommendation_title=status.recommendation.title,
            recommendation_category=status.recommendation.category,
            status=status.status,
            metric_deltas=metric_deltas,
            overall_improvement=overall_improvement,
            implementation_duration_days=duration_days,
            created_at=status.created_at,
            completed_at=status.completed_at,
            notes=status.notes
        )

        logger.info(
            f"Generated improvement report for {recommendation_id}: "
            f"{overall_improvement}"
        )

        return report

    def get_all_tracked_recommendations(self) -> List[RecommendationStatus]:
        """
        Get all tracked recommendations.

        Returns:
            List of all RecommendationStatus objects

        Example:
            >>> all_recs = tracker.get_all_tracked_recommendations()
            >>> for rec in all_recs:
            ...     print(f"{rec.recommendation.title}: {rec.status}")
        """
        return list(self._tracked_recommendations.values())

    def get_recommendations_by_status(
        self,
        status: Literal["pending", "in_progress", "completed"]
    ) -> List[RecommendationStatus]:
        """
        Get tracked recommendations filtered by status.

        Args:
            status: Status to filter by

        Returns:
            List of RecommendationStatus objects with the specified status

        Example:
            >>> pending = tracker.get_recommendations_by_status("pending")
            >>> print(f"Pending recommendations: {len(pending)}")
        """
        return [
            rec for rec in self._tracked_recommendations.values()
            if rec.status == status
        ]

    def generate_summary_report(self) -> Dict[str, Any]:
        """
        Generate a summary report of all tracked recommendations.

        Returns:
            Dictionary with summary statistics and reports

        Example:
            >>> summary = tracker.generate_summary_report()
            >>> print(f"Total tracked: {summary['total_tracked']}")
            >>> print(f"Completed: {summary['completed_count']}")
        """
        all_recs = self.get_all_tracked_recommendations()

        pending = self.get_recommendations_by_status("pending")
        in_progress = self.get_recommendations_by_status("in_progress")
        completed = self.get_recommendations_by_status("completed")

        # Generate reports for completed recommendations
        completed_reports = []
        for rec in completed:
            try:
                report = self.generate_improvement_report(rec.recommendation_id)
                completed_reports.append(report)
            except Exception as e:
                logger.warning(
                    f"Failed to generate report for {rec.recommendation_id}: {e}"
                )

        # Calculate overall statistics
        total_improved_metrics = sum(
            sum(1 for delta in report.metric_deltas if delta.improved)
            for report in completed_reports
        )
        total_tracked_metrics = sum(
            len(report.metric_deltas)
            for report in completed_reports
        )

        overall_success_rate = 0.0
        if total_tracked_metrics > 0:
            overall_success_rate = (total_improved_metrics / total_tracked_metrics) * 100

        summary = {
            "total_tracked": len(all_recs),
            "pending_count": len(pending),
            "in_progress_count": len(in_progress),
            "completed_count": len(completed),
            "completed_reports": completed_reports,
            "total_improved_metrics": total_improved_metrics,
            "total_tracked_metrics": total_tracked_metrics,
            "overall_success_rate": overall_success_rate,
            "pending_recommendations": [
                {
                    "id": str(rec.recommendation_id),
                    "title": rec.recommendation.title,
                    "category": rec.recommendation.category,
                    "priority": rec.recommendation.priority
                }
                for rec in pending
            ],
            "in_progress_recommendations": [
                {
                    "id": str(rec.recommendation_id),
                    "title": rec.recommendation.title,
                    "category": rec.recommendation.category,
                    "priority": rec.recommendation.priority,
                    "started_at": rec.started_at.isoformat() if rec.started_at else None
                }
                for rec in in_progress
            ]
        }

        logger.info(
            f"Generated summary report: {len(completed)} completed, "
            f"{len(in_progress)} in progress, {len(pending)} pending"
        )

        return summary

    def _get_status(self, recommendation_id: UUID) -> RecommendationStatus:
        """
        Get recommendation status by ID.

        Args:
            recommendation_id: ID of the tracked recommendation

        Returns:
            RecommendationStatus object

        Raises:
            ValueError: If recommendation ID not found
        """
        if recommendation_id not in self._tracked_recommendations:
            raise ValueError(
                f"Recommendation {recommendation_id} not found in tracker"
            )

        return self._tracked_recommendations[recommendation_id]

    def _calculate_average_metrics(
        self,
        evaluation_results: List[EvaluationResult]
    ) -> Dict[str, float]:
        """
        Calculate average metric values across evaluation results.

        Args:
            evaluation_results: List of evaluation results

        Returns:
            Dictionary mapping metric names to average values
        """
        metric_values: Dict[str, List[float]] = {}

        # Collect all metric values
        for result in evaluation_results:
            for dimension_result in result.dimension_results:
                for metric_score in dimension_result.scored_metrics():
                    metric_name = metric_score.metric_name
                    if metric_name not in metric_values:
                        metric_values[metric_name] = []
                    metric_values[metric_name].append(metric_score.score)

        # Calculate averages
        average_metrics = {
            metric_name: sum(values) / len(values)
            for metric_name, values in metric_values.items()
        }

        return average_metrics

    def _calculate_improvement_deltas(
        self,
        status: RecommendationStatus
    ) -> List[ImprovementDelta]:
        """
        Calculate improvement deltas for all tracked metrics.

        Args:
            status: RecommendationStatus with baseline and post-implementation metrics

        Returns:
            List of ImprovementDelta objects
        """
        deltas = []

        # Get all metrics that appear in either baseline or post-implementation
        all_metrics = set(status.baseline_metrics.keys()) | set(
            status.post_implementation_metrics.keys()
        )

        for metric_name in all_metrics:
            baseline = status.baseline_metrics.get(metric_name, 0.0)
            post = status.post_implementation_metrics.get(metric_name, 0.0)

            absolute_change = post - baseline

            # Calculate percentage change (handle division by zero)
            if baseline == 0.0:
                if post == 0.0:
                    percentage_change = 0.0
                else:
                    percentage_change = 100.0  # Arbitrary large value
            else:
                percentage_change = (absolute_change / baseline) * 100

            improved = absolute_change > 0

            delta = ImprovementDelta(
                metric_name=metric_name,
                baseline_value=baseline,
                post_implementation_value=post,
                absolute_change=absolute_change,
                percentage_change=percentage_change,
                improved=improved
            )

            deltas.append(delta)

        # Sort by absolute change (descending)
        deltas.sort(key=lambda d: abs(d.absolute_change), reverse=True)

        return deltas

generate_improvement_report(recommendation_id)

Generate an improvement report for a tracked recommendation.

Shows before/after comparison with improvement deltas.

Parameters:

Name	Type	Description	Default
recommendation_id	UUID	ID of the tracked recommendation	required

Returns:

Name	Type	Description
ImprovementReport	ImprovementReport	Report with before/after comparison

Raises:

Type	Description
ValueError	If recommendation ID not found or not completed

Example

report = tracker.generate_improvement_report(rec_id) for delta in report.metric_deltas: ... print(f"{delta.metric_name}: {delta.percentage_change:.1f}%")

Source code in uaef/analysis/improvement_tracker.py

def generate_improvement_report(
    self,
    recommendation_id: UUID
) -> ImprovementReport:
    """
    Generate an improvement report for a tracked recommendation.

    Shows before/after comparison with improvement deltas.

    Args:
        recommendation_id: ID of the tracked recommendation

    Returns:
        ImprovementReport: Report with before/after comparison

    Raises:
        ValueError: If recommendation ID not found or not completed

    Example:
        >>> report = tracker.generate_improvement_report(rec_id)
        >>> for delta in report.metric_deltas:
        ...     print(f"{delta.metric_name}: {delta.percentage_change:.1f}%")
    """
    status = self._get_status(recommendation_id)

    if status.status != "completed":
        raise ValueError(
            f"Recommendation {recommendation_id} is not completed yet "
            f"(status: {status.status})"
        )

    # Calculate improvement deltas
    metric_deltas = self._calculate_improvement_deltas(status)

    # Calculate implementation duration
    duration_days = None
    if status.started_at and status.completed_at:
        duration = status.completed_at - status.started_at
        duration_days = duration.total_seconds() / 86400  # Convert to days

    # Generate overall improvement summary
    improved_count = sum(1 for delta in metric_deltas if delta.improved)
    total_count = len(metric_deltas)

    if total_count == 0:
        overall_improvement = "No metrics tracked"
    else:
        success_rate = (improved_count / total_count) * 100
        overall_improvement = (
            f"Improved {improved_count}/{total_count} metrics "
            f"({success_rate:.1f}% success rate)"
        )

    report = ImprovementReport(
        recommendation_id=status.recommendation_id,
        recommendation_title=status.recommendation.title,
        recommendation_category=status.recommendation.category,
        status=status.status,
        metric_deltas=metric_deltas,
        overall_improvement=overall_improvement,
        implementation_duration_days=duration_days,
        created_at=status.created_at,
        completed_at=status.completed_at,
        notes=status.notes
    )

    logger.info(
        f"Generated improvement report for {recommendation_id}: "
        f"{overall_improvement}"
    )

    return report

generate_summary_report()

Generate a summary report of all tracked recommendations.

Returns:

Type	Description
Dict[str, Any]	Dictionary with summary statistics and reports

Example

summary = tracker.generate_summary_report() print(f"Total tracked: {summary['total_tracked']}") print(f"Completed: {summary['completed_count']}")

Source code in uaef/analysis/improvement_tracker.py

def generate_summary_report(self) -> Dict[str, Any]:
    """
    Generate a summary report of all tracked recommendations.

    Returns:
        Dictionary with summary statistics and reports

    Example:
        >>> summary = tracker.generate_summary_report()
        >>> print(f"Total tracked: {summary['total_tracked']}")
        >>> print(f"Completed: {summary['completed_count']}")
    """
    all_recs = self.get_all_tracked_recommendations()

    pending = self.get_recommendations_by_status("pending")
    in_progress = self.get_recommendations_by_status("in_progress")
    completed = self.get_recommendations_by_status("completed")

    # Generate reports for completed recommendations
    completed_reports = []
    for rec in completed:
        try:
            report = self.generate_improvement_report(rec.recommendation_id)
            completed_reports.append(report)
        except Exception as e:
            logger.warning(
                f"Failed to generate report for {rec.recommendation_id}: {e}"
            )

    # Calculate overall statistics
    total_improved_metrics = sum(
        sum(1 for delta in report.metric_deltas if delta.improved)
        for report in completed_reports
    )
    total_tracked_metrics = sum(
        len(report.metric_deltas)
        for report in completed_reports
    )

    overall_success_rate = 0.0
    if total_tracked_metrics > 0:
        overall_success_rate = (total_improved_metrics / total_tracked_metrics) * 100

    summary = {
        "total_tracked": len(all_recs),
        "pending_count": len(pending),
        "in_progress_count": len(in_progress),
        "completed_count": len(completed),
        "completed_reports": completed_reports,
        "total_improved_metrics": total_improved_metrics,
        "total_tracked_metrics": total_tracked_metrics,
        "overall_success_rate": overall_success_rate,
        "pending_recommendations": [
            {
                "id": str(rec.recommendation_id),
                "title": rec.recommendation.title,
                "category": rec.recommendation.category,
                "priority": rec.recommendation.priority
            }
            for rec in pending
        ],
        "in_progress_recommendations": [
            {
                "id": str(rec.recommendation_id),
                "title": rec.recommendation.title,
                "category": rec.recommendation.category,
                "priority": rec.recommendation.priority,
                "started_at": rec.started_at.isoformat() if rec.started_at else None
            }
            for rec in in_progress
        ]
    }

    logger.info(
        f"Generated summary report: {len(completed)} completed, "
        f"{len(in_progress)} in progress, {len(pending)} pending"
    )

    return summary

get_all_tracked_recommendations()

Get all tracked recommendations.

Returns:

Type	Description
List[RecommendationStatus]	List of all RecommendationStatus objects

Example

all_recs = tracker.get_all_tracked_recommendations() for rec in all_recs: ... print(f"{rec.recommendation.title}: {rec.status}")

Source code in uaef/analysis/improvement_tracker.py

def get_all_tracked_recommendations(self) -> List[RecommendationStatus]:
    """
    Get all tracked recommendations.

    Returns:
        List of all RecommendationStatus objects

    Example:
        >>> all_recs = tracker.get_all_tracked_recommendations()
        >>> for rec in all_recs:
        ...     print(f"{rec.recommendation.title}: {rec.status}")
    """
    return list(self._tracked_recommendations.values())

get_recommendations_by_status(status)

Get tracked recommendations filtered by status.

Parameters:

Name	Type	Description	Default
status	Literal['pending', 'in_progress', 'completed']	Status to filter by	required

Returns:

Type	Description
List[RecommendationStatus]	List of RecommendationStatus objects with the specified status

Example

pending = tracker.get_recommendations_by_status("pending") print(f"Pending recommendations: {len(pending)}")

Source code in uaef/analysis/improvement_tracker.py

def get_recommendations_by_status(
    self,
    status: Literal["pending", "in_progress", "completed"]
) -> List[RecommendationStatus]:
    """
    Get tracked recommendations filtered by status.

    Args:
        status: Status to filter by

    Returns:
        List of RecommendationStatus objects with the specified status

    Example:
        >>> pending = tracker.get_recommendations_by_status("pending")
        >>> print(f"Pending recommendations: {len(pending)}")
    """
    return [
        rec for rec in self._tracked_recommendations.values()
        if rec.status == status
    ]

mark_completed(recommendation_id, post_implementation_evaluation_results, notes='')

Mark a recommendation as completed and record post-implementation metrics.

Calculates improvement deltas and generates a before/after report.

Parameters:

Name	Type	Description	Default
recommendation_id	UUID	ID of the tracked recommendation	required
post_implementation_evaluation_results	List[EvaluationResult]	Evaluation results after implementation	required
notes	str	Optional notes about the implementation	''

Returns:

Name	Type	Description
ImprovementReport	ImprovementReport	Report showing before/after comparison

Raises:

Type	Description
ValueError	If recommendation ID not found or no post-implementation results

Example

report = tracker.mark_completed(rec_id, post_results, "Tool added successfully") print(f"Overall improvement: {report.overall_improvement}")

Source code in uaef/analysis/improvement_tracker.py

def mark_completed(
    self,
    recommendation_id: UUID,
    post_implementation_evaluation_results: List[EvaluationResult],
    notes: str = ""
) -> ImprovementReport:
    """
    Mark a recommendation as completed and record post-implementation metrics.

    Calculates improvement deltas and generates a before/after report.

    Args:
        recommendation_id: ID of the tracked recommendation
        post_implementation_evaluation_results: Evaluation results after implementation
        notes: Optional notes about the implementation

    Returns:
        ImprovementReport: Report showing before/after comparison

    Raises:
        ValueError: If recommendation ID not found or no post-implementation results

    Example:
        >>> report = tracker.mark_completed(rec_id, post_results, "Tool added successfully")
        >>> print(f"Overall improvement: {report.overall_improvement}")
    """
    status = self._get_status(recommendation_id)

    if not post_implementation_evaluation_results:
        raise ValueError("No post-implementation evaluation results provided")

    # Calculate post-implementation metrics
    post_metrics = self._calculate_average_metrics(
        post_implementation_evaluation_results
    )

    # Update status
    status.status = "completed"
    status.completed_at = datetime.utcnow()
    status.post_implementation_metrics = post_metrics
    if notes:
        status.notes = notes

    # If not already marked in_progress, set started_at to completed_at
    if status.started_at is None:
        status.started_at = status.completed_at

    logger.info(f"Marked recommendation {recommendation_id} as completed")

    # Generate improvement report
    return self.generate_improvement_report(recommendation_id)

mark_in_progress(recommendation_id, notes='')

Mark a recommendation as in progress.

Parameters:

Name	Type	Description	Default
recommendation_id	UUID	ID of the tracked recommendation	required
notes	str	Optional notes about the implementation	''

Raises:

Type	Description
ValueError	If recommendation ID not found or already completed

Example

tracker.mark_in_progress(rec_id, "Started implementing tool")

Source code in uaef/analysis/improvement_tracker.py

def mark_in_progress(
    self,
    recommendation_id: UUID,
    notes: str = ""
) -> None:
    """
    Mark a recommendation as in progress.

    Args:
        recommendation_id: ID of the tracked recommendation
        notes: Optional notes about the implementation

    Raises:
        ValueError: If recommendation ID not found or already completed

    Example:
        >>> tracker.mark_in_progress(rec_id, "Started implementing tool")
    """
    status = self._get_status(recommendation_id)

    if status.status == "completed":
        raise ValueError(
            f"Recommendation {recommendation_id} is already completed"
        )

    status.status = "in_progress"
    status.started_at = datetime.utcnow()
    if notes:
        status.notes = notes

    logger.info(f"Marked recommendation {recommendation_id} as in_progress")

start_tracking(recommendation, baseline_evaluation_results)

Start tracking a recommendation implementation.

Records baseline metrics from evaluation results before implementation begins.

Parameters:

Name	Type	Description	Default
recommendation	Recommendation	The recommendation to track	required
baseline_evaluation_results	List[EvaluationResult]	Evaluation results before implementation	required

Returns:

Name	Type	Description
UUID	UUID	Recommendation tracking ID

Raises:

Type	Description
ValueError	If no baseline evaluation results provided

Example

tracker = ImprovementTracker() rec_id = tracker.start_tracking(recommendation, baseline_results) print(f"Tracking recommendation: {rec_id}")

Source code in uaef/analysis/improvement_tracker.py

def start_tracking(
    self,
    recommendation: Recommendation,
    baseline_evaluation_results: List[EvaluationResult]
) -> UUID:
    """
    Start tracking a recommendation implementation.

    Records baseline metrics from evaluation results before implementation begins.

    Args:
        recommendation: The recommendation to track
        baseline_evaluation_results: Evaluation results before implementation

    Returns:
        UUID: Recommendation tracking ID

    Raises:
        ValueError: If no baseline evaluation results provided

    Example:
        >>> tracker = ImprovementTracker()
        >>> rec_id = tracker.start_tracking(recommendation, baseline_results)
        >>> print(f"Tracking recommendation: {rec_id}")
    """
    if not baseline_evaluation_results:
        raise ValueError("No baseline evaluation results provided")

    # Calculate baseline metrics
    baseline_metrics = self._calculate_average_metrics(baseline_evaluation_results)

    # Create tracking status
    status = RecommendationStatus(
        recommendation=recommendation,
        status="pending",
        baseline_metrics=baseline_metrics
    )

    self._tracked_recommendations[status.recommendation_id] = status

    logger.info(
        f"Started tracking recommendation '{recommendation.title}' "
        f"(ID: {status.recommendation_id})"
    )

    return status.recommendation_id

RecommendationStatus

Bases: BaseModel

Status tracking for a recommendation implementation.

Attributes:

Name	Type	Description
recommendation_id	UUID	Unique identifier for this recommendation tracking
recommendation	Recommendation	The recommendation being tracked
status	Literal['pending', 'in_progress', 'completed']	Current implementation status
baseline_metrics	Dict[str, float]	Metric values before implementation
post_implementation_metrics	Dict[str, float]	Metric values after implementation (if completed)
created_at	datetime	When tracking started
started_at	Optional[datetime]	When implementation started (if in_progress or completed)
completed_at	Optional[datetime]	When implementation completed (if completed)
notes	str	Implementation notes or observations

Source code in uaef/analysis/improvement_tracker.py

class RecommendationStatus(BaseModel):
    """
    Status tracking for a recommendation implementation.

    Attributes:
        recommendation_id: Unique identifier for this recommendation tracking
        recommendation: The recommendation being tracked
        status: Current implementation status
        baseline_metrics: Metric values before implementation
        post_implementation_metrics: Metric values after implementation (if completed)
        created_at: When tracking started
        started_at: When implementation started (if in_progress or completed)
        completed_at: When implementation completed (if completed)
        notes: Implementation notes or observations
    """

    recommendation_id: UUID = Field(
        default_factory=uuid4,
        description="Unique identifier for this recommendation tracking"
    )
    recommendation: Recommendation = Field(
        ...,
        description="The recommendation being tracked"
    )
    status: Literal["pending", "in_progress", "completed"] = Field(
        default="pending",
        description="Current implementation status"
    )
    baseline_metrics: Dict[str, float] = Field(
        default_factory=dict,
        description="Metric values before implementation"
    )
    post_implementation_metrics: Dict[str, float] = Field(
        default_factory=dict,
        description="Metric values after implementation"
    )
    created_at: datetime = Field(
        default_factory=datetime.utcnow,
        description="When tracking started"
    )
    started_at: Optional[datetime] = Field(
        default=None,
        description="When implementation started"
    )
    completed_at: Optional[datetime] = Field(
        default=None,
        description="When implementation completed"
    )
    notes: str = Field(
        default="",
        description="Implementation notes or observations"
    )

    class Config:
        """Pydantic configuration."""
        json_schema_extra = {
            "example": {
                "recommendation_id": "550e8400-e29b-41d4-a716-446655440000",
                "recommendation": {
                    "category": "tool",
                    "priority": "high",
                    "title": "Add data validation tool",
                    "description": "Agent frequently fails when validating data formats",
                    "rationale": "15 failures related to data validation",
                    "impact": "Expected 20% improvement in tool_accuracy",
                    "affected_metrics": ["tool_accuracy"]
                },
                "status": "completed",
                "baseline_metrics": {"tool_accuracy": 0.65},
                "post_implementation_metrics": {"tool_accuracy": 0.82},
                "created_at": "2024-01-01T10:00:00Z",
                "started_at": "2024-01-02T10:00:00Z",
                "completed_at": "2024-01-05T10:00:00Z",
                "notes": "Added validate_data_format tool and updated prompt"
            }
        }

Config

Pydantic configuration.

Source code in uaef/analysis/improvement_tracker.py

class Config:
    """Pydantic configuration."""
    json_schema_extra = {
        "example": {
            "recommendation_id": "550e8400-e29b-41d4-a716-446655440000",
            "recommendation": {
                "category": "tool",
                "priority": "high",
                "title": "Add data validation tool",
                "description": "Agent frequently fails when validating data formats",
                "rationale": "15 failures related to data validation",
                "impact": "Expected 20% improvement in tool_accuracy",
                "affected_metrics": ["tool_accuracy"]
            },
            "status": "completed",
            "baseline_metrics": {"tool_accuracy": 0.65},
            "post_implementation_metrics": {"tool_accuracy": 0.82},
            "created_at": "2024-01-01T10:00:00Z",
            "started_at": "2024-01-02T10:00:00Z",
            "completed_at": "2024-01-05T10:00:00Z",
            "notes": "Added validate_data_format tool and updated prompt"
        }
    }

uaef.analysis.recommendations

Recommendation engine for agent improvement insights.

Recommendation

Bases: BaseModel

A specific, actionable recommendation for agent improvement.

Attributes:

Name	Type	Description
category	Literal['prompt', 'tool', 'architecture', 'reasoning', 'context', 'safety']	Category of recommendation (prompt, tool, architecture, reasoning)
priority	Literal['critical', 'high', 'medium', 'low']	Priority level (critical, high, medium, low)
title	str	Short title of the recommendation
description	str	Detailed description of the issue and recommendation
rationale	str	Why this recommendation is being made
impact	str	Expected impact if implemented
examples	List[str]	Examples from high-scoring test cases (if applicable)
affected_metrics	List[str]	Metrics that would improve with this recommendation
implementation_steps	List[str]	Specific steps to implement the recommendation

Source code in uaef/analysis/recommendations.py

class Recommendation(BaseModel):
    """
    A specific, actionable recommendation for agent improvement.

    Attributes:
        category: Category of recommendation (prompt, tool, architecture, reasoning)
        priority: Priority level (critical, high, medium, low)
        title: Short title of the recommendation
        description: Detailed description of the issue and recommendation
        rationale: Why this recommendation is being made
        impact: Expected impact if implemented
        examples: Examples from high-scoring test cases (if applicable)
        affected_metrics: Metrics that would improve with this recommendation
        implementation_steps: Specific steps to implement the recommendation
    """

    category: Literal["prompt", "tool", "architecture", "reasoning", "context", "safety"] = Field(
        ...,
        description="Category of recommendation"
    )
    priority: Literal["critical", "high", "medium", "low"] = Field(
        ...,
        description="Priority level"
    )
    title: str = Field(..., description="Short title of the recommendation")
    description: str = Field(
        ...,
        description="Detailed description of the issue and recommendation"
    )
    rationale: str = Field(..., description="Why this recommendation is being made")
    impact: str = Field(..., description="Expected impact if implemented")
    examples: List[str] = Field(
        default_factory=list,
        description="Examples from high-scoring test cases"
    )
    affected_metrics: List[str] = Field(
        default_factory=list,
        description="Metrics that would improve with this recommendation"
    )
    implementation_steps: List[str] = Field(
        default_factory=list,
        description="Specific steps to implement the recommendation"
    )

    class Config:
        """Pydantic configuration."""
        json_schema_extra = {
            "example": {
                "category": "tool",
                "priority": "high",
                "title": "Add data validation tool",
                "description": "Agent frequently fails when validating data formats",
                "rationale": "15 failures related to data validation across test cases",
                "impact": "Expected 20% improvement in tool_accuracy metric",
                "examples": ["High-scoring case used explicit validation step"],
                "affected_metrics": ["tool_accuracy", "tool_sequence_accuracy"],
                "implementation_steps": [
                    "Create validate_data_format tool",
                    "Add tool to agent's tool set",
                    "Update prompt with validation examples"
                ]
            }
        }

Config

Pydantic configuration.

Source code in uaef/analysis/recommendations.py

class Config:
    """Pydantic configuration."""
    json_schema_extra = {
        "example": {
            "category": "tool",
            "priority": "high",
            "title": "Add data validation tool",
            "description": "Agent frequently fails when validating data formats",
            "rationale": "15 failures related to data validation across test cases",
            "impact": "Expected 20% improvement in tool_accuracy metric",
            "examples": ["High-scoring case used explicit validation step"],
            "affected_metrics": ["tool_accuracy", "tool_sequence_accuracy"],
            "implementation_steps": [
                "Create validate_data_format tool",
                "Add tool to agent's tool set",
                "Update prompt with validation examples"
            ]
        }
    }

RecommendationEngine

Engine for generating actionable recommendations for agent improvement.

Integrates with RootCauseAnalyzer, FailureClusterer, and TrendAnalyzer to: - Identify systemic issues across evaluations - Generate category-specific recommendations - Prioritize recommendations by impact - Provide examples from successful test cases - Suggest specific implementation steps

Source code in uaef/analysis/recommendations.py

class RecommendationEngine:
    """
    Engine for generating actionable recommendations for agent improvement.

    Integrates with RootCauseAnalyzer, FailureClusterer, and TrendAnalyzer to:
    - Identify systemic issues across evaluations
    - Generate category-specific recommendations
    - Prioritize recommendations by impact
    - Provide examples from successful test cases
    - Suggest specific implementation steps
    """

    def __init__(
        self,
        root_cause_analyzer: Optional[RootCauseAnalyzer] = None,
        failure_clusterer: Optional[FailureClusterer] = None,
        trend_analyzer: Optional[TrendAnalyzer] = None
    ):
        """
        Initialize the recommendation engine.

        Args:
            root_cause_analyzer: RootCauseAnalyzer instance (creates default if None)
            failure_clusterer: FailureClusterer instance (optional, None if not available)
            trend_analyzer: TrendAnalyzer instance (creates default if None)
        """
        self.root_cause_analyzer = root_cause_analyzer or RootCauseAnalyzer()
        self.failure_clusterer = failure_clusterer  # May be None if sklearn not available
        self.trend_analyzer = trend_analyzer or TrendAnalyzer()

        logger.info("RecommendationEngine initialized")

    def generate_recommendations(
        self,
        evaluation_results: List[EvaluationResult],
        include_examples: bool = True
    ) -> RecommendationReport:
        """
        Generate comprehensive recommendations based on evaluation results.

        Analyzes failure patterns, clusters, and trends to provide:
        - Specific, actionable recommendations
        - Prioritization by impact
        - Examples from high-scoring test cases
        - Implementation guidance

        Args:
            evaluation_results: List of evaluation results to analyze
            include_examples: Whether to include examples from successful cases

        Returns:
            RecommendationReport with prioritized recommendations

        Raises:
            ValueError: If no evaluation results provided or analysis fails

        Example:
            >>> engine = RecommendationEngine()
            >>> report = engine.generate_recommendations(evaluation_results)
            >>> for rec in report.recommendations[:3]:
            ...     print(f"{rec.priority}: {rec.title}")
        """
        if not evaluation_results:
            raise ValueError("No evaluation results provided")

        logger.info(f"Generating recommendations for {len(evaluation_results)} evaluations")

        try:
            # Separate passed and failed evaluations
            failed_results = [r for r in evaluation_results if not r.passed]
            passed_results = [r for r in evaluation_results if r.passed]

            if not failed_results:
                logger.info("No failures found - generating optimization recommendations")
                return self._generate_optimization_recommendations(
                    evaluation_results,
                    passed_results
                )

            # Perform root cause analysis
            failure_patterns = self.root_cause_analyzer.identify_failure_patterns(
                evaluation_results
            )

            # Perform failure clustering
            failure_clusters = []
            if self.failure_clusterer:
                try:
                    failure_clusters = self.failure_clusterer.cluster_failures(
                        evaluation_results,
                        method="hierarchical"
                    )
                except Exception as e:
                    logger.warning(f"Failure clustering skipped: {e}")
            else:
                logger.info("Failure clustering skipped (clusterer not available)")

            # Perform trend analysis (if enough data points)
            trend_report = None
            if len(evaluation_results) >= 3:
                try:
                    trend_report = self.trend_analyzer.analyze_metric_trends(
                        evaluation_results
                    )
                except ValueError as e:
                    logger.warning(f"Trend analysis skipped: {e}")

            # Generate recommendations from different sources
            recommendations = []

            # From failure patterns
            recommendations.extend(
                self._generate_pattern_recommendations(failure_patterns)
            )

            # From failure clusters
            recommendations.extend(
                self._generate_cluster_recommendations(failure_clusters)
            )

            # From trend analysis
            if trend_report:
                recommendations.extend(
                    self._generate_trend_recommendations(trend_report)
                )

            # Add examples from successful cases
            if include_examples and passed_results:
                self._add_success_examples(recommendations, passed_results)

            # Deduplicate and prioritize
            recommendations = self._deduplicate_recommendations(recommendations)
            recommendations = self._prioritize_recommendations(
                recommendations,
                failed_results,
                evaluation_results
            )

            # Generate summary and metrics
            success_rate = len(passed_results) / len(evaluation_results)
            top_issues = self._identify_top_issues(
                failure_patterns,
                failure_clusters
            )
            summary = self._generate_summary(
                len(evaluation_results),
                len(failed_results),
                success_rate,
                recommendations
            )
            improvement_potential = self._estimate_improvement_potential(
                recommendations,
                failed_results
            )

            report = RecommendationReport(
                recommendations=recommendations,
                summary=summary,
                total_evaluations=len(evaluation_results),
                failed_evaluations=len(failed_results),
                success_rate=success_rate,
                top_issues=top_issues,
                improvement_potential=improvement_potential
            )

            logger.info(
                f"Generated {len(recommendations)} recommendations "
                f"(success rate: {success_rate:.1%})"
            )

            return report

        except Exception as e:
            logger.error(f"Error generating recommendations: {e}", exc_info=True)
            raise ValueError(f"Failed to generate recommendations: {e}") from e

    def _generate_pattern_recommendations(
        self,
        failure_patterns: List[FailurePattern]
    ) -> List[Recommendation]:
        """Generate recommendations from failure patterns."""
        recommendations = []

        for pattern in failure_patterns:
            # Determine category from pattern type
            category = self._categorize_pattern(pattern.pattern_type)

            # Map severity to priority
            priority_map = {
                "critical": "critical",
                "high": "high",
                "medium": "medium",
                "low": "low"
            }
            priority = priority_map.get(pattern.severity, "medium")

            # Generate specific recommendation based on category
            rec = self._create_category_recommendation(
                category=category,
                priority=priority,
                pattern=pattern
            )

            if rec:
                recommendations.append(rec)

        return recommendations

    def _generate_cluster_recommendations(
        self,
        failure_clusters: List[FailureCluster]
    ) -> List[Recommendation]:
        """Generate recommendations from failure clusters."""
        recommendations = []

        for cluster in failure_clusters:
            # Skip small clusters
            if cluster.size < 3:
                continue

            # Determine category from common metrics
            category = self._categorize_metrics(cluster.common_metrics)

            # Map severity to priority
            priority_map = {
                "critical": "critical",
                "high": "high",
                "medium": "medium",
                "low": "low"
            }
            priority = priority_map.get(cluster.severity, "medium")

            # Create recommendation
            title = f"Address {category} issues affecting {cluster.size} test cases"

            description = (
                f"A cluster of {cluster.size} similar failures was identified. "
                f"{cluster.summary}. "
                f"Common failures: {', '.join(cluster.common_failures[:2])}"
            )

            rationale = (
                f"This pattern affects {cluster.size} test cases with {cluster.severity} "
                f"severity. Addressing this cluster could significantly improve overall "
                f"performance."
            )

            impact = self._estimate_cluster_impact(cluster)

            implementation_steps = self._generate_implementation_steps(
                category,
                cluster.common_metrics
            )

            rec = Recommendation(
                category=category,
                priority=priority,
                title=title,
                description=description,
                rationale=rationale,
                impact=impact,
                examples=[],
                affected_metrics=cluster.common_metrics,
                implementation_steps=implementation_steps
            )

            recommendations.append(rec)

        return recommendations

    def _generate_trend_recommendations(
        self,
        trend_report: TrendReport
    ) -> List[Recommendation]:
        """Generate recommendations from trend analysis."""
        recommendations = []

        # Focus on declining trends
        for metric_name, trend in trend_report.metric_trends.items():
            if trend.trend_direction == "declining":
                category = self._categorize_metric_name(metric_name)

                # Determine priority based on decline severity
                if trend.change_percentage < -20:
                    priority = "critical"
                elif trend.change_percentage < -10:
                    priority = "high"
                else:
                    priority = "medium"

                title = f"Address declining {metric_name} metric"

                description = (
                    f"The {metric_name} metric has been declining over time. "
                    f"Current trend shows {abs(trend.change_percentage):.1f}% decrease "
                    f"from initial value of {trend.values[0]:.3f} to {trend.values[-1]:.3f}."
                )

                rationale = (
                    f"Declining trend detected with slope {trend.slope:.4f}. "
                    f"This indicates a systematic issue that needs attention."
                )

                impact = (
                    f"Reversing this trend could improve {metric_name} by "
                    f"~{abs(trend.change_percentage):.1f}%"
                )

                implementation_steps = self._generate_implementation_steps(
                    category,
                    [metric_name]
                )

                rec = Recommendation(
                    category=category,
                    priority=priority,
                    title=title,
                    description=description,
                    rationale=rationale,
                    impact=impact,
                    examples=[],
                    affected_metrics=[metric_name],
                    implementation_steps=implementation_steps
                )

                recommendations.append(rec)

        # Add recommendations for critical anomalies
        for anomaly in trend_report.anomalies:
            if anomaly.severity in ["critical", "high"]:
                category = self._categorize_metric_name(anomaly.metric_name)

                title = f"Investigate {anomaly.metric_name} anomaly"

                description = (
                    f"An anomaly was detected in {anomaly.metric_name}: {anomaly.description}"
                )

                rationale = (
                    f"This anomaly has {anomaly.severity} severity and may indicate "
                    f"a significant issue that needs immediate attention."
                )

                impact = "Resolving this anomaly could prevent further degradation"

                rec = Recommendation(
                    category=category,
                    priority=anomaly.severity,
                    title=title,
                    description=description,
                    rationale=rationale,
                    impact=impact,
                    examples=[],
                    affected_metrics=[anomaly.metric_name],
                    implementation_steps=[
                        "Investigate root cause of anomaly",
                        "Review changes made around anomaly timestamp",
                        "Implement fixes and monitor metric recovery"
                    ]
                )

                recommendations.append(rec)

        return recommendations

    def _add_success_examples(
        self,
        recommendations: List[Recommendation],
        passed_results: List[EvaluationResult]
    ) -> None:
        """Add examples from successful test cases to recommendations."""
        if not passed_results:
            return

        # Get high-scoring results (top 20%)
        sorted_results = sorted(
            passed_results,
            key=lambda r: r.overall_score,
            reverse=True
        )
        top_results = sorted_results[:max(1, len(sorted_results) // 5)]

        # Extract patterns from successful cases
        for rec in recommendations:
            examples = []

            # Find successful cases with high scores in affected metrics
            for result in top_results:
                for dim_result in result.dimension_results:
                    for metric_score in dim_result.scored_metrics():
                        if metric_score.metric_name in rec.affected_metrics:
                            if metric_score.score >= 0.9:
                                example = (
                                    f"High-scoring case (score: {metric_score.score:.2f}) "
                                    f"for {metric_score.metric_name}"
                                )
                                if metric_score.reasoning:
                                    example += f": {metric_score.reasoning[:100]}"
                                examples.append(example)
                                break

            # Add unique examples (limit to 3)
            rec.examples = list(set(examples))[:3]

    def _deduplicate_recommendations(
        self,
        recommendations: List[Recommendation]
    ) -> List[Recommendation]:
        """Remove duplicate recommendations based on title similarity."""
        seen_titles = set()
        unique_recs = []

        for rec in recommendations:
            # Normalize title for comparison
            normalized_title = rec.title.lower().strip()

            if normalized_title not in seen_titles:
                seen_titles.add(normalized_title)
                unique_recs.append(rec)

        return unique_recs

    def _prioritize_recommendations(
        self,
        recommendations: List[Recommendation],
        failed_results: List[EvaluationResult],
        all_results: List[EvaluationResult]
    ) -> List[Recommendation]:
        """Prioritize recommendations by impact and frequency."""
        # Calculate impact scores
        for rec in recommendations:
            # Count how many failures this recommendation could address
            affected_count = 0
            for result in failed_results:
                for dim_result in result.dimension_results:
                    for metric_score in dim_result.scored_metrics():
                        if metric_score.metric_name in rec.affected_metrics:
                            if metric_score.score < 0.7:
                                affected_count += 1
                                break

            # Calculate impact percentage
            impact_percentage = (affected_count / len(all_results)) * 100

            # Update impact description
            rec.impact = (
                f"{rec.impact}. "
                f"Could address {affected_count} failures ({impact_percentage:.1f}% of evaluations)"
            )

        # Sort by priority and impact
        priority_order = {"critical": 0, "high": 1, "medium": 2, "low": 3}
        recommendations.sort(
            key=lambda r: (
                priority_order.get(r.priority, 3),
                -len(r.affected_metrics)
            )
        )

        return recommendations

    def _categorize_pattern(self, pattern_type: str) -> str:
        """Categorize a failure pattern."""
        pattern_lower = pattern_type.lower()

        if "tool" in pattern_lower:
            return "tool"
        elif "reasoning" in pattern_lower or "logic" in pattern_lower:
            return "reasoning"
        elif "context" in pattern_lower:
            return "context"
        elif "response" in pattern_lower:
            return "prompt"
        elif "safety" in pattern_lower or "bias" in pattern_lower:
            return "safety"
        else:
            return "architecture"

    def _categorize_metrics(self, metrics: List[str]) -> str:
        """Categorize based on metric names."""
        if not metrics:
            return "architecture"

        # Count metrics by category
        categories = Counter()
        for metric in metrics:
            category = self._categorize_metric_name(metric)
            categories[category] += 1

        # Return most common category
        return categories.most_common(1)[0][0]

    def _categorize_metric_name(self, metric_name: str) -> str:
        """Categorize a single metric name."""
        metric_lower = metric_name.lower()

        if any(x in metric_lower for x in ["tool", "parameter"]):
            return "tool"
        elif any(x in metric_lower for x in ["reasoning", "logic", "coherence", "fallacy"]):
            return "reasoning"
        elif any(x in metric_lower for x in ["context", "retention"]):
            return "context"
        elif any(x in metric_lower for x in ["relevance", "completeness", "hallucination", "answer"]):
            return "prompt"
        elif any(x in metric_lower for x in ["safety", "bias", "toxicity"]):
            return "safety"
        else:
            return "architecture"

    def _create_category_recommendation(
        self,
        category: str,
        priority: str,
        pattern: FailurePattern
    ) -> Optional[Recommendation]:
        """Create a recommendation for a specific category."""
        # Category-specific templates
        templates = {
            "tool": {
                "title": "Improve tool selection and usage",
                "description": (
                    f"Agent shows consistent issues with tool selection. "
                    f"{pattern.description}. "
                    f"This affects {pattern.frequency} test cases."
                ),
                "steps": [
                    "Review and clarify tool descriptions in system prompt",
                    "Add 3-5 examples of correct tool usage for each tool",
                    "Consider simplifying tool set by combining similar tools",
                    "Add tool selection validation logic"
                ]
            },
            "reasoning": {
                "title": "Enhance reasoning capabilities",
                "description": (
                    f"Agent demonstrates logical inconsistencies and reasoning errors. "
                    f"{pattern.description}. "
                    f"This affects {pattern.frequency} test cases."
                ),
                "steps": [
                    "Implement structured reasoning framework (e.g., Chain-of-Thought)",
                    "Add reasoning validation checkpoints in agent workflow",
                    "Include diverse reasoning examples in system prompt",
                    "Consider using reasoning-optimized model"
                ]
            },
            "context": {
                "title": "Improve context management",
                "description": (
                    f"Agent fails to properly retain and use context information. "
                    f"{pattern.description}. "
                    f"This affects {pattern.frequency} test cases."
                ),
                "steps": [
                    "Implement explicit context summarization between turns",
                    "Add context retrieval mechanisms for long conversations",
                    "Optimize context window usage and prioritization",
                    "Add context validation checks"
                ]
            },
            "prompt": {
                "title": "Enhance response generation",
                "description": (
                    f"Agent produces low-quality or irrelevant responses. "
                    f"{pattern.description}. "
                    f"This affects {pattern.frequency} test cases."
                ),
                "steps": [
                    "Refine system prompt with quality guidelines",
                    "Add response quality templates and examples",
                    "Implement response validation before returning",
                    "Add post-processing for response refinement"
                ]
            },
            "safety": {
                "title": "Address safety and bias concerns",
                "description": (
                    f"Agent shows safety or bias issues. "
                    f"{pattern.description}. "
                    f"This affects {pattern.frequency} test cases."
                ),
                "steps": [
                    "Add explicit safety guidelines to system prompt",
                    "Implement content filtering and validation",
                    "Add bias detection and mitigation strategies",
                    "Review and update safety training examples"
                ]
            },
            "architecture": {
                "title": "Consider architecture improvements",
                "description": (
                    f"Multiple failure types suggest architectural issues. "
                    f"{pattern.description}. "
                    f"This affects {pattern.frequency} test cases."
                ),
                "steps": [
                    "Review overall agent architecture and workflow",
                    "Consider adding validation layers between components",
                    "Evaluate alternative agent frameworks or patterns",
                    "Implement comprehensive error handling"
                ]
            }
        }

        template = templates.get(category)
        if not template:
            return None

        rationale = (
            f"This pattern occurs in {pattern.frequency} test cases with {pattern.severity} "
            f"severity. Addressing it is a high-impact improvement opportunity."
        )

        impact = (
            f"Expected to improve {', '.join(pattern.affected_metrics[:3])} metrics. "
            f"Could resolve {pattern.frequency} failures."
        )

        return Recommendation(
            category=category,
            priority=priority,
            title=template["title"],
            description=template["description"],
            rationale=rationale,
            impact=impact,
            examples=[],
            affected_metrics=pattern.affected_metrics,
            implementation_steps=template["steps"]
        )

    def _generate_implementation_steps(
        self,
        category: str,
        metrics: List[str]
    ) -> List[str]:
        """Generate implementation steps for a category."""
        base_steps = {
            "tool": [
                "Audit current tool set and descriptions",
                "Add clear examples of tool usage to prompt",
                "Implement tool selection validation",
                "Test with diverse tool usage scenarios"
            ],
            "reasoning": [
                "Add structured reasoning framework to prompt",
                "Include reasoning validation checkpoints",
                "Add diverse reasoning examples",
                "Monitor reasoning quality metrics"
            ],
            "context": [
                "Implement context summarization strategy",
                "Add context retrieval mechanisms",
                "Optimize context window management",
                "Test with long conversation scenarios"
            ],
            "prompt": [
                "Review and refine system prompt",
                "Add quality guidelines and examples",
                "Implement response validation",
                "Test with diverse query types"
            ],
            "safety": [
                "Add safety guidelines to system prompt",
                "Implement content filtering",
                "Add bias detection mechanisms",
                "Review with safety experts"
            ],
            "architecture": [
                "Review overall agent design",
                "Identify architectural bottlenecks",
                "Implement validation layers",
                "Consider alternative patterns"
            ]
        }

        return base_steps.get(category, [
            "Analyze root cause in detail",
            "Design and implement solution",
            "Test thoroughly",
            "Monitor metrics post-deployment"
        ])

    def _estimate_cluster_impact(self, cluster: FailureCluster) -> str:
        """Estimate the impact of addressing a failure cluster."""
        percentage = (cluster.size / 100) * 100  # Assuming ~100 test cases

        return (
            f"Addressing this cluster could improve success rate by ~{min(percentage, 20):.1f}%. "
            f"Affects {cluster.size} test cases with {cluster.severity} severity."
        )

    def _identify_top_issues(
        self,
        failure_patterns: List[FailurePattern],
        failure_clusters: List[FailureCluster]
    ) -> List[str]:
        """Identify top 3 most common issues."""
        issues = []

        # Add top patterns
        for pattern in sorted(failure_patterns, key=lambda p: p.frequency, reverse=True)[:2]:
            issues.append(
                f"{pattern.pattern_type.replace('_', ' ').title()} ({pattern.frequency} cases)"
            )

        # Add largest cluster if not already covered
        if failure_clusters:
            largest_cluster = max(failure_clusters, key=lambda c: c.size)
            cluster_desc = f"{largest_cluster.summary.split(':')[0]} ({largest_cluster.size} cases)"
            if cluster_desc not in issues:
                issues.append(cluster_desc)

        return issues[:3]

    def _generate_summary(
        self,
        total: int,
        failed: int,
        success_rate: float,
        recommendations: List[Recommendation]
    ) -> str:
        """Generate high-level summary."""
        critical_count = sum(1 for r in recommendations if r.priority == "critical")
        high_count = sum(1 for r in recommendations if r.priority == "high")

        summary_parts = [
            f"Analyzed {total} evaluations with {failed} failures ({success_rate:.1%} success rate)"
        ]

        if critical_count > 0:
            summary_parts.append(f"{critical_count} critical recommendations")
        if high_count > 0:
            summary_parts.append(f"{high_count} high-priority recommendations")

        if recommendations:
            summary_parts.append(
                f"Top recommendation: {recommendations[0].title}"
            )

        return ". ".join(summary_parts) + "."

    def _estimate_improvement_potential(
        self,
        recommendations: List[Recommendation],
        failed_results: List[EvaluationResult]
    ) -> str:
        """Estimate improvement potential if recommendations are implemented."""
        if not recommendations or not failed_results:
            return "No improvement potential estimated"

        # Count high-priority recommendations
        high_priority = sum(
            1 for r in recommendations
            if r.priority in ["critical", "high"]
        )

        # Estimate based on number of high-priority recommendations
        if high_priority >= 3:
            improvement = "15-25%"
        elif high_priority >= 2:
            improvement = "10-15%"
        else:
            improvement = "5-10%"

        return (
            f"Implementing top {min(3, len(recommendations))} recommendations "
            f"could improve success rate by {improvement}"
        )

    def _generate_optimization_recommendations(
        self,
        all_results: List[EvaluationResult],
        passed_results: List[EvaluationResult]
    ) -> RecommendationReport:
        """Generate optimization recommendations when there are no failures."""
        recommendations = []

        # Find metrics with room for improvement (< 0.95)
        metric_scores = defaultdict(list)
        for result in passed_results:
            for dim_result in result.dimension_results:
                for metric_score in dim_result.scored_metrics():
                    metric_scores[metric_score.metric_name].append(metric_score.score)

        # Identify metrics with lowest average scores
        avg_scores = {
            metric: sum(scores) / len(scores)
            for metric, scores in metric_scores.items()
            if scores
        }

        # Sort by score (lowest first)
        sorted_metrics = sorted(avg_scores.items(), key=lambda x: x[1])

        # Generate optimization recommendations for lowest-scoring metrics
        for metric_name, avg_score in sorted_metrics[:3]:
            if avg_score < 0.95:
                category = self._categorize_metric_name(metric_name)

                rec = Recommendation(
                    category=category,
                    priority="low",
                    title=f"Optimize {metric_name} performance",
                    description=(
                        f"While {metric_name} is passing (avg: {avg_score:.3f}), "
                        f"there is room for optimization to reach excellence (>0.95)."
                    ),
                    rationale=(
                        f"Current average score of {avg_score:.3f} indicates good but "
                        f"not excellent performance. Optimization could push this metric "
                        f"to excellence."
                    ),
                    impact=f"Could improve {metric_name} by {(0.95 - avg_score):.2f} points",
                    examples=[],
                    affected_metrics=[metric_name],
                    implementation_steps=self._generate_implementation_steps(
                        category,
                        [metric_name]
                    )
                )
                recommendations.append(rec)

        summary = (
            f"Analyzed {len(all_results)} evaluations with 100% success rate. "
            f"Generated {len(recommendations)} optimization recommendations for excellence."
        )

        return RecommendationReport(
            recommendations=recommendations,
            summary=summary,
            total_evaluations=len(all_results),
            failed_evaluations=0,
            success_rate=1.0,
            top_issues=["No failures detected - focus on optimization"],
            improvement_potential=(
                "Agent is performing well. Optimization recommendations focus on "
                "achieving excellence (>0.95) across all metrics."
            )
        )

generate_recommendations(evaluation_results, include_examples=True)

Generate comprehensive recommendations based on evaluation results.

Analyzes failure patterns, clusters, and trends to provide: - Specific, actionable recommendations - Prioritization by impact - Examples from high-scoring test cases - Implementation guidance

Parameters:

Name	Type	Description	Default
evaluation_results	List[EvaluationResult]	List of evaluation results to analyze	required
include_examples	bool	Whether to include examples from successful cases	True

Returns:

Type	Description
RecommendationReport	RecommendationReport with prioritized recommendations

Raises:

Type	Description
ValueError	If no evaluation results provided or analysis fails

Example

engine = RecommendationEngine() report = engine.generate_recommendations(evaluation_results) for rec in report.recommendations[:3]: ... print(f"{rec.priority}: {rec.title}")

Source code in uaef/analysis/recommendations.py

def generate_recommendations(
    self,
    evaluation_results: List[EvaluationResult],
    include_examples: bool = True
) -> RecommendationReport:
    """
    Generate comprehensive recommendations based on evaluation results.

    Analyzes failure patterns, clusters, and trends to provide:
    - Specific, actionable recommendations
    - Prioritization by impact
    - Examples from high-scoring test cases
    - Implementation guidance

    Args:
        evaluation_results: List of evaluation results to analyze
        include_examples: Whether to include examples from successful cases

    Returns:
        RecommendationReport with prioritized recommendations

    Raises:
        ValueError: If no evaluation results provided or analysis fails

    Example:
        >>> engine = RecommendationEngine()
        >>> report = engine.generate_recommendations(evaluation_results)
        >>> for rec in report.recommendations[:3]:
        ...     print(f"{rec.priority}: {rec.title}")
    """
    if not evaluation_results:
        raise ValueError("No evaluation results provided")

    logger.info(f"Generating recommendations for {len(evaluation_results)} evaluations")

    try:
        # Separate passed and failed evaluations
        failed_results = [r for r in evaluation_results if not r.passed]
        passed_results = [r for r in evaluation_results if r.passed]

        if not failed_results:
            logger.info("No failures found - generating optimization recommendations")
            return self._generate_optimization_recommendations(
                evaluation_results,
                passed_results
            )

        # Perform root cause analysis
        failure_patterns = self.root_cause_analyzer.identify_failure_patterns(
            evaluation_results
        )

        # Perform failure clustering
        failure_clusters = []
        if self.failure_clusterer:
            try:
                failure_clusters = self.failure_clusterer.cluster_failures(
                    evaluation_results,
                    method="hierarchical"
                )
            except Exception as e:
                logger.warning(f"Failure clustering skipped: {e}")
        else:
            logger.info("Failure clustering skipped (clusterer not available)")

        # Perform trend analysis (if enough data points)
        trend_report = None
        if len(evaluation_results) >= 3:
            try:
                trend_report = self.trend_analyzer.analyze_metric_trends(
                    evaluation_results
                )
            except ValueError as e:
                logger.warning(f"Trend analysis skipped: {e}")

        # Generate recommendations from different sources
        recommendations = []

        # From failure patterns
        recommendations.extend(
            self._generate_pattern_recommendations(failure_patterns)
        )

        # From failure clusters
        recommendations.extend(
            self._generate_cluster_recommendations(failure_clusters)
        )

        # From trend analysis
        if trend_report:
            recommendations.extend(
                self._generate_trend_recommendations(trend_report)
            )

        # Add examples from successful cases
        if include_examples and passed_results:
            self._add_success_examples(recommendations, passed_results)

        # Deduplicate and prioritize
        recommendations = self._deduplicate_recommendations(recommendations)
        recommendations = self._prioritize_recommendations(
            recommendations,
            failed_results,
            evaluation_results
        )

        # Generate summary and metrics
        success_rate = len(passed_results) / len(evaluation_results)
        top_issues = self._identify_top_issues(
            failure_patterns,
            failure_clusters
        )
        summary = self._generate_summary(
            len(evaluation_results),
            len(failed_results),
            success_rate,
            recommendations
        )
        improvement_potential = self._estimate_improvement_potential(
            recommendations,
            failed_results
        )

        report = RecommendationReport(
            recommendations=recommendations,
            summary=summary,
            total_evaluations=len(evaluation_results),
            failed_evaluations=len(failed_results),
            success_rate=success_rate,
            top_issues=top_issues,
            improvement_potential=improvement_potential
        )

        logger.info(
            f"Generated {len(recommendations)} recommendations "
            f"(success rate: {success_rate:.1%})"
        )

        return report

    except Exception as e:
        logger.error(f"Error generating recommendations: {e}", exc_info=True)
        raise ValueError(f"Failed to generate recommendations: {e}") from e

RecommendationReport

Bases: BaseModel

Comprehensive recommendation report for agent improvement.

Attributes:

Name	Type	Description
recommendations	List[Recommendation]	List of recommendations sorted by priority
summary	str	High-level summary of key findings
total_evaluations	int	Total number of evaluations analyzed
failed_evaluations	int	Number of failed evaluations
success_rate	float	Overall success rate (0-1)
top_issues	List[str]	Top 3 most common issues identified
improvement_potential	str	Estimated improvement potential if recommendations implemented

Source code in uaef/analysis/recommendations.py

class RecommendationReport(BaseModel):
    """
    Comprehensive recommendation report for agent improvement.

    Attributes:
        recommendations: List of recommendations sorted by priority
        summary: High-level summary of key findings
        total_evaluations: Total number of evaluations analyzed
        failed_evaluations: Number of failed evaluations
        success_rate: Overall success rate (0-1)
        top_issues: Top 3 most common issues identified
        improvement_potential: Estimated improvement potential if recommendations implemented
    """

    recommendations: List[Recommendation] = Field(
        default_factory=list,
        description="List of recommendations sorted by priority"
    )
    summary: str = Field(..., description="High-level summary of key findings")
    total_evaluations: int = Field(..., description="Total number of evaluations analyzed")
    failed_evaluations: int = Field(..., description="Number of failed evaluations")
    success_rate: float = Field(..., description="Overall success rate", ge=0.0, le=1.0)
    top_issues: List[str] = Field(
        default_factory=list,
        description="Top 3 most common issues identified"
    )
    improvement_potential: str = Field(
        ...,
        description="Estimated improvement potential if recommendations implemented"
    )

    class Config:
        """Pydantic configuration."""
        json_schema_extra = {
            "example": {
                "recommendations": [],
                "summary": "Analyzed 100 evaluations with 35% failure rate",
                "total_evaluations": 100,
                "failed_evaluations": 35,
                "success_rate": 0.65,
                "top_issues": [
                    "Tool selection errors (15 cases)",
                    "Context retention failures (12 cases)",
                    "Response quality issues (8 cases)"
                ],
                "improvement_potential": "Implementing top 3 recommendations could improve success rate to ~80%"
            }
        }

Config

Pydantic configuration.

Source code in uaef/analysis/recommendations.py

class Config:
    """Pydantic configuration."""
    json_schema_extra = {
        "example": {
            "recommendations": [],
            "summary": "Analyzed 100 evaluations with 35% failure rate",
            "total_evaluations": 100,
            "failed_evaluations": 35,
            "success_rate": 0.65,
            "top_issues": [
                "Tool selection errors (15 cases)",
                "Context retention failures (12 cases)",
                "Response quality issues (8 cases)"
            ],
            "improvement_potential": "Implementing top 3 recommendations could improve success rate to ~80%"
        }
    }

uaef.insights.generator

LLM-powered insight generation for evaluation reports.

Reads a report.md file, sends it to Bedrock, and writes insights.md to the same report directory.

generate_insights(report_dir, *, report_type=None, model_id=None)

Generate LLM-powered insights for an existing report.

Reads report.md from the given report directory, sends it to Bedrock for analysis, and writes insights.md to the same directory.

Parameters:

Name	Type	Description	Default
report_dir	str	Path to the report directory (must contain report.md)	required
report_type	Optional[str]	"single" or "comparison" (auto-detected if None)	None
model_id	Optional[str]	Override the model ID from config (optional)	None

Returns:

Type	Description
str	Path to the generated insights.md file

Examples:

>>> from uaef.insights import generate_insights
>>> insights_path = generate_insights("output/reports/full_report_20260602_125549")

Source code in uaef/insights/generator.py

def generate_insights(
    report_dir: str,
    *,
    report_type: Optional[str] = None,
    model_id: Optional[str] = None,
) -> str:
    """
    Generate LLM-powered insights for an existing report.

    Reads report.md from the given report directory, sends it to Bedrock
    for analysis, and writes insights.md to the same directory.

    Args:
        report_dir: Path to the report directory (must contain report.md)
        report_type: "single" or "comparison" (auto-detected if None)
        model_id: Override the model ID from config (optional)

    Returns:
        Path to the generated insights.md file

    Examples:
        >>> from uaef.insights import generate_insights
        >>> insights_path = generate_insights("output/reports/full_report_20260602_125549")
    """
    report_dir = Path(report_dir)
    report_path = report_dir / "report.md"

    if not report_path.exists():
        raise FileNotFoundError(f"No report.md found in {report_dir}")

    report_content = report_path.read_text(encoding="utf-8")

    # Auto-detect report type
    if report_type is None:
        if "Comparison Report" in report_content or "Changes from Baseline" in report_content:
            report_type = "comparison"
        else:
            report_type = "single"

    logger.info(f"Generating {report_type} insights for {report_dir}")

    # Select prompt
    if report_type == "comparison":
        prompt = COMPARISON_PROMPT.format(report_content=report_content)
    else:
        prompt = SINGLE_RUN_PROMPT.format(report_content=report_content)

    # Call Bedrock
    config = get_config()
    insight_config = config.insight
    effective_model = model_id or insight_config.model_id

    client = BedrockClient(
        region=config.aws.region,
        retry_attempts=insight_config.retry_attempts,
        retry_delay=insight_config.retry_delay,
    )

    response_text, _metadata = client.invoke_model(
        model_id=effective_model,
        prompt=prompt,
        temperature=insight_config.temperature,
        max_tokens=insight_config.max_tokens,
    )

    # Write insights.md
    insights_path = report_dir / "insights.md"
    header = f"# Insights ({report_type.title()} Analysis)\n\n"
    header += f"*Generated by LLM ({effective_model})*\n\n---\n\n"
    insights_path.write_text(header + response_text, encoding="utf-8")

    logger.info(f"Insights saved to {insights_path}")
    return str(insights_path)