AI Agents · Pelajaran

Membangun Agen Asisten Email Sederhana

Dari awal hingga akhir: baca kotak masuk → rangkum → buat draf balasan → tunggu persetujuan.

Pelajaran 4 dari 413 langkah

Membangun Agen Asisten Email Sederhana adalah pelajaran AI Agents gratis di CoddyKit. Ini adalah pelajaran 4 dari 4. Kamu bisa membaca pelajaran lengkapnya di bawah secara gratis — lalu praktikkan langsung di browser dengan editor kode bawaan dan tutor AI 24/7. Ini adalah bagian dari jalur belajar AI Agents, dan progresmu tersinkronisasi di web dan aplikasi CoddyKit. Kursus AI Agents mencakup 4 pelajaran total.

Arsitektur Agen Asisten Email

Agen asisten email mengikuti alur kerja standar: Ambil → Klasifikasikan → Putuskan → Buat Draf → Setujui → Kirim. Agen membaca email yang belum dibaca, menggunakan LLM untuk mengklasifikasikan dan membuat draf balasan, lalu menunggu persetujuan manusia sebelum mengirimkannya. Desain dengan manusia dalam proses ini mencegah kesalahan besar akibat pengiriman email yang sepenuhnya otonom.

# Email Assistant Pipeline:
#
# 1. FETCH: Pull unread emails from Gmail API
# 2. CLASSIFY: LLM labels each email
#    - 'action_needed': requires a reply
#    - 'fyi': informational, no reply needed
#    - 'spam': should be archived
# 3. DRAFT: LLM generates reply for 'action_needed' emails
# 4. APPROVE: Human reviews drafts in Gmail UI
# 5. SEND: Agent sends approved drafts
#
# Tools: Gmail API, Anthropic/OpenAI API, json, base64

print('Email assistant pipeline: fetch -> classify -> draft -> approve -> send')

Mendefinisikan Alat Agen

Susun agen dengan definisi alat yang jelas. Setiap alat merupakan fungsi Python dengan tanggung jawab tertentu. Modularitas ini membuat agen mudah diuji, didiagnosis, dan diperluas dengan kemampuan baru.

class EmailAssistantTools:
    def __init__(self, gmail_service, llm_client):
        self.gmail = gmail_service
        self.llm = llm_client

    def fetch_unread(self, max_emails=10):
        '''Fetch unread emails from inbox.'''
        pass

    def classify_email(self, subject, body, sender):
        '''Ask LLM to classify: action_needed / fyi / spam.'''
        pass

    def draft_reply(self, subject, body, sender, context):
        '''Ask LLM to draft a reply to an email.'''
        pass

    def create_draft(self, message_id, reply_text):
        '''Save draft reply in Gmail for human review.'''
        pass

    def send_approved_drafts(self):
        '''Send all drafts marked as approved.'''
        pass

print('Tool-based architecture enables testing each step independently')

Langkah 1: Mengambil dan Mengurai Email

Alat pertama mengambil email yang belum dibaca dan mengekstrak bidang penting yang diperlukan LLM: subjek, pengirim, dan isi teks biasa. Buat isi email tetap singkat—potong hingga 2000 karakter pertama agar tetap berada dalam batas konteks LLM dan mengurangi biaya API.

import base64

def fetch_emails_for_classification(gmail_service, max_emails=10):
    messages_list = gmail_service.users().messages().list(
        userId='me',
        q='is:unread label:inbox',
        maxResults=max_emails
    ).execute().get('messages', [])

    emails = []
    for ref in messages_list:
        msg = gmail_service.users().messages().get(
            userId='me', id=ref['id'], format='full'
        ).execute()

        headers = {h['name'].lower(): h['value']
                   for h in msg['payload'].get('headers', [])}
        body = extract_plain_text_body(msg)[:2000]  # trim for LLM

        emails.append({
            'id': msg['id'],
            'thread_id': msg['threadId'],
            'from': headers.get('from', ''),
            'subject': headers.get('subject', '(no subject)'),
            'body': body
        })

    print(f'Fetched {len(emails)} unread emails')
    return emails

Langkah 2: Klasifikasi oleh LLM

Kirim setiap email ke LLM untuk diklasifikasikan. Gunakan prompt terstruktur yang meminta model mengembalikan objek JSON berisi kategori dan alasan. Minta keluaran JSON secara eksplisit—hasilnya lebih andal daripada mengurai teks bebas.

import json
import anthropic
import os

client = anthropic.Anthropic(api_key=os.environ['ANTHROPIC_API_KEY'])

