On this superior tutorial, we goal to build a multi-agent task automation system utilizing the PrimisAI Nexus framework, which is totally built-in with the OpenAI API. Our major goal is to reveal how hierarchical supervision, clever device utilization, and structured outputs can facilitate the coordination of a number of AI brokers to carry out complicated duties, starting from planning and growth to high quality assurance and knowledge evaluation. As we stroll via every part, we don’t simply construct particular person brokers; we architect a collaborative ecosystem the place every agent has a transparent position, obligations, and good instruments to perform the duty.
!pip set up primisai openai nest-asyncio
import os
import nest_asyncio
from primisai.nexus.core import AI, Agent, Supervisor
from primisai.nexus.utils.debugger import Debugger
import json
nest_asyncio.apply()We start by putting in the core dependencies: Primisai for agent orchestration, OpenAI for LLM entry, and nest_asyncio to deal with Colab’s occasion loop quirks. After making use of nest_asyncio, we make sure the pocket book is able to execute asynchronous duties seamlessly, a key requirement for multi-agent execution.
print("🚀 PrimisAI Nexus Superior Tutorial with OpenAI API")
print("=" * 55)
os.environ["OPENAI_API_KEY"] = "Use Your Personal API Key Here5"
# llm_config = {
# "api_key": os.environ["OPENAI_API_KEY"],
# "mannequin": "gpt-4o-mini",
# "base_url": "https://api.openai.com/v1",
# "temperature": 0.7
# }
llm_config = {
"api_key": os.environ["OPENAI_API_KEY"],
"mannequin": "gpt-3.5-turbo",
"base_url": "https://api.openai.com/v1",
"temperature": 0.7
}
print("📋 API Configuration:")
print(f"• Mannequin: {llm_config['model']}")
print(f"• Base URL: {llm_config['base_url']}")
print("• Word: OpenAI has restricted free tokens via April 2025")
print("• Different: Think about Puter.js for limitless free entry")To energy our brokers, we connect with OpenAI’s fashions, beginning with gpt-3.5-turbo for cost-efficient duties. We retailer our API key in surroundings variables and assemble a configuration dictionary specifying the mannequin, temperature, and base URL. This part permits us to flexibly change between fashions, equivalent to gpt-4o-mini or gpt-4o, relying on job complexity and price.
code_schema = {
"sort": "object",
"properties": {
"description": {"sort": "string", "description": "Code rationalization"},
"code": {"sort": "string", "description": "Python code implementation"},
"language": {"sort": "string", "description": "Programming language"},
"complexity": {"sort": "string", "enum": ["beginner", "intermediate", "advanced"]},
"test_cases": {"sort": "array", "gadgets": {"sort": "string"}, "description": "Instance utilization"}
},
"required": ["description", "code", "language"]
}
analysis_schema = {
"sort": "object",
"properties": {
"abstract": {"sort": "string", "description": "Temporary evaluation abstract"},
"insights": {"sort": "array", "gadgets": {"sort": "string"}, "description": "Key insights"},
"suggestions": {"sort": "array", "gadgets": {"sort": "string"}, "description": "Motion gadgets"},
"confidence": {"sort": "quantity", "minimal": 0, "most": 1},
"methodology": {"sort": "string", "description": "Evaluation method used"}
},
"required": ["summary", "insights", "confidence"]
}
planning_schema = {
"sort": "object",
"properties": {
"duties": {"sort": "array", "gadgets": {"sort": "string"}, "description": "Checklist of duties to finish"},
"precedence": {"sort": "string", "enum": ["low", "medium", "high"]},
"estimated_time": {"sort": "string", "description": "Time estimate"},
"dependencies": {"sort": "array", "gadgets": {"sort": "string"}, "description": "Job dependencies"}
},
"required": ["tasks", "priority"]
}
We outline JSON schemas for 3 agent varieties: CodeWriter, Knowledge Analyst, and Challenge Planner. These schemas implement construction within the agent’s responses, making the output machine-readable and predictable. It helps us be sure that the system returns constant knowledge, equivalent to code blocks, insights, or mission timelines, even when completely different LLMs are behind the scenes.
def calculate_metrics(data_str):
"""Calculate complete statistics for numerical knowledge"""
attempt:
knowledge = json.masses(data_str) if isinstance(data_str, str) else data_str
if isinstance(knowledge, listing) and all(isinstance(x, (int, float)) for x in knowledge):
import statistics
return {
"imply": statistics.imply(knowledge),
"median": statistics.median(knowledge),
"mode": statistics.mode(knowledge) if len(set(knowledge)) < len(knowledge) else "No mode",
"std_dev": statistics.stdev(knowledge) if len(knowledge) > 1 else 0,
"max": max(knowledge),
"min": min(knowledge),
"depend": len(knowledge),
"sum": sum(knowledge)
}
return {"error": "Invalid knowledge format - anticipating array of numbers"}
besides Exception as e:
return {"error": f"Couldn't parse knowledge: {str(e)}"}
def validate_code(code):
"""Superior code validation with syntax and primary safety checks"""
attempt:
dangerous_imports = ['os', 'subprocess', 'eval', 'exec', '__import__']
security_warnings = []
for hazard in dangerous_imports:
if hazard in code:
security_warnings.append(f"Probably harmful: {hazard}")
compile(code, '', 'exec')
return {
"legitimate": True,
"message": "Code syntax is legitimate",
"security_warnings": security_warnings,
"traces": len(code.cut up('n'))
}
besides SyntaxError as e:
return {
"legitimate": False,
"message": f"Syntax error: {e}",
"line": getattr(e, 'lineno', 'unknown'),
"security_warnings": []
}
def search_documentation(question):
"""Simulate looking out documentation (placeholder perform)"""
docs = {
"python": "Python is a high-level programming language",
"listing": "Lists are ordered, mutable collections in Python",
"perform": "Capabilities are reusable blocks of code",
"class": "Lessons outline objects with attributes and strategies"
}
outcomes = []
for key, worth in docs.gadgets():
if question.decrease() in key.decrease():
outcomes.append(f"{key}: {worth}")
return {
"question": question,
"outcomes": outcomes if outcomes else ["No documentation found"],
"total_results": len(outcomes)
} Subsequent, we add customized instruments that brokers might name, equivalent to calculate_metrics for statistical summaries, validate_code for syntax and safety checks, and search_documentation for simulated programming assist. These instruments lengthen the brokers’ skills, turning them from easy chatbots into interactive, utility-driven staff able to autonomous reasoning and validation.
print("n📋 Establishing Multi-Agent Hierarchy with OpenAI")
main_supervisor = Supervisor(
identify="ProjectManager",
llm_config=llm_config,
system_message="You're a senior mission supervisor coordinating growth and evaluation duties. Delegate appropriately, present clear summaries, and guarantee high quality supply. All the time contemplate time estimates and dependencies."
)
dev_supervisor = Supervisor(
identify="DevManager",
llm_config=llm_config,
is_assistant=True,
system_message="You handle growth duties. Coordinate between coding, testing, and code overview. Guarantee greatest practices and safety."
)
analysis_supervisor = Supervisor(
identify="AnalysisManager",
llm_config=llm_config,
is_assistant=True,
system_message="You handle knowledge evaluation and analysis duties. Guarantee thorough evaluation, statistical rigor, and actionable insights."
)
qa_supervisor = Supervisor(
identify="QAManager",
llm_config=llm_config,
is_assistant=True,
system_message="You handle high quality assurance and testing. Guarantee thorough validation and documentation."
)
To simulate a real-world administration construction, we create a multi-tiered hierarchy. A ProjectManager serves as the foundation supervisor, overseeing three assistant supervisors (DevManager, AnalysisManager, and QAManager), every answerable for domain-specific brokers. This modular hierarchy permits duties to circulation down from high-level technique to granular execution.
code_agent = Agent(
identify="CodeWriter",
llm_config=llm_config,
system_message="You're an professional Python developer. Write clear, environment friendly, well-documented code with correct error dealing with. All the time embody check circumstances and comply with PEP 8 requirements.",
output_schema=code_schema,
instruments=[{
"metadata": {
"function": {
"name": "validate_code",
"description": "Validates Python code syntax and checks for security issues",
"parameters": {
"type": "object",
"properties": {
"code": {"type": "string", "description": "Python code to validate"}
},
"required": ["code"]
}
}
},
"device": validate_code
}, {
"metadata": {
"perform": {
"identify": "search_documentation",
"description": "Seek for programming documentation and examples",
"parameters": {
"sort": "object",
"properties": {
"question": {"sort": "string", "description": "Documentation subject to seek for"}
},
"required": ["query"]
}
}
},
"device": search_documentation
}],
use_tools=True
)
review_agent = Agent(
identify="CodeReviewer",
llm_config=llm_config,
system_message="You're a senior code reviewer. Analyze code for greatest practices, effectivity, safety, maintainability, and potential points. Present constructive suggestions and ideas.",
keep_history=True,
instruments=[{
"metadata": {
"function": {
"name": "validate_code",
"description": "Validates code syntax and security",
"parameters": {
"type": "object",
"properties": {
"code": {"type": "string", "description": "Code to validate"}
},
"required": ["code"]
}
}
},
"device": validate_code
}],
use_tools=True
)
analyst_agent = Agent(
identify="DataAnalyst",
llm_config=llm_config,
system_message="You're a knowledge scientist specializing in statistical evaluation and insights era. Present thorough evaluation with confidence metrics and actionable suggestions.",
output_schema=analysis_schema,
instruments=[{
"metadata": {
"function": {
"name": "calculate_metrics",
"description": "Calculates comprehensive statistics for numerical data",
"parameters": {
"type": "object",
"properties": {
"data_str": {"type": "string", "description": "JSON string of numerical data array"}
},
"required": ["data_str"]
}
}
},
"device": calculate_metrics
}],
use_tools=True
)
planner_agent = Agent(
identify="ProjectPlanner",
llm_config=llm_config,
system_message="You're a mission planning specialist. Break down complicated initiatives into manageable duties with reasonable time estimates and clear dependencies.",
output_schema=planning_schema
)
tester_agent = Agent(
identify="QATester",
llm_config=llm_config,
system_message="You're a QA specialist centered on complete testing methods, edge circumstances, and high quality assurance.",
instruments=[{
"metadata": {
"function": {
"name": "validate_code",
"description": "Validates code for testing",
"parameters": {
"type": "object",
"properties": {
"code": {"type": "string", "description": "Code to test"}
},
"required": ["code"]
}
}
},
"device": validate_code
}],
use_tools=True
)
We then construct a various set of specialised brokers: CodeWriter for producing Python code, CodeReviewer for reviewing logic and safety, DataAnalyst for performing structured knowledge evaluation, ProjectPlanner for job breakdown, and QATester for high quality checks. Every agent has domain-specific instruments, output schemas, and system directions tailor-made to their position.
dev_supervisor.register_agent(code_agent)
dev_supervisor.register_agent(review_agent)
analysis_supervisor.register_agent(analyst_agent)
qa_supervisor.register_agent(tester_agent)
main_supervisor.register_agent(dev_supervisor)
main_supervisor.register_agent(analysis_supervisor)
main_supervisor.register_agent(qa_supervisor)
main_supervisor.register_agent(planner_agent)All brokers are registered beneath their respective supervisors, and the assistant supervisors are, in flip, registered with the primary supervisor. This setup creates a completely linked agent ecosystem, the place directions might cascade from the top-level agent to any specialist agent within the community.
print("n🌳 Agent Hierarchy:")
main_supervisor.display_agent_graph()
print("n🧪 Testing Full Multi-Agent Communication")
print("-" * 45)
attempt:
test_response = main_supervisor.chat("Hi there! Please introduce your group and clarify the way you coordinate complicated initiatives.")
print(f"✅ Supervisor communication check profitable!")
print(f"Response preview: {test_response[:200]}...")
besides Exception as e:
print(f"❌ Supervisor check failed: {str(e)}")
print("Falling again to direct agent testing...")We visualize your entire hierarchy utilizing display_agent_graph() to verify our construction. It presents a transparent view of how every agent is related throughout the broader job administration circulation, a useful diagnostic earlier than deployment.
print("n🎯 Complicated Multi-Agent Job Execution")
print("-" * 40)
complex_task = """Create a Python perform that implements a binary search algorithm,
have it reviewed for optimization, examined totally, and supply a mission plan
for integrating it into a bigger search system."""
print(f"Complicated Job: {complex_task}")
attempt:
complex_response = main_supervisor.chat(complex_task)
print(f"✅ Complicated job accomplished")
print(f"Response: {complex_response[:300]}...")
besides Exception as e:
print(f"❌ Complicated job failed: {str(e)}")We give the total system a real-world job: create a binary search perform, overview it, check it, and plan its integration into a bigger mission. The ProjectManager seamlessly coordinates brokers throughout growth, QA, and planning, demonstrating the true energy of hierarchical, tool-driven agent orchestration.
print("n🔧 Software Integration & Structured Outputs")
print("-" * 43)
print("Testing Code Agent with instruments...")
attempt:
code_response = code_agent.chat("Create a perform to calculate fibonacci numbers with memoization")
print(f"✅ Code Agent with instruments: Working")
print(f"Response sort: {sort(code_response)}")
if isinstance(code_response, str) and code_response.strip().startswith('{'):
code_data = json.masses(code_response)
print(f" - Description: {code_data.get('description', 'N/A')[:50]}...")
print(f" - Language: {code_data.get('language', 'N/A')}")
print(f" - Complexity: {code_data.get('complexity', 'N/A')}")
else:
print(f" - Uncooked response: {code_response[:100]}...")
besides Exception as e:
print(f"❌ Code Agent error: {str(e)}")
print("nTesting Analyst Agent with instruments...")
attempt:
analysis_response = analyst_agent.chat("Analyze this gross sales knowledge: [100, 150, 120, 180, 200, 175, 160, 190, 220, 185]. What traits do you see?")
print(f"✅ Analyst Agent with instruments: Working")
if isinstance(analysis_response, str) and analysis_response.strip().startswith('{'):
analysis_data = json.masses(analysis_response)
print(f" - Abstract: {analysis_data.get('abstract', 'N/A')[:50]}...")
print(f" - Confidence: {analysis_data.get('confidence', 'N/A')}")
print(f" - Insights depend: {len(analysis_data.get('insights', []))}")
else:
print(f" - Uncooked response: {analysis_response[:100]}...")
besides Exception as e:
print(f"❌ Analyst Agent error: {str(e)}")We instantly check the capabilities of two specialised brokers utilizing actual prompts. We first ask the CodeWriter agent to generate a Fibonacci perform with memoization and validate that it returns structured output containing a code description, language, and complexity stage. Then, we consider the DataAnalyst agent by feeding it pattern gross sales knowledge to extract traits.
print("n🔨 Handbook Software Utilization")
print("-" * 22)
# Check all instruments manually
sample_data = "[95, 87, 92, 88, 91, 89, 94, 90, 86, 93]"
metrics_result = calculate_metrics(sample_data)
print(f"Statistics for {sample_data}:")
for key, worth in metrics_result.gadgets():
print(f" {key}: {worth}")
print("nCode validation check:")
test_code = """
def binary_search(arr, goal):
left, proper = 0, len(arr) - 1
whereas left <= proper:
mid = (left + proper) // 2
if arr[mid] == goal:
return mid
elif arr[mid] < goal:
left = mid + 1
else:
proper = mid - 1
return -1
"""
validation_result = validate_code(test_code)
print(f"Validation outcome: {validation_result}")
print("nDocumentation search check:")
doc_result = search_documentation("python perform")
print(f"Search outcomes: {doc_result}")We step exterior the agent framework to check every device instantly. First, we use the calculate_metrics device on a dataset of ten numbers, confirming it appropriately returned statistics equivalent to imply, median, mode, and customary deviation. Subsequent, we run the validate_code device on a pattern binary search perform, which confirms each syntactic correctness and flags no safety warnings. Lastly, we check the search_documentation device with the question “python perform” and obtain related documentation snippets, verifying its skill to effectively simulate contextual lookup.
print("n🚀 Superior Multi-Agent Workflow")
print("-" * 35)
workflow_stages = [
("Planning", "Create a project plan for building a web scraper for news articles"),
("Development", "Implement the web scraper with error handling and rate limiting"),
("Review", "Review the web scraper code for security and efficiency"),
("Testing", "Create comprehensive test cases for the web scraper"),
("Analysis", "Analyze sample scraped data: [45, 67, 23, 89, 12, 56, 78, 34, 91, 43]")
]
workflow_results = {}
for stage, job in workflow_stages:
print(f"n{stage} Stage: {job}")
attempt:
if stage == "Planning":
response = planner_agent.chat(job)
elif stage == "Growth":
response = code_agent.chat(job)
elif stage == "Evaluate":
response = review_agent.chat(job)
elif stage == "Testing":
response = tester_agent.chat(job)
elif stage == "Evaluation":
response = analyst_agent.chat(job)
workflow_results[stage] = response
print(f"✅ {stage} accomplished: {response[:80]}...")
besides Exception as e:
print(f"❌ {stage} failed: {str(e)}")
workflow_results[stage] = f"Error: {str(e)}"We simulate a five-stage mission lifecycle: planning, growth, overview, testing, and evaluation. Every job is handed to essentially the most related agent, and responses are collected to guage efficiency. This demonstrates the framework’s functionality to handle end-to-end workflows with out guide intervention.
print("n📊 System Monitoring & Efficiency")
print("-" * 37)
debugger = Debugger(identify="OpenAITutorialDebugger")
debugger.log("Superior OpenAI tutorial execution accomplished efficiently")
print(f"Most important Supervisor ID: {main_supervisor.workflow_id}")We activate the Debugger device to trace the efficiency of our session and log system occasions. We additionally print the primary supervisor’s workflow_id as a traceable identifier, helpful when managing a number of workflows in manufacturing.
In conclusion, we now have efficiently constructed a completely automated, OpenAI-compatible multi-agent system utilizing PrimisAI Nexus. Every agent operates with readability, precision, and autonomy, whether or not writing code, validating logic, analyzing knowledge, or breaking down complicated workflows. Our hierarchical construction permits for seamless job delegation and modular scalability. PrimisAI Nexus framework establishes a sturdy basis for automating real-world duties, whether or not in software program growth, analysis, planning, or knowledge operations, via clever collaboration between specialised brokers.
Try the Codes. All credit score for this analysis goes to the researchers of this mission. Additionally, be happy to comply with us on Twitter, Youtube and Spotify and don’t overlook to affix our 100k+ ML SubReddit and Subscribe to our 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.