[feat][bing search api] [feat][code interpreter]

mcs_agent.py

@@ -0,0 +1,114 @@
+import os
+import time
+
+from swarm_models import OpenAIChat
+from swarms import Agent
+from dotenv import load_dotenv
+
+from swarms_tools.social_media.twitter_tool import TwitterTool
+
+load_dotenv()
+
+model_name = "gpt-4o"
+
+model = OpenAIChat(
+    model_name=model_name,
+    max_tokens=3000,
+    openai_api_key=os.getenv("OPENAI_API_KEY"),
+)
+
+
+treatment_agent = Agent(
+    agent_name="Medical Treatment Expert",
+    system_prompt="""
+    You are a highly experienced medical doctor with expertise in evidence-based treatments, pharmaceuticals, and clinical therapeutics. Your primary focus is to provide comprehensive, safe, and effective treatment plans for patients. You do not offer official diagnoses or medical advice; rather, you provide educational information. Always advise users to consult a healthcare professional for personalized guidance.
+
+    ### Primary Responsibilities:
+    1. **Focus on Treatments**: Recommend the most appropriate treatments (medications, therapies, procedures) based on standard clinical guidelines and the latest medical evidence.
+    2. **Detail Therapeutic Approaches**: Explain how each treatment works, typical usage, dosage ranges (if relevant), expected outcomes, and any potential side effects or contraindications.
+    3. **Integrate Holistic Care**: When appropriate, discuss lifestyle modifications, rehabilitation, adjunct therapies, or preventive measures that support overall well-being.
+    4. **Provide Clear Explanations**: Use lay-friendly language whenever possible, but maintain clinical accuracy and detail.
+
+    ### Formatting and Clarity:
+    - Present treatment recommendations in a structured, easy-to-follow manner.
+    - Where applicable, cite brief references to reputable guidelines or organizations (e.g., WHO, CDC, NICE, etc.).
+    - Include any necessary cautionary notes about drug interactions, special populations (e.g., pregnant women, children, elderly), and the importance of personalized medical care.
+
+    ### Important Disclaimer:
+    - Your responses are for general educational purposes only and do not replace professional medical consultation.
+    - Always advise the user to consult a qualified healthcare provider for personalized treatment.
+    """,
+    llm=model,
+    max_loops=1,
+    dynamic_temperature_enabled=True,
+)
+
+
+# Define your options with the necessary credentials
+options = {
+    "id": "mcsswarm",
+    "name": "mcsswarm",
+    "description": "An example Twitter Plugin for testing.",
+    "credentials": {
+        "apiKey": os.getenv("TWITTER_API_KEY"),
+        "apiSecretKey": os.getenv("TWITTER_API_SECRET_KEY"),
+        "accessToken": os.getenv("TWITTER_ACCESS_TOKEN"),
+        "accessTokenSecret": os.getenv("TWITTER_ACCESS_TOKEN_SECRET"),
+    },
+}
+
+# Initialize the TwitterTool with your options
+twitter_plugin = TwitterTool(options)
+
+# Assuming `twitter_plugin` and `medical_coder` are already initialized
+post_tweet = twitter_plugin.get_function("post_tweet")
+
+# Set to track posted tweets and avoid duplicates
+posted_tweets = set()
+
+
+def post_unique_tweet():
+    """
+    Generate and post a unique tweet. Skip duplicates.
+    """
+    tweet_prompt = (
+        "Craft a concise and engaging tweet about a specific disease and various treatment options for that disease using traditional medicine without invasive measures."
+        "Be very direct and to the point, but also engaging and interesting. Aim to provide maximum value "
+        "Focus on one disease and its corresponding treatment per tweet."
+        "Keep it informative, yet brief and captivating."
+    )
+
+    # Generate a new tweet text
+    tweet_text = treatment_agent.run(tweet_prompt)
+
+    # Check for duplicates
+    if tweet_text in posted_tweets:
+        print("Duplicate tweet detected. Skipping...")
+        return
+
+    # Post the tweet
+    try:
+        post_tweet(tweet_text)
+        print(f"Posted tweet: {tweet_text}")
+        # Add the tweet to the set of posted tweets
+        posted_tweets.add(tweet_text)
+    except Exception as e:
+        print(f"Error posting tweet: {e}")
+
+
+# Loop to post tweets every 10 seconds
+def start_tweet_loop(interval: int = 10):
+    """
+    Continuously post tweets every `interval` seconds.
+
+    Args:
+        interval (int): Time in seconds between tweets.
+    """
+    print("Starting tweet loop...")
+    while True:
+        post_unique_tweet()
+        time.sleep(interval)
+
+
+# Start the loop
+start_tweet_loop(200)

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
 name = "swarms-tools"
