Архитектура Agentic AI в ERP: Реализация мультиагентной закупки на LangGraph

От теории к практической реализации интеллектуального модуля Procure-to-Pay с использованием паттерна Supervisor и параллельным выполнением агентов.

Проблема традиционной автоматизации

Современные ERP-системы (SAP, Oracle, 1С) используют RPA для автоматизации рутинных операций, но эти решения сталкиваются с фундаментальными ограничениями при обработке сложных закупочных сценариев:

• Отсутствие контекстуальной памяти: RPA-боты не хранят историю взаимодействий, что критично при переговорах с поставщиками.
• Жёсткие流程: Каждое отклонение от сценария (например, отсутствие товара) требует ручного вмешательства.
• Нет итеративного планирования: Системы не могут генерировать и оценивать альтернативные варианты (замена поставщика, изменение сроков).
• Изоляция данных: Информация о бюджете, контрактах и рисках хранится в разных модулях без единого контекста.

В результате время согласования сложной закупки занимает 2–5 рабочих дней, а не 5–10 минут, как в идеале. Agentic AI, построенный на графах состояний, решает эти проблемы через:

1. Состояние (State): Единый контекст (запрос, бюджет, поставщики, решения) передаётся между агентами.
2. Параллельное выполнение: Несколько агентов работают одновременно (например, Scout и Auditor).
3. Итеративные циклы: Возможность возврата к предыдущим шагам при недостатке данных.
4. Human-in-the-loop: Автоматический эскалация при низкой уверенности модели.

Архитектура мультиагентной системы: Supervisor + Specialized Agents

Мы используем паттерн Supervisor, где Orchestrator выступает как диспетчер, распределяющий задачи между специализированными агентами. Каждый агент — это LLM (например, GPT-4o или Claude 3.5 Sonnet) с системным промптом и набором инструментов (Tools).

Реализация на LangGraph: StateGraph и параллельные узлы

LangGraph позволяет создавать циклические графы, где состояние (State) передаётся между узлами. Вот упрощённый пример для Procure-to-Pay:

from langgraph.graph import StateGraph, END
from typing import TypedDict, List, Annotated
import operator

# Определение состояния
class ProcurementState(TypedDict):
    request: dict                    # Исходный запрос на закупку
    suppliers: List[dict]            # Найденные поставщики
    risks: dict                      # Оценка рисков
    budget_check: dict               # Результат проверки бюджета
    decision: str                    # "approve", "reject", "escalate"
    confidence: float                # Уверенность в решении (0-1)
    human_intervention: bool         # Нужно ли вмешательство человека

# Узлы (агенты)
def scout_node(state: ProcurementState):
    """Поиск поставщиков через внешние API"""
    suppliers = search_suppliers(state["request"]["item"])
    return {"suppliers": suppliers}

def auditor_node(state: ProcurementState):
    """Оценка рисков для каждого поставщика"""
    risks = {}
    for supplier in state["suppliers"]:
        risk_score = calculate_risk_score(supplier["id"])
        risks[supplier["id"]] = risk_score
    return {"risks": risks}

def budget_node(state: ProcurementState):
    """Проверка бюджета"""
    total = sum(s["price"] for s in state["suppliers"])
    check = check_budget_balance(state["request"]["cost_center"], total)
    return {"budget_check": check}

def orchestrator_node(state: ProcurementState):
    """Синтез и принятие решения"""
    # Взвешивание факторов
    avg_risk = sum(state["risks"].values()) / len(state["risks"])
    budget_ok = state["budget_check"]["available"] > state["budget_check"]["required"]
    
    # Простая логика (в реальности — ML-модель)
    if avg_risk < 30 and budget_ok:
        decision = "approve"
        confidence = 0.9
    elif avg_risk > 70:
        decision = "reject"
        confidence = 0.85
    else:
        decision = "escalate"
        confidence = 0.6
    
    return {
        "decision": decision,
        "confidence": confidence,
        "human_intervention": confidence < 0.8
    }

# Построение графа
workflow = StateGraph(ProcurementState)

# Добавление узлов
workflow.add_node("scout", scout_node)
workflow.add_node("auditor", auditor_node)
workflow.add_node("budget", budget_node)
workflow.add_node("orchestrator", orchestrator_node)

# Параллельное выполнение scout и auditor
workflow.add_edge("scout", "auditor")
workflow.add_edge("auditor", "budget")
workflow.add_edge("budget", "orchestrator")
workflow.add_edge("orchestrator", END)

# Компиляция
app = workflow.compile()

# Запуск
result = app.invoke({
    "request": {"item": "ноутбук Dell XPS 15", "quantity": 10, "cost_center": "IT-2024"}
})
print(f"Решение: {result['decision']} (уверенность: {result['confidence']:.2f})")

Ключевые особенности реализации

Инструменты и API: Подключение внешних данных

Каждый агент использует набор инструментов (Tools) — функций, которые вызываются LLM для получения данных. Важно: инструменты должны быть идемпотентными и иметь валидацию.

Пример инструмента для Scout

from langchain.tools import tool
import requests

@tool
def search_suppliers(item: str, location: str = "RU") -> List[dict]:
    """Поиск поставщиков по каталогу. Возвращает список словарей с id, name, price, delivery_time."""
    response = requests.get(
        "https://api.supplier-catalog.com/v1/search",
        params={"q": item, "country": location},
        headers={"Authorization": "Bearer API_KEY"}
    )
    response.raise_for_status()
    suppliers = response.json()["suppliers"]
    
    # Фильтрация и сортировка
    filtered = [s for s in suppliers if s["in_stock"] > 0]
    return sorted(filtered, key=lambda x: x["price"])[:10]  # Топ-10

Интеграция с ERP-модулями

Инструменты должны взаимодействовать с существующими ERP-системами через:

• REST/GraphQL API: Для модулей бюджетирования, склада, контрактов.
• SOAP (legacy): Для старых ERP (SAP ECC). Используйте адаптеры.
• Прямой доступ к БД: Только для чтения (реплика) с строгими правами.
• События (Kafka/RabbitMQ): Для асинхронных уведомлений (например, "поставка задержана").

Безопасность и Аудит: Валидация и логирование

Agentic AI должен соответствовать требованиям compliance (GDPR, ФЗ-152, SOX). Ключевые практики:

1. Валидация перед записью

Никогда не доверяйте решениям агента без проверки. Используйте двухэтапную валидацию:

1. Схемы данных (Pydantic): Все входные/выходные данные агентов должны описываться через Pydantic модели. Пример:

from pydantic import BaseModel, Field
from typing import List

class Supplier(BaseModel):
    id: str = Field(..., description="Идентификатор поставщика в ERP")
    name: str
    price: float = Field(gt=0, description="Цена за единицу")
    delivery_days: int = Field(ge=1, description="Срок поставки в днях")
    risk_score: int = Field(ge=0, le=100, description="Скор риска от Auditor")

2. Аудит всех шагов

Логируйте каждое действие агента в отдельную таблицу ai_audit_log:

3. RBAC для эскалации

Human-in-the-loop запросы направляются не любому оператору, а только тем, у кого есть роль procurement_approver и права на конкретный центр затрат. Интеграция через workflow-систему (Camunda, Temporal):

• Orchestrator создаёт задачу в Camunda с данными запроса.
• Camunda назначает задачу approver'у на основе политик (например, сумма > 1M руб → CFO).
• После решения оператора Camunda возвращает управление графу.

Кейс: Sudden Outage — как агенты реагируют на срыв поставки

Сценарий: Закупка критичного компонента для производства. За 2 дня до поставки основной поставщик сообщает о форс-мажоре (забастовка).