consultoria-as/social-media-automation
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Add Content Generation Engine v2 with quality scoring

Browse Source 
Major improvements to AI content generation:

## New Components (app/services/ai/)
- PromptLibrary: YAML-based prompt templates with inheritance
- ContextEngine: Anti-repetition and best performers tracking
- ContentGeneratorV2: Enhanced generation with dynamic parameters
- PlatformAdapter: Platform-specific content adaptation
- ContentValidator: AI-powered quality scoring (0-100)

## Prompt Library (app/prompts/)
- 3 personalities: default, educational, promotional
- 5 templates: tip_tech, product_post, service_post, thread, response
- 4 platform configs: x, threads, instagram, facebook
- Few-shot examples by category: ia, productividad, seguridad

## Database Changes
- New table: content_memory (tracks generated content)
- New columns in posts: quality_score, score_breakdown, generation_attempts

## New API Endpoints (/api/v2/generate/)
- POST /generate - Generation with quality check
- POST /generate/batch - Batch generation
- POST /quality/evaluate - Evaluate content quality
- GET /templates, /personalities, /platforms - List configs

## Celery Tasks
- update_engagement_scores (every 6h)
- cleanup_old_memory (monthly)
- refresh_best_posts_yaml (weekly)

## Tests
- Comprehensive tests for all AI engine components

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Consultoría AS
2026-01-28 20:55:28 +00:00
parent f458f809ca
commit 11b0ba46fa
36 changed files with 6266 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

416
app/services/content_generator.py
View File

