Разработка AI-агента для автоматизации бизнес-процессов

Разработка AI-агента для автоматизации бизнес-процессов

AI-агент для BPA (Business Process Automation) обрабатывает повторяющиеся многошаговые задачи: сбор данных из нескольких систем, принятие структурированных решений по правилам, заполнение документов, маршрутизация задач, уведомления. В отличие от классической RPA, LLM-агент понимает неструктурированный контент и адаптируется к вариациям входных данных.

Типовые бизнес-процессы для AI-агента

Обработка входящих заявок: агент разбирает email/форму, классифицирует, проверяет данные, маршрутизирует нужному исполнителю, создаёт задачу в трекере.

Онбординг нового сотрудника: создание учётных записей в системах, рассылка приветственных материалов, назначение ментора, постановка начальных задач.

Обработка счетов на оплату: извлечение реквизитов из PDF/фото, сверка с договором, проверка бюджета, маршрутизация на подпись, отметка в ERP.

Ежемесячная отчётность: сбор данных из нескольких источников, расчёт показателей, формирование отчёта, рассылка стейкхолдерам.

Архитектура агента для обработки заявок

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import TypedDict, Optional

class ApplicationState(TypedDict):
    raw_input: str              # Входящий текст заявки
    applicant_name: str
    application_type: str       # тип заявки
    extracted_data: dict        # извлечённые данные
    validation_result: dict     # результат проверки
    routing_decision: str       # куда направить
    task_id: Optional[str]      # ID созданной задачи
    notification_sent: bool

llm = ChatOpenAI(model="gpt-4o", temperature=0)

def classify_and_extract(state: ApplicationState) -> ApplicationState:
    """Классификация заявки и извлечение данных"""
    response = llm.invoke(f"""Проанализируй входящую заявку и извлеки структурированные данные.

Заявка:
{state['raw_input']}

Верни JSON:
{{
  "application_type": "vacation|expense|equipment|access|other",
  "applicant_name": "...",
  "department": "...",
  "details": {{}},  // специфичные поля по типу
  "urgency": "normal|urgent|critical",
  "missing_info": []  // чего не хватает
}}""")

    import json
    data = json.loads(response.content)
    return {
        **state,
        "application_type": data["application_type"],
        "applicant_name": data.get("applicant_name", ""),
        "extracted_data": data,
    }

def validate_application(state: ApplicationState) -> ApplicationState:
    """Проверка полноты и соответствия политикам"""
    app_type = state["application_type"]
    extracted = state["extracted_data"]

    validation = {"valid": True, "errors": [], "warnings": []}

    if app_type == "vacation":
        # Проверяем остаток отпуска
        days = extracted["details"].get("days", 0)
        balance = hr_api.get_vacation_balance(state["applicant_name"])
        if days > balance:
            validation["valid"] = False
            validation["errors"].append(f"Недостаточно дней отпуска: запрошено {days}, доступно {balance}")

    elif app_type == "expense":
        amount = extracted["details"].get("amount", 0)
        if amount > 50000:  # Лимит самостоятельного согласования
            validation["warnings"].append("Требует согласования руководителя")

    return {**state, "validation_result": validation}

def route_application(state: ApplicationState) -> ApplicationState:
    """Определение маршрута обработки"""
    app_type = state["application_type"]
    validation = state["validation_result"]
    urgency = state["extracted_data"].get("urgency", "normal")

    if not validation["valid"]:
        routing = "reject_with_explanation"
    elif app_type == "vacation":
        routing = "hr_manager"
    elif app_type == "expense" and state["extracted_data"]["details"].get("amount", 0) > 50000:
        routing = "director_approval"
    elif app_type == "access":
        routing = "it_department"
    else:
        routing = "auto_approve"

    return {**state, "routing_decision": routing}

def execute_routing(state: ApplicationState) -> ApplicationState:
    """Выполнение маршрутизации"""
    routing = state["routing_decision"]

    if routing == "auto_approve":
        task_id = jira_api.create_task(
            title=f"Авто-одобрено: {state['application_type']} от {state['applicant_name']}",
            status="Done",
            assignee="system",
        )
    elif routing in ["hr_manager", "director_approval", "it_department"]:
        assignee_map = {
            "hr_manager": "[email protected]",
            "director_approval": "[email protected]",
            "it_department": "[email protected]",
        }
        task_id = jira_api.create_task(
            title=f"Заявка на {state['application_type']} от {state['applicant_name']}",
            assignee=assignee_map[routing],
            description=json.dumps(state["extracted_data"], ensure_ascii=False),
            priority="High" if state["extracted_data"].get("urgency") == "urgent" else "Normal",
        )
    else:
        task_id = None

    notification_service.send(
        to=state["applicant_name"],
        message=f"Ваша заявка принята. Routing: {routing}. ID: {task_id}"
    )

    return {**state, "task_id": task_id, "notification_sent": True}

# Построение графа процесса
graph = StateGraph(ApplicationState)
graph.add_node("classify_and_extract", classify_and_extract)
graph.add_node("validate", validate_application)
graph.add_node("route", route_application)
graph.add_node("execute", execute_routing)

graph.set_entry_point("classify_and_extract")
graph.add_edge("classify_and_extract", "validate")
graph.add_edge("validate", "route")
graph.add_edge("route", "execute")
graph.add_edge("execute", END)

application_agent = graph.compile()

Практический кейс: обработка входящих счетов

Задача: 180+ счетов на оплату в месяц. До автоматизации: 3.5 часа в день главного бухгалтера.

Пайплайн агента:

1. Извлечение текста из PDF (pdfplumber / LlamaParse)
2. LLM извлекает: поставщик, ИНН, сумма, НДС, дата, номер, договор
3. Сверка с реестром договоров (векторный поиск)
4. Проверка в 1С: остаток по договору, бюджет статьи
5. При OK → создание платёжного поручения в 1С
6. При расхождении → задача бухгалтеру с объяснением

Метрики после 3 месяцев:

• Автоматически обработано без вмешательства: 73%
• Точность извлечения реквизитов: 96%
• Ошибки (неправильная привязка к договору): 1.2%
• Экономия времени: 2.5 часа/день

Human-in-the-loop: когда агент запрашивает подтверждение

def requires_human_approval(state: ApplicationState) -> bool:
    """Определяет, нужно ли вмешательство человека"""
    return (
        not state["validation_result"]["valid"] or
        state["extracted_data"].get("amount", 0) > 100000 or
        state["application_type"] == "termination" or
        state["extracted_data"].get("urgency") == "critical"
    )

# В LangGraph: interrupt_before для HITL
agent = graph.compile(
    interrupt_before=["execute"],  # Прерывание перед выполнением
    checkpointer=MemorySaver(),
)

Сроки

• Анализ бизнес-процесса и проектирование: 1–2 недели
• Разработка агента с основными интеграциями: 3–5 недель
• Тестирование и HITL настройка: 1–2 недели
• Итого: 5–9 недель