"""
Pattern 268: Benchmarking Pattern with LangChain/LangGraph

Purpose: Performance measurement and comparison with statistical analysis.
Provides intelligent benchmark orchestration with LLM-powered performance insights.
"""

from typing import Dict, Any, List, Optional, Callable, TypedDict
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import time
import statistics
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
import json


class BenchmarkType(Enum):
    """Types of benchmarks."""
    EXECUTION_TIME = "execution_time"
    THROUGHPUT = "throughput"
    LATENCY = "latency"
    RESOURCE = "resource"
    COMPARISON = "comparison"
    STRESS = "stress"


class PerformanceLevel(Enum):
    """Performance classification."""
    EXCELLENT = "excellent"
    GOOD = "good"
    ACCEPTABLE = "acceptable"
    POOR = "poor"
    CRITICAL = "critical"


@dataclass
class BenchmarkResult:
    """Result of a benchmark run."""
    name: str
    benchmark_type: BenchmarkType
    iterations: int
    mean_time: float
    median_time: float
    min_time: float
    max_time: float
    std_dev: float
    throughput: float
    percentile_95: float
    percentile_99: float
    performance_level: PerformanceLevel
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class BenchmarkInsights:
    """LLM-generated benchmark insights."""
    performance_assessment: str
    bottlenecks: List[str]
    recommendations: List[str]
    comparison_summary: str


class BenchmarkState(TypedDict):
    """State for benchmark workflow."""
    benchmark_name: str
    function: Optional[Callable]
    iterations: int
    benchmark_type: BenchmarkType
    execution_times: List[float]
    mean_time: float
    median_time: float
    std_dev: float
    throughput: float
    percentile_95: float
    percentile_99: float
    performance_level: PerformanceLevel
    outliers: List[float]
    insights: Optional[BenchmarkInsights]
    comparison_data: Dict[str, Any]


