Evaluation Flywheel

Build a continuous improvement cycle using evaluations to drive model quality over time.

Overview

The evaluation flywheel is a systematic approach to continuously improving AI system quality through iterative measurement, analysis, and optimization.

┌─────────────────────────────────────────────────────────────────────┐
│                    THE EVALUATION FLYWHEEL                           │
├─────────────────────────────────────────────────────────────────────┤
│                                                                      │
│                         ┌───────────┐                               │
│                         │  DEPLOY   │                               │
│                         │  MODEL    │                               │
│                         └─────┬─────┘                               │
│                               │                                      │
│                               ▼                                      │
│        ┌───────────┐    ┌───────────┐                               │
│        │  ITERATE  │◄───│  COLLECT  │                               │
│        │  & IMPROVE│    │  DATA     │                               │
│        └─────┬─────┘    └─────┬─────┘                               │
│              │                │                                      │
│              ▼                ▼                                      │
│        ┌───────────┐    ┌───────────┐                               │
│        │  ANALYZE  │◄───│ EVALUATE  │                               │
│        │  RESULTS  │    │  QUALITY  │                               │
│        └───────────┘    └───────────┘                               │
│                                                                      │
│  Each rotation:                                                      │
│  ├─► Improves model quality                                         │
│  ├─► Builds evaluation dataset                                      │
│  ├─► Refines metrics                                                │
│  └─► Accelerates future iterations                                  │
│                                                                      │
└─────────────────────────────────────────────────────────────────────┘

Flywheel Components

1. Evaluation Framework

from dataclasses import dataclass
from typing import Callable, Any
from enum import Enum

class MetricType(Enum):
    BINARY = "binary"          # Pass/fail
    SCORE = "score"            # 0-1 continuous
    CATEGORICAL = "categorical"  # Multi-class

@dataclass
class Metric:
    name: str
    type: MetricType
    evaluator: Callable[[str, str], float]
    threshold: float
    weight: float = 1.0

@dataclass
class EvaluationResult:
    example_id: str
    input_text: str
    output_text: str
    expected: str
    metrics: dict[str, float]
    passed: bool
    feedback: str = ""

class EvaluationFramework:
    def __init__(self, metrics: list[Metric]):
        self.metrics = {m.name: m for m in metrics}
        self.history: list[EvaluationResult] = []

    def evaluate(
        self,
        example_id: str,
        input_text: str,
        output_text: str,
        expected: str = None
    ) -> EvaluationResult:
        """Evaluate a single output."""

        metric_scores = {}
        weighted_sum = 0
        total_weight = 0

        for name, metric in self.metrics.items():
            score = metric.evaluator(output_text, expected or input_text)
            metric_scores[name] = score
            weighted_sum += score * metric.weight
            total_weight += metric.weight

        overall_score = weighted_sum / total_weight if total_weight > 0 else 0

        result = EvaluationResult(
            example_id=example_id,
            input_text=input_text,
            output_text=output_text,
            expected=expected,
            metrics=metric_scores,
            passed=overall_score >= 0.8  # Configurable threshold
        )

        self.history.append(result)
        return result

    def get_summary(self) -> dict:
        """Get evaluation summary statistics."""

        if not self.history:
            return {}

        passed_count = sum(1 for r in self.history if r.passed)

        metric_avgs = {}
        for name in self.metrics:
            scores = [r.metrics.get(name, 0) for r in self.history]
            metric_avgs[name] = sum(scores) / len(scores)

        return {
            "total_evaluated": len(self.history),
            "passed": passed_count,
            "pass_rate": passed_count / len(self.history),
            "metric_averages": metric_avgs
        }

2. Data Collection Pipeline

import json
from datetime import datetime

class DataCollector:
    def __init__(self, storage_path: str):
        self.storage_path = storage_path
        self.collected = []

    def log_interaction(
        self,
        input_text: str,
        output_text: str,
        user_feedback: str = None,
        metadata: dict = None
    ):
        """Log a production interaction."""

        entry = {
            "timestamp": datetime.now().isoformat(),
            "input": input_text,
            "output": output_text,
            "feedback": user_feedback,
            "metadata": metadata or {}
        }

        self.collected.append(entry)

        # Persist immediately
        with open(self.storage_path, "a") as f:
            f.write(json.dumps(entry) + "\n")

    def get_for_evaluation(self, n: int = 100) -> list[dict]:
        """Get recent interactions for evaluation."""

        # Load from storage
        entries = []
        with open(self.storage_path) as f:
            for line in f:
                if line.strip():
                    entries.append(json.loads(line))

        # Return most recent n
        return entries[-n:]

    def get_negative_examples(self) -> list[dict]:
        """Get examples with negative feedback for analysis."""

        negative = []
        for entry in self.collected:
            if entry.get("feedback") in ["bad", "incorrect", "unhelpful"]:
                negative.append(entry)
        return negative

3. Analysis Engine

from collections import Counter
import numpy as np

