AI-агент логістики — цифровий співробітник

AI Агент логістики — цифровий співробітник

Логістичні компанії витрачають до 70% часу операторів на моніторинг відправлень та розбір винятків. При 100 відправленнях на день оператор вручну перевіряє статуси 5–10 разів, на що йде до 4 годин. Кожна затримка знижує лояльність клієнтів і збільшує штрафи. Ми розробили AI-агента, який автоматизує операційні завдання ланцюга постачання: планування маршрутів, трекінг відправлень, взаємодія з перевізниками, обробка винятків (затримки, пошкодження, недостачі), моніторинг KPI та генерація звітів. Оператор-людина підключається тільки до нестандартних ситуацій, що потребують переговорів. Результат — скорочення ручної праці на 55% і зростання on-time delivery до 89%.

Як AI-агент знижує навантаження на логістів?

Агент працює в три етапи: збір даних, аналіз за допомогою LLM та автоматичне виконання дій. Для трекінгу він підключається до API перевізників (DPD, СДЭК, ПЭК та ін.) кожні 2 години і звіряє фактичний статус із очікуваним. При виявленні затримки система розраховує час простою та приймає рішення: повідомити отримувача, перебронювати доставку або ескалювати менеджеру. У середньому агент обробляє виняток у 3 рази швидше за людину.

from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from typing import TypedDict, Annotated, Optional import operator llm = ChatOpenAI(model="gpt-4o", temperature=0) class ShipmentState(TypedDict): shipment_id: str shipment_data: dict tracking_history: list[dict] anomalies: Annotated[list, operator.add] actions_taken: Annotated[list, operator.add] escalation_required: bool escalation_reason: Optional[str] class ShipmentMonitor: async def check_shipment(self, shipment_id: str) -> ShipmentState: """Повна перевірка статусу відправлення""" # Отримуємо дані shipment = await logistics_db.get_shipment(shipment_id) tracking = await carrier_api.get_tracking(shipment["tracking_number"]) expected_eta = shipment["expected_delivery"] current_eta = tracking.get("estimated_delivery") anomalies = [] # Перевірка затримки if current_eta and current_eta > expected_eta: delay_hours = (current_eta - expected_eta).total_seconds() / 3600 anomalies.append({ "type": "delivery_delay", "severity": "high" if delay_hours > 24 else "medium", "details": f"Затримка {delay_hours:.0f} годин, нова дата: {current_eta}", }) # Відсутність оновлень last_update = tracking.get("last_event_time") hours_since_update = (datetime.now() - last_update).total_seconds() / 3600 if last_update else 99 if hours_since_update > 48: anomalies.append({ "type": "no_tracking_update", "severity": "medium", "details": f"Немає оновлень {hours_since_update:.0f} годин", }) # LLM-аналіз при аномаліях escalation_required = False escalation_reason = None if anomalies: assessment = await self.assess_anomalies(shipment, anomalies) escalation_required = assessment["requires_escalation"] escalation_reason = assessment.get("reason") return ShipmentState( shipment_id=shipment_id, shipment_data=shipment, tracking_history=tracking.get("events", []), anomalies=anomalies, actions_taken=[], escalation_required=escalation_required, escalation_reason=escalation_reason, ) async def assess_anomalies(self, shipment: dict, anomalies: list) -> dict: """LLM оцінює необхідність ескалації""" response = await llm.ainvoke(f"""Оціни ситуацію з відправленням. Відправлення: {json.dumps(shipment, ensure_ascii=False)} Аномалії: {json.dumps(anomalies, ensure_ascii=False)} Визнач: 1. Чи потребує ситуація негайної ескалації менеджеру? 2. Які автоматичні дії можна вжити? 3. Чи потрібно повідомляти отримувача? Поверни JSON: {{"requires_escalation": bool, "reason": "...", "auto_actions": [...], "notify_recipient": bool}}""") return json.loads(response.content) 

Маршрутизація використовує гібридний підхід: для малих задач (до 20 точок) — LLM з chain-of-thought, для великих — алгоритмічний вирішувач з постобробкою через AI. Це дає оптимальні маршрути з урахуванням часових вікон, вантажопідйомності та заторів. Детальніше про Vehicle Routing Problem.