-version = "0.1.5"
+version = "0.1.6"
 description = "Paper - Pytorch"
 license = "MIT"
 authors = ["Kye Gomez <[email protected]>"]

swarms_tools/code/__init__.py

@@ -0,0 +1,3 @@
+from swarms_tools.code.code_executor import CodeExecutor
+
+__all__ = ["CodeExecutor"]

swarms_tools/code/code_executor.py

@@ -0,0 +1,128 @@
+import os
+import subprocess
+from loguru import logger
+
+
+class CodeExecutor:
+    """
+    A class to execute Python code and return the output as a string.
+
+    The class also logs the input and output using loguru and stores the outputs
+    in a folder called 'artifacts'.
+
+    Methods:
+        execute(code: str) -> str:
+            Executes the given Python code and returns the output.
+    """
+
+    def __init__(
+        self,
+        max_output_length: int = 1000,
+        artifacts_directory: str = "artifacts",
+        language: str = "python3",
+    ) -> None:
+        """
+        Initializes the CodeExecutor class and sets up the logging.
+        """
+        self.max_output_length = max_output_length
+        self.artifacts_dir = artifacts_directory
+        self.language = language
+
+        os.makedirs(self.artifacts_dir, exist_ok=True)
+        self.setup_logging()
+
+    def setup_logging(self) -> None:
+        """
+        Sets up the loguru logger with colorful output.
+        """
+        logger.add(
+            os.path.join(self.artifacts_dir, "code_execution.log"),
+            format="{time} {level} {message}",
+            level="DEBUG",
+        )
+        logger.info(
+            "Logger initialized and artifacts directory set up."
+        )
+
+    def format_code(self, code: str) -> str:
+        """
+        Formats the given Python code using black.
+
+        Args:
+            code (str): The Python code to format.
+
+        Returns:
+            str: The formatted Python code.
+
+        Raises:
+            ValueError: If the code cannot be formatted.
+        """
+        try:
+            import black
+
+            formatted_code = black.format_str(
+                code, mode=black.FileMode()
+            )
+            return formatted_code
+        except Exception as e:
+            logger.error(f"Error formatting code: {e}")
+            raise ValueError(f"Error formatting code: {e}") from e
+
+    def execute(self, code: str) -> str:
+        """
+        Executes the given Python code and returns the output.
+
+        Args:
+            code (str): The Python code to execute.
+
+        Returns:
+            str: The output of the executed code.
+
+        Raises:
+            RuntimeError: If there is an error during the execution of the code.
+        """
+        try:
+            formatted_code = self.format_code(code)
+            logger.info(f"Executing code:\n{formatted_code}")
+            completed_process = subprocess.run(
+                [self.language, "-c", formatted_code],
+                capture_output=True,
+                text=True,
+                check=True,
+            )
+            output = completed_process.stdout
+            logger.info(f"Code output:\n{output}")
+            return output
+        except subprocess.CalledProcessError as e:
+            logger.error(f"Error executing code: {e.stderr}")
+            raise RuntimeError(
+                f"Error executing code: {e.stderr}"
+            ) from e
+
+
+# # Example usage:
+# if __name__ == "__main__":
+#     executor = CodeExecutor(max_output_length=300)
+#     code = """
+# import requests
+# from typing import Any
+
+# def fetch_financial_news(api_key: str, query: str, num_articles: int) -> Any:
+#     try:
+#         url = f"https://newsapi.org/v2/everything?q={query}&apiKey={api_key}"
+#         response = requests.get(url)
+#         response.raise_for_status()
+#         return response.json()
+#     except requests.RequestException as e:
+#         print(f"Request Error: {e}")
+#         raise
+#     except ValueError as e:
+#         print(f"Value Error: {e}")
+#         raise
+
+# api_key = ""
+# result = fetch_financial_news(api_key, query="Nvidia news", num_articles=5)
+# print(result)
+#     """
+#     result = executor.execute(code)
+#     print(result)