Generative Simulation Benchmarking for circular manufacturing supply chains under real-time policy constraints

Generative Simulation Benchmarking for circular manufacturing supply chains under real-time policy constraints

It started with a frustrating paradox. I was building a reinforcement learning agent to optimize a closed-loop manufacturing supply chain — the kind where returned products get refurbished and fed back into production. My agent was performing beautifully in simulation, reducing waste by 22% and cutting raw material costs by 18%. Then I deployed it against a live digital twin of an actual facility, and it collapsed. The policies it generated violated environmental compliance windows, missed regulatory reporting deadlines, and completely ignored the fact that certain materials could only be processed during specific hours due to emissions caps.

I spent weeks debugging, and the culprit wasn't my neural network architecture or my reward shaping. It was the simulation itself. I was benchmarking against static, pre-computed scenarios that didn't capture the dynamic, policy-constrained reality of circular manufacturing. That experience sent me down a rabbit hole that led to what I now call Generative Simulation Benchmarking — a framework that combines generative AI, real-time policy constraint encoding, and adaptive scenario synthesis to create evaluation environments that actually reflect operational reality.

In this article, I want to share what I learned, the code I wrote, the failures I hit, and the architecture that finally worked. This isn't a theoretical overview — it's a hands-on account of building, testing, and iterating on a benchmarking system for circular supply chains under real-time constraints.

The Problem with Traditional Simulation Benchmarking

Before diving into my solution, let me be clear about why traditional approaches fail. Standard supply chain simulators like SimPy or AnyLogic are excellent at modeling stochastic processes — demand fluctuations, lead times, machine breakdowns. But they treat policy constraints as static parameters. You define a constraint, and it's baked into the simulation as a fixed rule.

In circular manufacturing, constraints are dynamic. Consider these examples:

• Carbon credit thresholds that reset monthly based on cumulative emissions
• Recycling quotas that adjust based on market prices for recovered materials
• Labor availability windows that shift with shift schedules and union agreements
• Regulatory compliance deadlines that trigger audits when violated

When I explored the literature on simulation-based benchmarking for circular supply chains, I kept hitting the same wall: papers would validate their proposed algorithms against static datasets like the classic Beer Game or generic benchmark problems. These evaluations told you how an algorithm performed in a simplified world, not how it would perform in your facility with your constraints changing in real-time.

My exploration of agentic AI systems revealed something interesting though. The same techniques used to make AI agents adapt to dynamic environments — curriculum learning, generative adversarial training, and constraint-aware reward shaping — could be repurposed to create better benchmarks themselves. Instead of asking "how does my algorithm perform on this fixed scenario?", we can ask "how does my algorithm perform across a distribution of dynamically-generated scenarios that respect real-time policy constraints?"

Generative Simulation: The Core Concept

The core idea is deceptively simple. Instead of manually designing simulation scenarios, we use a generative model to create them. But the generative model isn't producing arbitrary scenarios — it's producing scenarios that are:

1. Physically plausible (respecting material flow conservation, capacity limits)
2. Policy-constrained (respecting real-time regulatory and operational constraints)
3. Adversarially challenging (probing weaknesses in the algorithm being benchmarked)

This creates an interesting feedback loop. The generative model learns to produce scenarios that challenge the algorithm, and the algorithm learns to handle those scenarios. The benchmark becomes a co-evolutionary system.

As I was experimenting with this approach, I came across a beautiful connection to quantum computing. The constraint satisfaction problem at the heart of generating valid circular supply chain scenarios is NP-hard in the general case. For small instances, classical solvers work fine. But when you have hundreds of material types, multiple recovery pathways, and real-time policy constraints, the search space explodes. Quantum annealing and QAOA (Quantum Approximate Optimization Algorithm) offer a potential path forward for finding near-optimal constraint-satisfying scenarios, though my experimentation has been limited to classical approximations so far.

Architecture Overview

Let me walk you through the architecture I settled on after several iterations. The system has four main components:

class GenerativeSimulationBenchmark:
    def __init__(self, supply_chain_model, policy_engine, generator, evaluator):
        self.supply_chain_model = supply_chain_model  # Simulator of the physical system
        self.policy_engine = policy_engine            # Real-time policy constraint checker
        self.generator = generator                     # Generative model for scenarios
        self.evaluator = evaluator                     # Algorithm evaluation harness

    def run_benchmark(self, algorithm, num_rounds=100):
        results = []
        scenario_history = []

        for round_idx in range(num_rounds):
            # Generate a scenario that respects current policy constraints
            scenario = self.generator.generate(
                policy_state=self.policy_engine.current_state()
            )

            # Run the algorithm against this scenario
            performance = self.evaluator.evaluate(
                algorithm=algorithm,
                scenario=scenario,
                policy_engine=self.policy_engine
            )

            # Feed performance back to improve future scenario generation
            self.generator.update(performance)

            results.append(performance)
            scenario_history.append(scenario)

            # Update policy state based on simulation outcomes
            self.policy_engine.update(scenario.outcomes)

        return results, scenario_history

The key insight here is the feedback loop. The generator doesn't just create random scenarios — it creates scenarios that are increasingly tailored to expose weaknesses in the algorithm being tested. This makes the benchmark adaptive and stressful in a way that static benchmarks can never be.

Policy Constraint Encoding

The heart of the system is how we encode real-time policy constraints. In my experimentation, I found that a hybrid approach works best: use a formal constraint language for hard constraints (things that must never be violated) and learned penalty functions for soft constraints (things that should be minimized but are not absolute).

Here's the policy engine I built:

from dataclasses import dataclass
from typing import Dict, List, Callable
import numpy as np

@dataclass
class PolicyConstraint:
    name: str
    constraint_type: str  # 'hard' or 'soft'
    check_fn: Callable[[Dict], bool]  # Returns True if satisfied
    penalty_fn: Callable[[Dict], float]  # Returns violation magnitude
    temporal_scope: str  # 'instant', 'window', 'cumulative'

class RealTimePolicyEngine:
    def __init__(self):
        self.constraints = []
        self.state_history = []
        self.cumulative_metrics = {}

    def add_constraint(self, constraint: PolicyConstraint):
        self.constraints.append(constraint)

    def check_state(self, state: Dict) -> Dict[str, bool]:
        """Check if a given state satisfies all constraints."""
        results = {}
        for constraint in self.constraints:
            if constraint.temporal_scope == 'instant':
                results[constraint.name] = constraint.check_fn(state)
            elif constraint.temporal_scope == 'window':
                # Check against recent history
                window_data = self._get_window_data(
                    state,
                    lookback=constraint.temporal_window
                )
                results[constraint.name] = constraint.check_fn(window_data)
            elif constraint.temporal_scope == 'cumulative':
                # Check against cumulative metrics
                cumulative = self._get_cumulative(state)
                results[constraint.name] = constraint.check_fn(cumulative)
        return results

    def compute_violation_penalties(self, state: Dict) -> float:
        """Compute total penalty for soft constraint violations."""
        total_penalty = 0.0
        for constraint in self.constraints:
            if constraint.constraint_type == 'soft':
                penalty = constraint.penalty_fn(state)
                total_penalty += penalty
        return total_penalty

    def update(self, state: Dict):
        """Update state history and cumulative metrics after each simulation step."""
        self.state_history.append(state)
        for key, value in state.get('cumulative_metrics', {}).items():
            self.cumulative_metrics[key] = self.cumulative_metrics.get(key, 0) + value

A concrete example of a real-time policy constraint in circular manufacturing: consider a facility that refurbishes electronic components. The constraint might state that the total energy consumption from refurbishment processes cannot exceed a monthly cap, but the cap resets based on renewable energy credits earned through recycling programs.