class IntelligentBenchmark:
    """
    Intelligent Benchmark with LangGraph workflow integration.
    
    Features:
    - Execution time measurement
    - Statistical analysis
    - Outlier detection
    - Percentile calculation
    - LLM-powered insights
    - Performance comparison
    - Bottleneck identification
    - Optimization recommendations
    """
    
    def __init__(self, name: str, model_name: str = "gpt-4", temperature: float = 0.5):
        self.name = name
        self.llm = ChatOpenAI(model=model_name, temperature=temperature)
        
        # Results storage
        self.results: Dict[str, BenchmarkResult] = {}
        
        # Build workflow
        self.workflow = self._build_workflow()
        
        # Statistics
        self.stats = {
            "total_benchmarks": 0,
            "benchmark_types": {t.value: 0 for t in BenchmarkType},
            "total_iterations": 0,
            "avg_execution_time": 0.0
        }
    
    def _build_workflow(self) -> StateGraph:
        """Build LangGraph workflow."""
        workflow = StateGraph(BenchmarkState)
        
        workflow.add_node("initialize_benchmark", self._initialize_benchmark_node)
        workflow.add_node("run_iterations", self._run_iterations_node)
        workflow.add_node("analyze_statistics", self._analyze_statistics_node)
        workflow.add_node("detect_outliers", self._detect_outliers_node)
        workflow.add_node("generate_comparison", self._generate_comparison_node)
        
        workflow.set_entry_point("initialize_benchmark")
        workflow.add_edge("initialize_benchmark", "run_iterations")
        workflow.add_edge("run_iterations", "analyze_statistics")
        workflow.add_edge("analyze_statistics", "detect_outliers")
        workflow.add_edge("detect_outliers", "generate_comparison")
        workflow.add_edge("generate_comparison", END)
        
        return workflow.compile()
    
    def _initialize_benchmark_node(self, state: BenchmarkState) -> BenchmarkState:
        """Initialize benchmark state."""
        state["execution_times"] = []
        state["outliers"] = []
        state["comparison_data"] = {}
        return state
    
    def _run_iterations_node(self, state: BenchmarkState) -> BenchmarkState:
        """Run benchmark iterations."""
        function = state["function"]
        iterations = state["iterations"]
        
        if not function:
            return state
        
        times: List[float] = []
        
        # Warm-up run
        try:
            function()
        except:
            pass
        
        # Benchmark iterations
        for _ in range(iterations):
            start = time.perf_counter()
            try:
                function()
            except:
                pass
            end = time.perf_counter()
            times.append(end - start)
        
        state["execution_times"] = times
        return state
    
    def _analyze_statistics_node(self, state: BenchmarkState) -> BenchmarkState:
        """Analyze benchmark statistics."""
        times = state["execution_times"]
        
        if not times:
            state["mean_time"] = 0.0
            state["median_time"] = 0.0
            state["std_dev"] = 0.0
            state["throughput"] = 0.0
            state["percentile_95"] = 0.0
            state["percentile_99"] = 0.0
            state["performance_level"] = PerformanceLevel.CRITICAL
            return state
        
        # Calculate statistics
        state["mean_time"] = statistics.mean(times)
        state["median_time"] = statistics.median(times)
        state["std_dev"] = statistics.stdev(times) if len(times) > 1 else 0.0
        state["throughput"] = 1.0 / state["mean_time"] if state["mean_time"] > 0 else 0.0
        
        # Calculate percentiles
        sorted_times = sorted(times)
        p95_idx = int(len(sorted_times) * 0.95)
        p99_idx = int(len(sorted_times) * 0.99)
        state["percentile_95"] = sorted_times[min(p95_idx, len(sorted_times) - 1)]
        state["percentile_99"] = sorted_times[min(p99_idx, len(sorted_times) - 1)]
        
        # Classify performance
        mean_ms = state["mean_time"] * 1000
        if mean_ms < 10:
            state["performance_level"] = PerformanceLevel.EXCELLENT
        elif mean_ms < 50:
            state["performance_level"] = PerformanceLevel.GOOD
        elif mean_ms < 200:
            state["performance_level"] = PerformanceLevel.ACCEPTABLE
        elif mean_ms  BenchmarkState:
        """Detect outlier measurements."""
        times = state["execution_times"]
        
        if not times or len(times)  2 standard deviations)
        outliers = []
        for t in times:
            if abs(t - mean) > 2 * std_dev:
                outliers.append(t)
        
        state["outliers"] = outliers
        return state
    
    def _generate_comparison_node(self, state: BenchmarkState) -> BenchmarkState:
        """Generate performance insights using LLM."""
        benchmark_name = state["benchmark_name"]
        mean_time = state["mean_time"]
        std_dev = state["std_dev"]
        throughput = state["throughput"]
        performance_level = state["performance_level"]
        outliers = state["outliers"]
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are an expert in performance analysis and optimization."),
            ("user", """Analyze this benchmark result and provide insights:

Benchmark: {benchmark_name}
Mean Time: {mean_time:.6f}s ({mean_ms:.3f}ms)
Std Dev: {std_dev:.6f}s
Throughput: {throughput:.2f} ops/sec
Performance Level: {performance_level}
Outliers Detected: {outlier_count}

Provide analysis as JSON with:
- performance_assessment: Overall performance evaluation
- bottlenecks: List of potential bottlenecks (2-3 items)
- recommendations: List of optimization recommendations (2-3 items)
- comparison_summary: One-line summary for comparison

Focus on actionable insights.""")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({
            "benchmark_name": benchmark_name,
            "mean_time": mean_time,
            "mean_ms": mean_time * 1000,
            "std_dev": std_dev,
            "throughput": throughput,
            "performance_level": performance_level.value,
            "outlier_count": len(outliers)
        })
        
        try:
            insights_data = json.loads(response.content)
            state["insights"] = BenchmarkInsights(
                performance_assessment=insights_data.get("performance_assessment", "Performance assessed"),
                bottlenecks=insights_data.get("bottlenecks", []),
                recommendations=insights_data.get("recommendations", []),
                comparison_summary=insights_data.get("comparison_summary", "Benchmark completed")
            )
        except:
            state["insights"] = BenchmarkInsights(
                performance_assessment=f"Performance level: {performance_level.value}",
                bottlenecks=["Analysis unavailable"],
                recommendations=["Re-run benchmark with more iterations"],
                comparison_summary="Benchmark completed"
            )
        
        return state
    
    def benchmark(
        self,
        name: str,
        function: Callable,
        iterations: int = 100,
        benchmark_type: BenchmarkType = BenchmarkType.EXECUTION_TIME
    ) -> BenchmarkResult:
        """Run benchmark through workflow."""
        initial_state: BenchmarkState = {
            "benchmark_name": name,
            "function": function,
            "iterations": iterations,
            "benchmark_type": benchmark_type,
            "execution_times": [],
            "mean_time": 0.0,
            "median_time": 0.0,
            "std_dev": 0.0,
            "throughput": 0.0,
            "percentile_95": 0.0,
            "percentile_99": 0.0,
            "performance_level": PerformanceLevel.ACCEPTABLE,
            "outliers": [],
            "insights": None,
            "comparison_data": {}
        }
        
        final_state = self.workflow.invoke(initial_state)
        
        # Create result
        result = BenchmarkResult(
            name=name,
            benchmark_type=benchmark_type,
            iterations=iterations,
            mean_time=final_state["mean_time"],
            median_time=final_state["median_time"],
            min_time=min(final_state["execution_times"]) if final_state["execution_times"] else 0.0,
            max_time=max(final_state["execution_times"]) if final_state["execution_times"] else 0.0,
            std_dev=final_state["std_dev"],
            throughput=final_state["throughput"],
            percentile_95=final_state["percentile_95"],
            percentile_99=final_state["percentile_99"],
            performance_level=final_state["performance_level"]
        )
        
        # Store result
        self.results[name] = result
        
        # Update statistics
        self.stats["total_benchmarks"] += 1
        self.stats["benchmark_types"][benchmark_type.value] += 1
        self.stats["total_iterations"] += iterations
        
        # Update average execution time
        total_time = sum(r.mean_time for r in self.results.values())
        self.stats["avg_execution_time"] = total_time / len(self.results)
        
        return result
    
    def compare(
        self,
        implementations: Dict[str, Callable],
        iterations: int = 100
    ) -> Dict[str, BenchmarkResult]:
        """Compare multiple implementations."""
        comparison: Dict[str, BenchmarkResult] = {}
        
        for name, func in implementations.items():
            result = self.benchmark(name, func, iterations, BenchmarkType.COMPARISON)
            comparison[name] = result
        
        return comparison
    
    def generate_report(self, comparison: Dict[str, BenchmarkResult]) -> str:
        """Generate comparison report."""
        if not comparison:
            return "No benchmarks to compare"
        
        report = "\n=== Benchmark Comparison Report ===\n"
        
        # Find fastest
        fastest = min(comparison.values(), key=lambda r: r.mean_time)
        
        for name, result in sorted(comparison.items(), key=lambda x: x[1].mean_time):
            ratio = result.mean_time / fastest.mean_time if fastest.mean_time > 0 else 1.0
            
            report += f"\n{name}:"
            report += f"\n  Mean Time: {result.mean_time*1000:.3f}ms"
            report += f"\n  Median Time: {result.median_time*1000:.3f}ms"
            report += f"\n  Min/Max: {result.min_time*1000:.3f}ms / {result.max_time*1000:.3f}ms"
            report += f"\n  Std Dev: {result.std_dev*1000:.3f}ms"
            report += f"\n  Throughput: {result.throughput:.2f} ops/sec"
            report += f"\n  95th Percentile: {result.percentile_95*1000:.3f}ms"
            report += f"\n  Performance: {result.performance_level.value.upper()}"
            report += f"\n  Relative: {ratio:.2f}x"
            if name == fastest.name:
                report += " ⚡ FASTEST"
            report += "\n"
        
        return report
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get benchmark statistics."""
        return {
            "name": self.name,
            **self.stats,
            "results_count": len(self.results)
        }


class BenchmarkPattern:
    """
    Comprehensive Benchmark Pattern coordinator with LangChain/LangGraph.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.benchmarks: Dict[str, IntelligentBenchmark] = {}
        self.model_name = model_name
    
    def create_benchmark(self, name: str) -> IntelligentBenchmark:
        """Create a benchmark."""
        benchmark = IntelligentBenchmark(name, model_name=self.model_name)
        self.benchmarks[name] = benchmark
        return benchmark
    
    def run(
        self,
        benchmark_name: str,
        test_name: str,
        function: Callable,
        iterations: int = 100
    ) -> Dict[str, Any]:
        """Run a benchmark."""
        if benchmark_name not in self.benchmarks:
            return {"error": f"Benchmark {benchmark_name} not found"}
        
        benchmark = self.benchmarks[benchmark_name]
        result = benchmark.benchmark(test_name, function, iterations)
        
        return {
            "benchmark": benchmark_name,
            "test": test_name,
            "result": {
                "mean_time": result.mean_time,
                "median_time": result.median_time,
                "throughput": result.throughput,
                "performance_level": result.performance_level.value
            }
        }
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get overall statistics."""
        total_benchmarks = sum(b.stats["total_benchmarks"] for b in self.benchmarks.values())
        total_iterations = sum(b.stats["total_iterations"] for b in self.benchmarks.values())
        
        return {
            "total_benchmark_suites": len(self.benchmarks),
            "total_benchmarks": total_benchmarks,
            "total_iterations": total_iterations
        }


if __name__ == "__main__":
    print("=== Benchmark Pattern with LangChain/LangGraph Demo ===\n")
    
    print("Scenario 1: Simple Performance Comparison")
    print("-" * 50)
    pattern = BenchmarkPattern(model_name="gpt-4")
    
    bench = pattern.create_benchmark("PerformanceSuite")
    
    # Define test functions
    def fast_operation():
        """Fast operation."""
        return sum(range(100))
    
    def medium_operation():
        """Medium operation."""
        return sum(range(1000))
    
    def slow_operation():
        """Slow operation."""
        return sum(range(10000))
    
    print("\nRunning individual benchmarks...")
    
    result1 = pattern.run("PerformanceSuite", "FastOp", fast_operation, iterations=1000)
    print(f"\nFastOp:")
    print(f"  Mean: {result1['result']['mean_time']*1000:.3f}ms")
    print(f"  Throughput: {result1['result']['throughput']:.2f} ops/sec")
    print(f"  Performance: {result1['result']['performance_level']}")
    
    result2 = pattern.run("PerformanceSuite", "MediumOp", medium_operation, iterations=1000)
    print(f"\nMediumOp:")
    print(f"  Mean: {result2['result']['mean_time']*1000:.3f}ms")
    print(f"  Throughput: {result2['result']['throughput']:.2f} ops/sec")
    print(f"  Performance: {result2['result']['performance_level']}")
    
    result3 = pattern.run("PerformanceSuite", "SlowOp", slow_operation, iterations=1000)
    print(f"\nSlowOp:")
    print(f"  Mean: {result3['result']['mean_time']*1000:.3f}ms")
    print(f"  Throughput: {result3['result']['throughput']:.2f} ops/sec")
    print(f"  Performance: {result3['result']['performance_level']}")
    
    print("\n" + "=" * 50)
    print("\nScenario 2: Implementation Comparison")
    print("-" * 50)
    
    # Compare implementations
    implementations = {
        "Fast": fast_operation,
        "Medium": medium_operation,
        "Slow": slow_operation
    }
    
    print("\nComparing implementations...")
    comparison = bench.compare(implementations, iterations=500)
    report = bench.generate_report(comparison)
    print(report)
    
    print("=" * 50)
    print("\nScenario 3: List Processing Comparison")
    print("-" * 50)
    
    list_bench = pattern.create_benchmark("ListProcessing")
    
    # Different list processing approaches
    test_list = list(range(1000))
    
    def list_comprehension():
        return [x * 2 for x in test_list]
    
    def map_function():
        return list(map(lambda x: x * 2, test_list))
    
    def for_loop():
        result = []
        for x in test_list:
            result.append(x * 2)
        return result
    
    print("\nBenchmarking list processing approaches...")
    
    list_implementations = {
        "ListComprehension": list_comprehension,
        "MapFunction": map_function,
        "ForLoop": for_loop
    }
    
    list_comparison = list_bench.compare(list_implementations, iterations=500)
    list_report = list_bench.generate_report(list_comparison)
    print(list_report)
    
    print("=" * 50)
    print("\nScenario 4: String Operations Benchmark")
    print("-" * 50)
    
    string_bench = pattern.create_benchmark("StringOperations")
    
    test_strings = ["test" + str(i) for i in range(100)]
    
    def string_join():
        return "".join(test_strings)
    
    def string_concat():
        result = ""
        for s in test_strings:
            result += s
        return result
    
    print("\nBenchmarking string operations...")
    
    string_implementations = {
        "Join": string_join,
        "Concatenation": string_concat
    }
    
    string_comparison = string_bench.compare(string_implementations, iterations=500)
    string_report = string_bench.generate_report(string_comparison)
    print(string_report)
    
    print("=" * 50)
    print("\nStatistics")
    print("-" * 50)
    
    overall_stats = pattern.get_statistics()
    print(f"\nOverall Statistics:")
    print(f"  Total Benchmark Suites: {overall_stats['total_benchmark_suites']}")
    print(f"  Total Benchmarks Run: {overall_stats['total_benchmarks']}")
    print(f"  Total Iterations: {overall_stats['total_iterations']}")
    
    print(f"\nPerformance Suite Statistics:")
    perf_stats = bench.get_statistics()
    for key, value in perf_stats.items():
        print(f"  {key}: {value}")
    
    print("\n" + "=" * 50)
    print("Benchmark pattern demonstration complete!")
    print("=" * 50)

    

Agentic Architecture Pattern: Benchmarking

Pattern 268: Benchmarking Pattern – Line-by-Line Explanation

Overview

This pattern implements comprehensive performance benchmarking with statistical analysis, outlier detection, and LLM-powered insights. It measures execution time, memory usage, and provides detailed performance characteristics.

Code Explanation

Lines 1-17: Imports

"""
Pattern 268: Benchmarking Pattern with LangChain/LangGraph

