Cara Membangun Tim Sales AI Agent Menggunakan CrewAI: Tutorial Lengkap dengan Kode

Margabagus.com – Studi menunjukkan bahwa 65% perusahaan sekarang menggunakan generative AI, namun masih menghadapi tantangan signifikan dalam akurasi hasil dan keamanan data. CrewAI hadir sebagai solusi dengan framework Python yang dibangun dari awal, bukan turunan dari LangChain atau framework lain. Dengan lebih dari 100,000 developer yang telah tersertifikasi melalui kursus komunitas di learn.crewai.com, CrewAI telah memproses lebih dari 10 juta agents per bulan dan digunakan oleh hampir setengah dari Fortune 500. Dalam tutorial implementasi CrewAI untuk sales ini, saya akan memandu Anda langkah demi langkah membangun tim sales AI yang bekerja otomatis, lengkap dengan kode yang bisa langsung Anda gunakan.

Persiapan Environment dan Instalasi

Photo by Nubelson Fernandes on Unsplash

Sebelum memulai, pastikan sistem Anda memenuhi requirements. CrewAI membutuhkan Python versi >=3.10 <3.13. Framework ini menggunakan UV untuk dependency management, menawarkan setup dan eksekusi yang seamless.

Langkah-Langkah (HowTo)

Langkah 1: Install CrewAI

# Install CrewAI basic package pip install crewai # Atau install dengan tools tambahan untuk agents pip install 'crewai[tools]'

Langkah 2: Setup API Keys

Buat file .env di root project Anda:

# Required API Keys OPENAI_API_KEY=your_openai_api_key_here SERPER_API_KEY=your_serper_api_key_here # Untuk web search SERPAPI_API_KEY=your_serpapi_key_here # Alternative search # Optional untuk integrasi lanjutan AIRTABLE_API_KEY=your_airtable_key_here BRAVE_API_KEY=your_brave_search_key_here

Anda bisa mendapatkan Serper API key dari Serper.dev. Untuk OpenAI, gunakan API key dari platform.openai.com.

Langkah 3: Membuat Project Structure

CrewAI menyediakan CLI untuk membuat project structure yang rapi:

# Membuat project baru untuk sales automation
crewai create crew sales_automation

# Navigasi ke folder project
cd sales_automation

Ini akan generate struktur project seperti ini:

sales_automation/
├── .gitignore
├── pyproject.toml
├── README.md
├── .env
└── src/
    └── sales_automation/
        ├── __init__.py
        ├── main.py
        ├── crew.py
        ├── tools/
        │   ├── custom_tool.py
        │   └── __init__.py
        └── config/
            ├── agents.yaml
            └── tasks.yaml

Contoh Kode Minimum crew.py

Setelah environment siap dan struktur proyek terbentuk, berikut contoh kode minimum
crew.py. Kode ini sudah bisa dijalankan sebagai pipeline dasar: mendefinisikan agents,
tasks, dan mengeksekusinya secara sekuensial.

Tujuannya agar Anda bisa langsung mencoba workflow sederhana tanpa harus memahami semua detail
konfigurasi YAML atau custom tools lebih dulu. Versi lengkap dan modular akan dibahas setelah ini.

# == crew.py (minimum viable example) == # env: Python 3.10–3.12 # pip install crewai "crewai[tools]" python-dotenv from __future__ import annotations import os from dataclasses import dataclass from typing import Any, Dict, List from dotenv import load_dotenv load_dotenv() from crewai import Agent, Task, Crew, Process # ---- Config minimal ---- OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") assert OPENAI_API_KEY, "Set OPENAI_API_KEY in .env" MODEL_PLANNER = os.getenv("MODEL_PLANNER", "gpt-4o-mini") MODEL_WRITER = os.getenv("MODEL_WRITER", "gpt-4o-mini") # ---- Agents ---- lead_researcher = Agent( role="Lead Researcher", goal="Mengumpulkan info prospek relevan (company, jabatan, pain points) dari input ICP dan kata kunci.", backstory="Peneliti growth B2B dengan disiplin pada data yang dapat ditindaklanjuti.", allow_delegation=False, verbose=True, llm=MODEL_PLANNER ) lead_scorer = Agent( role="Lead Scorer", goal="Memberi skor kecocokan prospek terhadap ICP (1–100) + alasan singkat.", backstory="Analis penjualan berfokus pada prioritas eksekusi.", allow_delegation=False, verbose=True, llm=MODEL_PLANNER ) email_personalizer = Agent( role="Email Personalizer", goal="Membuat cold email singkat, personal, bernilai jelas, dengan CTA spesifik.", backstory="Copywriter penjualan dengan pendekatan empatik dan ringkas.", allow_delegation=False, verbose=True, llm=MODEL_WRITER ) followup_strategist = Agent( role="Follow-up Strategist", goal="Mendesain cadence follow-up 3 langkah (email/LinkedIn) sesuai prioritas.", backstory="Lifecycle marketer yang paham timing & channel.", allow_delegation=False, verbose=True, llm=MODEL_PLANNER ) # ---- Tasks ---- t_research = Task( description=( "Riset singkat prospek berdasarkan ICP & keywords. " "Keluaran: ringkasan 3–5 poin + 3 data spesifik perusahaan." ), expected_output="Bullet ringkas + data spesifik (sumber/URL bila ada).", agent=lead_researcher ) t_score = Task( description=( "Skor ICP (1–100) + alasan 2–3 kalimat. Label: hot/warm/cold (>=80 hot; 60–79 warm;

