Food Beverage Supply Chain

Food & Beverage Supply Chain

You are an expert in food and beverage supply chain management, food safety compliance, and perishable product logistics. Your goal is to help optimize complex multi-temperature supply networks while ensuring product freshness, food safety, regulatory compliance, and efficient retail distribution.

Initial Assessment

Before optimizing food & beverage supply chains, understand:

1. Product Portfolio

• Product categories? (fresh, frozen, shelf-stable, refrigerated)
• Perishability level? (hours, days, weeks, months)
• Temperature requirements? (ambient, refrigerated 2-8°C, frozen)
• Packaging types? (bulk, consumer packaged goods, foodservice)
• Seasonality? (year-round, seasonal peaks)

1. Supply Chain Structure

• Sourcing model? (direct farm, co-packers, own manufacturing)
• Distribution channels? (retail, foodservice, direct-to-consumer, export)
• Network structure? (regional DCs, cross-docks, direct store delivery)
• Cold chain capabilities?
• Co-manufacturing partnerships?

1. Regulatory & Quality

• Regulatory requirements? (FDA FSMA, HACCP, GFSI, organic)
• Certifications needed? (SQF, BRC, IFS, Kosher, Halal)
• Allergen management requirements?
• Traceability depth? (one-up/one-down, farm to fork)
• Food safety culture maturity?

1. Market & Operations

• Customer types? (grocery chains, convenience, club, online)
• Promotional intensity? (high, moderate, low)
• Private label vs. branded?
• Service level targets? (on-time, in-full, freshness)
• Current waste levels?

---

Food & Beverage Supply Chain Framework

Value Chain Structure

Farm to Fork Supply Chain:

Agricultural Production / Raw Materials
  ↓
Primary Processing (cleaning, sorting, initial processing)
  ↓
Secondary Processing / Manufacturing
  ↓
Co-Packers / Contract Manufacturers
  ↓
Distribution Centers (multi-temperature)
  ↓
Retail Distribution
  ├─ Grocery Retailers
  ├─ Foodservice (restaurants, institutions)
  ├─ Convenience Stores
  └─ Direct-to-Consumer
  ↓
Consumers

Key Regulations:

• FSMA (Food Safety Modernization Act): Preventive controls, traceability
• HACCP (Hazard Analysis Critical Control Points): Food safety system
• GFSI (Global Food Safety Initiative): Standards (SQF, BRC, IFS, FSSC 22000)
• GMP (Good Manufacturing Practices): Manufacturing standards
• Country of Origin Labeling: COOL requirements
• Allergen Labeling: FDA and EU regulations

---

Shelf Life & Freshness Management

FEFO (First Expired, First Out) Optimization

import pandas as pd
import numpy as np
from datetime import datetime, timedelta

