Loading...
Loading...
Generate articles, reports, blog posts, or marketing copy with AI. Use when writing blog posts, creating product descriptions, generating newsletters, drafting reports, producing marketing copy, creating documentation, writing email campaigns, or any task where AI writes long-form content from a topic or brief. Powered by DSPy content generation pipelines.
npx skill4agent add lebsral/dspy-programming-not-prompting-lms-skills ai-writing-contentimport dspy
from pydantic import BaseModel, Field
class Section(BaseModel):
heading: str = Field(description="Section heading")
key_points: list[str] = Field(description="Main points to cover in this section")
class ContentOutline(BaseModel):
title: str
sections: list[Section]
class GenerateOutline(dspy.Signature):
"""Create a structured outline for the content."""
topic: str = dspy.InputField(desc="The topic or brief to write about")
content_type: str = dspy.InputField(desc="Type: blog post, report, product description, etc.")
audience: str = dspy.InputField(desc="Who will read this content")
outline: ContentOutline = dspy.OutputField()
outliner = dspy.ChainOfThought(GenerateOutline)class GenerateResearchedOutline(dspy.Signature):
"""Create a structured outline grounded in the provided research."""
topic: str = dspy.InputField()
content_type: str = dspy.InputField()
audience: str = dspy.InputField()
research: list[str] = dspy.InputField(desc="Research sources and key facts")
outline: ContentOutline = dspy.OutputField()class WriteSection(dspy.Signature):
"""Write one section of the article based on the outline."""
topic: str = dspy.InputField(desc="Overall article topic")
section_heading: str = dspy.InputField(desc="This section's heading")
key_points: list[str] = dspy.InputField(desc="Points to cover in this section")
previous_sections: str = dspy.InputField(desc="What's been written so far, for continuity")
tone: str = dspy.InputField(desc="Writing tone and style")
section_text: str = dspy.OutputField(desc="The written section (2-4 paragraphs)")
class ContentWriter(dspy.Module):
def __init__(self):
self.outline = dspy.ChainOfThought(GenerateOutline)
self.write_section = dspy.ChainOfThought(WriteSection)
def forward(self, topic, content_type="blog post", audience="general", tone="professional"):
# Step 1: Generate outline
plan = self.outline(topic=topic, content_type=content_type, audience=audience)
# Step 2: Write each section
sections = []
running_text = ""
for section in plan.outline.sections:
result = self.write_section(
topic=topic,
section_heading=section.heading,
key_points=section.key_points,
previous_sections=running_text[-2000:], # last 2000 chars for context
tone=tone,
)
sections.append(f"## {section.heading}\n\n{result.section_text}")
running_text += result.section_text + "\n\n"
full_article = f"# {plan.outline.title}\n\n" + "\n\n".join(sections)
return dspy.Prediction(
title=plan.outline.title,
outline=plan.outline,
article=full_article,
)class ResearchTopic(dspy.Signature):
"""Generate search queries to research this topic."""
topic: str = dspy.InputField()
key_points: list[str] = dspy.InputField(desc="Points that need factual backing")
queries: list[str] = dspy.OutputField(desc="Search queries to find supporting facts")
class WriteSectionWithSources(dspy.Signature):
"""Write a section using the provided sources for factual claims."""
section_heading: str = dspy.InputField()
key_points: list[str] = dspy.InputField()
sources: list[str] = dspy.InputField(desc="Research passages to ground claims in")
previous_sections: str = dspy.InputField()
tone: str = dspy.InputField()
section_text: str = dspy.OutputField(desc="Section text with claims grounded in sources")
class ResearchedWriter(dspy.Module):
def __init__(self):
self.outline = dspy.ChainOfThought(GenerateOutline)
self.research = dspy.ChainOfThought(ResearchTopic)
self.retrieve = dspy.Retrieve(k=3)
self.write = dspy.ChainOfThought(WriteSectionWithSources)
def forward(self, topic, content_type="blog post", audience="general", tone="professional"):
plan = self.outline(topic=topic, content_type=content_type, audience=audience)
sections = []
running_text = ""
for section in plan.outline.sections:
# Research this section
queries = self.research(
topic=topic, key_points=section.key_points
).queries
sources = []
for query in queries:
sources.extend(self.retrieve(query).passages)
# Write with sources
result = self.write(
section_heading=section.heading,
key_points=section.key_points,
sources=sources,
previous_sections=running_text[-2000:],
tone=tone,
)
sections.append(f"## {section.heading}\n\n{result.section_text}")
running_text += result.section_text + "\n\n"
return dspy.Prediction(
title=plan.outline.title,
article=f"# {plan.outline.title}\n\n" + "\n\n".join(sections),
)class CritiqueContent(dspy.Signature):
"""Critique the written content and suggest improvements."""
content: str = dspy.InputField(desc="The content to critique")
content_type: str = dspy.InputField()
audience: str = dspy.InputField()
is_good_enough: bool = dspy.OutputField(desc="Is this ready to publish?")
feedback: str = dspy.OutputField(desc="Specific feedback for improvement")
class ImproveContent(dspy.Signature):
"""Improve the content based on the feedback."""
content: str = dspy.InputField(desc="Current draft")
feedback: str = dspy.InputField(desc="Feedback to address")
improved_content: str = dspy.OutputField(desc="Improved version")
class QualityWriter(dspy.Module):
def __init__(self, max_revisions=2):
self.writer = ContentWriter()
self.critic = dspy.ChainOfThought(CritiqueContent)
self.improver = dspy.ChainOfThought(ImproveContent)
self.max_revisions = max_revisions
def forward(self, topic, content_type="blog post", audience="general", tone="professional"):
# Generate first draft
draft = self.writer(
topic=topic, content_type=content_type, audience=audience, tone=tone
)
article = draft.article
# Critique-improve loop
for _ in range(self.max_revisions):
critique = self.critic(
content=article, content_type=content_type, audience=audience
)
if critique.is_good_enough:
break
improved = self.improver(content=article, feedback=critique.feedback)
article = improved.improved_content
return dspy.Prediction(
title=draft.title,
article=article,
)class StyledWriter(dspy.Module):
def __init__(self, brand_rules=None):
self.writer = ContentWriter()
self.brand_rules = brand_rules or []
def forward(self, topic, content_type="blog post", audience="general", tone="professional"):
result = self.writer(topic=topic, content_type=content_type, audience=audience, tone=tone)
# Enforce brand rules
article_lower = result.article.lower()
for rule in self.brand_rules:
if rule.get("type") == "forbidden_word":
dspy.Assert(
rule["word"].lower() not in article_lower,
f"Content must not use the word '{rule['word']}'. "
f"Use '{rule.get('alternative', 'a different word')}' instead."
)
elif rule.get("type") == "required_section":
dspy.Assert(
rule["heading"].lower() in article_lower,
f"Content must include a '{rule['heading']}' section."
)
# General style checks
sentences = result.article.split(".")
avg_sentence_len = sum(len(s.split()) for s in sentences) / max(len(sentences), 1)
dspy.Suggest(
avg_sentence_len < 25,
"Sentences are too long on average. Aim for shorter, punchier sentences."
)
return result
# Usage
writer = StyledWriter(brand_rules=[
{"type": "forbidden_word", "word": "utilize", "alternative": "use"},
{"type": "forbidden_word", "word": "leverage", "alternative": "use"},
{"type": "required_section", "heading": "Conclusion"},
])def readability_metric(example, prediction, trace=None):
words = prediction.article.split()
sentences = prediction.article.split(".")
if not sentences or not words:
return 0.0
avg_sentence_len = len(words) / len(sentences)
# Penalize very long or very short sentences
readability = 1.0 if 10 < avg_sentence_len < 20 else 0.5
# Penalize very short articles
length_ok = 1.0 if len(words) > 200 else 0.5
return (readability + length_ok) / 2class JudgeContent(dspy.Signature):
"""Judge the quality of generated content."""
content: str = dspy.InputField()
content_type: str = dspy.InputField()
topic: str = dspy.InputField()
relevance: float = dspy.OutputField(desc="0.0-1.0 — stays on topic")
coherence: float = dspy.OutputField(desc="0.0-1.0 — flows well, logically structured")
engagement: float = dspy.OutputField(desc="0.0-1.0 — interesting to read")
def content_quality_metric(example, prediction, trace=None):
judge = dspy.Predict(JudgeContent)
result = judge(
content=prediction.article,
content_type=example.content_type,
topic=example.topic,
)
return (result.relevance + result.coherence + result.engagement) / 3optimizer = dspy.BootstrapFewShot(metric=content_quality_metric, max_bootstrapped_demos=4)
optimized = optimizer.compile(QualityWriter(), trainset=trainset)/ai-summarizing/ai-building-pipelines/ai-improving-accuracy