AI Content Generation Agents: Blog Posts, Social Media, and Email Marketing at Scale

The Content Production Problem

Marketing teams face a relentless demand for content across channels: blog posts for SEO, social media for engagement, and email campaigns for nurturing leads. An AI content generation agent does not replace writers — it amplifies them by handling first drafts, format adaptation, and channel-specific optimization at scale. The key is maintaining consistent brand voice while automating the repetitive parts of the content pipeline.

Brand Voice Configuration

Every piece of content should sound like it came from your brand. We encode brand voice as a structured specification that gets injected into every generation prompt.

flowchart LR
    USERS(["Traffic"])
    LB["Geo LB plus<br/>Anycast"]
    EDGE["Edge cache plus<br/>rate limit"]
    APP["Stateless app pods<br/>HPA on QPS"]
    QUEUE[(Async work queue)]
    WORKER["Worker pool<br/>GPU or CPU"]
    CACHE[("Redis cache<br/>LLM responses")]
    DB[("Read replicas<br/>and primary")]
    OBS[(Observability)]
    USERS --> LB --> EDGE --> APP
    APP --> CACHE
    APP --> QUEUE --> WORKER
    APP --> DB
    APP --> OBS
    style LB fill:#4f46e5,stroke:#4338ca,color:#fff
    style WORKER fill:#ede9fe,stroke:#7c3aed,color:#1e1b4b
    style CACHE fill:#f59e0b,stroke:#d97706,color:#1f2937
    style OBS fill:#0ea5e9,stroke:#0369a1,color:#fff

from dataclasses import dataclass, field

@dataclass
class BrandVoice:
    company_name: str
    tone: str  # e.g., "professional but approachable"
    audience: str  # e.g., "B2B SaaS decision makers"
    vocabulary_use: list[str] = field(default_factory=list)  # preferred terms
    vocabulary_avoid: list[str] = field(default_factory=list)  # banned terms
    style_rules: list[str] = field(default_factory=list)

    def to_prompt_block(self) -> str:
        rules = "\n".join(f"- {r}" for r in self.style_rules)
        use = ", ".join(self.vocabulary_use)
        avoid = ", ".join(self.vocabulary_avoid)
        return f"""Brand: {self.company_name}
Tone: {self.tone}
Target audience: {self.audience}
Preferred terms: {use}
Terms to avoid: {avoid}
Style rules:
{rules}"""

voice = BrandVoice(
    company_name="Acme AI",
    tone="professional but approachable, data-driven",
    audience="B2B SaaS CTOs and engineering leaders",
    vocabulary_use=["AI agents", "automation", "workflow"],
    vocabulary_avoid=["synergy", "disrupt", "leverage", "utilize"],
    style_rules=[
        "Use active voice",
        "Keep sentences under 25 words",
        "Include specific numbers and examples",
        "Avoid buzzwords and jargon",
    ],
)

Blog Post Generation with SEO Optimization

Blog posts need more than good writing — they need keyword targeting, proper heading structure, and meta descriptions that drive click-through from search results.

from openai import AsyncOpenAI
from dataclasses import dataclass

client = AsyncOpenAI()

@dataclass
class BlogBrief:
    topic: str
    target_keyword: str
    secondary_keywords: list[str]
    word_count: int = 1200
    target_audience: str = ""

BLOG_PROMPT = """Write a blog post based on this brief.

{brand_voice}

Topic: {topic}
Primary keyword: {keyword} (use naturally 3-5 times)
Secondary keywords: {secondary} (use each at least once)
Target length: {word_count} words

Structure requirements:
- H1 title that includes the primary keyword
- Meta description (under 155 characters)
- 4-6 H2 sections with descriptive headings
- Opening paragraph that hooks the reader with a specific stat or question
- Closing section with clear next steps or CTA

Return the post in markdown format with a YAML frontmatter block
containing title, meta_description, and keywords.
"""

async def generate_blog_post(
    brief: BlogBrief, brand: BrandVoice
) -> str:
    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "user",
            "content": BLOG_PROMPT.format(
                brand_voice=brand.to_prompt_block(),
                topic=brief.topic,
                keyword=brief.target_keyword,
                secondary=", ".join(brief.secondary_keywords),
                word_count=brief.word_count,
            ),
        }],
        max_tokens=4000,
    )
    return response.choices[0].message.content

Social Media Adaptation

A single blog post can generate multiple social media posts across platforms. Each platform has different character limits, tone expectations, and formatting conventions.

SOCIAL_PROMPT = """Adapt this blog post into {platform} posts.

{brand_voice}

Blog content:
{blog_excerpt}

Platform rules:
{platform_rules}

Generate {count} variations. Each should:
- Stand alone (reader has NOT read the blog)
- Include a hook in the first line
- End with a call to action or question
- Include relevant hashtags for {platform}

Return as a JSON array of strings.
"""

