AI Food Traceability System

We encountered a situation where a recall of dairy products affected 15% of the market—every hour of downtime cost the manufacturer millions of rubles. The lack of pinpoint traceability turned a local defect into a large-scale crisis. With AI and graph analysis, the time to locate a problematic lot is reduced from weeks to 15 minutes. Savings on a single such incident are up to 95% of recall costs. Below is how we build such systems and what problems we solve.

Problems We Solve

Mass recalls due to inaccurate localization. Traditional one-up/one-down accounting does not allow quick building of a complete graph of affected lots. An AI agent automatically collects data from all supply chain participants and calculates the risk subset in seconds. Graph analysis based on NetworkX processes a chain of 10 million nodes 100 times faster than SQL queries.

Manual labeling and input errors. When integrating with Chestny Znak, DataMatrix codes need to be applied, entry into circulation confirmed, and reconciled with physical presence. We automate request generation, code reception, and reconciliation, eliminating human error. Discrepancies drop from 5–10% to 0.5%.

Unaccounted cold chain breaches. Transport and storage with temperature violations reduce shelf life. Our ML model calculates Mean Kinetic Temperature (MKT) from IoT logger data and recalculates the remaining shelf life of each lot in real time. (MKT calculation per ICH Q1A standard)

Traceability Architecture

The graph links all participants—from raw material supplier to store shelf. The code below implements basic classes for recording a lot and building the 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
            )
        }

How AI Connects the Supply Chain?

Each participant—from farmer to distributor—submits their data to the graph via API or file uploads. An ML model cleans and normalizes records, automatically detecting duplicates and inconsistencies. The graph is then analyzed for cycles (closed loops) and isolated nodes, which often indicate data gaps. This provides a complete picture of product movement. Graph build time for 10 million nodes is under 3 seconds (p99).

Why is Chestny Znak Integration Critical?

According to regulatory requirements, dairy products, water, and other categories are subject to mandatory labeling. Our AI system automates:

• generating a code request in GIS MT;
• receiving codes and applying them to packaging (with position control);
• confirming entry into circulation upon shipment;
• automatic reconciliation—matching codes in the system with physical shipments.

If the discrepancy exceeds 2%, the system blocks shipment and notifies the responsible person. Through automation, we reduce labeling time by 80%.

Comparison of Approaches

Typical Risks and Their Mitigation

Process of Work

1. Analysis and audit — study current processes, data, labeling (2–3 weeks).
2. Architecture design — choose stack: Python, NetworkX, PostgreSQL + DataMatrix generator, ML server (PyTorch/TensorFlow), vector DB for logs.
3. Implementation — develop traceability graph, ML quality model, API for integration with accounting system and GIS MT.
4. Testing — stress test on historical data: recall simulation, response time check (p99 no more than 3 seconds for a 10M-node graph).
5. Deployment and training — deploy on client infrastructure (on-prem or cloud), train operators.

What's Included in the Work

We deliver:

• architecture and API documentation;
• staff training (2–3 days);
• graph and ML model code;
• Power BI dashboards for monitoring;
• SLA for support from 3 months.

Timeline and Cost

Estimated implementation time — from 4 to 6 months, depending on data volume and number of integrations. Cost is calculated individually after audit: considering number of SKUs, shipment points, and required ML accuracy. Get a consultation — we will evaluate your project within 3 business days. Order an audit now to reduce the risk of mass recalls.

We guarantee reduction of problem lot localization time to 15 minutes and reduction of recall volume by 95% compared to a mass approach.