Crafting Scalable Custom AI Agents for Enterprise Automation with Monitoring and Orchestration

Building a Custom AI Agent Framework

This tutorial guides you through designing and implementing a custom AI agent framework using PyTorch and essential Python libraries. It covers wrapping core functionalities into monitored CustomTool classes, orchestrating multiple agents with system prompts, and defining end-to-end workflows for automating tasks such as competitive website analysis and data-processing pipelines. The framework includes retry logic, logging, and performance metrics to ensure robust deployment and scalability within enterprise environments.

Core Libraries and Setup

The initial setup involves installing and importing key libraries including PyTorch, Transformers, pandas, NumPy, BeautifulSoup for web scraping, and scikit-learn for machine learning. A standardized logging configuration is established along with global constants for API timeouts and retry limits to guarantee predictable tool behavior in production.

CustomTool and ToolResult Classes

A ToolResult dataclass standardizes the outcome of tool executions, capturing success status, execution time, returned data, and any errors. The CustomTool base class wraps individual functions with an execute method that tracks call counts, computes average execution times, and logs errors, promoting consistent monitoring and observability across all custom utilities.

@dataclass
class ToolResult:
   """Standardized tool result structure"""
   success: bool
   data: Any
   error: Optional[str] = None
   execution_time: float = 0.0
   metadata: Dict[str, Any] = None
 
class CustomTool:
   def __init__(self, name: str, description: str, func: Callable):
       self.name = name
       self.description = description
       self.func = func
       self.calls = 0
       self.avg_execution_time = 0.0
       self.error_rate = 0.0
  
   def execute(self, *args, **kwargs) -> ToolResult:
       start_time = time.time()
       self.calls += 1
       try:
           result = self.func(*args, **kwargs)
           execution_time = time.time() - start_time
           self.avg_execution_time = ((self.avg_execution_time * (self.calls - 1)) + execution_time) / self.calls
           return ToolResult(success=True, data=result, execution_time=execution_time, metadata={'tool_name': self.name, 'call_count': self.calls})
       except Exception as e:
           execution_time = time.time() - start_time
           self.error_rate = (self.error_rate * (self.calls - 1) + 1) / self.calls
           logger.error(f"Tool {self.name} failed: {str(e)}")
           return ToolResult(success=False, data=None, error=str(e), execution_time=execution_time, metadata={'tool_name': self.name, 'call_count': self.calls})

CustomAgent for Task Execution

The CustomAgent class manages tools, system prompts, conversation history, and performance metrics. It routes tasks to appropriate tools based on keyword matching and aggregates results with execution summaries.

class CustomAgent:
   def __init__(self, name: str, system_prompt: str = "", max_iterations: int = 5):
       self.name = name
       self.system_prompt = system_prompt
       self.max_iterations = max_iterations
       self.tools = {}
       self.conversation_history = []
       self.performance_metrics = {}
  
   def add_tool(self, tool: CustomTool):
       self.tools[tool.name] = tool
  
   def run(self, task: str) -> Dict[str, Any]:
       logger.info(f"Agent {self.name} executing task: {task}")
       task_lower = task.lower()
       results = []
       if any(keyword in task_lower for keyword in ['analyze', 'website', 'url', 'web']):
           if 'advanced_web_intelligence' in self.tools:
               import re
               url_pattern = r'https?://[^\s]+'
               urls = re.findall(url_pattern, task)
               if urls:
                   result = self.tools['advanced_web_intelligence'].execute(urls[0])
                   results.append(result)
       elif any(keyword in task_lower for keyword in ['data', 'analyze', 'stats', 'csv']):
           if 'advanced_data_science_toolkit' in self.tools:
               if 'name,age,salary' in task:
                   data_start = task.find('name,age,salary')
                   data_part = task[data_start:]
                   result = self.tools['advanced_data_science_toolkit'].execute(data_part, 'stats')
                   results.append(result)
       elif any(keyword in task_lower for keyword in ['generate', 'code', 'api', 'client']):
           if 'advanced_code_generator' in self.tools:
               result = self.tools['advanced_code_generator'].execute(task)
               results.append(result)
       return {
           'agent': self.name,
           'task': task,
           'results': [r.data if r.success else {'error': r.error} for r in results],
           'execution_summary': {
               'tools_used': len(results),
               'success_rate': sum(1 for r in results if r.success) / len(results) if results else 0,
               'total_time': sum(r.execution_time for r in results)
           }
       }

Specialized Tools with Performance Monitoring

Three key tools wrapped with a @performance_monitor decorator deliver advanced capabilities:

advanced_web_intelligence: Performs comprehensive or sentiment-based website analysis using requests and BeautifulSoup.
advanced_data_science_toolkit: Executes statistical analysis and clustering on CSV or JSON data with pandas and scikit-learn.
advanced_code_generator: Generates production-ready code templates supporting API clients and data processors.

Agent Orchestration and Workflow Management

The AgentOrchestrator manages multiple agents, enabling workflow coordination with multi-step tasks that span specialized expertise areas. It supports creating domain-specific agents, executing workflows, and retrieving system status metrics.

Production Examples and System Metrics

Demonstrations include web analysis, data science pipelines, code generation, and multi-agent workflow execution. The system outputs performance metrics such as call counts, average execution times, and error rates for each tool, ensuring observability and reliability.

This framework provides a robust blueprint for building scalable, monitored, and orchestrated AI agents tailored for enterprise-grade automation and analytics.