def create_energy_credit_constraint(max_monthly_energy, credits_per_ton_recycled):
    def check_fn(state):
        monthly_energy = state['cumulative_metrics']['monthly_energy_kwh']
        recycling_credits = state['cumulative_metrics']['recycled_tonnes'] * credits_per_ton_recycled
        effective_cap = max_monthly_energy + recycling_credits
        return monthly_energy <= effective_cap

    return PolicyConstraint(
        name='energy_credit_constraint',
        constraint_type='hard',
        check_fn=check_fn,
        penalty_fn=lambda s: max(0, s['cumulative_metrics']['monthly_energy_kwh'] -
                                  (max_monthly_energy + s['cumulative_metrics']['recycled_tonnes'] * credits_per_ton_recycled)),
        temporal_scope='cumulative'
    )

This constraint is dynamic because the effective cap grows as the facility recycles more materials. A static benchmark would miss this entirely.

The Generative Model for Scenario Synthesis

The most challenging part was building the generative model that creates realistic, constraint-respecting scenarios. I explored several approaches, but the one that worked best was a conditional variational autoencoder (CVAE) combined with a constraint-satisfaction post-processing layer.

import torch
import torch.nn as nn

class ScenarioGenerator(nn.Module):
    def __init__(self, latent_dim=32, state_dim=64, condition_dim=16):
        super().__init__()
        self.encoder = nn.Sequential(
            nn.Linear(state_dim + condition_dim, 128),
            nn.ReLU(),
            nn.Linear(128, 64),
            nn.ReLU(),
            nn.Linear(64, latent_dim * 2)  # mean and log variance
        )

        self.decoder = nn.Sequential(
            nn.Linear(latent_dim + condition_dim, 64),
            nn.ReLU(),
            nn.Linear(64, 128),
            nn.ReLU(),
            nn.Linear(128, state_dim),
            nn.Sigmoid()  # Output in [0, 1] range
        )

    def encode(self, state, condition):
        x = torch.cat([state, condition], dim=-1)
        params = self.encoder(x)
        mean, log_var = params.chunk(2, dim=-1)
        return mean, log_var

    def reparameterize(self, mean, log_var):
        std = torch.exp(0.5 * log_var)
        eps = torch.randn_like(std)
        return mean + eps * std

    def decode(self, z, condition):
        x = torch.cat([z, condition], dim=-1)
        return self.decoder(x)

    def forward(self, state, condition):
        mean, log_var = self.encode(state, condition)
        z = self.reparameterize(mean, log_var)
        reconstructed = self.decode(z, condition)
        return reconstructed, mean, log_var

The condition vector encodes the current policy state — things like remaining carbon budget, current recycling rates, labor availability, and regulatory deadlines. The generator learns to produce scenarios that are consistent with these conditions.

But here's the critical part: the raw output of the CVAE doesn't guarantee constraint satisfaction. That's where the post-processing layer comes in. I use a differentiable constraint projection layer that adjusts the generated scenario to satisfy hard constraints.

class ConstraintProjectionLayer(nn.Module):
    def __init__(self, policy_engine, projection_steps=10):
        super().__init__()
        self.policy_engine = policy_engine
        self.projection_steps = projection_steps

    def forward(self, scenario_tensor):
        # scenario_tensor: [batch_size, state_dim]
        projected = scenario_tensor.clone()

        for _ in range(self.projection_steps):
            # Check which constraints are violated
            violations = self._check_constraints(projected)

            if not violations.any():
                break

            # Compute gradient direction to reduce violations
            gradient = self._compute_constraint_gradient(projected, violations)

            # Project back to feasible region
            projected = projected - 0.1 * gradient

        return projected

    def _check_constraints(self, scenario):
        # Convert tensor to dict for policy engine
        scenario_dict = self._tensor_to_dict(scenario)
        results = self.policy_engine.check_state(scenario_dict)
        violations = ~torch.tensor(list(results.values()))
        return violations

    def _compute_constraint_gradient(self, scenario, violations):
        # Approximate gradient via finite differences
        gradient = torch.zeros_like(scenario)
        epsilon = 0.01

        for i in range(scenario.shape[-1]):
            scenario_plus = scenario.clone()
            scenario_plus[:, i] += epsilon
            scenario_minus = scenario.clone()
            scenario_minus[:, i] -= epsilon

            violation_plus = self._check_constraints(scenario_plus)
            violation_minus = self._check_constraints(scenario_minus)

            gradient[:, i] = (violation_plus.float() - violation_minus.float()) / (2 * epsilon)

        return gradient