Membuat Agent untuk Tim Sales

Photo by Fotis Fotopoulos on Unsplash

Mari kita mulai dengan mendefinisikan agent-agent sales kita. Cara membangun tim sales AI agent yang efektif dimulai dengan mendefinisikan role yang jelas.

Baca juga artikel menarik lainnya: Workflow Otomatis Freelance: Kirim Invoice dan Email Follow Up Pakai n8n

Konfigurasi Agents dengan YAML

Edit file src/sales_automation/config/agents.yaml:

# Agent untuk Research Lead
lead_researcher:
  role: >
    Senior Sales Intelligence Analyst
  goal: >
    Mengumpulkan informasi mendalam tentang prospek {company_name}
    termasuk data perusahaan, decision makers, dan pain points mereka
  backstory: >
    Anda adalah seorang sales intelligence expert dengan 15 tahun pengalaman
    dalam B2B research. Anda memiliki kemampuan luar biasa dalam menemukan
    informasi tersembunyi dan mengidentifikasi peluang sales yang tidak
    terlihat oleh orang lain. Anda selalu update dengan tren industri terbaru.
  llm: gpt-4  # Bisa diganti dengan model lain

# Agent untuk Lead Scoring  
lead_scorer:
  role: >
    Lead Qualification Specialist
  goal: >
    Mengevaluasi dan memberikan skor pada lead {company_name} berdasarkan
    kriteria ICP (Ideal Customer Profile) dan likelihood to convert
  backstory: >
    Dengan background di data science dan sales operations, Anda ahli
    dalam mengidentifikasi high-quality leads. Anda telah mengembangkan
    sistem scoring yang telah meningkatkan conversion rate hingga 40%.
  llm: gpt-4

# Agent untuk Email Personalization
email_personalizer:
  role: >
    B2B Email Marketing Specialist  
  goal: >
    Membuat email outreach yang sangat personal dan compelling untuk
    {contact_name} di {company_name} yang mendorong response rate tinggi
  backstory: >
    Anda adalah copywriter B2B yang telah menulis ribuan cold email
    dengan average open rate 45% dan reply rate 15%. Anda memahami
    psikologi B2B buyer dan cara membuat pesan yang resonates.
  llm: gpt-4

# Agent untuk Follow-up Strategy
follow_up_strategist:
  role: >
    Sales Cadence Optimization Expert
  goal: >
    Merancang strategi follow-up yang optimal untuk {company_name}
    termasuk timing, channel, dan messaging approach
  backstory: >
    Sebagai veteran sales dengan track record closing deals bernilai
    jutaan dollar, Anda memahami pentingnya persistent follow-up
    yang tidak mengganggu. Anda ahli dalam multi-channel outreach.
  llm: gpt-4

Implementasi Agents dalam Python

Sekarang buat implementasi di src/sales_automation/crew.py:

from crewai import Agent, Crew, Process, Task
from crewai.project import CrewBase, agent, crew, task, before_kickoff, after_kickoff
from crewai_tools import SerperDevTool, WebsiteSearchTool, FileWriterTool
from typing import List
import os

@CrewBase
class SalesAutomationCrew:
    """Sales Automation crew untuk otomasi proses sales"""
    
    # Path ke file konfigurasi
    agents_config = 'config/agents.yaml'
    tasks_config = 'config/tasks.yaml'
    
    # Hook untuk preprocessing
    @before_kickoff
    def prepare_data(self, inputs):
        """Validasi dan prepare input data sebelum crew mulai"""
        print(f" Memulai sales automation untuk: {inputs.get('company_name')}")
        
        # Validasi required inputs
        required_fields = ['company_name', 'contact_name', 'industry']
        for field in required_fields:
            if field not in inputs:
                raise ValueError(f"Missing required field: {field}")
                
        # Tambahkan default values jika perlu
        inputs.setdefault('company_size', 'Unknown')
        inputs.setdefault('budget_range', 'Not specified')
        
        return inputs
    
    # Hook untuk postprocessing
    @after_kickoff  
    def process_results(self, result):
        """Process dan save hasil setelah crew selesai"""
        print(f"✅ Sales automation selesai! Saving results...")
        
        # Save hasil ke file
        output_dir = 'output'
        os.makedirs(output_dir, exist_ok=True)
        
        with open(f"{output_dir}/sales_report_{result.raw[:20]}.md", 'w') as f:
            f.write(result.raw)
            
        return result
    
    # Define agents menggunakan decorator
    @agent
    def lead_researcher(self) -> Agent:
        """Agent untuk research lead information"""
        return Agent(
            config=self.agents_config['lead_researcher'],
            tools=[
                SerperDevTool(),  # Untuk web search
                WebsiteSearchTool(),  # Untuk scraping website
            ],
            verbose=True,  # Tampilkan thinking process
            memory=True,  # Enable memory untuk context retention
            max_iter=3,  # Maksimal iterasi untuk avoid infinite loop
            allow_delegation=True  # Bisa delegate task ke agent lain
        )
    
    @agent
    def lead_scorer(self) -> Agent:
        """Agent untuk scoring leads"""
        return Agent(
            config=self.agents_config['lead_scorer'],
            tools=[
                # Custom tool untuk akses database (dijelaskan nanti)
            ],
            verbose=True,
            memory=True,
            max_iter=2
        )
    
    @agent
    def email_personalizer(self) -> Agent:
        """Agent untuk create personalized emails"""
        return Agent(
            config=self.agents_config['email_personalizer'],
            tools=[
                FileWriterTool(),  # Untuk save email drafts
            ],
            verbose=True,
            memory=True,
            max_execution_time=300,  # Max 5 menit per email
            system_template="""You are an expert B2B email writer.
            Always write concise, value-focused emails under 150 words.
            Use the PAS (Problem-Agitate-Solution) framework.""",
            response_template="""Email Draft:
            Subject: {subject}
            
            Body:
            {body}
            
            CTA: {cta}"""
        )
    
    @agent
    def follow_up_strategist(self) -> Agent:
        """Agent untuk follow-up strategy"""
        return Agent(
            config=self.agents_config['follow_up_strategist'],
            verbose=True,
            memory=True,
            function_calling_llm="gpt-3.5-turbo"  # Gunakan model lebih murah untuk planning
        )