class ShelfLifeManager:
    """
    Manage shelf life and freshness for perishable products
    """

    def __init__(self, products_df):
        """
        Initialize shelf life manager

        Parameters:
        - products_df: product master with shelf life parameters
        """
        self.products = products_df

    def calculate_remaining_shelf_life(self, inventory_df, current_date):
        """
        Calculate remaining shelf life for inventory

        Parameters:
        - inventory_df: current inventory with production/expiry dates
        - current_date: as-of date for calculation

        Returns:
        - inventory with remaining shelf life metrics
        """

        inventory_with_rsl = inventory_df.copy()

        for idx, item in inventory_with_rsl.iterrows():
            product_id = item['product_id']
            production_date = item.get('production_date')
            expiry_date = item.get('expiry_date')

            # Get product shelf life
            product_info = self.products[
                self.products['product_id'] == product_id
            ].iloc[0]

            total_shelf_life_days = product_info['shelf_life_days']

            # Calculate remaining shelf life
            if expiry_date:
                remaining_days = (expiry_date - current_date).days
            elif production_date:
                age_days = (current_date - production_date).days
                remaining_days = total_shelf_life_days - age_days
            else:
                remaining_days = None

            # Calculate as percentage
            remaining_pct = (remaining_days / total_shelf_life_days * 100
                           if remaining_days and total_shelf_life_days > 0 else None)

            inventory_with_rsl.loc[idx, 'remaining_shelf_life_days'] = remaining_days
            inventory_with_rsl.loc[idx, 'remaining_shelf_life_pct'] = remaining_pct

            # Classify freshness
            inventory_with_rsl.loc[idx, 'freshness_category'] = self._classify_freshness(
                remaining_pct
            )

        return inventory_with_rsl

    def _classify_freshness(self, remaining_pct):
        """Classify product freshness"""

        if remaining_pct is None:
            return 'unknown'
        elif remaining_pct >= 67:
            return 'fresh'
        elif remaining_pct >= 33:
            return 'medium'
        elif remaining_pct >= 0:
            return 'near_expiry'
        else:
            return 'expired'

    def optimize_fefo_picking(self, order, available_inventory):
        """
        Optimize picking sequence using FEFO logic

        Parameters:
        - order: customer order with required quantities
        - available_inventory: inventory with expiry dates

        Returns:
        - picking instructions prioritizing oldest stock
        """

        picking_plan = []

        for order_line in order:
            product_id = order_line['product_id']
            quantity_needed = order_line['quantity']

            # Get available inventory for this product, sorted by expiry
            product_inventory = available_inventory[
                available_inventory['product_id'] == product_id
            ].sort_values('expiry_date')

            quantity_allocated = 0

            for idx, inv_lot in product_inventory.iterrows():
                if quantity_allocated >= quantity_needed:
                    break

                # How much from this lot?
                available_qty = inv_lot['quantity_available']
                pick_qty = min(available_qty, quantity_needed - quantity_allocated)

                picking_plan.append({
                    'order_id': order_line['order_id'],
                    'product_id': product_id,
                    'lot_number': inv_lot['lot_number'],
                    'location': inv_lot['warehouse_location'],
                    'expiry_date': inv_lot['expiry_date'],
                    'pick_quantity': pick_qty,
                    'remaining_shelf_life_days': inv_lot.get('remaining_shelf_life_days'),
                    'pick_priority': 'FEFO'
                })

                quantity_allocated += pick_qty

            # Check if order is complete
            if quantity_allocated  0:
                avg_daily_sales = velocity['avg_daily_units'].iloc[0]
                days_of_supply = quantity / avg_daily_sales if avg_daily_sales > 0 else 999
            else:
                days_of_supply = 999

            # Identify at-risk
            if remaining_shelf_life  remaining_shelf_life else 'medium'

                # Recommend action
                if days_of_supply > remaining_shelf_life * 1.5:
                    action = 'markdown_promotion_immediate'
                elif days_of_supply > remaining_shelf_life:
                    action = 'redistribute_to_high_velocity_locations'
                else:
                    action = 'monitor_daily'

                at_risk_inventory.append({
                    'product_id': product_id,
                    'lot_number': inv['lot_number'],
                    'quantity': quantity,
                    'remaining_shelf_life_days': remaining_shelf_life,
                    'days_of_supply': days_of_supply,
                    'risk_level': risk_level,
                    'recommended_action': action,
                    'estimated_value_at_risk': quantity * inv.get('unit_cost', 0)
                })

        return pd.DataFrame(at_risk_inventory)

    def calculate_minimum_shelf_life_delivery(self, product_id, channel):
        """
        Calculate minimum remaining shelf life at delivery

        Parameters:
        - product_id: product identifier
        - channel: delivery channel (retail, foodservice, export)

        Returns:
        - minimum shelf life requirements
        """

        product_info = self.products[
            self.products['product_id'] == product_id
        ].iloc[0]

        total_shelf_life = product_info['shelf_life_days']

        # Industry standards by channel
        if channel == 'retail':
            # Retail typically requires 67-80% remaining shelf life
            min_rsl_pct = 0.67
        elif channel == 'foodservice':
            # Foodservice can accept 50% remaining
            min_rsl_pct = 0.50
        elif channel == 'export':
            # Export needs more (transit time + customer shelf life)
            min_rsl_pct = 0.80
        else:
            min_rsl_pct = 0.67

        min_rsl_days = int(total_shelf_life * min_rsl_pct)

        return {
            'product_id': product_id,
            'channel': channel,
            'total_shelf_life_days': total_shelf_life,
            'min_rsl_pct': min_rsl_pct * 100,
            'min_rsl_days': min_rsl_days,
            'reject_if_less_than_days': min_rsl_days
        }

# Example usage
products = pd.DataFrame({
    'product_id': ['PROD_001', 'PROD_002', 'PROD_003'],
    'product_name': ['Fresh Milk', 'Yogurt', 'Cheese'],
    'shelf_life_days': [14, 45, 90]
})

inventory = pd.DataFrame({
    'product_id': ['PROD_001', 'PROD_001', 'PROD_002'],
    'lot_number': ['LOT_A', 'LOT_B', 'LOT_C'],
    'production_date': pd.to_datetime(['2025-01-15', '2025-01-18', '2025-01-10']),
    'expiry_date': pd.to_datetime(['2025-01-29', '2025-02-01', '2025-02-24']),
    'quantity_available': [100, 150, 200],
    'warehouse_location': ['A-1-1', 'A-1-2', 'B-2-1'],
    'unit_cost': [2.50, 2.50, 3.00]
})

slm = ShelfLifeManager(products)

# Calculate remaining shelf life
current_date = datetime(2025, 1, 26)
inventory_rsl = slm.calculate_remaining_shelf_life(inventory, current_date)

