09. Workflow Agents

A workflow agent chains multiple AI tasks into a complete business process. The agent decides which actions to take based on input, handles the sequencing, and produces a final output.

The agent architecture uses a loop: receive input → determine next action → execute action → evaluate result → repeat until complete:

class WorkflowAgent:
    def __init__(self, model='llama3.1:8b'):
        self.model = model
        Self.tools = {
            'classify_email': classify_email,
            'generate_response': generate_response,
            'extract_data': extract_invoice_data,
            'find_knowledge': answer_faq,
            'summarize': summarize_text
        }
        self.history = []
    
    def execute(self, task, input_data):
        """Execute a task using appropriate tools."""
        self.history = [{'task': task, 'input': input_data}]
        
        # Determine actions needed
        plan = self.plan_actions(task, input_data)
        
        # Execute each action in sequence
        results = {}
        for step in plan:
            tool_name = step['tool']
            tool_input = step['input']
            
            if tool_name not in self.tools:
                return {'error': f'Unknown tool: {tool_name}'}
            
            try:
                result = self.tools[tool_name](tool_input)
                results[tool_name] = result
                self.history.append({'step': tool_name, 'result': result})
            except Exception as e:
                return {'error': f'Tool {tool_name} failed: {str(e)}'}
        
        return {'status': 'complete', 'results': results}
    
    def plan_actions(self, task, input_data):
        """Determine which tools to use and in what order."""
        plan_prompt = f"""Analyze this task and determine the sequence of tools to use.

Available tools:
- classify_email: Categorize an email
- generate_response: Draft an email response
- extract_data: Extract structured data from documents
- find_knowledge: Answer questions from knowledge base
- summarize: Create a summary of text

Task: {task}
Input: {str(input_data)[:1000]}

Respond with JSON array of steps:
[{{"tool": "tool_name", "input": "what to pass to the tool"}}]

Return only the JSON array, no explanation:
"""
        
        response = chat(model=self.model, messages=[
            {'role': 'user', 'content': plan_prompt}
        ])
        
        import json
        try:
            return json.loads(response['message']['content'])
        except json.JSONDecodeError:
            return [{'tool': 'summarize', 'input': input_data}]

A practical workflow for processing incoming vendor invoices:

def process_invoice_workflow(pdf_path):
    """Complete workflow for invoice processing."""
    agent = WorkflowAgent()
    
    # Step 1: Extract data from invoice
    text = extract_text_from_pdf(pdf_path)
    extraction_result = agent.execute(
        task="Extract invoice data",
        input_data={'text': text, 'format': 'invoice'}
    )
    
    if 'error' in extraction_result:
        return {'status': 'failed', 'reason': extraction_result['error']}
    
    invoice_data = extraction_result['results']['extract_data']
    
    # Step 2: Check against knowledge base for vendor info
    vendor_info = agent.execute(
        task="Find vendor policies",
        input_data={'query': f"vendor {invoice_data.get('vendor_name', 'unknown')}"}
    )
    
    # Step 3: Create summary for accountant
    summary = f"""
Invoice received from: {invoice_data.get('vendor_name')}
Amount: ${invoice_data.get('total', 0):.2f}
Due: {invoice_data.get('due_date')}

Recommend: {'Approve for payment' if invoice_data.get('total', 0) < 10000 else 'Manager approval required'}
"""
    
    return {
        'status': 'complete',
        'invoice': invoice_data,
        'summary': summary
    }

Monitoring agent execution reveals where improvements are needed. Log each step, track token usage, and measure end-to-end time. Patterns in failures point to specific tool or prompt issues.