Music generator (#2103)

## Description

Generates music from a prompt.

Author: mishramonalisha76

cookbook/playground/multimodal_agents.py

@@ -57,6 +57,32 @@
     ),
 )
 
+ml_music_agent = Agent(
+    name="ModelsLab Music Agent",
+    agent_id="ml_music_agent",
+    model=OpenAIChat(id="gpt-4o"),
+    tools=[ModelsLabTools(wait_for_completion=True, file_type=FileType.MP3)],
+    description="You are an AI agent that can generate music using the ModelsLabs API.",
+    instructions=[
+        "When generating music, use the `generate_media` tool with detailed prompts that specify:",
+        "- The genre and style of music (e.g., classical, jazz, electronic)",
+        "- The instruments and sounds to include",
+        "- The tempo, mood and emotional qualities",
+        "- The structure (intro, verses, chorus, bridge, etc.)",
+        "Create rich, descriptive prompts that capture the desired musical elements.",
+        "Focus on generating high-quality, complete instrumental pieces.",
+        "Keep responses simple and only confirm when music is generated successfully.",
+        "Do not include any file names, URLs or technical details in responses.",
+    ],
+    markdown=True,
+    debug_mode=True,
+    add_history_to_messages=True,
+    add_datetime_to_instructions=True,
+    storage=SqliteAgentStorage(
+        table_name="ml_music_agent", db_file=image_agent_storage_file
+    ),
+)
+
 ml_video_agent = Agent(
     name="ModelsLab Video Agent",
     agent_id="ml_video_agent",
@@ -147,6 +173,7 @@
     agents=[
         image_agent,
         ml_gif_agent,
+        ml_music_agent,
         ml_video_agent,
         fal_agent,
         gif_agent,

libs/agno/agno/models/response.py

@@ -38,3 +38,4 @@ class ModelResponse:
 class FileType(str, Enum):
     MP4 = "mp4"
     GIF = "gif"
+    MP3 = "mp3"

libs/agno/agno/tools/models_labs.py

@@ -1,124 +1,161 @@
 import json
 import time
 from os import getenv
-from typing import Optional
+from typing import Any, Dict, Optional
 from uuid import uuid4
 
 from agno.agent import Agent
-from agno.media import ImageArtifact, VideoArtifact
+from agno.media import AudioArtifact, ImageArtifact, VideoArtifact
 from agno.models.response import FileType
 from agno.tools import Toolkit
 from agno.utils.log import logger
 
 try:
     import requests
+    from requests.exceptions import RequestException
 except ImportError:
     raise ImportError("`requests` not installed. Please install using `pip install requests`")
 
+MODELS_LAB_URLS = {
+    "MP4": "https://modelslab.com/api/v6/video/text2video",
+    "MP3": "https://modelslab.com/api/v6/voice/music_gen",
+    "GIF": "https://modelslab.com/api/v6/video/text2video",
+}
+
+MODELS_LAB_FETCH_URLS = {
+    "MP4": "https://modelslab.com/api/v6/video/fetch",
+    "MP3": "https://modelslab.com/api/v6/voice/fetch",
+    "GIF": "https://modelslab.com/api/v6/video/fetch",
+}
+
 
 class ModelsLabTools(Toolkit):
     def __init__(
         self,
         api_key: Optional[str] = None,
-        url: str = "https://modelslab.com/api/v6/video/text2video",
-        fetch_url: str = "https://modelslab.com/api/v6/video/fetch",
-        # Whether to wait for the video to be ready
         wait_for_completion: bool = False,
-        # Time to add to the ETA to account for the time it takes to fetch the video
         add_to_eta: int = 15,
-        # Maximum time to wait for the video to be ready
         max_wait_time: int = 60,
         file_type: FileType = FileType.MP4,
     ):
         super().__init__(name="models_labs")
 
-        self.url = url
-        self.fetch_url = fetch_url
+        file_type_str = file_type.value.upper()
+        self.url = MODELS_LAB_URLS[file_type_str]
+        self.fetch_url = MODELS_LAB_FETCH_URLS[file_type_str]
         self.wait_for_completion = wait_for_completion
         self.add_to_eta = add_to_eta
         self.max_wait_time = max_wait_time
         self.file_type = file_type
         self.api_key = api_key or getenv("MODELS_LAB_API_KEY")
+
         if not self.api_key:
             logger.error("MODELS_LAB_API_KEY not set. Please set the MODELS_LAB_API_KEY environment variable.")
 
         self.register(self.generate_media)
 
-    def generate_media(self, agent: Agent, prompt: str) -> str:
-        """Use this function to generate a video or image given a prompt.
-
-        Args:
-            prompt (str): A text description of the desired video.
+    def _create_payload(self, prompt: str) -> Dict[str, Any]:
+        """Create payload based on file type."""
+        base_payload: Dict[str, Any] = {
+            "key": self.api_key,
+            "prompt": prompt,
+            "webhook": None,
+            "track_id": None,
+        }
+
+        if self.file_type in [FileType.MP4, FileType.GIF]:
+            video_template = {
+                "height": 512,
+                "width": 512,
+                "num_frames": 25,
+                "negative_prompt": "low quality",
+                "model_id": "cogvideox",
+                "instant_response": False,
+                "output_type": self.file_type.value,
+            }
+            base_payload |= video_template  # Use |= instead of update()
+        else:
+            audio_template = {
+                "base64": False,
+                "temp": False,
+            }
+            base_payload |= audio_template  # Use |= instead of update()
+
+        return base_payload
+
+    def _add_media_artifact(self, agent: Agent, media_id: str, media_url: str, eta: Optional[str] = None) -> None:
+        """Add appropriate media artifact based on file type."""
+        if self.file_type == FileType.MP4:
+            agent.add_video(VideoArtifact(id=str(media_id), url=media_url, eta=str(eta)))
+        elif self.file_type == FileType.GIF:
+            agent.add_image(ImageArtifact(id=str(media_id), url=media_url))
+        elif self.file_type == FileType.MP3:
+            agent.add_audio(AudioArtifact(id=str(media_id), url=media_url))
+
+    def _wait_for_media(self, media_id: str, eta: int) -> bool:
+        """Wait for media generation to complete."""
+        time_to_wait = min(eta + self.add_to_eta, self.max_wait_time)
+        logger.info(f"Waiting for {time_to_wait} seconds for {self.file_type.value} to be ready")
+
+        for seconds_waited in range(time_to_wait):
+            try:
+                fetch_response = requests.post(
+                    f"{self.fetch_url}/{media_id}",
+                    json={"key": self.api_key},
+                    headers={"Content-Type": "application/json"},
+                )
+                fetch_result = fetch_response.json()
+
+                if fetch_result.get("status") == "success":
+                    return True
+
+                time.sleep(1)
+
+            except RequestException as e:
+                logger.warning(f"Error during fetch attempt {seconds_waited}: {e}")
+
+        return False
 
-        Returns:
-            str: A message indicating if the video has been generated successfully or an error message.
-        """
+    def generate_media(self, agent: Agent, prompt: str) -> str:
+        """Generate media (video, image, or audio) given a prompt."""
         if not self.api_key:
             return "Please set the MODELS_LAB_API_KEY"
 
         try:
-            payload = json.dumps(
-                {
-                    "key": self.api_key,
-                    "prompt": prompt,
-                    "height": 512,
-                    "width": 512,
-                    "num_frames": 25,
-                    "webhook": None,
-                    "output_type": self.file_type.value,
-                    "track_id": None,
-                    "negative_prompt": "low quality",
-                    "model_id": "cogvideox",
-                    "instant_response": False,
-                }
-            )
-
+            payload = json.dumps(self._create_payload(prompt))
             headers = {"Content-Type": "application/json"}
-            logger.debug(f"Generating video for prompt: {prompt}")
-            response = requests.request("POST", self.url, data=payload, headers=headers)
+
+            logger.debug(f"Generating {self.file_type.value} for prompt: {prompt}")
+            response = requests.post(self.url, data=payload, headers=headers)
             response.raise_for_status()
 
             result = response.json()
+
             if "error" in result:
-                logger.error(f"Failed to generate video: {result['error']}")
+                error_msg = f"Failed to generate {self.file_type.value}: {result['error']}"
+                logger.error(error_msg)
                 return f"Error: {result['error']}"
 
             eta = result["eta"]
             url_links = result["future_links"]
-            logger.info(f"Media will be ready in {eta} seconds")
-            logger.info(f"Media URLs: {url_links}")
+            media_id = str(uuid4())
 
-            video_id = str(uuid4())
-
-            logger.debug(f"Result: {result}")
             for media_url in url_links:
-                if self.file_type == FileType.MP4:
-                    agent.add_video(VideoArtifact(id=str(video_id), url=media_url, eta=str(eta)))
-                elif self.file_type == FileType.GIF:
-                    agent.add_image(ImageArtifact(id=str(video_id), url=media_url))
+                self._add_media_artifact(agent, media_id, media_url, str(eta))
 
             if self.wait_for_completion and isinstance(eta, int):
-                video_ready = False
-                seconds_waited = 0
-                time_to_wait = min(eta + self.add_to_eta, self.max_wait_time)
-                logger.info(f"Waiting for {time_to_wait} seconds for video to be ready")
-                while not video_ready and seconds_waited < time_to_wait:
-                    time.sleep(1)
-                    seconds_waited += 1
-                    # Fetch the video from the ModelsLabs API
-                    fetch_payload = json.dumps({"key": self.api_key})
-                    fetch_headers = {"Content-Type": "application/json"}
-                    logger.debug(f"Fetching video from {self.fetch_url}/{video_id}")
-                    fetch_response = requests.request(
-                        "POST", f"{self.fetch_url}/{video_id}", data=fetch_payload, headers=fetch_headers
-                    )
-                    fetch_result = fetch_response.json()
-                    logger.debug(f"Fetch result: {fetch_result}")
-                    if fetch_result.get("status") == "success":
-                        video_ready = True
-                        break
-
-            return f"Video has been generated successfully and will be ready in {eta} seconds"
+                if self._wait_for_media(media_id, eta):
+                    logger.info("Media generation completed successfully")
+                else:
+                    logger.warning("Media generation timed out")
+
+            return f"{self.file_type.value.capitalize()} has been generated successfully and will be ready in {eta} seconds"
+
+        except RequestException as e:
+            error_msg = f"Network error while generating {self.file_type.value}: {e}"
+            logger.error(error_msg)
+            return f"Error: {error_msg}"
         except Exception as e:
-            logger.error(f"Failed to generate video: {e}")
-            return f"Error: {e}"
+            error_msg = f"Unexpected error while generating {self.file_type.value}: {e}"
+            logger.error(error_msg)
+            return f"Error: {error_msg}"