Innovology - Cross-Industry Innovation at PE Speed

Task-based automation is dead. If you're still writing Zapier flows or Make.com scenarios, you're building yesterday's solutions. Agent-based systems now handle 10x complexity with dynamic decision-making, self-correction, and goal-oriented execution. Here's how to build them.

Architecture Evolution: From Pipelines to Agents

Traditional Task Pipeline:

// Old way: Brittle, sequential pipeline
const processOrder = async (order) => {
  try {
    const validated = await validateOrder(order);
    const inventory = await checkInventory(validated.items);
    const payment = await processPayment(validated.payment);
    const shipment = await createShipment(validated, payment);
    await notifyCustomer(shipment);
  } catch (error) {
    await handleError(error); // Usually just logs and fails
  }
};

Agent-Based Approach:

# New way: Goal-oriented agent
from agentframework import Agent, Goal, Context

class OrderAgent(Agent):
    def __init__(self):
        super().__init__()
        self.capabilities = [
            ValidateOrderCapability(),
            InventoryCapability(),
            PaymentCapability(),
            ShippingCapability(),
            CustomerCommsCapability()
        ]

    async def achieve_goal(self, goal: Goal, context: Context):
        # Agent determines optimal path
        plan = await self.plan(goal, context)

        while not goal.is_achieved():
            next_action = plan.get_next_action()
            result = await self.execute(next_action)

            # Dynamic replanning on failure
            if result.failed:
                plan = await self.replan(goal, context, result)

            # Learn from execution
            self.update_knowledge(result)

        return goal.outcome

# Usage
agent = OrderAgent()
goal = Goal("Process order #12345 ensuring customer satisfaction")
result = await agent.achieve_goal(goal, context=order_data)

Platform Comparison: Build vs. Buy

n8n: Open-Source Agent Platform

Pros:

Self-hosted control
Extensible with custom nodes
Active community (40k+ GitHub stars)
Fair-code license

Implementation:

// Custom n8n agent node
import { IExecuteFunctions, INodeExecutionData, INodeType } from 'n8n-workflow';

export class IntelligentAgent implements INodeType {
    description = {
        displayName: 'Intelligent Agent',
        name: 'intelligentAgent',
        group: ['transform'],
        version: 1,
        description: 'Execute goal-based automation',
        defaults: {
            name: 'Intelligent Agent',
        },
        inputs: ['main'],
        outputs: ['main'],
        properties: [
            {
                displayName: 'Goal',
                name: 'goal',
                type: 'string',
                default: '',
                required: true,
            },
            {
                displayName: 'Model',
                name: 'model',
                type: 'options',
                options: [
                    { name: 'GPT-4', value: 'gpt-4' },
                    { name: 'Claude', value: 'claude' },
                    { name: 'Local LLM', value: 'local' },
                ],
                default: 'gpt-4',
            },
        ],
    };

    async execute(this: IExecuteFunctions): Promise<INodeExecutionData[][]> {
        const items = this.getInputData();
        const goal = this.getNodeParameter('goal', 0) as string;
        const model = this.getNodeParameter('model', 0) as string;

        const agent = new Agent({
            model,
            tools: this.getAvailableTools(),
            memory: this.getWorkflowStaticData('agent_memory'),
        });

        const results = await Promise.all(
            items.map(async (item) => {
                const result = await agent.run(goal, item.json);
                return {
                    json: result,
                    binary: item.binary,
                };
            })
        );

        // Persist agent memory
        this.setWorkflowStaticData('agent_memory', agent.memory);

        return this.prepareOutputData(results);
    }
}

Salesforce Agentforce: Enterprise Agent Platform

Setup:

// Agentforce Agent Definition
public class CustomerServiceAgent extends AgentForce.Agent {

    @AgentCapability
    public AgentResponse handleCustomerInquiry(AgentContext context) {
        String inquiry = context.getInput('customer_message');

        // Define goal
        Goal goal = new Goal()
            .setObjective('Resolve customer inquiry')
            .addConstraint('Maintain brand voice')
            .addConstraint('Follow support policies')
            .setSuccessCriteria('Customer satisfied');

        // Agent determines approach
        Plan plan = this.createPlan(goal, context);

        // Execute with built-in Salesforce integrations
        return plan.execute(new ExecutionContext()
            .withCRM(context.getRecordId())
            .withKnowledgeBase('Support_KB')
            .withEscalationPath('Tier2_Queue')
        );
    }

