AI Food Traceability System

AI-based food traceability system

Traceability is a key requirement for food safety: when a problem is identified, all affected batches must be localized within minutes. AI automates chain data collection and ensures that recalls are targeted rather than mass.

Traceability architecture

One-up/One-down principle:

Each participant in the chain knows who received the product from and who gave it to. AI connects all the links into a graph:

from datetime import datetime
import json
import hashlib

class TraceabilityRecord:
    """Запись прослеживаемости для партии продукта"""

    def __init__(self, lot_id, product_code, quantity_kg, timestamp=None):
        self.lot_id = lot_id
        self.product_code = product_code
        self.quantity_kg = quantity_kg
        self.timestamp = timestamp or datetime.now().isoformat()
        self.inputs = []          # из каких партий сделан (сырьё)
        self.processing_params = {}  # производственные параметры
        self.outputs = []         # в какие партии ушёл (полуфабрикат, готовая)
        self.shipments = []       # кому отгружен

    def add_input_lot(self, input_lot_id, quantity_used, quality_params=None):
        """Добавить входящую партию сырья"""
        self.inputs.append({
            'lot_id': input_lot_id,
            'quantity_kg': quantity_used,
            'quality': quality_params or {},
            'timestamp': datetime.now().isoformat()
        })

    def record_processing(self, params):
        """Записать производственные параметры"""
        self.processing_params = {
            **params,
            'recorded_at': datetime.now().isoformat()
        }

    def add_shipment(self, destination_id, quantity, transport_conditions=None):
        """Записать отгрузку"""
        self.shipments.append({
            'destination': destination_id,
            'quantity_kg': quantity,
            'transport': transport_conditions or {},
            'timestamp': datetime.now().isoformat()
        })

    def generate_lot_hash(self):
        """Хэш записи для верификации целостности"""
        data = json.dumps({
            'lot_id': self.lot_id,
            'inputs': self.inputs,
            'processing': self.processing_params
        }, sort_keys=True)
        return hashlib.sha256(data.encode()).hexdigest()


class TraceabilityGraph:
    """Граф прослеживаемости для анализа и отзывов"""

    def __init__(self):
        import networkx as nx
        self.graph = nx.DiGraph()

    def add_lot(self, lot: TraceabilityRecord):
        self.graph.add_node(lot.lot_id, data=lot.__dict__)
        for inp in lot.inputs:
            self.graph.add_edge(inp['lot_id'], lot.lot_id, quantity=inp['quantity_kg'])
        for shipment in lot.shipments:
            self.graph.add_edge(lot.lot_id, shipment['destination'], quantity=shipment['quantity_kg'])

    def recall_simulation(self, problem_lot_id):
        """Выявить все партии, затронутые отзывом"""
        import networkx as nx
        # Все потомки (вниз по цепи): куда ушёл проблемный продукт
        downstream = nx.descendants(self.graph, problem_lot_id)
        # Все предки (вверх по цепи): из какого сырья сделан
        upstream = nx.ancestors(self.graph, problem_lot_id)

        return {
            'problem_lot': problem_lot_id,
            'downstream_lots': list(downstream),
            'upstream_lots': list(upstream),
            'total_kg_at_risk': sum(
                self.graph.nodes[lot]['data']['quantity_kg']
                for lot in downstream
                if lot in self.graph.nodes
            )
        }

Integration with CHESTNY ZANK

Food product labeling:

• Milk and dairy products: from June 1, 2021 (Resolution 1550) - Packaged water: from 2022 - Business: registration in the GIS MT, printing of DataMatrix codes

AI automation of labeling: - Generating a code request → receiving codes → applying them to packaging - Automatic confirmation of entry into circulation upon shipment - Reconciliation: codes in the system vs. physical presence → identifying discrepancies

Analytics and problem prevention

Predictive quality control:

ML model on production data + recall history: - Process parameters out of tolerance → what % of these batches have historically resulted in complaints? - Proactive quarantine: the batch undergoes enhanced inspection before release - Economics: 1 preventive quarantine > 1 recall (10–50 times cheaper)

Cold chain monitoring:

IoT loggers in transport and refrigeration: - Temperature violation → automatic expiration date update - ML calculation of Mean Kinetic Temperature (MKT) based on temperature history - Recalculation of the remaining shelf life of the batch

Development period: 4–6 months for a traceability system with CHESTNY ZNAK integration, a traceability graph, and cold chain monitoring.