def classify_email(subject, sender, body):
    prompt = (
        'Classify this email as exactly one of: action_needed, fyi, spam.\n'
        'Return JSON only: {"category": "...", "reason": "..."}\n\n'
        f'From: {sender}\n'
        f'Subject: {subject}\n\n'
        f'Body:\n{body[:1500]}'
    )

    response = client.messages.create(
        model='claude-opus-4-5',
        max_tokens=200,
        messages=[{'role': 'user', 'content': prompt}]
    )

    try:
        result = json.loads(response.content[0].text)
        return result.get('category', 'fyi'), result.get('reason', '')
    except json.JSONDecodeError:
        return 'fyi', 'Could not parse LLM response'

Langkah 3: Membuat Draf Balasan

Untuk email yang diklasifikasikan sebagai action_needed, minta LLM membuat draf balasan. Berikan konteks tentang peran dan nada agen. Minta balasan yang profesional tetapi ringkas—sertakan penanda tempat untuk apa pun yang perlu diisi oleh manusia.

def draft_reply(subject, sender, body, agent_context):
    prompt = (
        'You are an email assistant drafting a professional reply.\n'
        'Guidelines:\n'
        '- Be concise and professional\n'
        '- Use [FILL IN] for any info you don\'t know\n'
        '- Start with a greeting, end with a sign-off\n\n'
        f'Context about the recipient\'s work: {agent_context}\n\n'
        f'Original email from {sender}:\n'
        f'Subject: {subject}\n\n'
        f'{body[:1500]}\n\n'
        'Draft a reply:'
    )

    response = client.messages.create(
        model='claude-opus-4-5',
        max_tokens=500,
        messages=[{'role': 'user', 'content': prompt}]
    )

    return response.content[0].text.strip()

Langkah 4: Menyimpan Draf untuk Ditinjau Manusia

Jangan pernah mengirim email yang dibuat AI tanpa peninjauan manusia. Simpan email tersebut sebagai draf Gmail agar manusia dapat membuka Gmail, meninjau, mengedit jika diperlukan, lalu mengirimkannya. API pembuatan draf sama dengan API pengiriman, tetapi menggunakan drafts().create().

import base64
from email.mime.text import MIMEText

def save_draft_reply(gmail_service, original_message, reply_text):
    sender_header = next(
        (h['value'] for h in original_message['payload'].get('headers', [])
         if h['name'].lower() == 'from'), ''
    )
    subject = next(
        (h['value'] for h in original_message['payload'].get('headers', [])
         if h['name'].lower() == 'subject'), ''
    )
    msg_id_header = next(
        (h['value'] for h in original_message['payload'].get('headers', [])
         if h['name'].lower() == 'message-id'), ''
    )

    mime_msg = MIMEText(reply_text, 'plain', 'utf-8')
    mime_msg['To'] = sender_header
    mime_msg['Subject'] = 'Re: ' + subject
    mime_msg['In-Reply-To'] = msg_id_header
    mime_msg['References'] = msg_id_header

    raw = base64.urlsafe_b64encode(mime_msg.as_bytes()).decode()
    draft = gmail_service.users().drafts().create(
        userId='me',
        body={'message': {'raw': raw, 'threadId': original_message['threadId']}}
    ).execute()

    print(f'Draft saved: {draft["id"]}')
    return draft['id']

# --- demo: minimal stand-in for the Gmail API's service object ---
class _Exec:
    def __init__(self, result):
        self._result = result
    def execute(self):
        return self._result

class _FakeDrafts:
    def create(self, userId, body):
        return _Exec({'id': 'r9000abc'})

class _FakeUsers:
    def drafts(self):
        return _FakeDrafts()

class _FakeGmailService:
    def users(self):
        return _FakeUsers()

original_message = {
    'threadId': 'thread_1',
    'payload': {'headers': [
        {'name': 'From', 'value': 'customer@example.com'},
        {'name': 'Subject', 'value': 'Question about my order'},
        {'name': 'Message-ID', 'value': '<abc123@mail.example.com>'}
    ]}
}
save_draft_reply(_FakeGmailService(), original_message, 'Thanks for reaching out, we will look into it.')

Langkah 5: Menandai Email yang Telah Diproses

Setelah memproses email (diklasifikasikan dan draf dibuat atau diarsipkan), tandai email tersebut untuk mencegah pemrosesan ulang. Tambahkan label khusus seperti AgentProcessed dan hapus label UNREAD. Buat label tersebut sekali jika belum ada.

def get_or_create_label(gmail_service, label_name):
    labels = gmail_service.users().labels().list(userId='me').execute()
    for label in labels.get('labels', []):
        if label['name'] == label_name:
            return label['id']

    # Create the label
    new_label = gmail_service.users().labels().create(
        userId='me',
        body={
            'name': label_name,
            'labelListVisibility': 'labelShow',
            'messageListVisibility': 'show'
        }
    ).execute()
    print(f'Created label: {label_name}')
    return new_label['id']

