10. Approval Workflows

Business processes frequently require human decision points. Approval workflows bridge the gap between automated data processing and human judgment, ensuring that consequential actions receive appropriate review before execution.

Architecture Overview

An approval workflow consists of three components: a triggering event, a notification mechanism, and a resolution handler. The triggering event initiates the workflow when conditions are met. The notification mechanism alerts the appropriate approver through their preferred channel. The resolution handler executes the corresponding action based on the approver's decision.

# approval_workflow.py
import json
import time
from datetime import datetime
from pathlib import Path
import ollama

class ApprovalWorkflow:
    def __init__(self, config_path: str = "workflow_config.json"):
        self.config = self._load_config(config_path)
        self.pending_approvals = self._load_pending()
        self.approved_actions = []
        self.rejected_actions = []
    
    def _load_config(self, path: str) -> dict:
        with open(path) as f:
            return json.load(f)
    
    def _load_pending(self) -> dict:
        pending_path = Path(self.config.get("pending_file", "pending_approvals.jsonl"))
        if pending_path.exists():
            with open(pending_path) as f:
                return {json.loads(line)["id"]: json.loads(line) for line in f}
        return {}
    
    def trigger_approval(self, action_type: str, action_data: dict) -> str:
        """Create a new approval request."""
        approval_id = f"{action_type}_{int(time.time())}"
        
        # Generate AI summary for approver
        summary = self._generate_summary(action_type, action_data)
        
        approval_request = {
            "id": approval_id,
            "type": action_type,
            "data": action_data,
            "summary": summary,
            "created_at": datetime.now().isoformat(),
            "status": "pending",
            "required_approver": self._get_approver(action_type)
        }
        
        self.pending_approvals[approval_id] = approval_request
        self._persist_pending()
        self._notify_approver(approval_request)
        
        return approval_id
    
    def _generate_summary(self, action_type: str, action_data: dict) -> str:
        """Use local LLM to generate human-readable summary."""
        prompt = f"""Summarize this {action_type} request for an approver:
{json.dumps(action_data, indent=2)}

Format as: SUMMARY: <one sentence>
IMPACT: <one sentence>
RISK: <low/medium/high>
"""
        response = ollama.chat(
            model=self.config.get("summary_model", "llama3"),
            messages=[{"role": "user", "content": prompt}]
        )
        return response["message"]["content"]

Escalation and Timeouts

Unchecked approvals create risk. A production workflow requires timeout handling and escalation paths.

    def process_approval(self, approval_id: str, decision: str, 
                        approver_id: str, comments: str = "") -> dict:
        """Process an approval decision."""
        if approval_id not in self.pending_approvals:
            return {"status": "error", "message": "Approval not found"}
        
        approval = self.pending_approvals[approval_id]
        
        if decision not in ["approved", "rejected"]:
            return {"status": "error", "message": "Invalid decision"}
        
        approval["status"] = decision
        approval["decided_at"] = datetime.now().isoformat()
        approval["approver_id"] = approver_id
        approval["comments"] = comments
        
        # Remove from pending and record in history
        del self.pending_approvals[approval_id]
        self._persist_pending()
        
        if decision == "approved":
            self.approved_actions.append(approval)
            self._execute_action(approval)
        else:
            self.rejected_actions.append(approval)
            self._notify_requester(approval, rejected=True)
        
        return {"status": "success", "approval": approval}
    
    def check_timeouts(self) -> list:
        """Find and handle timed-out approvals."""
        timeout_hours = self.config.get("timeout_hours", 24)
        timed_out = []
        
        for approval in self.pending_approvals.values():
            created = datetime.fromisoformat(approval["created_at"])
            age_hours = (datetime.now() - created).total_seconds() / 3600
            
            if age_hours > timeout_hours:
                approval["status"] = "timeout"
                timed_out.append(approval)
                self._escalate(approval)
        
        return timed_out

Integration Patterns

For critical workflows, integrate with existing systems through webhooks.

    def _notify_approver(self, approval: dict):
        """Send notification to approver."""
        webhook_url = approval.get("required_approver", {}).get("webhook")
        if webhook_url:
            import requests
            requests.post(webhook_url, json={
                "approval_id": approval["id"],
                "summary": approval["summary"],
                "action_type": approval["type"],
                "created_at": approval["created_at"]
            })