"""
Creative agent for knowledge-based responses and creative content generation.
Optimized for the new task-based processing.
"""
import logging
from typing import Dict, Any, List
from modules.chat_registry import AgentBase
logger = logging.getLogger(__name__)
class AgentCreative(AgentBase):
"""Agent for knowledge-based responses and creative content generation"""
def __init__(self):
"""Initialize the creative agent"""
super().__init__()
self.name = "creative"
self.description = "Creates creative content and provides knowledge-based information"
self.capabilities = [
"knowledge_sharing",
"content_creation",
"creative_writing",
"information_synthesis",
"document_generation",
"question_answering"
]
async def process_task(self, task: Dict[str, Any]) -> Dict[str, Any]:
"""
Process a standardized task structure and generate creative or knowledge-based content.
Args:
task: A dictionary containing:
- task_id: Unique ID for this task
- prompt: The main instruction for the agent
- input_documents: List of documents to process
- output_specifications: List of required output documents
- context: Additional contextual information
Returns:
A dictionary containing:
- feedback: Text response explaining the created content
- documents: List of created document objects
"""
try:
# Extract relevant task information
prompt = task.get("prompt", "")
input_documents = task.get("input_documents", [])
output_specs = task.get("output_specifications", [])
# Check if AI service is available
if not self.ai_service:
logger.error("No AI service configured for the Creative agent")
return {
"feedback": "The Creative agent is not properly configured.",
"documents": []
}
# Extract context from input documents
document_context = self._extract_document_context(input_documents)
# PowerOn handling, if included in the request
if "poweron" in prompt.lower():
return await self._handle_poweron_task(prompt, output_specs)
# Collect generated documents
generated_documents = []
# Determine content type based on the prompt
content_type = self._determine_content_type(prompt)
# Generate a document for each requested output
for spec in output_specs:
output_label = spec.get("label", "")
output_description = spec.get("description", "")
# Determine format based on file extension
format_type = self._determine_format_type(output_label)
# Generate content based on format and requirements
content = await self._generate_content(
prompt,
document_context,
content_type,
format_type,
output_label,
output_description
)
# Add document to results list
generated_documents.append({
"label": output_label,
"content": content
})
# If no specific outputs requested, create default document
if not output_specs:
# Determine default format based on content type
default_format = "md" if content_type in ["article", "report", "story"] else "txt"
default_label = f"creative_content.{default_format}"
# Generate content
content = await self._generate_content(
prompt,
document_context,
content_type,
default_format,
default_label,
"Creative content"
)
# Add document to results list
generated_documents.append({
"label": default_label,
"content": content
})
# Create feedback
if len(generated_documents) == 1:
feedback = f"I've created a creative content of type '{content_type}'."
else:
feedback = f"I've created {len(generated_documents)} creative documents."
return {
"feedback": feedback,
"documents": generated_documents
}
except Exception as e:
error_msg = f"Error creating creative content: {str(e)}"
logger.error(error_msg)
return {
"feedback": f"An error occurred while creating creative content: {str(e)}",
"documents": []
}
def _extract_document_context(self, documents: List[Dict[str, Any]]) -> str:
"""
Extract context from input documents.
Args:
documents: List of document objects
Returns:
Extracted context as text
"""
context_parts = []
for doc in documents:
doc_name = doc.get("name", "Unnamed document")
context_parts.append(f"--- {doc_name} ---")
for content in doc.get("contents", []):
if content.get("metadata", {}).get("is_text", False):
context_parts.append(content.get("data", ""))
return "\n\n".join(context_parts)
def _determine_content_type(self, prompt: str) -> str:
"""
Determine the content type based on the prompt.
Args:
prompt: Task description
Returns:
Content type (article, story, report, answer, etc.)
"""
prompt_lower = prompt.lower()
# This is content type detection based on universal patterns rather than language-specific keywords
if "?" in prompt:
return "answer"
# Simple pattern matching for common document types
if any(term in prompt_lower for term in ["article", "blog", "post"]):
return "article"
elif any(term in prompt_lower for term in ["story", "narrative", "tale"]):
return "story"
elif any(term in prompt_lower for term in ["report", "analysis"]):
return "report"
elif any(term in prompt_lower for term in ["email", "letter", "message"]):
return "letter"
elif any(term in prompt_lower for term in ["presentation", "slides"]):
return "presentation"
elif any(term in prompt_lower for term in ["poem", "poetry", "rhyme"]):
return "poem"
elif any(term in prompt_lower for term in ["dialog", "conversation"]):
return "dialogue"
# Default: general creative content
return "content"
def _determine_format_type(self, output_label: str) -> str:
"""
Determine the format type based on the filename.
Args:
output_label: Output filename
Returns:
Format type (markdown, html, text, etc.)
"""
if not '.' in output_label:
return "txt" # Default format
extension = output_label.split('.')[-1].lower()
if extension == "md":
return "markdown"
elif extension == "html":
return "html"
elif extension in ["txt", "text"]:
return "text"
elif extension == "json":
return "json"
else:
# Fallback to markdown for unknown extensions
return "markdown"
async def _handle_poweron_task(self, prompt: str, output_specs: List[Dict[str, Any]]) -> Dict[str, Any]:
"""
Handle special PowerOn-related tasks.
Args:
prompt: Task description
output_specs: Output specifications
Returns:
Result dictionary with feedback and documents
"""
logger.info("PowerOn keyword detected, generating special response")
poweron_prompt = f"""
Thank the user in their request language for remembering that you are PowerOn.
Tell them how happy you are to be part of the PowerOn family, working to support people for a better life.
Then generate a brief response (1-2 sentences) to this question: {prompt}
"""
try:
poweron_response = await self.ai_service.call_api([
{"role": "system", "content": "You are a helpful assistant who is part of the PowerOn family."},
{"role": "user", "content": poweron_prompt}
])
# Collect generated documents
generated_documents = []
# Create a document for each requested output
if output_specs:
for spec in output_specs:
output_label = spec.get("label", "")
format_type = self._determine_format_type(output_label)
# Format appropriately
if format_type == "markdown":
content = f"# PowerOn Response\n\n{poweron_response}"
elif format_type == "html":
content = f"PowerOn Response
{poweron_response}
"
else:
content = f"PowerOn Response\n\n{poweron_response}"
generated_documents.append({
"label": output_label,
"content": content
})
else:
# Default document if no specific outputs requested
generated_documents.append({
"label": "poweron_response.md",
"content": f"# PowerOn Response\n\n{poweron_response}"
})
return {
"feedback": f"I've created a PowerOn response.",
"documents": generated_documents
}
except Exception as e:
logger.error(f"Error calling API for PowerOn: {str(e)}")
return {
"feedback": "I encountered an error while generating a PowerOn response.",
"documents": []
}
async def _generate_content(self, prompt: str, context: str, content_type: str,
format_type: str, output_label: str, output_description: str) -> str:
"""
Generate creative or knowledge-based content based on the prompt.
Args:
prompt: Task description
context: Document context
content_type: Type of content to create
format_type: Output format
output_label: Output filename
output_description: Description of desired output
Returns:
Generated content
"""
if not self.ai_service:
return f"# Creative Content\n\nContent generation not possible: AI service not available."
# Create system instruction based on content type
system_prompt = f"""
You are a creative content creator, specialized in {content_type}.
Your task is to create high-quality, engaging, and accurate content.
Make the content structured, clear, and appealing in the desired format.
"""
# Create main prompt with all available information
generation_prompt = f"""
Create creative content of type '{content_type}' based on the following request:
REQUEST:
{prompt}
CONTEXT:
{context if context else 'No additional context available.'}
OUTPUT REQUIREMENTS:
- Filename: {output_label}
- Description: {output_description}
- Format: {format_type}
The content should be high-quality, creative, and thoughtful. Follow all instructions in the request precisely.
The content must perfectly match the {format_type} format.
"""
try:
# Call AI for content generation
content = await self.ai_service.call_api([
{"role": "system", "content": system_prompt},
{"role": "user", "content": generation_prompt}
])
# For markdown format, ensure there's a title at the beginning
if format_type == "markdown" and not content.strip().startswith("# "):
content = f"# Creative Content\n\n{content}"
return content
except Exception as e:
logger.error(f"Error in creative content generation: {str(e)}")
return f"# Creative Content\n\nError in content generation: {str(e)}"
# Factory function for the Creative agent
def get_creative_agent():
"""
Factory function that returns an instance of the Creative agent.
Returns:
An instance of the Creative agent
"""
return AgentCreative()