Building a Social Media AI Agent: A Step-by-Step Guide

mkdir social_media_agent
cd social_media_agent
python -m venv venv
source venv/bin/activate  # On Windows, use venv\Scripts\activate

openai==1.3.0
agentops==0.1.7
python-dotenv==1.0.0
tweepy==4.14.0
facebook-sdk==3.1.0
instagram-private-api==1.6.0
linkedin-api==2.0.0
aschedule==1.1.0

pip install -r requirements.txt

OPENAI_API_KEY=your_openai_api_key_here
AGENTOPS_API_KEY=your_agentops_api_key_here
TWITTER_API_KEY=your_twitter_api_key
TWITTER_API_SECRET=your_twitter_api_secret
TWITTER_ACCESS_TOKEN=your_twitter_access_token
TWITTER_ACCESS_TOKEN_SECRET=your_twitter_access_token_secret
TWITTER_USERNAME=your_twitter_username
FACEBOOK_ACCESS_TOKEN=your_facebook_access_token
FACEBOOK_PAGE_ID=your_facebook_page_id
INSTAGRAM_USERNAME=your_instagram_username
INSTAGRAM_PASSWORD=your_instagram_password
LINKEDIN_EMAIL=your_linkedin_email
LINKEDIN_PASSWORD=your_linkedin_password
LINKEDIN_COMPANY_ID=your_linkedin_company_id

import os
import json
import time
from datetime import datetime, timedelta
import random
import threading
import agentops
import schedule
import tweepy
from facebook import GraphAPI
from instagram_private_api import Client as InstagramClient
from linkedin_api import Linkedin
from openai import OpenAI

class SocialMediaAgent:
def init(self, openai_client):
self.openai_client = openai_client
self.industry = "technology" # Set your industry here
self.last_trends_check = None
self.current_trends = []
self.fine_tuned_model = None
self.recent_posts = []
    # Set up API clients
    self.twitter_client = self.setup_twitter_client()
    self.facebook_client = self.setup_facebook_client()
    self.instagram_client = self.setup_instagram_client()
    self.linkedin_client = self.setup_linkedin_client()

    self.scheduled_posts = []
    self.scheduler_thread = threading.Thread(target=self.run_scheduler)
    self.scheduler_thread.start()

def setup_twitter_client(self):
    auth = tweepy.OAuthHandler(os.getenv('TWITTER_API_KEY'), os.getenv('TWITTER_API_SECRET'))
    auth.set_access_token(os.getenv('TWITTER_ACCESS_TOKEN'), os.getenv('TWITTER_ACCESS_TOKEN_SECRET'))
    return tweepy.API(auth)

def setup_facebook_client(self):
    return GraphAPI(os.getenv('FACEBOOK_ACCESS_TOKEN'))

def setup_instagram_client(self):
    return InstagramClient(os.getenv('INSTAGRAM_USERNAME'), os.getenv('INSTAGRAM_PASSWORD'))

def setup_linkedin_client(self):
    return Linkedin(os.getenv('LINKEDIN_EMAIL'), os.getenv('LINKEDIN_PASSWORD'))

def run_scheduler(self):
    while True:
        schedule.run_pending()
        time.sleep(1)

def collect_training_data(self):
    training_data = []

    # Collect tweets
    tweets = self.twitter_client.user_timeline(screen_name=os.getenv('TWITTER_USERNAME'), count=500)
    for tweet in tweets:
        if tweet.created_at < datetime(2023, 1, 1):  # Adjust this date as needed
            training_data.append(tweet.text)

    # Collect Facebook posts
    facebook_posts = self.facebook_client.get_connections(os.getenv('FACEBOOK_PAGE_ID'), 'posts')
    for post in facebook_posts['data']:
        if 'message' in post and post['created_time'] < '2023-01-01T00:00:00+0000':
            training_data.append(post['message'])

    # Collect LinkedIn posts
    linkedin_posts = self.linkedin_client.get_company_updates(os.getenv('LINKEDIN_COMPANY_ID'))
    for post in linkedin_posts:
        if 'content' in post and 'title' in post['content'] and post['created']['time'] < 1672531200000:
            training_data.append(post['content']['title'])

    return training_data