@@ -1,20 +1,65 @@
"""
Servicio de generación de contenido con DeepSeek API.
Este archivo mantiene la interfaz original (ContentGenerator) para
compatibilidad con código existente, pero internamente usa el nuevo
motor modular (ContentGeneratorV2) cuando está disponible.
Para nuevas integraciones, usar directamente:
from app.services.ai import ContentGeneratorV2, content_generator_v2
"""
import json
from typing import Optional, List, Dict
from typing import Optional, List, Dict, Any
from openai import OpenAI
from sqlalchemy.orm import Session
from app.core.config import settings
# Importar nuevo motor
try:
from app.services.ai import (
ContentGeneratorV2,
content_generator_v2,
ContextEngine,
context_engine,
ContentValidator,
content_validator,
)
NEW_ENGINE_AVAILABLE = True
except ImportError:
NEW_ENGINE_AVAILABLE = False
class ContentGenerator:
"""Generador de contenido usando DeepSeek API."""
"""
Generador de contenido usando DeepSeek API.
def __init__(self):
Esta clase mantiene la interfaz original para compatibilidad.
Internamente delega al nuevo motor cuando está disponible.
Para nuevas funcionalidades, usar ContentGeneratorV2 directamente.
"""
def __init__(self, use_new_engine: bool = True):
"""
Inicializar el generador.
Args:
use_new_engine: Si usar el nuevo motor v2 (default: True)
"""
self._client = None
self.model = "deepseek-chat"
self._use_new_engine = use_new_engine and NEW_ENGINE_AVAILABLE
if self._use_new_engine:
self._v2 = content_generator_v2
self._validator = content_validator
self._context = context_engine
else:
self._v2 = None
self._validator = None
self._context = None
@property
def client(self):
@@ -30,6 +75,15 @@ class ContentGenerator:
def _get_system_prompt(self) -> str:
"""Obtener el prompt del sistema con la personalidad de la marca."""
# Si hay nuevo motor, usar su prompt
if self._use_new_engine:
try:
from app.services.ai import prompt_library
return prompt_library.get_system_prompt()
except Exception:
pass
# Fallback al prompt original
return f"""Eres el Community Manager de {settings.BUSINESS_NAME}, una empresa de tecnología ubicada en {settings.BUSINESS_LOCATION}.
SOBRE LA EMPRESA:
@@ -54,13 +108,70 @@ REGLAS:
- Enfócate en ayudar, no en vender directamente
- Adapta el contenido a cada plataforma"""
# === Métodos principales con nuevo motor ===
async def generate_tip_tech(
self,
category: str,
platform: str,
template: Optional[str] = None,
db: Optional[Session] = None,
validate: bool = True
) -> str:
"""
Generar un tip tech.
Args:
category: Categoría del tip
platform: Plataforma destino
template: Template opcional (ignorado en v2)
db: Sesión de DB para context engine
validate: Si validar el contenido generado
Returns:
Contenido del tip
"""
if self._use_new_engine:
result = await self._v2.generate_tip(
category=category,
platform=platform,
db=db
)
content = result["adapted_content"]
# Validar y regenerar si es necesario
if validate and self._validator:
quality = await self._validator.evaluate(content, platform)
if quality.final_decision == "regenerate":
# Regenerar con hints
hints = self._validator.get_regeneration_hints(quality)
result = await self._v2.generate(
template_name="tip_tech",
variables={
"category": category,
"difficulty_level": "principiante",
"target_audience": "profesionales tech"
},
platform=platform,
db=db,
temperature_override=0.9 # Más creatividad en retry
)
content = result["adapted_content"]
return content
# Fallback a implementación original
return await self._generate_tip_tech_legacy(category, platform, template)
async def _generate_tip_tech_legacy(
self,
category: str,
platform: str,
template: Optional[str] = None
) -> str:
"""Generar un tip tech."""
"""Implementación original de generate_tip_tech."""
char_limits = {
"x": 280,
"threads": 500,
@@ -97,11 +208,55 @@ Responde SOLO con el texto del post, sin explicaciones."""
return response.choices[0].message.content.strip()
async def generate_product_post(
self,
product: Dict,
platform: str,
db: Optional[Session] = None,
validate: bool = True
) -> str:
"""
Generar post para un producto.
Args:
product: Dict con datos del producto
platform: Plataforma destino
db: Sesión de DB
validate: Si validar contenido
Returns:
Contenido del post
"""
if self._use_new_engine:
result = await self._v2.generate_product_post(
product=product,
platform=platform,
db=db
)
content = result["adapted_content"]
if validate and self._validator:
quality = await self._validator.evaluate(content, platform)
if quality.final_decision == "regenerate":
result = await self._v2.generate_product_post(
product=product,
platform=platform,
db=db,
temperature_override=0.9
)
content = result["adapted_content"]
return content
# Fallback a implementación original
return await self._generate_product_post_legacy(product, platform)
async def _generate_product_post_legacy(
self,
product: Dict,
platform: str
) -> str:
"""Generar post para un producto."""
"""Implementación original de generate_product_post."""
char_limits = {
"x": 280,
"threads": 500,
@@ -144,11 +299,55 @@ Responde SOLO con el texto del post."""
return response.choices[0].message.content.strip()
async def generate_service_post(
self,
service: Dict,
platform: str,
db: Optional[Session] = None,
validate: bool = True
) -> str:
"""
Generar post para un servicio.
Args:
service: Dict con datos del servicio
platform: Plataforma destino
db: Sesión de DB
validate: Si validar contenido
Returns:
Contenido del post
"""
if self._use_new_engine:
result = await self._v2.generate_service_post(
service=service,
platform=platform,
db=db
)
content = result["adapted_content"]
if validate and self._validator:
quality = await self._validator.evaluate(content, platform)
if quality.final_decision == "regenerate":
result = await self._v2.generate_service_post(
service=service,
platform=platform,
db=db,
temperature_override=0.9
)
content = result["adapted_content"]
return content
# Fallback
return await self._generate_service_post_legacy(service, platform)
async def _generate_service_post_legacy(
self,
service: Dict,
platform: str
) -> str:
"""Generar post para un servicio."""
"""Implementación original de generate_service_post."""
char_limits = {
"x": 280,
"threads": 500,
@@ -191,11 +390,38 @@ Responde SOLO con el texto del post."""
return response.choices[0].message.content.strip()
async def generate_thread(
self,
topic: str,
num_posts: int = 5,
db: Optional[Session] = None
) -> List[str]:
"""
Generar un hilo educativo.
Args:
topic: Tema del hilo
num_posts: Número de posts
db: Sesión de DB
Returns:
Lista de posts del hilo
"""
if self._use_new_engine:
return await self._v2.generate_thread(
topic=topic,
num_posts=num_posts,
db=db
)
# Fallback
return await self._generate_thread_legacy(topic, num_posts)
async def _generate_thread_legacy(
self,
topic: str,
num_posts: int = 5
) -> List[str]:
"""Generar un hilo educativo."""
"""Implementación original de generate_thread."""
prompt = f"""Genera un hilo educativo de {num_posts} posts sobre: {topic}
REQUISITOS:
@@ -222,7 +448,6 @@ Responde con cada post en una línea separada, sin explicaciones."""
)
content = response.choices[0].message.content.strip()
# Separar posts por líneas no vacías
posts = [p.strip() for p in content.split('\n') if p.strip()]
return posts
@@ -233,7 +458,36 @@ Responde con cada post en una línea separada, sin explicaciones."""
interaction_type: str,
context: Optional[str] = None
) -> List[str]:
"""Generar sugerencias de respuesta para una interacción."""
"""
Generar sugerencias de respuesta para una interacción.
Args:
interaction_content: Contenido de la interacción
interaction_type: Tipo de interacción
context: Contexto adicional
Returns:
Lista de 3 opciones de respuesta
"""
if self._use_new_engine:
return await self._v2.generate_response(
interaction_content=interaction_content,
interaction_type=interaction_type,
context=context
)
# Fallback
return await self._generate_response_legacy(
interaction_content, interaction_type, context
)
async def _generate_response_legacy(
self,
interaction_content: str,
interaction_type: str,
context: Optional[str] = None
) -> List[str]:
"""Implementación original de generate_response_suggestion."""
prompt = f"""Un usuario escribió esto en redes sociales:
"{interaction_content}"
@@ -267,21 +521,46 @@ Responde con las 3 opciones numeradas, una por línea."""
content = response.choices[0].message.content.strip()
suggestions = [s.strip() for s in content.split('\n') if s.strip()]
# Limpiar numeración si existe
cleaned = []
for s in suggestions:
if s[0].isdigit() and (s[1] == '.' or s[1] == ')'):
s = s[2:].strip()
cleaned.append(s)
return cleaned[:3] # Máximo 3 sugerencias
return cleaned[:3]
async def adapt_content_for_platform(
self,
content: str,
target_platform: str
) -> str:
"""Adaptar contenido existente a una plataforma específica."""
"""
Adaptar contenido existente a una plataforma específica.
Args:
content: Contenido original
target_platform: Plataforma destino
Returns:
Contenido adaptado
"""
if self._use_new_engine:
# Detectar plataforma de origen (asumimos la más genérica)
return await self._v2.adapt_content(
content=content,
source_platform="instagram", # Asume origen genérico
target_platform=target_platform
)
# Fallback
return await self._adapt_content_legacy(content, target_platform)
async def _adapt_content_legacy(
self,
content: str,
target_platform: str
) -> str:
"""Implementación original de adapt_content_for_platform."""
char_limits = {
"x": 280,
"threads": 500,
@@ -318,6 +597,119 @@ Responde SOLO con el contenido adaptado."""
return response.choices[0].message.content.strip()
# === Nuevos métodos (solo v2) ===
async def generate_with_quality_check(
self,
template_name: str,
variables: Dict[str, Any],
platform: str,
db: Optional[Session] = None,
max_attempts: int = 2
) -> Dict[str, Any]:
"""
Generar contenido con validación y regeneración automática.
Solo disponible con el nuevo motor.
Args:
template_name: Nombre del template
variables: Variables para el template
platform: Plataforma destino
db: Sesión de DB
max_attempts: Máximo intentos de regeneración
Returns:
Dict con contenido, score, y metadata
"""
if not self._use_new_engine:
raise RuntimeError(
"Este método requiere el nuevo motor. "
"Asegúrate de que app.services.ai esté disponible."
)
attempt = 0
temperature = 0.7
while attempt < max_attempts:
attempt += 1
# Generar
result = await self._v2.generate(
template_name=template_name,
variables=variables,
platform=platform,
db=db,
temperature_override=temperature
)
content = result["adapted_content"]
# Evaluar calidad
quality = await self._validator.evaluate(content, platform)
# Si pasa, retornar
if quality.final_decision == "accept":
return {
"content": content,
"quality_score": quality.scoring.total_score if quality.scoring else None,
"score_breakdown": quality.scoring.breakdown if quality.scoring else None,
"is_top_performer": quality.scoring.is_top_performer if quality.scoring else False,
"attempts": attempt,
"metadata": result["metadata"]
}
# Si debe regenerar, aumentar temperature
temperature = min(1.0, temperature + 0.1)
# Si llegamos aquí, usar el último intento aunque no sea ideal
return {
"content": content,
"quality_score": quality.scoring.total_score if quality.scoring else None,
"score_breakdown": quality.scoring.breakdown if quality.scoring else None,
"is_top_performer": False,
"attempts": attempt,
"metadata": result["metadata"],
"warning": "Contenido aceptado después de máximos intentos"
}
async def save_to_memory(
self,
db: Session,
post_id: int,
content: str,
content_type: str,
platform: str,
quality_score: Optional[int] = None,
quality_breakdown: Optional[Dict] = None
):
"""
Guardar contenido en memoria para tracking.
Solo disponible con el nuevo motor.
Args:
db: Sesión de DB
post_id: ID del post
content: Contenido generado
content_type: Tipo de contenido
platform: Plataforma
quality_score: Score de calidad
quality_breakdown: Breakdown del score
"""
if not self._use_new_engine:
return # Silenciosamente ignorar si no hay nuevo motor
self._context.save_to_memory(
db=db,
post_id=post_id,
content=content,
content_type=content_type,
platform=platform,
quality_score=quality_score,
quality_breakdown=quality_breakdown
)
# Instancia global
content_generator = ContentGenerator()