class AnalysisEngine:
    def __init__(self, framework: EvaluationFramework):
        self.framework = framework

    def identify_failure_patterns(self) -> dict:
        """Identify common failure patterns."""

        failed = [r for r in self.framework.history if not r.passed]

        # Analyze which metrics failed most
        metric_failures = Counter()
        for result in failed:
            for metric_name, score in result.metrics.items():
                threshold = self.framework.metrics[metric_name].threshold
                if score < threshold:
                    metric_failures[metric_name] += 1

        # Analyze input characteristics of failures
        failed_input_lengths = [len(r.input_text.split()) for r in failed]
        passed_input_lengths = [len(r.input_text.split()) for r in self.framework.history if r.passed]

        return {
            "total_failures": len(failed),
            "failure_rate": len(failed) / max(len(self.framework.history), 1),
            "metrics_causing_failures": dict(metric_failures.most_common()),
            "avg_failed_input_length": np.mean(failed_input_lengths) if failed_input_lengths else 0,
            "avg_passed_input_length": np.mean(passed_input_lengths) if passed_input_lengths else 0,
        }

    def suggest_improvements(self) -> list[str]:
        """Generate improvement suggestions based on analysis."""

        patterns = self.identify_failure_patterns()
        suggestions = []

        # Metric-based suggestions
        for metric, count in patterns["metrics_causing_failures"].items():
            if count > len(self.framework.history) * 0.2:  # >20% failures
                suggestions.append(f"Focus on improving '{metric}' - causing {count} failures")

        # Length-based suggestions
        if patterns["avg_failed_input_length"] > patterns["avg_passed_input_length"] * 1.5:
            suggestions.append("Model struggles with longer inputs - consider chunking or summarization")

        if patterns["failure_rate"] > 0.3:
            suggestions.append("High failure rate (>30%) - consider prompt optimization or fine-tuning")

        return suggestions

    def generate_training_data(self, n: int = 100) -> list[dict]:
        """Generate training data from successful evaluations."""

        passed = [r for r in self.framework.history if r.passed]

        training_data = []
        for result in passed[:n]:
            training_data.append({
                "messages": [
                    {"role": "user", "content": result.input_text},
                    {"role": "assistant", "content": result.output_text}
                ]
            })

        return training_data

4. Iteration Manager

from datetime import datetime
from typing import Optional

@dataclass
class IterationRecord:
    iteration_id: str
    timestamp: datetime
    model_version: str
    metrics: dict
    changes_made: list[str]
    improvement: float

class IterationManager:
    def __init__(self):
        self.iterations: list[IterationRecord] = []
        self.current_baseline: Optional[dict] = None

    def start_iteration(self, model_version: str) -> str:
        """Start a new improvement iteration."""

        iteration_id = f"iter_{len(self.iterations)+1}_{datetime.now().strftime('%Y%m%d')}"
        return iteration_id

    def record_iteration(
        self,
        iteration_id: str,
        model_version: str,
        metrics: dict,
        changes: list[str]
    ):
        """Record iteration results."""

        # Calculate improvement over baseline
        improvement = 0
        if self.current_baseline:
            baseline_score = self.current_baseline.get("overall_score", 0)
            current_score = metrics.get("overall_score", 0)
            improvement = current_score - baseline_score

        record = IterationRecord(
            iteration_id=iteration_id,
            timestamp=datetime.now(),
            model_version=model_version,
            metrics=metrics,
            changes_made=changes,
            improvement=improvement
        )

        self.iterations.append(record)

        # Update baseline if improved
        if improvement > 0:
            self.current_baseline = metrics

    def get_progress_report(self) -> dict:
        """Generate progress report across iterations."""

        if not self.iterations:
            return {"message": "No iterations recorded"}

        first = self.iterations[0]
        last = self.iterations[-1]

        return {
            "total_iterations": len(self.iterations),
            "first_iteration": {
                "date": first.timestamp.isoformat(),
                "metrics": first.metrics
            },
            "latest_iteration": {
                "date": last.timestamp.isoformat(),
                "metrics": last.metrics
            },
            "total_improvement": sum(i.improvement for i in self.iterations),
            "best_performing_changes": self._get_best_changes()
        }

    def _get_best_changes(self) -> list[str]:
        """Identify changes that led to biggest improvements."""

        change_impacts = {}
        for iteration in self.iterations:
            if iteration.improvement > 0:
                for change in iteration.changes_made:
                    if change not in change_impacts:
                        change_impacts[change] = 0
                    change_impacts[change] += iteration.improvement

        return sorted(change_impacts.keys(), key=lambda x: change_impacts[x], reverse=True)[:5]

Complete Flywheel Implementation