def fine_tune_model(self, training_data):
    prepared_data = [{"prompt": "Create a social media post:", "completion": post} for post in training_data]

    with open('training_data.jsonl', 'w') as f:
        for entry in prepared_data:
            json.dump(entry, f)
            f.write('\n')

    file = self.openai_client.File.create(file=open('training_data.jsonl', 'rb'), purpose='fine-tune')
    fine_tune_job = self.openai_client.FineTune.create(training_file=file.id, model="davinci")

    while fine_tune_job.status != "succeeded":
        fine_tune_job = self.openai_client.FineTune.retrieve(id=fine_tune_job.id)
        print(f"Fine-tuning status: {fine_tune_job.status}")
        time.sleep(60)

    return fine_tune_job.fine_tuned_model

@agentops.record_function('monitor_trends')
def monitor_trends(self):
    if self.last_trends_check is None or datetime.now() - self.last_trends_check > timedelta(hours=6):
        prompt = f"List 5 current trending topics in the {self.industry} industry, separated by commas."
        response = self.openai_client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "user", "content": prompt}]
        )
        self.current_trends = [trend.strip() for trend in response.choices[0].message.content.split(',')]
        self.last_trends_check = datetime.now()

    return self.current_trends

@agentops.record_function('generate_content_ideas')
def generate_content_ideas(self, trends):
    ideas = []
    for trend in trends:
        prompt = f"Generate a content idea for a social media post about '{trend}' in the {self.industry} industry. Include a suggested hashtag."
        response = self.openai_client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "user", "content": prompt}]
        )
        ideas.append(response.choices[0].message.content)
    return ideas