class RouteOptimizer: async def optimize_delivery_routes( self, deliveries: list[dict], # [{id, address, time_window, weight}] vehicles: list[dict], # [{id, capacity, location}] date: str, ) -> dict: """Оптимізує маршрути доставки (Vehicle Routing Problem)""" # Для малих задач — через LLM з роздумом if len(deliveries) <= 20: return await self.llm_route_optimizer(deliveries, vehicles) # Для великих — алгоритмічний підхід + LLM для винятків return await self.algorithmic_route_optimizer(deliveries, vehicles) async def llm_route_optimizer(self, deliveries: list, vehicles: list) -> dict: response = await llm.ainvoke(f"""Склади оптимальні маршрути доставки. Доставки: {json.dumps(deliveries, ensure_ascii=False, indent=2)} Транспортні засоби: {json.dumps(vehicles, ensure_ascii=False, indent=2)} Врахуй: часові вікна, вантажопідйомність, мінімізацію загального пробігу. Поверни JSON: {{"routes": [{{"vehicle_id": "...", "stops": [delivery_ids_in_order]}}]}}""") return json.loads(response.content) 

Обробка винятків побудована на playbooks — попередньо визначених сценаріях для типових ситуацій: затримка, пошкодження, митна затримка, неправильна адреса. Агент виконує автоматичні дії (повідомлення, створення претензій, перебронювання) і ескалює тільки коли автоматика безсила.

class ExceptionHandler: EXCEPTION_PLAYBOOKS = { "delivery_delay": { "auto_actions": ["notify_recipient", "update_crm", "rebook_if_urgent"], "escalate_if": lambda hours: hours > 72, }, "damaged_goods": { "auto_actions": ["create_claim", "notify_sender", "photo_request"], "escalate_always": True, }, "customs_hold": { "auto_actions": ["get_customs_details", "notify_broker"], "escalate_if": lambda days: days > 3, }, "address_not_found": { "auto_actions": ["contact_recipient", "check_database"], "escalate_if": lambda attempts: attempts > 2, }, } async def handle_exception(self, exception: dict) -> dict: exception_type = exception["type"] playbook = self.EXCEPTION_PLAYBOOKS.get(exception_type) if not playbook: return await self.generic_exception_handler(exception) actions_taken = [] # Виконуємо автоматичні дії for action in playbook.get("auto_actions", []): result = await self.execute_action(action, exception) actions_taken.append({"action": action, "result": result}) # Перевіряємо чи потрібна ескалація escalate = playbook.get("escalate_always", False) if not escalate and "escalate_if" in playbook: escalate_fn = playbook["escalate_if"] # Викликаємо з потрібним параметром із даних винятку escalate = escalate_fn(exception.get("delay_hours") or exception.get("hold_days") or exception.get("attempts", 0)) if escalate: await self.escalate_to_manager(exception, actions_taken) return {"actions_taken": actions_taken, "escalated": escalate} 

Analytics-модуль щоденно збирає ключові метрики та генерує KPI-звіт природною мовою. У звіті — відхилення від норми та рекомендації щодо покращення.

class LogisticsAnalytics: async def daily_kpi_report(self) -> str: """Щоденний звіт по KPI логістики""" # Дані з БД metrics = await asyncio.gather( self.get_on_time_delivery_rate(), self.get_damage_rate(), self.get_carrier_performance(), self.get_cost_per_shipment(), self.get_exception_rate(), ) report = await llm.ainvoke(f"""Створи KPI-звіт по логістиці за {datetime.now().strftime('%d.%m.%Y')}. Метрики: - On-time delivery: {metrics[0]['rate']:.1%} (ціль: {metrics[0]['target']:.1%}) - Damage rate: {metrics[1]['rate']:.3%} - Top carriers by performance: {metrics[2]} - Cost per shipment: {metrics[3]['avg']:,.0f} руб - Exception rate: {metrics[4]['rate']:.2%} Формат: короткий summary (3 речення), відхилення від норми, рекомендації.""") return report.content 

Що дає гібридна маршрутизація?

Гібридний підхід дозволяє обробляти до 500 точок доставки за хвилину, поєднуючи швидкість алгоритмів та гнучкість LLM. Для стандартних маршрутів з часовими вікнами використовується алгоритмічний оптимізатор, а для нестандартних запитів (термінова доставка, зміна пріоритетів) підключається LLM з роздумом. Це скорочує середній час маршрутизації на 40% у порівнянні з чисто алгоритмічним підходом.

Результати впровадження: скорочення ручної праці на 55%

Порівняння часу обробки винятків:

Ці цифри — з нашого практичного кейсу з дистриб'ютором FMCG, який обробляє 500 відправлень на день. Наші фахівці мають значний досвід в інтеграції AI-рішень у логістику, використовують перевірені стеки (LangGraph, OpenAI GPT-4o, ChromaDB) та дають гарантію на кожен етап.

Етапи впровадження

1. Аудит поточних процесів та інтеграцій — 1 тиждень.
2. Підключення до API перевізників та складів — 1–2 тижні.
3. Розробка трекінг-модуля та playbooks винятків — 2–3 тижні.
4. Налаштування маршрутизації та KPI-аналітики — 2–3 тижні.
5. Тестування на історичних даних та pilot-запуск — 1–2 тижні.
6. Повноцінний запуск та навчання операторів — 1 тиждень.

Підсумковий термін: 8–12 тижнів залежно від кількості перевізників та складності винятків.

Що входить у результат?

• Працюючий AI-агент у вашій інфраструктурі (on-prem або хмара).
• Інтеграція з 3–5 перевізниками (розширюється за запитом).
• Playbook для 5+ типів винятків (з можливістю доналаштування).
• Щоденний KPI-звіт українською мовою.
• Документація та навчання команди операторів.
• Підтримка протягом 30 днів після запуску.

Ми гарантуємо зниження ручної обробки винятків мінімум на 50%. Окупність рішення — від 6 місяців при обсязі від 200 відправлень на день.

Для оцінки придатності AI-агента у ваших процесах зв'яжіться з нами. Отримайте консультацію щодо впровадження — ми проаналізуємо ваші логістичні процеси та запропонуємо оптимальне рішення під ключ.