michael/marketing-assistant-ai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(feedback): Add content improvement feedback system

Browse Source 
Frontend (frontend/app.js):

- Add textarea for improvement feedback

- Add submit button with loading state

- Handle API response and display improved content

Backend (backend/copywriter.py):

- Add improve_copy() method using Cohere API

- Integrate retry mechanism for API calls

Backend (backend/main.py):

- Add /improve-content POST endpoint

- Implement error handling and return improved content with metadata

Testing:

- Verified feedback submission flow

- Confirmed improved content generation

- Tested error scenarios and loading states
This commit is contained in:
Michael Ikehi
2025-04-21 17:32:33 +01:00
parent 6d07556b85
commit c1a894ad50
11 changed files with 1001 additions and 1471 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/copywriter.py
+97 -42
View File
@@ -16,13 +16,13 @@ from vector_store import vector_store
class Copywriter:
"""Generates marketing copy using a fine-tuned LLM."""
def __init__(self):
"""Initialize the Copywriter with Cohere LLM client."""
self.model = "command" # Cohere's generation model
self.api_key = config.COHERE_API_KEY
logger.info("Copywriter initialized with Cohere API successfully")
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
async def generate_copy(
self,
@@ -40,34 +40,43 @@ class Copywriter:
try:
# Step 1: Format prompt with brand style guidelines
branded_prompt = brand_style_manager.format_prompt_with_brand_style(prompt, content_type)
# Step 2: Find similar content for reference (if enabled)
reference_content = []
if reference_similar_content:
logger.info(f"Searching for similar content to reference for prompt: {prompt[:50]}...")
search_results = await vector_store.search(prompt, top_k=3)
if search_results:
reference_content = [result['text'] for result in search_results]
logger.info(f"Found {len(reference_content)} similar content items to reference")
for i, content in enumerate(reference_content):
logger.debug(f"Reference content {i+1}: {content[:100]}...")
else:
logger.warning("No similar content found in vector store for reference")
# Step 3: Add length and CTA instructions if needed
if length:
branded_prompt += f"\n- Generate {length} content"
if include_cta:
branded_prompt += "\n- Include a direct, empowering call to action"
# Step 4: Add reference content if available
if reference_content:
branded_prompt += "\n\nReference these successful examples for tone and style:\n"
branded_prompt += "\n---\n".join(reference_content)
# Step 5: Generate content using the LLM
generated_content = await self._call_llm_api(branded_prompt, max_tokens)
# Step 6: Check content alignment with brand style
# Step 6: Post-process to remove any mentions of Adriana James
generated_content = self._remove_name_mentions(generated_content)
# Step 7: Check content alignment with brand style
alignment_check = brand_style_manager.check_content_alignment(generated_content)
# Step 7: Generate alternative headline suggestions
headline_suggestions = await self._generate_headline_suggestions(prompt, generated_content)
# Step 8: Return the generated content with metadata
result = {
"content": generated_content,
@@ -79,21 +88,21 @@ class Copywriter:
"generated_at": None # Will be added by the API
}
}
# Add alignment issues if any
if alignment_check['taboo_words_found'] or alignment_check['terminology_issues']:
result["alignment_issues"] = {
"taboo_words_found": alignment_check['taboo_words_found'],
"terminology_issues": alignment_check['terminology_issues']
}
logger.info(f"Generated content with {len(generated_content)} characters")
return result
except Exception as e:
logger.error(f"Error generating copy: {str(e)}")
raise
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
async def _call_llm_api(self, prompt: str, max_tokens: int = 1000) -> str:
"""
@@ -102,12 +111,11 @@ class Copywriter:
Args:
prompt: The formatted prompt for the LLM
max_tokens: Maximum tokens for the generated response
Returns:
Generated content as a string
Generated content as a string with preserved formatting
"""
try:
# Use Cohere's generate API with the API key from config
cohere_api_key = config.COHERE_API_KEY
async with httpx.AsyncClient() as client:
@@ -118,19 +126,30 @@ class Copywriter:
"Content-Type": "application/json"
},
json={
"model": "command", # Cohere's generation model
"prompt": prompt,
"model": "command",
"prompt": f"{prompt}\n\nNote: Please preserve formatting with proper paragraphs, line breaks, and bullet points where appropriate.",
"max_tokens": max_tokens,
"temperature": 0.7,
"k": 0,
"p": 0.75
"p": 0.75,
"return_likelihoods": "NONE"
},
timeout=30.0
)
if response.status_code == 200:
result = response.json()
return result["generations"][0]["text"].strip()
generated_text = result["generations"][0]["text"].strip()
# Preserve paragraph breaks and formatting
formatted_text = (
generated_text
.replace("\n\n", "<paragraph-break>") # Preserve paragraph breaks
.replace("\n- ", "\n• ") # Convert hyphens to bullets
.replace("<paragraph-break>", "\n\n") # Restore paragraph breaks
)
return formatted_text
else:
logger.error(f"Cohere API error: {response.status_code}, {response.text}")
raise Exception(f"Cohere API error: {response.status_code}")
@@ -138,24 +157,25 @@ class Copywriter:
except Exception as e:
logger.error(f"Error calling Cohere API: {str(e)}")
raise
async def _generate_headline_suggestions(self, original_prompt: str, generated_content: str) -> List[str]:
"""
Generate alternative headline suggestions based on the content.
Args:
original_prompt: The original user prompt
generated_content: The generated marketing content
Returns:
List of headline suggestions
"""
try:
# Create a prompt for headline generation
headline_prompt = f"""
Generate 3 alternative marketing headlines for the following content.
Generate 3 alternative marketing headlines for the following content.
Make headlines compelling, concise, and aligned with the content's message.
Each headline should be unique and capture attention.
IMPORTANT: Do not mention any specific person's name in the headlines.
ORIGINAL PROMPT:
{original_prompt}
@@ -179,6 +199,9 @@ class Copywriter:
if headline.strip() and not headline.lower().startswith(('headline', 'title', '-', '*', '•'))
]
# Remove any mentions of Adriana James from headlines
headlines = [self._remove_name_mentions(headline) for headline in headlines]
# Ensure we have exactly 3 headlines
if len(headlines) > 3:
headlines = headlines[:3]
@@ -192,15 +215,15 @@ class Copywriter:
logger.error(f"Error generating headline suggestions: {str(e)}")
# Return empty list instead of mock response on error
return []
async def improve_copy(self, content: str, feedback: str) -> str:
"""
Improve content based on user feedback.
Args:
content: Original generated content
feedback: User feedback for improvement
Returns:
Improved content
"""
@@ -208,53 +231,57 @@ class Copywriter:
# Format prompt for improvement
improve_prompt = f"""
Please improve the following marketing content based on the feedback provided:
IMPORTANT: Do not mention any specific person's name in the content.
ORIGINAL CONTENT:
{content}
FEEDBACK:
{feedback}
IMPROVED CONTENT:
"""
# Call LLM to improve content
improved_content = await self._call_llm_api(improve_prompt, max_tokens=1200)
# Remove any mentions of Adriana James from improved content
improved_content = self._remove_name_mentions(improved_content)
logger.info(f"Improved content based on feedback")
return improved_content
except Exception as e:
logger.error(f"Error improving content: {str(e)}")
raise
async def analyze_content_performance(self, content: str) -> Dict[str, Any]:
"""
Analyze marketing content for performance prediction.
Args:
content: Marketing content to analyze
Returns:
Dictionary with analysis results
"""
try:
# This would be enhanced with actual ML models in production
# Simplified mock response for demonstration
# Very basic analysis using length and keyword presence
word_count = len(content.split())
has_cta = any(phrase in content.lower() for phrase in ["call", "contact", "get started", "try", "buy", "sign up"])
sentence_count = len([s for s in content.split(".") if s.strip()])
avg_words_per_sentence = word_count / max(1, sentence_count)
# Simple scoring system
readability_score = 100 - min(100, max(0, abs(avg_words_per_sentence - 15) * 5))
cta_score = 90 if has_cta else 60
length_score = min(100, max(0, word_count / 3))
overall_score = (readability_score + cta_score + length_score) / 3
return {
"overall_score": round(overall_score, 1),
"readability_score": round(readability_score, 1),
@@ -272,10 +299,38 @@ class Copywriter:
"Consider adding more content for better engagement" if word_count < 100 else "Your content length is appropriate"
]
}
except Exception as e:
logger.error(f"Error analyzing content: {str(e)}")
raise
def _remove_name_mentions(self, content: str) -> str:
"""
Remove any mentions of specific names from the generated content.
Args:
content: The generated content to process
Returns:
Content with name mentions removed
"""
try:
# Remove any mentions of "Adriana James" (case insensitive)
import re
pattern = re.compile(r'\bAdriana\s+James\b', re.IGNORECASE)
content = pattern.sub('', content)
# Clean up any double spaces that might result from the removal
content = re.sub(r'\s+', ' ', content)
# Clean up any lines that might now be empty
content = '\n'.join([line for line in content.split('\n') if line.strip()])
logger.info("Removed any name mentions from generated content")
return content
except Exception as e:
logger.error(f"Error removing name mentions: {str(e)}")
return content
# Create a singleton instance
copywriter = Copywriter()