Introduction: What is a Content Writing Agent?
A Content Writing Agent is an AI-powered system that automates the end-to-end process of researching, outlining, drafting, reviewing, and refining written content. Unlike a simple prompt-and-response interaction with an LLM, a content writing agent orchestrates multiple steps—each handled by specialized components—to produce high-quality, structured, and polished articles, blog posts, reports, or marketing copy.
Built with LangGraph, this agent becomes a stateful, multi-step workflow where each node performs a distinct task (e.g., gathering research, generating an outline, writing a draft, reviewing for quality), and edges define how information flows between them. The graph-based architecture allows for conditional branching, loops for revision, and human-in-the-loop interventions, making it far more robust than a linear chain of prompts.
Why LangGraph for Content Writing Agents?
LangGraph provides a framework for building cyclic, stateful, and controllable LLM applications. For content writing, this matters because:
- Multi-step orchestration: Research → Outline → Draft → Review → Revise → Finalize is naturally a graph, not a single pipe.
- Conditional logic: You can route back to revision if the review finds issues, creating iterative improvement loops.
- Persistent state: The entire writing context (topic, research notes, outline, draft, feedback) travels through the graph as a typed dictionary, giving every node full visibility.
- Tool integration: Nodes can call external APIs (web search, fact-checking databases, grammar checkers) as part of their processing.
- Human-in-the-loop: You can pause execution at any node (like review) to get human approval before proceeding.
- Observability & debugging: LangGraph's built-in tracing lets you inspect exactly what each node produced and how the state evolved.
Core Architecture Overview
Our content writing agent will have the following graph structure:
- research_node: Gathers information about the topic using web search tools.
- outline_node: Creates a structured outline based on research.
- draft_node: Writes the full first draft from the outline and research.
- review_node: Evaluates the draft against quality criteria and produces feedback.
- revision_node: Revises the draft based on feedback.
- finalize_node: Polishes and formats the final output.
The graph flows linearly from research → outline → draft → review, then uses a conditional edge: if the review approves the draft, it proceeds to finalize; otherwise, it loops back to revision, which then feeds back to review for re-evaluation. This loop continues until quality thresholds are met or a maximum revision count is reached.
Setting Up the Environment
First, install the required dependencies. We'll use LangGraph, LangChain, an LLM provider (OpenAI), and a search tool (Tavily) for research.
# Install core dependencies
pip install langgraph langchain langchain-openai langchain-community tavily-python
# For environment variables (store your API keys)
pip install python-dotenvSet up your environment variables in a .env file:
OPENAI_API_KEY=your-openai-api-key-here
TAVILY_API_KEY=your-tavily-api-key-hereNow load them and initialize the LLM and tools:
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_community.tools.tavily_search import TavilySearchResults
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List, Optional, Literal
import operator
load_dotenv()
# Initialize the LLM — using GPT-4o for high-quality writing
llm = ChatOpenAI(
model="gpt-4o",
temperature=0.7,
max_tokens=4096
)
# Initialize a lower-temperature LLM for structured tasks (outline, review)
structured_llm = ChatOpenAI(
model="gpt-4o",
temperature=0.2,
max_tokens=2048
)
# Research tool — Tavily performs web searches
research_tool = TavilySearchResults(
max_results=5,
include_answer=True,
include_raw_content=False
)Defining the State and Data Models
The state is the central data structure that flows through every node. We define it as a TypedDict with fields for each piece of content and metadata the agent needs to track.
class ContentState(TypedDict):
"""State schema for the content writing agent."""
# Input
topic: str
target_audience: str
content_type: str # e.g., "blog_post", "technical_article", "tutorial"
tone: str # e.g., "professional", "conversational", "technical"
word_count_target: int
# Research phase
research_queries: Annotated[List[str], operator.add]
research_notes: Optional[str]
# Outline phase
outline: Optional[str]
# Draft phase
draft: Optional[str]
# Review phase
review_feedback: Optional[str]
review_score: Optional[float] # 1-10 quality score
passed_review: bool
# Revision tracking
revision_count: int
max_revisions: int
# Final output
final_content: Optional[str]
# Flow control
next_step: strThe Annotated[List[str], operator.add] annotation on research_queries tells LangGraph to append to this list rather than overwrite it when multiple nodes contribute to it. This is useful if we want to accumulate queries across iterations.
Building the Agent Nodes
1. Research Node
The research node takes the topic and generates targeted search queries, then executes them using Tavily and compiles the findings into structured research notes.
def research_node(state: ContentState) -> dict:
"""
Gathers research on the topic using web search.
Generates search queries, executes them, and compiles notes.
"""
topic = state["topic"]
audience = state["target_audience"]
content_type = state["content_type"]
# Step 1: Generate smart search queries based on the topic
query_gen_prompt = f"""You are a research planner. Given the topic below, generate
3-4 specific web search queries that would yield the most relevant, up-to-date
information for writing a {content_type} targeted at {audience}.
Topic: {topic}
Return ONLY the queries, one per line, no numbering or bullets."""
query_response = structured_llm.invoke(query_gen_prompt)
queries = [q.strip() for q in query_response.content.strip().split('\n') if q.strip()]
# Step 2: Execute searches
all_research = []
for query in queries:
try:
results = research_tool.invoke({"query": query})
# Format results into a readable summary
formatted = f"### Query: {query}\n"
if isinstance(results, list):
for r in results:
formatted += f"- {r.get('content', str(r))}\n"
else:
formatted += f"{results}\n"
all_research.append(formatted)
except Exception as e:
all_research.append(f"### Query: {query}\n- Error: {str(e)}")
research_notes_raw = "\n\n".join(all_research)
# Step 3: Synthesize research into structured notes
synthesis_prompt = f"""You are a research synthesizer. Based on the following raw
research results, create comprehensive, well-organized research notes for writing a
{content_type} about "{topic}" for {audience}.
Include:
- Key facts and statistics
- Different perspectives or viewpoints
- Recent developments or trends
- Practical examples or case studies
- Any controversies or nuances
Raw research:
{research_notes_raw}
Format the notes with clear headings and bullet points. Be thorough but concise."""
synthesis_response = structured_llm.invoke(synthesis_prompt)
research_notes = synthesis_response.content
return {
"research_queries": queries,
"research_notes": research_notes,
"next_step": "outline"
}2. Outline Node
The outline node creates a structured skeleton for the content, defining sections, subsections, and key points to cover.
def outline_node(state: ContentState) -> dict:
"""
Creates a detailed content outline based on research notes.
"""
topic = state["topic"]
research_notes = state["research_notes"]
content_type = state["content_type"]
audience = state["target_audience"]
word_count = state["word_count_target"]
outline_prompt = f"""You are an expert content strategist. Create a detailed outline
for a {content_type} about "{topic}" targeting {audience}.
Target word count: approximately {word_count} words.
Research notes available:
{research_notes}
Requirements for the outline:
1. Include a compelling title (and subtitle if appropriate)
2. List all major sections with H2 headings
3. Under each section, list H3 subheadings and bullet points of key content
4. Note where to include examples, data points, or quotes from research
5. Include an introduction section that hooks the reader
6. Include a conclusion section with actionable takeaways
7. Estimate word count allocation per section
Make the outline detailed enough that a writer could produce a full draft from it
without additional guidance."""
outline_response = structured_llm.invoke(outline_prompt)
outline = outline_response.content
return {
"outline": outline,
"next_step": "draft"
}3. Draft Node
The draft node writes the complete first draft, following the outline and incorporating research. This uses the higher-temperature LLM for more creative, fluent prose.
def draft_node(state: ContentState) -> dict:
"""
Writes the full first draft based on the outline and research notes.
"""
topic = state["topic"]
outline = state["outline"]
research_notes = state["research_notes"]
content_type = state["content_type"]
audience = state["target_audience"]
tone = state["tone"]
word_count = state["word_count_target"]
draft_prompt = f"""You are a professional content writer. Write a complete,
polished first draft for a {content_type} based on the outline and research below.
Topic: {topic}
Target audience: {audience}
Tone: {tone}
Target word count: approximately {word_count} words
OUTLINE:
{outline}
RESEARCH NOTES:
{research_notes}
Writing instructions:
1. Follow the outline exactly — use the same headings and structure
2. Write in full, flowing prose under each heading
3. Incorporate facts, examples, and data from the research naturally
4. Maintain the specified tone consistently throughout
5. Write a compelling introduction that hooks readers
6. Write a substantive conclusion with actionable takeaways
7. Use transitional phrases between sections for smooth flow
8. Aim for the target word count (±10% is acceptable)
9. Do NOT include meta-commentary like "as discussed above" — write as a published piece
10. Format with proper Markdown headings (##, ###)"""
draft_response = llm.invoke(draft_prompt)
draft = draft_response.content
return {
"draft": draft,
"next_step": "review"
}4. Review Node
The review node evaluates the draft against multiple quality criteria and produces structured feedback with a score. This is where we decide whether to proceed to finalization or loop back for revision.
def review_node(state: ContentState) -> dict:
"""
Reviews the draft against quality criteria and provides structured feedback.
Sets passed_review based on score threshold.
"""
draft = state["draft"]
topic = state["topic"]
content_type = state["content_type"]
audience = state["target_audience"]
word_count_target = state["word_count_target"]
revision_count = state.get("revision_count", 0)
max_revisions = state.get("max_revisions", 3)
review_prompt = f"""You are a strict content editor. Review the following draft
for a {content_type} about "{topic}" for {audience}. Target word count: {word_count_target}.
Evaluate the draft on these criteria, scoring each 1-10:
1. Structure & Organization: Does it follow a logical flow with clear headings?
2. Content Depth & Accuracy: Is the information substantive and factually sound?
3. Readability & Engagement: Is the prose clear, engaging, and well-paced?
4. Tone Consistency: Does the tone match the target audience and content type?
5. Completeness: Does it cover the topic adequately for the target word count?
6. Grammar & Mechanics: Are there any language errors?
For each criterion, provide:
- Score (1-10)
- Specific praise (what works well)
- Specific criticism (what needs improvement)
- Concrete suggestion for revision
After the individual criteria, provide:
- Overall score (average of the six scores, rounded to 1 decimal)
- Summary feedback paragraph
- A bullet list of the top 3-5 actionable revisions needed
DRAFT TO REVIEW:
{draft}
IMPORTANT: If the overall score is 8.0 or higher, the draft PASSES review and can proceed
to finalization. Otherwise, it needs revision. Revision count so far: {revision_count}
out of {max_revisions} maximum."""
review_response = structured_llm.invoke(review_prompt)
feedback = review_response.content
# Parse the overall score from the feedback
import re
score_match = re.search(r'Overall score[:\s]*([0-9]+\\.?[0-9]*)', feedback)
if score_match:
overall_score = float(score_match.group(1))
else:
# Fallback: look for any number near "score" in the last portion
scores = re.findall(r'[0-9]+\\.?[0-9]*', feedback)
if scores:
overall_score = float(scores[-1])
else:
overall_score = 5.0 # Conservative default
passed = overall_score >= 8.0 or revision_count >= max_revisions
return {
"review_feedback": feedback,
"review_score": overall_score,
"passed_review": passed,
"revision_count": revision_count,
"next_step": "finalize" if passed else "revise"
}5. Revision Node
The revision node takes the review feedback and rewrites the draft, addressing each piece of criticism. It increments the revision counter.
def revision_node(state: ContentState) -> dict:
"""
Revises the draft based on review feedback.
"""
draft = state["draft"]
feedback = state["review_feedback"]
topic = state["topic"]
content_type = state["content_type"]
audience = state["target_audience"]
tone = state["tone"]
outline = state["outline"]
revision_count = state.get("revision_count", 0)
revision_prompt = f"""You are a meticulous content reviser. Revise the draft below
based on the editor's feedback. Address EVERY actionable suggestion in the feedback.
Topic: {topic}
Content type: {content_type}
Target audience: {audience}
Tone: {tone}
ORIGINAL OUTLINE (maintain this structure):
{outline}
EDITOR FEEDBACK:
{feedback}
ORIGINAL DRAFT:
{draft}
Revision instructions:
1. Address every specific criticism and suggestion in the feedback
2. Maintain the original outline structure
3. Preserve content that was praised — only fix what needs fixing
4. Improve weak sections, add missing information, fix tone inconsistencies
5. Fix any grammar or mechanical issues noted
6. Produce the COMPLETE revised draft — not just the changed sections
7. Output the full revised draft with proper Markdown formatting"""
revised_response = llm.invoke(revision_prompt)
revised_draft = revised_response.content
new_revision_count = revision_count + 1
return {
"draft": revised_draft,
"revision_count": new_revision_count,
"next_step": "review"
}6. Finalize Node
The finalize node performs a final polish pass—fixing minor issues, adding a title if needed, formatting metadata, and producing the finished piece.
def finalize_node(state: ContentState) -> dict:
"""
Final polish and formatting of the approved draft.
"""
draft = state["draft"]
topic = state["topic"]
content_type = state["content_type"]
audience = state["target_audience"]
tone = state["tone"]
finalize_prompt = f"""You are a final polish editor. Take the approved draft below
and perform final polishing for a {content_type} about "{topic}" for {audience}.
Tasks:
1. Ensure the title is compelling and SEO-friendly (add or improve if needed)
2. Add a brief "summary" or "TL;DR" section right after the introduction if appropriate
3. Check that all headings use consistent formatting
4. Ensure smooth transitions between all sections
5. Do a final grammar and style pass
6. Add a subtle call-to-action or next-step suggestion in the conclusion if appropriate
7. Preserve ALL substantive content — only make surface-level improvements
Output the complete, finalized piece with proper Markdown formatting.
DRAFT TO POLISH:
{draft}"""
final_response = llm.invoke(finalize_prompt)
final_content = final_response.content
return {
"final_content": final_content,
"next_step": "end"
}Assembling the Graph
Now we wire all the nodes together into a LangGraph StateGraph. We define the nodes, edges, and a conditional routing function that decides whether to loop back for revision or proceed to finalization.
from langgraph.graph import StateGraph, END
def build_content_writing_graph() -> StateGraph:
"""
Constructs and compiles the content writing agent graph.
"""
# Create the graph with our state schema
workflow = StateGraph(ContentState)
# Add all nodes
workflow.add_node("research", research_node)
workflow.add_node("outline", outline_node)
workflow.add_node("draft", draft_node)
workflow.add_node("review", review_node)
workflow.add_node("revise", revision_node)
workflow.add_node("finalize", finalize_node)
# Define the entry point — where execution begins
workflow.set_entry_point("research")
# Add linear edges for the main pipeline
workflow.add_edge("research", "outline")
workflow.add_edge("outline", "draft")
workflow.add_edge("draft", "review")
# Conditional edge: after review, either finalize or revise
workflow.add_conditional_edges(
"review",
route_after_review,
{
"finalize": "finalize",
"revise": "revise"
}
)
# After revision, go back to review for re-evaluation
workflow.add_edge("revise", "review")
# After finalize, end the workflow
workflow.add_edge("finalize", END)
return workflow.compile()
def route_after_review(state: ContentState) -> Literal["finalize", "revise"]:
"""
Conditional routing function.
Returns 'finalize' if the draft passed review or max revisions reached,
otherwise returns 'revise' for another iteration.
"""
passed = state.get("passed_review", False)
revision_count = state.get("revision_count", 0)
max_revisions = state.get("max_revisions", 3)
if passed or revision_count >= max_revisions:
return "finalize"
return "revise"
# Compile the graph
content_agent = build_content_writing_graph()Running the Agent
With the graph compiled, we can invoke it with an initial state. The agent will execute the entire pipeline—researching, outlining, drafting, reviewing, potentially revising multiple times, and finalizing—all in one invocation.
# Define the initial state for a content writing task
initial_state = {
"topic": "Building Production-Ready RAG Systems with LangChain and Pinecone",
"target_audience": "Software developers and ML engineers with intermediate Python experience",
"content_type": "technical tutorial",
"tone": "technical but accessible, with practical code examples",
"word_count_target": 2500,
"research_queries": [],
"research_notes": "",
"outline": "",
"draft": "",
"review_feedback": "",
"review_score": 0.0,
"passed_review": False,
"revision_count": 0,
"max_revisions": 3,
"final_content": "",
"next_step": "research"
}
# Run the agent
print("🚀 Starting Content Writing Agent...")
print(f"Topic: {initial_state['topic']}")
print(f"Target: {initial_state['word_count_target']} words | Max revisions: {initial_state['max_revisions']}")
print("-" * 60)
final_state = content_agent.invoke(initial_state)
print("\n✅ Agent finished!")
print(f"Revisions performed: {final_state['revision_count']}")
print(f"Final review score: {final_state.get('review_score', 'N/A')}")
print(f"Final content length: {len(final_state.get('final_content', ''))} characters")
print("-" * 60)
print("\n📄 FINAL CONTENT:")
print(final_state.get('final_content', 'No content produced')[:500] + "...")
For more control, you can stream the execution step-by-step to observe each node's output:
# Stream execution to watch each step
print("Streaming agent execution...\n")
for step_output in content_agent.stream(initial_state):
# step_output is a dict like {"research": {...}} or {"outline": {...}}
node_name = list(step_output.keys())[0]
node_result = step_output[node_name]
print(f"\n{'='*40}")
print(f"📌 Node: {node_name}")
print(f"{'='*40}")
# Print a summary of what changed
for key, value in node_result.items():
if value and key not in ['next_step', 'passed_review']:
if isinstance(value, str) and len(value) > 200:
print(f" {key}: [updated — {len(value)} chars]")
elif isinstance(value, list):
print(f" {key}: [list with {len(value)} items]")
else:
print(f" {key}: {value}")
Adding Human-in-the-Loop Review
One of LangGraph's most powerful features is the ability to interrupt execution and wait for human input. Here's how to add an interrupt before the review node so a human editor can provide feedback manually:
# Compile with an interrupt point before the "review" node
content_agent_with_human = build_content_writing_graph()
content_agent_with_human = content_agent_with_human.compile(
interrupt_before=["review"]
)
# Run until the interrupt
print("Agent running until review...")
interrupted_state = content_agent_with_human.invoke(initial_state)
# At this point, execution pauses. You can inspect the draft:
print("\n📝 Draft produced. Here's a preview:")
print(interrupted_state['draft'][:800] + "...\n")
# Human provides feedback (simulated here, but could come from a UI)
human_feedback = """
The draft looks solid overall. However:
- The code examples in section 3 need more context
- The performance optimization section should mention caching strategies
- Add a troubleshooting subsection
Score: 7.5 — needs revision.
"""
# Update state with human feedback and resume
interrupted_state['review_feedback'] = human_feedback
interrupted_state['review_score'] = 7.5
interrupted_state['passed_review'] = False
interrupted_state['next_step'] = 'revise'
# Resume execution from the interrupt point
final_state = content_agent_with_human.invoke(
interrupted_state,
config={"thread_id": "human-review-session-1"}
)
print("\n✅ Resumed and completed with human feedback incorporated!")
print(f"Final content: {final_state.get('final_content', '')[:500]}...")
Best Practices for Content Writing Agents
1. Design Nodes with Single Responsibilities
Each node should do exactly one thing well. Resist the temptation to combine research and outlining into one node—separation gives you more control, better debugging, and the ability to swap out individual components later.
2. Use Temperature Strategically
Use lower temperature settings (0.0–0.3) for structured tasks like outline generation, review, and scoring. Use higher temperatures (0.5–0.8) for creative writing in the draft node. This prevents outlines from being too rigid while keeping evaluations consistent.
3. Implement Robust Score Parsing
LLMs can be inconsistent in how they format scores. Always include fallback parsing logic (regex with multiple patterns, default values) so a malformed review response doesn't crash the entire pipeline.
4. Set a Maximum Revision Cap
Always set max_revisions (3–5 is reasonable) to prevent infinite loops. Combine this with a forced finalization path when the cap is reached, even if the score is below threshold.
5. Preserve Context Across Revisions
The state carries the original outline and research notes through every revision. Always include these in the revision prompt so the reviser has full context and doesn't drift from the original structure.
6. Add Logging and Observability
Use LangGraph's built-in tracing or add explicit logging at each node to record: node name, timestamp, token usage, and a summary of state changes. This is invaluable for debugging and cost tracking.
import logging
import time
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def instrumented_node(node_func, node_name: str):
"""Wrapper that adds logging to any node function."""
def wrapped(state):
logger.info(f"[{node_name}] Starting...")
start = time.time()
result = node_func(state)
elapsed = time.time() - start
logger.info(f"[{node_name}] Completed in {elapsed:.2f}s")
# Log key state changes
for key in ['research_notes', 'outline', 'draft', 'review_feedback', 'final_content']:
if key in result and result[key]:
logger.info(f"[{node_name}] Produced {key}: {len(str(result[key]))} chars")
return result
return wrapped7. Handle Tool Failures Gracefully
External tools like web search APIs can fail, rate-limit, or return empty results. Always wrap tool calls in try/except blocks and provide fallback behavior (e.g., proceeding with whatever research was gathered, or using cached data).
8. Use Typed State for Clarity
Define your state as a TypedDict with explicit types rather than a plain dict. This catches type errors early, serves as living documentation, and helps IDEs provide autocompletion when writing node functions.
9. Test Each Node in Isolation
Before running the full graph, test each node function independently with mock state. This dramatically speeds up iteration and makes debugging tractable.
# Example: testing the outline node in isolation
test_state = {
"topic": "Test Topic",
"research_notes": "Some research notes here...",
"content_type": "blog_post",
"target_audience": "general readers",
"word_count_target": 1000,
"outline": "",
"next_step": "outline"
}
result = outline_node(test_state)
print(result["outline"][:500])
assert "introduction" in result["outline"].lower(), "Outline missing introduction"10. Consider Parallel Nodes Where Appropriate
If you need both web research and internal knowledge base search, you can run them as parallel nodes using LangGraph's fan-out patterns, then merge results with a join node. This cuts total execution time.
Conclusion
Building a content writing agent with LangGraph transforms what could be a brittle, single-prompt approach into a robust, iterative, and controllable writing pipeline. By decomposing the writing process into discrete nodes—research, outline, draft, review, revise, and finalize—you gain the ability to inspect, debug, and improve each stage independently. The graph-based architecture with conditional routing enables automatic revision loops that systematically improve content quality, while human-in-the-loop interrupts let you inject editorial judgment at critical decision points.
The patterns demonstrated here—typed state management, strategic temperature use, graceful tool error handling, revision caps, and isolated node testing—form a solid foundation that you can extend for domain-specific needs. Whether you're generating technical tutorials, marketing copy, or long-form reports, a LangGraph-powered content writing agent gives you the structure, observability, and control to produce consistently high-quality written content at scale.