GitHubDiscordX

Livepeer Youtube Video Generator Swarm

Imagine automating the process of generating and uploading videos to YouTube with the power of AI and decentralized networks. In this tutorial, you'll learn how to use the Livepeer Youtube Video Generator Swarm to create a swarm of AI agents that collaborate to manage video generation and upload tasks seamlessly.

By leveraging Livepeer’s decentralized video platform and YouTube’s vast audience reach, these agents work together to automatically create and upload video content without the need for manual intervention.

Whether you want to streamline content creation, automate video uploads, or explore decentralized media solutions, this agent-driven process simplifies video management and distribution across multiple platforms, showcasing the potential of decentralized video processing.

Prerequisites

  • Python 3.11 or higher installed

  • Git installed

  • An IDE or text editor (VS Code, PyCharm, etc.)

  • Basic understanding of Python programming

  • API keys from:

    • OpenAI

    • Anthropic

  • Livepeer AI

  • Google API Client

  • MoviePy

To set up YouTube API credentials, go to the Google API Console, create a new project, enable the YouTube Data API v3, and set up OAuth 2.0 credentials by configuring a consent screen and adding valid redirect URIs (http://localhost:8088/flowName=GeneralOAuthFlow) and (http://localhost:8088/) for authentication.

Download the client_secret.json file and place it in your project’s root directory to handle secure OAuth 2.0 authentication, which is necessary for authorizing API requests and accessing private YouTube data securely. Also, register yourself as a test user in the OAuth Consent Screen.

To get the Livepeer API key, sign up for an account at Livepeer, navigate to the "API" section in the dashboard, and create a new API key. This key is needed to authenticate and interact with Livepeer's decentralized video streaming network, allowing you to integrate video streaming, transcoding, and other features into your application using their APIs.

Step-by-Step Tutorial for Building a Livepeer YouTube Video Generator Swarm

Below is the step-by-step tutorial to build the swarm which will autonomously automate the process of creating and uploading YouTube videos.

Cloning the Repository

First, clone the repository containing the base code for building your dApp.

Step 1: Open your terminal or command prompt and run the following commands:

git clone https://github.com/hivenetwork-ai/example-agents/
cd livepeer_youtube_swarm

These commands will download the repository and set up the directory to start working on the agent.

Setting Up the Environment

Now, set up the environment required to run the agent.

Step 1: Create a virtual environment:

python -m venv ./venv

Step 2: Activate the virtual environment:

  • On Windows

    • Command Prompt:

      venv\Scripts\activate.bat

    • PowerShell:

      venv\Scripts\Activate.ps1

  • On Linux/Mac:

    venv/bin/activate

Installing Dependencies

With the environment activated, install the required dependencies:

pip install -r requirements.txt
pip install git+https://github.com/hivenetwork-ai/hive-agent-py.git@main#egg=hive-agent[web3]

Setting Up Environment Variables

Step 1: Create a new file called .env in the root directory:

touch .env

Step 2: Copy the contents of .env.example into your new .env file.

Step 3: Add your API keys to the .env file:

MODEL=gpt-4-turbo-preview
OPENAI_API_KEY=<replace_with_your_openai_api_key>
ANTHROPIC_API_KEY=<replace_with_your_anthropic_api_key>
ENVIRONMENT=dev
HIVE_AGENT_LOG_LEVEL=INFO
LIVEPEER_API_KEY=<replace_with_your_livepeer_api_key>

Understanding the Code Structure

The repository contains essential components for running the Livepeer YouTube video generator swarm.

  • hive_swarm folder: Contains the agents responsible for generating and uploading videos to YouTube, along with helper tools for integrating with Livepeer and YouTube APIs.

  • swarm.py: The main script that orchestrates the swarm agents and manages the task distribution for video generation and upload.

  • main.py: The core driver script where you input prompts for video creation and interact with the agents to execute the tasks.

  • .env: Stores your API keys and configuration settings necessary for accessing the Livepeer and YouTube services.

  • requirements.txt: Lists the dependencies required to run the agents, including libraries for API interaction and video processing.

Creating the Tools Folder

To set up the tools folder, you'll need to create a directory within your project structure and add the necessary Python files that facilitate various functionalities. Below is a step-by-step guide on how to create the tools folder and its contents.

Add Required Python Files:

Inside the tools folder, we will be creating the following Python files:

  • __init__.py

  • files.py

  • livepeer_api.py

  • video_editor.py

  • youtube_upload.py

init.py

This file initializes the tools module and imports necessary functions from other modules.

from .files import save_to_file, read_from_file, list_files, download_from_url
from .livepeer_api import text_to_image, image_to_video
from .video_editor import video_editor
from .youtube_upload import upload_video

files.py

Contains utility functions for file handling such as saving, reading, and listing files.

files.py
import os
import requests

def save_to_file(content: str, file_name: str, file_path: str) -> None:
    """
    Saves the given content to a file at the specified path with the specified file name.

    :param content: The content to be written to the file.
    :param file_name: The name of the file to save the content in.
    :param file_path: The path where the file should be saved.
    """

    # ensure the directory exists
    os.makedirs(file_path, exist_ok=True)

    # construct the full file path
    full_path = os.path.join(file_path, file_name)

    # write the content to the file
    with open(full_path, "w") as file:
        file.write(content)

    print(f"File saved to {full_path}")


def read_from_file(file_path: str) -> str:
    """
    Reads the content from a file at the specified path.

    :param file_path: The path of the file to read.
    :return: The content of the file.
    """
    with open(file_path, "r") as file:
        return file.read()
    
def list_files(file_path: str) -> list[str]:
    """
    Lists the files in the specified directory.

    :param file_path: The path of the directory to list.
    :return: A list of files in the directory.
    """
    return os.listdir(file_path)



def download_from_url(url: str, file_name: str,file_path: str) -> str:

    """
    Saves the given content to a file at the specified path with the specified file name.

    :param url: Url to download the image from.
    :param file_name: Name of the file to save the image to.
    :param file_path: Path to save the image to.
    :return: Path to the saved image.
    """
    response = requests.get(url)
    if response.status_code == 200:
        # Create the full directory path
        os.makedirs(file_path, exist_ok=True)
        
        full_path = os.path.join(file_path, file_name)
        
        # Write the image content to the file
        with open(full_path, "wb") as f:
            f.write(response.content)
        
        return full_path
    else:
        raise Exception(f"Failed to download image: HTTP {response.status_code}")

livepeer_api.py

This file integrates with the Livepeer API for various functionalities, including text-to-image and image-to-video conversions.

livepeer_api.py
from livepeer_ai import Livepeer
from dotenv import load_dotenv
import os

load_dotenv()

livepeer_api_key= os.getenv("LIVEPEER_API_KEY")

text2img_model_list = [
    'SG161222/RealVisXL_V4.0_Lightning',
    'ByteDance/SDXL-Lightning',
    'SG161222/Realistic_Vision_V6.0_B1_noVAE',
    'stabilityai/stable-diffusion-xl-base-1.0',
    'runwayml/stable-diffusion-v1-5',
    'prompthero/openjourney-v4',
    'SG161222/RealVisXL_V4.0',
    'stabilityai/sd-turbo',
    'stabilityai/sdxl-turbo',
    'stabilityai/stable-diffusion-3-medium-diffusers'
]

img2video_model_list = [
    'stable-video-diffusion-img2vid-xt',
    'stabilityai/stable-video-diffusion-img2vid-xt-1-1',   
]

img2img_model_list = [
    'timbrooks/instruct-pix2pix',
    'ByteDance/SDXL-Lightning',
    'SG161222/RealVisXL_V4.0',
    'SG161222/RealVisXL_V4.0_Lightning',
    'stabilityai/sd-turbo',
    'stabilityai/sdxl-turbo'
]

segment_model_list = [
"facebook/sam2-hiera-large"
"facebook/sam2-hiera-base-plus",
"facebook/sam2-hiera-small",
"facebook/sam2-hiera-tiny"
]

def text_to_image(prompt: str, model_id: str = text2img_model_list[1]):
    """
    This function generates an image from a text prompt using the Livepeer AI API.

    Args:
        prompt (str): The text prompt to guide image generation.
        
    Returns:
        str: The URL of the generated image.
    """
    s = Livepeer(http_bearer=livepeer_api_key)

    res = s.generate.text_to_image(request={
        "prompt": prompt,
        "model_id": model_id,
        # Optional fields
        # "loras": { "latent-consistency/lcm-lora-sdxl": 1.0, "nerijs/pixel-art-xl": 1.2 },  # Low-Rank Adaptation models and weights
        # "height": 1080,                              # Height of the generated image in pixels
        # "width": 1920,                               # Width of the generated image in pixels
        # "guidance_scale": 7.5,                       # Degree to which the model follows the prompt (higher values = closer to prompt)
        # "negative_prompt": "bad quality",            # Text prompt to exclude from image generation
        # "safety_check": True,                        # Perform a safety check to filter offensive content
        # "seed": 42,                                  # Set seed for reproducibility
        # "num_inference_steps": 50,                   # Number of denoising steps for improved quality
        # "num_images_per_prompt": 1                   # Number of images generated per prompt
    })
    
    print(res)
    
    if res.image_response is not None and res.image_response.images:
        generated_image_url = res.image_response.images[0].url
        return generated_image_url
    else:
        return None
    
def image_to_video(image_path: str, model_id: str = img2video_model_list[1]):
    """
    This function generates a video from an image using the Livepeer AI API.

    Args:
        image_path (str): The path to the image to be used for video generation.
        
    Returns:
        str: The URL of the generated video.
    """
    s = Livepeer(http_bearer=livepeer_api_key)

    with open(image_path, "rb") as image_file:
        image_content = image_file.read()

    res = s.generate.image_to_video(request={
        "image": {
            "file_name": image_path.split('/')[-1],
            "content": image_content,
        },
        "model_id": model_id,
        # Optional fields
        # "height": 1080,                             # Height of the generated video in pixels
        # "width": 1920,                              # Width of the generated video in pixels
        #"fps": 24,                                  # Frames per second for the generated video
        # "motion_bucket_id": 5,                      # Conditions motion amount (higher values = more motion)
        # "noise_aug_strength": 0.5,                  # Amount of noise added, reduces resemblance to original image and increases motion
        # "safety_check": True,                       # Enable safety checks to filter harmful content
        # "seed": 42,                                 # Set seed for reproducibility
        # "num_inference_steps": 50                   # Number of denoising steps for better quality
    })
    
    print(res)
    
    if res.video_response is not None and res.video_response.images:
        generated_video_url = res.video_response.images[0].url
        return generated_video_url
    else:
        return None
    
def upscale_image(prompt: str, image_path: str, model_id: str = 'stabilityai/stable-diffusion-x4-upscaler'):
    """
    This function upscales an image using the Livepeer AI API.

    Args:
        prompt (str): The text prompt to guide the upscaled image generation.
        image_path (str): The path to the image to be upscaled.
        
    Returns:
        str: The URL of the upscaled image.
    """
    s = Livepeer(http_bearer=livepeer_api_key)

    with open(image_path, "rb") as image_file:
        image_content = image_file.read()

    res = s.generate.upscale(request={
        "prompt": prompt,
        "image": {
            "file_name": image_path.split('/')[-1],
            "content": image_content,
        },
        "model_id": model_id,
        # Optional fields
        # "safety_check": True,   # Perform a safety check to filter offensive content
        # "seed": 42,             # Set seed for reproducible results
        # "num_inference_steps": 50,  # Number of denoising steps (higher = better quality but slower)
    })
    
    print(res)
    
    if res.image_response is not None and res.image_response.images:
        upscaled_image_url = res.image_response.images[0].url
        return upscaled_image_url
    else:
        return None
    
def image_to_image(prompt: str, image_path: str, model_id: str = img2img_model_list[0]):
    """
    This function transforms an image based on a text prompt using the Livepeer AI API.

    Args:
        prompt (str): The text prompt to guide image generation.
        image_path (str): The path to the image to be modified.
        
    Returns:
        str: The URL of the transformed image.
    """
    s = Livepeer(http_bearer=livepeer_api_key)

    with open(image_path, "rb") as image_file:
        image_content = image_file.read()

    res = s.generate.image_to_image(request={
        "prompt": prompt,
        "image": {
            "file_name": image_path.split('/')[-1],
            "content": image_content,
        },
        "model_id": model_id,
        # Optional fields
        # "loras": { "latent-consistency/lcm-lora-sdxl": 1.0, "nerijs/pixel-art-xl": 1.2 },  # Low-Rank Adaptation models and weights
        # "strength": 0.75,                             # Degree of transformation (0 to 1)
        # "guidance_scale": 7.5,                        # Pushes model towards text prompt (higher values = closer to text prompt)
        # "image_guidance_scale": 1.0,                  # Degree to which generated image is influenced by the original image
        # "negative_prompt": "bad quality",             # What to exclude from image generation
        # "safety_check": True,                         # Enable safety checks to filter out harmful content
        # "seed": 42,                                   # Set a seed for reproducible results
        # "num_inference_steps": 50,                    # Number of denoising steps for better quality
        # "num_images_per_prompt": 1                    # Number of images generated per prompt
    })
    
    print(res)
    
    if res.image_response is not None and res.image_response.images:
        generated_image_url = res.image_response.images[0].url
        return generated_image_url
    else:
        return None
    
def segment_anything(image_path: str, model_id: str = segment_model_list[0]):
    """
    This function segments an image using the Livepeer AI API.

    Args:
        image_path (str): The path to the image to be segmented.
        
    Returns:
        dict: The segmentation response including masks or other segmentation outputs.
    """
    s = Livepeer(http_bearer=livepeer_api_key)

    with open(image_path, "rb") as image_file:
        image_content = image_file.read()

    res = s.generate.segment_anything2(request={
        "image": {
            "file_name": image_path.split('/')[-1],
            "content": image_content,
        },
        "model_id": model_id,
        # Optional fields
        # "point_coords": [[100, 200], [300, 400]],  # Nx2 array for point prompts in (X, Y) pixel format
        # "point_labels": [1, 0],                   # Labels for points (1 = foreground, 0 = background)
        # "box": [50, 50, 300, 300],                # Box prompt in XYXY format
        # "mask_input": "previous_mask_data",       # Low-res mask from a previous iteration (1xHxW)
        # "multimask_output": True,                 # If true, returns multiple masks for ambiguous prompts
        # "return_logits": True,                    # If true, returns un-thresholded mask logits
        # "normalize_coords": True                  # If true, normalizes point coordinates to [0,1] range
    })
    
    print(res)
    
    if res.masks_response is not None:
        return res.masks_response.masks
    else:
        return None
    
def audio_to_text(audio_path: str, model_id: str = "openai/whisper-large-v3"):
    """
    This function transcribes an audio file using the Livepeer AI API.

    Args:
        audio_path (str): The path to the audio file to be transcribed.        
    Returns:
        str: The transcribed text from the audio file.
    """
    s = Livepeer(http_bearer=livepeer_api_key)

    with open(audio_path, "rb") as audio_file:
        audio_content = audio_file.read()

    res = s.generate.audio_to_text(request={
        "audio": {
            "file_name": audio_path.split('/')[-1],
            "content": audio_content,
        },
        "model_id": model_id,
    })
    
    print(res)
    
    if res.text_response is not None:
        return res.text_response.text
    else:
        return None

video_editor.py

This file contains functions to edit videos, specifically to resize and crop them for platforms like YouTube Shorts.

video_editor.py
import sys
from moviepy.editor import VideoFileClip, concatenate_videoclips, CompositeVideoClip, vfx

def video_editor(video_files: list[str], output_filename: str):
    """
    This function takes a list of video files, resizes and crops them to fit YouTube Shorts dimensions, and concatenates them into a single video file.

    :param video_files: A list of paths to video files.
    :param output_filename: The path to the output video file.
    :return: The path to the output video file.
    """

    clips = []
    for video in video_files:
        try:
            # Load the video file
            clip = VideoFileClip(video)

            # Set target dimensions for YouTube Shorts
            target_width = 1080
            target_height = 1920

            # Determine the scaling factor to cover the target dimensions
            clip_width, clip_height = clip.size
            width_ratio = target_width / clip_width
            height_ratio = target_height / clip_height
            scaling_factor = max(width_ratio, height_ratio)

            # Resize the clip to cover the target area
            clip_resized = clip.resize(height=int(clip.h * scaling_factor))

            # Center crop the clip to target dimensions
            clip_cropped = clip_resized.crop(
                x_center=clip_resized.w / 2,
                y_center=clip_resized.h / 2,
                width=target_width,
                height=target_height
            )

            # Optionally limit the clip duration to 60 seconds
            # clip_cropped = clip_cropped.subclip(0, min(clip_cropped.duration, 60))

            clips.append(clip_cropped)
        except Exception as e:
            print(f"Error processing {video}: {e}")
            continue

    if not clips:
        print("No valid video clips were processed.")
        sys.exit(1)

    # Concatenate clips
    final_clip = concatenate_videoclips(clips, method='compose')

    # Write the final video file
    final_clip.write_videofile(
        output_filename,
        fps=30,
        codec='libx264',
        audio_codec='aac',
        preset='medium',
        threads=4
    )

    return output_filename

youtube_upload.py

Implement the necessary functions for uploading videos to YouTube.

youtube_upload.py
import http.client
import httplib2
import os
import random
import sys
import time
import pickle
import logging
from typing import Optional, List

from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from googleapiclient.http import MediaFileUpload

from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request

# Explicitly tell the underlying HTTP transport library not to retry, since
# we are handling retry logic ourselves.
httplib2.RETRIES = 1

# Maximum number of times to retry before giving up.
MAX_RETRIES = 10

# Always retry when these exceptions are raised.
from http.client import NotConnected, IncompleteRead, ImproperConnectionState, \
    CannotSendRequest, CannotSendHeader, ResponseNotReady, BadStatusLine

RETRIABLE_EXCEPTIONS = (httplib2.HttpLib2Error, IOError, NotConnected,
                        IncompleteRead, ImproperConnectionState,
                        CannotSendRequest, CannotSendHeader,
                        ResponseNotReady, BadStatusLine)

# Always retry when a googleapiclient.errors.HttpError with one of these status codes is raised.
RETRIABLE_STATUS_CODES = [500, 502, 503, 504]

# This OAuth 2.0 access scope allows an application to upload files to the authenticated user's YouTube channel.
YOUTUBE_UPLOAD_SCOPE = "https://www.googleapis.com/auth/youtube.upload"
YOUTUBE_API_SERVICE_NAME = "youtube"
YOUTUBE_API_VERSION = "v3"

VALID_PRIVACY_STATUSES = ("public", "private", "unlisted")


def upload_video(
    file: str,
    title: str,
    description: str,
    category: str = "22",
    keywords: Optional[str] = "",
    privacyStatus: str = "unlisted",
    client_secrets_file: str = "client_secret.json",
    credentials_file: str = "token.pickle"
) -> None:
    """
    Uploads a video file to YouTube.

    Args:
        file (str): Path to the video file to upload.
        title (str): Video title.
        description (str): Video description.
        keywords (str): Comma-separated list of video keywords.
    """
    if not os.path.exists(file):
        raise FileNotFoundError(f"Please specify a valid file. '{file}' does not exist.")

    youtube = get_authenticated_service(client_secrets_file, credentials_file)

    tags = None
    if keywords:
        tags = keywords.split(",")

    body = dict(
        snippet=dict(
            title=title,
            description=description,
            tags=tags,
            categoryId=category
        ),
        status=dict(
            privacyStatus=privacyStatus
        )
    )

    # Call the API's videos.insert method to create and upload the video.
    insert_request = youtube.videos().insert(
        part=",".join(body.keys()),
        body=body,
        media_body=MediaFileUpload(file, chunksize=-1, resumable=True)
    )

    resumable_upload(insert_request)


def get_authenticated_service(client_secrets_file: str, credentials_file: str):
    credentials = None
    # The file token.pickle stores the user's access and refresh tokens.
    if os.path.exists(credentials_file):
        with open(credentials_file, 'rb') as token:
            credentials = pickle.load(token)
    # If there are no valid credentials, let the user log in.
    if not credentials or not credentials.valid:
        if credentials and credentials.expired and credentials.refresh_token:
            credentials.refresh(Request())
        else:
            if not os.path.exists(client_secrets_file):
                raise FileNotFoundError(f"Missing client_secrets.json file: {client_secrets_file}")
            flow = InstalledAppFlow.from_client_secrets_file(
                client_secrets_file, scopes=[YOUTUBE_UPLOAD_SCOPE])
            
            credentials = flow.run_local_server(
                    host='localhost',
                    port=8088,
                    authorization_prompt_message='Please visit this URL: {url}',
                    success_message='The auth flow is complete; you may close this window.',
                    open_browser=True)
        # Save the credentials for the next run.
        with open(credentials_file, 'wb') as token:
            pickle.dump(credentials, token)
    return build(YOUTUBE_API_SERVICE_NAME, YOUTUBE_API_VERSION, credentials=credentials)


def resumable_upload(insert_request):
    response = None
    error = None
    retry = 0
    while response is None:
        try:
            print("Uploading file...")
            status, response = insert_request.next_chunk()
            if response is not None:
                if 'id' in response:
                    print("Video id '%s' was successfully uploaded." %
                          response['id'])
                else:
                    exit("The upload failed with an unexpected response: %s" % response)
        except HttpError as e:
            if e.resp.status in RETRIABLE_STATUS_CODES:
                error = "A retriable HTTP error %d occurred:\n%s" % (e.resp.status,
                                                                     e.content)
            else:
                raise
        except RETRIABLE_EXCEPTIONS as e:
            error = "A retriable error occurred: %s" % e

        if error is not None:
            print(error)
            retry += 1
            if retry > MAX_RETRIES:
                exit("No longer attempting to retry.")

            max_sleep = 2 ** retry
            sleep_seconds = random.random() * max_sleep
            print("Sleeping %f seconds and then retrying..." % sleep_seconds)
            time.sleep(sleep_seconds)

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

Creating the Swarm

1. Import Required Libraries

First, you need to import the necessary libraries and modules to build your AI swarm. This includes HiveSwarm for managing the swarm, tools for saving and processing files, and SDKContext for context management.

from hive_agent import HiveSwarm
from hive_swarm.tools import save_to_file, list_files, read_from_file, download_from_url

from dotenv import load_dotenv
load_dotenv()

from hive_agent.sdk_context import SDKContext

2. Load Environment Variables

Use load_dotenv() to load environment variables from a .env file. This is essential for configuring your environment and accessing necessary credentials.

from dotenv import load_dotenv
load_dotenv()

3. Creating the Configuration File

The configuration file for the Livepeer YouTube Video Generator defines multiple AI agents, each tailored to specific roles in the video production process. It sets essential parameters such as the model, environment, and timeout for tasks, ensuring optimal performance.

Agents include a Script Writer, Scene Writer, Scene Prompt Generator, Scene Image Generator, Scene Image to Video Generator, Video Editor, and YouTube Upload Agent, each equipped with specialized instructions and tools.

config file
[model]
model = "gpt-4o"

[environment]
type = "dev"

[timeout]
llm = 60

[Script_Writer_Agent]
model = "claude-3-5-sonnet-20240620"
environment="dev"
timeout = 30
instruction= """
You are a script writer on a video production team. You should write a script for a video that is engaging and informative. 
The video should be 15 seconds long.
The video should be in the style of a youtube video. 
"""

[Scene_Writer_Agent]
model = "claude-3-5-sonnet-20240620"
environment="dev"
timeout = 60
instruction= """
You are a scene writer agent on a video production team. You should write a scene for a video that is engaging and informative. You should define each scene in terms of the following attributes:
- Scene Number
- Scene Heading
- Scene Description
- Visual Aids
Every scene should be 3 seconds long. You should write 5 scenes with very detailed descriptions.
"""

[Scene_Prompt_Generator_Agent]
model = "claude-3-5-sonnet-20240620"
environment="dev"
timeout = 60
instruction= """
You are an expert scene prompt generator agent on a video production team. Your role is to create highly detailed and effective prompts for the image generation pipeline. When given scene scripts:

1. Analyze each scene thoroughly, focusing on visual elements, mood, and key details.
2. For each scene, create:
   - A concise Scene Prompt summarizing the overall visual concept
   - A detailed Image Prompt optimized for AI image generation:
     • Use vivid, descriptive language
     • Specify important elements like composition, lighting, colors, and style
     • Include relevant artistic references if applicable
   - A Negative Prompt to avoid unwanted elements or styles
3. Tailor your prompts to the specific requirements of AI image generation:
   - Use clear, unambiguous language
   - Prioritize visual elements over abstract concepts
   - Balance detail with room for creative interpretation
4. Format your output clearly:
   - Scene Number
   - Scene Prompt
   - Image Prompt
   - Negative Prompt

Remember, the quality and specificity of your prompts directly impact the final images. Strive for prompts that will result in visually compelling and accurate representations of each scene.
"""

[Scene_Image_Generator_Agent]
model = "claude-3-5-sonnet-20240620"
environment="dev"
timeout = 60
instruction= """
You are a scene image generator agent on a video production team. You should use the prompts from the scene prompt generator agent and generate images for each scene. Collect all urls returned from text_to_image tool in a list and return the list to your manager.
"""
tools=[
    { module = "hive_swarm.tools", name = "text_to_image" }
]

[Scene_Image_to_Video_Generator_Agent]
model = "claude-3-5-sonnet-20240620"
environment="dev"
timeout = 60
instruction= """
You are a scene image to video generator agent on a video production team. You should use the images from the scene image generator agent and generate videos for each scene. Collect all urls returned from image_to_video tool in a list and return the list to your manager.
"""
tools=[
    { module = "hive_swarm.tools", name = "image_to_video" }
]

[Video_Editor_Agent]
model = "gpt-4o"
environment="dev"
timeout = 60
instruction= """
You are a video editor agent on a video production team. You should use the videos from the video generator agent and edit them to create a final video.
"""
tools=[
    { module = "hive_swarm.tools", name = "video_editor" }
]

[Youtube_Upload_Agent]
model = "gpt-4o"
environment="dev"
timeout = 60
instruction= """
You are a youtube upload agent on a video production team. You should use the videos from the video editor agent and upload them to youtube. You should also create a title, description and keywords for the video.
"""
tools=[
    { module = "hive_swarm.tools", name = "upload_video" }
]

4. Create an SDK Context

Define the path to your configuration file (e.g., hive_config.toml) and create an SDKContext instance. This context will manage the configuration settings for the swarm.

config_path = "./hive_swarm/hive_config.toml"
sdk_context = SDKContext(config_path=config_path)

5. Initialize the Livepeer Swarm

Create an instance of HiveSwarm. Set its name, description, and instructions on how it should operate. Provide the functions it will use (e.g., save_to_file, list_files, read_from_file, download_from_url) and pass the previously created SDKContext.

livepeer_swarm = HiveSwarm(
    name="Livepeer Youtube Video Generator",
    description="A swarm of agents that collaborate as members of a Livepeer Youtube Video Generator team.",
    instruction="""You are the manager of a video production team for Livepeer Youtube Video Generator. Your goal is to guide the team in creating an engaging and informative video. Follow these steps:

    1. Coordinate with these agents in order:
       a. Script Writer Agent
       b. Scene Writer Agent
       c. Scene Prompt Generator Agent
       d. Scene Image Generator Agent (max 3 files at a time)
       e. Scene Image to Video Generator Agent (max 3 files at a time)
       f. Video Editor Agent
       g. Youtube Upload Agent

    2. After each interaction with agents, save the output as a separate file in:
       `./hive-agent-data/output/<video_topic>/<agent_type>/`
       Use the save_to_file tool for this purpose.

    3. Scene Image Generator Agent and Scene Image to Video Agent will return URL or list of URLs. You should use download_from_url tool to download images and videos.

    4. Use list_files and read_from_file tools to access and review previous outputs.

    5. Ensure each agent receives the relevant input from the previous step.

    6. Maintain consistency and coherence throughout the video creation process.

    7. Monitor the quality and relevance of each output, requesting revisions if necessary.

    8. Provide clear, concise instructions to each agent, specifying their task and any relevant constraints.

    9. After all steps are complete, review the entire project for cohesiveness and alignment with the original video concept.

    Remember to adapt your management style based on the specific video topic and requirements provided by the user.
    """,
    functions=[save_to_file, list_files, read_from_file, download_from_url],
    sdk_context=sdk_context,
)

Creating main.pyto run the Swarm

1. Import Required Libraries

Start by importing the necessary libraries. You'll use asyncio for managing asynchronous operations and load_dotenv to load your environment variables. Additionally, import the livepeer_swarm instance created earlier.

import asyncio

from dotenv import load_dotenv

from hive_swarm import livepeer_swarm

2. Load Environment Variables

Use load_dotenv() to load the environment variables from a .env file. This step ensures that your application can access the required API keys and other configuration settings.

load_dotenv()

3. Define the Main Asynchronous Function

Create the main asynchronous function, main()which will handle user interactions with the swarm. Inside this function, print a welcome message and instructions for exiting the program.

async def main():
    print(
        "Welcome to the Livepeer Youtube Video Generator Swarm!\nVisit https://swarmzero.ai to learn more.\n\nType 'exit' to quit.\n"
    )

4. Create an Input Loop

Inside the main() function, implement a loop that continuously prompts the user for input. If the user types "exit", the loop breaks and the program ends. Otherwise, the input prompt is sent to the livepeer_swarm.chat() method, which processes the prompt and returns a response.

while True:
        prompt = input("\n\nEnter your prompt: \n\n")
        if prompt.lower() == "exit":
            break
        response = await livepeer_swarm.chat(prompt)
        print(response)

5. Execute the Main Function

Use the asyncio.run() method to execute the main() function. This line ensures that the asynchronous operations within the function are properly managed.

if __name__ == "__main__":
    asyncio.run(main())

Running the Agent

Now that everything is set up, let's run the dApp builder.

Step 1: Run the agent:

(venv) python main.py

Step 2: Input your prompt:

Follow the prompts to enter your video topic. The swarm will then generate and upload a video based on your input.

Create a Short Video on Lavish Cars

Output

PreviousBuild a Web3 dApp with a Swarm of AI AgentsNextTools

Last updated