זרימות עבודה חכמות עם הסתעפות: אוטומציה מונעת בינה מלאכותית

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()