PLATFORM_RULES = {
    "linkedin": "Professional tone. 150-300 words. Use line breaks for readability.",
    "twitter": "Under 280 characters. Punchy and direct. 2-3 hashtags max.",
    "instagram": "Conversational. 100-200 words. 5-10 hashtags at the end.",
}

async def adapt_for_social(
    blog_content: str, platform: str, brand: BrandVoice, count: int = 3
) -> list[str]:
    import json
    response = await client.chat.completions.create(
        model="gpt-4o",
        response_format={"type": "json_object"},
        messages=[
            {"role": "system", "content": "Return a JSON object with key 'posts' containing an array of strings."},
            {
                "role": "user",
                "content": SOCIAL_PROMPT.format(
                    platform=platform,
                    brand_voice=brand.to_prompt_block(),
                    blog_excerpt=blog_content[:2000],
                    platform_rules=PLATFORM_RULES.get(platform, ""),
                    count=count,
                ),
            },
        ],
    )
    result = json.loads(response.choices[0].message.content)
    return result["posts"]

Email Campaign Generation

Email marketing requires subject line optimization, preview text, and segmented body content. The agent generates multiple subject line variants for A/B testing.

@dataclass
class EmailBrief:
    campaign_goal: str  # e.g., "webinar registration"
    audience_segment: str
    key_message: str
    cta_text: str
    cta_url: str

async def generate_email_campaign(
    brief: EmailBrief, brand: BrandVoice
) -> dict:
    import json
    prompt = f"""Generate an email campaign.

{brand.to_prompt_block()}

Goal: {brief.campaign_goal}
Segment: {brief.audience_segment}
Key message: {brief.key_message}
CTA: {brief.cta_text} -> {brief.cta_url}

Return JSON with:
- "subject_lines": array of 3 subject line variants (under 50 chars each)
- "preview_text": under 90 characters
- "body_html": email body in simple HTML
- "body_plain": plain text version
"""
    response = await client.chat.completions.create(
        model="gpt-4o",
        response_format={"type": "json_object"},
        messages=[
            {"role": "system", "content": "Return valid JSON only."},
            {"role": "user", "content": prompt},
        ],
    )
    return json.loads(response.choices[0].message.content)

Approval Workflow

No AI-generated content should go live without human review. An approval queue lets editors review, edit, and approve content before publication.

from enum import Enum
from datetime import datetime

class ContentStatus(Enum):
    DRAFT = "draft"
    PENDING_REVIEW = "pending_review"
    APPROVED = "approved"
    REJECTED = "rejected"
    PUBLISHED = "published"

class ApprovalQueue:
    def __init__(self, db_pool):
        self.pool = db_pool

    async def submit_for_review(
        self, content: str, content_type: str, metadata: dict
    ) -> str:
        async with self.pool.acquire() as conn:
            row = await conn.fetchrow(
                """INSERT INTO content_queue
                   (content, content_type, metadata, status, created_at)
                   VALUES ($1, $2, $3, $4, $5)
                   RETURNING id""",
                content, content_type,
                json.dumps(metadata),
                ContentStatus.PENDING_REVIEW.value,
                datetime.utcnow(),
            )
            return str(row["id"])

    async def approve(self, content_id: str, reviewer: str) -> None:
        async with self.pool.acquire() as conn:
            await conn.execute(
                """UPDATE content_queue
                   SET status = $1, reviewed_by = $2, reviewed_at = $3
                   WHERE id = $4""",
                ContentStatus.APPROVED.value,
                reviewer, datetime.utcnow(), content_id,
            )

FAQ

How do I prevent AI-generated content from sounding generic?

Feed the LLM specific examples of your best-performing content as few-shot examples in the prompt. Include concrete data points, customer quotes, and industry-specific terminology in your briefs. Generic output usually results from generic input — the more specific context you provide, the more distinctive the output.

What is the best way to handle SEO keyword density without stuffing?

Specify exact keyword usage counts in the prompt (e.g., "use the primary keyword 3-5 times naturally") and run a post-generation check that counts keyword occurrences. If the count falls outside the target range, regenerate or use a follow-up prompt asking the LLM to adjust specific paragraphs.

How do I maintain consistency across hundreds of generated posts?

Use the brand voice configuration as a constant system prompt across all generation calls. Maintain a style guide document that the LLM references. Run a brand voice compliance check on each piece of generated content by asking a separate LLM call to score adherence to your style rules and flag deviations.

---

#ContentGeneration #MarketingAI #SEO #BrandVoice #Python #AgenticAI #LearnAI #AIEngineering