    @AgentTool
    public Object searchKnowledge(String query) {
        return [
            SELECT Id, Title, Body, Solution__c 
            FROM Knowledge__kav 
            WHERE PublishStatus = 'Online' 
            AND Title LIKE :('%' + query + '%')
            LIMIT 5
        ];
    }
}

// Deployment
AgentForce.deploy(new CustomerServiceAgent())
    .withTrigger(AgentTrigger.CASE_CREATED)
    .withPermissions(new AgentPermissions()
        .allowRead(Account.SObjectType)
        .allowWrite(Case.SObjectType)
    )
    .activate();

Custom Agent Implementation

Using LangChain + FastAPI:

from fastapi import FastAPI, BackgroundTasks
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain.tools import Tool
from langchain.memory import ConversationSummaryBufferMemory
from langchain_openai import ChatOpenAI
import asyncio

app = FastAPI()

class BusinessAgent:
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4", temperature=0)
        self.memory = ConversationSummaryBufferMemory(
            llm=self.llm,
            max_token_limit=2000
        )
        self.tools = self._init_tools()
        self.agent = self._init_agent()

    def _init_tools(self):
        return [
            Tool(
                name="crm_lookup",
                func=self.crm_lookup,
                description="Look up customer information in CRM"
            ),
            Tool(
                name="inventory_check",
                func=self.inventory_check,
                description="Check product inventory levels"
            ),
            Tool(
                name="order_process",
                func=self.process_order,
                description="Process a customer order"
            ),
        ]

    def _init_agent(self):
        prompt = """You are a business operations agent. 
        Your goal is to handle business processes autonomously.
        Always verify data before taking actions.
        Escalate to humans when confidence < 80%.

        Available tools: {tools}
        Conversation history: {chat_history}

        Task: {input}

        Think step by step and use tools as needed.
        {agent_scratchpad}
        """

        agent = create_openai_tools_agent(
            llm=self.llm,
            tools=self.tools,
            prompt=prompt
        )

        return AgentExecutor(
            agent=agent,
            tools=self.tools,
            memory=self.memory,
            verbose=True,
            max_iterations=10,
            early_stopping_method="generate"
        )

    async def crm_lookup(self, customer_id: str) -> dict:
        # Integrate with actual CRM
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"https://crm.api/customers/{customer_id}",
                headers={"Authorization": f"Bearer {CRM_TOKEN}"}
            )
            return response.json()

    async def run(self, task: str, context: dict = None):
        # Add context to memory if provided
        if context:
            self.memory.chat_memory.add_user_message(
                f"Context: {json.dumps(context)}"
            )

        # Execute agent
        result = await self.agent.ainvoke({"input": task})

        # Extract and return structured output
        return {
            "result": result["output"],
            "actions_taken": self._extract_actions(result),
            "confidence": self._calculate_confidence(result)
        }

# API Endpoints
agent = BusinessAgent()

@app.post("/agent/execute")
async def execute_task(
    task: str, 
    context: dict = None,
    background_tasks: BackgroundTasks = BackgroundTasks()
):
    result = await agent.run(task, context)

    # Log execution for monitoring
    background_tasks.add_task(log_execution, task, result)

    return result

@app.get("/agent/status")
async def get_agent_status():
    return {
        "memory_size": len(agent.memory.chat_memory.messages),
        "tools_available": [tool.name for tool in agent.tools],
        "model": "gpt-4",
        "status": "operational"
    }

Key Implementation Patterns

1. Goal-Oriented Architecture

class Goal:
    def __init__(self, objective: str):
        self.objective = objective
        self.constraints = []
        self.success_criteria = []
        self.state = "pending"

    def is_achieved(self) -> bool:
        return all(criterion.is_met() for criterion in self.success_criteria)

    def add_constraint(self, constraint: Constraint):
        self.constraints.append(constraint)
        return self

2. Dynamic Tool Selection

class ToolSelector {
    constructor(tools) {
        this.tools = tools;
        this.embeddings = new OpenAIEmbeddings();
    }

    async selectTools(task, context) {
        // Embed task description
        const taskEmbedding = await this.embeddings.embed(task);

        // Calculate similarity with tool descriptions
        const scores = await Promise.all(
            this.tools.map(async (tool) => {
                const toolEmbedding = await this.embeddings.embed(tool.description);
                return {
                    tool,
                    score: cosineSimilarity(taskEmbedding, toolEmbedding)
                };
            })
        );

        // Return top relevant tools
        return scores
            .sort((a, b) => b.score - a.score)
            .slice(0, 5)
            .map(item => item.tool);
    }
}

