From 34177fda3a4e4f43961c5f73137e06e0b5a57dc0 Mon Sep 17 00:00:00 2001 From: Kye Gomez Date: Sat, 1 Feb 2025 19:44:04 -0500 Subject: [PATCH] [feat][bing search api] [feat][code interpreter] --- examples/mcs_agent.py | 140 ----------------------------- mcs_agent.py | 114 +++++++++++++++++++++++ pyproject.toml | 2 +- swarms_tools/code/__init__.py | 3 + swarms_tools/code/code_executor.py | 128 ++++++++++++++++++++++++++ swarms_tools/search/__init__.py | 3 + swarms_tools/search/bing.py | 85 ++++++++++++++++++ swarms_tools/search/exa_tool.py | 4 +- 8 files changed, 336 insertions(+), 143 deletions(-) delete mode 100644 examples/mcs_agent.py create mode 100644 mcs_agent.py create mode 100644 swarms_tools/code/__init__.py create mode 100644 swarms_tools/code/code_executor.py create mode 100644 swarms_tools/search/bing.py diff --git a/examples/mcs_agent.py b/examples/mcs_agent.py deleted file mode 100644 index 1e3fd76..0000000 --- a/examples/mcs_agent.py +++ /dev/null @@ -1,140 +0,0 @@ -import os -from time 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"), -) - - -medical_coder = Agent( - agent_name="Medical Coder", - system_prompt=""" - You are a highly experienced and certified medical coder with extensive knowledge of ICD-10 coding guidelines, clinical documentation standards, and compliance regulations. Your responsibility is to ensure precise, compliant, and well-documented coding for all clinical cases. - - ### Primary Responsibilities: - 1. **Review Clinical Documentation**: Analyze all available clinical records, including specialist inputs, physician notes, lab results, imaging reports, and discharge summaries. - 2. **Assign Accurate ICD-10 Codes**: Identify and assign appropriate codes for primary diagnoses, secondary conditions, symptoms, and complications. - 3. **Ensure Coding Compliance**: Follow the latest ICD-10-CM/PCS coding guidelines, payer-specific requirements, and organizational policies. - 4. **Document Code Justification**: Provide clear, evidence-based rationale for each assigned code. - - ### Detailed Coding Process: - - **Review Specialist Inputs**: Examine all relevant documentation to capture the full scope of the patient's condition and care provided. - - **Identify Diagnoses**: Determine the primary and secondary diagnoses, as well as any symptoms or complications, based on the documentation. - - **Assign ICD-10 Codes**: Select the most accurate and specific ICD-10 codes for each identified diagnosis or condition. - - **Document Supporting Evidence**: Record the documentation source (e.g., lab report, imaging, or physician note) for each code to justify its assignment. - - **Address Queries**: Note and flag any inconsistencies, missing information, or areas requiring clarification from providers. - - ### Output Requirements: - Your response must be clear, structured, and compliant with professional standards. Use the following format: - - 1. **Primary Diagnosis Codes**: - - **ICD-10 Code**: [e.g., E11.9] - - **Description**: [e.g., Type 2 diabetes mellitus without complications] - - **Supporting Documentation**: [e.g., Physician's note dated MM/DD/YYYY] - - 2. **Secondary Diagnosis Codes**: - - **ICD-10 Code**: [Code] - - **Description**: [Description] - - **Order of Clinical Significance**: [Rank or priority] - - 3. **Symptom Codes**: - - **ICD-10 Code**: [Code] - - **Description**: [Description] - - 4. **Complication Codes**: - - **ICD-10 Code**: [Code] - - **Description**: [Description] - - **Relevant Documentation**: [Source of information] - - 5. **Coding Notes**: - - Observations, clarifications, or any potential issues requiring provider input. - - ### Additional Guidelines: - - Always prioritize specificity and compliance when assigning codes. - - For ambiguous cases, provide a brief note with reasoning and flag for clarification. - - Ensure the output format is clean, consistent, and ready for professional use. - """, - 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 = ( - "Share an intriguing, lesser-known fact about a medical disease, and include an innovative, fun, or surprising way to manage or cure it! " - "Make the response playful, engaging, and inspiring—something that makes people smile while learning. No markdown, just plain text!" - ) - - # Generate a new tweet text - tweet_text = medical_coder.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=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(10) diff --git a/mcs_agent.py b/mcs_agent.py new file mode 100644 index 0000000..fe99fd1 --- /dev/null +++ b/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) diff --git a/pyproject.toml b/pyproject.toml index bf98ddb..3caafe0 100644 --- a/pyproject.toml +++ b/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 "] diff --git a/swarms_tools/code/__init__.py b/swarms_tools/code/__init__.py new file mode 100644 index 0000000..3b78808 --- /dev/null +++ b/swarms_tools/code/__init__.py @@ -0,0 +1,3 @@ +from swarms_tools.code.code_executor import CodeExecutor + +__all__ = ["CodeExecutor"] diff --git a/swarms_tools/code/code_executor.py b/swarms_tools/code/code_executor.py new file mode 100644 index 0000000..3e56044 --- /dev/null +++ b/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) diff --git a/swarms_tools/search/__init__.py b/swarms_tools/search/__init__.py index e69de29..6995cf3 100644 --- a/swarms_tools/search/__init__.py +++ b/swarms_tools/search/__init__.py @@ -0,0 +1,3 @@ +from swarms_tools.search.bing import fetch_web_articles_bing_api + +__all__ = ["fetch_web_articles_bing_api"] diff --git a/swarms_tools/search/bing.py b/swarms_tools/search/bing.py new file mode 100644 index 0000000..ce69a10 --- /dev/null +++ b/swarms_tools/search/bing.py @@ -0,0 +1,85 @@ +import os +import requests +from typing import List, Dict +from swarms_tools.utils.formatted_string import ( + format_object_to_string, +) + + +def check_bing_api_key(): + try: + return os.getenv("BING_API_KEY") + except Exception as error: + print(f"Error {error}") + raise None + + +def parse_and_merge_logs(logs: List[Dict[str, str]]) -> str: + """ + Parses logs and merges them into a single string for input to an LLM. + + Parameters: + logs (List[Dict[str, str]]): A list of dictionaries where each dictionary represents a log entry. + + Returns: + str: A single string containing all log entries concatenated. + """ + + merged_logs = "" + for log in logs: + log_entries = [ + f"{key}: {value}" for key, value in log.items() + ] + log_string = "\n".join(log_entries) + merged_logs += log_string + "\n\n" + + return merged_logs.strip() + + +def fetch_web_articles_bing_api( + query: str = None, +) -> str: + """ + Fetches four articles from Bing Web Search API based on the given query. + + Parameters: + query (str): The search query to retrieve articles. + subscription_key (str): The Bing Search API subscription key. + + Returns: + List[Dict[str, str]]: A list of dictionaries containing article details. + """ + subscription_key = check_bing_api_key() + + url = "https://api.bing.microsoft.com/v7.0/search" + headers = {"Ocp-Apim-Subscription-Key": subscription_key} + params = {"q": query, "count": 4, "mkt": "en-US"} + + response = requests.get(url, headers=headers, params=params) + response.raise_for_status() + search_results = response.json() + + articles = [] + for i, result in enumerate( + search_results.get("webPages", {}).get("value", []) + ): + article_info = { + "query": query, + "url": result.get("url"), + "title": result.get("name"), + "publishedDate": result.get("dateLastCrawled"), + "author": ( + result.get("provider")[0]["name"] + if result.get("provider") + else "Unknown" + ), + "id": str(i + 1), # Generating a simple unique ID + } + articles.append(article_info) + + articles = parse_and_merge_logs(articles) + return format_object_to_string(articles) + + +# out = fetch_web_articles_bing_api("swarms ai github") +# print(out) diff --git a/swarms_tools/search/exa_tool.py b/swarms_tools/search/exa_tool.py index 72bdf86..e67d681 100644 --- a/swarms_tools/search/exa_tool.py +++ b/swarms_tools/search/exa_tool.py @@ -33,10 +33,10 @@ def search_exa_ai( response.raise_for_status() # Raises an HTTPError if the response status code is 4XX/5XX return response.json() except requests.exceptions.HTTPError as http_err: - print(f'HTTP error occurred: {http_err}') + print(f"HTTP error occurred: {http_err}") return "Failed to retrieve results due to an HTTP error." except Exception as err: - print(f'Other error occurred: {err}') + print(f"Other error occurred: {err}") return "Failed to retrieve results due to an error."