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 тижнів.

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