The Social Media Agent analyzes your Twitter writing style, stores it in Memori, generates new tweets that sound exactly like you using Nebius AI, and posts them automatically using Composioโs Twitter integration.
from langchain_nebius import ChatNebiusimport jsonnebius_chat = ChatNebius( api_key=os.getenv("NEBIUS_API_KEY"), model="zai-org/GLM-4.5-Air", temperature=0.6, top_p=0.95,)def analyze_tweeting_style(tweets: list) -> dict: """Analyze tweeting style using Nebius LLM.""" # Prepare tweets for analysis tweets_text = "" for i, tweet in enumerate(tweets, 1): tweets_text += f"Tweet {i}: {tweet.get('description', 'N/A')}\n" prompt = f""" Analyze the following tweets and extract the author's tweeting style characteristics. Focus on: 1. Tone (casual, professional, humorous, serious, etc.) 2. Language style (formal, informal, slang, technical, etc.) 3. Common hashtags and their themes 4. Emoji usage patterns 5. Tweet structure and length preferences 6. Topics and interests they tweet about 7. Writing personality and voice 8. Common phrases or expressions they use Tweets to analyze: {tweets_text} Provide your analysis in JSON format with these keys: - tone: string describing the tone - language_style: string describing language formality and style - hashtag_patterns: list of common hashtags and themes - emoji_usage: string describing emoji patterns - tweet_structure: string describing tweet length and structure preferences - common_topics: list of topics they tweet about - personality: string describing overall personality - common_phrases: list of phrases or expressions they use - writing_habits: list of specific tweeting habits or patterns """ response = nebius_chat.invoke(prompt) analysis_text = response.content # Extract JSON from response if "```json" in analysis_text: json_start = analysis_text.find("```json") + 7 json_end = analysis_text.find("```", json_start) json_content = analysis_text[json_start:json_end].strip() else: json_start = analysis_text.find("{") json_end = analysis_text.rfind("}") + 1 json_content = analysis_text[json_start:json_end] return json.loads(json_content)
def store_tweeting_style_in_memori( memory_system, style_analysis: dict, twitter_handle: str, profile_image_url: str, handle: str, username: str,): """Store tweeting style analysis in Memori.""" # Create a conversation about the tweeting style user_input = f""" Hi AI, here is my Twitter style from @{twitter_handle}: {style_analysis.get('tone', 'N/A')} tone, {style_analysis.get('personality', 'N/A')} personality, {style_analysis.get('language_style', 'N/A')} language style. """ ai_response = f""" I understand your Twitter style! You tweet with a {style_analysis.get('tone', 'N/A')} tone and {style_analysis.get('personality', 'N/A')} personality. Your language is {style_analysis.get('language_style', 'N/A')}. I'll use this to generate tweets that sound exactly like you. """ # Record in Memori with metadata memory_system.record_conversation( user_input=user_input, ai_output=ai_response, model="nebius-glm-4.5-air", metadata={ "type": "twitter_style_profile", "twitter_handle": twitter_handle, "style_data": style_analysis, "analysis_timestamp": "now", "profile_image_url": profile_image_url, "username": username, "handle": handle, }, ) return ai_response
def generate_tweet_with_style(memory_tool, topic: str) -> str: """Generate tweet using stored tweeting style from memory.""" # Get tweeting style context from memory writing_style_context = "" try: context_result = memory_tool.execute(query="twitter style tone personality") if context_result and "No relevant memories found" not in str(context_result): writing_style_context = str(context_result)[:300] except Exception: pass # Create prompt based on whether we have style if writing_style_context: prompt = f"Write a tweet about '{topic}' in this exact style: {writing_style_context}. Keep it casual, under 275 characters. NO quotes. Just the tweet." else: prompt = f"Write a casual tweet about '{topic}'. Under 275 characters. NO quotes. Just the tweet." response = nebius_chat.invoke(prompt) tweet_content = response.content.strip() # Clean up formatting if tweet_content.startswith('"') and tweet_content.endswith('"'): tweet_content = tweet_content[1:-1] # Ensure character limit if len(tweet_content) > 275: tweet_content = tweet_content[:275] + "..." return tweet_content