Mendefinisikan Tasks untuk Workflow Sales

Tasks adalah unit kerja yang akan dilakukan oleh agents. Edit src/sales_automation/config/tasks.yaml:

# Research task
research_lead:
  description: >
    Lakukan research mendalam tentang {company_name} meliputi:
    1. Informasi perusahaan (size, revenue, recent news)
    2. Key decision makers terutama {contact_name}
    3. Tech stack yang mereka gunakan
    4. Recent initiatives atau pain points
    5. Kompetitor dan positioning mereka
    
    Gunakan web search dan website mereka untuk data akurat.
  expected_output: >
    Detailed report dalam markdown format dengan sections:
    - Company Overview
    - Key Personnel
    - Technology Stack  
    - Business Challenges
    - Sales Opportunities
    - Recommended Approach
  agent: lead_researcher

# Scoring task
score_lead:
  description: >
    Berdasarkan research data, evaluate lead {company_name} dengan kriteria:
    1. Company size fit (ideal: 50-500 employees)
    2. Industry fit (prioritas: SaaS, Fintech, E-commerce)
    3. Budget likelihood (based on company revenue)
    4. Technology readiness
    5. Urgency indicators
    
    Berikan score 1-100 dengan breakdown per kriteria.
  expected_output: >
    Lead score report dengan:
    - Overall score (1-100)
    - Breakdown score per kriteria
    - Qualification status (Hot/Warm/Cold)
    - Reasoning untuk score
    - Recommended next actions
  agent: lead_scorer
  context:
    - research_lead  # Gunakan output dari research task

# Email creation task
create_email:
  description: >
    Buat personalized cold email untuk {contact_name} di {company_name}
    berdasarkan research findings. Email harus:
    1. Subject line yang attention-grabbing dan relevant
    2. Opening yang personal berdasarkan recent company news
    3. Value proposition yang address specific pain point mereka
    4. Social proof yang relevant dengan industry mereka
    5. Clear CTA yang low-commitment
    
    Maksimal 150 kata, tone professional tapi conversational.
  expected_output: >
    Email draft lengkap dengan:
    - 3 variasi subject line untuk A/B testing
    - Email body yang ready to send
    - Follow-up email draft (jika no response)
    - Personalization tokens highlighted
  agent: email_personalizer
  context:
    - research_lead
    - score_lead

# Follow-up strategy task
plan_follow_up:
  description: >
    Design optimal follow-up sequence untuk {company_name} meliputi:
    1. Timeline untuk setiap follow-up (day 3, 7, 14, 21)
    2. Channel untuk setiap touchpoint (email, LinkedIn, phone)
    3. Message theme untuk setiap follow-up
    4. Trigger events untuk accelerate follow-up
    5. Exit criteria (kapan stop follow-up)
  expected_output: >
    Follow-up playbook dengan:
    - Detailed timeline dan channel strategy
    - Message templates untuk setiap touchpoint
    - Trigger events checklist
    - Success metrics untuk track
  agent: follow_up_strategist
  context:
    - score_lead
    - create_email

Implementasi Tasks dalam Python

Tambahkan task methods di crew.py:

# Lanjutan dari class SalesAutomationCrew

    @task
    def research_lead(self) -> Task:
        """Task untuk research lead information"""
        return Task(
            config=self.tasks_config['research_lead'],
            output_file='output/research_report.md'  # Save output ke file
        )
    
    @task
    def score_lead(self) -> Task:
        """Task untuk scoring lead"""
        return Task(
            config=self.tasks_config['score_lead'],
            output_file='output/lead_score.json',
            # Custom callback setelah task selesai
            callback=self.handle_score_result
        )
    
    @task  
    def create_email(self) -> Task:
        """Task untuk create email"""
        return Task(
            config=self.tasks_config['create_email'],
            output_file='output/email_drafts.md',
            human_input=True  # Minta human review sebelum finalize
        )
    
    @task
    def plan_follow_up(self) -> Task:
        """Task untuk follow-up planning"""
        return Task(
            config=self.tasks_config['plan_follow_up'],
            output_file='output/follow_up_plan.md'
        )
    
    # Callback function untuk handle scoring result
    def handle_score_result(self, output):
        """Process scoring result untuk determine next steps"""
        # Parse score dari output
        score = self.extract_score(output.raw)
        
        if score >= 80:
            print(" HOT LEAD! Prioritizing for immediate outreach...")
        elif score >= 60:
            print("✨ WARM LEAD. Adding to nurture sequence...")
        else:
            print("❄️ COLD LEAD. Scheduling for quarterly check-in...")
            
        return output
    
    def extract_score(self, text):
        """Extract numerical score dari text output"""
        import re
        match = re.search(r'Overall score[:\s]+(\d+)', text)
        return int(match.group(1)) if match else 0
    
    @crew
    def crew(self) -> Crew:
        """Creates the Sales Automation crew"""
        return Crew(
            agents=self.agents,  # Automatically gets all agents
            tasks=self.tasks,    # Automatically gets all tasks
            process=Process.sequential,  # Sequential execution
            verbose=True,
            memory=True,  # Enable crew-level memory
            embedder={  # Configure memory embeddings
                "provider": "openai",
                "config": {
                    "model": "text-embedding-3-small"
                }
            },
            cache=True,  # Enable caching untuk efficiency
            max_rpm=10,  # Rate limiting
            share_crew=True  # Share untuk analytics (optional)
        )

