In this advanced tutorial, we aim to Create a multi-agent task automation system Using the Nexus Primisai Framework, which is entirely integrated into the Openai API. Our main objective is to demonstrate how hierarchical supervision, intelligent use of tools and structured outputs can facilitate the coordination of several AI agents to perform complex tasks, ranging from planning and development to quality insurance and data analysis. While we cross each phase, we do not only build individual agents; We architect a collaborative ecosystem where each agent has a clear role, responsibilities and intelligent tools to accomplish the task.
!pip install 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 installing basic outbuildings: Primisai for agent orchestration, OpenAI for LLM Access and Nest_asyncio to manage the quirks of the Colab event loop. After applying Nest_asyncio, we make sure that the notebook is ready to perform asynchronous tasks in a transparent manner, a key requirement for multi-agent execution.
print("🚀 PrimisAI Nexus Advanced Tutorial with OpenAI API")
print("=" * 55)
os.environ("OPENAI_API_KEY") = "Use Your Own API Key Here5"
# llm_config = {
# "api_key": os.environ("OPENAI_API_KEY"),
# "model": "gpt-4o-mini",
# "base_url": "https://api.openai.com/v1",
# "temperature": 0.7
# }
llm_config = {
"api_key": os.environ("OPENAI_API_KEY"),
"model": "gpt-3.5-turbo",
"base_url": "https://api.openai.com/v1",
"temperature": 0.7
}
print("📋 API Configuration:")
print(f"• Model: {llm_config('model')}")
print(f"• Base URL: {llm_config('base_url')}")
print("• Note: OpenAI has limited free tokens through April 2025")
print("• Alternative: Consider Puter.js for unlimited free access")To feed our agents, we connect to OpenAi models, starting with GPT-3.5-Turbo for profitable tasks. We store our API key in environmental variables and build a configuration dictionary specifying the model, temperature and basic URL. This section allows us to switch flexibly between the models, such as GPT-4O-MINI or GPT-4O, depending on the complexity and cost of tasks.
code_schema = {
"type": "object",
"properties": {
"description": {"type": "string", "description": "Code explanation"},
"code": {"type": "string", "description": "Python code implementation"},
"language": {"type": "string", "description": "Programming language"},
"complexity": {"type": "string", "enum": ("beginner", "intermediate", "advanced")},
"test_cases": {"type": "array", "items": {"type": "string"}, "description": "Example usage"}
},
"required": ("description", "code", "language")
}
analysis_schema = {
"type": "object",
"properties": {
"summary": {"type": "string", "description": "Brief analysis summary"},
"insights": {"type": "array", "items": {"type": "string"}, "description": "Key insights"},
"recommendations": {"type": "array", "items": {"type": "string"}, "description": "Action items"},
"confidence": {"type": "number", "minimum": 0, "maximum": 1},
"methodology": {"type": "string", "description": "Analysis approach used"}
},
"required": ("summary", "insights", "confidence")
}
planning_schema = {
"type": "object",
"properties": {
"tasks": {"type": "array", "items": {"type": "string"}, "description": "List of tasks to complete"},
"priority": {"type": "string", "enum": ("low", "medium", "high")},
"estimated_time": {"type": "string", "description": "Time estimate"},
"dependencies": {"type": "array", "items": {"type": "string"}, "description": "Task dependencies"}
},
"required": ("tasks", "priority")
}
We define the JSON diagrams for three types of agents: CODEWRITER, data analyst and project planner. These patterns apply the structure in the agent's responses, which makes the exit readable by machine and predictable. It helps us to ensure that the system returns consistent data, such as code blocks, information or project times, even when different LLMs are behind the scenes.
def calculate_metrics(data_str):
"""Calculate comprehensive statistics for numerical data"""
try:
data = json.loads(data_str) if isinstance(data_str, str) else data_str
if isinstance(data, list) and all(isinstance(x, (int, float)) for x in data):
import statistics
return {
"mean": statistics.mean(data),
"median": statistics.median(data),
"mode": statistics.mode(data) if len(set(data)) < len(data) else "No mode",
"std_dev": statistics.stdev(data) if len(data) > 1 else 0,
"max": max(data),
"min": min(data),
"count": len(data),
"sum": sum(data)
}
return {"error": "Invalid data format - expecting array of numbers"}
except Exception as e:
return {"error": f"Could not parse data: {str(e)}"}
def validate_code(code):
"""Advanced code validation with syntax and basic security checks"""
try:
dangerous_imports = ('os', 'subprocess', 'eval', 'exec', '__import__')
security_warnings = ()
for danger in dangerous_imports:
if danger in code:
security_warnings.append(f"Potentially dangerous: {danger}")
compile(code, '', 'exec')
return {
"valid": True,
"message": "Code syntax is valid",
"security_warnings": security_warnings,
"lines": len(code.split('\n'))
}
except SyntaxError as e:
return {
"valid": False,
"message": f"Syntax error: {e}",
"line": getattr(e, 'lineno', 'unknown'),
"security_warnings": ()
}
def search_documentation(query):
"""Simulate searching documentation (placeholder function)"""
docs = {
"python": "Python is a high-level programming language",
"list": "Lists are ordered, mutable collections in Python",
"function": "Functions are reusable blocks of code",
"class": "Classes define objects with attributes and methods"
}
results = ()
for key, value in docs.items():
if query.lower() in key.lower():
results.append(f"{key}: {value}")
return {
"query": query,
"results": results if results else ("No documentation found"),
"total_results": len(results)
} Then, we add personalized tools that agents could call, such as calculate_metrics for statistical summaries, validate_code for syntax and security checks, and Search_documentation for simulated programming aid. These tools extend the capacities of agents, by transforming them simple chatbots into interactive workers and focused on public services capable of autonomous reasoning and validation.
print("\n📋 Setting up Multi-Agent Hierarchy with OpenAI")
main_supervisor = Supervisor(
name="ProjectManager",
llm_config=llm_config,
system_message="You are a senior project manager coordinating development and analysis tasks. Delegate appropriately, provide clear summaries, and ensure quality delivery. Always consider time estimates and dependencies."
)
dev_supervisor = Supervisor(
name="DevManager",
llm_config=llm_config,
is_assistant=True,
system_message="You manage development tasks. Coordinate between coding, testing, and code review. Ensure best practices and security."
)
analysis_supervisor = Supervisor(
name="AnalysisManager",
llm_config=llm_config,
is_assistant=True,
system_message="You manage data analysis and research tasks. Ensure thorough analysis, statistical rigor, and actionable insights."
)
qa_supervisor = Supervisor(
name="QAManager",
llm_config=llm_config,
is_assistant=True,
system_message="You manage quality assurance and testing. Ensure thorough validation and documentation."
)
To simulate a real world management structure, we create a hierarchy on several levels. ProjectManager serves as a root supervisor, supervising three assistant supervisors (DevManager, Analysis and Qamanager), each in charge of agents specific to the domain. This modular hierarchy allows tasks to flow from high -level strategy to granular execution.
code_agent = Agent(
name="CodeWriter",
llm_config=llm_config,
system_message="You are an expert Python developer. Write clean, efficient, well-documented code with proper error handling. Always include test cases and follow PEP 8 standards.",
output_schema=code_schema,
tools=({
"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")
}
}
},
"tool": validate_code
}, {
"metadata": {
"function": {
"name": "search_documentation",
"description": "Search for programming documentation and examples",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "Documentation topic to search for"}
},
"required": ("query")
}
}
},
"tool": search_documentation
}),
use_tools=True
)
review_agent = Agent(
name="CodeReviewer",
llm_config=llm_config,
system_message="You are a senior code reviewer. Analyze code for best practices, efficiency, security, maintainability, and potential issues. Provide constructive feedback and suggestions.",
keep_history=True,
tools=({
"metadata": {
"function": {
"name": "validate_code",
"description": "Validates code syntax and security",
"parameters": {
"type": "object",
"properties": {
"code": {"type": "string", "description": "Code to validate"}
},
"required": ("code")
}
}
},
"tool": validate_code
}),
use_tools=True
)
analyst_agent = Agent(
name="DataAnalyst",
llm_config=llm_config,
system_message="You are a data scientist specializing in statistical analysis and insights generation. Provide thorough analysis with confidence metrics and actionable recommendations.",
output_schema=analysis_schema,
tools=({
"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")
}
}
},
"tool": calculate_metrics
}),
use_tools=True
)
planner_agent = Agent(
name="ProjectPlanner",
llm_config=llm_config,
system_message="You are a project planning specialist. Break down complex projects into manageable tasks with realistic time estimates and clear dependencies.",
output_schema=planning_schema
)
tester_agent = Agent(
name="QATester",
llm_config=llm_config,
system_message="You are a QA specialist focused on comprehensive testing strategies, edge cases, and quality assurance.",
tools=({
"metadata": {
"function": {
"name": "validate_code",
"description": "Validates code for testing",
"parameters": {
"type": "object",
"properties": {
"code": {"type": "string", "description": "Code to test"}
},
"required": ("code")
}
}
},
"tool": validate_code
}),
use_tools=True
)
We then build a diversified set of specialized agents: Codewriter to generate Python code, CoderereViewer to examine logic and security, dataalyst to carry out a structured data analysis, Projectplanner for the breakdown of tasks and Qateter for quality checks. Each agent has specific tools, output schemes and system instructions adapted to its role.
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 agents are recorded under their respective supervisors, and assistant supervisors are, in turn, recorded with the main supervisor. This configuration creates an entirely linked agent ecosystem, where the instructions could pass from the level of higher level to any specialized network agent.
print("\n🌳 Agent Hierarchy:")
main_supervisor.display_agent_graph()
print("\n🧪 Testing Full Multi-Agent Communication")
print("-" * 45)
try:
test_response = main_supervisor.chat("Hello! Please introduce your team and explain how you coordinate complex projects.")
print(f"✅ Supervisor communication test successful!")
print(f"Response preview: {test_response(:200)}...")
except Exception as e:
print(f"❌ Supervisor test failed: {str(e)}")
print("Falling back to direct agent testing...")We view the whole hierarchy using the display_agent_graph () to confirm our structure. It offers a clear view of how each agent is connected in the wider task management flow, a useful diagnosis before deployment.
print("\n🎯 Complex Multi-Agent Task Execution")
print("-" * 40)
complex_task = """Create a Python function that implements a binary search algorithm,
have it reviewed for optimization, tested thoroughly, and provide a project plan
for integrating it into a larger search system."""
print(f"Complex Task: {complex_task}")
try:
complex_response = main_supervisor.chat(complex_task)
print(f"✅ Complex task completed")
print(f"Response: {complex_response(:300)}...")
except Exception as e:
print(f"❌ Complex task failed: {str(e)}")We give the complete system a real world task: create a binary research function, revise it, test it and plan its integration into a larger project. ProjectManager perfectly coordinates agents through development, AQ and planning, demonstrating the true power of hierarchical orchestration and focused on tools.
print("\n🔧 Tool Integration & Structured Outputs")
print("-" * 43)
print("Testing Code Agent with tools...")
try:
code_response = code_agent.chat("Create a function to calculate fibonacci numbers with memoization")
print(f"✅ Code Agent with tools: Working")
print(f"Response type: {type(code_response)}")
if isinstance(code_response, str) and code_response.strip().startswith('{'):
code_data = json.loads(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" - Raw response: {code_response(:100)}...")
except Exception as e:
print(f"❌ Code Agent error: {str(e)}")
print("\nTesting Analyst Agent with tools...")
try:
analysis_response = analyst_agent.chat("Analyze this sales data: (100, 150, 120, 180, 200, 175, 160, 190, 220, 185). What trends do you see?")
print(f"✅ Analyst Agent with tools: Working")
if isinstance(analysis_response, str) and analysis_response.strip().startswith('{'):
analysis_data = json.loads(analysis_response)
print(f" - Summary: {analysis_data.get('summary', 'N/A')(:50)}...")
print(f" - Confidence: {analysis_data.get('confidence', 'N/A')}")
print(f" - Insights count: {len(analysis_data.get('insights', ()))}")
else:
print(f" - Raw response: {analysis_response(:100)}...")
except Exception as e:
print(f"❌ Analyst Agent error: {str(e)}")We directly test the capacities of two specialized agents by using real prompts. We first ask the Codewriter agent to generate a fibonacci function with memorization and validate that it returns a structured output containing a description of code, language and a level of complexity. Then, we assess the data agent by fueling it samples sales data to extract trends.
print("\n🔨 Manual Tool Usage")
print("-" * 22)
# Test all tools 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, value in metrics_result.items():
print(f" {key}: {value}")
print("\nCode validation test:")
test_code = """
def binary_search(arr, target):
left, right = 0, len(arr) - 1
while left <= right:
mid = (left + right) // 2
if arr(mid) == target:
return mid
elif arr(mid) < target:
left = mid + 1
else:
right = mid - 1
return -1
"""
validation_result = validate_code(test_code)
print(f"Validation result: {validation_result}")
print("\nDocumentation search test:")
doc_result = search_documentation("python function")
print(f"Search results: {doc_result}")We leave the agent's executive to directly test each tool. First of all, we use the Calculate_metrics tool on a set of data of ten numbers, confirming the properly returned statistics such as the average, the median, the mode and the standard deviation. Then, we execute the Validate_Code tool on an example of a binary research function, which confirms both syntactic accuracy and indicators without security warnings. Finally, we test the Search_documentation tool with the “Python function” of the request and receive relevant documentation extracts, checking its ability to effectively simulate contextual research.
print("\n🚀 Advanced 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, task in workflow_stages:
print(f"\n{stage} Stage: {task}")
try:
if stage == "Planning":
response = planner_agent.chat(task)
elif stage == "Development":
response = code_agent.chat(task)
elif stage == "Review":
response = review_agent.chat(task)
elif stage == "Testing":
response = tester_agent.chat(task)
elif stage == "Analysis":
response = analyst_agent.chat(task)
workflow_results(stage) = response
print(f"✅ {stage} completed: {response(:80)}...")
except Exception as e:
print(f"❌ {stage} failed: {str(e)}")
workflow_results(stage) = f"Error: {str(e)}"We simulate a life cycle of the project in five stages: planning, development, examination, test and analysis. Each task is transmitted to the most relevant agent and the responses are collected to assess performance. This demonstrates the capacity of the framework to manage workflows from start to finish without manual intervention.
print("\n📊 System Monitoring & Performance")
print("-" * 37)
debugger = Debugger(name="OpenAITutorialDebugger")
debugger.log("Advanced OpenAI tutorial execution completed successfully")
print(f"Main Supervisor ID: {main_supervisor.workflow_id}")We activate the debugger tool to follow the performance of our session events and the newspaper system. We also print the workflow_id of the main supervisor as a traceable identifier, useful when managing several workflows in production.
In conclusion, we have managed to build a fully automated and compatible Openai compatible multi-agent system using Primisai Nexus. Each agent works with clarity, precision and autonomy, whether by writing code, validating logic, analyzing data or decomposing complex workflows. Our hierarchical structure allows the delegation of transparent task and modular scalability. Primisai Nexus Framework establishes a robust base for the automation of real world tasks, whether in software development, research, planning or data operations, thanks to intelligent collaboration between specialized agents.
Discover the Codes. All the merit of this research goes to researchers in this project. Also, don't hesitate to follow us Twitter,, YouTube And Spotify And don't forget to join our Subseubdredit 100k + ml and subscribe to Our newsletter.
Asif Razzaq is the CEO of Marktechpost Media Inc .. as a visionary entrepreneur and engineer, AIF undertakes to exploit the potential of artificial intelligence for social good. His most recent company is the launch of an artificial intelligence media platform, Marktechpost, which stands out from its in-depth coverage of automatic learning and in-depth learning news which are both technically solid and easily understandable by a large audience. The platform has more than 2 million monthly views, illustrating its popularity with the public.
