AI-система відстеження харчових продуктів
Ми стикалися з ситуацією, коли відклик партії молочної продукції зачепив 15% ринку — кожна година простою коштувала виробнику мільйони рублів. Відсутність точкового відстеження перетворювала локальний дефект на масштабну кризу. З AI та графовим аналізом час локалізації проблемної партії скорочується з тижнів до 15 хвилин. Економія на одному такому інциденті — до 95% витрат на відклик. Нижче — як ми будуємо такі системи і які проблеми вирішуємо.
Проблеми, які вирішуємо
Масові відклики через неточну локалізацію. Традиційний one-up/one-down облік не дозволяє швидко побудувати повний граф зачеплених партій. AI-агент автоматично збирає дані від усіх учасників ланцюга та за секунди обчислює підмножину ризику. Графовий аналіз на основі NetworkX обробляє ланцюг із 10 млн вузлів у 100 разів швидше за SQL-запити.
Ручне маркування та помилки введення. При інтеграції з «Чесним знаком» коди DataMatrix потрібно наносити, підтверджувати введення в обіг і звіряти з фізичною наявністю. Ми автоматизуємо генерацію заявок, прийом кодів та reconciliation, виключаючи людський фактор. Розбіжності знижуються з 5–10% до 0,5%.
Невраховані порушення холодового ланцюга. Транспортування та зберігання з порушеннями температури скорочують термін придатності. Наша ML-модель розраховує Mean Kinetic Temperature (MKT) за даними IoT-логерів та перераховує залишковий термін кожної партії в реальному часі. (Розрахунок MKT за стандартом ICH Q1A)
Архітектура відстеження
Граф пов'язує всіх учасників — від постачальника сировини до полиці магазину. Код нижче реалізує базові класи для запису партії та побудови графа.
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 ) } Як AI пов'язує ланцюг постачання?
Кожен учасник — від фермера до дистриб'ютора — передає в граф свої дані через API або завантаження файлів. ML-модель очищає та нормалізує записи, автоматично визначаючи дублікати та невідповідності. Потім граф аналізується на предмет циклів (замкнених поставок) та ізольованих вузлів, які часто вказують на прогалини в даних. Це дає повну картину руху продуктів. Час побудови графа для 10 млн вузлів — менше 3 секунд (p99).
Чому інтеграція з ЧЕСНИЙ ЗНАК критична?
Згідно з вимогами регулятора, молочна продукція, вода та інші категорії підлягають обов'язковому маркуванню. Наша AI-система автоматизує:
- генерацію заявки на коди в ГІС МТ;
- прийом кодів та їх нанесення на упаковку (з контролем позиціонування);
- підтвердження введення в обіг при відвантаженні;
- автоматичну звірку — reconciliation кодів у системі та фізичних відвантажень.
Якщо розбіжність перевищує 2%, система блокує відвантаження та повідомляє відповідального. За рахунок автоматизації ми скорочуємо час на маркування на 80%.
Порівняння підходів
| Аспект | Традиційний облік | AI-відстеження |
|---|---|---|
| Час пошуку партії | дні–тижні | хвилини |
| Точність відклику | карантин усієї партії | точкова заміна |
| Прогнозування ризиків | немає | ML-модель якості |
| Інтеграція з маркуванням | ручне введення | автоматична |
| Ціна помилки | до 10% виручки | 0.1% виручки |
Типові ризики та їх нейтралізація
| Проблема | Рішення AI | Ефект |
|---|---|---|
| Забруднення сировини | ML-модель предиктивної якості | зниження браку на 70% |
| Помилки в DataMatrix | CV-контроль нанесення | відсоток помилок < 0,1% |
| Розрив холодового ланцюга | MKT-моніторинг у реальному часі | збільшення терміну придатності на 15% |
Процес роботи
- Аналітика та аудит — вивчаємо поточні процеси, дані, маркування (2–3 тижні).
- Проектування архітектури — обираємо стек: Python, NetworkX, PostgreSQL + DataMatrix генератор, ML-сервер (PyTorch/TensorFlow), векторна БД для логів.
- Реалізація — розробляємо граф відстеження, ML-модель якості, API інтеграції з обліковою системою та ГІС МТ.
- Тестування — стрес-тестуємо на історичних даних: симуляція відклику, перевірка часу відповіді (p99 не більше 3 секунд для графа з 10 млн вузлів).
- Деплой та навчання — розгортаємо на інфраструктурі замовника (on-prem або хмара), навчаємо операторів.
Типові терміни етапів
- Аудит: 2–3 тижні
- MVP: 2–3 місяці
- Повна система з інтеграцією: 4–6 місяців
- Підтримка та донавчання: щомісячно
Що входить в роботу
Ми передаємо:
- документацію з архітектури та API;
- навчання персоналу (2–3 дні);
- код графа та ML-моделі;
- дашборди в Power BI для моніторингу;
- SLA на підтримку від 3 місяців.
Терміни та вартість
Орієнтовний час реалізації — від 4 до 6 місяців, залежно від обсягу даних та кількості інтеграцій. Вартість розраховується індивідуально після аудиту: врахування кількості SKU, точок відвантаження та необхідної точності ML. Отримайте консультацію — оцінимо ваш проект за 3 робочих дні. Замовте аудит прямо зараз, щоб знизити ризики масових відкликів.
Ми гарантуємо зниження часу локалізації проблемної партії до 15 хвилин та скорочення обсягів відклику на 95% відносно масового підходу.