@agentops.record_function('create_timely_post')
def create_timely_post(self, content_idea):
    prompt = f"Create a compelling social media post based on this idea: {content_idea}. The post should be engaging and no more than 280 characters."
    response = self.openai_client.chat.completions.create(
        model=self.fine_tuned_model,
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

@agentops.record_function('schedule_post')
def schedule_post(self, post, platform, scheduled_time):
    job = schedule.every().day.at(scheduled_time.strftime("%H:%M")).do(
        self.post_to_social_media, post, platform
    )
    self.scheduled_posts.append({
        "content": post,
        "platform": platform,
        "scheduled_time": scheduled_time,
        "job": job
    })
    return f"Post scheduled for {platform} at {scheduled_time}"

@agentops.record_function('post_to_social_media')
def post_to_social_media(self, post, platform):
    if platform.lower() == 'twitter':
        return self.post_to_twitter(post)
    elif platform.lower() == 'facebook':
        return self.post_to_facebook(post)
    elif platform.lower() == 'instagram':
        return self.post_to_instagram(post)
    elif platform.lower() == 'linkedin':
        return self.post_to_linkedin(post)
    else:
        raise ValueError(f"Unsupported platform: {platform}")

def post_to_twitter(self, post):
    try:
        tweet = self.twitter_client.update_status(post)
        return f"Posted to Twitter. Tweet ID: {tweet.id}"
    except Exception as e:
        return f"Failed to post to Twitter: {str(e)}"

def post_to_facebook(self, post):
    try:
        page_id = os.getenv('FACEBOOK_PAGE_ID')
        result = self.facebook_client.put_object(page_id, "feed", message=post)
        return f"Posted to Facebook. Post ID: {result['id']}"
    except Exception as e:
        return f"Failed to post to Facebook: {str(e)}"

def post_to_instagram(self, post):
    try:  # Note: This is a simplified version. In reality, you'd need to create an image.
        result = self.instagram_client.post_photo(post, caption=post)
        return f"Posted to Instagram. Media ID: {result['media']['id']}"
    except Exception as e:
        return f"Failed to post to Instagram: {str(e)}"

def post_to_linkedin(self, post):
    try:
        company_id = os.getenv('LINKEDIN_COMPANY_ID')
        result = self.linkedin_client.create_company_share(company_id, post)
        return f"Posted to LinkedIn. Activity ID: {result['activity_id']}"
    except Exception as e:
        return f"Failed to post to LinkedIn: {str(e)}"

@agentops.record_function('run_content_cycle')
def run_content_cycle(self):
    trends = self.monitor_trends()
    content_ideas = self.generate_content_ideas(trends)
    posts = []
    for idea in content_ideas:
        post = self.create_timely_post(idea)
        for platform in ["Twitter", "Facebook", "Instagram", "LinkedIn"]:
            if self.should_post_immediately(platform):
                result = self.post_to_social_media(post, platform)
                posts.append({"content": post, "platform": platform, "result": result, "type": "immediate"})
            else:
                scheduled_time = self.get_optimal_posting_time(platform)
                result = self.schedule_post(post, platform, scheduled_time)
                posts.append({"content": post, "platform": platform, "result": result, "type": "scheduled"})
        self.recent_posts.append(post)
    return posts

def should_post_immediately(self, platform):
    # This is a placeholder. Implement your own logic here.
    return random.choice([True, False])

def get_optimal_posting_time(self, platform):
    # This is a placeholder. Implement your own logic here.
    return datetime.now() + timedelta(hours=random.randint(1, 24))

def get_scheduled_posts(self):
    return [{"content": post["content"],
             "platform": post["platform"],
             "scheduled_time": post["scheduled_time"]}
            for post in self.scheduled_posts]

def cancel_scheduled_post(self, index):
    if 0 <= index < len(self.scheduled_posts):
        post = self.scheduled_posts.pop(index)
        schedule.cancel_job(post["job"])
        return f"Cancelled scheduled post for {post['platform']} at {post['scheduled_time']}"
    else:
        return "Invalid post index"

def perform_style_check(self, recent_posts):
    prompt = "Analyze the following recent posts and determine if they match our brand voice and style:\n\n"
    prompt += "\n".join(recent_posts)
    prompt += "\n\nDo these posts match our established style? Provide a yes/no answer and a brief explanation."
    response = self.openai_client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}]
    )

    return response.choices[0].message.content

def get_recent_posts(self, count=10):
    return self.recent_posts[-count:]

import time
import atexit
from openai import OpenAI
import agentops
import os
from dotenv import load_dotenv
from social_media_agent import SocialMediaAgent

load_dotenv()

openai = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
agentops.init(os.getenv("AGENTOPS_API_KEY"))

def main():
    agent = SocialMediaAgent(openai)
   
    if agent.fine_tuned_model is None:
        training_data = agent.collect_training_data()
        agent.fine_tuned_model = agent.fine_tune_model(training_data)
   
    style_check_interval = 7 * 24 * 3600  # 7 days
    last_style_check = time.time()

    def cleanup():
        print("Shutting down scheduler...")
        agent.scheduler_thread.join(timeout=1)
   
    atexit.register(cleanup)

    while True:
        try:
            agent.run_content_cycle()
           
            if time.time() - last_style_check > style_check_interval:
                recent_posts = agent.get_recent_posts()
                style_analysis = agent.perform_style_check(recent_posts)
                print(f"Style Check Results: {style_analysis}")
                last_style_check = time.time()
           
            time.sleep(3600)  # Wait an hour before the next cycle
        except Exception as e:
            print(f"An error occurred: {e}")
            agentops.end_session("Fail")
       
        print(f"Cycle completed. Sleeping for 1 hour.")

if __name__ == "__main__":
    main()

FROM python:3.9-slim
WORKDIR /app
COPY . /app
RUN pip install --no-cache-dir -r requirements.txt
EXPOSE 80
ENV NAME SocialMediaAgent
CMD ["python", "main.py"]

docker build -t social-media-agent

docker run -d --name social-media-agent --env-file .env social-media-agent