import gradio as gr import numpy as np import random import time import json from collections import deque from smolagents import Tool, HfApiModel import plotly.graph_objects as go import plotly.express as px from plotly.subplots import make_subplots # ==================== SUPPLY CHAIN TOOLS ==================== class SupplyTool(Tool): name = "supply_tool" description = "Supplies raw materials to the manufacturer." inputs = { "demand": {"type": "number", "description": "Demand from manufacturer"}, "inventory": {"type": "number", "description": "Supplier's current inventory"} } output_type = "number" def forward(self, demand: int, inventory: int) -> int: supply = min(inventory, demand) return supply class ManufactureTool(Tool): name = "manufacture_tool" description = "Manufactures goods from raw materials." inputs = { "raw_material": {"type": "number", "description": "Available raw materials"}, "capacity": {"type": "number", "description": "Manufacturing capacity"}, "demand": {"type": "number", "description": "Demand for manufactured goods"} } output_type = "number" def forward(self, raw_material: int, capacity: int, demand: int) -> int: production = min(raw_material, capacity, demand) return production class DistributeTool(Tool): name = "distribute_tool" description = "Distributes goods from manufacturer to retailers." inputs = { "inventory": {"type": "number", "description": "Available inventory"}, "demand": {"type": "number", "description": "Retailer demand"} } output_type = "number" def forward(self, inventory: int, demand: int) -> int: distribution = min(inventory, demand) return distribution class RetailTool(Tool): name = "retail_tool" description = "Handles retail sales to customers." inputs = { "customer_demand": {"type": "number", "description": "Customer demand"}, "available_stock": {"type": "number", "description": "Available retail stock"} } output_type = "number" def forward(self, customer_demand: int, available_stock: int) -> int: sales = min(customer_demand, available_stock) return sales # ==================== DEMAND FORECASTING ==================== class DemandForecast: def __init__(self, window_size=10): self.demand_history = deque(maxlen=window_size) self.window_size = window_size def update(self, actual_demand): self.demand_history.append(actual_demand) def forecast(self): if len(self.demand_history) < 2: return 30 # Default demand if not enough history # Simple exponential smoothing with trend alpha = 0.3 # Smoothing factor trend = (sum(np.diff(list(self.demand_history))) / (len(self.demand_history) - 1)) last_demand = self.demand_history[-1] forecast = last_demand + alpha * trend return max(int(forecast), 0) # ==================== PERFORMANCE METRICS ==================== class PerformanceMetrics: def __init__(self): self.total_demand = 0 self.fulfilled_demand = 0 self.backorders = 0 self.inventory_history = [] self.total_costs = 0 self.daily_metrics = [] def update_fill_rate(self, demand, fulfilled): self.total_demand += demand self.fulfilled_demand += fulfilled self.backorders += demand - fulfilled def update_inventory(self, inventory_levels): # Calculate total inventory across all stages total_inventory = sum(inventory_levels.values()) self.inventory_history.append(total_inventory) def update_costs(self, new_costs): self.total_costs += new_costs def log_daily_metrics(self, step, state, costs): self.daily_metrics.append({ 'step': step, 'supplier_inv': state['supplier_inventory'], 'manufacturer_inv': state['manufacturer_inventory'], 'distributor_inv': state['distributor_inventory'], 'retail_inv': state['retail_inventory'], 'backorders': state['backorders'], 'daily_costs': costs, 'cumulative_costs': self.total_costs }) def calculate_metrics(self): fill_rate = (self.fulfilled_demand / self.total_demand * 100) if self.total_demand > 0 else 0 avg_inventory = sum(self.inventory_history) / len(self.inventory_history) if self.inventory_history else 1 inventory_turnover = self.fulfilled_demand / avg_inventory if avg_inventory > 0 else 0 return { "fill_rate": round(fill_rate, 2), "inventory_turnover": round(inventory_turnover, 2), "backorders": self.backorders, "total_costs": round(self.total_costs, 2), "average_inventory": round(avg_inventory, 2) } # ==================== COST CALCULATIONS ==================== costs = { "raw_material": 10, "manufacturing": 15, "distribution": 5, "holding": 2, "backorder": 20 } def calculate_daily_costs(state, supply, production, distribution): daily_costs = ( supply * costs["raw_material"] + production * costs["manufacturing"] + distribution * costs["distribution"] + (state["supplier_inventory"] + state["manufacturer_inventory"] + state["distributor_inventory"] + state["retail_inventory"]) * costs["holding"] + state["backorders"] * costs["backorder"] ) return daily_costs # ==================== SUPPLY CHAIN SIMULATION ==================== class SupplyChainSimulator: def __init__(self): self.supply_tool = SupplyTool() self.manufacture_tool = ManufactureTool() self.distribute_tool = DistributeTool() self.retail_tool = RetailTool() self.demand_forecast = DemandForecast() self.metrics = PerformanceMetrics() self.reset_state() def reset_state(self): self.state = { "supplier_inventory": 100, "manufacturer_inventory": 20, "manufacturer_capacity": 50, "distributor_inventory": 15, "retail_inventory": 10, "retailer_customer_demand": 30, "backorders": 0, "forecast_demand": 30 } self.demand_forecast = DemandForecast() self.metrics = PerformanceMetrics() def run_simulation(self, steps=5, progress_callback=None): results = [] for step in range(steps): if progress_callback: progress_callback((step + 1) / steps) step_result = self.run_single_step(step) results.append(step_result) time.sleep(0.1) # Small delay for UI updates return results def run_single_step(self, step): initial_demand = self.state["retailer_customer_demand"] # 1. Update demand forecast self.state["forecast_demand"] = self.demand_forecast.forecast() # 2. Supply raw materials manufacturer_demand = max(self.state["forecast_demand"] - self.state["manufacturer_inventory"], 0) supply = self.supply_tool.forward(manufacturer_demand, self.state["supplier_inventory"]) self.state["supplier_inventory"] -= supply # 3. Manufacturing production = self.manufacture_tool.forward( raw_material=supply, capacity=self.state["manufacturer_capacity"], demand=manufacturer_demand ) self.state["manufacturer_capacity"] -= production self.state["manufacturer_inventory"] += production # 4. Distribution distributor_intake = min(self.state["manufacturer_inventory"], 50 - self.state["distributor_inventory"]) self.state["manufacturer_inventory"] -= distributor_intake self.state["distributor_inventory"] += distributor_intake retail_supply = self.distribute_tool.forward( inventory=self.state["distributor_inventory"], demand=self.state["retailer_customer_demand"] + self.state["backorders"] ) self.state["distributor_inventory"] -= retail_supply self.state["retail_inventory"] += retail_supply # 5. Retail sales and backorder management total_demand = self.state["retailer_customer_demand"] + self.state["backorders"] fulfilled_demand = self.retail_tool.forward( customer_demand=total_demand, available_stock=self.state["retail_inventory"] ) self.state["retail_inventory"] -= fulfilled_demand # Update backorders new_backorders = total_demand - fulfilled_demand self.state["backorders"] = new_backorders # Update metrics self.metrics.update_fill_rate(initial_demand, fulfilled_demand) self.metrics.update_inventory({ "supplier": self.state["supplier_inventory"], "manufacturer": self.state["manufacturer_inventory"], "distributor": self.state["distributor_inventory"], "retail": self.state["retail_inventory"] }) daily_costs = calculate_daily_costs(self.state, supply, production, retail_supply) self.metrics.update_costs(daily_costs) self.metrics.log_daily_metrics(step + 1, self.state, daily_costs) # 6. Daily updates self.state["manufacturer_capacity"] = 50 self.state["supplier_inventory"] += random.randint(10, 20) self.state["retailer_customer_demand"] = max(30 + random.randint(-5, 5), 0) self.demand_forecast.update(self.state["retailer_customer_demand"]) return { "step": step + 1, "state": self.state.copy(), "actions": { "supply": supply, "production": production, "distribution": retail_supply, "fulfilled_demand": fulfilled_demand }, "metrics": self.metrics.calculate_metrics(), "daily_costs": daily_costs } # ==================== VISUALIZATION FUNCTIONS ==================== def create_inventory_plot(metrics_data): if not metrics_data: return go.Figure() steps = [d['step'] for d in metrics_data] supplier_inv = [d['supplier_inv'] for d in metrics_data] manufacturer_inv = [d['manufacturer_inv'] for d in metrics_data] distributor_inv = [d['distributor_inv'] for d in metrics_data] retail_inv = [d['retail_inv'] for d in metrics_data] fig = go.Figure() fig.add_trace(go.Scatter(x=steps, y=supplier_inv, name='Supplier', line=dict(color='#FF6B6B', width=3))) fig.add_trace(go.Scatter(x=steps, y=manufacturer_inv, name='Manufacturer', line=dict(color='#FFA500', width=3))) fig.add_trace(go.Scatter(x=steps, y=distributor_inv, name='Distributor', line=dict(color='#FFD700', width=3))) fig.add_trace(go.Scatter(x=steps, y=retail_inv, name='Retail', line=dict(color='#FF69B4', width=3))) fig.update_layout( title='Inventory Levels Over Time', xaxis_title='Step', yaxis_title='Inventory Level', paper_bgcolor='rgba(0,0,0,0)', plot_bgcolor='rgba(0,0,0,0)', font=dict(color='#8B4513'), legend=dict(bgcolor='rgba(255,255,255,0.8)') ) return fig def create_costs_plot(metrics_data): if not metrics_data: return go.Figure() steps = [d['step'] for d in metrics_data] daily_costs = [d['daily_costs'] for d in metrics_data] cumulative_costs = [d['cumulative_costs'] for d in metrics_data] fig = make_subplots(specs=[[{"secondary_y": True}]]) fig.add_trace( go.Bar(x=steps, y=daily_costs, name='Daily Costs', marker_color='#FF7F50', opacity=0.7), secondary_y=False, ) fig.add_trace( go.Scatter(x=steps, y=cumulative_costs, name='Cumulative Costs', line=dict(color='#DC143C', width=3)), secondary_y=True, ) fig.update_xaxes(title_text="Step") fig.update_yaxes(title_text="Daily Costs", secondary_y=False) fig.update_yaxes(title_text="Cumulative Costs", secondary_y=True) fig.update_layout( title='Cost Analysis', paper_bgcolor='rgba(0,0,0,0)', plot_bgcolor='rgba(0,0,0,0)', font=dict(color='#8B4513') ) return fig # ==================== GRADIO INTERFACE ==================== # Initialize simulator simulator = SupplyChainSimulator() def run_simulation_interface(steps, progress=gr.Progress()): simulator.reset_state() def update_progress(p): progress(p, desc="Running simulation...") results = simulator.run_simulation(steps, update_progress) # Create summary final_metrics = simulator.metrics.calculate_metrics() summary = f""" ## 📊 Simulation Complete! **Performance Summary:** - **Fill Rate:** {final_metrics['fill_rate']}% - **Inventory Turnover:** {final_metrics['inventory_turnover']} - **Total Backorders:** {final_metrics['backorders']} - **Total Costs:** ${final_metrics['total_costs']} - **Average Inventory:** {final_metrics['average_inventory']} units """ # Create plots inventory_plot = create_inventory_plot(simulator.metrics.daily_metrics) costs_plot = create_costs_plot(simulator.metrics.daily_metrics) # Create detailed results table results_data = [] for result in results: results_data.append([ result['step'], result['state']['supplier_inventory'], result['state']['manufacturer_inventory'], result['state']['distributor_inventory'], result['state']['retail_inventory'], result['state']['backorders'], f"${result['daily_costs']:.2f}" ]) return summary, inventory_plot, costs_plot, results_data def reset_simulation(): simulator.reset_state() return "Simulation reset successfully!", go.Figure(), go.Figure(), [] # Custom CSS for warm professional theme custom_css = """ .gradio-container { background: linear-gradient(135deg, #FFF8DC 0%, #FFE4B5 50%, #FFDAB9 100%) !important; font-family: 'Segoe UI', Tahoma, Geneva, Verdana, sans-serif; } .gr-button { background: linear-gradient(45deg, #FF6B6B, #FFA500) !important; border: none !important; color: white !important; font-weight: bold !important; border-radius: 10px !important; box-shadow: 0 4px 15px rgba(255, 107, 107, 0.3) !important; transition: all 0.3s ease !important; } .gr-button:hover { transform: translateY(-2px) !important; box-shadow: 0 6px 20px rgba(255, 107, 107, 0.4) !important; } .gr-panel { background: rgba(255, 255, 255, 0.9) !important; border-radius: 15px !important; border: 2px solid #FFB347 !important; box-shadow: 0 8px 32px rgba(255, 179, 71, 0.2) !important; } h1, h2, h3 { color: #B22222 !important; text-shadow: 2px 2px 4px rgba(0,0,0,0.1) !important; } .gr-textbox { border: 2px solid #FFB347 !important; border-radius: 8px !important; } .gr-slider { accent-color: #FF6B6B !important; } """ # Create Gradio interface with gr.Blocks(css=custom_css, title="🏭 AI Supply Chain Agent") as demo: gr.HTML("""

