ai-mirow-projects
/
swarms
mirror of https://github.com/kyegomez/swarms
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

[TWITTER AGENTS]

Browse Source
pull/741/merge
Kye Gomez 2 weeks ago
parent c9b2079710
commit be62d30303
1 changed files with 429 additions and 249 deletions
Show Stats Download Patch File Download Diff File

678
docs/swarms_tools/twitter.md
Unescape View File

@ -1,353 +1,533 @@
# TwitterTool Documentation # TwitterTool Technical Documentation
## Overview ## Class 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. The TwitterTool class provides a comprehensive interface for interacting with Twitter's API. It encapsulates common Twitter operations and provides error handling, logging, and integration capabilities with AI agents.
## Installation ## Installation Requirements
Before using the TwitterTool, ensure you have the required dependencies installed:
```bash ```bash
pip install tweepy swarms-tools pip install swarms-tools
``` ```
## Basic Configuration ## Class Definition
The TwitterTool requires Twitter API credentials for authentication. Here's how to set up the basic configuration: ### Constructor Parameters
```python The `options` dictionary accepts the following keys:
from swarms_tools.social_media.twitter_tool import TwitterTool
import os | Parameter | Type | Required | Default | Description |
|-----------|------|----------|----------|-------------|
| id | str | No | "twitter_plugin" | Unique identifier for the tool instance |
| name | str | No | "Twitter Plugin" | Display name for the tool |
| description | str | No | Default description | Tool description |
| credentials | Dict[str, str] | Yes | None | Twitter API credentials |
options = { The `credentials` dictionary requires:
"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) | Credential | Type | Required | Description |
|------------|------|----------|-------------|
| apiKey | str | Yes | Twitter API key |
| apiSecretKey | str | Yes | Twitter API secret key |
| accessToken | str | Yes | Twitter access token |
| accessTokenSecret | str | Yes | Twitter access token secret |
## Public Methods
### available_functions
```python
@property
def available_functions(self) -> List[str]
``` ```
For security, it's recommended to use environment variables for credentials: Returns a list of available function names that can be executed by the tool.
**Returns:**
- List[str]: Names of available functions ['get_metrics', 'reply_tweet', 'post_tweet', 'like_tweet', 'quote_tweet']
**Example:**
```python ```python
import os twitter_tool = TwitterTool(options)
from dotenv import load_dotenv functions = twitter_tool.available_functions
print(f"Available functions: {functions}")
```
load_dotenv() ### get_function
options = { ```python
"id": "twitter_bot", def get_function(self, fn_name: str) -> Callable
"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 Retrieves a specific function by name for execution.
The TwitterTool provides five main functions: **Parameters:**
- fn_name (str): Name of the function to retrieve
1. **Posting Tweets**: Create new tweets **Returns:**
2. **Replying to Tweets**: Respond to existing tweets - Callable: The requested function
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 **Raises:**
- ValueError: If the function name is not found
**Example:**
```python ```python
# Get a specific function
post_tweet = twitter_tool.get_function('post_tweet') post_tweet = twitter_tool.get_function('post_tweet')
reply_tweet = twitter_tool.get_function('reply_tweet') metrics_fn = twitter_tool.get_function('get_metrics')
quote_tweet = twitter_tool.get_function('quote_tweet') ```
like_tweet = twitter_tool.get_function('like_tweet')
get_metrics = twitter_tool.get_function('get_metrics') ## Protected Methods
# Post a tweet ### _get_metrics
post_tweet("Hello, Twitter!")
# Reply to a tweet ```python
reply_tweet(tweet_id=123456789, reply="Great point!") def _get_metrics(self) -> Dict[str, int]
```
# Quote a tweet Fetches user metrics including followers, following, and tweet counts.
quote_tweet(tweet_id=123456789, quote="Interesting perspective!")
# Like a tweet **Returns:**
like_tweet(tweet_id=123456789) - Dict[str, int]: Dictionary containing metrics
- followers: Number of followers
- following: Number of accounts following
- tweets: Total tweet count
# Get account metrics **Example:**
metrics = get_metrics() ```python
metrics = twitter_tool._get_metrics()
print(f"Followers: {metrics['followers']}") print(f"Followers: {metrics['followers']}")
print(f"Following: {metrics['following']}")
print(f"Total tweets: {metrics['tweets']}")
``` ```
## Integration with Agents ### _reply_tweet
The TwitterTool can be particularly powerful when integrated with AI agents. Here are several examples of agent integrations: ```python
def _reply_tweet(self, tweet_id: int, reply: str) -> None
```
### 1. Medical Information Bot Posts a reply to a specific tweet.
This example shows how to create a medical information bot that shares health facts: **Parameters:**
- tweet_id (int): ID of the tweet to reply to
- reply (str): Text content of the reply
**Example:**
```python ```python
from swarms import Agent twitter_tool._reply_tweet(
from swarms_models import OpenAIChat tweet_id=1234567890,
reply="Thank you for sharing this insight!"
# 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 ### _post_tweet
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 ```python
def post_medical_fact(): def _post_tweet(self, tweet: str) -> Dict[str, Any]
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 Creates a new tweet.
**Parameters:**
- tweet (str): Text content of the tweet
This example demonstrates how to create a bot that summarizes news articles: **Returns:**
- Dict[str, Any]: Response from Twitter API
**Example:**
```python ```python
# Create a news summarization agent twitter_tool._post_tweet(
news_agent = Agent( tweet="Exploring the fascinating world of AI and machine learning! #AI #ML"
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): ### _like_tweet
# Generate summary
prompt = f"Summarize this news article in a tweet-length format: {article_url}" ```python
summary = news_agent.run(prompt) def _like_tweet(self, tweet_id: int) -> None
# Post the summary
post_tweet = twitter_tool.get_function('post_tweet')
post_tweet(f"{summary} Source: {article_url}")
``` ```
### 3. Interactive Q&A Bot Likes a specific tweet.
This example shows how to create a bot that responds to user questions: **Parameters:**
- tweet_id (int): ID of the tweet to like
**Example:**
```python ```python
class TwitterQABot: twitter_tool._like_tweet(tweet_id=1234567890)
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() ### _quote_tweet
qa_bot.handle_question(123456789, "What causes climate change?")
```python
def _quote_tweet(self, tweet_id: int, quote: str) -> None
``` ```
## Best Practices Creates a quote tweet.
When using the TwitterTool, especially with agents, consider these best practices: **Parameters:**
- tweet_id (int): ID of the tweet to quote
- quote (str): Text content to add to the quote
1. **Rate Limiting**: Implement delays between tweets to comply with Twitter's rate limits: **Example:**
```python ```python
import time twitter_tool._quote_tweet(
tweet_id=1234567890,
def post_with_rate_limit(tweet_text, delay=60): quote="This is a fascinating perspective on AI development!"
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: ## Integration Examples
```python
posted_tweets = set()
def post_unique_tweet(tweet_text): ### Basic Usage with Environment Variables
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 ```python
def safe_tweet(tweet_text): import os
try: from dotenv import load_dotenv
post_tweet = twitter_tool.get_function('post_tweet') from swarms_tools import TwitterTool
post_tweet(tweet_text)
except Exception as e: load_dotenv()
logging.error(f"Failed to post tweet: {e}")
# Implement retry logic or fallback behavior options = {
"id": "twitter_bot",
"name": "AI Twitter Bot",
"description": "An AI-powered 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"),
}
}
twitter_tool = TwitterTool(options)
``` ```
4. **Content Validation**: Validate content before posting: ### Integration with AI Agent for Content Generation
```python ```python
def validate_and_post(tweet_text): from swarms import Agent
if len(tweet_text) > 280: from swarms_models import OpenAIChat
tweet_text = tweet_text[:277] + "..." from swarms_tools import TwitterTool
import os
# Check for prohibited content
prohibited_terms = ["spam", "inappropriate"] from dotenv import load_dotenv
if any(term in tweet_text.lower() for term in prohibited_terms):
return False load_dotenv()
# Initialize the model
model = OpenAIChat(
model_name="gpt-4",
max_tokens=3000,
openai_api_key=os.getenv("OPENAI_API_KEY")
)
# Create content generation agent
content_agent = Agent(
agent_name="Content Creator",
system_prompt="""
You are an expert content creator for Twitter.
Create engaging, informative tweets that:
1. Are under 280 characters
2. Use appropriate hashtags
3. Maintain a professional tone
4. Include relevant calls to action
""",
llm=model
)
class TwitterContentBot:
def __init__(self, twitter_tool: TwitterTool, agent: Agent):
self.twitter_tool = twitter_tool
self.agent = agent
self.post_tweet = twitter_tool.get_function('post_tweet')
post_tweet = twitter_tool.get_function('post_tweet') def generate_and_post(self, topic: str) -> None:
post_tweet(tweet_text) """
return True Generates and posts content about a specific topic.
Args:
topic (str): The topic to create content about
"""
prompt = f"Create an engaging tweet about {topic}"
tweet_content = self.agent.run(prompt)
self.post_tweet(tweet_content)
# Usage
bot = TwitterContentBot(twitter_tool, content_agent)
bot.generate_and_post("artificial intelligence")
``` ```
## Advanced Features ### Automated Engagement System
### Scheduled Posting ```python
import time
from typing import List, Dict
Implement scheduled posting using Python's built-in scheduling capabilities: class TwitterEngagementSystem:
def __init__(self, twitter_tool: TwitterTool):
self.twitter_tool = twitter_tool
self.like_tweet = twitter_tool.get_function('like_tweet')
self.reply_tweet = twitter_tool.get_function('reply_tweet')
self.get_metrics = twitter_tool.get_function('get_metrics')
# Track engagement history
self.engagement_history: List[Dict] = []
def engage_with_tweet(self, tweet_id: int, should_like: bool = True,
reply_text: Optional[str] = None) -> None:
"""
Engages with a specific tweet through likes and replies.
Args:
tweet_id (int): The ID of the tweet to engage with
should_like (bool): Whether to like the tweet
reply_text (Optional[str]): Text to reply with, if any
"""
try:
if should_like:
self.like_tweet(tweet_id)
if reply_text:
self.reply_tweet(tweet_id, reply_text)
# Record engagement
self.engagement_history.append({
'timestamp': time.time(),
'tweet_id': tweet_id,
'actions': {
'liked': should_like,
'replied': bool(reply_text)
}
})
except Exception as e:
print(f"Failed to engage with tweet {tweet_id}: {e}")
# Usage
engagement_system = TwitterEngagementSystem(twitter_tool)
engagement_system.engage_with_tweet(
tweet_id=1234567890,
should_like=True,
reply_text="Great insights! Thanks for sharing."
)
```
### Scheduled Content System
```python ```python
from datetime import datetime
import schedule import schedule
from datetime import datetime, timedelta
def scheduled_tweet_job(): class ScheduledTwitterBot:
twitter_tool = TwitterTool(options) def __init__(self, twitter_tool: TwitterTool, agent: Agent):
post_tweet = twitter_tool.get_function('post_tweet') self.twitter_tool = twitter_tool
self.agent = agent
self.post_tweet = twitter_tool.get_function('post_tweet')
self.posted_tweets: List[str] = []
# Generate content using an agent def generate_daily_content(self) -> None:
content = medical_expert.run("Generate a health tip of the day") """
post_tweet(content) Generates and posts daily content based on the current date.
"""
# Schedule tweets for specific times today = datetime.now()
schedule.every().day.at("10:00").do(scheduled_tweet_job) prompt = f"""
schedule.every().day.at("15:00").do(scheduled_tweet_job) Create an engaging tweet for {today.strftime('%A, %B %d')}.
Focus on technology trends and insights.
"""
tweet_content = self.agent.run(prompt)
# Avoid duplicate content
if tweet_content not in self.posted_tweets:
self.post_tweet(tweet_content)
self.posted_tweets.append(tweet_content)
# Keep only last 100 tweets in memory
if len(self.posted_tweets) > 100:
self.posted_tweets.pop(0)
# Usage
scheduled_bot = ScheduledTwitterBot(twitter_tool, content_agent)
# Schedule daily posts
schedule.every().day.at("10:00").do(scheduled_bot.generate_daily_content)
schedule.every().day.at("15:00").do(scheduled_bot.generate_daily_content)
# Run the scheduler
while True: while True:
schedule.run_pending() schedule.run_pending()
time.sleep(60) time.sleep(60)
``` ```
### Analytics Integration ## Error Handling
Track the performance of your tweets: The TwitterTool implements comprehensive error handling through try-except blocks. All methods catch and handle `tweepy.TweepyException` for Twitter API-specific errors. Here's an example of implementing custom error handling:
```python ```python
class TweetAnalytics: class TwitterToolWithCustomErrors(TwitterTool):
def __init__(self, twitter_tool): def _post_tweet(self, tweet: str) -> Dict[str, Any]:
self.twitter_tool = twitter_tool """
self.metrics_history = [] Enhanced tweet posting with custom error handling.
def record_metrics(self): Args:
get_metrics = self.twitter_tool.get_function('get_metrics') tweet (str): The tweet content to post
current_metrics = get_metrics()
self.metrics_history.append({ Returns:
'timestamp': datetime.now(), Dict[str, Any]: Response from Twitter API
'metrics': current_metrics
}) Raises:
ValueError: If tweet exceeds character limit
def get_growth_rate(self): """
if len(self.metrics_history) < 2: if len(tweet) > 280:
return None raise ValueError("Tweet exceeds 280 character limit")
try:
return super()._post_tweet(tweet)
except tweepy.TweepyException as e:
self.logger.error(f"Twitter API error: {e}")
raise
```
## Rate Limiting
Twitter's API has rate limits that should be respected. Here's an example implementation of rate limiting:
```python
from time import time, sleep
from collections import deque
class RateLimitedTwitterTool(TwitterTool):
def __init__(self, options: Dict[str, Any]) -> None:
super().__init__(options)
self.tweet_timestamps = deque(maxlen=300) # Track last 300 tweets
self.max_tweets_per_15min = 300
latest = self.metrics_history[-1]['metrics'] def _check_rate_limit(self) -> None:
previous = self.metrics_history[-2]['metrics'] """
Checks if we're within rate limits and waits if necessary.
"""
now = time()
return { # Remove timestamps older than 15 minutes
'followers_growth': latest['followers'] - previous['followers'], while self.tweet_timestamps and self.tweet_timestamps[0] < now - 900:
'tweets_growth': latest['tweets'] - previous['tweets'] self.tweet_timestamps.popleft()
}
if len(self.tweet_timestamps) >= self.max_tweets_per_15min:
sleep_time = 900 - (now - self.tweet_timestamps[0])
if sleep_time > 0:
sleep(sleep_time)
self.tweet_timestamps.append(now)
def _post_tweet(self, tweet: str) -> Dict[str, Any]:
self._check_rate_limit()
return super()._post_tweet(tweet)
``` ```
## Troubleshooting ## Best Practices
Common issues and their solutions: 1. Always use environment variables for credentials:
```python
from dotenv import load_dotenv
load_dotenv()
1. **Authentication Errors**: Double-check your API credentials and ensure they're properly loaded from environment variables. options = {
"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")
}
}
```
2. **Rate Limiting**: If you encounter rate limit errors, implement exponential backoff: 2. Implement proper logging:
```python ```python
import time import logging
from random import uniform
def exponential_backoff(attempt): logging.basicConfig(
wait_time = min(300, (2 ** attempt) + uniform(0, 1)) level=logging.INFO,
time.sleep(wait_time) format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
```
def retry_post(tweet_text, max_attempts=5): 3. Use type hints consistently:
for attempt in range(max_attempts): ```python
from typing import Optional, Dict, Any, List
def process_tweet(
tweet_id: int,
action: str,
content: Optional[str] = None
) -> Dict[str, Any]:
pass
```
4. Handle API rate limits gracefully:
```python
from time import sleep
def retry_with_backoff(func, max_retries: int = 3):
for attempt in range(max_retries):
try: try:
post_tweet = twitter_tool.get_function('post_tweet') return func()
post_tweet(tweet_text) except tweepy.TweepyException as e:
return True if attempt == max_retries - 1:
except Exception as e: raise
if attempt < max_attempts - 1: sleep_time = (2 ** attempt) + random.uniform(0, 1)
exponential_backoff(attempt) sleep(sleep_time)
else: ```
raise e
5. Validate input data:
```python
def validate_tweet(tweet: str) -> bool:
if not tweet or len(tweet) > 280:
return False
return True
``` ```
3. **Content Length Issues**: Implement automatic content truncation: ## Testing
Example of a basic test suite for the TwitterTool:
```python ```python
def truncate_tweet(text, max_length=280): from swarms_tools import TwitterTool
if len(text) <= max_length: import os
return text from dotenv import load_dotenv
load_dotenv()
def test_twitter_tool():
# Test configuration
options = {
"id": "test_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")
}
}
# Initialize tool
tool = TwitterTool(options)
# Test available functions
assert 'post_tweet' in tool.available_functions
assert 'get_metrics' in tool.available_functions
# Try to break at last space before limit # Test function retrieval
truncated = text[:max_length-3] post_tweet = tool.get_function('post_tweet')
last_space = truncated.rfind(' ') assert callable(post_tweet)
if last_space > 0:
truncated = truncated[:last_space]
return truncated + "..." # Test metrics
metrics = tool._get_metrics()
assert isinstance(metrics, dict)
assert 'followers' in metrics
assert 'following' in metrics
assert 'tweets' in metrics
``` ```
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. Remember to always test your code thoroughly before deploying it in a production environment. The TwitterTool is designed to be extensible, so you can add new features and customizations as needed for your specific use case.
Write Preview
Loading…
Cancel
Save