What is an AI Content Agency?
An AI Content Agency is a software system that uses autonomous AI agents to handle the entire content production pipeline — from research and strategy through writing, editing, and publishing — with minimal human intervention. Instead of hiring a team of writers, editors, and project managers, you deploy a swarm of specialized agents that collaborate to produce high-quality content at scale.
Each agent is an independent unit with its own prompt context, tool access, and decision-making capabilities. They communicate via structured message passing, share a common knowledge base, and can be orchestrated by a central coordinator or operate in a decentralized fashion. The result is a system that can handle hundreds of client accounts simultaneously, where a traditional agency would need dozens of human staff members.
Think of it as a content factory built on LLMs, vector databases, and workflow automation — capable of producing blog posts, social media content, email newsletters, product descriptions, and technical documentation across multiple industries and brand voices.
Why It Matters
The Economics of Content Production
A traditional content agency might serve 10–15 clients with a team of 5 writers, 2 editors, and 1 project manager. Salaries, benefits, sick days, turnover, and training costs eat into margins. An AI-powered agency collapses this ratio: one developer can build a system that serves 150+ clients with better turnaround times and consistent quality.
24/7 Operation Without Burnout
AI agents don't sleep, don't need weekends off, and don't experience creative fatigue. When a client in a different time zone requests urgent content at 2 AM, your agent system picks it up immediately, processes it, and delivers by morning — no overtime pay required.
Consistency Across Clients
Human writers bring inconsistent voice, varying quality, and subjective interpretation of briefs. Agents enforce brand guidelines programmatically. Every piece of content adheres to the same structural rules, tone specifications, and quality checks, eliminating the variance that plagues human teams.
Data-Driven Optimization
Agents can integrate directly with analytics platforms, SEO tools, and social media APIs. They don't just create content — they analyze performance metrics, identify what's working, and automatically adjust future content strategies. This closed feedback loop is impossible to achieve manually at scale.
Core Architecture: Agent Types and Their Responsibilities
The Strategic Planner Agent
This agent analyzes the client's industry, competitors, and audience to build content strategies. It connects to SEO APIs, scrapes competitor sites, and produces structured content calendars. It doesn't write — it plans what should be written and why.
The Research Agent
Given a topic, the Research Agent scours internal knowledge bases, performs web searches via tools like Tavily or SerpAPI, and compiles factual briefs. It outputs structured research documents with citations, statistics, and source URLs that downstream agents can reference.
The Writer Agent
The Writer takes a content brief and research document as input and produces the actual draft. It follows strict formatting rules, adheres to brand voice guidelines stored in a vector database, and can write in multiple formats: long-form articles, social media captions, ad copy, or email sequences.
The Editor Agent
This agent performs grammar correction, fact-checking against the research document, tone alignment, and structural improvements. It acts as a quality gate — if the content doesn't pass its checks, it sends revision notes back to the Writer Agent for a rewrite cycle.
The Media Agent
Responsible for sourcing or generating images, creating simple graphics, and ensuring proper alt-text and metadata. It can query stock image APIs, generate images via DALL-E or Stable Diffusion, or pull from the client's media library.
The Publishing Agent
Handles the final step: formatting content for the target platform (WordPress, Shopify, Medium, social media schedulers), uploading via REST APIs, scheduling posts, and logging the publication in the system's audit trail.
Building Your First AI Content Agent System
Setting Up the Environment
We'll build a modular system using Python, LangChain for agent orchestration, and a vector database (ChromaDB) for storing client brand guidelines and past content. Each agent is a class with a defined interface.
# requirements.txt
langchain==0.3.0
langchain-openai==0.2.0
chromadb==0.5.0
pydantic==2.8.0
tavily-python==0.5.0
httpx==0.27.0
asyncio==3.4.3
The Base Agent Class
Every agent in our system extends a common base class that provides shared functionality: structured output parsing, retry logic, and a consistent interface for receiving tasks and returning results.
from pydantic import BaseModel, Field
from typing import Optional, List, Dict, Any
import asyncio
from datetime import datetime
import uuid
class AgentTask(BaseModel):
"""A unit of work passed between agents."""
task_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
task_type: str
client_id: str
payload: Dict[str, Any]
parent_task_id: Optional[str] = None
created_at: datetime = Field(default_factory=datetime.utcnow)
class AgentResult(BaseModel):
"""Structured output from any agent."""
task_id: str
success: bool
output: Any
metadata: Dict[str, Any] = Field(default_factory=dict)
errors: List[str] = Field(default_factory=list)
tokens_used: int = 0
cost: float = 0.0
class BaseAgent:
"""Foundation for all content agents."""
def __init__(self, name: str, model: str = "gpt-4o-mini"):
self.name = name
self.model = model
self.retry_limit = 3
async def process(self, task: AgentTask) -> AgentResult:
"""Override in subclass with specific logic."""
raise NotImplementedError
async def process_with_retry(self, task: AgentTask) -> AgentResult:
for attempt in range(self.retry_limit):
try:
result = await self.process(task)
if result.success:
return result
# Log failure and retry
print(f"[{self.name}] Attempt {attempt+1} failed: {result.errors}")
except Exception as e:
print(f"[{self.name}] Exception on attempt {attempt+1}: {str(e)}")
await asyncio.sleep(2 ** attempt) # Exponential backoff
return AgentResult(
task_id=task.task_id,
success=False,
output=None,
errors=[f"Failed after {self.retry_limit} attempts"]
)
The Research Agent Implementation
The Research Agent takes a topic and client context, performs web searches, and compiles a structured research brief. It uses the Tavily search API for real-time web access.
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from tavily import TavilyClient
import json
class ResearchAgent(BaseAgent):
"""Gathers facts, statistics, and sources for content creation."""
def __init__(self, tavily_api_key: str, **kwargs):
super().__init__(name="ResearchAgent", **kwargs)
self.tavily = TavilyClient(api_key=tavily_api_key)
self.llm = ChatOpenAI(model=self.model, temperature=0.3)
self.synthesis_prompt = ChatPromptTemplate.from_messages([
("system", """You are an expert research analyst. Given raw search results
and a content topic, synthesize a structured research brief containing:
1. Key facts and statistics (with source URLs)
2. Competing viewpoints or perspectives
3. Notable quotes or expert opinions
4. Recommended structure for the content piece
5. Any important caveats or context
Format your response as valid JSON with these exact keys:
"facts", "perspectives", "quotes", "structure", "caveats"
Each value should be a list of strings."""),
("human", """Topic: {topic}
Client Industry: {industry}
Target Audience: {audience}
Raw Search Results: {search_results}""")
])
async def perform_search(self, query: str, depth: int = 3) -> List[Dict]:
"""Execute multi-angle searches for comprehensive coverage."""
search_angles = [
f"{query} latest statistics data",
f"{query} expert analysis trends",
f"{query} challenges problems issues",
f"{query} best practices guide",
]
all_results = []
for angle in search_angles[:depth]:
response = self.tavily.search(query=angle, search_depth="advanced")
all_results.extend(response.get("results", []))
return all_results
async def process(self, task: AgentTask) -> AgentResult:
topic = task.payload.get("topic")
client_context = task.payload.get("client_context", {})
if not topic:
return AgentResult(
task_id=task.task_id,
success=False,
output=None,
errors=["Missing 'topic' in payload"]
)
try:
# Step 1: Perform web research
raw_results = await self.perform_search(
query=topic,
depth=task.payload.get("research_depth", 3)
)
# Step 2: Synthesize into structured brief
formatted_results = json.dumps([
{"title": r.get("title"), "content": r.get("content"),
"url": r.get("url")}
for r in raw_results[:15]
], indent=2)
response = await self.llm.ainvoke(
self.synthesis_prompt.format_messages(
topic=topic,
industry=client_context.get("industry", "General"),
audience=client_context.get("target_audience", "General"),
search_results=formatted_results
)
)
# Parse the structured brief
content = response.content
if "json" in content:
content = content.split("json")[1].split("")[0]
elif "" in content:
content = content.split("")[1].split("")[0]
brief = json.loads(content.strip())
return AgentResult(
task_id=task.task_id,
success=True,
output={
"research_brief": brief,
"source_urls": [r.get("url") for r in raw_results[:15]
if r.get("url")],
"raw_search_count": len(raw_results)
},
metadata={"tokens_used": response.response_metadata.get(
"token_usage", {}).get("total_tokens", 0)}
)
except json.JSONDecodeError as e:
return AgentResult(
task_id=task.task_id,
success=False,
output=None,
errors=[f"Failed to parse research brief JSON: {str(e)}"]
)
except Exception as e:
return AgentResult(
task_id=task.task_id,
success=False,
output=None,
errors=[str(e)]
)
The Writer Agent Implementation
The Writer Agent consumes a research brief and produces the actual draft. It retrieves client brand voice guidelines from a vector store and applies them during generation.
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
import chromadb
from chromadb.utils import embedding_functions
class WriterAgent(BaseAgent):
"""Produces content drafts based on research briefs and brand guidelines."""
def __init__(self, chroma_client: chromadb.Client, **kwargs):
super().__init__(name="WriterAgent", **kwargs)
self.llm = ChatOpenAI(model=self.model, temperature=0.7)
self.chroma = chroma_client
self.default_embedding = embedding_functions.OpenAIEmbeddingFunction(
api_key=kwargs.get("openai_api_key", ""),
model_name="text-embedding-3-small"
)
self.writing_prompt = ChatPromptTemplate.from_messages([
("system", """You are a professional content writer. Your task is to
write a complete, polished draft based on the provided research brief.
CRITICAL RULES:
- Write in the exact brand voice described below
- Follow the content structure from the research brief
- Use specific facts and statistics from the brief (cite them inline)
- Maintain engaging, human-sounding prose — do NOT sound robotic
- Include a compelling headline and introduction
- End with a conclusion or call-to-action as appropriate
BRAND VOICE GUIDELINES:
{brand_voice}
FORMAT REQUIREMENTS:
{format_requirements}"""),
("human", """Content Type: {content_type}
Topic: {topic}
Target Audience: {audience}
Desired Word Count: {word_count}
RESEARCH BRIEF:
{research_brief}
ADDITIONAL INSTRUCTIONS:
{additional_instructions}""")
])
async def retrieve_brand_voice(self, client_id: str) -> str:
"""Fetch client brand guidelines from vector store."""
collection = self.chroma.get_collection(
f"brand_guidelines_{client_id}",
embedding_function=self.default_embedding
)
results = collection.query(
query_texts=["brand voice tone style guidelines"],
n_results=3
)
if results and results.get("documents"):
return "\n".join([doc for doc in results["documents"][0]])
return "Professional, clear, and conversational tone. Avoid jargon."
async def retrieve_format_requirements(self, content_type: str,
client_id: str) -> str:
"""Fetch formatting rules for the content type."""
collection = self.chroma.get_collection(
f"format_templates_{client_id}",
embedding_function=self.default_embedding
)
results = collection.query(
query_texts=[f"{content_type} format template structure"],
n_results=1
)
if results and results.get("documents"):
return results["documents"][0][0]
return "Standard article format: headline, introduction, body sections, conclusion"
async def process(self, task: AgentTask) -> AgentResult:
required_fields = ["topic", "research_brief", "content_type", "client_id"]
for field in required_fields:
if field not in task.payload:
return AgentResult(
task_id=task.task_id,
success=False,
output=None,
errors=[f"Missing required field: {field}"]
)
try:
# Retrieve client-specific guidelines
client_id = task.payload["client_id"]
brand_voice = await self.retrieve_brand_voice(client_id)
format_reqs = await self.retrieve_format_requirements(
task.payload["content_type"], client_id
)
# Format research brief for prompt
brief = task.payload["research_brief"]
if isinstance(brief, dict):
brief_str = json.dumps(brief, indent=2)
else:
brief_str = str(brief)
# Invoke LLM
response = await self.llm.ainvoke(
self.writing_prompt.format_messages(
brand_voice=brand_voice,
format_requirements=format_reqs,
content_type=task.payload["content_type"],
topic=task.payload["topic"],
audience=task.payload.get("target_audience", "General"),
word_count=task.payload.get("word_count", 1200),
research_brief=brief_str,
additional_instructions=task.payload.get(
"additional_instructions", "None"
)
)
)
draft = response.content
# Extract headline (first line or H1)
lines = draft.strip().split("\n")
headline = lines[0].lstrip("#").strip() if lines else "Untitled"
return AgentResult(
task_id=task.task_id,
success=True,
output={
"draft": draft,
"headline": headline,
"word_count_estimate": len(draft.split()),
"brand_voice_used": brand_voice[:100] + "..."
},
metadata={
"tokens_used": response.response_metadata.get(
"token_usage", {}).get("total_tokens", 0)
}
)
except Exception as e:
return AgentResult(
task_id=task.task_id,
success=False,
output=None,
errors=[f"Writer agent error: {str(e)}"]
)
The Editor Agent Implementation
The Editor Agent validates drafts against the original brief, checks for factual accuracy, and enforces quality standards. It can either approve or return revision requests.
class EditorAgent(BaseAgent):
"""Reviews content and provides actionable feedback or approval."""
def __init__(self, **kwargs):
super().__init__(name="EditorAgent", **kwargs)
self.llm = ChatOpenAI(model=self.model, temperature=0.1)
self.review_prompt = ChatPromptTemplate.from_messages([
("system", """You are a meticulous content editor. Review the draft
against the research brief and quality standards. You must output
a valid JSON object with EXACTLY these fields:
{
"approved": true/false,
"overall_score": 1-10,
"grammar_score": 1-10,
"factual_accuracy": 1-10,
"tone_alignment": 1-10,
"structure_score": 1-10,
"issues_found": ["list of specific issues"],
"revision_instructions": "detailed revision notes if not approved",
"strengths": ["what works well"]
}
QUALITY THRESHOLDS:
- overall_score >= 7 required for approval
- factual_accuracy >= 8 required (no hallucinated facts)
- grammar_score >= 7 required
- If any threshold is not met, set approved=false and provide
clear revision_instructions"""),
("human", """ORIGINAL RESEARCH BRIEF:
{research_brief}
DRAFT TO REVIEW:
{draft}
EXPECTED WORD COUNT: {expected_word_count}
CONTENT TYPE: {content_type}
CLIENT INDUSTRY: {industry}""")
])
async def process(self, task: AgentTask) -> AgentResult:
draft = task.payload.get("draft")
research_brief = task.payload.get("research_brief")
if not draft or not research_brief:
return AgentResult(
task_id=task.task_id,
success=False,
output=None,
errors=["Missing draft or research_brief in payload"]
)
try:
brief_str = json.dumps(research_brief, indent=2) if isinstance(
research_brief, dict) else str(research_brief)
response = await self.llm.ainvoke(
self.review_prompt.format_messages(
research_brief=brief_str,
draft=draft,
expected_word_count=task.payload.get("word_count", 1200),
content_type=task.payload.get("content_type", "article"),
industry=task.payload.get("industry", "General")
)
)
content = response.content
if "json" in content:
content = content.split("json")[1].split("")[0]
elif "" in content:
content = content.split("")[1].split("")[0]
review = json.loads(content.strip())
return AgentResult(
task_id=task.task_id,
success=True,
output=review,
metadata={
"approved": review.get("approved", False),
"overall_score": review.get("overall_score", 0),
"tokens_used": response.response_metadata.get(
"token_usage", {}).get("total_tokens", 0)
}
)
except json.JSONDecodeError:
return AgentResult(
task_id=task.task_id,
success=False,
output={"approved": False, "revision_instructions":
"Editor output could not be parsed."},
errors=["JSON parse error in editor response"]
)
The Publishing Agent Implementation
The Publishing Agent formats content for target platforms and pushes it live via platform APIs. We'll implement WordPress publishing as an example.
import httpx
from base64 import b64encode
class PublishingAgent(BaseAgent):
"""Publishes content to target platforms via their APIs."""
def __init__(self, **kwargs):
super().__init__(name="PublishingAgent", **kwargs)
# Platform API credentials stored per-client
self.platform_credentials: Dict[str, Dict] = {}
def register_client_platform(
self, client_id: str, platform: str, credentials: Dict[str, str]
):
"""Store API credentials for a client's platform."""
key = f"{client_id}:{platform}"
self.platform_credentials[key] = credentials
async def publish_wordpress(
self, client_id: str, content: Dict[str, Any]
) -> Dict[str, Any]:
"""Publish a post to WordPress REST API."""
creds = self.platform_credentials.get(f"{client_id}:wordpress")
if not creds:
raise ValueError(f"No WordPress credentials for client {client_id}")
site_url = creds["site_url"]
username = creds["username"]
app_password = creds["app_password"]
# WordPress API auth
auth_string = b64encode(
f"{username}:{app_password}".encode()
).decode()
headers = {
"Authorization": f"Basic {auth_string}",
"Content-Type": "application/json"
}
# Build WordPress post payload
post_payload = {
"title": content.get("headline", "Untitled"),
"content": content.get("html_body", content.get("draft", "")),
"status": content.get("status", "draft"),
"categories": content.get("categories", []),
"tags": content.get("tags", []),
"meta": {
"seo_title": content.get("seo_title", ""),
"seo_description": content.get("seo_description", ""),
}
}
# Handle featured image if provided
if content.get("featured_media_id"):
post_payload["featured_media"] = content["featured_media_id"]
async with httpx.AsyncClient() as client:
response = await client.post(
f"{site_url}/wp-json/wp/v2/posts",
headers=headers,
json=post_payload,
timeout=30.0
)
if response.status_code in (200, 201):
data = response.json()
return {
"published": True,
"post_id": data.get("id"),
"post_url": data.get("link"),
"status": data.get("status"),
"response_code": response.status_code
}
else:
return {
"published": False,
"error": f"HTTP {response.status_code}: {response.text[:300]}",
"response_code": response.status_code
}
async def process(self, task: AgentTask) -> AgentResult:
platform = task.payload.get("platform", "wordpress")
client_id = task.payload.get("client_id")
content = task.payload.get("content", {})
try:
if platform == "wordpress":
result = await self.publish_wordpress(client_id, content)
else:
return AgentResult(
task_id=task.task_id,
success=False,
output=None,
errors=[f"Unsupported platform: {platform}"]
)
return AgentResult(
task_id=task.task_id,
success=result.get("published", False),
output=result,
metadata={"platform": platform, "client_id": client_id}
)
except Exception as e:
return AgentResult(
task_id=task.task_id,
success=False,
output=None,
errors=[f"Publishing error: {str(e)}"]
)
Orchestrating Multiple Agents: The Workflow Engine
Individual agents are powerful, but the real magic comes from chaining them together in a robust workflow. The Orchestrator manages task routing, handles revision loops (Writer ↔ Editor), and tracks the state of every content job across all clients.
from enum import Enum
from typing import Callable
import asyncio
from collections import defaultdict
class ContentJobStatus(Enum):
QUEUED = "queued"
RESEARCHING = "researching"
WRITING = "writing"
EDITING = "editing"
REVISING = "revising"
MEDIA_SOURCING = "media_sourcing"
PUBLISHING = "publishing"
COMPLETED = "completed"
FAILED = "failed"
class ContentJob:
"""Tracks a single content piece through the pipeline."""
def __init__(self, job_id: str, client_id: str, topic: str,
content_type: str, platform: str):
self.job_id = job_id
self.client_id = client_id
self.topic = topic
self.content_type = content_type
self.platform = platform
self.status = ContentJobStatus.QUEUED
self.research_brief = None
self.draft = None
self.final_content = None
self.revision_count = 0
self.max_revisions = 3
self.errors: List[str] = []
self.created_at = datetime.utcnow()
self.completed_at = None
class Orchestrator:
"""Central coordinator that routes tasks between agents."""
def __init__(
self,
research_agent: ResearchAgent,
writer_agent: WriterAgent,
editor_agent: EditorAgent,
publishing_agent: PublishingAgent
):
self.research = research_agent
self.writer = writer_agent
self.editor = editor_agent
self.publisher = publishing_agent
self.jobs: Dict[str, ContentJob] = {}
self.active_tasks: Dict[str, asyncio.Task] = {}
self.completed_queue: List[ContentJob] = []
self.max_concurrent = 10
self._semaphore = asyncio.Semaphore(self.max_concurrent)
async def submit_job(self, job: ContentJob) -> str:
"""Submit a new content job and begin processing."""
self.jobs[job.job_id] = job
task = asyncio.create_task(self._process_job(job))
self.active_tasks[job.job_id] = task
return job.job_id
async def _process_job(self, job: ContentJob):
"""Full pipeline: Research → Write → Edit → (Revise) → Publish."""
async with self._semaphore:
try:
# Phase 1: Research
job.status = ContentJobStatus.RESEARCHING
research_task = AgentTask(
task_type="research",
client_id=job.client_id,
payload={
"topic": job.topic,
"content_type": job.content_type,
"research_depth": 3,
"client_context": await self._get_client_context(
job.client_id)
}
)
research_result = await self.research.process_with_retry(
research_task)
if not research_result.success:
job.status = ContentJobStatus.FAILED
job.errors = research_result.errors
return
job.research_brief = research_result.output["research_brief"]
# Phase 2: Write
job.status = ContentJobStatus.WRITING
write_task = AgentTask(
task_type="write",
client_id=job.client_id,
payload={
"topic": job.topic,
"content_type": job.content_type,
"research_brief": job.research_brief,
"target_audience": "General",
"word_count": 1200,
"additional_instructions": ""
}
)
write_result = await self.writer.process_with_retry(write_task)
if not write_result.success:
job.status = ContentJobStatus.FAILED
job.errors = write_result.errors
return
job.draft = write_result.output["draft"]
# Phase 3: Edit with revision loop
while job.revision_count <= job.max_revisions:
job.status = ContentJobStatus.EDITING if job.revision_count == 0 \
else ContentJobStatus.REVISING
edit_task = AgentTask(
task_type="edit",
client_id=job.client_id,
payload={
"draft": job.draft,
"research_brief": job.research_brief,
"word_count": 1200,
"content_type": job.content_type,
"industry": "Technology"
}
)
edit_result = await self.editor.process_with_retry(edit_task)
if not edit_result.success:
job.status = ContentJobStatus.FAILED
job.errors = edit_result.errors
return
review = edit_result.output
if review.get("approved"):
job.final_content = job.draft
break
else:
# Revision cycle
job.revision_count += 1
revision_instructions = review.get(
"revision_instructions", "Improve overall quality")
revise_task = AgentTask(
task_type="write",
client_id=job.client_id,
payload={
"topic": job.topic,
"content_type": job.content_type,
"research_brief": job.research_brief,
"target_audience": "General",
"word_count": 1200,
"additional_instructions":
f"REVISION INSTRUCTIONS: {revision_instructions}"
}
)
revise_result = await self.writer.process_with_retry(
revise_task)
if revise_result.success:
job.draft = revise_result.output["draft"]
else:
job.errors.append(
f"Revision {job.revision_count} failed")
if not job.final_content:
job.status = ContentJobStatus.FAILED
job.errors.append(
f"Max revisions ({job.max_revisions}) exceeded")
return
# Phase 4: Publish
job.status = ContentJobStatus.PUBLISHING
publish_task = AgentTask(
task_type="publish",
client_id=job.client_id,
payload={
"platform": job.platform,
"content": {
"headline": "Extracted Headline",
"html_body": job.final_content,
"status": "publish",
"categories": [],
"tags": []
}
}
)
publish_result = await self.publisher.process_with_retry(
publish_task)
if publish_result.success:
job.status = ContentJobStatus.COMPLETED
job.completed_at = datetime.utcnow()
self.completed_queue.append(job)
else:
job.status = ContentJobStatus.FAILED
job.errors.append(
f"Publish failed: {publish_result.errors}")
except Exception as e:
job.status = ContentJobStatus.FAILED
job.errors.append(f"Pipeline error: {str(e)}")
async def _get_client_context(self, client_id: str) -> Dict[str, Any]:
"""Fetch client metadata for research context."""
# In production, this queries your client database
return {
"industry": "Technology",
"target_audience": "Software developers",
"brand_name": "ExampleCorp"
}
async def submit_batch(self, jobs: List[ContentJob]) -> List[str]:
"""Submit multiple jobs for concurrent processing."""
job_ids = []
for job in jobs:
job_id = await self.submit_job(job)
job_ids.append(job_id)
return job_ids
def get_job_status(self, job_id: str) -> Optional[Dict[str, Any]]:
"""Retrieve current status of any job."""
job = self.jobs.get(job_id)
if not job:
return None
return {
"job_id": job.job_id,
"client_id": job.client_id,
"topic": job.topic,
"status": job.status.value,
"revision_count": job.revision_count,
"errors": job.errors,
"created_at": job.created_at.isoformat(),
"completed_at": job.completed_at.isoformat() if job.completed_at else None
}
Scaling to 10x Client Capacity
Parallel Processing Architecture
The key to serving 150+ clients instead of 15 is eliminating bottlenecks. Our Orchestrator already uses asyncio.Semaphore to control concurrency, but in production you'll want to distribute work across multiple worker processes or machines. Here's how to scale the architecture:
import multiprocessing
from concurrent.futures import ProcessPoolExecutor, as_completed
import redis
import pickle
class DistributedOrchestrator:
"""Scales content processing across multiple workers using Redis queue."""
def __init__(self, redis_url: str, num_workers: int = 4):
self.redis = redis.from_url(redis_url)
self.num_workers = num_workers
self.job_queue_key = "content_jobs:queue"
self.result_queue_key = "content_jobs:results"
async def enqueue_job(self, job: ContentJob):
"""Push a job to the Redis queue for distributed processing."""
serialized = pickle.dumps(job)
self.redis.lpush(self.job_queue_key, serialized)
def worker_process(self, worker_id: int):
"""Runs in a separate process, pulling jobs from Redis."""
# Each worker needs its own agent instances
research = ResearchAgent(tavily_api_key=os.environ["TAVILY_KEY"])
writer = WriterAgent(chroma_client=chromadb.PersistentClient(
path=f"/tmp/worker_{worker_id}_chroma"))
editor = EditorAgent()
publisher = PublishingAgent()
orchestrator = Orchestrator(research, writer, editor, publisher)
local_redis = redis.from_url(os.environ["REDIS_URL"])
print(f"Worker {worker