Розробка AI-цифрового логіста (AI Logistics Agent)

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

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

Моніторинг та трекинг відправлень

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)

Маршрутизація та оптимізація

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)

Обробка виключень

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 звіт по логістиці.

Метрики:
- 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%}

Формат: коротке резюме (3 речення), відхилення від норми, рекомендації.""")

        return report.content

Практичний кейс: дистрибьютор FMCG, 500 відправлень/день

Ситуація: 4 логістів, 500 відправлень на день, оператори витрачали 70% часу на трекинг та обробку виключень.

AI Агент взяв на себе:

• Автоматичний трекинг кожні 2 години через API перевозників
• Сповіщення адресатам про статус та затримки
• Автоматичне відкриття претензій при пошкодженнях
• Перебронювання при затримках > 4 годин (критичні відправлення)
• Щоденний KPI-звіт та щотижневий carrier scorecard

Результати:

• Ручна обробка виключень: 180 випадків/день → 45 (агент закрив автоматично)
• On-time delivery rate: 82% → 89% (краща маршрутизація + проактивне управління виключеннями)
• Претензії, що підтверджені (правильно оформлені): 62% → 91%
• Час логістів на оперативну роботу: -55%

Терміни

• Трекинг та моніторинг відправлень: 2–3 тижні
• Обробник виключень з playbooks: 2–3 тижні
• Маршрутизація та оптимізація: 2–3 тижні
• Analytics та KPI-звіти: 1 тиждень
• Інтеграція з перевозниками: 1–2 тижні
• Всього: 8–12 тижнів