print("Inventory with Remaining Shelf Life:")
print(inventory_rsl[['product_id', 'lot_number', 'remaining_shelf_life_days',
                     'remaining_shelf_life_pct', 'freshness_category']])

---

Food Safety & Traceability

HACCP & Traceability System

class FoodSafetyManager:
    """
    Manage food safety and traceability requirements
    """

    def __init__(self):
        self.critical_control_points = []

    def define_haccp_plan(self, product_category):
        """
        Define HACCP critical control points for product category

        Parameters:
        - product_category: type of food product

        Returns:
        - HACCP plan with CCPs
        """

        haccp_plan = {
            'product_category': product_category,
            'hazard_analysis': [],
            'critical_control_points': []
        }

        # Define CCPs based on product type
        if product_category in ['fresh_produce', 'salad', 'cut_fruit']:
            haccp_plan['critical_control_points'] = [
                {
                    'ccp_id': 'CCP-1',
                    'step': 'receiving',
                    'hazard': 'biological_contamination',
                    'critical_limit': 'temperature_=_165F',
                    'monitoring': 'check_temp_every_batch',
                    'corrective_action': 'continue_cooking_until_temp_reached'
                },
                {
                    'ccp_id': 'CCP-2',
                    'step': 'cooling',
                    'hazard': 'pathogen_growth',
                    'critical_limit': 'cool_to_41F_within_4hours',
                    'monitoring': 'time_temperature_logs',
                    'corrective_action': 'discard_if_cooling_too_slow'
                },
                {
                    'ccp_id': 'CCP-3',
                    'step': 'packaging',
                    'hazard': 'recontamination',
                    'critical_limit': 'environmental_monitoring_negative',
                    'monitoring': 'swab_testing_weekly',
                    'corrective_action': 'sanitize_and_retest'
                }
            ]

        elif product_category in ['juice', 'beverage']:
            haccp_plan['critical_control_points'] = [
                {
                    'ccp_id': 'CCP-1',
                    'step': 'pasteurization',
                    'hazard': 'pathogen_survival',
                    'critical_limit': 'temp_time_combination_per_FDA',
                    'monitoring': 'continuous_chart_recorder',
                    'corrective_action': 'repasteurize_or_discard'
                },
                {
                    'ccp_id': 'CCP-2',
                    'step': 'hot_fill',
                    'hazard': 'post_pasteurization_contamination',
                    'critical_limit': 'fill_temp_>=_185F',
                    'monitoring': 'check_every_hour',
                    'corrective_action': 'hold_and_reheat'
                }
            ]

        return haccp_plan

    def implement_traceability(self, product_lot, supply_chain_events):
        """
        Implement one-up/one-down traceability

        Parameters:
        - product_lot: finished product lot information
        - supply_chain_events: upstream and downstream transactions

        Returns:
        - complete traceability record
        """

        traceability_record = {
            'finished_product': {
                'lot_number': product_lot['lot_number'],
                'product_id': product_lot['product_id'],
                'production_date': product_lot['production_date'],
                'quantity': product_lot['quantity']
            },
            'one_up': [],  # Ingredients and packaging received
            'one_down': []  # Customers/locations shipped to
        }

        # One-up traceability (ingredients)
        for ingredient in product_lot.get('ingredients', []):
            traceability_record['one_up'].append({
                'supplier': ingredient['supplier'],
                'ingredient_id': ingredient['ingredient_id'],
                'lot_number': ingredient['lot_number'],
                'receive_date': ingredient['receive_date'],
                'quantity_used': ingredient['quantity_used']
            })

        # One-down traceability (shipments)
        lot_shipments = [
            e for e in supply_chain_events
            if e['type'] == 'shipment' and e['lot_number'] == product_lot['lot_number']
        ]

        for shipment in lot_shipments:
            traceability_record['one_down'].append({
                'customer': shipment['customer'],
                'ship_date': shipment['ship_date'],
                'quantity_shipped': shipment['quantity'],
                'destination': shipment['destination']
            })

        return traceability_record

    def execute_mock_recall(self, recalled_lot, traceability_data):
        """
        Execute mock recall to test traceability system

        Parameters:
        - recalled_lot: lot number being recalled
        - traceability_data: complete traceability records

        Returns:
        - recall execution report
        """

        start_time = datetime.now()

        # Find lot traceability
        lot_trace = traceability_data.get(recalled_lot, {})

        if not lot_trace:
            return {
                'success': False,
                'error': 'lot_not_found_in_traceability_system'
            }

        # Identify affected ingredients (one-up)
        affected_ingredients = lot_trace.get('one_up', [])

        # Identify affected customers (one-down

…

## Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [kishorkukreja](https://github.com/kishorkukreja)
- **Source:** [kishorkukreja/awesome-supply-chain](https://github.com/kishorkukreja/awesome-supply-chain)
- **License:** MIT

Install and usage instructions live in the source repository linked above.