AI Workflow з розгалуженням та умовною логікою

Розробка AI Workflow з розгалуженням та умовною логікою

При обробці 500+ вхідних документів щодня — рахунки, договори, скарги, заявки — критично миттєво маршрутизувати кожен документ до потрібного спеціаліста. Лінійний пайплайн тут неефективний: помилки класифікації сягають 15%, а час обробки одного документа — до 45 хвилин. В результаті термінові рахунки губляться, скарги зависають, а договори підписуються із затримками. Особливо гостро проблема стоїть у компаніях з високим документарним навантаженням: ритейл, логістика, фінанси. Щодня десятки спеціалістів вручну сортують документи — це не лише повільно, а й дорого: на сортування йде до 30% робочого часу. AI Workflow (також machine learning workflow) з умовним розгалуженням AI на базі LangGraph усуває ці втрати. Ми проєктуємо такі системи під ключ з гарантією результату: автообробка до 80% потоку, точність класифікації — від 90%, час маршрутизації — секунди замість годин.

Як обрати тип розгалуження для вашого workflow?

Усі методи розгалуження (conditional branching AI) діляться на три категорії. Порівняємо їх у таблиці:

Вибір типу залежить від задачі: для передбачуваних сценаріїв достатньо детермінованого, для складних неструктурованих — hybrid дає 2-3x приріст точності порівняно з чисто LLM-підходом. Також можна комбінувати з RAG pipeline для збагачення контексту.

Що таке hybrid-розгалуження і коли його застосовувати?

Hybrid-розгалуження об'єднує детерміновані правила та LLM-based рішення в одному workflow. Наприклад, при обробці рахунку детермінована нода перевіряє формат та числові поля, а LLM витягує неструктуровані дані (найменування товарів, умови). Якщо LLM не може знайти суму, спрацьовує retry-цикл зі зміненим промптом. Такий підхід знижує latency на 30% порівняно з чисто LLM-рішенням і забезпечує точність 95% навіть на зашумлених даних. Hybrid-розгалуження незамінне, коли в одному потоці зустрічаються і структуровані, і неструктуровані документи.

Реалізація з LangGraph

Згідно з офіційною документацією LangGraph, умовні ребра дозволяють маршрутизацію на основі стану. Розглянемо приклад графа для обробки документів:

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

class WorkflowState(TypedDict):
    input_document: str
    document_type: Optional[str]
    extracted_data: Optional[dict]
    validation_errors: list[str]
    processing_path: str
    output: Optional[dict]

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

# Нода 1: Класифікація документа
def classify_document(state: WorkflowState) -> WorkflowState:
    response = llm.invoke(f"""Визнач тип документа. Типи: invoice, contract, complaint, inquiry, other
Документ: {state['input_document'][:500]}
Відповідь одним словом:""")
    return {**state, "document_type": response.content.strip().lower()}

# Нода 2a: Обробка рахунку
def process_invoice(state: WorkflowState) -> WorkflowState:
    response = llm.invoke(f"""Витягни дані рахунку.
{state['input_document']}
Поверни JSON: {{vendor, amount, date, due_date, items}}""")
    return {**state, "extracted_data": json.loads(response.content), "processing_path": "invoice"}

# Нода 2b: Обробка договору
def process_contract(state: WorkflowState) -> WorkflowState:
    response = llm.invoke(f"""Витягни ключові умови договору.
{state['input_document']}
Поверни JSON: {{parties, subject, amount, duration, key_conditions}}""")
    return {**state, "extracted_data": json.loads(response.content), "processing_path": "contract"}

# Нода 2c: Обробка скарги
def process_complaint(state: WorkflowState) -> WorkflowState:
    response = llm.invoke(f"""Класифікуй скаргу.
{state['input_document']}
Поверни JSON: {{category, severity: low/medium/high/critical, requires_immediate_action: bool}}""")
    return {**state, "extracted_data": json.loads(response.content), "processing_path": "complaint"}

# Нода 3: Валідація
def validate_data(state: WorkflowState) -> WorkflowState:
    errors = []
    data = state.get("extracted_data", {})
    if state["document_type"] == "invoice":
        if not data.get("amount"):
            errors.append("Missing invoice amount")
        if not data.get("vendor"):
            errors.append("Missing vendor information")
    return {**state, "validation_errors": errors}

# Нода 4a: Успішне завершення
def finalize_success(state: WorkflowState) -> WorkflowState:
    return {**state, "output": {
        "status": "processed",
        "path": state["processing_path"],
        "data": state["extracted_data"],
    }}

# Нода 4b: Обробка помилок
def handle_validation_errors(state: WorkflowState) -> WorkflowState:
    return {**state, "output": {
        "status": "validation_failed",
        "errors": state["validation_errors"],
        "requires_manual_review": True,
    }}