3. Self-Correction Mechanisms

class SelfCorrectingAgent:
    def __init__(self):
        self.validator = OutputValidator()
        self.max_retries = 3

    async def execute_with_correction(self, task):
        for attempt in range(self.max_retries):
            result = await self.execute(task)

            validation = self.validator.validate(result)
            if validation.is_valid:
                return result

            # Learn from failure
            self.add_to_context(f"Attempt {attempt + 1} failed: {validation.error}")

            # Adjust approach
            task = self.refine_task(task, validation.feedback)

        raise MaxRetriesExceeded(f"Failed after {self.max_retries} attempts")

Production Considerations

Monitoring and Observability

from opentelemetry import trace
from prometheus_client import Counter, Histogram

# Metrics
agent_executions = Counter('agent_executions_total', 'Total agent executions')
agent_duration = Histogram('agent_duration_seconds', 'Agent execution duration')
agent_errors = Counter('agent_errors_total', 'Total agent errors')

class MonitoredAgent(Agent):
    @agent_duration.time()
    async def run(self, goal):
        tracer = trace.get_tracer(__name__)

        with tracer.start_as_current_span("agent_execution") as span:
            span.set_attribute("goal", goal.objective)
            span.set_attribute("agent_id", self.id)

            try:
                agent_executions.inc()
                result = await super().run(goal)
                span.set_attribute("success", True)
                return result
            except Exception as e:
                agent_errors.inc()
                span.set_attribute("success", False)
                span.record_exception(e)
                raise

Testing Agent Behaviors

import pytest
from unittest.mock import Mock, patch

class TestOrderAgent:
    @pytest.fixture
    def agent(self):
        return OrderAgent()

    @pytest.mark.asyncio
    async def test_handles_out_of_stock(self, agent):
        # Mock inventory service
        with patch.object(agent, 'check_inventory') as mock_inventory:
            mock_inventory.return_value = {"available": False}

            goal = Goal("Process order for out-of-stock item")
            result = await agent.achieve_goal(goal, context={"item": "widget"})

            # Agent should find alternative solution
            assert result.success
            assert "alternative" in result.actions_taken
            assert result.customer_notified

    async def test_escalates_complex_scenarios(self, agent):
        goal = Goal("Handle return for customized product")
        result = await agent.achieve_goal(goal, context=complex_return_data)

        assert result.escalated_to_human
        assert result.escalation_reason == "Policy exception required"

Security Considerations

class SecureAgent(Agent):
    def __init__(self):
        super().__init__()
        self.permission_manager = PermissionManager()
        self.audit_logger = AuditLogger()

    async def execute_action(self, action, context):
        # Check permissions
        if not self.permission_manager.can_execute(action, context.user):
            raise PermissionDenied(f"User lacks permission for {action}")

        # Log action attempt
        self.audit_logger.log_attempt(action, context)

        # Execute with sandboxing
        sandbox = Sandbox(
            allowed_apis=self.permission_manager.get_allowed_apis(context.user),
            timeout=30,
            memory_limit="512MB"
        )

        result = await sandbox.execute(super().execute_action, action, context)

        # Log result
        self.audit_logger.log_result(action, result)

        return result

Migration Strategy

Moving from task automation to agents:

Identify Agent-Ready Processes
- High exception rate (>20%)
- Multiple decision points
- Cross-system orchestration
- Natural language inputs

Parallel Run Strategy

async def migrate_with_shadow_mode(task_automation, agent):
   # Run both systems
   task_result = await task_automation.run()
   agent_result = await agent.run()

   # Compare results
   if results_match(task_result, agent_result):
       confidence += 0.1
   else:
       log_divergence(task_result, agent_result)

   # Switch when confident
   if confidence > 0.95:
       return agent_result
   else:
       return task_result

Gradual Rollout
- Start with read-only operations
- Add write capabilities incrementally
- Expand scope as confidence grows

The shift to agent-based automation isn't just a technical upgrade—it's a fundamental rethinking of how we build business systems. Start small, measure everything, and prepare for a world where configuration is conversation.

Building Agent-Based Automation: A Developer's Guide to the Post-Task Era