def mark_processed(gmail_service, message_id, agent_label_id):
    gmail_service.users().messages().modify(
        userId='me',
        id=message_id,
        body={
            'addLabelIds': [agent_label_id],
            'removeLabelIds': ['UNREAD']
        }
    ).execute()

# --- demo: minimal stand-in for the Gmail API's service object ---
class _Exec:
    def __init__(self, result):
        self._result = result
    def execute(self):
        return self._result

class _FakeUsers:
    def __init__(self):
        self._labels = [{'id': 'Label_1', 'name': 'Processed'}]
    def labels(self):
        return self
    def list(self, userId):
        return _Exec({'labels': self._labels})
    def create(self, userId, body):
        print(f'Created label: {body["name"]}')
        return _Exec({'id': 'Label_2', 'name': body['name']})
    def messages(self):
        return self
    def modify(self, userId, id, body):
        print(f'Marked {id} processed with {body}')
        return _Exec({'id': id})

class _FakeGmailService:
    def users(self):
        return _FakeUsers()

gmail_service = _FakeGmailService()
label_id = get_or_create_label(gmail_service, 'Processed')
print(f'Label id: {label_id}')
new_label_id = get_or_create_label(gmail_service, 'AgentHandled')
mark_processed(gmail_service, 'msg_99', new_label_id)

Siklus Eksekusi Lengkap Agen

Hubungkan semua langkah ke dalam satu metode run(). Agen mengulang setiap email, mengklasifikasikan masing-masing email, membuat draf untuk item yang memerlukan tindakan, mengarsipkan spam, dan menandai semuanya sebagai telah diproses. Catat rangkuman di akhir.

def run_email_agent(gmail_service, agent_context, max_emails=10):
    processed_label = get_or_create_label(gmail_service, 'AgentProcessed')
    emails = fetch_emails_for_classification(gmail_service, max_emails)

    summary = {'action_needed': 0, 'fyi': 0, 'spam': 0, 'drafts_created': 0}

    for email in emails:
        category, reason = classify_email(
            email['subject'], email['from'], email['body']
        )
        summary[category] += 1
        print(f'[{category}] {email["subject"][:60]} - {reason[:50]}')

        if category == 'action_needed':
            # Fetch full message for reply context
            full_msg = gmail_service.users().messages().get(
                userId='me', id=email['id'], format='full'
            ).execute()
            reply = draft_reply(
                email['subject'], email['from'],
                email['body'], agent_context
            )
            save_draft_reply(gmail_service, full_msg, reply)
            summary['drafts_created'] += 1

        mark_processed(gmail_service, email['id'], processed_label)

    print('\nAgent run complete:')
    for k, v in summary.items():
        print(f'  {k}: {v}')
    return summary

Penanganan Kesalahan dan Ketangguhan

Bungkus setiap langkah pemrosesan email dalam try/except. Kegagalan pada satu email tidak boleh menghentikan agen memproses email lainnya. Catat kesalahan beserta ID dan subjek email agar dapat ditinjau nanti.

from googleapiclient.errors import HttpError

def process_email_safely(gmail_service, email, agent_context, label_id):
    try:
        category, reason = classify_email(
            email['subject'], email['from'], email['body']
        )

        if category == 'action_needed':
            full_msg = gmail_service.users().messages().get(
                userId='me', id=email['id'], format='full'
            ).execute()
            reply = draft_reply(
                email['subject'], email['from'],
                email['body'], agent_context
            )
            save_draft_reply(gmail_service, full_msg, reply)

        mark_processed(gmail_service, email['id'], label_id)
        return category

    except HttpError as e:
        print(f'Gmail API error on {email["id"]}: {e.resp.status}')
        return 'error'
    except Exception as e:
        print(f'Unexpected error on {email["subject"][:50]}: {e}')
        return 'error'

Menjadwalkan Agen agar Berjalan Secara Berkala

Jalankan agen email sesuai jadwal menggunakan perulangan sederhana dengan time.sleep() di antara setiap proses. Untuk produksi, gunakan penjadwal tugas seperti cron, APScheduler, atau fungsi cloud. Selalu tambahkan pemeriksaan agar agen tidak berjalan di luar jam kerja dan menghormati zona waktu penerima.

import time
import datetime

def is_business_hours():
    now = datetime.datetime.now()
    # Mon-Fri, 9am-6pm local time
    return (now.weekday() < 5 and 9 <= now.hour < 18)

def run_scheduler(gmail_service, agent_context,
                  interval_minutes=30, max_emails=20):
    print(f'Email agent started. Checking every {interval_minutes} min.')

    while True:
        if is_business_hours():
            print(f'\n[{datetime.datetime.now().strftime("%H:%M")}] Running agent...')
            try:
                run_email_agent(gmail_service, agent_context, max_emails)
            except Exception as e:
                print(f'Agent run failed: {e}')
        else:
            print('Outside business hours, skipping run')

        time.sleep(interval_minutes * 60)

