Умные workflow с ветвлением: автоматизация на базе AI

Разработка 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 недель.

Получите консультацию инженера — мы оценим ваш кейс и предложим решение. Свяжитесь с нами для расчета стоимости и сроков вашего проекта.