Розробка AI-системи збагачення даних про клієнта з відкритих джерел

Розробка AI-системи збагачення даних клієнта з відкритих джерел

AI-система збагачення даних (data enrichment) автоматично доповнює профіль клієнта інформацією з відкритих джерел: LinkedIn, Crunchbase, GitHub, новинні сайти, реєстри компаній. Це прискорює onboarding, покращує якість lead scoring та персоналізацію.

Архітектура системи

from dataclasses import dataclass
from typing import Optional

@dataclass
class EnrichedProfile:
    # Вихідні дані
    email: str
    company_name: str
    # Збагачені дані
    linkedin_profile: Optional[dict] = None
    company_info: Optional[dict] = None
    news_mentions: Optional[list] = None
    tech_stack: Optional[list] = None
    funding_info: Optional[dict] = None
    enrichment_score: float = 0.0  # Впевненість у даних

class DataEnrichmentPipeline:
    def __init__(self):
        self.enrichers = [
            LinkedInEnricher(),
            ClearbitEnricher(),
            CrunchbaseEnricher(),
            NewsEnricher(),
            GitHubEnricher(),
        ]

    async def enrich(self, email: str, company: str) -> EnrichedProfile:
        profile = EnrichedProfile(email=email, company_name=company)

        # Паралельне збагачення з усіх джерел
        tasks = [enricher.enrich(email, company) for enricher in self.enrichers]
        results = await asyncio.gather(*tasks, return_exceptions=True)

        for enricher, result in zip(self.enrichers, results):
            if isinstance(result, Exception):
                continue  # Одне джерело не зламає весь pipeline
            profile = enricher.apply_to_profile(profile, result)

        profile.enrichment_score = self._compute_score(profile)
        return profile

Збагачення LinkedIn через ProxyCurl

import httpx

class LinkedInEnricher:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://nubela.co/proxycurl/api"

    async def enrich(self, email: str, company: str) -> dict:
        async with httpx.AsyncClient() as client:
            # Пошук профілю за email
            response = await client.get(
                f"{self.base_url}/linkedin/profile/resolve/email",
                params={"email": email},
                headers={"Authorization": f"Bearer {self.api_key}"}
            )

            if response.status_code != 200:
                return {}

            profile_url = response.json().get('linkedin_profile_url')
            if not profile_url:
                return {}

            # Отримання повного профілю
            profile_response = await client.get(
                f"{self.base_url}/v2/linkedin",
                params={"url": profile_url, "skills": "include"},
                headers={"Authorization": f"Bearer {self.api_key}"}
            )

            return profile_response.json()

AI-вилучення технологічного стека

class TechStackExtractor:
    def __init__(self):
        self.llm = Anthropic()

    async def extract_from_website(self, domain: str) -> list[str]:
        """Вилучення tech stack з веб-сайту компанії через AI"""
        # Збір контенту з веб-сайту
        job_postings = await self._scrape_job_postings(domain)
        about_page = await self._scrape_page(f"https://{domain}/about")

        combined_text = ' '.join([about_page] + job_postings[:5])

        response = self.llm.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=300,
            messages=[{
                "role": "user",
                "content": f"""Extract technology stack from this company information.
Return JSON array of technology names (programming languages, frameworks, cloud platforms, databases).
Only include clearly mentioned technologies.

Text: {combined_text[:3000]}"""
            }]
        )

        return json.loads(response.content[0].text)

Якість та дедупликація збагачених даних

Різні джерела надають різні дані для однієї компанії. Потрібна логіка узгодження:

def reconcile_company_info(sources: list[dict]) -> dict:
    """Об'єднання даних про компанію з кількох джерел"""
    reconciled = {}

    # Пріоритет джерел: офіційні реєстри > Clearbit > Web scraping
    priority_order = ['company_registry', 'clearbit', 'linkedin', 'web_scraping']

    for field in ['employee_count', 'founded_year', 'industry', 'headquarters']:
        for source_name in priority_order:
            source = next((s for s in sources if s.get('source') == source_name), None)
            if source and field in source:
                reconciled[field] = source[field]
                break

    return reconciled

Типовий результат: збагачення 80-90% контактів CRM додатковими даними за 2-5 секунд на запис. Якість даних: 85-90% точність для базових полів (посада, галузь, розмір компанії).