From 283a3f0e0c65f1fc142b6bb3d6043aeaa797c2f7 Mon Sep 17 00:00:00 2001 From: Kye Gomez Date: Thu, 23 Jan 2025 20:37:11 -0500 Subject: [PATCH] [TWITTER AGENTS] --- docs/mkdocs.yml | 1 + docs/swarms_tools/twitter.md | 353 ++++++++++++++++++++++++++++++++ litellm_tool_example.py | 51 +++++ swarms/tools/base_tool.py | 60 ++---- swarms/utils/litellm_wrapper.py | 5 +- tests/tools/test_base_tool.py | 126 ++++++++++++ 6 files changed, 556 insertions(+), 40 deletions(-) create mode 100644 docs/swarms_tools/twitter.md create mode 100644 litellm_tool_example.py create mode 100644 tests/tools/test_base_tool.py diff --git a/docs/mkdocs.yml b/docs/mkdocs.yml index ca55140bd..889060057 100644 --- a/docs/mkdocs.yml +++ b/docs/mkdocs.yml @@ -213,6 +213,7 @@ nav: - Agent with HTX + CoinGecko: "swarms/examples/swarms_tools_htx.md" - Agent with HTX + CoinGecko Function Calling: "swarms/examples/swarms_tools_htx_gecko.md" - Agent with Yahoo Finance: "swarms/examples/yahoo_finance.md" + - Twitter Agents: "swarms/examples/twitter.md" - Meme Agents: - Bob The Builder: "swarms/examples/bob_the_builder.md" - Meme Agent Builder: "swarms/examples/meme_agents.md" diff --git a/docs/swarms_tools/twitter.md b/docs/swarms_tools/twitter.md new file mode 100644 index 000000000..c6d6b1d68 --- /dev/null +++ b/docs/swarms_tools/twitter.md @@ -0,0 +1,353 @@ +# TwitterTool Documentation + +## Overview + +The TwitterTool is a powerful Python-based interface for interacting with Twitter's API, designed specifically for integration with autonomous agents and AI systems. It provides a streamlined way to perform common Twitter operations while maintaining proper error handling and logging capabilities. + +## Installation + +Before using the TwitterTool, ensure you have the required dependencies installed: + +```bash +pip install tweepy swarms-tools +``` + +## Basic Configuration + +The TwitterTool requires Twitter API credentials for authentication. Here's how to set up the basic configuration: + +```python +from swarms_tools.social_media.twitter_tool import TwitterTool + +import os + +options = { + "id": "your_unique_id", + "name": "your_tool_name", + "description": "Your tool description", + "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") + } +} + +twitter_tool = TwitterTool(options) +``` + +For security, it's recommended to use environment variables for credentials: + +```python +import os +from dotenv import load_dotenv + +load_dotenv() + +options = { + "id": "twitter_bot", + "name": "Twitter Bot", + "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") + } +} +``` + +## Core Functionality + +The TwitterTool provides five main functions: + +1. **Posting Tweets**: Create new tweets +2. **Replying to Tweets**: Respond to existing tweets +3. **Quoting Tweets**: Share tweets with additional commentary +4. **Liking Tweets**: Engage with other users' content +5. **Fetching Metrics**: Retrieve account statistics + +### Basic Usage Examples + +```python +# Get a specific function +post_tweet = twitter_tool.get_function('post_tweet') +reply_tweet = twitter_tool.get_function('reply_tweet') +quote_tweet = twitter_tool.get_function('quote_tweet') +like_tweet = twitter_tool.get_function('like_tweet') +get_metrics = twitter_tool.get_function('get_metrics') + +# Post a tweet +post_tweet("Hello, Twitter!") + +# Reply to a tweet +reply_tweet(tweet_id=123456789, reply="Great point!") + +# Quote a tweet +quote_tweet(tweet_id=123456789, quote="Interesting perspective!") + +# Like a tweet +like_tweet(tweet_id=123456789) + +# Get account metrics +metrics = get_metrics() +print(f"Followers: {metrics['followers']}") +``` + +## Integration with Agents + +The TwitterTool can be particularly powerful when integrated with AI agents. Here are several examples of agent integrations: + +### 1. Medical Information Bot + +This example shows how to create a medical information bot that shares health facts: + +```python +from swarms import Agent +from swarms_models import OpenAIChat + +# Initialize the AI model +model = OpenAIChat( + model_name="gpt-4", + max_tokens=3000, + openai_api_key=os.getenv("OPENAI_API_KEY") +) + +# Create a medical expert agent +medical_expert = Agent( + agent_name="Medical Expert", + system_prompt=""" + You are a medical expert sharing evidence-based health information. + Your tweets should be: + - Accurate and scientifically sound + - Easy to understand + - Engaging and relevant + - Within Twitter's character limit + """, + llm=model +) + +# Function to generate and post medical tweets +def post_medical_fact(): + prompt = "Share an interesting medical fact that would be helpful for the general public." + tweet_text = medical_expert.run(prompt) + post_tweet = twitter_tool.get_function('post_tweet') + post_tweet(tweet_text) +``` + +### 2. News Summarization Bot + +This example demonstrates how to create a bot that summarizes news articles: + +```python +# Create a news summarization agent +news_agent = Agent( + agent_name="News Summarizer", + system_prompt=""" + You are a skilled news editor who excels at creating concise, + accurate summaries of news articles while maintaining the key points. + Your summaries should be: + - Factual and unbiased + - Clear and concise + - Properly attributed + - Under 280 characters + """, + llm=model +) + +def summarize_and_tweet(article_url): + # Generate summary + prompt = f"Summarize this news article in a tweet-length format: {article_url}" + summary = news_agent.run(prompt) + + # Post the summary + post_tweet = twitter_tool.get_function('post_tweet') + post_tweet(f"{summary} Source: {article_url}") +``` + +### 3. Interactive Q&A Bot + +This example shows how to create a bot that responds to user questions: + +```python +class TwitterQABot: + def __init__(self): + self.twitter_tool = TwitterTool(options) + self.qa_agent = Agent( + agent_name="Q&A Expert", + system_prompt=""" + You are an expert at providing clear, concise answers to questions. + Your responses should be: + - Accurate and informative + - Conversational in tone + - Limited to 280 characters + - Include relevant hashtags when appropriate + """, + llm=model + ) + + def handle_question(self, tweet_id, question): + # Generate response + response = self.qa_agent.run(f"Answer this question: {question}") + + # Reply to the tweet + reply_tweet = self.twitter_tool.get_function('reply_tweet') + reply_tweet(tweet_id=tweet_id, reply=response) + +qa_bot = TwitterQABot() +qa_bot.handle_question(123456789, "What causes climate change?") +``` + +## Best Practices + +When using the TwitterTool, especially with agents, consider these best practices: + +1. **Rate Limiting**: Implement delays between tweets to comply with Twitter's rate limits: +```python +import time + +def post_with_rate_limit(tweet_text, delay=60): + post_tweet = twitter_tool.get_function('post_tweet') + post_tweet(tweet_text) + time.sleep(delay) # Wait 60 seconds between tweets +``` + +2. **Content Tracking**: Maintain a record of posted content to avoid duplicates: +```python +posted_tweets = set() + +def post_unique_tweet(tweet_text): + if tweet_text not in posted_tweets: + post_tweet = twitter_tool.get_function('post_tweet') + post_tweet(tweet_text) + posted_tweets.add(tweet_text) +``` + +3. **Error Handling**: Implement robust error handling for API failures: +```python +def safe_tweet(tweet_text): + try: + post_tweet = twitter_tool.get_function('post_tweet') + post_tweet(tweet_text) + except Exception as e: + logging.error(f"Failed to post tweet: {e}") + # Implement retry logic or fallback behavior +``` + +4. **Content Validation**: Validate content before posting: +```python +def validate_and_post(tweet_text): + if len(tweet_text) > 280: + tweet_text = tweet_text[:277] + "..." + + # Check for prohibited content + prohibited_terms = ["spam", "inappropriate"] + if any(term in tweet_text.lower() for term in prohibited_terms): + return False + + post_tweet = twitter_tool.get_function('post_tweet') + post_tweet(tweet_text) + return True +``` + +## Advanced Features + +### Scheduled Posting + +Implement scheduled posting using Python's built-in scheduling capabilities: + +```python +from datetime import datetime +import schedule + +def scheduled_tweet_job(): + twitter_tool = TwitterTool(options) + post_tweet = twitter_tool.get_function('post_tweet') + + # Generate content using an agent + content = medical_expert.run("Generate a health tip of the day") + post_tweet(content) + +# Schedule tweets for specific times +schedule.every().day.at("10:00").do(scheduled_tweet_job) +schedule.every().day.at("15:00").do(scheduled_tweet_job) + +while True: + schedule.run_pending() + time.sleep(60) +``` + +### Analytics Integration + +Track the performance of your tweets: + +```python +class TweetAnalytics: + def __init__(self, twitter_tool): + self.twitter_tool = twitter_tool + self.metrics_history = [] + + def record_metrics(self): + get_metrics = self.twitter_tool.get_function('get_metrics') + current_metrics = get_metrics() + self.metrics_history.append({ + 'timestamp': datetime.now(), + 'metrics': current_metrics + }) + + def get_growth_rate(self): + if len(self.metrics_history) < 2: + return None + + latest = self.metrics_history[-1]['metrics'] + previous = self.metrics_history[-2]['metrics'] + + return { + 'followers_growth': latest['followers'] - previous['followers'], + 'tweets_growth': latest['tweets'] - previous['tweets'] + } +``` + +## Troubleshooting + +Common issues and their solutions: + +1. **Authentication Errors**: Double-check your API credentials and ensure they're properly loaded from environment variables. + +2. **Rate Limiting**: If you encounter rate limit errors, implement exponential backoff: +```python +import time +from random import uniform + +def exponential_backoff(attempt): + wait_time = min(300, (2 ** attempt) + uniform(0, 1)) + time.sleep(wait_time) + +def retry_post(tweet_text, max_attempts=5): + for attempt in range(max_attempts): + try: + post_tweet = twitter_tool.get_function('post_tweet') + post_tweet(tweet_text) + return True + except Exception as e: + if attempt < max_attempts - 1: + exponential_backoff(attempt) + else: + raise e +``` + +3. **Content Length Issues**: Implement automatic content truncation: +```python +def truncate_tweet(text, max_length=280): + if len(text) <= max_length: + return text + + # Try to break at last space before limit + truncated = text[:max_length-3] + last_space = truncated.rfind(' ') + if last_space > 0: + truncated = truncated[:last_space] + + return truncated + "..." +``` + +Remember to regularly check Twitter's API documentation for any updates or changes to rate limits and functionality. The TwitterTool is designed to be extensible, so you can add new features as needed for your specific use case. \ No newline at end of file diff --git a/litellm_tool_example.py b/litellm_tool_example.py new file mode 100644 index 000000000..e79b5655a --- /dev/null +++ b/litellm_tool_example.py @@ -0,0 +1,51 @@ +from swarms.tools.base_tool import BaseTool + +import requests +from swarms.utils.litellm_wrapper import LiteLLM + + +def get_stock_data(symbol: str) -> str: + """ + Fetches stock data from Yahoo Finance for a given stock symbol. + + Args: + symbol (str): The stock symbol to fetch data for (e.g., 'AAPL' for Apple Inc.). + + Returns: + Dict[str, Any]: A dictionary containing stock data, including price, volume, and other relevant information. + + Raises: + ValueError: If the stock symbol is invalid or data cannot be retrieved. + """ + url = f"https://query1.finance.yahoo.com/v7/finance/quote?symbols={symbol}" + response = requests.get(url) + + if response.status_code != 200: + raise ValueError(f"Error fetching data for symbol: {symbol}") + + data = response.json() + if ( + "quoteResponse" not in data + or not data["quoteResponse"]["result"] + ): + raise ValueError(f"No data found for symbol: {symbol}") + + return str(data["quoteResponse"]["result"][0]) + + +tool_schema = BaseTool( + tools=[get_stock_data] +).convert_tool_into_openai_schema() + +tool_schema = tool_schema["functions"][0] + +llm = LiteLLM( + model_name="gpt-4o", +) + +print( + llm.run( + "What is the stock data for Apple Inc. (AAPL)?", + tools=[tool_schema], + ) +) diff --git a/swarms/tools/base_tool.py b/swarms/tools/base_tool.py index 04319db81..ae47a1a17 100644 --- a/swarms/tools/base_tool.py +++ b/swarms/tools/base_tool.py @@ -3,7 +3,6 @@ from pydantic import BaseModel, Field -from swarms.tools.func_calling_executor import openai_tool_executor from swarms.tools.func_to_str import function_to_str, functions_to_str from swarms.tools.function_util import process_tool_docs from swarms.tools.py_func_to_openai_func_str import ( @@ -15,6 +14,7 @@ multi_base_model_to_openai_function, ) from swarms.utils.loguru_logger import initialize_logger +from swarms.tools.tool_parse_exec import parse_and_execute_json logger = initialize_logger(log_folder="base_tool") @@ -178,16 +178,14 @@ def get_docs_from_callable(self, item): def execute_tool( self, + response: str, *args: Any, **kwargs: Any, ) -> Callable: try: - return openai_tool_executor( - self.list_of_dicts, - self.function_map, - self.verbose, - *args, - **kwargs, + return parse_and_execute_json( + self.tools, + response, ) except Exception as e: logger.error(f"An error occurred in execute_tool: {e}") @@ -253,6 +251,7 @@ def dynamic_run(self, input: Any) -> str: def execute_tool_by_name( self, tool_name: str, + response: str, ) -> Any: """ Search for a tool by name and execute it. @@ -268,31 +267,16 @@ def execute_tool_by_name( ValueError: If the tool with the specified name is not found. TypeError: If the tool name is not mapped to a function in the function map. """ - # Search for the tool by name - tool = next( - ( - tool - for tool in self.tools - if tool.get("name") == tool_name - ), - None, - ) - - # If the tool is not found, raise an error - if tool is None: - raise ValueError(f"Tool '{tool_name}' not found") - - # Get the function associated with the tool + # Step 1. find the function in the function map func = self.function_map.get(tool_name) - # If the function is not found, raise an error - if func is None: - raise TypeError( - f"Tool '{tool_name}' is not mapped to a function" - ) + execution = parse_and_execute_json( + functions=[func], + json_string=response, + verbose=self.verbose, + ) - # Execute the tool - return func(**tool.get("parameters", {})) + return execution def execute_tool_from_text(self, text: str) -> Any: """ @@ -415,16 +399,14 @@ def convert_tool_into_openai_schema(self): ) # Combine all tool schemas into a single schema - if tool_schemas: - combined_schema = { - "type": "function", - "functions": [ - schema["function"] for schema in tool_schemas - ], - } - return json.dumps(combined_schema, indent=4) - - return None + combined_schema = { + "type": "function", + "functions": [ + schema["function"] for schema in tool_schemas + ], + } + + return combined_schema def check_func_if_have_docs(self, func: callable): if func.__doc__ is not None: diff --git a/swarms/utils/litellm_wrapper.py b/swarms/utils/litellm_wrapper.py index 2dbdc97ee..9252966fc 100644 --- a/swarms/utils/litellm_wrapper.py +++ b/swarms/utils/litellm_wrapper.py @@ -25,6 +25,8 @@ def __init__( temperature: float = 0.5, max_tokens: int = 4000, ssl_verify: bool = False, + *args, + **kwargs, ): """ Initialize the LiteLLM with the given parameters. @@ -64,7 +66,7 @@ def _prepare_messages(self, task: str) -> list: return messages - def run(self, task: str, *args, **kwargs): + def run(self, task: str, tools: any = [], *args, **kwargs): """ Run the LLM model for the given task. @@ -86,6 +88,7 @@ def run(self, task: str, *args, **kwargs): stream=self.stream, temperature=self.temperature, max_tokens=self.max_tokens, + tools=tools, *args, **kwargs, ) diff --git a/tests/tools/test_base_tool.py b/tests/tools/test_base_tool.py new file mode 100644 index 000000000..1b6cdeeb4 --- /dev/null +++ b/tests/tools/test_base_tool.py @@ -0,0 +1,126 @@ +from pydantic import BaseModel +from typing import Optional +import json + +from swarms.tools.base_tool import BaseTool + + +class TestModel(BaseModel): + name: str + age: int + email: Optional[str] = None + + +def sample_function(x: int, y: int) -> int: + """Test function for addition.""" + return x + y + + +def test_func_to_dict(): + print("Testing func_to_dict") + tool = BaseTool() + + result = tool.func_to_dict( + function=sample_function, + name="sample_function", + description="Test function", + ) + + assert result["type"] == "function" + assert result["function"]["name"] == "sample_function" + assert "parameters" in result["function"] + print("func_to_dict test passed") + + +def test_base_model_to_dict(): + print("Testing base_model_to_dict") + tool = BaseTool() + + result = tool.base_model_to_dict(TestModel) + + assert "type" in result + assert "properties" in result["properties"] + assert "name" in result["properties"]["properties"] + print("base_model_to_dict test passed") + + +def test_detect_tool_input_type(): + print("Testing detect_tool_input_type") + tool = BaseTool() + + model = TestModel(name="Test", age=25) + assert tool.detect_tool_input_type(model) == "Pydantic" + + dict_input = {"key": "value"} + assert tool.detect_tool_input_type(dict_input) == "Dictionary" + + assert tool.detect_tool_input_type(sample_function) == "Function" + print("detect_tool_input_type test passed") + + +def test_execute_tool_by_name(): + print("Testing execute_tool_by_name") + tool = BaseTool( + function_map={"sample_function": sample_function}, + verbose=True, + ) + + response = json.dumps( + {"name": "sample_function", "parameters": {"x": 1, "y": 2}} + ) + + result = tool.execute_tool_by_name("sample_function", response) + assert result == 3 + print("execute_tool_by_name test passed") + + +def test_check_str_for_functions_valid(): + print("Testing check_str_for_functions_valid") + tool = BaseTool(function_map={"test_func": lambda x: x}) + + valid_json = json.dumps( + {"type": "function", "function": {"name": "test_func"}} + ) + + assert tool.check_str_for_functions_valid(valid_json) is True + + invalid_json = json.dumps({"type": "invalid"}) + assert tool.check_str_for_functions_valid(invalid_json) is False + print("check_str_for_functions_valid test passed") + + +def test_convert_funcs_into_tools(): + print("Testing convert_funcs_into_tools") + tool = BaseTool(tools=[sample_function]) + + tool.convert_funcs_into_tools() + assert "sample_function" in tool.function_map + assert callable(tool.function_map["sample_function"]) + print("convert_funcs_into_tools test passed") + + +def run_all_tests(): + print("Starting all tests") + + tests = [ + test_func_to_dict, + test_base_model_to_dict, + test_detect_tool_input_type, + test_execute_tool_by_name, + test_check_str_for_functions_valid, + test_convert_funcs_into_tools, + ] + + for test in tests: + try: + test() + except AssertionError as e: + print(f"Test {test.__name__} failed: {str(e)}") + except Exception as e: + print(f"Unexpected error in {test.__name__}: {str(e)}") + + print("All tests completed") + + +if __name__ == "__main__": + run_all_tests()