In my research of generative models for simulation, I discovered that the projection layer is often the bottleneck. The finite-difference gradient computation is slow for high-dimensional state spaces. I experimented with using a learned constraint-violation predictor to speed this up, but the accuracy trade-off wasn't worth it for my use case. For production systems, I'd recommend using a differentiable constraint solver or a learned projection network trained via imitation learning.

Adversarial Scenario Generation

One of the most interesting findings from my experimentation was the power of adversarial generation. Instead of just generating valid scenarios, we want to generate scenarios that are challenging for the algorithm being benchmarked. This turns the benchmarking process into a game.

class AdversarialScenarioGenerator:
    def __init__(self, generator, discriminator, policy_engine):
        self.generator = generator
        self.discriminator = discriminator  # Predicts algorithm performance
        self.policy_engine = policy_engine

    def generate_challenging_scenario(self, current_algorithm_state):
        # Condition on algorithm's current performance
        condition = self._encode_algorithm_state(current_algorithm_state)

        # Generate candidate scenarios
        candidates = []
        for _ in range(10):
            scenario = self.generator.generate(condition=condition)
            # Project to satisfy constraints
            scenario = self._project_to_feasible(scenario)
            candidates.append(scenario)

        # Select the scenario that the discriminator predicts will be hardest
        hardest_scenario = None
        hardest_score = float('inf')

        for scenario in candidates:
            predicted_performance = self.discriminator.predict(scenario)
            if predicted_performance < hardest_score:
                hardest_score = predicted_performance
                hardest_scenario = scenario

        return hardest_scenario

This adversarial approach ensures that the benchmark doesn't become stale. As the algorithm improves, the generator learns to create harder scenarios, which forces further improvement. It's a continuous co-evolution that mirrors how real-world supply chains become more efficient under regulatory pressure.

Benchmarking Metrics for Circular Manufacturing

Through studying circular economy metrics, I realized that traditional supply chain KPIs (on-time delivery, inventory turns, cost) are insufficient. Circular manufacturing requires additional metrics that capture the health of the closed-loop system:

def compute_circularity_metrics(simulation_results):
    metrics = {}

    # Material circularity indicator (MCI)
    total_material_in = sum(s['raw_material_in'] for s in simulation_results)
    recycled_material = sum(s['recycled_material'] for s in simulation_results)
    metrics['material_circularity'] = recycled_material / max(total_material_in, 1e-6)

    # Recovery rate
    total_waste = sum(s['waste_generated'] for s in simulation_results)
    total_recovered = sum(s['waste_recovered'] for s in simulation_results)
    metrics['recovery_rate'] = total_recovered / max(total_waste, 1e-6)

    # Policy compliance score
    total_steps = len(simulation_results)
    compliant_steps = sum(1 for s in simulation_results if s['policy_compliant'])
    metrics['policy_compliance'] = compliant_steps / max(total_steps, 1e-6)

    # Resource efficiency
    total_energy = sum(s['energy_consumed'] for s in simulation_results)
    total_output = sum(s['units_produced'] for s in simulation_results)
    metrics['energy_per_unit'] = total_energy / max(total_output, 1e-6)

    # Dynamic constraint responsiveness
    constraint_responses = [s['constraint_response_time'] for s in simulation_results
                           if s.get('constraint_triggered', False)]
    metrics['avg_constraint_response_ms'] = np.mean(constraint_responses) if constraint_responses else 0

    return metrics

Real-World Applications and Case Study

During my investigation of real-world circular supply chains, I focused on the electronics refurbishment industry. This sector is particularly interesting because it has:

1. Complex material flows (precious metals, rare earth elements, hazardous materials)
2. Stringent regulations (WEEE directive in EU, various state-level e-waste laws)
3. Real-time pricing dynamics (commodity prices for recovered materials)
4. High uncertainty (return rates, quality of returned products)

I built a digital twin of a mid-sized electronics refurbishment facility and used