AI Content Manager as Digital Worker
AI Content Manager automates content publication, moderation, and analytics: scheduled posting, adaptation across channels, comment responses, engagement reports. Manages content 24/7 without days off.
Agent Architecture
from openai import AsyncOpenAI
from datetime import datetime
import asyncio
client = AsyncOpenAI()
class ContentManagerAgent:
def __init__(self, channels: list[str], brand_context: dict):
self.channels = channels # ['telegram', 'vk', 'instagram', 'website']
self.brand = brand_context
self.scheduler = ContentScheduler()
self.publisher = MultiChannelPublisher()
self.moderator = CommentModerator()
async def run(self):
"""Main agent loop"""
tasks = [
self.process_content_queue(),
self.moderate_comments(),
self.generate_daily_report(),
]
await asyncio.gather(*tasks)
Content Adaptation for Channels
CHANNEL_FORMATS = {
"telegram": {"max_len": 4096, "supports_html": True, "emoji": True},
"vk": {"max_len": 16384, "supports_markup": True, "emoji": True},
"instagram": {"max_len": 2200, "hashtags": 30, "emoji": True},
"twitter_x": {"max_len": 280, "supports_threads": True},
"linkedin": {"max_len": 3000, "tone": "professional"},
}
async def adapt_content_for_channel(
original_content: str,
channel: str,
media_urls: list[str] = None
) -> dict:
fmt = CHANNEL_FORMATS[channel]
response = await client.chat.completions.create(
model="gpt-4o",
messages=[{
"role": "system",
"content": f"""Adapt content for {channel}.
Max characters: {fmt['max_len']}.
Tone: {fmt.get('tone', 'conversational')}.
{'Add relevant hashtags.' if fmt.get('hashtags') else ''}
{'Use emoji moderately.' if fmt.get('emoji') else 'No emoji.'}
Preserve meaning, change format."""
}, {
"role": "user",
"content": original_content
}]
)
return {
"channel": channel,
"text": response.choices[0].message.content,
"media": media_urls or []
}
Automatic Comment Moderation
async def moderate_comment(comment: str, post_context: str) -> dict:
response = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[{
"role": "system",
"content": """Moderate comment. Return JSON:
{
"action": "approve|delete|flag_review|auto_reply",
"reason": "...",
"auto_reply": "response text if action=auto_reply"
}
Delete: spam, ads, insults, profanity.
Auto-reply: product questions, thanks, complaints (initial response).
Review: borderline content, legal questions."""
}, {
"role": "user",
"content": f"Post context: {post_context[:200]}\nComment: {comment}"
}],
response_format={"type": "json_object"}
)
return json.loads(response.choices[0].message.content)
Cross-posting via APIs
class MultiChannelPublisher:
async def publish_telegram(self, text: str, media: list = None, channel_id: str = None):
import telegram
bot = telegram.Bot(token=TELEGRAM_TOKEN)
if media:
await bot.send_photo(chat_id=channel_id, photo=media[0], caption=text)
else:
await bot.send_message(chat_id=channel_id, text=text, parse_mode="HTML")
async def publish_vk(self, text: str, media: list = None, group_id: str = None):
import vk_api
vk = vk_api.VkApi(token=VK_TOKEN).get_api()
vk.wall.post(owner_id=f"-{group_id}", message=text)
Analytics and Reporting
async def generate_weekly_content_report(analytics: dict) -> str:
response = await client.chat.completions.create(
model="gpt-4o",
messages=[{
"role": "system",
"content": "Create weekly content manager report. Structure: top posts, reach/engagement, best formats, recommendations."
}, {
"role": "user",
"content": json.dumps(analytics, ensure_ascii=False)
}]
)
return response.choices[0].message.content
Timeline: basic agent with scheduled posting and 2–3 channel adaptation — 2–3 weeks. Full system with moderation, analytics, and integrations — 6–8 weeks.