Project: Multi-Agent Content Pipeline

Project: Content Pipeline Agent

This final project builds a multi-stage content pipeline agent that takes a topic or URL, conducts research, writes a long-form article, creates social media variations, and produces a complete content package — ready to publish.

What We're Building

A content pipeline that:

1. Accepts a topic (or URL to base content on)
2. Researches the topic with web search
3. Creates a detailed content outline
4. Writes the full article (with SEO in mind)
5. Generates social media variations (LinkedIn, Twitter/X, email newsletter)
6. Saves everything to a structured output

Architecture

Topic Input
    ↓
[Research Node] — Web search for current information
    ↓
[Outline Node] — Create detailed article structure
    ↓
[Write Article Node] — Draft the full article
    ↓
[Parallel Social Content]
    ├── LinkedIn Post
    ├── Twitter Thread
    └── Email Newsletter Snippet
    ↓
Output Package (folder with all files)

Complete Implementation

# content_pipeline.py
import os
import json
import time
from typing import TypedDict, Annotated
import operator
from pathlib import Path
from datetime import datetime
from langchain_openai import ChatOpenAI
from langchain_community.tools import TavilySearchResults
from langchain_core.messages import SystemMessage, HumanMessage
from langgraph.graph import StateGraph, END
from langgraph.types import Send
from dotenv import load_dotenv

load_dotenv()

llm = ChatOpenAI(model="gpt-4o", temperature=0.7)
precise_llm = ChatOpenAI(model="gpt-4o", temperature=0)
search = TavilySearchResults(max_results=5, search_depth="advanced", include_answer=True)

# === STATE ===

class ContentState(TypedDict):
    topic: str
    audience: str
    tone: str
    target_keyword: str
    research: str
    sources: list[str]
    outline: str
    article: str
    social_content: Annotated[dict, lambda x, y: {**x, **y}]  # Merge dicts
    metadata: dict

# === PIPELINE NODES ===

def research_topic(state: ContentState) -> ContentState:
    """Research the topic using web search."""
    print(f"\n🔍 Researching: {state['topic']}")
    
    # Multiple targeted searches
    queries = [
        state["topic"],
        f"{state['topic']} statistics 2024 2025",
        f"{state['topic']} expert insights trends",
    ]
    
    all_results = []
    all_sources = []
    
    for query in queries:
        results = search.invoke(query)
        if isinstance(results, list):
            for r in results:
                if isinstance(r, dict):
                    if r.get("content"):
                        all_results.append(f"[{r.get('url', '')}]\n{r['content']}")
                    if r.get("url"):
                        all_sources.append(r["url"])
    
    raw_research = "\n\n---\n\n".join(all_results[:8])  # Limit context size
    
    # Synthesize research
    synthesis = precise_llm.invoke([
        SystemMessage(content="You synthesize web research into structured notes for content creation."),
        HumanMessage(content=f"""Topic: {state['topic']}
Target audience: {state['audience']}
Target keyword: {state['target_keyword']}

Web search results:
{raw_research[:6000]}

Create concise research notes:
1. Key facts, statistics, and data points (with approximate sources)
2. Current trends and recent developments
3. Expert perspectives or common views
4. Common questions or misconceptions about this topic
5. What's unique or surprising about this topic

Format as structured bullet points. Be specific — include numbers and examples.""")
    ])
    
    print(f"   ✓ Research complete ({len(all_sources)} sources)")
    
    return {
        **state,
        "research": synthesis.content,
        "sources": list(set(all_sources))[:15]
    }

def create_outline(state: ContentState) -> ContentState:
    """Create a detailed content outline."""
    print("\n📋 Creating outline...")
    
    outline = precise_llm.invoke([
        SystemMessage(content="You create detailed, SEO-optimized article outlines for content writers."),
        HumanMessage(content=f"""Create a detailed outline for a long-form article.

Topic: {state['topic']}
Target keyword: {state['target_keyword']}
Audience: {state['audience']}
Tone: {state['tone']}

Research notes:
{state['research']}

Create a complete article outline with:
- Compelling title with the target keyword
- Meta description (155 characters, includes keyword)
- Introduction hook idea
- 4-6 main sections with H2 headings (include keyword naturally in 1-2)
- 2-3 subsections (H3) per main section
- Key points to cover in each section (based on research)
- Conclusion approach
- CTA suggestion

Format as clean markdown with the structure above.""")
    ])
    
    print("   ✓ Outline created")
    return {**state, "outline": outline.content}