🏭 AI Supply Chain Management Agent

Intelligent Multi-Agent Supply Chain Optimization & Simulation

""") with gr.Row(): with gr.Column(scale=1): gr.HTML("""

🎛️ Simulation Controls

""") steps_slider = gr.Slider( minimum=1, maximum=20, value=5, step=1, label="Number of Simulation Steps", info="More steps = longer simulation" ) with gr.Row(): run_btn = gr.Button("🚀 Run Simulation", variant="primary", size="lg") reset_btn = gr.Button("🔄 Reset", variant="secondary") gr.HTML("""

📋 How it works:

""") with gr.Column(scale=2): summary_output = gr.Markdown("Click 'Run Simulation' to start!", elem_classes=["summary-box"]) with gr.Tabs(): with gr.Tab("📈 Inventory Tracking"): inventory_plot = gr.Plot(label="Inventory Levels") with gr.Tab("💰 Cost Analysis"): costs_plot = gr.Plot(label="Cost Breakdown") with gr.Tab("📊 Detailed Results"): results_table = gr.Dataframe( headers=["Step", "Supplier Inv", "Manufacturer Inv", "Distributor Inv", "Retail Inv", "Backorders", "Daily Cost"], label="Step-by-Step Results" ) gr.HTML("""

🤖 Powered by AI Agents | 📊 Real-time Analytics | 🔄 Dynamic Optimization

""") # Event handlers run_btn.click( fn=run_simulation_interface, inputs=[steps_slider], outputs=[summary_output, inventory_plot, costs_plot, results_table] ) reset_btn.click( fn=reset_simulation, outputs=[summary_output, inventory_plot, costs_plot, results_table] ) # Launch the app if __name__ == "__main__": demo.launch( server_name="0.0.0.0", server_port=7860, share=True, show_error=True )