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

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

AI workflow з розгалуженням — це система, де потік виконання визначається динамічно на основі проміжних результатів, вхідних даних або рішень LLM. Це складніше, ніж лінійні конвеєри, але необхідно для реальних бізнес-процесів, де різні вхідні дані потребують різних шляхів обробки.

Типи умовного розгалуження

Детерміноване розгалуження: умова визначається кодом на основі даних (if/else, switch).

LLM-базоване розгалуження: умова визначається рішенням мовної моделі (класифікація, маршрутизація).

Гібридне: код обробляє структуровані умови, LLM — неструктуровані.

Реалізація з 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,
    }}

# Функції маршрутизації
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 та correction loops

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",
})

Практичний приклад: Workflow обробки вхідної кореспонденції

Завдання: автоматична обробка 500+ вхідних документів на день (email-додатки, завантаження через портал).

Граф розгалуження:

Вхідний документ
    → Класифікація (8 типів)
    ↓
Тип: рахунок   → Екстракт деталей → Зіставлення з контрактом → [OK: 1C] / [помилка: бухгалтер]
Тип: контракт  → Екстракт умов   → Юридичний ризик-скор    → [low: архів] / [medium+: юрист]
Тип: скарга    → Оцінка важливості → [critical: негайно] / [normal: черга]
Тип: запит     → Тема             → Маршрут до відділу

Метрики:

• Автоматична обробка без ручного втручання: 71%
• Час маршрутизації: миттєво (порівняно з 45 хв вручну)
• Точність класифікації: 94%
• Помилки маршрутизації: 2.1%

Візуалізація графа

LangGraph надає вбудовану візуалізацію:

from IPython.display import Image

Image(workflow.get_graph().draw_mermaid_png())

Графік реалізації

• Проектування workflow-графа: 1 тиждень
• Реалізація вузлів та розгалуження: 2–3 тижні
• Тестування граничних випадків: 1–2 тижні
• Всього: 4–6 тижнів