Розробка 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}. 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. Перевірка в ERP: залишок по договору, бюджет статті
5. При OK → створення платіжного доручення в ERP
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 тижнів