AI Workflow Automation in Practice

automation_evolution:
  rule_based:
    capabilities:
      - If X then Y
      - Pattern matching
      - Structured data

    limitations:
      - Can't handle ambiguity
      - Brittle to variations
      - Every case must be coded

  ai_powered:
    capabilities:
      - Natural language understanding
      - Context-aware decisions
      - Handles variations
      - Learns patterns

    new_possibilities:
      - Email triage and routing
      - Document processing
      - Approval workflows
      - Customer request handling

class AIDecisionAutomation:
    """Automate decisions with AI."""

    async def route_request(
        self,
        request: Request
    ) -> RoutingDecision:
        # Classify request
        classification = await self.llm.generate(
            prompt=f"""Classify this request and determine routing.

Request: {request.content}
Context: {request.metadata}

Categories: {self.categories}

Output JSON:
{{
  "category": "...",
  "priority": "high/medium/low",
  "route_to": "team name",
  "reasoning": "..."
}}"""
        )

        decision = json.loads(classification)

        # Validate decision
        if not self._validate_decision(decision):
            return RoutingDecision(
                route_to="manual_review",
                reason="Could not classify confidently"
            )

        return RoutingDecision(**decision)

    async def approve_request(
        self,
        request: ApprovalRequest
    ) -> ApprovalDecision:
        # Check against policy
        policy_check = await self._check_policy(request)

        if policy_check.clear_approve:
            return ApprovalDecision(
                approved=True,
                auto=True,
                reason="Meets all criteria"
            )

        if policy_check.clear_reject:
            return ApprovalDecision(
                approved=False,
                auto=True,
                reason=policy_check.rejection_reason
            )

        # Ambiguous - escalate to human
        return ApprovalDecision(
            approved=None,
            auto=False,
            reason="Requires human review",
            context=policy_check.concerns
        )

class AIDocumentProcessor:
    """Process documents with AI."""

    async def process_invoice(
        self,
        document: Document
    ) -> ProcessedInvoice:
        # Extract structured data
        extracted = await self.llm.generate(
            prompt=f"""Extract invoice information.

Document text:
{document.text}

Extract as JSON:
{{
  "vendor": "...",
  "invoice_number": "...",
  "date": "YYYY-MM-DD",
  "total": 0.00,
  "line_items": [...],
  "payment_terms": "..."
}}"""
        )

        data = json.loads(extracted)

        # Validate against existing records
        validation = await self._validate_extraction(data)

        if validation.issues:
            # Flag for review
            return ProcessedInvoice(
                data=data,
                status="needs_review",
                issues=validation.issues
            )

        # Auto-process if valid
        return ProcessedInvoice(
            data=data,
            status="processed"
        )

class AIWorkflowEngine:
    """Orchestrate workflows with AI decisions."""

    async def execute_workflow(
        self,
        workflow: Workflow,
        trigger: Trigger
    ) -> WorkflowResult:
        context = {"trigger": trigger, "results": {}}

        for step in workflow.steps:
            if step.type == "ai_decision":
                result = await self._ai_decision(step, context)
            elif step.type == "action":
                result = await self._execute_action(step, context)
            elif step.type == "condition":
                result = await self._evaluate_condition(step, context)

            context["results"][step.name] = result

            # Handle branching
            next_step = self._determine_next(step, result)
            if next_step == "end":
                break

        return WorkflowResult(
            success=True,
            context=context
        )

# Example workflow definition
workflow = {
    "name": "expense_approval",
    "trigger": "expense_submitted",
    "steps": [
        {
            "name": "classify_expense",
            "type": "ai_decision",
            "prompt": "Classify this expense..."
        },
        {
            "name": "check_policy",
            "type": "ai_decision",
            "prompt": "Does this comply with policy..."
        },
        {
            "name": "route_or_approve",
            "type": "condition",
            "rules": {
                "amount < 100 and policy_compliant": "auto_approve",
                "amount >= 100": "manager_review",
                "not policy_compliant": "flag_violation"
            }
        }
    ]
}

human_in_loop_patterns:
  confidence_threshold:
    - Auto-process if confidence > 95%
    - Human review if 70-95%
    - Escalate if < 70%

  sampling:
    - Random sample for quality check
    - 100% review for new categories
    - Reduced review as confidence grows

  exception_handling:
    - Clear escalation path
    - Context preservation
    - Feedback capture