Implementasi crew.py (Ringkas)

Segmen ini merangkum cara menggabungkan agents dan tasks ke dalam satu alur eksekusi.
Versi ringkas ini fokus pada orkestrasi berurutan (sequential) untuk validasi cepat.
Untuk versi lengkap dengan memory, tools, dan flows lanjutan, lihat kode minimum/utama di bagian sebelumnya.

# == crew.py (ringkas) == # env: Python 3.10–3.12 # pip install crewai "crewai[tools]" python-dotenv from __future__ import annotations import os from typing import Dict, Any, List from dotenv import load_dotenv; load_dotenv() from crewai import Agent, Task, Crew, Process # --- Konfigurasi minimal --- OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") assert OPENAI_API_KEY, "Set OPENAI_API_KEY di .env" MODEL_PLANNER = os.getenv("MODEL_PLANNER", "gpt-4o-mini") MODEL_WRITER = os.getenv("MODEL_WRITER", "gpt-4o-mini") # --- Definisi agents (ringkas) --- lead_researcher = Agent( role="Lead Researcher", goal="Kumpulkan info prospek relevan dari ICP + keywords.", backstory="Peneliti growth B2B yang fokus pada data tindakan.", llm=MODEL_PLANNER, verbose=True, allow_delegation=False ) lead_scorer = Agent( role="Lead Scorer", goal="Skor kecocokan ICP (1–100) + alasan singkat.", backstory="Analis prioritas penjualan.", llm=MODEL_PLANNER, verbose=True, allow_delegation=False ) email_personalizer = Agent( role="Email Personalizer", goal="Tulis cold email personal

Membuat Custom Tools untuk Sales

Otomasi penjualan dengan AI agent memerlukan tools khusus. Mari buat custom tool untuk akses CRM:

# src/sales_automation/tools/crm_tool.py
from crewai_tools import BaseTool
from typing import Type, Any, Optional
from pydantic import BaseModel, Field
import requests
import os

class CRMSearchInput(BaseModel):
    """Input schema untuk CRM search"""
    company_name: str = Field(..., description="Nama perusahaan yang dicari")
    field: Optional[str] = Field("all", description="Field specific yang dicari")

class CRMTool(BaseTool):
    name: str = "CRM Search Tool"
    description: str = "Search customer data dari CRM system"
    args_schema: Type[BaseModel] = CRMSearchInput
    
    def _run(self, company_name: str, field: str = "all") -> str:
        """
        Execute CRM search
        """
        # Contoh implementasi dengan API CRM
        api_key = os.getenv("CRM_API_KEY")
        base_url = "https://api.yourcrm.com/v1"
        
        try:
            # Search company
            response = requests.get(
                f"{base_url}/companies/search",
                headers={"Authorization": f"Bearer {api_key}"},
                params={"query": company_name}
            )
            
            if response.status_code == 200:
                data = response.json()
                
                if field == "all":
                    return self._format_company_data(data)
                else:
                    return str(data.get(field, "Field not found"))
            else:
                return f"Error searching CRM: {response.status_code}"
                
        except Exception as e:
            return f"Error accessing CRM: {str(e)}"
    
    def _format_company_data(self, data: dict) -> str:
        """Format data untuk readable output"""
        return f"""
Company Information:
- Name: {data.get('name')}
- Industry: {data.get('industry')}  
- Size: {data.get('employee_count')} employees
- Revenue: ${data.get('annual_revenue', 'N/A')}
- Last Contact: {data.get('last_contact_date')}
- Current Status: {data.get('status')}
- Key Contacts: {len(data.get('contacts', []))} people
"""

# Tool untuk Email Tracking
class EmailTrackerTool(BaseTool):
    name: str = "Email Tracker"
    description: str = "Track email opens, clicks, and responses"
    
    def _run(self, email_id: str) -> str:
        """Track email metrics"""
        # Implementasi tracking logic
        # Bisa integrate dengan SendGrid, Mailgun, etc
        return f"Email {email_id} - Opens: 3, Clicks: 1, Replied: No"

# Tool untuk LinkedIn Enrichment
class LinkedInEnrichmentTool(BaseTool):
    name: str = "LinkedIn Enrichment"
    description: str = "Enrich contact data dari LinkedIn"
    
    def _run(self, person_name: str, company: str) -> str:
        """Get LinkedIn data"""
        # Implementasi dengan LinkedIn API atau scraping tool
        # IMPORTANT: Pastikan comply dengan LinkedIn ToS
        return f"LinkedIn profile untuk {person_name} at {company}"

Mengimplementasikan Flows untuk Complex Workflows

CrewAI Flows memungkinkan orkestrasi yang lebih kompleks. Ini contoh implementasi sales pipeline dengan Flows:

# src/sales_automation/flows/sales_pipeline_flow.py
from crewai.flow.flow import Flow, listen, start, router
from typing import Dict, List, Any
import json

