"""
outreach_generator.py -- Generiert personalisierte Kaltakquise-Emails via Claude.
Fuer jeden Kontakt wird eine Email generiert die:
- Sich konkret auf die Branche des Empfaengers bezieht
- Ein reales Problem benennt
- Das passende Produkt (LP Lueftung ODER AG KI) anbietet
- Einen messbaren Personalisierungs-Score hat (>= 0.6 Pflicht)
Verwendung:
python outreach_generator.py --step 1 --campaign 1
python outreach_generator.py --step 1 --all
python outreach_generator.py --regenerate 42 # Einzelne Sequence-ID neu generieren
"""
import os
import json
import re
import logging
import argparse
from typing import Dict, List, Optional, Tuple
import anthropic
from cold_outreach_db import ColdOutreachDB
from knowledge_retriever import KnowledgeRetriever
logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
logger = logging.getLogger(__name__)
CLAUDE_MODEL = 'claude-sonnet-4-6'
MIN_PERSONALIZATION_SCORE = 0.6
MAX_RETRIES = 2
# ---------------------------------------------------------------------------
# Prompt-Texte
# ---------------------------------------------------------------------------
SYSTEM_PROMPT = """Du bist ein Email-Copywriter fuer personalisierte B2B-Kaltakquise.
GOLDENE REGELN:
- Schreibe auf Deutsch (oder Spanisch wenn der Kontakt es ist)
- Maximal 150 Woerter pro Email
- Kein Marketing-Sprech, kein "Wir sind die Besten"
- Schreibe wie ein Mensch, nicht wie eine Maschine
- Erster Satz MUSS sich auf die konkrete Situation des Empfaengers beziehen
- Keine generischen Floskeln: "Ich hoffe diese Email erreicht Sie wohlauf" = VERBOTEN
- Call-to-Action: Immer eine konkrete, niedrigschwellige Frage am Ende
- Kein Anhang, keine Links in Email 1
- Signatur: Nur Vorname + Firma + Telefon
- Firmenname des Empfaengers MUSS im Text vorkommen (nicht nur in Anrede)
OUTPUT: Ausschliesslich JSON, kein zusaetzlicher Text:
{
"subject": "...",
"body": "...",
"personalization_score": 0.0-1.0,
"personalization_elements": ["element1", "element2"]
}
personalization_score berechnen nach:
- 0.2 Punkte: Firmenname im Textkörper
- 0.2 Punkte: Branche/Beruf konkret benannt
- 0.2 Punkte: Spezifisches Problem der Branche
- 0.2 Punkte: Lokaler/konkreter Bezug (kein Generic)
- 0.2 Punkte: Echter Nutzen mit Zahl/Beispiel ODER Wissens-Snippet
personalization_elements: Liste der konkreten Bezugspunkte die du verwendet hast."""
LP_CONTEXT = """
ABSENDER: Karlo, Mallorca AirServices (Lueftungsprofi)
PRODUKT: Dezentrale Lueftungssysteme (getAir), CO2/Feuchte-Sensoren, Schimmelpraevention
REGION: Mallorca (Balearen) -- lokaler Spezialist
ZIELGRUPPEN: Hotels, Immobilienverwalter, Buerogebaeude, Bauunternehmen, Wohnungsbau
ALLEINSTELLUNGSMERKMAL: Einziger deutschsprachiger getAir-Spezialist auf Mallorca, persoenlicher Service
TON: Professionell aber persoenlich. Regional verankert. Nicht verkaeuferhaft. Wie ein Nachbar der Experte ist.
SIGNATUR: Karlo | Mallorca AirServices | +34 644 931 559
BRANCHENSPEZIFISCHE PROBLEME:
- Hotels: Gaestebeschwerden ueber stickige Zimmer, Schimmel im Bad, hohe Klimaanlagen-Kosten
- Immobilienverwalter: Schimmelreklamationen, Sanierungskosten 3.000-8.000 EUR/Einheit
- Bueros: CO2 ueber 1000ppm senkt Mitarbeiter-Produktivitaet um 15%
- Bauunternehmen: CTE/RITE-Vorschriften Spanien seit 2020 fuer mechanische Lueftung
"""
AG_CONTEXT = """
ABSENDER: Karlo, Agent Solutions
PRODUKTE (waehle das passende basierend auf icp_recommended_product):
- email_agent: KI-basierte Email-Automatisierung (Klassifikation, Auto-Reply, Routing). 87% guenstiger als Fonica.
- telefon_agent: KI-Telefonassistent (IVR, Spracherkennung, automatische Anrufbearbeitung). 100 Anrufe/Tag fuer 42-90 EUR/Monat.
- ki_kurse: Online-Kurse zu KI-Themen fuer Unternehmer
ZIELGRUPPEN: KMUs, Mittelstaendler, kleine Dienstleister
ALLEINSTELLUNGSMERKMAL: DIY-Loesungen, transparent, kein Vendor-Lock-In, 10x guenstiger als Enterprise-Software
TON: Technisch kompetent aber verstaendlich. Kein Buzzword-Bingo. Konkrete Zahlen (Kosten, Zeit).
SIGNATUR: Karlo | Agent Solutions | +34 644 931 559
BRANCHENSPEZIFISCHE PROBLEME:
- Handwerker/Dienstleister: Verpassen Anrufe waehrend Arbeit, verlieren Auftraege
- Arztpraxen: Telefon klingelt permanent, Rezeption ueberfordert
- Rechtsanwaelte/Steuerberater: Email-Flut, Erst-Anfragen kosten zu viel Zeit
- E-Commerce: Kunden-Support-Emails fressen Zeit
"""
STEP_INSTRUCTIONS = {
1: """
EMAIL 1 -- ERSTER KONTAKT (Aufmerksamkeit wecken):
- Satz 1: KONKRETER Bezug zur Firma/Branche (zeige dass du recherchiert hast)
- Satz 2-3: Problem benennen das der Empfaenger wahrscheinlich hat
- Satz 4: Andeuten dass es eine Loesung gibt (nicht verkaufen!)
- Satz 5: Einfache Frage ("Ist das ein Thema bei Ihnen?")
- Laenge: 80-120 Woerter
- Kein Link, kein Anhang, kein Druck
""",
2: """
EMAIL 2 -- FOLLOW-UP MIT MEHRWERT (4 Tage spaeter):
- Satz 1: Kurzer Bezug auf Email 1 (NICHT "haben Sie meine Email gelesen")
Beispiel: "Kurz nochmal wegen [Thema] --"
- Satz 2-4: Konkretes Wissens-Nugget oder Statistik aus dem Knowledge-Snippet (wenn vorhanden)
ODER eine konkrete Erfolgsgeschichte/Zahl aus dem Produkt
- Satz 5: "Falls Sie X Minuten haben, zeige ich Ihnen gerne wie Y funktioniert"
- Laenge: 100-150 Woerter
- Optional: Ein kurzer Link zur Website
""",
3: """
EMAIL 3 -- LETZTER VERSUCH (Tag 8, kurz und respektvoll):
- Satz 1: "Ich moechte nicht laestig sein -- deshalb ist das meine letzte Nachricht."
- Satz 2: Nochmal den Kern-Nutzen in EINEM Satz
- Satz 3: "Falls der Zeitpunkt nicht passt -- kein Problem. Ich bin da wenn es soweit ist."
- Optional: Kurze Social Proof-Referenz ("Wir helfen bereits X Hotels in Ihrer Region")
- Laenge: 60-80 Woerter -- kurz und klar
- Kein Druck, kein "letzte Chance"
""",
}
# Blacklisted Phrases fuer automatische Validierung
BLACKLIST_PHRASES = [
'ich hoffe diese email erreicht sie',
'i hope this email finds you',
'als marktfuehrer',
'wir sind die nummer eins',
'exklusives angebot',
'nur fuer kurze zeit',
'unverbindlich und kostenlos',
'revolutionaer',
'game-changer',
'synergie',
'holistisch',
'paradigmenwechsel',
'ich wende mich an sie',
'darf ich ihre aufmerksamkeit',
'unvergessliches erlebnis',
]
# ---------------------------------------------------------------------------
# Generator
# ---------------------------------------------------------------------------
class OutreachGenerator:
"""Generiert personalisierte Outreach-Emails via Claude."""
def __init__(self, db_path: str = 'email_agent.db',
anthropic_api_key: str = None,
knowledge_db_path: str = None):
self.db = ColdOutreachDB(db_path)
self.claude = anthropic.Anthropic(
api_key=anthropic_api_key or os.getenv('ANTHROPIC_API_KEY')
)
self.retriever = KnowledgeRetriever(knowledge_db_path)
# ── Oeffentliche API ─────────────────────────────────────────────────────
def generate_step(self, step: int, campaign_id: int = None) -> Dict:
"""
Generiert Emails fuer alle faelligen Kontakte eines Steps.
Returns:
Dict: {generated, accepted, rejected_low_score, errors}
"""
if step not in (1, 2, 3):
raise ValueError(f"Ungueltiger Step: {step} (erwartet 1, 2 oder 3)")
# Enriched Kontakte aktivieren (Sequenzen erstellen + Status -> active)
activated = self.db.activate_enriched_contacts(campaign_id=campaign_id)
if activated:
logger.info(f"{activated} Kontakte aktiviert (Sequenzen erstellt).")
# Faellige Sequenzen holen
due = self.db.get_due_emails()
step_due = [s for s in due if s['sequence_step'] == step]
if campaign_id:
step_due = [s for s in step_due
if self._contact_in_campaign(s['contact_id'], campaign_id)]
stats = {'generated': 0, 'accepted': 0, 'rejected_low_score': 0, 'errors': 0}
logger.info(f"Generiere Emails fuer Step {step}: {len(step_due)} faellige Kontakte")
for seq in step_due:
try:
result = self._generate_for_sequence(seq, step)
stats['generated'] += 1
if result.get('accepted'):
stats['accepted'] += 1
else:
stats['rejected_low_score'] += 1
logger.warning(
f"Email fuer {seq['email']} abgelehnt "
f"(Score: {result.get('score', 0):.2f})"
)
except Exception as e:
logger.error(f"Fehler bei Seq {seq['id']} ({seq['email']}): {e}")
stats['errors'] += 1
logger.info(f"Generierung Step {step} abgeschlossen: {stats}")
return stats
def regenerate(self, seq_id: int) -> Dict:
"""
Einzelne Sequence-Email neu generieren (nach manueller Ablehnung im Review).
"""
# Sequenz + Kontakt laden
due = self.db.get_due_emails()
seq = next((s for s in due if s['id'] == seq_id), None)
# Auch bereits generierte finden (get_pending_review gibt alle mit body)
if not seq:
pending = self.db.get_pending_review()
seq = next((s for s in pending if s['id'] == seq_id), None)
if not seq:
# Manuelle DB-Abfrage als Fallback
logger.error(f"Sequenz {seq_id} nicht gefunden.")
return {'error': 'not_found'}
step = seq['sequence_step']
result = self._generate_for_sequence(seq, step)
return result
# ── Interne Logik ────────────────────────────────────────────────────────
def _generate_for_sequence(self, seq: Dict, step: int) -> Dict:
"""Generiert Email fuer eine Sequence-Zeile. Retried bei niedrigem Score."""
icp = seq.get('icp', 'AG')
industry = seq.get('industry', '')
pain_point = seq.get('pain_point', '')
# Knowledge-Snippet holen (nur fuer Step 2)
knowledge_snippet = None
if step == 2:
knowledge_snippet = self.retriever.get_knowledge_snippet(
industry=industry,
pain_point=pain_point,
icp=icp,
)
for attempt in range(MAX_RETRIES + 1):
result = self._call_claude(seq, step, knowledge_snippet, attempt)
if not result:
continue
score = result.get('personalization_score', 0.0)
problems = self._validate_email(
subject=result.get('subject', ''),
body=result.get('body', ''),
contact=seq,
)
if score >= MIN_PERSONALIZATION_SCORE and not problems:
# Akzeptiert -- in DB speichern
self.db.save_generated_email(
seq_id=seq['id'],
subject=result['subject'],
body=result['body'],
personalization_score=score,
personalization_elements=result.get('personalization_elements', []),
knowledge_snippet=knowledge_snippet,
)
logger.info(
f"Email generiert: {seq['email']} | Step {step} | "
f"Score {score:.2f} | Produkt: {seq.get('icp_recommended_product', icp)}"
)
return {'accepted': True, 'score': score, 'seq_id': seq['id']}
elif attempt < MAX_RETRIES:
hint = f"Letzter Versuch war zu generisch (Score {score:.2f})."
if problems:
hint += f" Probleme: {'; '.join(problems[:2])}"
logger.debug(f"Retry {attempt + 1} fuer {seq['email']}: {hint}")
knowledge_snippet = f"{hint}\n\n{knowledge_snippet or ''}"
else:
# Endgueltig abgelehnt -- trotzdem speichern aber als rejected markieren
if result:
self.db.save_generated_email(
seq_id=seq['id'],
subject=result.get('subject', ''),
body=result.get('body', ''),
personalization_score=score,
personalization_elements=result.get('personalization_elements', []),
knowledge_snippet=knowledge_snippet,
)
# Review-Status auf 'rejected' -- muss manuell regeneriert werden
self.db.reject_email(seq['id'])
return {'accepted': False, 'score': score, 'seq_id': seq['id'],
'problems': problems}
return {'accepted': False, 'score': 0.0, 'seq_id': seq['id']}
def _call_claude(self, contact: Dict, step: int,
knowledge_snippet: Optional[str],
attempt: int) -> Optional[Dict]:
"""
Baut den Prompt und ruft Claude auf.
"""
icp = contact.get('icp', 'AG')
icp_context = LP_CONTEXT if icp == 'LP' else AG_CONTEXT
# Produkt-Spezifikation
product = contact.get('icp_recommended_product', '')
product_hint = f"\nEMPFOHLENES PRODUKT: {product}" if product else ''
# Knowledge-Snippet
knowledge_section = ''
if knowledge_snippet and step == 2:
knowledge_section = f"""
WISSENS-SNIPPET (einbetten in Email 2 als konkreten Mehrwert):
"{knowledge_snippet}"
Formuliere es natuerlich um -- nicht woertlich kopieren.
"""
# Retry-Hinweis
retry_hint = ''
if attempt > 0:
retry_hint = f"""
WICHTIG: Vorheriger Versuch war zu generisch.
Sei NOCH spezifischer. Nutze ECHTE Details aus Firma/Branche/Beschreibung.
Vermeide allgemeine Aussagen. Schreibe als haettest du die Website des Empfaengers gelesen.
"""
notes = contact.get('notes', '')
notes_section = f"\n- Recherche-Notiz/Outreach-Winkel: {notes}" if notes else ''
prompt = f"""{icp_context}{product_hint}
KONTAKT-DETAILS:
- Vorname: {contact.get('first_name', '')}
- Nachname: {contact.get('last_name', '')}
- Firma: {contact.get('company', '')}
- Branche: {contact.get('industry', '')}
- Firmenbeschreibung: {contact.get('company_description', '')}
- Schmerzpunkt: {contact.get('pain_point', '')}
- Sprache: {contact.get('language', 'de')}{notes_section}
{STEP_INSTRUCTIONS[step]}
{knowledge_section}
{retry_hint}
Generiere jetzt die Email als JSON."""
try:
response = self.claude.messages.create(
model=CLAUDE_MODEL,
max_tokens=600,
system=SYSTEM_PROMPT,
messages=[{'role': 'user', 'content': prompt}],
)
raw = response.content[0].text.strip()
start = raw.find('{')
end = raw.rfind('}') + 1
if start == -1:
return None
return json.loads(raw[start:end])
except (json.JSONDecodeError, Exception) as e:
logger.error(f"Claude-Generierung fehlgeschlagen fuer {contact.get('email')}: {e}")
return None
def _validate_email(self, subject: str, body: str, contact: Dict) -> List[str]:
"""
Prueft Email gegen harte Regeln.
Gibt Liste von Problemen zurueck (leer = OK).
"""
problems = []
# Laenge
word_count = len(body.split())
if word_count > 200:
problems.append(f"Zu lang: {word_count} Woerter (max 200)")
if word_count < 40:
problems.append(f"Zu kurz: {word_count} Woerter (min 40)")
# Blacklisted Phrasen
body_lower = body.lower()
for phrase in BLACKLIST_PHRASES:
if phrase in body_lower:
problems.append(f"Verbotene Phrase: '{phrase}'")
# Firmenname muss vorkommen
company = (contact.get('company') or '').strip()
if company and len(company) > 3:
if company.lower() not in body_lower and company.lower() not in subject.lower():
problems.append("Firmenname fehlt im Text")
# Vorname in Anrede
first_name = (contact.get('first_name') or '').strip()
if first_name and first_name not in body:
problems.append("Vorname fehlt in Anrede")
# Call-to-Action (Fragezeichen)
if '?' not in body:
problems.append("Kein Call-to-Action (keine Frage)")
# Nicht leer
if not subject.strip():
problems.append("Betreff fehlt")
if not body.strip():
problems.append("Email-Text fehlt")
return problems
def _contact_in_campaign(self, contact_id: int, campaign_id: int) -> bool:
"""Prueft ob ein Kontakt zur Kampagne gehoert."""
contact = self.db.get_contact(contact_id)
return contact and contact.get('campaign_id') == campaign_id
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def main():
parser = argparse.ArgumentParser(
description='Generiert personalisierte Outreach-Emails via Claude.'
)
group = parser.add_mutually_exclusive_group(required=True)
group.add_argument('--step', type=int, choices=[1, 2, 3],
help='Emails fuer diesen Sequence-Step generieren')
group.add_argument('--regenerate', type=int, metavar='SEQ_ID',
help='Einzelne Email neu generieren')
parser.add_argument('--campaign', type=int, metavar='ID',
help='Nur fuer diese Kampagne generieren')
parser.add_argument('--all', action='store_true',
help='Alle Kampagnen bearbeiten')
parser.add_argument('--db', default='email_agent.db', help='Pfad zur DB-Datei')
args = parser.parse_args()
generator = OutreachGenerator(db_path=args.db)
if args.step:
campaign_id = args.campaign if not args.all else None
print(f"\nGeneriere Step-{args.step} Emails...")
if campaign_id:
print(f"Kampagne: {campaign_id}")
stats = generator.generate_step(step=args.step, campaign_id=campaign_id)
print(f"\nErgebnis:")
print(f" Generiert: {stats['generated']}")
print(f" Akzeptiert: {stats['accepted']}")
print(f" Zu generisch: {stats['rejected_low_score']}")
print(f" Fehler: {stats['errors']}")
if stats['accepted'] > 0:
print(f"\nNaechster Schritt:")
print(f" python review_cli.py")
elif args.regenerate:
print(f"\nRegeneriere Sequence {args.regenerate}...")
result = generator.regenerate(args.regenerate)
if result.get('accepted'):
print(f" OK -- Score: {result['score']:.2f}")
else:
print(f" Abgelehnt -- Score: {result.get('score', 0):.2f}")
if result.get('problems'):
for p in result['problems']:
print(f" - {p}")
if __name__ == '__main__':
main()