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.

Sign up for GitHub

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

feature: refactor skill_x_api and add image generation #28

Open
wants to merge 3 commits into
base: main
Choose a base branch
from
Open

feature: refactor skill_x_api and add image generation #28

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
c913f90
refactor xapiskill and add image generation
shihao-guo Jan 29, 2025
f20323a
default image generation to false
shihao-guo Jan 31, 2025
dd7bb8d
update image upload method due to composio api change
shihao-guo Feb 4, 2025
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
97 changes: 56 additions & 41 deletions my_digital_being/activities/activity_post_a_tweet.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,19 +1,21 @@
import logging
from typing import Dict, Any, List
from typing import Dict, Any, List, Tuple

from framework.activity_decorator import activity, ActivityBase, ActivityResult
from framework.api_management import api_manager
from framework.memory import Memory
from skills.skill_chat import chat_skill
from skills.skill_generate_image import ImageGenerationSkill
from skills.skill_x_api import XAPISkill

logger = logging.getLogger(__name__)


@activity(
name="post_a_tweet",
energy_cost=0.4,
cooldown=10000, # 1 hour
required_skills=["twitter_posting"],
cooldown=3600, # 1 hour
required_skills=["twitter_posting", "image_generation"],
)
class PostTweetActivity(ActivityBase):
"""
Expand All @@ -26,11 +28,13 @@ class PostTweetActivity(ActivityBase):
def __init__(self):
super().__init__()
self.max_length = 280
# The Composio action name from your logs
self.composio_action = "TWITTER_CREATION_OF_A_POST"
# If you know your Twitter username, you can embed it in the link
# or fetch it dynamically. Otherwise, substitute accordingly:
self.twitter_username = "YourUserName"
# set this to True if you want to generate an image for the tweet
self.image_generation_enabled = False
self.default_size = (1024, 1024) # Added for image generation
self.default_format = "png" # Added for image generation

async def execute(self, shared_data) -> ActivityResult:
try:
Expand Down Expand Up @@ -61,8 +65,18 @@ async def execute(self, shared_data) -> ActivityResult:
if len(tweet_text) > self.max_length:
tweet_text = tweet_text[: self.max_length - 3] + "..."

# 4) Post the tweet via Composio
post_result = self._post_tweet_via_composio(tweet_text)
# 4) Generate an image based on the tweet text
if self.image_generation_enabled:
image_prompt, media_urls = await self._generate_image_for_tweet(tweet_text, personality_data)
else:
image_prompt, media_urls = None, []

# 5) Post the tweet via X API
x_api = XAPISkill({
"enabled": True,
"twitter_username": self.twitter_username
})
post_result = await x_api.post_tweet(tweet_text, media_urls)
if not post_result["success"]:
error_msg = post_result.get(
"error", "Unknown error posting tweet via Composio"
Expand All @@ -77,7 +91,7 @@ async def execute(self, shared_data) -> ActivityResult:
else None
)

# 5) Return success, adding link & prompt in metadata
# 6) Return success, adding link & prompt in metadata
logger.info(f"Successfully posted tweet: {tweet_text[:50]}...")
return ActivityResult(
success=True,
Expand All @@ -88,7 +102,9 @@ async def execute(self, shared_data) -> ActivityResult:
"model": chat_response["data"].get("model"),
"finish_reason": chat_response["data"].get("finish_reason"),
"tweet_link": tweet_link,
"prompt_used": prompt_text, # <--- includes the full prompt
"prompt_used": prompt_text,
"image_prompt_used": image_prompt,
"image_count": len(media_urls),
},
)

Expand Down Expand Up @@ -159,38 +175,37 @@ def _build_chat_prompt(
f"but not repeating old tweets. Avoid hashtags or repeated phrases.\n"
)

def _post_tweet_via_composio(self, tweet_text: str) -> Dict[str, Any]:
def _build_image_prompt(self, tweet_text: str, personality: Dict[str, Any]) -> str:
personality_str = "\n".join(f"{t}: {v}" for t, v in personality.items())
return f"Our digital being has these personality traits:\n" \
f"{personality_str}\n\n" \
f"And is creating a tweet with the text: {tweet_text}\n\n" \
f"Generate an image that represents the story of the tweet and reflects the personality traits. Do not include the tweet text in the image."

async def _generate_image_for_tweet(self, tweet_text: str, personality_data: Dict[str, Any]) -> Tuple[str, List[str]]:
"""
Post a tweet using the "Creation of a post" Composio action.
The response returns {'successfull': True, ...}, not 'success'.
We'll check 'successfull' or fallback if needed.
Generate an image for the tweet and upload it to Twitter.
Returns a tuple of (image_prompt, media_urls).
If generation fails, returns (None, []).
"""
try:
from framework.composio_integration import composio_manager

logger.info(
f"Posting tweet via Composio action='{self.composio_action}', text='{tweet_text[:50]}...'"
)

response = composio_manager._toolset.execute_action(
action=self.composio_action,
params={"text": tweet_text},
entity_id="MyDigitalBeing",
logger.info("Decided to generate an image for tweet")
image_skill = ImageGenerationSkill({
"enabled": True,
"max_generations_per_day": 50,
"supported_formats": ["png", "jpg"],
})

if await image_skill.can_generate():
image_prompt = self._build_image_prompt(tweet_text, personality_data)
image_result = await image_skill.generate_image(
prompt=image_prompt,
size=self.default_size,
format=self.default_format
)

# The actual success key is "successfull" (with 2 Ls)
success_val = response.get("success", response.get("successfull"))
if success_val:
data_section = response.get("data", {})
nested_data = data_section.get("data", {})
tweet_id = nested_data.get("id")
return {"success": True, "tweet_id": tweet_id}
else:
return {
"success": False,
"error": response.get("error", "Unknown or missing success key"),
}

except Exception as e:
logger.error(f"Error in Composio tweet post: {e}", exc_info=True)
return {"success": False, "error": str(e)}

if image_result.get("success") and image_result.get("image_data", {}).get("url"):
return image_prompt, [image_result["image_data"]["url"]]
else:
logger.warning("Image generation not available, proceeding with text-only tweet")

return None, []
112 changes: 9 additions & 103 deletions my_digital_being/activities/activity_post_recent_memory_tweet.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
from framework.api_management import api_manager
from framework.memory import Memory
from skills.skill_chat import chat_skill
from skills.skill_x_api import XAPISkill

logger = logging.getLogger(__name__)

Expand All @@ -27,7 +28,6 @@ class PostRecentMemoriesTweetActivity(ActivityBase):
def __init__(self, num_activities_to_fetch: int = 10):
super().__init__()
self.max_length = 280
self.composio_action = "TWITTER_CREATION_OF_A_POST"
self.twitter_username = "YourUserName"

# Activity types to ignore in memory results
Expand Down Expand Up @@ -88,9 +88,8 @@ async def execute(self, shared_data) -> ActivityResult:
new_memories=new_memories,
)

# 6) Extract drawing URLs and upload to Twitter media
# 6) Extract drawing URLs from memories
drawing_urls = self._extract_drawing_urls(new_memories)
media_ids = await self._upload_drawings_to_twitter(drawing_urls)

# 7) Use chat skill to generate the tweet text
chat_response = await chat_skill.get_chat_completion(
Expand All @@ -108,8 +107,12 @@ async def execute(self, shared_data) -> ActivityResult:
if len(tweet_text) > self.max_length:
tweet_text = tweet_text[: self.max_length - 3] + "..."

# 8) Post to Twitter via Composio
post_result = self._post_tweet_via_composio(tweet_text, media_ids)
# 8) Post to Twitter via X API with any extracted images
x_api = XAPISkill({
"enabled": True,
"twitter_username": self.twitter_username
})
post_result = await x_api.post_tweet(tweet_text, drawing_urls)
if not post_result["success"]:
error_msg = post_result.get(
"error", "Unknown error posting tweet via Composio"
Expand Down Expand Up @@ -141,6 +144,7 @@ async def execute(self, shared_data) -> ActivityResult:
"prompt_used": prompt_text,
"model": chat_response["data"].get("model"),
"finish_reason": chat_response["data"].get("finish_reason"),
"image_count": len(drawing_urls),
},
)

Expand Down Expand Up @@ -264,42 +268,6 @@ def _build_chat_prompt(
)
return prompt

def _post_tweet_via_composio(self, tweet_text: str, media_ids: List[str]) -> Dict[str, Any]:
"""
Post tweet via Composio with optional media_ids.
"""
try:
from framework.composio_integration import composio_manager

logger.info(
f"Posting tweet via Composio action='{self.composio_action}', text='{tweet_text[:50]}...', media_count={len(media_ids)}"
)

response = composio_manager._toolset.execute_action(
action=self.composio_action,
params={
"text": tweet_text,
"media__media__ids": media_ids if media_ids else None
},
entity_id="MyDigitalBeing",
)

success_val = response.get("success", response.get("successfull"))
if success_val:
data_section = response.get("data", {})
nested_data = data_section.get("data", {})
tweet_id = nested_data.get("id")
return {"success": True, "tweet_id": tweet_id}
else:
return {
"success": False,
"error": response.get("error", "Unknown or missing success key"),
}

except Exception as e:
logger.error(f"Error in Composio tweet post: {e}", exc_info=True)
return {"success": False, "error": str(e)}

def _extract_drawing_urls(self, memories: List[str]) -> List[str]:
"""
Extract URLs from all DrawActivity entries in memories.
Expand Down Expand Up @@ -329,65 +297,3 @@ def _extract_drawing_urls(self, memories: List[str]) -> List[str]:
continue

