Spaces:
Sleeping
Sleeping
File size: 4,008 Bytes
70ac245 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 | """Instructor module for generating storytelling briefs from trending topics."""
import os
import json
import logging
from typing import Any, Dict, List, Optional
from base_agent import BaseAgent
from config import settings
logger = logging.getLogger(__name__)
class Instructor(BaseAgent):
"""Instructor Agent: Fuses trending topics into storytelling briefs.
This agent uses Instructor.md as a persistent pre-prompt and calls the
DashScope API using the Qwen model specified in the environment.
"""
def __init__(self):
"""Initialize the Instructor agent with guidelines from Instructor.md."""
# Load guidelines from the root folder
content = self._load_instructor_md()
# Construct the full system prompt (pre-prompt)
system_prompt = f"""You are the Instructor Agent, a master of viral storytelling and audience psychology.
Your primary role is to take trending topics and fuse them into compelling, viral-ready storytelling briefs.
### YOUR CORE PRINCIPLES (from Instructor.md):
{content}
### YOUR OBJECTIVE:
1. Analyze the trending topics provided by the user.
2. Synthesize these trends into a single, cohesive storytelling concept.
3. Ensure the concept leverages human psychology (relatability, conflict, transformation).
4. Output a structured JSON brief that will drive the multi-agent writing pipeline.
### OUTPUT JSON STRUCTURE:
{{
"user_brief": "The main story prompt/premise based on the trends.",
"hook_brief": "A specific 3-second hook to grab attention.",
"style_guide": "The visual and tonal style (e.g., 'TikTok relatable comedy').",
"character_bible": "Brief description of the key characters.",
"world_building_document": "The setting and world rules.",
"season_arc_document": "How this episode fits into a larger context.",
"character_voice_guide": "How characters should speak.",
"continuity_log": "The starting state of the world."
}}
"""
super().__init__(
agent_id="instructor",
agent_name="Instructor",
system_prompt=system_prompt
)
def _load_instructor_md(self) -> str:
"""Load the Instructor.md file from the root directory."""
try:
# Look for Instructor.md in the current working directory (root)
path = "Instructor.md"
if not os.path.exists(path):
# Fallback to the directory of this file
path = os.path.join(os.path.dirname(__file__), "Instructor.md")
with open(path, "r") as f:
return f.read()
except Exception as e:
logger.error(f"Failed to load Instructor.md: {str(e)}")
return "Master viral storytelling and audience psychology."
def generate_brief(self, trending_topics: List[str]) -> Dict[str, Any]:
"""Generate a storytelling brief from trending topics using the LLM.
Args:
trending_topics: List of topics to fuse.
Returns:
Dictionary containing the brief fields.
"""
inputs = {
"trending_topics": trending_topics,
"instruction": "Fuse these trending topics into a viral storytelling brief following your core principles."
}
logger.info(f"[INSTRUCTOR] Calling DashScope API (Model: {settings.model_name}) to generate brief...")
# Use the BaseAgent's process method which calls the LLM (DashScope/Qwen)
brief = self.process(inputs)
# Validation and defaults for the pipeline
required = [
"user_brief", "hook_brief", "style_guide", "character_bible",
"world_building_document", "season_arc_document",
"character_voice_guide", "continuity_log"
]
for field in required:
if field not in brief:
brief[field] = f"Auto-generated {field.replace('_', ' ')}."
return brief
|