def write_article(state: ContentState) -> ContentState:
    """Write the full article from the outline."""
    print("\n✍️  Writing article (this takes a moment)...")
    
    article = llm.invoke([
        SystemMessage(content=f"""You are a professional content writer. Write engaging, informative articles for {state['audience']}.
Tone: {state['tone']}
Write in a way that is natural, not AI-sounding. Use specific examples and data."""),
        HumanMessage(content=f"""Write a complete long-form article based on this outline.

Target keyword: {state['target_keyword']} (use naturally 4-6 times)
Target length: 1200-1800 words

Outline:
{state['outline']}

Research to draw from:
{state['research']}

Requirements:
- Start with a compelling hook (stat, question, or bold statement)
- Use subheadings to break up the content
- Include specific examples, data points from the research
- Write in active voice
- Short paragraphs (3-4 sentences max)
- Include a strong conclusion with key takeaways
- Add a sources section at the end

Write the complete article now:""")
    ])
    
    print(f"   ✓ Article written ({len(article.content.split())} words)")
    return {**state, "article": article.content}

# === PARALLEL SOCIAL CONTENT NODES ===

def write_linkedin_post(state: dict) -> dict:
    """Write a LinkedIn post from the article."""
    print("   📱 Writing LinkedIn post...")
    
    post = llm.invoke([
        SystemMessage(content="You write high-performing LinkedIn posts that drive engagement."),
        HumanMessage(content=f"""Write a LinkedIn post promoting this article.

Article topic: {state['topic']}
Target audience: {state['audience']}

Article summary:
{state['article'][:1500]}

LinkedIn post requirements:
- 150-250 words
- Start with a hook that stops the scroll (stat, bold claim, or compelling question)
- 3-4 key insights from the article
- End with a question to drive comments
- Include the article premise clearly
- 5-7 relevant hashtags at the end
- Professional but conversational tone

Write the post now:""")
    ])
    
    return {"social_content": {"linkedin": post.content}}

def write_twitter_thread(state: dict) -> dict:
    """Write a Twitter/X thread from the article."""
    print("   🐦 Writing Twitter thread...")
    
    thread = llm.invoke([
        SystemMessage(content="You write viral Twitter/X threads that get shared and discussed."),
        HumanMessage(content=f"""Write a Twitter/X thread about this article.

Topic: {state['topic']}

Article summary:
{state['article'][:1500]}

Thread requirements:
- 6-8 tweets
- Tweet 1: Strong hook (the most surprising fact or bold statement)
- Tweets 2-6: One insight per tweet, each valuable standalone
- Tweet 7: "The thread:" + link placeholder
- Tweet 8 (optional): Call to action

Format each tweet as:
[1/7] tweet text here
[2/7] tweet text here

Keep each tweet under 260 characters.
Make each tweet standalone — useful even without reading the others.""")
    ])
    
    return {"social_content": {"twitter_thread": thread.content}}

def write_email_snippet(state: dict) -> dict:
    """Write an email newsletter snippet."""
    print("   📧 Writing email snippet...")
    
    email = llm.invoke([
        SystemMessage(content="You write compelling email newsletter snippets that drive clicks."),
        HumanMessage(content=f"""Write an email newsletter snippet for this article.

Topic: {state['topic']}
Audience: {state['audience']}

Article:
{state['article'][:1000]}

Email snippet requirements:
- Subject line (under 60 characters, creates curiosity)
- Preview text (40 characters, complements subject)
- Opening (1 sentence hook)
- 3-4 sentence teaser (key insight without giving everything away)
- "Read the full article →" CTA

Format:
Subject: [subject line]
Preview: [preview text]

[email body]

[Read the full article →]""")
    ])
    
    return {"social_content": {"email_snippet": email.content}}

# === ROUTING ===

def route_to_social_writers(state: ContentState):
    """Send article to all social writers in parallel."""
    return [
        Send("write_linkedin", dict(state)),
        Send("write_twitter", dict(state)),
        Send("write_email", dict(state)),
    ]

# === BUILD GRAPH ===

def build_content_pipeline():
    graph = StateGraph(ContentState)
    
    graph.add_node("research", research_topic)
    graph.add_node("outline", create_outline)
    graph.add_node("write_article", write_article)
    graph.add_node("write_linkedin", write_linkedin_post)
    graph.add_node("write_twitter", write_twitter_thread)
    graph.add_node("write_email", write_email_snippet)
    
    graph.set_entry_point("research")
    graph.add_edge("research", "outline")
    graph.add_edge("outline", "write_article")
    
    # After article, fan out to social writers in parallel
    graph.add_conditional_edges("write_article", route_to_social_writers, 
                                 ["write_linkedin", "write_twitter", "write_email"])
    
    graph.add_edge("write_linkedin", END)
    graph.add_edge("write_twitter", END)
    graph.add_edge("write_email", END)
    
    return graph.compile()

