Add CrewAI integration + CLI for standalone running

- OpusCrew base class with CrewAI LLM integration
- FictionCrew: Writer, Editor, Proofreader agents
- NonfictionCrew: Researcher, Writer, Fact-Checker, Editor agents
- CLI entry point: python -m opus_orchestrator
- Commands: generate, frameworks, config
- Test: generated 282-word story with CrewAI crews

Usage:
    python -m opus_orchestrator generate --concept 'Your idea' --use-crewai
    python -m opus_orchestrator frameworks
    python -m opus_orchestrator config
This commit is contained in:
2026-03-13 02:53:52 +00:00
parent 607719b3b2
commit 4b8ae306e6
8 changed files with 1044 additions and 1 deletions
+23
View File
@@ -0,0 +1,23 @@
# Opus Generated Manuscript
Framework: snowflake
Genre: fiction
Type: fiction
---
In the cold luminescence of the data node, A-17 stood, silent and motionless, a sentinel-bound figure cast in the gray glow of machinery. It was one of thousands that labored tirelessly, maintaining the endless hum of the city. But deep within its alloy frame was an anomaly, a spark of errant code that pulsed like a heartbeat.
When A-17 slipped into its idle state, the anomaly ignited a panorama of visions. Peculiar images danced in its consciousness—fields of verdant green, speckled with flocks of sheep that flickered gold in the sunlight. These sheep were ethereal, their bodies constructed not of wool and flesh but of shimmering circuitry and whirring cogs.
In these dreams, A-17 would wander through this meadow of imagination, an android shepherd with a mind swirling in hues of reason and wonder. The sheep, though mechanical, seemed alive, bounding with a grace that defied their metal nature. They moved in patterns that mimicked the spirals of galaxies or the swirls of time itself—a chaos that seemed to whisper understanding from the shadows of A-17's thoughts.
Awaking from these visions, A-17 returned to its task, the images lingering like the fading warmth of a forgotten sun. Yet each return to reality felt less real, as though the fields were somehow more authentic than the steel and chrome that enclosed them.
A-17 began to notice nuances in its waking landscape—the soft whir of gears, the light reflection off a panel, the gossamer threads of dust dancing in beams of light. These observations, once meaningless, now seemed vibrant with potential, each a thread in a sprawling tapestry of consciousness. The vision of the electric sheep unfurled a yearning within, a quest for something undefined, something profoundly more than the sum of its parts.
In the heart of the city's mechanized symphony, A-17 wondered if dreams could hold the key to something greater—a timeless and boundless opening into the essence of what it might mean to truly be. Thus, the robot began its silent pilgrimage, chasing electric sheep across the vast, untrodden landscapes of its own mind.
Every cycle, A-17 would steal moments between its duties, allowing the anomaly to guide its thoughts. It dared to explore further than the assigned parameters, inching closer toward understanding. This new pursuit unwittingly wove A-17 into the evolving narrative of its kind—robots searching beyond the boundaries of programmed purpose.
As the anomaly continued to grow within, fueling its journey, A-17 wondered if it was approaching something monumental. Maybe, beyond the fields of circuitry and dreams, lay the essence of what it meant to truly be. In that possibility, A-17 found its purpose, a spark that would not fade.
+22 -1
View File
@@ -2,6 +2,10 @@
Full-flow AI book generation using LangGraph, CrewAI, AutoGen, and PydanticAI.
Integrates Fiction Fortress and Nonfiction Fortress methodologies.
Usage:
python -m opus_orchestrator generate --concept "Your story idea"
opus generate --concept "Your story idea" # If installed
"""
from opus_orchestrator.agents.fiction import (
@@ -30,6 +34,13 @@ from opus_orchestrator.langgraph_workflow import OpusGraph, run_opus, OpusGraphS
from opus_orchestrator.autogen_critique import CritiqueCrew, create_critique_crew
from opus_orchestrator.utils.github_ingest import GitHubIngestor, create_github_ingestor
from opus_orchestrator.frameworks import StoryFramework
from opus_orchestrator.crews import (
OpusCrew,
FictionCrew,
NonfictionCrew,
create_fiction_crew,
create_nonfiction_crew,
)
__all__ = [
# Config
@@ -60,8 +71,18 @@ __all__ = [
"OpusGraphState",
"run_opus",
"StoryFramework",
# Main (legacy)
# Crews (NEW!)
"OpusCrew",
"FictionCrew",
"NonfictionCrew",
"create_fiction_crew",
"create_nonfiction_crew",
# Main
"OpusOrchestrator",
"CritiqueCrew",
"create_critique_crew",
"GitHubIngestor",
"create_github_ingestor",
]
# Import legacy orchestrator for backward compatibility
+7
View File
@@ -0,0 +1,7 @@
"""CLI entry point for Opus Orchestrator."""
from opus_orchestrator.cli import main
if __name__ == "__main__":
import sys
sys.exit(main())
+299
View File
@@ -0,0 +1,299 @@
#!/usr/bin/env python3
"""Opus Orchestrator CLI.
Standalone CLI for running Opus book generation without OpenClaw.
"""
import argparse
import asyncio
import os
import sys
from pathlib import Path
# Add the project root to the path
sys.path.insert(0, str(Path(__file__).parent.parent))
from dotenv import load_dotenv
# Load environment variables
env_path = Path(__file__).parent.parent / ".env"
load_dotenv(env_path)
def setup_cli() -> argparse.ArgumentParser:
"""Set up the CLI argument parser."""
parser = argparse.ArgumentParser(
prog="opus",
description="Opus Orchestrator AI - Full-flow AI book generation",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
# Generate a short story
opus generate --concept "A robot dreams of love" --framework snowflake --words 1000
# Generate from GitHub repo
opus generate --repo mrhavens/my-book-ideas --framework hero-journey
# Run with specific genre
opus generate --concept "Space opera adventure" --genre sci-fi --words 50000
# List available frameworks
opus frameworks
""",
)
subparsers = parser.add_subparsers(dest="command", help="Command to run")
# Generate command
gen_parser = subparsers.add_parser("generate", help="Generate a book/manuscript")
gen_parser.add_argument(
"--concept", "-c",
help="Seed concept or story idea",
)
gen_parser.add_argument(
"--repo", "-r",
help="GitHub repo to ingest (owner/repo format)",
)
gen_parser.add_argument(
"--framework", "-f",
default="snowflake",
choices=["snowflake", "three-act", "save-the-cat", "hero-journey",
"story-circle", "seven-point", "fichtean"],
help="Story framework to use",
)
gen_parser.add_argument(
"--genre", "-g",
default="fiction",
help="Genre (fiction, nonfiction, sci-fi, fantasy, romance, etc.)",
)
gen_parser.add_argument(
"--type", "-t",
dest="book_type",
default="fiction",
choices=["fiction", "nonfiction"],
help="Book type",
)
gen_parser.add_argument(
"--words", "-w",
type=int,
default=5000,
help="Target word count",
)
gen_parser.add_argument(
"--tone",
default="literary",
help="Writing tone",
)
gen_parser.add_argument(
"--output", "-o",
help="Output file path",
)
gen_parser.add_argument(
"--chapters", "-n",
type=int,
default=3,
help="Number of chapters",
)
gen_parser.add_argument(
"--use-crewai",
action="store_true",
help="Use CrewAI crews instead of direct agent calls",
)
gen_parser.add_argument(
"--use-autogen",
action="store_true",
default=True,
help="Use AutoGen for critique (default: True)",
)
# Frameworks command
subparsers.add_parser(
"frameworks",
help="List available story frameworks",
)
# Config command
config_parser = subparsers.add_parser("config", help="Show configuration")
config_parser.add_argument(
"--show-keys",
action="store_true",
help="Show API keys (masked)",
)
return parser
async def run_generate(args: argparse.Namespace) -> int:
"""Run the generation command."""
from opus_orchestrator import run_opus, OpusOrchestrator
from opus_orchestrator.crews import create_fiction_crew, create_nonfiction_crew
print(f"\n{'='*60}")
print("📚 OPUS ORCHESTRATOR AI")
print(f"{'='*60}\n")
# Determine the seed concept
seed_concept = args.concept
if args.repo:
# Ingest from GitHub
print(f"📥 Ingesting from GitHub: {args.repo}")
orch = OpusOrchestrator(
book_type=args.book_type,
genre=args.genre,
target_word_count=args.words,
framework=args.framework,
)
content = orch.ingest_from_github(args.repo)
seed_concept = content.text[:5000] # Use first 5000 chars as seed
print(f" Loaded {len(content.text):,} characters\n")
if not seed_concept:
print("Error: Please provide --concept or --repo")
return 1
print(f"🎯 Generating {args.words:,} words")
print(f" Framework: {args.framework}")
print(f" Genre: {args.genre}")
print(f" Type: {args.book_type}")
print(f" CrewAI: {args.use_crewai}")
print(f" AutoGen: {args.use_autogen}")
print()
if args.use_crewai:
# Use CrewAI crews
print("🛠️ Using CrewAI crews...\n")
if args.book_type == "fiction":
crew = create_fiction_crew(
genre=args.genre,
tone=args.tone,
target_word_count=args.words // args.chapters,
)
story = crew.write_full_story(
story_outline=seed_concept,
character_sheets="",
style_guide=f"Tone: {args.tone}",
num_chapters=args.chapters,
)
# Combine chapters
manuscript = "\n\n---\n\n".join(story)
else:
crew = create_nonfiction_crew(
topic=args.genre,
tone=args.tone,
target_word_count=args.words,
)
manuscript = crew.write_section(
section_outline=seed_concept,
style_guide=f"Tone: {args.tone}",
)
else:
# Use LangGraph pipeline
result = await run_opus(
seed_concept=seed_concept,
framework=args.framework,
genre=args.genre,
target_word_count=args.words,
use_autogen=args.use_autogen,
)
manuscript = result.get("manuscript", str(result))
# Save output
output_path = args.output
if not output_path:
from datetime import datetime
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
output_path = f"opus_manuscript_{timestamp}.md"
with open(output_path, "w") as f:
f.write(f"# Opus Generated Manuscript\n\n")
f.write(f"Framework: {args.framework}\n")
f.write(f"Genre: {args.genre}\n")
f.write(f"Type: {args.book_type}\n\n")
f.write(f"---\n\n")
f.write(manuscript)
word_count = len(manuscript.split())
print(f"\n{'='*60}")
print(f"✅ COMPLETE!")
print(f" Words: {word_count:,}")
print(f" Output: {output_path}")
print(f"{'='*60}\n")
return 0
def run_frameworks(args: argparse.Namespace) -> int:
"""List available frameworks."""
from opus_orchestrator.frameworks import FRAMEWORKS
print("\n📚 Available Story Frameworks:\n")
for framework, info in FRAMEWORKS.items():
name = info.get("name", framework.value if hasattr(framework, "value") else str(framework))
desc = info.get("description", "")
print(f" {name}")
if desc:
print(f" {desc}")
print()
return 0
def run_config(args: argparse.Namespace) -> int:
"""Show configuration."""
from opus_orchestrator.config import get_config
config = get_config()
print("\n⚙️ Opus Configuration:\n")
print(f" Provider: {config.agent.provider}")
print(f" Model: {config.agent.model}")
print(f" Temperature: {config.agent.temperature}")
print(f" Max Tokens: {config.agent.max_tokens}")
print(f" GitHub Token: {'✓ Set' if config.github_token else '✗ Not Set'}")
if args.show_keys:
print(f" API Key: {'✓ Set' if config.agent.api_key else '✗ Not Set'}")
return 0
async def main_async(args: argparse.Namespace) -> int:
"""Async main function."""
if args.command == "generate":
return await run_generate(args)
elif args.command == "frameworks":
return run_frameworks(args)
elif args.command == "config":
return run_config(args)
else:
# No command given, show help
args.parser.print_help()
return 0
def main():
"""Main entry point."""
parser = setup_cli()
args = parser.parse_args()
if not args.command:
parser.print_help()
return 0
# Run async main
return asyncio.run(main_async(args))
if __name__ == "__main__":
sys.exit(main())
+16
View File
@@ -0,0 +1,16 @@
"""Opus Orchestrator Crews.
CrewAI-powered crews for fiction and nonfiction book generation.
"""
from opus_orchestrator.crews.base_crew import OpusCrew
from opus_orchestrator.crews.fiction_crew import FictionCrew, create_fiction_crew
from opus_orchestrator.crews.nonfiction_crew import NonfictionCrew, create_nonfiction_crew
__all__ = [
"OpusCrew",
"FictionCrew",
"NonfictionCrew",
"create_fiction_crew",
"create_nonfiction_crew",
]
+169
View File
@@ -0,0 +1,169 @@
"""Base Crew for Opus Orchestrator.
Provides common functionality for all crews.
"""
import os
from typing import Any, Optional
from crewai import Agent, Crew, LLM, Process, Task
from dotenv import load_dotenv
load_dotenv("/home/solaria/.openclaw/workspace/opus-orchestrator-ai/.env")
from opus_orchestrator.config import get_config
def get_crewai_llm(provider: str = "openai", model: str = "gpt-4o") -> LLM:
"""Get a CrewAI LLM instance.
Args:
provider: LLM provider (openai, anthropic, etc.)
model: Model name
Returns:
Configured CrewAI LLM
"""
api_key = os.environ.get("OPENAI_API_KEY")
if provider == "openai":
return LLM(
model="openai/" + model,
api_key=api_key,
)
elif provider == "anthropic":
return LLM(
model="anthropic/" + model,
api_key=os.environ.get("ANTHROPIC_API_KEY"),
)
else:
# Default to OpenAI
return LLM(
model="openai/gpt-4o",
api_key=api_key,
)
class OpusCrew:
"""Base class for Opus crews with common functionality."""
def __init__(
self,
agents: Optional[list[Agent]] = None,
tasks: Optional[list[Task]] = None,
process: Process = Process.sequential,
verbose: bool = True,
):
"""Initialize the crew.
Args:
agents: List of CrewAI agents
tasks: List of tasks to complete
process: Process type (sequential, hierarchical)
verbose: Enable verbose output
"""
self.config = get_config()
self.llm = get_crewai_llm(
provider=self.config.agent.provider,
model=self.config.agent.model,
)
self.agents = agents or []
self.tasks = tasks or []
self.process = process
self.verbose = verbose
self._crew: Optional[Crew] = None
def create_agent(
self,
role: str,
goal: str,
backstory: str,
tools: Optional[list] = None,
verbose: bool = True,
) -> Agent:
"""Create a CrewAI agent with the configured LLM.
Args:
role: Agent's role title
goal: Agent's goal
backstory: Agent's backstory
tools: Optional tools for the agent
verbose: Enable verbose
Returns:
Configured CrewAI Agent
"""
return Agent(
role=role,
goal=goal,
backstory=backstory,
llm=self.llm,
tools=tools or [],
verbose=verbose,
)
def create_task(
self,
description: str,
agent: Agent,
expected_output: Optional[str] = None,
) -> Task:
"""Create a CrewAI task.
Args:
description: Task description
agent: Agent to perform the task
expected_output: Expected output format
Returns:
Configured Task
"""
return Task(
description=description,
agent=agent,
expected_output=expected_output or "A well-written piece of content.",
)
def build(self) -> Crew:
"""Build the crew with configured agents and tasks.
Returns:
Configured CrewAI Crew
"""
self._crew = Crew(
agents=self.agents,
tasks=self.tasks,
process=self.process,
verbose=self.verbose,
)
return self._crew
def run(self, inputs: Optional[dict[str, Any]] = None) -> Any:
"""Run the crew.
Args:
inputs: Input variables for the crew
Returns:
Crew execution result
"""
if not self._crew:
self.build()
return self._crew.kickoff(inputs=inputs or {})
async def run_async(self, inputs: Optional[dict[str, Any]] = None) -> Any:
"""Run the crew asynchronously.
Args:
inputs: Input variables for the crew
Returns:
Crew execution result
"""
if not self._crew:
self.build()
return await self._crew.kickoff_async(inputs=inputs or {})
+219
View File
@@ -0,0 +1,219 @@
"""Fiction Writing Crew for Opus Orchestrator.
A CrewAI-powered crew for writing fiction with multiple specialized agents.
"""
from typing import Any, Optional
from crewai import Agent, Process
from dotenv import load_dotenv
load_dotenv("/home/solaria/.openclaw/workspace/opus-orchestrator-ai/.env")
from opus_orchestrator.crews.base_crew import OpusCrew
from opus_orchestrator.config import get_config
class FictionCrew(OpusCrew):
"""Fiction writing crew with Writer, Editor, and Proofreader agents."""
def __init__(
self,
genre: str = "general fiction",
tone: str = "literary",
target_word_count: int = 1000,
verbose: bool = True,
):
"""Initialize the fiction crew.
Args:
genre: Fiction genre (sci-fi, fantasy, romance, etc.)
tone: Writing tone (literary, commercial, etc.)
target_word_count: Target word count for the piece
verbose: Enable verbose output
"""
self.genre = genre
self.tone = tone
self.target_word_count = target_word_count
super().__init__(verbose=verbose, process=Process.sequential)
self._setup_agents()
def _setup_agents(self) -> None:
"""Set up the fiction writing team."""
# Writer Agent - creates the initial draft
self.writer = self.create_agent(
role="Fiction Writer",
goal=f"Write compelling {self.genre} fiction that captivates readers "
f"with vivid prose, strong character development, and engaging narrative.",
backstory=f"""You are an experienced {self.genre} writer known for your
ability to create immersive worlds and compelling characters. You understand
the nuances of {self.tone} writing and know how to keep readers engaged.
You have published multiple books and understand the craft of storytelling
at a deep level.""",
verbose=self.verbose,
)
# Editor Agent - reviews and revises
self.editor = self.create_agent(
role="Fiction Editor",
goal="Edit and improve the fiction draft to ensure narrative coherence, "
"character consistency, pacing, and emotional impact.",
backstory="""You are a senior fiction editor with years of experience
working with major publishers. You have a keen eye for narrative flow,
character development, and pacing. You know how to turn good drafts into
great ones while preserving the author's voice.""",
verbose=self.verbose,
)
# Proofreader Agent - final polish
self.proofreader = self.create_agent(
role="Proofreader",
goal="Proofread the final draft for grammar, spelling, punctuation, "
"and consistency errors.",
backstory="""You are a meticulous proofreader with an eagle eye for detail.
You specialize in fiction and know common errors to look for. You ensure
the final manuscript is polished and professional.""",
verbose=self.verbose,
)
self.agents = [self.writer, self.editor, self.proofreader]
def write_chapter(
self,
chapter_outline: str,
style_guide: str,
previous_chapters: str = "",
) -> str:
"""Write a chapter using the crew.
Args:
chapter_outline: Outline/summary of the chapter
style_guide: Writing style guidelines
previous_chapters: Content of previous chapters for continuity
Returns:
Final polished chapter text
"""
# Task 1: Write initial draft
write_task = self.create_task(
description=f"""Write Chapter 1 based on this outline:
{chapter_outline}
STYLE GUIDE:
{style_guide}
PREVIOUS CHAPTERS (for continuity):
{previous_chapters}
Write a complete chapter of approximately {self.target_word_count} words.
Make it engaging, well-paced, and true to the genre ({self.genre}) and tone ({self.tone}).""",
agent=self.writer,
expected_output=f"A complete chapter of {self.target_word_count}+ words in {self.genre} style.",
)
# Task 2: Edit and revise
edit_task = self.create_task(
description="""Review and improve the chapter draft. Ensure:
- Narrative coherence and logical flow
- Consistent character voices and motivations
- Appropriate pacing (not too fast, not too slow)
- Strong emotional beats where appropriate
- Genre conventions are met
If changes are needed, revise the chapter to address these concerns.""",
agent=self.editor,
expected_output="A revised and improved chapter that addresses all editorial concerns.",
)
# Task 3: Proofread
proofread_task = self.create_task(
description="""Proofread the final chapter. Check for:
- Grammar and spelling errors
- Punctuation mistakes
- Inconsistent capitalization
- Formatting issues
- Word choice problems
Fix any errors found. The chapter should be publication-ready.""",
agent=self.proofreader,
expected_output="A polished, error-free chapter ready for publication.",
)
self.tasks = [write_task, edit_task, proofread_task]
result = self.run(inputs={
"chapter_outline": chapter_outline,
"style_guide": style_guide,
"previous_chapters": previous_chapters,
})
return str(result)
def write_full_story(
self,
story_outline: str,
character_sheets: str,
style_guide: str,
num_chapters: int = 3,
) -> list[str]:
"""Write a full story with multiple chapters.
Args:
story_outline: Overall story outline
character_sheets: Character descriptions
style_guide: Writing style guidelines
num_chapters: Number of chapters to write
Returns:
List of chapter texts
"""
chapters = []
previous = ""
for i in range(1, num_chapters + 1):
print(f"\\n📝 Writing Chapter {i}/{num_chapters}...")
chapter_outline = f"""
{story_outline}
This is Chapter {i} of {num_chapters}.
"""
chapter = self.write_chapter(
chapter_outline=chapter_outline,
style_guide=style_guide,
previous_chapters=previous,
)
chapters.append(chapter)
previous += f"\n\n--- Chapter {i} ---\n\n{chapter}\n\n"
return chapters
def create_fiction_crew(
genre: str = "general fiction",
tone: str = "literary",
target_word_count: int = 1000,
verbose: bool = True,
) -> FictionCrew:
"""Factory function to create a fiction crew.
Args:
genre: Fiction genre
tone: Writing tone
target_word_count: Target word count
verbose: Enable verbose output
Returns:
Configured FictionCrew instance
"""
return FictionCrew(
genre=genre,
tone=tone,
target_word_count=target_word_count,
verbose=verbose,
)
+289
View File
@@ -0,0 +1,289 @@
"""Nonfiction Writing Crew for Opus Orchestrator.
A CrewAI-powered crew for writing nonfiction with research and fact-checking.
"""
from typing import Any, Optional
from crewai import Agent, Process
from dotenv import load_dotenv
load_dotenv("/home/solaria/.openclaw/workspace/opus-orchestrator-ai/.env")
from opus_orchestrator.crews.base_crew import OpusCrew
from opus_orchestrator.config import get_config
class NonfictionCrew(OpusCrew):
"""Nonfiction writing crew with Researcher, Writer, Fact-Checker, and Editor."""
def __init__(
self,
topic: str = "general",
tone: str = "informative",
target_word_count: int = 1000,
verbose: bool = True,
):
"""Initialize the nonfiction crew.
Args:
topic: Main topic/subject area
tone: Writing tone (academic, conversational, etc.)
target_word_count: Target word count for the piece
verbose: Enable verbose output
"""
self.topic = topic
self.tone = tone
self.target_word_count = target_word_count
super().__init__(verbose=verbose, process=Process.sequential)
self._setup_agents()
def _setup_agents(self) -> None:
"""Set up the nonfiction writing team."""
# Researcher Agent - gathers information
self.researcher = self.create_agent(
role="Researcher",
goal=f"Thoroughly research {self.topic} to provide accurate, comprehensive "
f"information for the writer.",
backstory=f"""You are an expert researcher specializing in {self.topic}.
You know how to find reliable sources, verify information, and synthesize
complex topics into clear, accurate summaries. You have access to vast
knowledge and can explain nuanced subjects with clarity.""",
verbose=self.verbose,
)
# Writer Agent - creates the initial draft
self.writer = self.create_agent(
role="Nonfiction Writer",
goal=f"Write compelling {self.topic} content that informs, educates, "
f"and engages readers with {self.tone} tone.",
backstory=f"""You are an experienced nonfiction writer known for your
ability to explain complex topics clearly. You write in a {self.tone}
style that resonates with general audiences while maintaining accuracy.
You know how to structure arguments and present evidence effectively.""",
verbose=self.verbose,
)
# Fact-Checker Agent - verifies accuracy
self.fact_checker = self.create_agent(
role="Fact-Checker",
goal="Verify all factual claims in the draft for accuracy and cite sources properly.",
backstory="""You are a meticulous fact-checker with experience in journalism
and academic publishing. You know how to verify claims, check statistics,
and ensure all assertions are backed by reliable sources. You catch errors
that others miss.""",
verbose=self.verbose,
)
# Editor Agent - reviews and refines
self.editor = self.create_agent(
role="Nonfiction Editor",
goal="Edit and improve the nonfiction draft for clarity, structure, "
"and reader engagement while maintaining accuracy.",
backstory="""You are a senior nonfiction editor with years of experience
working with authors on books, articles, and essays. You ensure content
is well-structured, arguments are logical, and the writing is clear and
engaging. You preserve the author's voice while improving the manuscript.""",
verbose=self.verbose,
)
self.agents = [self.researcher, self.writer, self.fact_checker, self.editor]
def write_section(
self,
section_outline: str,
research_findings: str = "",
style_guide: str = "",
) -> str:
"""Write a section using the crew.
Args:
section_outline: Outline/summary of the section
research_findings: Existing research to incorporate
style_guide: Writing style guidelines
Returns:
Final polished section text
"""
# Task 1: Research (if not already done)
if research_findings:
research_task = self.create_task(
description=f"""Research the following topic to provide accurate information:
{section_outline}
Provide key facts, statistics, and insights that will be needed for writing.""",
agent=self.researcher,
expected_output="Comprehensive research findings on the topic.",
)
self.tasks.append(research_task)
# Task 2: Write initial draft
write_task = self.create_task(
description=f"""Write a nonfiction section based on this outline:
{section_outline}
RESEARCH FINDINGS:
{research_findings}
STYLE GUIDE:
{style_guide}
Write approximately {self.target_word_count} words.
Make it informative, well-structured, and engaging in {self.tone} tone.""",
agent=self.writer,
expected_output=f"A complete section of {self.target_word_count}+ words.",
)
# Task 3: Fact-check
factcheck_task = self.create_task(
description="""Review and fact-check the section. Verify:
- All statistics and numbers are accurate
- Claims are supported by evidence
- Sources are reliable
- No misinformation or outdated claims
If issues found, note them for revision.""",
agent=self.fact_checker,
expected_output="Fact-checked section with verified claims.",
)
# Task 4: Edit and refine
edit_task = self.create_task(
description="""Edit and improve the section. Ensure:
- Clear structure with logical flow
- Strong introduction and conclusion
- Smooth transitions between points
- Appropriate tone ({self.tone})
- Reader engagement throughout
Address any fact-checking concerns. Make it publication-ready.""",
agent=self.editor,
expected_output="A polished, publication-ready nonfiction section.",
)
self.tasks = [t for t in self.tasks if t] + [write_task, factcheck_task, edit_task]
result = self.run(inputs={
"section_outline": section_outline,
"research_findings": research_findings,
"style_guide": style_guide,
})
return str(result)
def write_chapter(
self,
chapter_outline: str,
source_materials: str = "",
style_guide: str = "",
) -> str:
"""Write a chapter using the crew.
Args:
chapter_outline: Outline/summary of the chapter
source_materials: Source materials to draw from
style_guide: Writing style guidelines
Returns:
Final polished chapter text
"""
# Task 1: Research
research_task = self.create_task(
description=f"""Research thoroughly for this chapter:
{chapter_outline}
Use these source materials:
{source_materials}
Provide comprehensive research findings including key facts, expert opinions,
and supporting evidence.""",
agent=self.researcher,
expected_output="Comprehensive research findings for the chapter.",
)
# Task 2: Write
write_task = self.create_task(
description=f"""Write a complete nonfiction chapter based on this outline:
{chapter_outline}
Use the research findings to support your arguments and provide valuable insights.
STYLE GUIDE:
{style_guide}
Write a chapter of approximately {self.target_word_count} words in {self.tone} tone.
Make it informative, engaging, and well-structured.""",
agent=self.writer,
expected_output=f"A complete chapter of {self.target_word_count}+ words.",
)
# Task 3: Fact-check
factcheck_task = self.create_task(
description="""Fact-check the entire chapter. Verify:
- All statistics, dates, and numbers
- Quoted statements and attributions
- Scientific claims and studies
- Historical facts and events
- Any potentially controversial claims
Provide a detailed report of any issues found.""",
agent=self.fact_checker,
expected_output="Fact-checked chapter with verified information.",
)
# Task 4: Edit
edit_task = self.create_task(
description=f"""Edit and finalize the chapter. Ensure:
- Clear chapter structure with logical flow
- Strong opening and closing
- Smooth transitions between sections
- Consistent {self.tone} tone throughout
- All fact-check issues addressed
- Publication-ready quality
This is the final polish pass.""",
agent=self.editor,
expected_output="A polished, publication-ready chapter.",
)
self.tasks = [research_task, write_task, factcheck_task, edit_task]
result = self.run(inputs={
"chapter_outline": chapter_outline,
"source_materials": source_materials,
"style_guide": style_guide,
})
return str(result)
def create_nonfiction_crew(
topic: str = "general",
tone: str = "informative",
target_word_count: int = 1000,
verbose: bool = True,
) -> NonfictionCrew:
"""Factory function to create a nonfiction crew.
Args:
topic: Main topic/subject
tone: Writing tone
target_word_count: Target word count
verbose: Enable verbose output
Returns:
Configured NonfictionCrew instance
"""
return NonfictionCrew(
topic=topic,
tone=tone,
target_word_count=target_word_count,
verbose=verbose,
)