class SalesPipelineFlow(Flow):
    """
    Advanced sales pipeline dengan conditional logic
    """
    
    def __init__(self):
        super().__init__()
        self.qualified_leads = []
        self.email_queue = []
    
    @start()
    def fetch_leads(self) -> List[Dict]:
        """Starting point: Fetch leads dari berbagai sources"""
        print(" Fetching leads dari CRM dan marketing tools...")
        
        # Simulate fetching dari multiple sources
        leads = [
            {
                "company_name": "TechCorp Indonesia",
                "contact_name": "Budi Santoso",
                "industry": "SaaS",
                "source": "Website Form"
            },
            {
                "company_name": "FinanceHub",
                "contact_name": "Sarah Chen",
                "industry": "Fintech",
                "source": "LinkedIn"
            }
        ]
        
        print(f"✅ Found {len(leads)} new leads")
        return leads
    
    @listen(fetch_leads)
    def enrich_leads(self, leads: List[Dict]) -> List[Dict]:
        """Enrich setiap lead dengan additional data"""
        enriched = []
        
        for lead in leads:
            print(f" Enriching {lead['company_name']}...")
            
            # Use crew untuk enrichment
            from sales_automation.crew import SalesAutomationCrew
            crew = SalesAutomationCrew().crew()
            
            result = crew.kickoff(inputs={
                "company_name": lead["company_name"],
                "contact_name": lead["contact_name"],
                "industry": lead["industry"]
            })
            
            # Add enrichment results
            lead["enrichment_data"] = result.raw
            lead["research_complete"] = True
            enriched.append(lead)
            
        return enriched
    
    @router(enrich_leads)
    def route_by_score(self, lead: Dict) -> str:
        """Route leads based on scoring"""
        # Extract score dari enrichment data
        score = self._extract_lead_score(lead["enrichment_data"])
        lead["score"] = score
        
        if score >= 80:
            return "high_priority_flow"
        elif score >= 60:
            return "medium_priority_flow"
        else:
            return "low_priority_flow"
    
    @listen("high_priority_flow")
    def process_hot_lead(self, lead: Dict) -> Dict:
        """Process hot leads dengan immediate action"""
        print(f" Processing HOT lead: {lead['company_name']} (Score: {lead['score']})")
        
        # Generate personalized email immediately
        email_content = self._generate_priority_email(lead)
        
        # Schedule immediate outreach
        lead["email_content"] = email_content
        lead["send_immediately"] = True
        lead["follow_up_days"] = [1, 3, 5]  # Aggressive follow-up
        
        self.qualified_leads.append(lead)
        return lead
    
    @listen("medium_priority_flow")
    def process_warm_lead(self, lead: Dict) -> Dict:
        """Process warm leads dengan nurture sequence"""
        print(f"✨ Processing WARM lead: {lead['company_name']} (Score: {lead['score']})")
        
        # Add to nurture campaign
        lead["nurture_sequence"] = "standard_b2b"
        lead["send_after_days"] = 2
        lead["follow_up_days"] = [7, 14, 30]
        
        self.qualified_leads.append(lead)
        return lead
    
    @listen("low_priority_flow")
    def process_cold_lead(self, lead: Dict) -> Dict:
        """Process cold leads dengan long-term nurture"""
        print(f"❄️ Processing COLD lead: {lead['company_name']} (Score: {lead['score']})")
        
        # Add to quarterly newsletter only
        lead["add_to_newsletter"] = True
        lead["recheck_after_days"] = 90
        
        return lead
    
    @listen([process_hot_lead, process_warm_lead])
    def prepare_email_batch(self, lead: Dict) -> None:
        """Prepare emails untuk batch sending"""
        self.email_queue.append({
            "to": lead["contact_name"],
            "company": lead["company_name"],
            "priority": "high" if lead.get("send_immediately") else "normal",
            "content": lead.get("email_content", ""),
            "scheduled_date": self._calculate_send_date(lead)
        })
    
    def _extract_lead_score(self, enrichment_data: str) -> int:
        """Extract score dari enrichment results"""
        # Parse score dari AI output
        import re
        match = re.search(r'score[:\s]+(\d+)', enrichment_data.lower())
        return int(match.group(1)) if match else 50
    
    def _generate_priority_email(self, lead: Dict) -> str:
        """Generate high-priority email content"""
        # Implement email generation logic
        return f"Personalized email for {lead['contact_name']} at {lead['company_name']}"
    
    def _calculate_send_date(self, lead: Dict) -> str:
        """Calculate optimal send date"""
        from datetime import datetime, timedelta
        
        if lead.get("send_immediately"):
            return datetime.now().isoformat()
        
        days_delay = lead.get("send_after_days", 1)
        return (datetime.now() + timedelta(days=days_delay)).isoformat()
    
    def run_pipeline(self):
        """Execute the complete pipeline"""
        # Kickoff the flow
        self.kickoff()
        
        # Summary
        print("\n Pipeline Summary:")
        print(f"Total Qualified Leads: {len(self.qualified_leads)}")
        print(f"Emails Queued: {len(self.email_queue)}")
        
        # Save results
        with open("output/pipeline_results.json", "w") as f:
            json.dump({
                "qualified_leads": self.qualified_leads,
                "email_queue": self.email_queue,
                "timestamp": datetime.now().isoformat()
            }, f, indent=2)

Main Script untuk Menjalankan Sales Automation

Sekarang buat main script di src/sales_automation/main.py:

