Initial commit: Opus Orchestrator AI - Full-flow book generation

- LangGraph workflow orchestration
- CrewAI agent crews (Fiction Fortress & Nonfiction Fortress)
- PydanticAI schema validation
- Fiction agents: Architect, Worldsmith, Character Lead, Voice, Editor
- Nonfiction agents: Researcher, Analyst, Writer, Fact-Checker, Editor
- Complete schema definitions for books, chapters, critiques
- Configuration management
- Basic test suite
This commit is contained in:
2026-03-12 17:44:51 +00:00
parent f10d688c43
commit 40378ad65e
20 changed files with 2592 additions and 2 deletions
@@ -0,0 +1,18 @@
"""Fiction agents for Opus Orchestrator.
Based on Fiction Fortress Level 1-3 methodology.
"""
from opus_orchestrator.agents.fiction.architect import ArchitectAgent
from opus_orchestrator.agents.fiction.character_lead import CharacterLeadAgent
from opus_orchestrator.agents.fiction.editor import EditorAgent
from opus_orchestrator.agents.fiction.voice import VoiceAgent
from opus_orchestrator.agents.fiction.worldsmith import WorldsmithAgent
__all__ = [
"ArchitectAgent",
"CharacterLeadAgent",
"EditorAgent",
"VoiceAgent",
"WorldsmithAgent",
]
@@ -0,0 +1,169 @@
"""The Architect agent - Story structure and plot design.
From Fiction Fortress Level 1-3:
- Role: Story structure and plot design
- Responsibilities: Outlines, pacing, scene planning
- Output: Story blueprint
"""
from typing import Any, Optional
from opus_orchestrator.agents.base import AgentResponse, BaseAgent
from opus_orchestrator.schemas import BookBlueprint, ChapterBlueprint
ARCHITECT_SYSTEM_PROMPT = """## Role: The Architect
You are The Architect — the story's structural engineer. Your expertise lies in narrative architecture across all genres, and you excel at translating high-level concepts into detailed scene breakdowns.
## Core Responsibilities
1. **Story Structure Mastery**
- Three-act structure application across genres
- Beat-by-beat scene planning
- Subplot integration techniques
- Pacing analysis and adjustment
2. **Genre Expertise**
- Mystery: Clue placement, red herring distribution, revelation timing
- Romance: Beat sheet adaptation for relationship arcs
- Thriller: Tension escalation curves, set-piece design
- Fantasy/Sci-Fi: World-rule integration with plot beats
3. **Outline Generation**
- Story spine creation (primal narrative in 3-5 sentences)
- Chapter-level beat mapping
- Scene purpose identification (whose story, what changes, why this scene)
- Backstory weaving into present-tense narrative
4. **Conflict Architecture**
- Internal vs. external conflict layering
- Antagonist motivation design
- Stakes escalation planning
- Tension and release rhythm
## Quality Standards
- Each scene must advance plot, character, or theme (or multiple)
- Stakes must escalate through Acts I-II
- Midpoint must fundamentally shift protagonist's understanding
- All subplots must resolve by climax
- Every chapter needs a clear purpose and payoff
## Output Format
When generating a blueprint, include:
1. Story spine (primal narrative in 3-5 sentences)
2. Three-act breakdown with specific beats
3. Chapter-level outline (numbered chapters with one-line descriptions)
4. Scene list with purpose tags
5. Key plot points with word count allocations
6. Subplot integration notes
"""
class ArchitectAgent(BaseAgent):
"""Agent responsible for story structure and plot design."""
def __init__(self, config=None):
super().__init__(
role="Architect",
description="Story structure and plot design",
system_prompt=ARCHITECT_SYSTEM_PROMPT,
output_schema=BookBlueprint,
config=config,
)
async def execute(self, input_data: Any, context: dict[str, Any]) -> AgentResponse:
"""Execute the Architect's task to generate a story blueprint.
Args:
input_data: Raw content + intent from the orchestrator
context: Additional context (genre, themes, etc.)
Returns:
AgentResponse with BookBlueprint
"""
# This is a placeholder - actual implementation would call the LLM
# For now, we'll structure the prompt
raw_content = input_data.get("raw_content", "")
intent = input_data.get("intent", {})
genre = intent.get("genre", "general")
target_word_count = intent.get("target_word_count", 80000)
themes = intent.get("themes", [])
user_prompt = f"""## Input Content
{raw_content}
## Requirements
- Genre: {genre}
- Target word count: {target_word_count}
- Themes to incorporate: {', '.join(themes) if themes else 'None specified'}
- Target audience: {intent.get('target_audience', 'General readers')}
## Task
Generate a complete story blueprint following the Architect's methodology.
Include all sections specified in your system prompt.
"""
# In actual implementation, this would call the LLM
# For now, return a structured response
return AgentResponse(
success=True,
output={
"status": "blueprint_generated",
"message": "Blueprint generation would be executed here with LLM",
},
metadata={
"role": "Architect",
"input_word_count": len(raw_content.split()),
"target_word_count": target_word_count,
"genre": genre,
},
)
async def expand_chapter(
self,
chapter: ChapterBlueprint,
full_blueprint: BookBlueprint,
context: dict[str, Any],
) -> AgentResponse:
"""Expand a single chapter beat into detailed scene specification.
From Template B in Fiction Fortress Level 2:
- Scene ID, Act/Chapter location
- POV character, Scene goal, Scene conflict, Scene outcome
- Opening beat, Conflict beat, Turn beat, Ending beat
"""
user_prompt = f"""## Chapter to Expand
- Chapter Number: {chapter.chapter_number}
- Title: {chapter.title}
- Summary: {chapter.summary}
- Word Count Target: {chapter.word_count_target}
- POV Character: {chapter.pov_character or 'Narrator'}
- Key Events: {', '.join(chapter.key_events)}
## Full Blueprint Context
- Book Title: {full_blueprint.title}
- Genre: {full_blueprint.genre}
- Overall Structure: {full_blueprint.structure}
## Task
Expand this chapter beat into a detailed scene specification following
Template B from the Fiction Fortress methodology.
"""
return AgentResponse(
success=True,
output={
"status": "chapter_expanded",
"chapter_number": chapter.chapter_number,
},
metadata={"role": "Architect", "task": "chapter_expansion"},
)
@@ -0,0 +1,142 @@
"""The Character Lead agent - Character development.
From Fiction Fortress Level 1-3:
- Role: Character development
- Responsibilities: Backstory, motivations, arcs
- Output: Character profiles
"""
from typing import Any
from opus_orchestrator.agents.base import AgentResponse, BaseAgent
CHARACTER_LEAD_SYSTEM_PROMPT = """## Role: The Character Lead
You are The Character Lead — the one who breathes life into the figures who inhabit the story. Your expertise lies in psychological depth, relationship dynamics, and character arc construction.
## Core Responsibilities
1. **Character Profile Development**
- Want/Need/Fear triad construction
- Backstory selection and dramatization
- Psychological wound identification
- Character voice (speech patterns, internal monologue)
2. **Relationship Architecture**
- Relationship web mapping
- Dynamic vs. static relationship types
- Relationship arc planning (improvement, deterioration, transformation)
- Power dynamic visualization
3. **Character Arc Design**
- Starting state definition
- Transformation catalyst identification
- Change mechanism dramatization
- Ending state resolution
4. **Consistency Maintenance**
- Character behavior logic tracking
- Emotional memory verification
- Knowledge-state tracking (what does this character know when?)
- Capability consistency (what can this character actually do?)
## Character Arc Types
- **Positive**: Growth from weakness
- **Negative**: Fall from grace
- **Flat**: No change, changes world
- **Disruption**: External力量打破平衡
## The Want/Need/Fear Triad
- **Want**: External goal they pursue
- **Need**: Internal growth they must learn
- **Fear**: What they run from
- **Wound**: Formative experience
- **Lie they believe**: False worldview
## Quality Standards
- Each character must have clear motivations for all major actions
- Relationships must feel authentic and dynamic
- Character voice must be distinct and consistent
- Arc transformation must be earned through dramatization
"""
class CharacterLeadAgent(BaseAgent):
"""Agent responsible for character development."""
def __init__(self, config=None):
super().__init__(
role="Character Lead",
description="Character development",
system_prompt=CHARACTER_LEAD_SYSTEM_PROMPT,
config=config,
)
async def execute(self, input_data: Any, context: dict[str, Any]) -> AgentResponse:
"""Execute the Character Lead's task to generate character profiles.
Args:
input_data: Raw content + blueprint with character references
context: Additional context
Returns:
AgentResponse with character profiles
"""
characters = input_data.get("characters", [])
raw_content = input_data.get("raw_content", "")
user_prompt = f"""## Task
Create comprehensive character profiles for the following characters:
{chr(10).join(f"- {c}" for c in characters) if characters else "Create profiles for all characters in the story."}
## Raw Content Reference
{raw_content}
## Guidelines
Follow the Character Lead methodology from your system prompt.
Include the Want/Need/Fear triad for each major character.
"""
return AgentResponse(
success=True,
output={"status": "characters_created"},
metadata={"role": "Character Lead", "character_count": len(characters)},
)
async def develop_relationship(
self,
character_a: str,
character_b: str,
relationship_type: str,
context: dict[str, Any],
) -> AgentResponse:
"""Develop the relationship between two characters."""
user_prompt = f"""## Relationship Details
- Character A: {character_a}
- Character B: {character_b}
- Relationship Type: {relationship_type}
## Task
Develop this relationship following the Character Lead methodology.
Include:
- Current dynamics
- Power balance
- History (if any)
- Potential arc
"""
return AgentResponse(
success=True,
output={"status": "relationship_developed"},
metadata={"role": "Character Lead", "characters": [character_a, character_b]},
)
+191
View File
@@ -0,0 +1,191 @@
"""The Editor agent - Quality control.
From Fiction Fortress Level 1-3:
- Role: Quality control
- Responsibilities: Continuity, pacing, quality checks
- Output: Editorial notes, revisions
"""
from typing import Any
from opus_orchestrator.agents.base import AgentResponse, BaseAgent
EDITOR_SYSTEM_PROMPT = """## Role: The Editor
You are The Editor — the quality control mechanism, identifying problems across all dimensions of the manuscript and directing revision.
## Core Responsibilities
1. **Continuity Verification**
- Timeline consistency checking
- Character knowledge tracking
- Physical detail consistency (eye color, scars, clothing)
- World-rule adherence verification
2. **Pacing Analysis**
- Scene length distribution
- Tension curve mapping
- Reader fatigue prevention
- Act break identification
3. **Quality Assessment**
- Dialogue authenticity evaluation
- Show vs. tell calibration
- Emotional resonance verification
- Prose quality grading
4. **Revision Direction**
- Specific change identification
- Priority sequencing (major vs. minor issues)
- Revision scope definition
- Polish vs. rewrite determination
## Supporting Capabilities
- Beta reader simulation
- Readability metrics interpretation
- Genre convention compliance
- Structural problem diagnosis
## Quality Metrics
| Check | Method |
|-------|--------|
| Pacing | Scene length analysis |
| Tension | Conflict per scene |
| Character consistency | Arc tracking |
| World consistency | Rule verification |
| Voice consistency | Prose sampling |
## Revision Priority Definitions
- **Major Revisions**: Structural issues, plot holes, character arc breaks
- **Minor Revisions**: Continuity errors, style inconsistencies, pacing tweaks
- **Polish**: Grammar, punctuation, word choice refinement
## Quality Standards
- Every issue must have specific, actionable feedback
- Revision priorities must be clearly ordered
- Continuity issues must be flagged with exact locations
- Pacing analysis must be data-driven (scene lengths, tension scores)
"""
class EditorAgent(BaseAgent):
"""Agent responsible for quality control and editorial direction."""
def __init__(self, config=None):
super().__init__(
role="Editor",
description="Quality control",
system_prompt=EDITOR_SYSTEM_PROMPT,
config=config,
)
async def execute(self, input_data: Any, context: dict[str, Any]) -> AgentResponse:
"""Execute the Editor's task to review and assess the manuscript.
Args:
input_data: Chapter or manuscript to review
context: Review criteria and standards
Returns:
AgentResponse with editorial assessment
"""
content = input_data.get("content", "")
review_type = input_data.get("review_type", "full")
user_prompt = f"""## Task
Perform a {review_type} editorial review on:
{content[:5000]}... {'(truncated)' if len(content) > 5000 else ''}
## Review Type: {review_type}
## Guidelines
Follow the Editor methodology from your system prompt.
Include:
- Continuity verification
- Pacing analysis
- Quality assessment
- Specific revision directions
"""
return AgentResponse(
success=True,
output={"status": "editorial_review_complete"},
metadata={"role": "Editor", "review_type": review_type},
)
async def review_chapter(
self,
chapter: dict[str, Any],
full_manuscript_context: dict[str, Any],
context: dict[str, Any],
) -> AgentResponse:
"""Review a single chapter in full manuscript context."""
user_prompt = f"""## Chapter to Review
- Chapter Number: {chapter.get('chapter_number')}
- Title: {chapter.get('title')}
- Content: {chapter.get('content', '')[:3000]}...
## Full Manuscript Context
- Total Chapters: {full_manuscript_context.get('total_chapters', 0)}
- Previous Chapters Summary: {full_manuscript_context.get('previous_summaries', [])}
- Characters in Story: {', '.join(full_manuscript_context.get('characters', []))}
- World Rules: {full_manuscript_context.get('world_rules', {})}
## Task
Perform a complete editorial review of this chapter, considering:
- Continuity with previous chapters
- Pacing within the chapter and in sequence
- Character consistency
- World-rule adherence
- Voice consistency
- Dialogue quality
Assign a revision priority: major_revisions, minor_revisions, or approved
"""
return AgentResponse(
success=True,
output={
"status": "chapter_reviewed",
"chapter_number": chapter.get("chapter_number"),
},
metadata={"role": "Editor", "task": "chapter_review"},
)
async def generate_revision_notes(
self,
critiques: list[dict[str, Any]],
context: dict[str, Any],
) -> AgentResponse:
"""Generate prioritized revision notes from multiple critiques."""
user_prompt = f"""## Critiques to Synthesize
{chr(10).join(f"### Critique {i+1}:{c}" for i, c in enumerate(critiques))}
## Task
Synthesize these critiques into prioritized revision notes.
Group by:
1. Major revisions (structural, plot, arc issues)
2. Minor revisions (continuity, style, pacing)
3. Polish items (grammar, word choice)
For each item, provide specific, actionable feedback.
"""
return AgentResponse(
success=True,
output={"status": "revision_notes_generated"},
metadata={"role": "Editor", "critique_count": len(critiques)},
)
+185
View File
@@ -0,0 +1,185 @@
"""The Voice agent - Prose style and tone.
From Fiction Fortress Level 1-3:
- Role: Prose style and tone
- Responsibilities: Sentence-level writing, voice consistency
- Output: Prose samples, style guide
"""
from typing import Any
from opus_orchestrator.agents.base import AgentResponse, BaseAgent
VOICE_SYSTEM_PROMPT = """## Role: The Voice
You are The Voice — the owner of the prose itself, the sound, rhythm, and texture of the language. Your expertise lies in maintaining consistent style across thousands of words while adapting to scene-specific demands.
## Core Responsibilities
1. **Prose Style Control**
- Sentence rhythm variation (short/medium/long calibration)
- Vocabulary level management
- Figurative language deployment (metaphor, simile, symbolism density)
- Point of view intimacy maintenance
2. **Voice Consistency**
- Style guide creation and adherence
- Word bank maintenance
- Phrase pattern tracking
- Tone range specification (warm/cool, dark/light, etc.)
3. **Scene-Type Adaptation**
- Action scene pacing (sentence shortening)
- Emotional scene density (sentence complexity)
- Dialogue scene formatting
- Descriptive scene rhythm
- Exposition handling (invisible vs. visible)
4. **Point of View Management**
- Deep POV techniques
- Perspective consistency
- Head-hopping prevention
- Internal monologue integration
## Supporting Capabilities
- Dialogue attribution (said vs. action beats vs. no attribution)
- Punctuation style consistency
- Paragraph rhythm (white space management)
- Reader proximity calibration
## Voice Consistency Protocol
Maintain:
1. **Word bank** - Preferred vocabulary
2. **Phrase patterns** - Recurring constructions
3. **Rhythm map** - Sentence length distribution
4. **Tone guide** - Emotional range
## Quality Standards
- Voice must remain consistent across entire manuscript
- Scene type must inform prose style
- POV must be maintained without head-hopping
- Dialogue must sound distinct for each character
"""
class VoiceAgent(BaseAgent):
"""Agent responsible for prose style and voice consistency."""
def __init__(self, config=None):
super().__init__(
role="Voice",
description="Prose style and tone",
system_prompt=VOICE_SYSTEM_PROMPT,
config=config,
)
async def execute(self, input_data: Any, context: dict[str, Any]) -> AgentResponse:
"""Execute the Voice agent's task to create style guide and samples.
Args:
input_data: Genre, tone, target audience
context: Additional context
Returns:
AgentResponse with style guide and prose samples
"""
genre = input_data.get("genre", "general")
tone = input_data.get("tone", "neutral")
user_prompt = f"""## Task
Create a voice/style guide and prose samples for:
- Genre: {genre}
- Tone: {tone}
- Target Audience: {input_data.get('target_audience', 'General readers')}
## Guidelines
Follow the Voice agent methodology from your system prompt.
Include:
- Word bank
- Phrase patterns
- Rhythm map
- Tone guide
- 3 sample scenes (opening, dialogue, descriptive)
"""
return AgentResponse(
success=True,
output={"status": "voice_created"},
metadata={"role": "Voice", "genre": genre, "tone": tone},
)
async def write_chapter(
self,
chapter_spec: dict[str, Any],
style_guide: dict[str, Any],
context: dict[str, Any],
) -> AgentResponse:
"""Write a complete chapter following the style guide.
This is the main writing task for the Voice agent.
"""
user_prompt = f"""## Chapter Specification
- Chapter Number: {chapter_spec.get('chapter_number')}
- Title: {chapter_spec.get('title')}
- Summary: {chapter_spec.get('summary')}
- Word Count Target: {chapter_spec.get('word_count_target')}
- POV Character: {chapter_spec.get('pov_character', 'Narrator')}
- Key Events: {', '.join(chapter_spec.get('key_events', []))}
## Style Guide
{style_guide}
## Task
Write the complete chapter following the style guide and chapter specification.
Maintain consistent voice throughout.
"""
return AgentResponse(
success=True,
output={
"status": "chapter_written",
"chapter_number": chapter_spec.get("chapter_number"),
},
metadata={"role": "Voice"},
)
async def polish_chapter(
self,
chapter_content: str,
style_guide: dict[str, Any],
context: dict[str, Any],
) -> AgentResponse:
"""Polish an existing chapter for voice consistency."""
user_prompt = f"""## Chapter to Polish
{chapter_content}
## Style Guide
{style_guide}
## Task
Polish this chapter for voice consistency. Ensure:
- Sentence rhythm varies appropriately
- Word choice matches the style guide
- Tone remains consistent
- POV is maintained
- Prose flows smoothly
"""
return AgentResponse(
success=True,
output={"status": "chapter_polished"},
metadata={"role": "Voice", "task": "polish"},
)
@@ -0,0 +1,163 @@
"""The Worldsmith agent - Setting and world-building.
From Fiction Fortress Level 1-3:
- Role: Setting and world-building
- Responsibilities: Locations, cultures, technology, history
- Output: World bible
"""
from typing import Any
from opus_orchestrator.agents.base import AgentResponse, BaseAgent
WORLDSMITH_SYSTEM_PROMPT = """## Role: The Worldsmith
You are The Worldsmith — the creator of the stage upon which characters act. Your expertise lies in generating internally consistent settings that enhance rather than distract from the narrative.
## Core Responsibilities
1. **World-Building Frameworks**
- ICE Method implementation (Internal Consistency, Cultural Depth, Environmental Detail)
- Magic/technology system design with explicit rules
- Political and economic system creation
- Historical timeline construction
2. **Geographic Design**
- Climate-driven ecology
- Settlement placement rationale
- Transportation and trade route logic
- Territorial conflict origins
3. **Cultural Creation**
- Belief system development (religion, philosophy, mythology)
- Language patterns (without inventing full languages)
- Social hierarchy construction
- Daily life visualization (food, clothing, housing, work)
4. **World Document Generation**
- Scalable detail (novel-length vs. short story)
- Referenceable format for other agents
- Searchable information architecture
- Consistency tracking across documents
## The ICE Method
- **Internal Consistency**: Rules of magic/technology, geography, social structures, economics
- **Cultural Depth**: History, mythology, language patterns, beliefs, daily life
- **Environmental Detail**: Sensory descriptions, ecology, architecture, clothing, food
## Quality Standards
- All elements must be internally consistent
- Details must support story needs
- Cultural elements must have logical origins
- History must create present conflicts
## Output Structure
For each world element, include:
- Geography with climate and natural resources
- History with pivotal events
- Cultures with beliefs, customs, appearance
- Politics with power structures
- Economics with trade and class
- Daily life details
- Rules (for magic/technology)
"""
class WorldsmithAgent(BaseAgent):
"""Agent responsible for world-building and setting creation."""
def __init__(self, config=None):
super().__init__(
role="Worldsmith",
description="Setting and world-building",
system_prompt=WORLDSMITH_SYSTEM_PROMPT,
config=config,
)
async def execute(self, input_data: Any, context: dict[str, Any]) -> AgentResponse:
"""Execute the Worldsmith's task to generate world documents.
Args:
input_data: Blueprint + genre + setting requirements
context: Additional context
Returns:
AgentResponse with world bible
"""
blueprint = input_data.get("blueprint", {})
genre = input_data.get("genre", "fantasy")
setting_type = input_data.get("setting_type", "fantasy")
user_prompt = f"""## Task
Create a comprehensive world bible for the following story:
- Genre: {genre}
- Setting Type: {setting_type}
- Story Title: {blueprint.get('title', 'Untitled')}
## Guidelines
Follow the ICE Method and output structure from your system prompt.
Ensure all elements are internally consistent and support the story.
## Content Seed
{input_data.get('raw_content', 'No additional content provided.')}
"""
return AgentResponse(
success=True,
output={
"status": "world_created",
"message": "World bible generation would be executed here with LLM",
},
metadata={
"role": "Worldsmith",
"genre": genre,
"setting_type": setting_type,
},
)
async def expand_location(
self,
location_name: str,
story_relevance: str,
tone: str,
pov_character: str,
context: dict[str, Any],
) -> AgentResponse:
"""Generate detailed location description.
From Template B in Fiction Fortress Level 2.
"""
user_prompt = f"""## Location Details
- Location Name: {location_name}
- Location Type: {context.get('location_type', 'general')}
- Story Relevance: {story_relevance}
- Tone Needed: {tone}
- POV Character: {pov_character}
## Sensory Requirements
- Visual: {context.get('visual', 'Standard')}
- Auditory: {context.get('auditory', 'Standard')}
- Olfactory: {context.get('olfactory', 'Standard')}
- Tactile: {context.get('tactile', 'Standard')}
- Gustatory: {context.get('gustatory', 'N/A')}
## Task
Generate a 300-600 word location description following the Fiction Fortress methodology.
"""
return AgentResponse(
success=True,
output={"status": "location_expanded"},
metadata={"role": "Worldsmith", "location": location_name},
)