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/761/head
Kye Gomez 3 months ago
parent be62d30303
commit a4284f02de
1 changed files with 229 additions and 440 deletions
Show Stats Download Patch File Download Diff File

669
docs/swarms_tools/twitter.md
Unescape View File

@ -1,533 +1,322 @@
# TwitterTool Technical Documentation
# Twitter Tool Documentation
## Class Overview
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 Requirements
## Overview
The Twitter Tool provides a convenient interface for interacting with Twitter's API through the swarms-tools package. This documentation covers the initialization process and available functions for posting, replying, liking, and quoting tweets, as well as retrieving metrics.
## Installation
```bash
pip install swarms-tools
```
## Class Definition
## Authentication
The Twitter Tool requires Twitter API credentials for authentication. These should be stored as environment variables:
### Constructor Parameters
The `options` dictionary accepts the following keys:
```python
TWITTER_API_KEY=your_api_key
TWITTER_API_SECRET_KEY=your_api_secret_key
TWITTER_ACCESS_TOKEN=your_access_token
TWITTER_ACCESS_TOKEN_SECRET=your_access_token_secret
```
| 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 |
## Initialization
The `credentials` dictionary requires:
### TwitterTool Configuration 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 |
| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| id | str | Yes | Unique identifier for the Twitter tool instance |
| name | str | Yes | Name of the Twitter tool instance |
| description | str | No | Description of the tool's purpose |
| credentials | dict | Yes | Dictionary containing Twitter API credentials |
## Public Methods
### Credentials Dictionary Structure
### available_functions
| Key | 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 |
```python
@property
def available_functions(self) -> List[str]
```
## Available Functions
Returns a list of available function names that can be executed by the tool.
### initialize_twitter_tool()
**Returns:**
- List[str]: Names of available functions ['get_metrics', 'reply_tweet', 'post_tweet', 'like_tweet', 'quote_tweet']
Creates and returns a new instance of the TwitterTool.
**Example:**
```python
twitter_tool = TwitterTool(options)
functions = twitter_tool.available_functions
print(f"Available functions: {functions}")
def initialize_twitter_tool() -> TwitterTool:
```
### get_function
Returns:
- TwitterTool: Initialized Twitter tool instance
```python
def get_function(self, fn_name: str) -> Callable
```
### post_tweet()
Retrieves a specific function by name for execution.
Posts a new tweet to Twitter.
**Parameters:**
- fn_name (str): Name of the function to retrieve
| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| tweet | str | Yes | Text content of the tweet to post |
**Returns:**
- Callable: The requested function
Raises:
- tweepy.TweepyException: If tweet posting fails
**Raises:**
- ValueError: If the function name is not found
### reply_tweet()
**Example:**
```python
post_tweet = twitter_tool.get_function('post_tweet')
metrics_fn = twitter_tool.get_function('get_metrics')
```
Replies to an existing tweet.
## Protected Methods
| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| tweet_id | int | Yes | ID of the tweet to reply to |
| reply | str | Yes | Text content of the reply |
### _get_metrics
Raises:
- tweepy.TweepyException: If reply posting fails
```python
def _get_metrics(self) -> Dict[str, int]
```
Fetches user metrics including followers, following, and tweet counts.
### like_tweet()
**Returns:**
- Dict[str, int]: Dictionary containing metrics
- followers: Number of followers
- following: Number of accounts following
- tweets: Total tweet count
Likes a specified tweet.
**Example:**
```python
metrics = twitter_tool._get_metrics()
print(f"Followers: {metrics['followers']}")
print(f"Following: {metrics['following']}")
print(f"Total tweets: {metrics['tweets']}")
```
| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| tweet_id | int | Yes | ID of the tweet to like |
### _reply_tweet
Raises:
- tweepy.TweepyException: If liking the tweet fails
```python
def _reply_tweet(self, tweet_id: int, reply: str) -> None
```
### quote_tweet()
Posts a reply to a specific tweet.
Creates a quote tweet.
**Parameters:**
- tweet_id (int): ID of the tweet to reply to
- reply (str): Text content of the reply
| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| tweet_id | int | Yes | ID of the tweet to quote |
| quote | str | Yes | Text content to add to the quoted tweet |
**Example:**
```python
twitter_tool._reply_tweet(
tweet_id=1234567890,
reply="Thank you for sharing this insight!"
)
```
Raises:
- tweepy.TweepyException: If quote tweet creation fails
### _post_tweet
### get_metrics()
```python
def _post_tweet(self, tweet: str) -> Dict[str, Any]
```
Retrieves Twitter metrics.
Creates a new tweet.
Returns:
- Dict[str, int]: Dictionary containing various Twitter metrics
**Parameters:**
- tweet (str): Text content of the tweet
Raises:
- tweepy.TweepyException: If metrics retrieval fails
**Returns:**
- Dict[str, Any]: Response from Twitter API
## Usage Examples
**Example:**
### Basic Tweet Posting
```python
twitter_tool._post_tweet(
tweet="Exploring the fascinating world of AI and machine learning! #AI #ML"
)
```
### _like_tweet
from swarms_tools.twitter import initialize_twitter_tool, post_tweet
```python
def _like_tweet(self, tweet_id: int) -> None
# Post a simple tweet
post_tweet("Hello, Twitter!")
```
Likes a specific tweet.
**Parameters:**
- tweet_id (int): ID of the tweet to like
**Example:**
### Interacting with Tweets
```python
twitter_tool._like_tweet(tweet_id=1234567890)
```
# Reply to a tweet
reply_tweet(12345, "Great point!")
### _quote_tweet
# Like a tweet
like_tweet(12345)
```python
def _quote_tweet(self, tweet_id: int, quote: str) -> None
# Quote a tweet
quote_tweet(12345, "Adding my thoughts on this!")
```
Creates a quote tweet.
**Parameters:**
- tweet_id (int): ID of the tweet to quote
- quote (str): Text content to add to the quote
**Example:**
### Retrieving Metrics
```python
twitter_tool._quote_tweet(
tweet_id=1234567890,
quote="This is a fascinating perspective on AI development!"
)
metrics = get_metrics()
print(f"Current metrics: {metrics}")
```
## Integration Examples
### Basic Usage with Environment Variables
## Error Handling
All functions include built-in error handling and will print error messages if operations fail. It's recommended to implement additional error handling in production environments:
```python
import os
from dotenv import load_dotenv
from swarms_tools import TwitterTool
load_dotenv()
try:
post_tweet("Hello, Twitter!")
except Exception as e:
logger.error(f"Tweet posting failed: {e}")
# Implement appropriate error handling
```
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)
```
## Production Example
### Integration with AI Agent for Content Generation
This is an example of how to use the TwitterTool in a production environment using Swarms.
```python
from swarms import Agent
from swarms_models import OpenAIChat
from swarms_tools import TwitterTool
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()
# Initialize the model
model_name = "gpt-4o"
model = OpenAIChat(
model_name="gpt-4",
model_name=model_name,
max_tokens=3000,
openai_api_key=os.getenv("OPENAI_API_KEY")
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')
def generate_and_post(self, topic: str) -> None:
"""
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")
```
### Automated Engagement System
```python
import time
from typing import List, Dict
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)
}
})
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]
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."
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,
)
```
### Scheduled Content System
```python
import schedule
from datetime import datetime, timedelta
class ScheduledTwitterBot:
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')
self.posted_tweets: List[str] = []
def generate_daily_content(self) -> None:
"""
Generates and posts daily content based on the current date.
"""
today = datetime.now()
prompt = f"""
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:
schedule.run_pending()
time.sleep(60)
```
## Error Handling
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
class TwitterToolWithCustomErrors(TwitterTool):
def _post_tweet(self, tweet: str) -> Dict[str, Any]:
"""
Enhanced tweet posting with custom error handling.
Args:
tweet (str): The tweet content to post
Returns:
Dict[str, Any]: Response from Twitter API
Raises:
ValueError: If tweet exceeds character limit
"""
if len(tweet) > 280:
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
def _check_rate_limit(self) -> None:
"""
Checks if we're within rate limits and waits if necessary.
"""
now = time()
# Remove timestamps older than 15 minutes
while self.tweet_timestamps and self.tweet_timestamps[0] < now - 900:
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)
```
## Best Practices
1. Always use environment variables for credentials:
```python
from dotenv import load_dotenv
load_dotenv()
# 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")
}
"accessTokenSecret": os.getenv("TWITTER_ACCESS_TOKEN_SECRET"),
},
}
```
2. Implement proper logging:
```python
import logging
# Initialize the TwitterTool with your options
twitter_plugin = TwitterTool(options)
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
```
# # Post a tweet
# post_tweet_fn = twitter_plugin.get_function('post_tweet')
# post_tweet_fn("Hello world!")
3. Use type hints consistently:
```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:
return func()
except tweepy.TweepyException as e:
if attempt == max_retries - 1:
raise
sleep_time = (2 ** attempt) + random.uniform(0, 1)
sleep(sleep_time)
```
# Assuming `twitter_plugin` and `medical_coder` are already initialized
post_tweet = twitter_plugin.get_function("post_tweet")
5. Validate input data:
```python
def validate_tweet(tweet: str) -> bool:
if not tweet or len(tweet) > 280:
return False
return True
```
# Set to track posted tweets and avoid duplicates
posted_tweets = set()
## Testing
Example of a basic test suite for the TwitterTool:
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!"
)
```python
from swarms_tools import TwitterTool
import os
from dotenv import load_dotenv
# Generate a new tweet text
tweet_text = medical_coder.run(tweet_prompt)
load_dotenv()
# Check for duplicates
if tweet_text in posted_tweets:
print("Duplicate tweet detected. Skipping...")
return
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
# Test function retrieval
post_tweet = tool.get_function('post_tweet')
assert callable(post_tweet)
# Test metrics
metrics = tool._get_metrics()
assert isinstance(metrics, dict)
assert 'followers' in metrics
assert 'following' in metrics
assert 'tweets' in metrics
# 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)
```
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.
## Best Practices
1. Always store credentials in environment variables
2. Implement rate limiting in production environments
3. Add proper logging for all operations
4. Handle errors gracefully
5. Validate tweet content before posting
6. Monitor API usage limits
## Rate Limits
Be aware of Twitter's API rate limits. Implement appropriate delays between requests in production environments to avoid hitting these limits.
## Dependencies
- tweepy
- python-dotenv
- swarms-tools
## Version Compatibility
- Python 3.7+
- Latest version of swarms-tools package
Write Preview
Loading…
Cancel
Save