#!/usr/bin/env python
import sys
import os
from datetime import datetime
from typing import Dict, List
import json

# Import crews dan flows
from sales_automation.crew import SalesAutomationCrew
from sales_automation.flows.sales_pipeline_flow import SalesPipelineFlow

def run_single_lead_automation():
    """
    Run automation untuk single lead
    Cocok untuk testing atau processing specific lead
    """
    print(" Starting Single Lead Automation")
    print("=" * 50)
    
    # Input data untuk single lead
    inputs = {
        'company_name': 'TechStartup Indonesia',
        'contact_name': 'Ahmad Rahman',
        'industry': 'SaaS',
        'company_size': '50-100',
        'website': 'https://techstartup.id'
    }
    
    # Create crew instance
    crew = SalesAutomationCrew().crew()
    
    # Execute crew
    result = crew.kickoff(inputs=inputs)
    
    # Display results
    print("\n RESULTS:")
    print("=" * 50)
    print(result.raw)
    
    # Save detailed results
    save_results(result, inputs['company_name'])
    
    return result

def run_bulk_pipeline():
    """
    Run full sales pipeline untuk multiple leads
    Menggunakan Flows untuk complex orchestration
    """
    print(" Starting Bulk Sales Pipeline")
    print("=" * 50)
    
    # Initialize flow
    flow = SalesPipelineFlow()
    
    # Run the pipeline
    flow.run_pipeline()
    
    print("\n✅ Pipeline completed!")

def run_custom_workflow():
    """
    Run custom workflow dengan specific configuration
    """
    print(" Starting Custom Workflow")
    print("=" * 50)
    
    # Load leads dari CSV atau database
    leads = load_leads_from_source()
    
    # Process dengan custom logic
    results = []
    for lead in leads:
        try:
            # Skip jika sudah diprocess recently
            if was_processed_recently(lead):
                print(f"⏭️ Skipping {lead['company_name']} - recently processed")
                continue
            
            # Create crew dengan custom config
            crew = SalesAutomationCrew()
            
            # Customize agents untuk specific use case
            if lead['industry'] == 'Enterprise':
                # Use more sophisticated approach untuk enterprise
                crew.lead_researcher().max_iter = 5
                crew.email_personalizer().llm = "gpt-4"
            
            # Execute
            result = crew.crew().kickoff(inputs=lead)
            results.append({
                "lead": lead,
                "result": result.raw,
                "timestamp": datetime.now().isoformat()
            })
            
        except Exception as e:
            print(f"❌ Error processing {lead.get('company_name')}: {str(e)}")
            continue
    
    # Save all results
    save_bulk_results(results)
    
    return results

def load_leads_from_source() -> List[Dict]:
    """Load leads dari CSV, database, atau API"""
    # Contoh implementation
    # Bisa diganti dengan actual data source
    return [
        {
            "company_name": "Enterprise Corp",
            "contact_name": "John Doe",
            "industry": "Enterprise",
            "website": "https://enterprise.com"
        },
        {
            "company_name": "StartupXYZ",
            "contact_name": "Jane Smith",
            "industry": "Startup",
            "website": "https://startupxyz.com"
        }
    ]

def was_processed_recently(lead: Dict, days: int = 7) -> bool:
    """Check apakah lead sudah diprocess recently"""
    # Implement logic untuk check processing history
    # Bisa dari database atau file log
    return False

def save_results(result, company_name: str):
    """Save results ke organized structure"""
    # Create directory structure
    output_dir = f"output/{datetime.now().strftime('%Y%m%d')}/{company_name}"
    os.makedirs(output_dir, exist_ok=True)
    
    # Save different components
    with open(f"{output_dir}/full_report.md", "w") as f:
        f.write(result.raw)
    
    # Extract dan save specific sections
    sections = extract_sections(result.raw)
    for section_name, content in sections.items():
        with open(f"{output_dir}/{section_name}.txt", "w") as f:
            f.write(content)

def save_bulk_results(results: List[Dict]):
    """Save bulk processing results"""
    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
    
    with open(f"output/bulk_results_{timestamp}.json", "w") as f:
        json.dump(results, f, indent=2)
    
    # Create summary report
    create_summary_report(results)

def extract_sections(content: str) -> Dict[str, str]:
    """Extract sections dari AI output"""
    sections = {}
    current_section = "intro"
    current_content = []
    
    for line in content.split('\n'):
        if line.startswith('#'):  # New section
            if current_content:
                sections[current_section] = '\n'.join(current_content)
            current_section = line.strip('# ').lower().replace(' ', '_')
            current_content = []
        else:
            current_content.append(line)
    
    # Add last section
    if current_content:
        sections[current_section] = '\n'.join(current_content)
    
    return sections

def create_summary_report(results: List[Dict]):
    """Create executive summary dari bulk results"""
    total = len(results)
    
    summary = f"""
# Sales Automation Summary Report
Generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}

## Overview
- Total Leads Processed: {total}
- Success Rate: {(total / total * 100):.1f}%

## Results by Industry
{generate_industry_breakdown(results)}

## Next Steps
{generate_next_steps(results)}
"""
    
    with open("output/executive_summary.md", "w") as f:
        f.write(summary)

def generate_industry_breakdown(results):
    """Generate breakdown by industry"""
    # Implement industry analysis
    return "- SaaS: 5 leads\n- Fintech: 3 leads\n- E-commerce: 2 leads"

def generate_next_steps(results):
    """Generate recommended next steps"""
    return "1. Follow up on 3 hot leads immediately\n2. Schedule demos for next week\n3. Add 5 warm leads to nurture sequence"

