On this tutorial, we stroll you thru the design and implementation of a customized agent framework constructed on PyTorch and key Python tooling, starting from net intelligence and knowledge science modules to superior code turbines. We’ll learn to wrap core functionalities in monitored CustomTool courses, orchestrate a number of brokers with tailor-made system prompts, and outline end-to-end workflows that automate duties like aggressive web site evaluation and data-processing pipelines. Alongside the best way, we exhibit real-world examples, full with retry logic, logging, and efficiency metrics, so you possibly can confidently deploy and scale these brokers inside your group’s present infrastructure.
!pip set up -q torch transformers datasets pillow requests beautifulsoup4 pandas numpy scikit-learn openai
import os, json, asyncio, threading, time
import torch, pandas as pd, numpy as np
from PIL import Picture
import requests
from io import BytesIO, StringIO
from concurrent.futures import ThreadPoolExecutor
from functools import wraps, lru_cache
from typing import Dict, Record, Elective, Any, Callable, Union
import logging
from dataclasses import dataclass
import examine
logging.basicConfig(degree=logging.INFO)
logger = logging.getLogger(__name__)
API_TIMEOUT = 15
MAX_RETRIES = 3We start by putting in and importing all of the core libraries, together with PyTorch and Transformers, in addition to knowledge dealing with libraries equivalent to pandas and NumPy, and utilities like BeautifulSoup for net scraping and scikit-learn for machine studying. We configure a standardized logging setup to seize info and error messages, and outline world constants for API timeouts and retry limits, making certain our instruments behave predictably in manufacturing.
@dataclass
class ToolResult:
"""Standardized instrument consequence construction"""
success: bool
knowledge: Any
error: Elective[str] = None
execution_time: float = 0.0
metadata: Dict[str, Any] = None
class CustomTool:
"""Base class for customized instruments"""
def __init__(self, title: str, description: str, func: Callable):
self.title = title
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:
"""Execute instrument with monitoring"""
start_time = time.time()
self.calls += 1
strive:
consequence = 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,
knowledge=consequence,
execution_time=execution_time,
metadata={'tool_name': self.title, 'call_count': self.calls}
)
besides Exception as e:
execution_time = time.time() - start_time
self.error_rate = (self.error_rate * (self.calls - 1) + 1) / self.calls
logger.error(f"Device {self.title} failed: {str(e)}")
return ToolResult(
success=False,
knowledge=None,
error=str(e),
execution_time=execution_time,
metadata={'tool_name': self.title, 'call_count': self.calls}
)We outline a ToolResult dataclass to encapsulate each execution’s end result, whether or not it succeeded, how lengthy it took, any returned knowledge, and error particulars if it failed. Our CustomTool base class then wraps particular person features with a unified execute methodology that tracks name counts, measures execution time, computes a median runtime, and logs any errors. By standardizing instrument outcomes and efficiency metrics this fashion, we guarantee consistency and observability throughout all our customized utilities.
class CustomAgent:
"""Customized agent implementation with instrument administration"""
def __init__(self, title: str, system_prompt: str = "", max_iterations: int = 5):
self.title = title
self.system_prompt = system_prompt
self.max_iterations = max_iterations
self.instruments = {}
self.conversation_history = []
self.performance_metrics = {}
def add_tool(self, instrument: CustomTool):
"""Add a instrument to the agent"""
self.instruments[tool.name] = instrument
def run(self, job: str) -> Dict[str, Any]:
"""Execute a job utilizing accessible instruments"""
logger.information(f"Agent {self.title} executing job: {job}")
task_lower = job.decrease()
outcomes = []
if any(key phrase in task_lower for key phrase in ['analyze', 'website', 'url', 'web']):
if 'advanced_web_intelligence' in self.instruments:
import re
url_pattern = r'https?://[^s]+'
urls = re.findall(url_pattern, job)
if urls:
consequence = self.instruments['advanced_web_intelligence'].execute(urls[0])
outcomes.append(consequence)
elif any(key phrase in task_lower for key phrase in ['data', 'analyze', 'stats', 'csv']):
if 'advanced_data_science_toolkit' in self.instruments:
if 'title,age,wage' in job:
data_start = job.discover('title,age,wage')
data_part = job[data_start:]
consequence = self.instruments['advanced_data_science_toolkit'].execute(data_part, 'stats')
outcomes.append(consequence)
elif any(key phrase in task_lower for key phrase in ['generate', 'code', 'api', 'client']):
if 'advanced_code_generator' in self.instruments:
consequence = self.instruments['advanced_code_generator'].execute(job)
outcomes.append(consequence)
return {
'agent': self.title,
'job': job,
'outcomes': [r.data if r.success else {'error': r.error} for r in results],
'execution_summary': {
'tools_used': len(outcomes),
'success_rate': sum(1 for r in outcomes if r.success) / len(outcomes) if outcomes else 0,
'total_time': sum(r.execution_time for r in outcomes)
}
}We encapsulate our AI logic in a CustomAgent class that holds a set of instruments, a system immediate, and execution historical past, then routes every incoming job to the best instrument primarily based on easy key phrase matching. Within the run() methodology, we log the duty, choose the suitable instrument (net intelligence, knowledge evaluation, or code era), execute it, and combination the outcomes right into a standardized response that features success charges and timing metrics. This design allows us to simply prolong brokers by including new instruments and maintains our orchestration as each clear and measurable.
print("🏗️ Constructing Superior Device Structure")
def performance_monitor(func):
"""Decorator for monitoring instrument efficiency"""
@wraps(func)
def wrapper(*args, **kwargs):
start_time = time.time()
strive:
consequence = func(*args, **kwargs)
execution_time = time.time() - start_time
logger.information(f"{func.__name__} executed in {execution_time:.2f}s")
return consequence
besides Exception as e:
logger.error(f"{func.__name__} failed: {str(e)}")
elevate
return wrapper
@performance_monitor
def advanced_web_intelligence(url: str, analysis_type: str = "complete") -> Dict[str, Any]:
"""
Superior net intelligence gathering with a number of evaluation modes.
Args:
url: Goal URL for evaluation
analysis_type: Sort of study (complete, sentiment, technical, web optimization)
Returns:
Dict containing structured evaluation outcomes
"""
strive:
response = requests.get(url, timeout=API_TIMEOUT, headers={
'Consumer-Agent': 'Mozilla/5.0'
})
from bs4 import BeautifulSoup
soup = BeautifulSoup(response.content material, 'html.parser')
title = soup.discover('title').textual content if soup.discover('title') else 'No title'
meta_desc = soup.discover('meta', attrs={'title': 'description'})
meta_desc = meta_desc.get('content material') if meta_desc else 'No description'
if analysis_type == "complete":
return {
'title': title,
'description': meta_desc,
'word_count': len(soup.get_text().break up()),
'image_count': len(soup.find_all('img')),
'link_count': len(soup.find_all('a')),
'headers': [h.text.strip() for h in soup.find_all(['h1', 'h2', 'h3'])[:5]],
'status_code': response.status_code,
'content_type': response.headers.get('content-type', 'unknown'),
'page_size': len(response.content material)
}
elif analysis_type == "sentiment":
textual content = soup.get_text()[:2000]
positive_words = ['good', 'great', 'excellent', 'amazing', 'wonderful', 'fantastic']
negative_words = ['bad', 'terrible', 'awful', 'horrible', 'disappointing']
pos_count = sum(textual content.decrease().depend(phrase) for phrase in positive_words)
neg_count = sum(textual content.decrease().depend(phrase) for phrase in negative_words)
return {
'sentiment_score': pos_count - neg_count,
'positive_indicators': pos_count,
'negative_indicators': neg_count,
'text_sample': textual content[:200],
'analysis_type': 'sentiment'
}
besides Exception as e:
return {'error': f"Evaluation failed: {str(e)}"}
@performance_monitor
def advanced_data_science_toolkit(knowledge: str, operation: str) -> Dict[str, Any]:
"""
Complete knowledge science operations with statistical evaluation.
Args:
knowledge: CSV-like string or JSON knowledge
operation: Sort of study (stats, correlation, forecast, clustering)
Returns:
Dict with evaluation outcomes
"""
strive:
if knowledge.startswith('{') or knowledge.startswith('['):
parsed_data = json.loads(data)
df = pd.DataFrame(parsed_data)
else:
df = pd.read_csv(StringIO(data))
if operation == "stats":
numeric_columns = df.select_dtypes(include=[np.number]).columns.tolist()
consequence = {
'form': df.form,
'columns': df.columns.tolist(),
'dtypes': {col: str(dtype) for col, dtype in df.dtypes.objects()},
'missing_values': df.isnull().sum().to_dict(),
'numeric_columns': numeric_columns
}
if len(numeric_columns) > 0:
consequence['summary_stats'] = df[numeric_columns].describe().to_dict()
if len(numeric_columns) > 1:
consequence['correlation_matrix'] = df[numeric_columns].corr().to_dict()
return consequence
elif operation == "clustering":
from sklearn.cluster import KMeans
from sklearn.preprocessing import StandardScaler
numeric_df = df.select_dtypes(embrace=[np.number])
if numeric_df.form[1] Dict[str, str]:
"""
Superior code era with a number of language help and optimization.
Args:
task_description: Description of coding job
language: Goal programming language
Returns:
Dict with generated code and metadata
"""
templates = {
'python': {
'api_client': '''
import requests
import json
import time
from typing import Dict, Any, Elective
class APIClient:
"""Manufacturing-ready API consumer with retry logic and error dealing with"""
def __init__(self, base_url: str, api_key: Elective[str] = None, timeout: int = 30):
self.base_url = base_url.rstrip('/')
self.timeout = timeout
self.session = requests.Session()
if api_key:
self.session.headers.replace({'Authorization': f'Bearer {api_key}'})
self.session.headers.replace({
'Content material-Sort': 'software/json',
'Consumer-Agent': 'CustomAPIClient/1.0'
})
def _make_request(self, methodology: str, endpoint: str, **kwargs) -> Dict[str, Any]:
"""Make HTTP request with retry logic"""
url = f'{self.base_url}/{endpoint.lstrip("/")}'
for try in vary(3):
strive:
response = self.session.request(methodology, url, timeout=self.timeout, **kwargs)
response.raise_for_status()
return response.json() if response.content material else {}
besides requests.exceptions.RequestException as e:
if try == 2: # Final try
elevate
time.sleep(2 ** try) # Exponential backoff
def get(self, endpoint: str, params: Elective[Dict] = None) -> Dict[str, Any]:
return self._make_request('GET', endpoint, params=params)
def put up(self, endpoint: str, knowledge: Elective[Dict] = None) -> Dict[str, Any]:
return self._make_request('POST', endpoint, json=knowledge)
def put(self, endpoint: str, knowledge: Elective[Dict] = None) -> Dict[str, Any]:
return self._make_request('PUT', endpoint, json=knowledge)
def delete(self, endpoint: str) -> Dict[str, Any]:
return self._make_request('DELETE', endpoint)
''',
'data_processor': '''
import pandas as pd
import numpy as np
from typing import Record, Dict, Any, Elective
import logging
logger = logging.getLogger(__name__)
class DataProcessor:
"""Superior knowledge processor with complete cleansing and evaluation"""
def __init__(self, knowledge: pd.DataFrame):
self.original_data = knowledge.copy()
self.processed_data = knowledge.copy()
self.processing_log = []
def clean_data(self, technique: str="auto") -> 'DataProcessor':
"""Clear knowledge with configurable methods"""
initial_shape = self.processed_data.form
# Take away duplicates
self.processed_data = self.processed_data.drop_duplicates()
# Deal with lacking values primarily based on technique
if technique == 'auto':
# For numeric columns, use imply
numeric_cols = self.processed_data.select_dtypes(embrace=[np.number]).columns
self.processed_data[numeric_cols] = self.processed_data[numeric_cols].fillna(
self.processed_data[numeric_cols].imply()
)
# For categorical columns, use mode
categorical_cols = self.processed_data.select_dtypes(embrace=['object']).columns
for col in categorical_cols:
mode_value = self.processed_data[col].mode()
if len(mode_value) > 0:
self.processed_data[col] = self.processed_data[col].fillna(mode_value[0])
final_shape = self.processed_data.form
self.processing_log.append(f"Cleaned knowledge: {initial_shape} -> {final_shape}")
return self
def normalize(self, methodology: str="minmax", columns: Elective[List[str]] = None) -> 'DataProcessor':
"""Normalize numerical columns"""
cols = columns or self.processed_data.select_dtypes(embrace=[np.number]).columns.tolist()
if methodology == 'minmax':
# Min-max normalization
for col in cols:
col_min, col_max = self.processed_data[col].min(), self.processed_data[col].max()
if col_max != col_min:
self.processed_data[col] = (self.processed_data[col] - col_min) / (col_max - col_min)
elif methodology == 'zscore':
# Z-score normalization
for col in cols:
mean_val, std_val = self.processed_data[col].imply(), self.processed_data[col].std()
if std_val != 0:
self.processed_data[col] = (self.processed_data[col] - mean_val) / std_val
self.processing_log.append(f"Normalized columns {cols} utilizing {methodology}")
return self
def get_insights(self) -> Dict[str, Any]:
"""Generate complete knowledge insights"""
insights = {
'basic_info': {
'form': self.processed_data.form,
'columns': self.processed_data.columns.tolist(),
'dtypes': {col: str(dtype) for col, dtype in self.processed_data.dtypes.objects()}
},
'data_quality': {
'missing_values': self.processed_data.isnull().sum().to_dict(),
'duplicate_rows': self.processed_data.duplicated().sum(),
'memory_usage': self.processed_data.memory_usage(deep=True).to_dict()
},
'processing_log': self.processing_log
}
# Add statistical abstract for numeric columns
numeric_data = self.processed_data.select_dtypes(embrace=[np.number])
if len(numeric_data.columns) > 0:
insights['statistical_summary'] = numeric_data.describe().to_dict()
return insights
'''
}
}
task_lower = task_description.decrease()
if any(key phrase in task_lower for key phrase in ['api', 'client', 'http', 'request']):
code = templates[language]['api_client']
description = "Manufacturing-ready API consumer with retry logic and complete error dealing with"
elif any(key phrase in task_lower for key phrase in ['data', 'process', 'clean', 'analyze']):
code = templates[language]['data_processor']
description = "Superior knowledge processor with cleansing, normalization, and perception era"
else:
code = f'''# Generated code template for: {task_description}
# Language: {language}
class CustomSolution:
"""Auto-generated resolution template"""
def __init__(self):
self.initialized = True
def execute(self, *args, **kwargs):
"""Primary execution methodology - implement your logic right here"""
return {{"message": "Implement your customized logic right here", "job": "{task_description}"}}
# Utilization instance:
# resolution = CustomSolution()
# consequence = resolution.execute()
'''
description = f"Customized template for {task_description}"
return {
'code': code,
'language': language,
'description': description,
'complexity': 'production-ready',
'estimated_lines': len(code.break up('n')),
'options': ['error_handling', 'logging', 'type_hints', 'documentation']
}We wrap every core perform in a @performance_monitor decorator so we will log execution instances and catch failures, then implement three specialised instruments: advanced_web_intelligence for complete or sentiment-driven net scraping, advanced_data_science_toolkit for statistical evaluation and clustering on CSV or JSON knowledge, and advanced_code_generator for producing production-ready code templates, making certain we monitor efficiency and keep consistency throughout all our analytics and code-generation utilities.
print("🤖 Organising Customized Agent Framework")
class AgentOrchestrator:
"""Manages a number of specialised brokers with workflow coordination"""
def __init__(self):
self.brokers = {}
self.workflows = {}
self.results_cache = {}
self.performance_metrics = {}
def create_specialist_agent(self, title: str, instruments: Record[CustomTool], system_prompt: str = None):
"""Create domain-specific brokers"""
agent = CustomAgent(
title=title,
system_prompt=system_prompt or f"You're a specialist {title} agent.",
max_iterations=5
)
for instrument in instruments:
agent.add_tool(instrument)
self.brokers[name] = agent
return agent
def execute_workflow(self, workflow_name: str, inputs: Dict) -> Dict:
"""Execute multi-step workflows throughout brokers"""
if workflow_name not in self.workflows:
elevate ValueError(f"Workflow {workflow_name} not discovered")
workflow = self.workflows[workflow_name]
outcomes = {}
workflow_start = time.time()
for step in workflow['steps']:
agent_name = step['agent']
job = step['task'].format(**inputs, **outcomes)
if agent_name in self.brokers:
step_start = time.time()
consequence = self.brokers[agent_name].run(job)
step_time = time.time() - step_start
outcomes[step['output_key']] = consequence
outcomes[f"{step['output_key']}_time"] = step_time
total_time = time.time() - workflow_start
return {
'workflow': workflow_name,
'inputs': inputs,
'outcomes': outcomes,
'metadata': {
'total_execution_time': total_time,
'steps_completed': len(workflow['steps']),
'success': True
}
}
def get_system_status(self) -> Dict[str, Any]:
"""Get complete system standing"""
return {
'brokers': {title: {'instruments': len(agent.instruments)} for title, agent in self.brokers.objects()},
'workflows': listing(self.workflows.keys()),
'cache_size': len(self.results_cache),
'total_tools': sum(len(agent.instruments) for agent in self.brokers.values())
}
orchestrator = AgentOrchestrator()
web_tool = CustomTool(
title="advanced_web_intelligence",
description="Superior net evaluation and intelligence gathering",
func=advanced_web_intelligence
)
data_tool = CustomTool(
title="advanced_data_science_toolkit",
description="Complete knowledge science and statistical evaluation",
func=advanced_data_science_toolkit
)
code_tool = CustomTool(
title="advanced_code_generator",
description="Superior code era and structure",
func=advanced_code_generator
)
web_agent = orchestrator.create_specialist_agent(
"web_analyst",
[web_tool],
"You're a net evaluation specialist. Present complete web site evaluation and insights."
)
data_agent = orchestrator.create_specialist_agent(
"data_scientist",
[data_tool],
"You're a knowledge science knowledgeable. Carry out statistical evaluation and machine studying duties."
)
code_agent = orchestrator.create_specialist_agent(
"code_architect",
[code_tool],
"You're a senior software program architect. Generate optimized, production-ready code."
)
We initialize an AgentOrchestrator to handle our suite of AI brokers, register every CustomTool implementation for net intelligence, knowledge science, and code era, after which spin up three domain-specific brokers: web_analyst, data_scientist, and code_architect. Every agent is seeded with its respective toolset and a transparent system immediate. This setup allows us to coordinate and execute multi-step workflows throughout specialised experience areas inside a single, unified framework.
print("⚡ Defining Superior Workflows")
orchestrator.workflows['competitive_analysis'] = {
'steps': [
{
'agent': 'web_analyst',
'task': 'Analyze website {target_url} with comprehensive analysis',
'output_key': 'website_analysis'
},
{
'agent': 'code_architect',
'task': 'Generate monitoring code for website analysis automation',
'output_key': 'monitoring_code'
}
]
}
orchestrator.workflows['data_pipeline'] = {
'steps': [
{
'agent': 'data_scientist',
'task': 'Analyze the following CSV data with stats operation: {data_input}',
'output_key': 'data_analysis'
},
{
'agent': 'code_architect',
'task': 'Generate data processing pipeline code',
'output_key': 'pipeline_code'
}
]
}
We outline two key multi-agent workflows: competitive_analysis, which includes our net analyst scraping and analyzing a goal URL earlier than passing insights to our code architect to generate monitoring scripts, and data_pipeline, the place our knowledge scientist runs statistical analyses on CSV inputs. Then our code architect crafts the corresponding ETL pipeline code. These declarative step sequences allow us to orchestrate advanced duties end-to-end with minimal boilerplate.
print("🚀 Operating Manufacturing Examples")
print("n📊 Superior Internet Intelligence Demo")
strive:
web_result = web_agent.run("Analyze https://httpbin.org/html with complete evaluation sort")
print(f"✅ Internet Evaluation Success: {json.dumps(web_result, indent=2)}")
besides Exception as e:
print(f"❌ Internet evaluation error: {e}")
print("n🔬 Knowledge Science Pipeline Demo")
sample_data = """title,age,wage,division
Alice,25,50000,Engineering
Bob,30,60000,Engineering
Carol,35,70000,Advertising
David,28,55000,Engineering
Eve,32,65000,Advertising"""
strive:
data_result = data_agent.run(f"Analyze this knowledge with stats operation: {sample_data}")
print(f"✅ Knowledge Evaluation Success: {json.dumps(data_result, indent=2)}")
besides Exception as e:
print(f"❌ Knowledge evaluation error: {e}")
print("n💻 Code Structure Demo")
strive:
code_result = code_agent.run("Generate an API consumer for knowledge processing duties")
print(f"✅ Code Technology Success: Generated {len(code_result['results'][0]['code'].break up())} strains of code")
besides Exception as e:
print(f"❌ Code era error: {e}")
print("n🔄 Multi-Agent Workflow Demo")
strive:
workflow_inputs = {'target_url': 'https://httpbin.org/html'}
workflow_result = orchestrator.execute_workflow('competitive_analysis', workflow_inputs)
print(f"✅ Workflow Success: Accomplished in {workflow_result['metadata']['total_execution_time']:.2f}s")
besides Exception as e:
print(f"❌ Workflow error: {e}")
We run a set of manufacturing demos to validate every element: first, our web_analyst performs a full-site evaluation; subsequent, our data_scientist crunches pattern CSV stats; then our code_architect generates an API consumer; and eventually we orchestrate the end-to-end aggressive evaluation workflow, capturing success indicators, outputs, and execution timing for every step.
print("n📈 System Efficiency Metrics")
system_status = orchestrator.get_system_status()
print(f"System Standing: {json.dumps(system_status, indent=2)}")
print("nTool Efficiency:")
for agent_name, agent in orchestrator.brokers.objects():
print(f"n{agent_name}:")
for tool_name, instrument in agent.instruments.objects():
print(f" - {tool_name}: {instrument.calls} calls, {instrument.avg_execution_time:.3f}s avg, {instrument.error_rate:.1%} error charge")
print("n✅ Superior Customized Agent Framework Full!")
print("🚀 Manufacturing-ready implementation with full monitoring and error dealing with!")
We end by retrieving and printing our orchestrator’s total system standing, itemizing registered brokers, workflows, and cache dimension, then loop via every agent’s instruments to show name counts, common execution instances, and error charges. This offers us a real-time view of efficiency and reliability earlier than we log a remaining affirmation that our production-ready agent framework is full.
In conclusion, we now have a blueprint for creating specialised AI brokers that carry out advanced analyses and generate production-quality code, and in addition self-monitor their execution well being and useful resource utilization. The AgentOrchestrator ties all the pieces collectively, enabling you to coordinate multi-step workflows and seize granular efficiency insights throughout brokers. Whether or not you’re automating market analysis, ETL duties, or API consumer era, this framework gives the extensibility, reliability, and observability required for enterprise-grade AI deployments.
Take a look at the Codes. All credit score for this analysis goes to the researchers of this venture. Additionally, be happy to comply with us on Twitter and don’t neglect to affix our 100k+ ML SubReddit and Subscribe to our E-newsletter.
Asif Razzaq is the CEO of Marktechpost Media Inc.. As a visionary entrepreneur and engineer, Asif is dedicated to harnessing the potential of Synthetic Intelligence for social good. His most up-to-date endeavor is the launch of an Synthetic Intelligence Media Platform, Marktechpost, which stands out for its in-depth protection of machine studying and deep studying information that’s each technically sound and simply comprehensible by a large viewers. The platform boasts of over 2 million month-to-month views, illustrating its recognition amongst audiences.