class EvaluationFlywheel:
    """Complete evaluation flywheel system."""

    def __init__(
        self,
        metrics: list[Metric],
        storage_path: str,
        model_name: str
    ):
        self.framework = EvaluationFramework(metrics)
        self.collector = DataCollector(storage_path)
        self.analyzer = AnalysisEngine(self.framework)
        self.iteration_manager = IterationManager()
        self.model_name = model_name

    def run_evaluation_cycle(self, test_data: list[dict]) -> dict:
        """Run a complete evaluation cycle."""

        # Start iteration
        iteration_id = self.iteration_manager.start_iteration(self.model_name)

        # Evaluate all test data
        for item in test_data:
            self.framework.evaluate(
                example_id=item.get("id", ""),
                input_text=item["input"],
                output_text=item["output"],
                expected=item.get("expected")
            )

        # Get summary
        summary = self.framework.get_summary()

        # Analyze failures
        analysis = self.analyzer.identify_failure_patterns()

        # Get suggestions
        suggestions = self.analyzer.suggest_improvements()

        return {
            "iteration_id": iteration_id,
            "summary": summary,
            "analysis": analysis,
            "suggestions": suggestions
        }

    def apply_improvements(
        self,
        iteration_id: str,
        changes: list[str],
        new_metrics: dict
    ):
        """Record improvements applied."""

        self.iteration_manager.record_iteration(
            iteration_id=iteration_id,
            model_version=self.model_name,
            metrics=new_metrics,
            changes=changes
        )

    def export_training_data(self, output_path: str, n: int = 100):
        """Export high-quality examples for fine-tuning."""

        training_data = self.analyzer.generate_training_data(n)

        with open(output_path, "w") as f:
            for item in training_data:
                f.write(json.dumps(item) + "\n")

        print(f"Exported {len(training_data)} training examples to {output_path}")

    def get_report(self) -> dict:
        """Get comprehensive flywheel report."""

        return {
            "current_evaluation": self.framework.get_summary(),
            "failure_analysis": self.analyzer.identify_failure_patterns(),
            "improvement_suggestions": self.analyzer.suggest_improvements(),
            "progress": self.iteration_manager.get_progress_report()
        }

Usage Example

from openai import OpenAI

client = OpenAI()

# Define metrics
metrics = [
    Metric(
        name="accuracy",
        type=MetricType.SCORE,
        evaluator=lambda output, expected: 1.0 if expected in output else 0.0,
        threshold=0.8,
        weight=2.0
    ),
    Metric(
        name="length_appropriate",
        type=MetricType.SCORE,
        evaluator=lambda output, _: min(1.0, len(output.split()) / 100),
        threshold=0.5,
        weight=1.0
    ),
    Metric(
        name="professional_tone",
        type=MetricType.BINARY,
        evaluator=check_professional_tone,  # Your custom function
        threshold=0.8,
        weight=1.5
    )
]

# Initialize flywheel
flywheel = EvaluationFlywheel(
    metrics=metrics,
    storage_path="./evaluation_data.jsonl",
    model_name="gpt-4o"
)

# Cycle 1: Initial evaluation
test_data = load_test_data()  # Your test data
cycle_results = flywheel.run_evaluation_cycle(test_data)

print(f"Pass rate: {cycle_results['summary']['pass_rate']:.1%}")
print(f"Suggestions: {cycle_results['suggestions']}")

# Apply improvements (e.g., prompt changes)
# ... make changes ...

# Cycle 2: Re-evaluate
cycle_results_2 = flywheel.run_evaluation_cycle(test_data)

# Record improvement
flywheel.apply_improvements(
    iteration_id=cycle_results_2["iteration_id"],
    changes=["Added few-shot examples", "Clarified output format"],
    new_metrics=cycle_results_2["summary"]
)

# Get progress report
report = flywheel.get_report()
print(f"Total improvement: {report['progress']['total_improvement']:.1%}")

Integration with OpenAI Evals

def run_openai_eval_cycle(eval_id: str, test_items: list[dict]) -> dict:
    """Run evaluation cycle using OpenAI Evals API."""

    # Create eval run
    data_source = {
        "type": "jsonl",
        "source": {
            "type": "file_content",
            "content": [{"item": item} for item in test_items]
        }
    }

    run = client.evals.runs.create(
        eval_id=eval_id,
        data_source=data_source
    )

    # Poll for completion
    while run.status not in ["completed", "failed"]:
        time.sleep(10)
        run = client.evals.runs.retrieve(eval_id=eval_id, run_id=run.id)

    # Get results
    results = client.evals.runs.output_items.list(eval_id=eval_id, run_id=run.id)

    # Analyze
    passed = sum(1 for r in results.data if all(
        result.passed for result in r.results
    ))

    return {
        "total": len(results.data),
        "passed": passed,
        "pass_rate": passed / len(results.data),
        "details": results.data
    }

Best Practices

Flywheel Velocity

Practice	Impact
Daily evaluations	Fast feedback loops
Automated metrics	Consistent measurement
Tracked iterations	Clear progress visibility
Exported training data	Continuous improvement fuel

Common Pitfalls

1. Evaluating too infrequently - Run evaluations at least weekly
2. Ignoring edge cases - Actively collect and evaluate failures
3. Not tracking changes - Always document what was changed
4. Optimizing single metrics - Balance multiple quality dimensions

See Also

• SOP-003: Eval Creation
• Fine-Tuning Techniques
• Workflow: Self-Evolving Loop