21. Multi-Agent Content Team

A content team deployment coordinates specialized agents to produce high-quality written content. This architecture demonstrates orchestration patterns for creative work: research, drafting, review, and revision cycles managed by multiple agents with distinct responsibilities.

Team Structure

The content team consists of a Research Agent that gathers and synthesizes information, a Drafting Agent that transforms research into structured content, an Editor Agent that reviews for quality and consistency, and a Publishing Agent that formats and finalizes output.

Workflow Orchestration

The orchestrator manages the content pipeline: research → draft → review → revision → publish. Each stage may require human approval checkpoints depending on content risk levels.

# content_team/orchestrator.py
from dataclasses import dataclass, field
from typing import Optional
from enum import Enum

class ContentStage(Enum):
    RESEARCH = "research"
    DRAFT = "draft"
    REVIEW = "review"
    REVISION = "revision"
    PUBLISH = "publish"

@dataclass
class ContentTask:
    task_id: str
    topic: str
    target_audience: str
    word_count_target: int
    current_stage: ContentStage = ContentStage.RESEARCH
    research_context: Optional[dict] = None
    draft_content: Optional[str] = None
    review_notes: Optional[str] = None
    revision_count: int = 0

class ContentTeamOrchestrator:
    def __init__(
        self,
        research_agent: any,
        drafting_agent: any,
        editor_agent: any,
        publishing_agent: any
    ):
        self.agents = {
            ContentStage.RESEARCH: research_agent,
            ContentStage.DRAFT: drafting_agent,
            ContentStage.REVIEW: editor_agent,
            ContentStage.PUBLISH: publishing_agent
        }
        self.max_revisions = 3
    
    async def process_content_request(self, task: ContentTask) -> dict:
        while task.current_stage != ContentStage.PUBLISH:
            agent = self.agents[task.current_stage]
            task = await self._execute_stage(agent, task)
            
            if task.current_stage == ContentStage.REVIEW:
                if self._needs_revision(task):
                    task = self._transition_to_revision(task)
                else:
                    task = self._transition_to_publish(task)
        
        return {"task_id": task.task_id, "content": task.draft_content}
    
    async def _execute_stage(self, agent, task: ContentTask) -> ContentTask:
        context = self._build_context(task)
        result = await agent.process(context)
        return self._apply_result(task, result)
    
    def _build_context(self, task: ContentTask) -> dict:
        return {
            "topic": task.topic,
            "audience": task.target_audience,
            "word_count": task.word_count_target,
            "research": task.research_context,
            "draft": task.draft_content,
            "review_notes": task.review_notes
        }
    
    def _needs_revision(self, task: ContentTask) -> bool:
        return task.review_notes and len(task.review_notes) > 0
    
    def _transition_to_revision(self, task: ContentTask) -> ContentTask:
        task.revision_count += 1
        task.current_stage = ContentStage.REVISION
        return task
    
    def _transition_to_publish(self, task: ContentTask) -> ContentTask:
        task.current_stage = ContentStage.PUBLISH
        return task
    
    def _apply_result(self, task: ContentTask, result: dict) -> ContentTask:
        if ContentStage.RESEARCH:
            task.research_context = result.get("research")
            task.current_stage = ContentStage.DRAFT
        elif ContentStage.DRAFT:
            task.draft_content = result.get("draft")
            task.current_stage = ContentStage.REVIEW
        elif ContentStage.REVIEW:
            task.review_notes = result.get("notes")
            task.current_stage = ContentStage.REVIEW
        return task

Quality Gates

Human reviewers approve content before publication. Automated quality scoring complements human review, flagging content that exceeds acceptable plagiarism scores, factuality issues, or brand voice violations.