Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: random tweet intervals #15

Closed
wants to merge 2 commits into from
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion agents/example.json
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
"This is an example tweet.",
"This is another example tweet."
],
"loop_delay": 900,
"loop_delay": "random",
"config": [
{
"name": "twitter",
Expand Down
42 changes: 26 additions & 16 deletions src/agent.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,16 +7,14 @@
from dotenv import load_dotenv
from src.connection_manager import ConnectionManager
from src.helpers import print_h_bar
from typing import Union

REQUIRED_FIELDS = ["name", "bio", "traits", "examples", "loop_delay", "config", "tasks"]

logger = logging.getLogger("agent")

class ZerePyAgent:
def __init__(
self,
agent_name: str
):
def __init__(self, agent_name: str):
try:
agent_path = Path("agents") / f"{agent_name}.json"
agent_dict = json.load(open(agent_path, "r"))
Expand All @@ -25,26 +23,37 @@ def __init__(
if missing_fields:
raise KeyError(f"Missing required fields: {', '.join(missing_fields)}")

self.name = agent_dict["name"]
self.bio = agent_dict["bio"]
self.traits = agent_dict["traits"]
self.examples = agent_dict["examples"]
self.loop_delay = agent_dict["loop_delay"]
self.connection_manager = ConnectionManager(agent_dict["config"])
self.name: str = agent_dict["name"]
self.bio: list[str] = agent_dict["bio"]
self.traits: list[str] = agent_dict["traits"]
self.examples: list[str] = agent_dict["examples"]
self.loop_delay: float = agent_dict["loop_delay"]
self.connection_manager: ConnectionManager = ConnectionManager(agent_dict["config"])

# Extract Twitter config
twitter_config = next((config for config in agent_dict["config"] if config["name"] == "twitter"), None)
if not twitter_config:
raise KeyError("Twitter configuration is required")

# TODO: These should probably live in the related task parameters
self.self_reply_chance = twitter_config.get("self_reply_chance", 0.05)
self.tweet_interval = twitter_config.get("tweet_interval", 900)

self.is_llm_set = False
self.self_reply_chance: float = twitter_config.get("self_reply_chance", 0.05)

# Extract and validate tweet interval
tweet_interval_config = twitter_config.get("tweet_interval", 900)
if isinstance(tweet_interval_config, str):
tweet_interval_config = tweet_interval_config.lower()
if tweet_interval_config != "random":
raise ValueError("String tweet interval must be 'random'")
elif not isinstance(tweet_interval_config, (int, float)):
raise ValueError("Tweet interval must be a number or 'random'")
elif tweet_interval_config <= 0:
raise ValueError("Tweet interval must be positive")
self.tweet_interval: Union[int, float, str] = tweet_interval_config

self.is_llm_set: bool = False

# Cache for system prompt
self._system_prompt = None
self._system_prompt: str = None

# Extract loop tasks
self.tasks = agent_dict.get("tasks", [])
Expand Down Expand Up @@ -138,7 +147,8 @@ def loop(self):
if action_name == "post-tweet":
# Check if it's time to post a new tweet
current_time = time.time()
if current_time - last_tweet_time >= self.tweet_interval:
interval = self.tweet_interval if self.tweet_interval != "random" else random.randint(300, 1800) # 5-30 minutes
if current_time - last_tweet_time >= interval:
logger.info("\n📝 GENERATING NEW TWEET")
print_h_bar()

Expand Down
4 changes: 2 additions & 2 deletions src/connections/twitter_connection.py
Original file line number Diff line number Diff line change
Expand Up @@ -216,8 +216,8 @@ def configure(self) -> None:
"\n📝 To get your Twitter API credentials:",
"1. Go to https://developer.twitter.com/en/portal/dashboard",
"2. Create a new project and app if you haven't already",
"3. In your app settings, enable OAuth 1.0a with read and write permissions",
"4. Get your API Key (consumer key) and API Key Secret (consumer secret)"
"3. In your app settings, enable OAuth 1.0a with read and write permissions (this may be enabled by default)",
"4. Get your API Key (consumer key) and API Key Secret (consumer secret) for the 'Keys and tokens' tab"
]
logger.info("\n".join(setup_instructions))
print_h_bar()
Expand Down