Autonomous Technical Blog Writer using LangGraph: Multi-Agent Planning, Research & Multimodal Generation
This comprehensive study note explores the architectural design and step-by-step implementation of an enterprise-grade AI Blog Writing Agent powered by LangGraph. Moving beyond reactive single-prompt models, this system implements a structured Planning Agent Pattern capable of dynamic web research using Tavily, parallel multi-worker drafting via LangGraph's dynamic fan-out mechanism, and autonomous image generation using Google's Gemini API integrated into a dedicated multi-node reducer subgraph.
1. Conceptual Foundation: Reactive AI vs. Planning Agents
Standard Large Language Model (LLM) implementations operate reactively: upon receiving an input task, the model jumps directly into output generation. While effective for simple queries, this zero-shot approach degrades significantly when handling complex, structured content generation such as technical documentation or comprehensive blogs.
The Planning Agent Paradigm
A Planning Agent explicitly decouples understanding and design from execution. Instead of immediate generation, it operates in two distinct phases:
- Planning & Decomposition Phase: The agent analyzes the objective, evaluates constraints, queries missing domain knowledge, and breaks down the master objective into distinct, non-overlapping sub-tasks.
- Execution & Synthesis Phase: The agent delegates each sub-task to specialized worker nodes (operating sequentially or in parallel) and subsequently synthesizes their independent outputs into a unified final artifact.
Real-Life Analogy: Software Application Engineering
Consider instructing an AI to build a full-stack Expense Tracker Application:
- Reactive Approach: The LLM immediately starts generating source code in a single stream. It inevitably misses architectural edge cases, forgets database schema details, or fails to maintain modular software design.
- Planning Approach: The AI acts as a Solutions Architect first. It breaks the project into sub-tasks (Database Schema -> Auth Middleware -> REST Endpoints -> Frontend UI components). Once the blueprint is established, individual code generation tasks execute against defined interface specifications, yielding robust software.
2. High-Level System Architecture
The technical blog writer uses a stateful, directed graph built on LangGraph. The workflow dynamically adapts based on whether the topic requires external real-time web research or can be generated from the model's internal parametric knowledge base.
Complete End-to-End Workflow
3. Structural Schemas & State Management in LangGraph
To guarantee deterministic data flow between distinct nodes, the system heavily relies on Pydantic data models and a global TypedDict graph state.
1. Planning Models (Pydantic)
To prevent arbitrary unstructured responses, the Orchestrator node is constrained to return a structured Plan object containing granular Task objects.
from pydantic import BaseModel, Field
from typing import List, Optional
class Task(BaseModel):
id: int = Field(description="Unique numeric index for the section")
title: str = Field(description="Heading title of the section")
goal: str = Field(description="Core objective or key takeaway for the reader")
bullet_points: List[str] = Field(description="3 to 5 non-overlapping sub-topics to cover")
target_words: int = Field(description="Target word count for this section")
type: str = Field(description="Section type e.g., 'intro', 'core', 'code', 'conclusion'")
requires_research: bool = Field(default=False, description="Whether web research evidence is needed")
class Plan(BaseModel):
blog_title: str = Field(description="Catchy, SEO-optimized title for the blog post")
audience: str = Field(description="Target reader persona (e.g., Software Engineers)")
tone: str = Field(description="Writing style (e.g., technical, conversational)")
tasks: List[Task] = Field(description="Ordered list of section tasks")
2. Global Graph State (AgentState)
The state object persists data across the entire execution graph. Notice the use of operator.add as a reducer on the sections key to aggregate parallel outputs from worker nodes.
from typing import TypedDict, Annotated, List, Dict, Any
import operator
class AgentState(TypedDict):
topic: str
needs_research: bool
mode: str
queries: List[str]
evidence: List[Dict[str, Any]]
plan: Plan
sections: Annotated[List[str], operator.add] # Appends parallel worker outputs
merged_markdown: str
markdown_with_placeholders: str
image_specs: List[Dict[str, Any]]
final_markdown: str
4. Stage 1: Basic Orchestrator-Worker Implementation
The primary build begins with a core deterministic pipeline: Orchestrator -> Fan-Out -> Parallel Workers -> Reducer.
The Orchestrator (Planner) Node
The Orchestrator receives the user's topic and enforces structural constraints using with_structured_output(Plan).
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.2)
def orchestrator_node(state: AgentState) -> Dict[str, Any]:
planner_llm = llm.with_structured_output(Plan)
system_prompt = SystemMessage(content=(
"You are an expert technical blog planner. Break down the given topic "
"into 5 to 7 logical, sequential section tasks. Ensure each section has "
"clear, non-overlapping bullet points and realistic word count targets."
))
human_prompt = HumanMessage(content=f"Topic: {state['topic']}")
generated_plan = planner_llm.invoke([system_prompt, human_prompt])
return {"plan": generated_plan}
5. Dynamic Workflows: Fan-Out & Fan-In Parallel Execution
A core limitation of traditional chain workflows is serial execution. Writing 7 blog sections sequentially takes time:
By leveraging dynamic Fan-Out via LangGraph's Send API, section generation executes in parallel, reducing execution time to:
1. Dynamic Fan-Out Logic (Send API)
Instead of hardcoding graph edges, a conditional routing function reads the generated plan from the state and dispatches individual tasks to parallel instances of the worker_node.
from langgraph.types import Send
def fan_out_to_workers(state: AgentState):
plan = state["plan"]
topic = state["topic"]
# Return a list of Send objects targeting the 'worker' node dynamically
return [
Send(
"worker",
{
"task": task,
"topic": topic,
"plan": plan,
"evidence": state.get("evidence", [])
}
)
for task in plan.tasks
]
2. The Worker Node (Section Writer)
Each worker node operates in isolation, processing its assigned task while referencing the master plan for global context to avoid repetitive transitions.
def worker_node(payload: Dict[str, Any]) -> Dict[str, Any]:
task: Task = payload["task"]
topic: str = payload["topic"]
plan: Plan = payload["plan"]
system_prompt = SystemMessage(content=(
f"You are a technical writer drafting section {task.id}: '{task.title}'. "
f"Overall Blog Title: '{plan.blog_title}'. Target Audience: {plan.audience}. "
f"Tone: {plan.tone}. Cover these bullet points strictly: {task.bullet_points}. "
"Output ONLY raw, clean Markdown for this specific section."
))
human_prompt = HumanMessage(content=f"Draft section details: {task.goal}")
response = llm.invoke([system_prompt, human_prompt])
# Appends result as a single-element list (reduced via operator.add)
return {"sections": [response.content]}
6. Stage 2: Web Research Capabilities via Dynamic Routing & Tavily API
To write accurate technical articles about fast-evolving technologies, the agent must determine whether it needs real-time web knowledge or if it can rely on internal parametric memory.
1. Classification & Query Generation Schema
The Router Decision model classifies the topic and generates optimized search queries when research is required.
class RouterDecision(BaseModel):
needs_research: bool = Field(description="True if external current data is needed")
mode: str = Field(description="'closed_book', 'hybrid', or 'open_book'")
queries: List[str] = Field(default=[], description="3 to 5 targeted Tavily search queries")
def router_node(state: AgentState) -> Dict[str, Any]:
router_llm = llm.with_structured_output(RouterDecision)
system_prompt = SystemMessage(content=(
"Analyze the user topic. Decide if web search research is required.\n"
"- closed_book: Fundamental concepts (e.g., 'What is Self-Attention').\n"
"- hybrid: Known concepts needing recent updates/examples (e.g., 'Open Source LLMs in 2026').\n"
"- open_book: Highly temporal topics (e.g., 'Top AI News this week')."
))
decision = router_llm.invoke([system_prompt, HumanMessage(content=state["topic"])])
return {
"needs_research": decision.needs_research,
"mode": decision.mode,
"queries": decision.queries
}
2. Intelligent Routing Matrix
7. Research Node & Tavily API Integration
When needs_research is True, the execution moves to the Research Node. This node uses the Tavily Search API to collect external evidence, which it cleans and formats into a structured EvidencePack.
from langchain_community.tools.tavily_search import TavilySearchResults
class EvidenceItem(BaseModel):
title: str
url: str
published_date: Optional[str] = None
snippet: str
class EvidencePack(BaseModel):
items: List[EvidenceItem]
def research_node(state: AgentState) -> Dict[str, Any]:
tavily_tool = TavilySearchResults(max_results=4)
raw_results = []
# Execute search queries generated by the router
for query in state["queries"]:
results = tavily_tool.invoke({"query": query})
raw_results.extend(results)
# Deduplicate and normalize into structured evidence items
deduped_evidence = []
seen_urls = set()
for res in raw_results:
url = res.get("url")
if url and url not in seen_urls:
seen_urls.add(url)
deduped_evidence.append({
"title": res.get("title", "Search Result"),
"url": url,
"snippet": res.get("content", "")
})
return {"evidence": deduped_evidence}
8. Stage 3: Visual Generation via Gemini Subgraph
To elevate the output from raw text to an illustrated technical article, the application uses a Reducer Subgraph. Instead of relying on a single monolith reducer function, it uses a specialized 3-node sequence to synthesize text, identify optimal image locations, and generate custom visuals using Google's Gemini API.
Pydantic Schemas for Visual Integration
class ImageSpec(BaseModel):
placeholder: str = Field(description="Tag inserted in markdown e.g. '[IMAGE_PLACEHOLDER_1]'")
filename: str = Field(description="Target output filename e.g. 'attention_diagram.png'")
prompt: str = Field(description="Detailed text-to-image prompt tailored for technical illustration")
class GlobalImagePlan(BaseModel):
markdown_with_placeholders: str = Field(description="Full text containing placeholder tags")
image_specs: List[ImageSpec] = Field(description="Metadata and prompts for up to 3 visuals")
9. Subgraph Implementation & Image Generation Nodes
Node 1: Merge Content
Combines raw worker outputs into a cohesive document string.
def merge_content_node(state: AgentState) -> Dict[str, Any]:
merged = "\n\n".join(state["sections"])
full_markdown = f"# {state['plan'].blog_title}\n\n{merged}"
return {"merged_markdown": full_markdown}
Node 2: Decide Images
Sends the unified text through an LLM to evaluate complex topics that benefit from visual illustrations (e.g., flowcharts, architecture diagrams, sequence steps).
def decide_images_node(state: AgentState) -> Dict[str, Any]:
image_planner = llm.with_structured_output(GlobalImagePlan)
system_prompt = SystemMessage(content=(
"You are an expert technical media editor. Review the blog post and determine "
"where visuals (diagrams, concepts) are necessary. Select AT MOST 3 places.\n"
"1. Insert placeholder tags like [IMAGE_PLACEHOLDER_1] into the markdown.\n"
"2. Write highly detailed, technical generation prompts for an image model."
))
result = image_planner.invoke([system_prompt, HumanMessage(content=state["merged_markdown"])])
return {
"markdown_with_placeholders": result.markdown_with_placeholders,
"image_specs": [spec.dict() for spec in result.image_specs]
}
Node 3: Generate and Place Images (Gemini API Integration)
Iterates through image_specs, invokes Google's Gemini Image Generation endpoint, saves the resulting binary bytes locally, and swaps the placeholder tags for Markdown image links ().
import google.generativeai as genai
import os
def generate_and_place_images_node(state: AgentState) -> Dict[str, Any]:
current_markdown = state["markdown_with_placeholders"]
os.makedirs("./images", exist_ok=True)
for spec in state["image_specs"]:
placeholder = spec["placeholder"]
filename = spec["filename"]
prompt = spec["prompt"]
local_path = f"./images/{filename}"
# Invoke Google Gemini API for image generation
try:
# Conceptual call to Gemini Imagen API wrapper
result_bytes = call_gemini_imagen_api(prompt)
with open(local_path, "wb") as f:
f.write(result_bytes)
# Replace placeholder tag with Markdown image reference
markdown_image_tag = f"\n*Figure: {prompt}*"
current_markdown = current_markdown.replace(placeholder, markdown_image_tag)
except Exception as e:
# Fallback handling in case of rate limits or generation failure
current_markdown = current_markdown.replace(placeholder, "")
return {"final_markdown": current_markdown}
10. Stage 4: User Interface & Enterprise Deployment Mindset
To turn this multi-agent workflow into an accessible product, the application uses a Streamlit dashboard. This provides full observability into the agent's internal state machine during generation.
Enterprise Observability & Operations
- State Node Highlighting: Streamlit polls the state machine step, highlighting the active node (
Router->Research->Orchestrator->Workers->Reducer). - Evidence & Citation Transparency: When a blog uses web search data (
needs_research = True), the UI builds a bibliography section dynamically mapping inline citations to the Tavily source links stored in theEvidencePack. - API Cost & Safety Management:
- Structured Output Fail-Safes: Using Pydantic models with
with_structured_outputguarantees syntax validity, eliminating downstream JSON parsing crashes. - Token Cost Boundaries: Applying strict max word count target parameters (
target_words) per worker section keeps overall context length predictable. - Multimodal Fallbacks: Image generation steps use graceful exception handling so that a media error never aborts the core text delivery.
- Structured Output Fail-Safes: Using Pydantic models with
11. Complete Code Blueprint & Graph Assembly
Below is the complete blueprint putting all components together into a compiled LangGraph workflow.
from langgraph.graph import StateGraph, START, END
# --- 1. INITIALIZE GRAPH BUILDER ---
builder = StateGraph(AgentState)
# --- 2. ADD MAIN NODES ---
builder.add_node("router", router_node)
builder.add_node("research", research_node)
builder.add_node("orchestrator", orchestrator_node)
builder.add_node("worker", worker_node)
# --- 3. BUILD REDUCER SUBGRAPH ---
reducer_subgraph = StateGraph(AgentState)
reducer_subgraph.add_node("merge_content", merge_content_node)
reducer_subgraph.add_node("decide_images", decide_images_node)
reducer_subgraph.add_node("generate_images", generate_and_place_images_node)
reducer_subgraph.add_edge(START, "merge_content")
reducer_subgraph.add_edge("merge_content", "decide_images")
reducer_subgraph.add_edge("decide_images", "generate_images")
reducer_subgraph.add_edge("generate_images", END)
compiled_reducer = reducer_subgraph.compile()
builder.add_node("reducer", compiled_reducer)
# --- 4. DEFINE CONDITIONAL ROUTING EDGES ---
def route_after_router(state: AgentState):
if state["needs_research"]:
return "research"
return "orchestrator"
builder.add_edge(START, "router")
builder.add_conditional_edges("router", route_after_router, {
"research": "research",
"orchestrator": "orchestrator"
})
builder.add_edge("research", "orchestrator")
# Dynamic Fan-Out from Orchestrator to Workers
builder.add_conditional_edges("orchestrator", fan_out_to_workers, ["worker"])
# Fan-In from Workers to Reducer Subgraph
builder.add_edge("worker", "reducer")
builder.add_edge("reducer", END)
# --- 5. COMPILE MASTER GRAPH ---
app = builder.compile()
12. Summary and Key Takeaways
- Planning vs. Reactive Execution: Reactive single-prompt approaches fail at complex content creation. Planning agents separate structural task decomposition from execution, yielding superior depth, organization, and technical accuracy.
- Granular Pydantic Modeling: Defining structured contracts (
Plan,Task,RouterDecision,ImageSpec) eliminates output ambiguity and guarantees downstream node compatibility. - Dynamic Worker Fan-Out: Using LangGraph's
SendAPI allows dynamic creation of parallel worker nodes mapped to plan sections, reducing end-to-end execution latency from to relative to total sections. - Intelligent Query Routing: Categorizing topics into
closed_book,hybrid, andopen_bookminimizes unnecessary web searches, saving API costs and latency for static evergreen concepts. - Normalized Research Layer: Standardizing raw search outputs into clean
EvidenceItemmodels prevents search noise from corrupting downstream worker prompts. - Modular Subgraph Reducers: Isolating complex synthesis steps—such as text merging, media planning, and image generation—into a dedicated subgraph keeps the master execution graph simple and maintainable.
- Production Multimodal Synthesis: Combining text generation via OpenAI with targeted web research via Tavily and custom visual synthesis via Google Gemini yields a complete, production-grade technical content platform.