Building an AI-Powered Lead Generation System — Architecture and Hard Lessons

class ProspectDiscoveryService:
    def __init__(self):
        self.sources = [
            LinkedInScraper(),
            ApolloAPIClient(api_key=settings.APOLLO_KEY),
            ClearbitEnrichment(api_key=settings.CLEARBIT_KEY),
        ]
        self.dedup_store = RedisBloomFilter("prospects:seen")
 
    async def discover(self, icp: ICPCriteria, limit: int = 100) -> list[RawProspect]:
        tasks = [source.search(icp, limit) for source in self.sources]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        prospects = []
        for result in results:
            if isinstance(result, Exception):
                logger.error("Source failed", exc_info=result)
                continue
            for p in result:
                # Bloom filter deduplication — O(1), probabilistic
                if not self.dedup_store.contains(p.email):
                    self.dedup_store.add(p.email)
                    prospects.append(p)
        
        return prospects[:limit]

ICP_QUALIFICATION_PROMPT = """
You are evaluating whether a company is a good fit for the following ideal customer profile (ICP):
 
ICP DEFINITION:
{icp_description}
 
COMPANY PROFILE:
Name: {company_name}
Industry: {industry}
Size: {employee_count} employees
Location: {location}
Tech stack: {tech_stack}
Recent funding: {funding}
Description: {description}
 
Evaluate this company against the ICP. Return JSON:
{{
  "score": <0-100 integer, 100 = perfect fit>,
  "tier": <"A" | "B" | "C" | "disqualified">,
  "reasoning": "<2-3 sentences explaining the score>",
  "flags": ["<any red flags>"],
  "hooks": ["<specific talking points for outreach>"]
}}
 
Be rigorous. A score above 70 means high confidence this is a good prospect.
"""
 
async def qualify_prospect(
    company: EnrichedCompany,
    icp: ICPDefinition
) -> QualificationResult:
    prompt = ICP_QUALIFICATION_PROMPT.format(
        icp_description=icp.description,
        **company.to_dict()
    )
    
    response = await openai_client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        response_format={"type": "json_object"},
        temperature=0.1,  # Low temperature for consistent scoring
    )
    
    return QualificationResult.model_validate_json(
        response.choices[0].message.content
    )

OUTREACH_PROMPT = """
Write a cold email from {sender_name} ({sender_role}) to {prospect_name} ({prospect_title}) at {company_name}.
 
CONTEXT ABOUT THE PROSPECT:
- Recent hooks: {hooks}
- Company situation: {company_context}
- Their likely challenges: {challenges}
 
SENDER'S VALUE PROPOSITION:
{value_prop}
 
REQUIREMENTS:
- Subject line: max 8 words, no emojis, no all-caps
- Body: 4-6 sentences maximum
- First sentence: must reference something specific to them (not generic)
- CTA: one specific, low-friction ask
- Tone: peer-to-peer, not salesy
- No phrases like: "I hope this finds you well", "I came across your profile"
 
Return JSON: {{"subject": "...", "body": "..."}}
"""

// Tracking webhook — receives open/click events
export async function POST(request: Request) {
  const event = await request.json() as EmailEvent
  
  await db.emailEvent.create({
    data: {
      emailId: event.emailId,
      type: event.type, // "open" | "click" | "reply"
      timestamp: new Date(event.timestamp),
      metadata: event.metadata,
    }
  })
  
  // Update campaign stats
  await db.$executeRaw`
    UPDATE campaign_stats
    SET ${event.type}_count = ${event.type}_count + 1,
        last_event_at = NOW()
    WHERE campaign_id = (
      SELECT campaign_id FROM emails WHERE id = ${event.emailId}
    )
  `
}