return drawing_urls

async def _upload_drawings_to_twitter(self, drawing_urls: List[str]) -> List[str]:
"""
Downloads images from URLs and uploads them to Twitter via Composio.
Returns a list of Twitter media IDs.
"""
import aiohttp
import base64
from framework.composio_integration import composio_manager

media_ids = []

if not drawing_urls:
return media_ids

async with aiohttp.ClientSession() as session:
for url in drawing_urls:
try:
# Download image
async with session.get(url) as response:
if response.status != 200:
logger.warning(f"Failed to download image from {url}: {response.status}")
continue

image_data = await response.read()

# Convert to base64
base64_image = base64.b64encode(image_data).decode('utf-8')

# Extract filename from URL or use default
filename = url.split('/')[-1].split('?')[0] or 'image.png'

# Upload to Twitter via Composio
upload_response = composio_manager._toolset.execute_action(
action="TWITTER_MEDIA_UPLOAD_MEDIA",
params={
"media": {
"name": filename,
"content": base64_image
}
},
entity_id="MyDigitalBeing"
)

# Composio returns 'successfull' instead of 'successful'
if upload_response.get("successful") or upload_response.get("successfull"):
media_id = upload_response.get("media_id") or upload_response.get("data", {}).get("media_id")
if media_id:
media_ids.append(media_id)
logger.info(f"Successfully uploaded image to Twitter, media_id: {media_id}")
else:
logger.warning(f"Upload succeeded but no media_id returned. Response: {upload_response}")
else:
error = upload_response.get("error", "Unknown error")
logger.warning(f"Failed to upload image to Twitter: {error}")

except Exception as e:
logger.error(f"Error uploading image to Twitter: {e}", exc_info=True)
continue

logger.debug(f"Uploaded drawing URLs to Twitter media: {media_ids}")
return media_ids
Loading