Purpose: Performance benchmarking with statistical analysis.
Provides intelligent performance measurement with LLM-powered insights.
"""

Explanation: Module docstring – pattern for measuring and analyzing performance.

from typing import Dict, Any, List, Optional, Callable, TypedDict
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import time
import statistics
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
import json

Explanation: Includes time for measurements, statistics for analysis.

Lines 19-28: Enum Definitions

class BenchmarkType(Enum):
    """Types of benchmarks."""
    PERFORMANCE = "performance"  # Execution time
    MEMORY = "memory"  # Memory usage
    THROUGHPUT = "throughput"  # Operations per second
    LATENCY = "latency"  # Response time

Explanation: Four benchmark types: – PERFORMANCE: How fast does it run? – MEMORY: How much memory does it use? – THROUGHPUT: How many operations per second? – LATENCY: How long until response?

class BenchmarkStatus(Enum):
    """Status of benchmark."""
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"

Explanation: Tracks benchmark execution state.

Lines 30-75: Dataclasses

@dataclass
class BenchmarkConfig:
    """Configuration for benchmark."""
    name: str
    function: Callable
    args: tuple = field(default_factory=tuple)
    kwargs: Dict[str, Any] = field(default_factory=dict)
    iterations: int = 100
    warmup_iterations: int = 10
    benchmark_type: BenchmarkType = BenchmarkType.PERFORMANCE

Explanation: Configuration for what to benchmark: – Function to test – Arguments to pass – Number of iterations (default 100) – Warmup iterations (default 10) – discard first runs to account for JIT compilation, cache warming

@dataclass
class BenchmarkMetrics:
    """Metrics from benchmark run."""
    mean: float
    median: float
    std_dev: float
    min_value: float
    max_value: float
    percentile_95: float
    percentile_99: float
    total_time: float
    iterations: int

Explanation: Statistical metrics: – Mean: Average execution time – Median: Middle value (less affected by outliers) – Std Dev: Variability – Min/Max: Range – 95th/99th percentile: “Most” requests complete within this time – Total time: Sum of all iterations – Iterations: Number of measurements

@dataclass
class BenchmarkResult:
    """Result of benchmark."""
    config: BenchmarkConfig
    metrics: BenchmarkMetrics
    raw_measurements: List[float]
    outliers: List[float]
    status: BenchmarkStatus
    insights: List[str] = field(default_factory=list)
    timestamp: datetime = field(default_factory=datetime.now)

Explanation: Complete benchmark result including raw data, outliers, and AI insights.

Lines 77-90: TypedDict State

class BenchmarkState(TypedDict):
    """State for benchmarking workflow."""
    config: BenchmarkConfig
    raw_measurements: List[float]
    metrics: Optional[BenchmarkMetrics]
    outliers: List[float]
    status: BenchmarkStatus
    insights: List[str]
    error: Optional[str]

Explanation: Workflow state tracking measurements, metrics, outliers, and status.

Lines 92-130: IntelligentBenchmark Initialization

class IntelligentBenchmark:
    """
    Intelligent Benchmarking with LangGraph workflow integration.

    Features:
    - Multiple benchmark types (performance, memory, throughput, latency)
    - Statistical analysis (mean, median, std dev, percentiles)
    - Outlier detection
    - LLM-powered performance insights
    - Comparison reports
    """

Explanation: Main benchmarking class with statistical analysis and AI insights.

def __init__(self, model_name: str = "gpt-4", temperature: float = 0.5):
    self.llm = ChatOpenAI(model=model_name, temperature=temperature)

    # Storage
    self.results: Dict[str, BenchmarkResult] = {}

    # Build workflow
    self.workflow = self._build_workflow()

    # Statistics
    self.stats = {
        "total_benchmarks": 0,
        "total_measurements": 0,
        "total_outliers": 0
    }

Explanation: Initialize LLM, results storage, workflow, and tracking statistics.

Lines 132-150: Workflow Building

def _build_workflow(self) -> StateGraph:
    """Build LangGraph workflow."""
    workflow = StateGraph(BenchmarkState)

    workflow.add_node("warmup", self._warmup_node)
    workflow.add_node("measure", self._measure_node)
    workflow.add_node("analyze_statistics", self._analyze_statistics_node)
    workflow.add_node("detect_outliers", self._detect_outliers_node)
    workflow.add_node("generate_insights", self._generate_insights_node)

Explanation: Create 5-node workflow: 1. Warmup: Run iterations that will be discarded 2. Measure: Collect timing data 3. Analyze: Calculate statistics 4. Detect outliers: Find anomalies 5. Generate insights: Use LLM to interpret results

    workflow.set_entry_point("warmup")
    workflow.add_edge("warmup", "measure")
    workflow.add_edge("measure", "analyze_statistics")
    workflow.add_edge("analyze_statistics", "detect_outliers")
    workflow.add_edge("detect_outliers", "generate_insights")
    workflow.add_edge("generate_insights", END)

    return workflow.compile()

Explanation: Linear flow through all nodes.

Lines 152-174: Warmup Node

def _warmup_node(self, state: BenchmarkState) -> BenchmarkState:
    """Warm up before measurement."""
    config = state["config"]

    state["status"] = BenchmarkStatus.RUNNING

    # Run warmup iterations
    for _ in range(config.warmup_iterations):
        try:
            config.function(*config.args, **config.kwargs)
        except Exception as e:
            state["status"] = BenchmarkStatus.FAILED
            state["error"] = f"Warmup failed: {str(e)}"
            return state

    state["insights"].append(f"Completed {config.warmup_iterations} warmup iterations")

    return state

Explanation: Execute function multiple times before actual measurement. This accounts for: – JIT compilation warmup (in languages like Python/Java) – Cache warming (file system, CPU cache) – Connection pool initialization – Eliminates “cold start” bias from measurements

Lines 176-217: Measure Node

def _measure_node(self, state: BenchmarkState) -> BenchmarkState:
    """Measure performance."""
    config = state["config"]
    measurements: List[float] = []

Explanation: Initialize measurement list.

    for i in range(config.iterations):
        if config.benchmark_type == BenchmarkType.PERFORMANCE:
            # Measure execution time
            start_time = time.perf_counter()
            try:
                config.function(*config.args, **config.kwargs)
            except Exception as e:
                state["status"] = BenchmarkStatus.FAILED
                state["error"] = f"Measurement failed at iteration {i}: {str(e)}"
                return state
            end_time = time.perf_counter()

            measurement = (end_time - start_time) * 1000  # Convert to milliseconds
            measurements.append(measurement)

Explanation: PERFORMANCE measurement: – Use time.perf_counter() – highest resolution timer available – Calculate elapsed time – Convert to milliseconds for readability – Append to measurements list

        elif config.benchmark_type == BenchmarkType.LATENCY:
            # Similar to performance but may include network time
            start_time = time.perf_counter()
            try:
                config.function(*config.args, **config.kwargs)
            except Exception as e:
                state["status"] = BenchmarkStatus.FAILED
                state["error"] = f"Measurement failed at iteration {i}: {str(e)}"
                return state
            end_time = time.perf_counter()

            measurement = (end_time - start_time) * 1000
            measurements.append(measurement)

Explanation: LATENCY – same as performance, but semantic difference (latency often includes network/IO).

    state["raw_measurements"] = measurements
    self.stats["total_measurements"] += len(measurements)

    return state

Explanation: Store measurements and update statistics.

Lines 219-257: Analyze Statistics Node

def _analyze_statistics_node(self, state: BenchmarkState) -> BenchmarkState:
    """Analyze benchmark statistics."""
    measurements = state["raw_measurements"]

    if not measurements:
        state["status"] = BenchmarkStatus.FAILED
        state["error"] = "No measurements to analyze"
        return state

Explanation: Validate we have data to analyze.

    # Calculate statistics
    sorted_measurements = sorted(measurements)

    metrics = BenchmarkMetrics(
        mean=statistics.mean(measurements),
        median=statistics.median(measurements),
        std_dev=statistics.stdev(measurements) if len(measurements) > 1 else 0.0,
        min_value=min(measurements),
        max_value=max(measurements),
        percentile_95=sorted_measurements[int(len(sorted_measurements) * 0.95)],
        percentile_99=sorted_measurements[int(len(sorted_measurements) * 0.99)],
        total_time=sum(measurements),
        iterations=len(measurements)
    )

Explanation: Calculate comprehensive statistics: – Mean: Average performance (sum / count) – Median: Middle value when sorted (robust against outliers) – Std Dev: Standard deviation (how spread out values are) – Min/Max: Best and worst cases – 95th percentile: 95% of requests faster than this – 99th percentile: 99% of requests faster than this (tail latency) – Total time: Sum of all measurements – Iterations: Count

Percentiles are key for SLA definitions: “99% of requests complete within 100ms”.

    state["metrics"] = metrics
    state["insights"].append(
        f"Mean: {metrics.mean:.2f}ms, Median: {metrics.median:.2f}ms, "
        f"Std Dev: {metrics.std_dev:.2f}ms, 95th: {metrics.percentile_95:.2f}ms"
    )

    return state

Explanation: Store metrics and add summary insight.

Lines 259-299: Detect Outliers Node

def _detect_outliers_node(self, state: BenchmarkState) -> BenchmarkState:
    """Detect outliers in measurements."""
    measurements = state["raw_measurements"]
    metrics = state["metrics"]

    if not metrics or len(measurements) < 3:
        state["outliers"] = []
        return state

Explanation: Need at least 3 measurements for outlier detection.

    # Use IQR method for outlier detection
    sorted_measurements = sorted(measurements)
    q1_idx = int(len(sorted_measurements) * 0.25)
    q3_idx = int(len(sorted_measurements) * 0.75)

    q1 = sorted_measurements[q1_idx]
    q3 = sorted_measurements[q3_idx]
    iqr = q3 - q1

Explanation: IQR (Interquartile Range) Method: – Q1 = 25th percentile – Q3 = 75th percentile – IQR = Q3 – Q1 (middle 50% of data)

Standard statistical method for outlier detection.

    # Outliers are values beyond Q1 - 1.5*IQR or Q3 + 1.5*IQR
    lower_bound = q1 - 1.5 * iqr
    upper_bound = q3 + 1.5 * iqr

    outliers = [m for m in measurements if m < lower_bound or m > upper_bound]

Explanation: Classic outlier definition: – Below Q1 – 1.5 × IQR (unusually fast) – Above Q3 + 1.5 × IQR (unusually slow)

1.5 is standard multiplier from Tukey’s box plot method.

    state["outliers"] = outliers
    self.stats["total_outliers"] += len(outliers)

    if outliers:
        state["insights"].append(
            f"Detected {len(outliers)} outliers ({len(outliers)/len(measurements)*100:.1f}%)"
        )

    return state

Explanation: Store outliers and report percentage.

Lines 301-365: Generate Insights Node (LLM-Powered)

def _generate_insights_node(self, state: BenchmarkState) -> BenchmarkState:
    """Generate LLM-powered performance insights."""
    config = state["config"]
    metrics = state["metrics"]
    outliers = state["outliers"]

    if not metrics:
        return state

Explanation: Validate we have metrics to analyze.

    # Prepare data for LLM
    prompt = ChatPromptTemplate.from_messages([
        ("system", "You are an expert in performance optimization and benchmarking analysis."),
        ("user", """Analyze these benchmark results:

Benchmark: {name}
Type: {benchmark_type}
Iterations: {iterations}

Statistics:
- Mean: {mean:.2f}ms
- Median: {median:.2f}ms
- Std Dev: {std_dev:.2f}ms
- Min: {min_value:.2f}ms
- Max: {max_value:.2f}ms
- 95th percentile: {percentile_95:.2f}ms
- 99th percentile: {percentile_99:.2f}ms

Outliers: {outlier_count} ({outlier_percent:.1f}%)

Provide analysis as JSON with:
- performance_grade: Letter grade A-F based on consistency and speed
- bottlenecks: List of potential bottlenecks
- recommendations: List of optimization recommendations
- consistency_analysis: How consistent is the performance?
- outlier_impact: Impact of outliers on overall performance""")
    ])

Explanation: Comprehensive prompt asking LLM to analyze: – Overall performance grade – Potential bottlenecks causing slowness – Specific optimization recommendations – Performance consistency (low std dev = good) – Whether outliers are problematic

    chain = prompt | self.llm
    response = chain.invoke({
        "name": config.name,
        "benchmark_type": config.benchmark_type.value,
        "iterations": metrics.iterations,
        "mean": metrics.mean,
        "median": metrics.median,
        "std_dev": metrics.std_dev,
        "min_value": metrics.min_value,
        "max_value": metrics.max_value,
        "percentile_95": metrics.percentile_95,
        "percentile_99": metrics.percentile_99,
        "outlier_count": len(outliers),
        "outlier_percent": (len(outliers) / metrics.iterations * 100) if metrics.iterations > 0 else 0
    })

Explanation: Invoke LLM with all statistical data.

    try:
        insights_data = json.loads(response.content)

        state["insights"].append(f"Performance Grade: {insights_data.get('performance_grade', 'N/A')}")
        state["insights"].append(f"Consistency: {insights_data.get('consistency_analysis', 'N/A')}")

        bottlenecks = insights_data.get('bottlenecks', [])
        if bottlenecks:
            state["insights"].append(f"Potential Bottlenecks: {', '.join(bottlenecks)}")

        recommendations = insights_data.get('recommendations', [])
        if recommendations:
            state["insights"].extend([f"Recommendation: {rec}" for rec in recommendations])

    except Exception as e:
        state["insights"].append(f"LLM analysis completed (JSON parsing note: {str(e)})")

Explanation: Parse JSON response and add insights to state. Fallback if parsing fails.

    state["status"] = BenchmarkStatus.COMPLETED
    return state

Lines 367-405: Public Benchmark Method

def benchmark(self, config: BenchmarkConfig) -> BenchmarkResult:
    """Run benchmark with workflow."""
    initial_state: BenchmarkState = {
        "config": config,
        "raw_measurements": [],
        "metrics": None,
        "outliers": [],
        "status": BenchmarkStatus.PENDING,
        "insights": [],
        "error": None
    }

Explanation: Initialize workflow state.

    final_state = self.workflow.invoke(initial_state)

Explanation: Execute workflow (warmup → measure → analyze → detect outliers → generate insights).

    # Create result
    result = BenchmarkResult(
        config=config,
        metrics=final_state["metrics"],
        raw_measurements=final_state["raw_measurements"],
        outliers=final_state["outliers"],
        status=final_state["status"],
        insights=final_state["insights"]
    )

    self.results[config.name] = result
    self.stats["total_benchmarks"] += 1

    return result

Explanation: Package results, store in history, update statistics, and return.

Lines 430-500: BenchmarkingPattern Coordinator

class BenchmarkingPattern:
    """
    Comprehensive Benchmarking Pattern coordinator with LangChain/LangGraph.
    """

    def __init__(self, model_name: str = "gpt-4"):
        self.benchmarker = IntelligentBenchmark(model_name=model_name)

Explanation: Simple coordinator wrapping benchmarker.

    def benchmark_function(
        self,
        name: str,
        function: Callable,
        args: tuple = (),
        kwargs: Dict[str, Any] = None,
        iterations: int = 100,
        warmup: int = 10,
        benchmark_type: BenchmarkType = BenchmarkType.PERFORMANCE
    ) -> BenchmarkResult:
        """Benchmark a function."""
        config = BenchmarkConfig(
            name=name,
            function=function,
            args=args,
            kwargs=kwargs or {},
            iterations=iterations,
            warmup_iterations=warmup,
            benchmark_type=benchmark_type
        )

        return self.benchmarker.benchmark(config)

Explanation: Convenience method to benchmark function with explicit parameters.

    def compare_results(self, names: List[str]) -> Dict[str, Any]:
        """Compare multiple benchmark results."""
        results = {name: self.benchmarker.results.get(name) for name in names if name in self.benchmarker.results}

        if not results:
            return {"error": "No results found for comparison"}

        comparison = {
            "summary": {},
            "rankings": {},
            "relative_performance": {}
        }

        # Calculate rankings by mean
        sorted_by_mean = sorted(results.items(), key=lambda x: x[1].metrics.mean if x[1].metrics else float('inf'))

        for rank, (name, result) in enumerate(sorted_by_mean, 1):
            comparison["rankings"][name] = rank

        # Calculate relative performance (vs fastest)
        fastest_mean = sorted_by_mean[0][1].metrics.mean if sorted_by_mean else 1.0

        for name, result in results.items():
            if result.metrics:
                comparison["relative_performance"][name] = {
                    "mean_ratio": result.metrics.mean / fastest_mean,
                    "slower_by": f"{((result.metrics.mean / fastest_mean - 1) * 100):.1f}%"
                }

        return comparison

Explanation: Compare multiple benchmarks: – Rank by mean execution time – Calculate relative performance (how much slower than fastest) – Useful for comparing algorithm implementations

Lines 502-599: Demo Code

Scenario 1: Simple Function

def simple_calculation(n: int) -> int:
    """Simple mathematical calculation."""
    result = 0
    for i in range(n):
        result += i * i
    return result

result1 = pattern.benchmark_function(
    name="SimpleCalculation",
    function=simple_calculation,
    args=(1000,),
    iterations=1000,
    warmup=50
)

Explanation: Benchmark O(n) loop with 1000 iterations, 50 warmup runs.

Scenario 2: String Operations

def string_operations(text: str) -> str:
    """String manipulation operations."""
    result = text.upper()
    result = result.replace("a", "A")
    result = "".join(reversed(result))
    return result

result2 = pattern.benchmark_function(
    name="StringOperations",
    function=string_operations,
    args=("hello world " * 100,),
    iterations=500
)

Explanation: Benchmark string operations on 1200-character string.

Scenario 3: List Comprehension vs Loop

def list_comprehension(n: int) -> List[int]:
    """List comprehension approach."""
    return [i * i for i in range(n)]

def loop_approach(n: int) -> List[int]:
    """Traditional loop approach."""
    result = []
    for i in range(n):
        result.append(i * i)
    return result

result3 = pattern.benchmark_function(
    name="ListComprehension",
    function=list_comprehension,
    args=(10000,),
    iterations=200
)

result4 = pattern.benchmark_function(
    name="LoopApproach",
    function=loop_approach,
    args=(10000,),
    iterations=200
)

# Compare the two approaches
comparison = pattern.compare_results(["ListComprehension", "LoopApproach"])

Explanation: Classic Python comparison – list comprehensions are typically faster. Compare results to see exact difference.

Key Design Patterns

1. Template Method: Common workflow with type-specific measurement
2. Strategy Pattern: Different benchmark types (performance, memory, throughput)
3. Chain of Responsibility: Sequential processing through workflow nodes
4. Observer Pattern: Statistics tracking observes all benchmarks
5. Command Pattern: BenchmarkConfig encapsulates execution

Performance Considerations

• Warmup: O(warmup_iterations × function_time) – essential for accurate measurements
• Measurement: O(iterations × function_time) – linear in iterations
• Analysis: O(n log n) – sorting for percentiles
• Outlier Detection: O(n log n) – sorting for IQR
• Memory: O(iterations) – stores all raw measurements

Best Practices: – Use 100+ iterations for stable statistics – Include warmup (10-50 iterations) for JIT/cache warmup – Use time.perf_counter() – most accurate timer – Report median for typical performance, 95th/99th for tail latency – Compare relative performance, not absolute (hardware varies)

Extension Points

1. Add memory profiling with tracemalloc
2. Implement throughput measurement (ops/sec)
3. Add multi-threaded benchmarking
4. Implement statistical tests (t-test for comparing benchmarks)
5. Add visualization (histograms, box plots)
6. Support async functions with asyncio
7. Add regression detection (compare against baseline)