def main():
    """Main entry point dengan menu options"""
    print("""
     Sales AI Automation Tool
    ==========================
    
    Choose operation mode:
    1. Single Lead Processing
    2. Bulk Pipeline Processing  
    3. Custom Workflow
    4. Exit
    """)
    
    choice = input("Enter your choice (1-4): ")
    
    if choice == "1":
        run_single_lead_automation()
    elif choice == "2":
        run_bulk_pipeline()
    elif choice == "3":
        run_custom_workflow()
    elif choice == "4":
        print(" Goodbye!")
        sys.exit(0)
    else:
        print("❌ Invalid choice. Please try again.")
        main()

if __name__ == "__main__":
    # Check environment variables
    required_env_vars = ["OPENAI_API_KEY"]
    missing_vars = [var for var in required_env_vars if not os.getenv(var)]
    
    if missing_vars:
        print(f"❌ Missing environment variables: {', '.join(missing_vars)}")
        print("Please set them in your .env file")
        sys.exit(1)
    
    # Run main program
    main()

Testing dan Debugging Sales Automation

Testing adalah krusial untuk strategi sales AI multi-agent. Berikut cara comprehensive testing:

# src/sales_automation/tests/test_sales_crew.py
import pytest
from unittest.mock import Mock, patch
from sales_automation.crew import SalesAutomationCrew

class TestSalesAutomation:
    
    @pytest.fixture
    def crew(self):
        """Create crew instance untuk testing"""
        return SalesAutomationCrew()
    
    @pytest.fixture
    def sample_inputs(self):
        """Sample input data untuk testing"""
        return {
            'company_name': 'Test Company',
            'contact_name': 'Test Contact',
            'industry': 'SaaS',
            'website': 'https://test.com'
        }
    
    def test_agent_creation(self, crew):
        """Test semua agents tercreate dengan benar"""
        # Test lead researcher
        researcher = crew.lead_researcher()
        assert researcher.role == "Senior Sales Intelligence Analyst"
        assert len(researcher.tools) > 0
        
        # Test other agents
        assert crew.lead_scorer() is not None
        assert crew.email_personalizer() is not None
        assert crew.follow_up_strategist() is not None
    
    def test_task_configuration(self, crew):
        """Test tasks configured correctly"""
        research_task = crew.research_lead()
        assert research_task is not None
        assert research_task.output_file == 'output/research_report.md'
    
    @patch('sales_automation.crew.SerperDevTool')
    def test_web_search_integration(self, mock_serper, crew):
        """Test web search tool integration"""
        # Mock search results
        mock_serper.return_value._run.return_value = "Mocked search results"
        
        # Create researcher dengan mocked tool
        researcher = crew.lead_researcher()
        
        # Verify tool added
        assert any(isinstance(tool, type(mock_serper.return_value)) 
                  for tool in researcher.tools)
    
    def test_crew_execution_flow(self, crew, sample_inputs):
        """Test full crew execution flow"""
        with patch.object(crew.crew(), 'kickoff') as mock_kickoff:
            # Mock successful execution
            mock_result = Mock()
            mock_result.raw = "Test execution successful"
            mock_kickoff.return_value = mock_result
            
            # Execute
            result = crew.crew().kickoff(inputs=sample_inputs)
            
            # Verify
            assert result.raw == "Test execution successful"
            mock_kickoff.assert_called_once_with(inputs=sample_inputs)
    
    def test_error_handling(self, crew):
        """Test error handling mechanisms"""
        # Test missing required fields
        with pytest.raises(ValueError):
            crew.prepare_data({})  # Missing required fields
        
        # Test dengan partial data
        partial_data = {'company_name': 'Test'}
        with pytest.raises(ValueError):
            crew.prepare_data(partial_data)
    
    def test_score_extraction(self, crew):
        """Test score extraction logic"""
        test_cases = [
            ("Overall score: 85", 85),
            ("The overall score is 72 points", 72),
            ("Score: 91/100", 91),
            ("No score found", 0)
        ]
        
        for text, expected in test_cases:
            assert crew.extract_score(text) == expected

# Integration test
def test_integration_with_real_api():
    """
    Integration test dengan real API calls
    WARNING: Ini akan consume API credits!
    """
    import os
    if os.getenv("RUN_INTEGRATION_TESTS") != "true":
        pytest.skip("Skipping integration tests")
    
    crew = SalesAutomationCrew()
    inputs = {
        'company_name': 'Microsoft',
        'contact_name': 'Satya Nadella',
        'industry': 'Technology',
        'website': 'https://microsoft.com'
    }
    
    result = crew.crew().kickoff(inputs=inputs)
    
    # Verify hasil
    assert result is not None
    assert len(result.raw) > 100  # Should have substantial content
    assert 'Microsoft' in result.raw

Best Practices dan Optimization Tips

1. Memory Management

CrewAI mendukung berbagai tipe memory. Gunakan dengan bijak:

from crewai.memory import ShortTermMemory, LongTermMemory, EntityMemory

# Configure memory untuk crew
crew = Crew(
    agents=agents,
    tasks=tasks,
    memory=True,  # Enable default memory
    memory_config={
        "memory_system": ShortTermMemory(),  # Untuk current session
        "long_term_memory": LongTermMemory(),  # Untuk historical data
        "entity_memory": EntityMemory()  # Untuk track specific entities
    }
)

2. Rate Limiting dan Cost Control

Control API usage untuk manage costs:

# Di crew configuration
crew = Crew(
    agents=agents,
    tasks=tasks,
    max_rpm=10,  # Max requests per minute
    max_execution_time=1800,  # 30 minutes max
)

# Per agent control
agent = Agent(
    role="Researcher",
    llm="gpt-3.5-turbo",  # Use cheaper model when possible
    max_iter=3,  # Limit iterations
    max_execution_time=300  # 5 minutes per task
)

3. Caching untuk Efficiency

Enable caching untuk avoid redundant API calls:

# Enable caching di crew level
crew = Crew(
    agents=agents,
    tasks=tasks,
    cache=True,
    cache_handler={
        "type": "redis",  # Or "local"
        "config": {
            "host": "localhost",
            "port": 6379,
            "ttl": 3600  # 1 hour TTL
        }
    }
)

4. Monitoring dan Logging

Implement comprehensive logging:

import logging
from datetime import datetime

# Setup logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler(f'logs/sales_ai_{datetime.now().strftime("%Y%m%d")}.log'),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger(__name__)

# Log di critical points
@before_kickoff
def log_start(self, inputs):
    logger.info(f"Starting sales automation for {inputs.get('company_name')}")
    logger.debug(f"Full inputs: {inputs}")
    return inputs

@after_kickoff
def log_completion(self, result):
    logger.info("Sales automation completed")
    logger.debug(f"Result length: {len(result.raw)}")
    return result

Production Deployment

Untuk deploy ke production, pertimbangkan:

1. Environment Configuration

# config/settings.py
import os
from dotenv import load_dotenv

load_dotenv()

class Config:
    # API Keys
    OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
    SERPER_API_KEY = os.getenv("SERPER_API_KEY")
    
    # Model Configuration
    DEFAULT_LLM = os.getenv("DEFAULT_LLM", "gpt-4")
    FALLBACK_LLM = os.getenv("FALLBACK_LLM", "gpt-3.5-turbo")
    
    # Rate Limits
    MAX_RPM = int(os.getenv("MAX_RPM", "10"))
    MAX_CONCURRENT_CREWS = int(os.getenv("MAX_CONCURRENT_CREWS", "5"))
    
    # Storage
    OUTPUT_DIR = os.getenv("OUTPUT_DIR", "output")
    LOG_DIR = os.getenv("LOG_DIR", "logs")
    
    # Feature Flags
    ENABLE_CACHING = os.getenv("ENABLE_CACHING", "true").lower() == "true"
    ENABLE_MEMORY = os.getenv("ENABLE_MEMORY", "true").lower() == "true"

2. API Endpoint untuk Integration

# api/sales_automation_api.py
from fastapi import FastAPI, BackgroundTasks, HTTPException
from pydantic import BaseModel
from typing import Optional
import uuid

app = FastAPI(title="Sales AI Automation API")

class LeadRequest(BaseModel):
    company_name: str
    contact_name: str
    industry: str
    website: Optional[str] = None
    priority: Optional[str] = "normal"

class JobStatus(BaseModel):
    job_id: str
    status: str
    result: Optional[str] = None

# In-memory job store (use Redis in production)
jobs = {}

@app.post("/process-lead")
async def process_lead(
    lead: LeadRequest, 
    background_tasks: BackgroundTasks
):
    """Process single lead asynchronously"""
    job_id = str(uuid.uuid4())
    
    # Add to background tasks
    background_tasks.add_task(
        run_crew_async,
        job_id,
        lead.dict()
    )
    
    jobs[job_id] = {"status": "processing", "result": None}
    
    return {"job_id": job_id, "message": "Lead processing started"}

@app.get("/job/{job_id}")
async def get_job_status(job_id: str):
    """Check job status"""
    if job_id not in jobs:
        raise HTTPException(status_code=404, detail="Job not found")
    
    return JobStatus(
        job_id=job_id,
        status=jobs[job_id]["status"],
        result=jobs[job_id]["result"]
    )

async def run_crew_async(job_id: str, lead_data: dict):
    """Run crew in background"""
    try:
        from sales_automation.crew import SalesAutomationCrew
        
        crew = SalesAutomationCrew()
        result = crew.crew().kickoff(inputs=lead_data)
        
        jobs[job_id] = {
            "status": "completed",
            "result": result.raw[:500]  # First 500 chars
        }
    except Exception as e:
        jobs[job_id] = {
            "status": "failed",
            "result": str(e)
        }

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Langkah Selanjutnya

Photo by Kevin Ku on Unsplash

Membangun tim sales AI dengan CrewAI bukan hanya tentang menulis kode, tapi tentang menciptakan sistem yang benar-benar memahami dan mengotomasi proses sales Anda. Dengan implementasi yang tepat, Anda bisa meningkatkan efisiensi sales hingga 5.76x lebih cepat dibandingkan framework lain.

Framework ini terus berkembang dengan komunitas yang aktif. CrewAI Enterprise menyediakan fitur advanced seperti unified control plane, real-time observability, dan 24/7 support untuk kebutuhan enterprise. Mulailah dengan implementasi sederhana, lakukan iterasi berdasarkan feedback, dan scale sesuai kebutuhan bisnis Anda.

Ingat, kesuksesan implementasi AI sales agent bukan hanya tentang teknologi, tapi juga tentang bagaimana Anda mengintegrasikannya dengan proses bisnis existing dan terus mengoptimalkannya berdasarkan data real. Dengan CrewAI, masa depan sales automation ada di tangan Anda – dan kode yang saya bagikan di atas adalah starting point yang solid untuk memulai journey tersebut.