# Routing функції
def route_by_document_type(state: WorkflowState) -> str:
    mapping = {
        "invoice": "process_invoice",
        "contract": "process_contract",
        "complaint": "process_complaint",
    }
    return mapping.get(state["document_type"], "process_unknown")

def route_after_validation(state: WorkflowState) -> str:
    return "handle_errors" if state["validation_errors"] else "finalize"

# Побудова графа
graph = StateGraph(WorkflowState)
graph.add_node("classify", classify_document)
graph.add_node("process_invoice", process_invoice)
graph.add_node("process_contract", process_contract)
graph.add_node("process_complaint", process_complaint)
graph.add_node("validate", validate_data)
graph.add_node("finalize", finalize_success)
graph.add_node("handle_errors", handle_validation_errors)
graph.set_entry_point("classify")
graph.add_conditional_edges("classify", route_by_document_type, {
    "process_invoice": "process_invoice",
    "process_contract": "process_contract",
    "process_complaint": "process_complaint",
    "process_unknown": "handle_errors",
})
graph.add_edge("process_invoice", "validate")
graph.add_edge("process_contract", "validate")
graph.add_edge("process_complaint", "validate")
graph.add_conditional_edges("validate", route_after_validation, {
    "finalize": "finalize",
    "handle_errors": "handle_errors",
})
graph.add_edge("finalize", END)
graph.add_edge("handle_errors", END)
workflow = graph.compile()

Як реалізувати циклічні розгалуження з retry?

У реальних сценаріях дані можуть бути неповними або низької якості. Ми додаємо retry-цикл з обмеженням спроб:

MAX_RETRIES = 3

def check_quality_and_retry(state: WorkflowState) -> str:
    """Вирішує: прийняти результат або відправити на переробку"""
    if state.get("retry_count", 0) >= MAX_RETRIES:
        return "accept"  # Приймаємо навіть неідеальний результат
    quality = assess_output_quality(state["output"])
    if quality < 0.8:
        return "retry"
    return "accept"

def increment_retry(state: WorkflowState) -> WorkflowState:
    return {**state, "retry_count": state.get("retry_count", 0) + 1}

# Додаємо цикл до графа
graph.add_conditional_edges("quality_check", check_quality_and_retry, {
    "retry": "processing_node",
    "accept": "finalize",
})

Чому hybrid-розгалуження ефективніше?

З нашої практики: workflow обробки вхідної кореспонденції. Замовник отримував 500+ документів на день (email attachments, завантаження через портал). Детерміноване розгалуження не справлялося — документи одного типу часто вимагали різних дій. Ми впровадили hybrid-систему, де код обробляє структуровані поля (номери, дати), а LLM класифікує неоднозначні випадки. Результат: авто-обробка без ручного втручання — 71%, час маршрутизації скоротився з 45 хвилин до миттєвої, точність класифікації — 94%, помилки маршрутизації — 2,1%. Такий підхід дозволив клієнту суттєво скоротити витрати на ручну обробку.

Замовте розробку AI Workflow під ключ — ми підготуємо проєкт з урахуванням ваших вимог.

Типові помилки при розробці AI Workflow

Процес роботи

1. Аналітика — вивчаємо вхідні дані, типи розгалужень, вимоги до latency та точності. Фіксуємо цільові метрики: accuracy, p99 latency, error rate.
2. Проєктування графа — малюємо схему, визначаємо ноди та conditional edges. Обираємо тип розгалуження для кожного вузла.
3. Реалізація — пишемо код на LangGraph, інтегруємо з LLM (GPT-4, Claude) та зовнішніми API. Додаємо логування та моніторинг.
4. Тестування — покриваємо unit-тестами кожну ноду, перевіряємо edge cases (порожні входи, помилки моделей). Проводимо навантажувальне тестування на реальних даних.
5. Деплой — розгортаємо під навантаженням, налаштовуємо алерти за метриками: accuracy, latency p99, throughput. Забезпечуємо hot-reload для швидких правок.

Що входить у розробку AI Workflow

• Проєктна документація: опис графа, специфікація нод та conditional edges, сценарії використання.
• Вихідний код на LangGraph з коментарями та інструкцією з розгортання.
• Доступ до репозиторію з версіонуванням (Git) та CI/CD.
• Навчання команди (до 2-х онлайн-сесій) з налаштування та модифікації workflow.
• Технічна підтримка на 3 місяці після впровадження.

Гарантії та досвід

Наші інженери мають 10+ років досвіду в AI/ML, реалізували понад 50 проєктів AI-автоматизації. 5 років на ринку. Ми гарантуємо: точність класифікації не нижче 90%, автообробку 70%+ потоку, час відгуку системи — менше 1 секунди на документ. Результати фіксуємо в SLA з щомісячним аудитом.

Терміни

• Проєктування: від 1 тижня.
• Реалізація нод та розгалужень: 2–3 тижні.
• Тестування edge cases: 1–2 тижні.
• Разом під ключ: 4–6 тижнів.

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