Pencatatan dan Jejak Audit

Simpan log JSON untuk setiap email yang diproses agen. Dengan demikian, tercipta jejak audit yang dapat Anda tinjau untuk memverifikasi perilaku agen, mendiagnosis kesalahan klasifikasi, dan menyempurnakan prompt seiring waktu.

import json
import datetime
from pathlib import Path

LOG_FILE = Path('agent_audit.jsonl')

def log_action(email_id, subject, sender, category, action, draft_id=None):
    entry = {
        'timestamp': datetime.datetime.now().isoformat(),
        'email_id': email_id,
        'subject': subject[:100],
        'from': sender,
        'category': category,
        'action': action,
        'draft_id': draft_id
    }
    with open(LOG_FILE, 'a', encoding='utf-8') as f:
        f.write(json.dumps(entry) + '\n')

# Usage in the main loop:
log_action(
    email_id='18abc123',
    subject='Invoice #1234',
    sender='billing@vendor.com',
    category='action_needed',
    action='draft_created',
    draft_id='r9000abc'
)

# --- demo ---
print('Audit log contents:')
print(LOG_FILE.read_text(encoding='utf-8'))

Cek Singkat: Manusia dalam Proses

Uji pemahaman Anda tentang desain asisten email.

Jawab di pelajaran interaktif

Rangkuman Agen Asisten Email

Anda telah membuat alur kerja agen asisten email yang lengkap:

Mengambil: mengambil email yang belum dibaca dengan kueri penelusuran Gmail API
Mengklasifikasikan: LLM mengategorikan setiap email sebagai action_needed/fyi/spam dengan keluaran JSON
Membuat draf: LLM membuat balasan dengan nada profesional dan penanda tempat [FILL IN]
Menyimpan draf: menyimpannya di Gmail untuk ditinjau manusia—jangan pernah mengirim otomatis
Menandai sebagai telah diproses: menambahkan label khusus dan menghapus UNREAD untuk mencegah pemrosesan ulang
Mengisolasi kesalahan: membungkus setiap email dalam try/except agar satu kegagalan tidak menghentikan kumpulan proses
Log audit: berkas JSONL yang mencatat setiap tindakan untuk ditinjau dan disempurnakan

Gratis untuk memulai

Belajar AI Agents dengan tutor AI — gratis

Tulis dan jalankan kode asli di browser kamu, dapatkan bantuan instan dari tutor AI 24/7, dan lanjutkan di mana kamu tinggalkan di web atau aplikasi.

Kursus: 60
Pelajaran: 239

Mulai belajar gratis Lihat kursus lengkapnya

Pertanyaan yang Sering Diajukan

Apakah pelajaran “Membangun Agen Asisten Email Sederhana” gratis?

Ya — teks lengkap “Membangun Agen Asisten Email Sederhana” gratis dibaca di sini di web. Untuk praktiknya secara interaktif (editor kode bawaan dan tutor AI 24/7) dan buka sisa kursus AI Agents, upgrade ke CoddyKit PRO. Kursus AI Agents mencakup 4 pelajaran total.

Apa yang akan aku pelajari di “Membangun Agen Asisten Email Sederhana”?

Dari awal hingga akhir: baca kotak masuk → rangkum → buat draf balasan → tunggu persetujuan. Kamu berlatih AI Agents dengan kode praktik yang langsung kamu jalankan di browser, dan tutor AI 24/7 menjawab pertanyaanmu saat kamu mengerjakan pelajaran ini.

Apakah aku perlu pengalaman untuk memulai AI Agents?

Tidak diperlukan pengalaman sebelumnya. AI Agents di CoddyKit dirancang untuk pemula hingga pelajar tingkat lanjut, jadi kamu bisa memulai di sini atau dari awal dan belajar sesuai kecepatan kamu sendiri. Ini adalah pelajaran 4 dari 4.

Berapa lama pelajaran “Membangun Agen Asisten Email Sederhana” memakan waktu?

Sebagian besar pelajaran CoddyKit memakan waktu sekitar 5–10 menit. Setiap pelajaran ringkas dan interaktif, jadi kamu membuat kemajuan stabil dan melanjutkan dari tempat kamu tinggalkan di web dan aplikasi.

Bisakah aku menulis dan menjalankan kode dalam pelajaran AI Agents ini?

Ya. Setiap pelajaran AI Agents menyertakan editor kode bawaan, jadi kamu menulis dan menjalankan kode nyata langsung di browser dan mendapatkan umpan balik AI instan — tidak diperlukan penyiapan lokal.

Semua pelajaran dalam kursus ini

Menghubungkan ke Gmail melalui API
Membaca dan Mengirim Email Secara Terprogram
Membuat dan Menanyakan Acara Kalender
Membangun Agen Asisten Email Sederhana

← Kembali ke AI Agents