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 internet intelligence and information science modules to superior code mills. 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 way in which, we reveal real-world examples, full with retry logic, logging, and efficiency metrics, so you may 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, Checklist, Non-obligatory, 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 information dealing with libraries akin to pandas and NumPy, and utilities like BeautifulSoup for internet scraping and scikit-learn for machine studying. We configure a standardized logging setup to seize info and error messages, and outline international constants for API timeouts and retry limits, making certain our instruments behave predictably in manufacturing.
@dataclass
class ToolResult:
"""Standardized instrument outcome construction"""
success: bool
information: Any
error: Non-obligatory[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
attempt:
outcome = 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,
information=outcome,
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"Instrument {self.title} failed: {str(e)}")
return ToolResult(
success=False,
information=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 information, and error particulars if it failed. Our CustomTool base class then wraps particular person capabilities 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, activity: str) -> Dict[str, Any]:
"""Execute a activity utilizing accessible instruments"""
logger.data(f"Agent {self.title} executing activity: {activity}")
task_lower = activity.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, activity)
if urls:
outcome = self.instruments['advanced_web_intelligence'].execute(urls[0])
outcomes.append(outcome)
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 activity:
data_start = activity.discover('title,age,wage')
data_part = activity[data_start:]
outcome = self.instruments['advanced_data_science_toolkit'].execute(data_part, 'stats')
outcomes.append(outcome)
elif any(key phrase in task_lower for key phrase in ['generate', 'code', 'api', 'client']):
if 'advanced_code_generator' in self.instruments:
outcome = self.instruments['advanced_code_generator'].execute(activity)
outcomes.append(outcome)
return {
'agent': self.title,
'activity': activity,
'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 activity to the proper instrument primarily based on easy key phrase matching. Within the run() methodology, we log the duty, choose the suitable instrument (internet intelligence, information evaluation, or code era), execute it, and combination the outcomes right into a standardized response that features success charges and timing metrics. This design permits us to simply lengthen brokers by including new instruments and maintains our orchestration as each clear and measurable.
print("🏗️ Constructing Superior Instrument Structure")
def performance_monitor(func):
"""Decorator for monitoring instrument efficiency"""
@wraps(func)
def wrapper(*args, **kwargs):
start_time = time.time()
attempt:
outcome = func(*args, **kwargs)
execution_time = time.time() - start_time
logger.data(f"{func.__name__} executed in {execution_time:.2f}s")
return outcome
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 internet intelligence gathering with a number of evaluation modes.
Args:
url: Goal URL for evaluation
analysis_type: Kind of research (complete, sentiment, technical, website positioning)
Returns:
Dict containing structured evaluation outcomes
"""
attempt:
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(information: str, operation: str) -> Dict[str, Any]:
"""
Complete information science operations with statistical evaluation.
Args:
information: CSV-like string or JSON information
operation: Kind of research (stats, correlation, forecast, clustering)
Returns:
Dict with evaluation outcomes
"""
attempt:
if information.startswith('{') or information.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()
outcome = {
'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:
outcome['summary_stats'] = df[numeric_columns].describe().to_dict()
if len(numeric_columns) > 1:
outcome['correlation_matrix'] = df[numeric_columns].corr().to_dict()
return outcome
elif operation == "clustering":
from sklearn.cluster import KMeans
from sklearn.preprocessing import StandardScaler
numeric_df = df.select_dtypes(embody=[np.number])
if numeric_df.form[1] < 2:
return {'error': 'Want not less than 2 numeric columns for clustering'}
scaler = StandardScaler()
scaled_data = scaler.fit_transform(numeric_df.fillna(0))
n_clusters = min(3, max(2, len(numeric_df) // 2))
kmeans = KMeans(n_clusters=n_clusters, random_state=42, n_init=10)
clusters = kmeans.fit_predict(scaled_data)
return {
'n_clusters': n_clusters,
'cluster_centers': kmeans.cluster_centers_.tolist(),
'cluster_labels': clusters.tolist(),
'inertia': float(kmeans.inertia_),
'feature_names': numeric_df.columns.tolist()
}
besides Exception as e:
return {'error': f"Knowledge evaluation failed: {str(e)}"}
@performance_monitor
def advanced_code_generator(task_description: str, language: str = "python") -> Dict[str, str]:
"""
Superior code era with a number of language assist and optimization.
Args:
task_description: Description of coding activity
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, Non-obligatory
class APIClient:
"""Manufacturing-ready API consumer with retry logic and error dealing with"""
def __init__(self, base_url: str, api_key: Non-obligatory[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-Kind': 'utility/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):
attempt:
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: Non-obligatory[Dict] = None) -> Dict[str, Any]:
return self._make_request('GET', endpoint, params=params)
def publish(self, endpoint: str, information: Non-obligatory[Dict] = None) -> Dict[str, Any]:
return self._make_request('POST', endpoint, json=information)
def put(self, endpoint: str, information: Non-obligatory[Dict] = None) -> Dict[str, Any]:
return self._make_request('PUT', endpoint, json=information)
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 Checklist, Dict, Any, Non-obligatory
import logging
logger = logging.getLogger(__name__)
class DataProcessor:
"""Superior information processor with complete cleansing and evaluation"""
def __init__(self, information: pd.DataFrame):
self.original_data = information.copy()
self.processed_data = information.copy()
self.processing_log = []
def clean_data(self, technique: str="auto") -> 'DataProcessor':
"""Clear information 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(embody=[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(embody=['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 information: {initial_shape} -> {final_shape}")
return self
def normalize(self, methodology: str="minmax", columns: Non-obligatory[List[str]] = None) -> 'DataProcessor':
"""Normalize numerical columns"""
cols = columns or self.processed_data.select_dtypes(embody=[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 information 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(embody=[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 information processor with cleansing, normalization, and perception era"
else:
code = f'''# Generated code template for: {task_description}
# Language: {language}
class CustomSolution:
"""Auto-generated answer template"""
def __init__(self):
self.initialized = True
def execute(self, *args, **kwargs):
"""Most important execution methodology - implement your logic right here"""
return {{"message": "Implement your customized logic right here", "activity": "{task_description}"}}
# Utilization instance:
# answer = CustomSolution()
# outcome = answer.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 internet scraping, advanced_data_science_toolkit for statistical evaluation and clustering on CSV or JSON information, and advanced_code_generator for producing production-ready code templates, making certain we monitor efficiency and preserve 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: Checklist[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']
activity = step['task'].format(**inputs, **outcomes)
if agent_name in self.brokers:
step_start = time.time()
outcome = self.brokers[agent_name].run(activity)
step_time = time.time() - step_start
outcomes[step['output_key']] = outcome
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': record(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 internet evaluation and intelligence gathering",
func=advanced_web_intelligence
)
data_tool = CustomTool(
title="advanced_data_science_toolkit",
description="Complete information 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 internet evaluation specialist. Present complete web site evaluation and insights."
)
data_agent = orchestrator.create_specialist_agent(
"data_scientist",
[data_tool],
"You're a information 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 internet intelligence, information 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 permits 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 internet 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 information 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 complicated duties end-to-end with minimal boilerplate.
print("🚀 Operating Manufacturing Examples")
print("n📊 Superior Internet Intelligence Demo")
attempt:
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"""
attempt:
data_result = data_agent.run(f"Analyze this information 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")
attempt:
code_result = code_agent.run("Generate an API consumer for information processing duties")
print(f"✅ Code Era 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")
attempt:
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 part: 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 general system standing, itemizing registered brokers, workflows, and cache dimension, then loop by way of 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 closing affirmation that our production-ready agent framework is full.
In conclusion, we now have a blueprint for creating specialised AI brokers that carry out complicated 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 mission. Additionally, be at liberty to observe 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.