# === SAVE OUTPUT ===

def save_content_package(state: ContentState, output_dir: str):
    """Save all content to an organized folder."""
    output_path = Path(output_dir)
    output_path.mkdir(parents=True, exist_ok=True)
    
    slug = state["topic"].lower().replace(" ", "-")[:40]
    timestamp = datetime.now().strftime("%Y%m%d")
    
    files_saved = []
    
    # Save article
    article_file = output_path / f"{timestamp}-{slug}.md"
    with open(article_file, "w", encoding="utf-8") as f:
        f.write(f"# {state['topic']}\n\n")
        f.write(f"**Target keyword:** {state['target_keyword']}\n")
        f.write(f"**Audience:** {state['audience']}\n\n")
        f.write("---\n\n")
        f.write(state["article"])
        f.write("\n\n---\n\n## Sources\n")
        for src in state["sources"]:
            f.write(f"- {src}\n")
    files_saved.append(str(article_file))
    
    # Save social content
    social_file = output_path / f"{timestamp}-{slug}-social.md"
    with open(social_file, "w", encoding="utf-8") as f:
        social = state.get("social_content", {})
        f.write(f"# Social Content: {state['topic']}\n\n")
        
        if "linkedin" in social:
            f.write("## LinkedIn Post\n\n")
            f.write(social["linkedin"])
            f.write("\n\n---\n\n")
        
        if "twitter_thread" in social:
            f.write("## Twitter/X Thread\n\n")
            f.write(social["twitter_thread"])
            f.write("\n\n---\n\n")
        
        if "email_snippet" in social:
            f.write("## Email Newsletter Snippet\n\n")
            f.write(social["email_snippet"])
    files_saved.append(str(social_file))
    
    return files_saved

# === MAIN ===

def run_content_pipeline(
    topic: str,
    audience: str = "business professionals and marketers",
    tone: str = "informative and practical",
    target_keyword: str = None,
    output_dir: str = "./content_output"
) -> dict:
    
    if not target_keyword:
        target_keyword = topic.lower()
    
    pipeline = build_content_pipeline()
    
    print(f"\n{'='*60}")
    print(f"🚀 Content Pipeline Starting")
    print(f"   Topic: {topic}")
    print(f"   Audience: {audience}")
    print(f"   Keyword: {target_keyword}")
    print(f"{'='*60}")
    
    start_time = time.time()
    
    result = pipeline.invoke({
        "topic": topic,
        "audience": audience,
        "tone": tone,
        "target_keyword": target_keyword,
        "research": "",
        "sources": [],
        "outline": "",
        "article": "",
        "social_content": {},
        "metadata": {}
    })
    
    # Save everything
    saved_files = save_content_package(result, output_dir)
    
    duration = time.time() - start_time
    print(f"\n{'='*60}")
    print(f"✅ Content pipeline complete in {duration:.0f}s")
    print(f"   Article: ~{len(result['article'].split())} words")
    print(f"   Sources: {len(result['sources'])}")
    print(f"   Social pieces: {len(result.get('social_content', {}))}")
    print(f"   Files saved:")
    for f in saved_files:
        print(f"     📄 {f}")
    print(f"{'='*60}")
    
    return result


if __name__ == "__main__":
    result = run_content_pipeline(
        topic="How AI agents are changing software development workflows",
        audience="software developers and engineering managers",
        tone="practical and insightful, with concrete examples",
        target_keyword="AI agents software development",
        output_dir="./content_output"
    )

Running the Pipeline

export OPENAI_API_KEY="sk-..."
export TAVILY_API_KEY="tvly-..."

python content_pipeline.py

The pipeline produces:

• content_output/20250526-how-ai-agents-are-changing.md — full article
• content_output/20250526-how-ai-agents-are-changing-social.md — all social versions

Customizing the Pipeline

Add an editor node: After writing, have another LLM pass check for clarity, flow, and brand voice.

Add image suggestions: Generate alt text and image search queries for each section.

Add SEO scoring: Run the article through a Surfer SEO-style check comparing against target keyword density.

Connect to a CMS: Add a final node that posts to WordPress, Webflow, or Contentful via API.

Schedule recurring content: Use a cron job to run the pipeline on a topic list automatically.

---

Congratulations — you've completed the AI Agent Development Course. You've built everything from basic LangChain chains to multi-agent parallel pipelines, with full evaluation, error handling, streaming, and deployment. You're equipped to build production-quality AI agents.