AI-система прослеживаемости продуктов питания
Прослеживаемость — ключевое требование пищевой безопасности: при выявлении проблемы необходимо за минуты локализовать все затронутые партии. AI автоматизирует сбор данных цепочки и делает отзыв точечным, а не массовым.
Архитектура прослеживаемости
One-up/One-down принцип:
Каждый участник цепи знает, от кого получил и кому отдал продукт. AI связывает все звенья в граф:
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
)
}
Интеграция с ЧЕСТНЫЙ ЗНАК
Маркировка пищевой продукции:
- Молоко и молочные продукты: с 1.06.2021 (Постановление 1550)
- Вода упакованная: с 2022
- Бизес: регистрация в ГИС МТ, печать кодов DataMatrix
AI-автоматизация работы с маркировкой:
- Генерация заявки на коды → приём кодов → нанесение на упаковку
- Автоматическое подтверждение ввода в оборот при отгрузке
- Reconciliation: коды в системе vs. физическое наличие → выявление расхождений
Аналитика и предотвращение проблем
Predictive quality control:
ML-модель на производственных данных + история отзывов:
- Параметры процесса, выходящие за допуск → какой % этих партий исторически приводил к жалобам?
- Проактивный карантин: партия проходит усиленный контроль до выпуска
- Экономика: 1 превентивный карантин > 1 отзыв с рынка (в 10–50 раз дешевле)
Мониторинг холодовой цепи:
IoT-логгеры в транспорте и холодильниках:
- Нарушение температурного режима → автоматическое обновление срока годности
- ML-расчёт Mean Kinetic Temperature (MKT) по истории температур
- Пересчёт остаточного срока годности партии
Срок разработки: 4–6 месяцев для системы прослеживаемости с интеграцией ЧЕСТНЫЙ ЗНАК, графом прослеживаемости и cold chain мониторингом.