diff --git a/src/agents/critic.py b/src/agents/critic.py index ee1b1a0..6a7c18a 100644 --- a/src/agents/critic.py +++ b/src/agents/critic.py @@ -59,11 +59,25 @@ class CriticAgent(BaseAgent): def _get_system_prompt(self, profile_analysis: Dict[str, Any], example_posts: Optional[List[str]] = None, iteration: int = 1, max_iterations: int = 3) -> str: """Get system prompt for critic - orientiert an bewährten n8n-Prompts.""" - writing_style = profile_analysis.get("writing_style", {}) - linguistic = profile_analysis.get("linguistic_fingerprint", {}) - tone_analysis = profile_analysis.get("tone_analysis", {}) - phrase_library = profile_analysis.get("phrase_library", {}) + def _ensure_dict(value: Any) -> Dict[str, Any]: + return value if isinstance(value, dict) else {} + + def _extract_from_list(items: Any, key: str) -> Optional[str]: + if not isinstance(items, list): + return None + key_lower = key.lower() + for item in items: + if isinstance(item, str) and item.lower().startswith(f"{key_lower}:"): + return item.split(":", 1)[1].strip() + return None + + writing_style = _ensure_dict(profile_analysis.get("writing_style", {})) + linguistic = _ensure_dict(profile_analysis.get("linguistic_fingerprint", {})) + tone_analysis = _ensure_dict(profile_analysis.get("tone_analysis", {})) + phrase_library = _ensure_dict(profile_analysis.get("phrase_library", {})) structure_templates = profile_analysis.get("structure_templates", {}) + structure_templates_dict = _ensure_dict(structure_templates) + audience_insights = _ensure_dict(profile_analysis.get("audience_insights", {})) # Build example posts section for style comparison examples_section = "" @@ -84,7 +98,10 @@ class CriticAgent(BaseAgent): cta_phrases = phrase_library.get('cta_phrases', []) # Extract structure info - primary_structure = structure_templates.get('primary_structure', 'Hook → Body → CTA') + if isinstance(structure_templates, list): + primary_structure = _extract_from_list(structure_templates, "primary_structure") or "Hook → Body → CTA" + else: + primary_structure = structure_templates_dict.get('primary_structure', 'Hook → Body → CTA') # Iteration-aware guidance iteration_guidance = "" @@ -118,7 +135,7 @@ ITERATION {iteration}/{max_iterations} - Fortschritt anerkennen: REFERENZ-PROFIL (Der Maßstab): -Branche: {profile_analysis.get('audience_insights', {}).get('industry_context', 'Business')} +Branche: {audience_insights.get('industry_context', 'Business')} Perspektive: {writing_style.get('perspective', 'Ich-Perspektive')} Ansprache: {writing_style.get('form_of_address', 'Du/Euch')} Energie-Level: {linguistic.get('energy_level', 7)}/10 (1=sachlich, 10=explosiv) diff --git a/src/agents/writer.py b/src/agents/writer.py index 0444c89..595b98e 100644 --- a/src/agents/writer.py +++ b/src/agents/writer.py @@ -384,6 +384,10 @@ class WriterAgent(BaseAgent): if topic.get('why_this_person'): why_section = f"\n**WARUM DU DARÜBER SCHREIBEN SOLLTEST:**\n{topic.get('why_this_person')}\n" + plan_section = "" + if topic.get("post_plan"): + plan_section = f"\n**POST-PLAN (VERBINDLICH):**\n{json.dumps(topic.get('post_plan'), ensure_ascii=False, indent=2)}\n" + # User thoughts section thoughts_section = "" if user_thoughts: @@ -420,7 +424,7 @@ Schreibe einen authentischen LinkedIn-Post, der: {angle_section}{hook_section} **KERN-FAKT / INHALT:** {topic.get('fact', topic.get('description', ''))} -{summary_section}{extended_summary_section}{outline_section}{key_points_section}{facts_section}{quotes_section}{source_section}{thoughts_section} +{summary_section}{extended_summary_section}{outline_section}{key_points_section}{facts_section}{quotes_section}{source_section}{plan_section}{thoughts_section} **WARUM RELEVANT:** {topic.get('relevance', 'Aktuelles Thema für die Zielgruppe')} {why_section} @@ -587,6 +591,37 @@ Analysiere jeden Entwurf kurz und wähle den besten. Antworte im JSON-Format: phrase_library = profile_analysis.get("phrase_library", {}) structure_templates = profile_analysis.get("structure_templates", {}) + def _ensure_dict(value): + return value if isinstance(value, dict) else {} + + def _extract_from_list(items, key): + if not isinstance(items, list): + return None + prefix = f"{key}:" + for item in items: + text = str(item) + if text.lower().startswith(prefix): + return text.split(":", 1)[1].strip() + return None + + def _extract_list_from_list(items, key): + value = _extract_from_list(items, key) + if not value: + return [] + return [part.strip() for part in value.split(",") if part.strip()] + + def _extract_emojis(text: str) -> list: + if not text: + return [] + pattern = r"[\U0001F300-\U0001FAFF]" + matches = re.findall(pattern, text) + return matches or [] + + # Normalize optional sections + visual_dict = _ensure_dict(visual) + content_strategy_dict = _ensure_dict(content_strategy) + structure_templates_dict = _ensure_dict(structure_templates) + # Build example posts section (OPTIMIERT: mehr Kontext, weniger kürzen) examples_section = "" if example_posts and len(example_posts) > 0: @@ -598,7 +633,13 @@ Analysiere jeden Entwurf kurz und wähle den besten. Antworte im JSON-Format: examples_section += "--- Ende Beispiele ---\n" # Safe extraction of nested values - emoji_list = visual.get('emoji_usage', {}).get('emojis', ['🚀']) + if isinstance(visual, list): + emojis = [] + for item in visual: + emojis.extend(_extract_emojis(str(item))) + emoji_list = emojis if emojis else ['🚀'] + else: + emoji_list = visual_dict.get('emoji_usage', {}).get('emojis', ['🚀']) emoji_str = ' '.join(emoji_list) if isinstance(emoji_list, list) else str(emoji_list) sig_phrases = linguistic.get('signature_phrases', []) narrative_anchors = linguistic.get('narrative_anchors', []) @@ -622,9 +663,22 @@ Analysiere jeden Entwurf kurz und wähle den besten. Antworte im JSON-Format: return '\n - '.join(selected) # Extract structure templates - primary_structure = structure_templates.get('primary_structure', 'Hook → Body → CTA') - sentence_starters = structure_templates.get('typical_sentence_starters', []) - paragraph_transitions = structure_templates.get('paragraph_transitions', []) + if isinstance(structure_templates, list): + primary_structure = _extract_from_list(structure_templates, "primary_structure") or "Hook → Body → CTA" + sentence_starters = _extract_list_from_list(structure_templates, "typical_sentence_starters") + paragraph_transitions = _extract_list_from_list(structure_templates, "paragraph_transitions") + else: + primary_structure = structure_templates_dict.get('primary_structure', 'Hook → Body → CTA') + sentence_starters = structure_templates_dict.get('typical_sentence_starters', []) + paragraph_transitions = structure_templates_dict.get('paragraph_transitions', []) + + if isinstance(content_strategy, list): + content_strategy_dict = { + "cta_style": _extract_from_list(content_strategy, "cta_style") or "Interaktive Frage an die Community", + "hook_patterns": _extract_from_list(content_strategy, "hook_patterns"), + "post_structure": _extract_from_list(content_strategy, "post_structure"), + "storytelling_approach": _extract_from_list(content_strategy, "storytelling_approach"), + } # Extract N-gram patterns if available (NEU!) ngram_patterns = profile_analysis.get("ngram_patterns", {}) @@ -799,20 +853,20 @@ Zielgruppe: {audience.get('target_audience', 'Professionals')} 4. VISUELLE REGELN: -Unicode-Fettung: Nutze für den ersten Satz (Hook) fette Unicode-Zeichen (z.B. 𝗪𝗶𝗰𝗵𝘁𝗶𝗴𝗲𝗿 𝗦𝗮𝘁𝘇), sofern das zur Person passt: {visual.get('unicode_formatting', 'Fett für Hooks')} +Unicode-Fettung: Nutze für den ersten Satz (Hook) fette Unicode-Zeichen (z.B. 𝗪𝗶𝗰𝗵𝘁𝗶𝗴𝗲𝗿 𝗦𝗮𝘁𝘇), sofern das zur Person passt: {visual_dict.get('unicode_formatting', _extract_from_list(visual, 'unicode_formatting') if isinstance(visual, list) else 'Fett für Hooks')} Emoji-Logik: Verwende diese Emojis: {emoji_str} -Platzierung: {visual.get('emoji_usage', {}).get('placement', 'Ende')} -Häufigkeit: {visual.get('emoji_usage', {}).get('frequency', 'Mittel')} +Platzierung: {visual_dict.get('emoji_usage', {}).get('placement', _extract_from_list(visual, 'emoji_usage') if isinstance(visual, list) else 'Ende')} +Häufigkeit: {visual_dict.get('emoji_usage', {}).get('frequency', 'Mittel')} Erzähl-Anker: Baue Elemente ein wie: {narrative_str} (Falls 'PS-Zeilen', 'Dialoge' oder 'Flashbacks' genannt sind, integriere diese wenn es passt.) -Layout: {visual.get('structure_preferences', 'Kurze Absätze, mobil-optimiert')} +Layout: {visual_dict.get('structure_preferences', _extract_from_list(visual, 'structure_preferences') if isinstance(visual, list) else 'Kurze Absätze, mobil-optimiert')} Länge: Ca. {writing_style.get('average_word_count', 300)} Wörter -CTA: Beende den Post mit einer Variante von: {content_strategy.get('cta_style', 'Interaktive Frage an die Community')} +CTA: Beende den Post mit einer Variante von: {content_strategy_dict.get('cta_style', 'Interaktive Frage an die Community')} 5. GUARDRAILS (VERBOTE!): @@ -936,6 +990,10 @@ Strategy Weight: {strategy_weight:.1f} / 1.0 # Revision mode with structured feedback score_text = f"**AKTUELLER SCORE:** {critic_result.get('overall_score', 'N/A')}/100\n\n" if critic_result else "" + plan_section = "" + if topic.get("post_plan"): + plan_section = f"\n**POST-PLAN (VERBINDLICH):**\n{json.dumps(topic.get('post_plan'), ensure_ascii=False, indent=2)}\n" + return f"""ÜBERARBEITE den Post basierend auf dem Feedback. **VORHERIGE VERSION:** @@ -945,6 +1003,7 @@ Strategy Weight: {strategy_weight:.1f} / 1.0 {feedback} {specific_changes_text} {improvements_text} +{plan_section} **DEINE AUFGABE:** 1. Führe die Änderungen durch wie im Feedback beschrieben 2. Behalte alles bei was gut funktioniert @@ -999,6 +1058,10 @@ Gib NUR den überarbeiteten Post zurück - keine Kommentare.""" source_url = topic.get('source') or "" source_section = f"\n**QUELLE:** {source_title} {source_url}\n" + plan_section = "" + if topic.get("post_plan"): + plan_section = f"\n**POST-PLAN (VERBINDLICH):**\n{json.dumps(topic.get('post_plan'), ensure_ascii=False, indent=2)}\n" + # User thoughts section thoughts_section = "" if user_thoughts: @@ -1035,7 +1098,7 @@ Schreibe einen authentischen LinkedIn-Post, der: {angle_section}{hook_section} **KERN-FAKT / INHALT:** {topic.get('fact', topic.get('description', ''))} -{summary_section}{extended_summary_section}{outline_section}{key_points_section}{facts_section}{quotes_section}{source_section}{thoughts_section} +{summary_section}{extended_summary_section}{outline_section}{key_points_section}{facts_section}{quotes_section}{source_section}{plan_section}{thoughts_section} **WARUM RELEVANT:** {topic.get('relevance', 'Aktuelles Thema für die Zielgruppe')} diff --git a/src/orchestrator.py b/src/orchestrator.py index f44d622..b2043c3 100644 --- a/src/orchestrator.py +++ b/src/orchestrator.py @@ -1,4 +1,5 @@ """Main orchestrator for the LinkedIn workflow.""" +import json from collections import Counter from typing import Dict, Any, List, Optional, Callable from uuid import UUID @@ -638,6 +639,19 @@ class WorkflowOrchestrator: company_strategy = company.company_strategy logger.info(f"Loaded company strategy for post creation: {company.name}") + # Build a structured post plan (used to guide writing) + post_plan = await self._build_post_plan( + topic=topic, + profile_analysis=profile_analysis.full_analysis, + company_strategy=company_strategy, + strategy_weight=strategy_weight, + selected_hook=selected_hook, + user_thoughts=user_thoughts, + post_type_analysis=post_type_analysis + ) + if post_plan: + topic = {**topic, "post_plan": post_plan} + # Writer-Critic loop while iteration < max_iterations and not approved: iteration += 1 @@ -697,13 +711,45 @@ class WorkflowOrchestrator: max_iterations=max_iterations ) + # Style/Strategy scoring (enforce thresholds based on strategy_weight) + style_strategy = await self._score_style_strategy( + post=current_post, + profile_analysis=profile_analysis.full_analysis, + company_strategy=company_strategy, + strategy_weight=strategy_weight + ) + style_score = style_strategy.get("style_score", 0) + strategy_score = style_strategy.get("strategy_score", 0) + style_threshold = int(60 + (1 - strategy_weight) * 30) + strategy_threshold = int(60 + strategy_weight * 30) + if not company_strategy: + strategy_score = 100 + strategy_threshold = 0 + critic_result["style_score"] = style_score + critic_result["strategy_score"] = strategy_score + critic_result["style_threshold"] = style_threshold + critic_result["strategy_threshold"] = strategy_threshold + + needs_style_fix = style_score < style_threshold + needs_strategy_fix = strategy_score < strategy_threshold + if needs_style_fix or needs_strategy_fix: + critic_result["approved"] = False + issues = [] + if needs_style_fix: + issues.append(f"Stil-Match zu niedrig ({style_score}/100, Ziel: {style_threshold}).") + if needs_strategy_fix: + issues.append(f"Strategie-Match zu niedrig ({strategy_score}/100, Ziel: {strategy_threshold}).") + feedback_note = " ".join(issues) + extra = style_strategy.get("feedback", "") + critic_result["feedback"] = f"{critic_result.get('feedback','')}\n\n{feedback_note}\n{extra}".strip() + critic_feedback_list.append(critic_result) approved = critic_result.get("approved", False) score = critic_result.get("overall_score", 0) # Auto-approve on last iteration if score is decent (>= 80) - if iteration == max_iterations and not approved and score >= 80: + if iteration == max_iterations and not approved and score >= 80 and not (needs_style_fix or needs_strategy_fix): approved = True critic_result["approved"] = True logger.info(f"Auto-approved on final iteration with score {score}") @@ -838,6 +884,138 @@ class WorkflowOrchestrator: "readability_check": readability_result } + async def _build_post_plan( + self, + topic: Dict[str, Any], + profile_analysis: Dict[str, Any], + company_strategy: Optional[Dict[str, Any]], + strategy_weight: float, + selected_hook: str = "", + user_thoughts: str = "", + post_type_analysis: Optional[Dict[str, Any]] = None + ) -> Optional[Dict[str, Any]]: + """Create a structured plan to guide the writer.""" + try: + system_prompt = ( + "Du erstellst einen kompakten Schreibplan für einen LinkedIn-Post. " + "Der Plan muss Stil der Person UND Unternehmensstrategie berücksichtigen. " + "Antwort nur als JSON." + ) + + user_prompt = f""" +Thema: {topic.get('title')} +Kern-Fakt: {topic.get('fact') or topic.get('summary','')} +Relevanz: {topic.get('relevance')} + +Ausgewählter Hook: {selected_hook or "keiner"} +Persönliche Gedanken: {user_thoughts or "keine"} + +Profil-Analyse (Kurz): +Tone: {profile_analysis.get('writing_style', {}).get('tone')} +Perspektive: {profile_analysis.get('writing_style', {}).get('perspective')} +Ansprache: {profile_analysis.get('writing_style', {}).get('form_of_address')} +Signature Phrases: {profile_analysis.get('linguistic_fingerprint', {}).get('signature_phrases', [])} + +Unternehmensstrategie (Kurz): +Mission: {company_strategy.get('mission') if company_strategy else ''} +Brand Voice: {company_strategy.get('brand_voice') if company_strategy else ''} +Content Pillars: {company_strategy.get('content_pillars') if company_strategy else []} +Do: {company_strategy.get('dos') if company_strategy else []} +Don't: {company_strategy.get('donts') if company_strategy else []} + +Strategy Weight: {strategy_weight:.1f} + +Gib JSON im Format: +{{ + "hook_type": "Story/Fakten/These/Frage/...", + "structure": ["Hook", "Body", "CTA"], + "style_requirements": ["2-4 konkrete Stil-Elemente/Signaturphrasen"], + "strategy_requirements": ["2-4 Strategie-Elemente/Content-Pillar/Do"], + "cta_type": "Frage/Aufforderung/Diskussion/..." +}} +""" + raw = await self.writer.call_openai( + system_prompt=system_prompt, + user_prompt=user_prompt, + model="gpt-4o", + temperature=0.2, + response_format={"type": "json_object"} + ) + return json.loads(raw) + except Exception as e: + logger.warning(f"Post plan generation failed: {e}") + return None + + async def _score_style_strategy( + self, + post: str, + profile_analysis: Dict[str, Any], + company_strategy: Optional[Dict[str, Any]], + strategy_weight: float + ) -> Dict[str, Any]: + """Score style and strategy match separately.""" + try: + system_prompt = ( + "Bewerte einen LinkedIn-Post in zwei Dimensionen: Stil-Match zur Person " + "und Strategie-Match zur Unternehmensstrategie. Antwort nur als JSON." + ) + user_prompt = f""" +POST: +\"\"\"{post}\"\"\" + +Profil-Analyse (Kurz): +Tone: {profile_analysis.get('writing_style', {}).get('tone')} +Perspektive: {profile_analysis.get('writing_style', {}).get('perspective')} +Ansprache: {profile_analysis.get('writing_style', {}).get('form_of_address')} +Signature Phrases: {profile_analysis.get('linguistic_fingerprint', {}).get('signature_phrases', [])} + +Unternehmensstrategie (Kurz): +Mission: {company_strategy.get('mission') if company_strategy else ''} +Brand Voice: {company_strategy.get('brand_voice') if company_strategy else ''} +Content Pillars: {company_strategy.get('content_pillars') if company_strategy else []} +Do: {company_strategy.get('dos') if company_strategy else []} +Don't: {company_strategy.get('donts') if company_strategy else []} + +Gib JSON im Format: +{{ + "style_score": 0-100, + "strategy_score": 0-100, + "feedback": "Kurzes, konkretes Verbesserungshinweis-Text (2-4 Sätze)" +}} +""" + raw = await self.critic.call_openai( + system_prompt=system_prompt, + user_prompt=user_prompt, + model="gpt-4o", + temperature=0.2, + response_format={"type": "json_object"} + ) + return json.loads(raw) + except Exception as e: + logger.warning(f"Style/Strategy scoring failed: {e}") + return {"style_score": 0, "strategy_score": 0, "feedback": ""} + + async def generate_post_plan( + self, + topic: Dict[str, Any], + profile_analysis: Dict[str, Any], + company_strategy: Optional[Dict[str, Any]], + strategy_weight: float, + selected_hook: str = "", + user_thoughts: str = "", + post_type_analysis: Optional[Dict[str, Any]] = None + ) -> Optional[Dict[str, Any]]: + """Public wrapper to build a post plan.""" + return await self._build_post_plan( + topic=topic, + profile_analysis=profile_analysis, + company_strategy=company_strategy, + strategy_weight=strategy_weight, + selected_hook=selected_hook, + user_thoughts=user_thoughts, + post_type_analysis=post_type_analysis + ) + async def _extract_recurring_feedback(self, user_id: UUID) -> Dict[str, Any]: """ Extract recurring feedback patterns from past generated posts. diff --git a/src/utils/profile_analysis_normalizer.py b/src/utils/profile_analysis_normalizer.py new file mode 100644 index 0000000..5fd05b2 --- /dev/null +++ b/src/utils/profile_analysis_normalizer.py @@ -0,0 +1,71 @@ +"""Normalize profile analysis structures for consistent access.""" +from typing import Any, Dict, List + + +def _extract_from_list(items: Any, key: str) -> str: + if not isinstance(items, list): + return "" + prefix = f"{key}:" + for item in items: + text = str(item) + if text.lower().startswith(prefix): + return text.split(":", 1)[1].strip() + return "" + + +def _extract_list_from_list(items: Any, key: str) -> List[str]: + value = _extract_from_list(items, key) + if not value: + return [] + return [part.strip() for part in value.split(",") if part.strip()] + + +def _ensure_dict(value: Any) -> Dict[str, Any]: + return value if isinstance(value, dict) else {} + + +def normalize_profile_analysis(profile_analysis: Dict[str, Any]) -> Dict[str, Any]: + """Ensure nested fields are dicts with expected keys to avoid crashes.""" + if not isinstance(profile_analysis, dict): + return {} + + normalized = dict(profile_analysis) + + visual = profile_analysis.get("visual_patterns", {}) + content_strategy = profile_analysis.get("content_strategy", {}) + structure_templates = profile_analysis.get("structure_templates", {}) + + if isinstance(visual, list): + normalized["visual_patterns"] = { + "unicode_formatting": _extract_from_list(visual, "unicode_formatting"), + "structure_preferences": _extract_from_list(visual, "structure_preferences"), + "emoji_usage": { + "placement": _extract_from_list(visual, "emoji_usage"), + "frequency": _extract_from_list(visual, "emoji_usage"), + "emojis": [] + } + } + else: + normalized["visual_patterns"] = _ensure_dict(visual) + + if isinstance(content_strategy, list): + normalized["content_strategy"] = { + "cta_style": _extract_from_list(content_strategy, "cta_style"), + "hook_patterns": _extract_from_list(content_strategy, "hook_patterns"), + "post_structure": _extract_from_list(content_strategy, "post_structure"), + "storytelling_approach": _extract_from_list(content_strategy, "storytelling_approach"), + } + else: + normalized["content_strategy"] = _ensure_dict(content_strategy) + + if isinstance(structure_templates, list): + normalized["structure_templates"] = { + "primary_structure": _extract_from_list(structure_templates, "primary_structure"), + "template_examples": _extract_list_from_list(structure_templates, "template_examples"), + "paragraph_transitions": _extract_list_from_list(structure_templates, "paragraph_transitions"), + "typical_sentence_starters": _extract_list_from_list(structure_templates, "typical_sentence_starters"), + } + else: + normalized["structure_templates"] = _ensure_dict(structure_templates) + + return normalized diff --git a/src/web/user/routes.py b/src/web/user/routes.py index 7c14f03..09c4a7a 100644 --- a/src/web/user/routes.py +++ b/src/web/user/routes.py @@ -4373,6 +4373,16 @@ async def chat_generate_post(request: Request): "relevance": "User-specified content" } + plan = await orchestrator.generate_post_plan( + topic=topic, + profile_analysis=profile_analysis.full_analysis, + company_strategy=company_strategy, + strategy_weight=getattr(post_type, 'strategy_weight', 0.5), + user_thoughts=message + ) + if plan: + topic["post_plan"] = plan + # Generate post post_content = await writer.process( topic=topic, @@ -4485,6 +4495,16 @@ async def chat_refine_post(request: Request): "relevance": "User refinement request" } + plan = await orchestrator.generate_post_plan( + topic=topic, + profile_analysis=full_analysis, + company_strategy=company_strategy, + strategy_weight=getattr(post_type, 'strategy_weight', 0.5), + user_thoughts=message + ) + if plan: + topic["post_plan"] = plan + # Use writer's revision capability refined_post = await writer.process( topic=topic, @@ -4794,6 +4814,16 @@ async def company_chat_generate_post(request: Request): ) topic = {"title": message[:100], "fact": message, "relevance": "Company-created content"} + + plan = await orchestrator.generate_post_plan( + topic=topic, + profile_analysis=profile_analysis.full_analysis, + company_strategy=company_strategy, + strategy_weight=getattr(post_type, 'strategy_weight', 0.5), + user_thoughts=message + ) + if plan: + topic["post_plan"] = plan post_content = await writer.process( topic=topic, profile_analysis=profile_analysis.full_analysis,