update endpoints
This commit is contained in:
Ayomide
+220
-14
@@ -1,24 +1,230 @@
|
||||
# backend/copywriter.py
|
||||
import json
|
||||
import os
|
||||
from datetime import datetime
|
||||
from typing import Dict, List, Optional
|
||||
from pathlib import Path
|
||||
import openai
|
||||
from .vector_store import VectorStore
|
||||
from .brand_style import BrandStyle
|
||||
import openai
|
||||
from .config import Config
|
||||
|
||||
class Copywriter:
|
||||
def __init__(self):
|
||||
self.vector_store = VectorStore()
|
||||
self.brand_style = BrandStyle()
|
||||
self.user_queries_path = Path("data/user_queries")
|
||||
self.user_queries_path.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Initialize OpenAI
|
||||
openai.api_key = Config.OPENAI_API_KEY
|
||||
|
||||
def generate_copy(self, request):
|
||||
# Move the generation logic from main.py here
|
||||
similar = self.vector_store.search(request.prompt, request.content_type)
|
||||
def generate_copy(self, request) -> Dict:
|
||||
"""Generate marketing copy and log user interaction"""
|
||||
try:
|
||||
# Log the user query first
|
||||
query_log = self._log_user_query(request)
|
||||
|
||||
# Get similar content from vector store
|
||||
similar = self.vector_store.search(request.prompt, request.content_type)
|
||||
|
||||
# Format similar content for context
|
||||
similar_content = ""
|
||||
if similar:
|
||||
similar_content = "\n\nSimilar past campaigns for reference:\n"
|
||||
for i, campaign in enumerate(similar[:3], 1):
|
||||
similar_content += f"{i}. {campaign.get('content', '')}\n"
|
||||
|
||||
# Generate with OpenAI
|
||||
system_prompt = self.brand_style.get_prompt(request)
|
||||
user_prompt = f"Create marketing copy for: {request.prompt}{similar_content}"
|
||||
|
||||
response = openai.ChatCompletion.create(
|
||||
model="gpt-3.5-turbo",
|
||||
messages=[
|
||||
{"role": "system", "content": system_prompt},
|
||||
{"role": "user", "content": user_prompt}
|
||||
],
|
||||
temperature=0.7,
|
||||
max_tokens=500
|
||||
)
|
||||
|
||||
generated_copy = response.choices[0].message.content
|
||||
|
||||
# Update the query log with the result
|
||||
self._update_query_log(query_log["query_id"], generated_copy, True)
|
||||
|
||||
# Store the generated copy for future reference
|
||||
new_campaign = {
|
||||
"content": generated_copy,
|
||||
"content_type": request.content_type,
|
||||
"metadata": {
|
||||
"prompt": request.prompt,
|
||||
"tone": request.tone,
|
||||
"generated_at": datetime.now().isoformat(),
|
||||
"query_id": query_log["query_id"]
|
||||
}
|
||||
}
|
||||
|
||||
# Add to vector store for future similarity searches
|
||||
self.vector_store.add_campaign(new_campaign)
|
||||
|
||||
return {
|
||||
"result": generated_copy,
|
||||
"query_id": query_log["query_id"],
|
||||
"similar_campaigns_used": len(similar)
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
# Log the error in user queries
|
||||
if 'query_log' in locals():
|
||||
self._update_query_log(query_log["query_id"], str(e), False)
|
||||
raise e
|
||||
|
||||
def _log_user_query(self, request) -> Dict:
|
||||
"""Log user query for AI training purposes"""
|
||||
query_id = f"query_{datetime.now().strftime('%Y%m%d_%H%M%S_%f')}"
|
||||
|
||||
response = openai.ChatCompletion.create(
|
||||
model="gpt-3.5-turbo",
|
||||
messages=[
|
||||
{"role": "system", "content": self.brand_style.get_prompt(request)},
|
||||
{"role": "user", "content": f"Prompt: {request.prompt}\n\nSimilar examples:\n{similar}"}
|
||||
],
|
||||
temperature=0.7
|
||||
)
|
||||
# Handle both dict and Pydantic model objects
|
||||
if hasattr(request, 'dict'):
|
||||
request_data = request.dict()
|
||||
else:
|
||||
request_data = request
|
||||
|
||||
return response.choices[0].message.content
|
||||
query_log = {
|
||||
"query_id": query_id,
|
||||
"timestamp": datetime.now().isoformat(),
|
||||
"user_input": {
|
||||
"prompt": request_data.get("prompt"),
|
||||
"content_type": request_data.get("content_type"),
|
||||
"tone": request_data.get("tone")
|
||||
},
|
||||
"status": "processing",
|
||||
"generated_output": None,
|
||||
"success": None,
|
||||
"processing_time": None,
|
||||
"similar_campaigns_count": 0
|
||||
}
|
||||
|
||||
# Save to user_queries folder
|
||||
query_file = self.user_queries_path / f"{query_id}.json"
|
||||
with open(query_file, 'w') as f:
|
||||
json.dump(query_log, f, indent=2)
|
||||
|
||||
return query_log
|
||||
|
||||
def _update_query_log(self, query_id: str, output: str, success: bool):
|
||||
"""Update the query log with results"""
|
||||
query_file = self.user_queries_path / f"{query_id}.json"
|
||||
|
||||
if query_file.exists():
|
||||
with open(query_file, 'r') as f:
|
||||
query_log = json.load(f)
|
||||
|
||||
query_log.update({
|
||||
"generated_output": output,
|
||||
"success": success,
|
||||
"status": "completed" if success else "failed",
|
||||
"completed_at": datetime.now().isoformat()
|
||||
})
|
||||
|
||||
with open(query_file, 'w') as f:
|
||||
json.dump(query_log, f, indent=2)
|
||||
|
||||
def get_query_history(self, limit: int = 10) -> List[Dict]:
|
||||
"""Get recent user queries for analysis"""
|
||||
query_files = list(self.user_queries_path.glob("*.json"))
|
||||
query_files.sort(key=lambda x: x.stat().st_mtime, reverse=True)
|
||||
|
||||
queries = []
|
||||
for file in query_files[:limit]:
|
||||
with open(file, 'r') as f:
|
||||
queries.append(json.load(f))
|
||||
|
||||
return queries
|
||||
|
||||
def get_query_analytics(self) -> Dict:
|
||||
"""Get analytics on user queries for AI improvement"""
|
||||
query_files = list(self.user_queries_path.glob("*.json"))
|
||||
|
||||
total_queries = len(query_files)
|
||||
successful_queries = 0
|
||||
failed_queries = 0
|
||||
content_type_counts = {}
|
||||
tone_counts = {}
|
||||
|
||||
for file in query_files:
|
||||
with open(file, 'r') as f:
|
||||
query = json.load(f)
|
||||
|
||||
if query.get("success"):
|
||||
successful_queries += 1
|
||||
elif query.get("success") is False:
|
||||
failed_queries += 1
|
||||
|
||||
content_type = query.get("user_input", {}).get("content_type", "unknown")
|
||||
content_type_counts[content_type] = content_type_counts.get(content_type, 0) + 1
|
||||
|
||||
tone = query.get("user_input", {}).get("tone", "default")
|
||||
tone_counts[tone] = tone_counts.get(tone, 0) + 1
|
||||
|
||||
return {
|
||||
"total_queries": total_queries,
|
||||
"successful_queries": successful_queries,
|
||||
"failed_queries": failed_queries,
|
||||
"success_rate": round(successful_queries / total_queries * 100, 2) if total_queries > 0 else 0,
|
||||
"content_type_distribution": content_type_counts,
|
||||
"tone_distribution": tone_counts
|
||||
}
|
||||
|
||||
def log_user_feedback(self, query_id: str, feedback: Dict):
|
||||
"""Log user feedback on generated copy for training"""
|
||||
query_file = self.user_queries_path / f"{query_id}.json"
|
||||
|
||||
if query_file.exists():
|
||||
with open(query_file, 'r') as f:
|
||||
query_log = json.load(f)
|
||||
|
||||
query_log["user_feedback"] = {
|
||||
"rating": feedback.get("rating"), # 1-5 scale
|
||||
"comments": feedback.get("comments"),
|
||||
"used_output": feedback.get("used_output", False),
|
||||
"modifications_made": feedback.get("modifications_made"),
|
||||
"feedback_timestamp": datetime.now().isoformat()
|
||||
}
|
||||
|
||||
with open(query_file, 'w') as f:
|
||||
json.dump(query_log, f, indent=2)
|
||||
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
def export_training_data(self, output_file: str = None) -> str:
|
||||
"""Export user queries and feedback for model training"""
|
||||
if not output_file:
|
||||
output_file = f"training_data_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
|
||||
|
||||
query_files = list(self.user_queries_path.glob("*.json"))
|
||||
training_data = []
|
||||
|
||||
for file in query_files:
|
||||
with open(file, 'r') as f:
|
||||
query = json.load(f)
|
||||
|
||||
# Only include successful queries with feedback for training
|
||||
if query.get("success") and query.get("user_feedback"):
|
||||
training_example = {
|
||||
"input": query["user_input"]["prompt"],
|
||||
"content_type": query["user_input"]["content_type"],
|
||||
"tone": query["user_input"]["tone"],
|
||||
"output": query["generated_output"],
|
||||
"rating": query["user_feedback"]["rating"],
|
||||
"used": query["user_feedback"]["used_output"]
|
||||
}
|
||||
training_data.append(training_example)
|
||||
|
||||
output_path = Path("data") / output_file
|
||||
with open(output_path, 'w') as f:
|
||||
json.dump(training_data, f, indent=2)
|
||||
|
||||
return str(output_path)
|
||||
Reference in New Issue
Block a user