AI Agents · 강의

간단한 이메일 도우미 에이전트 만들기

처음부터 끝까지 진행합니다. 받은 편지함 읽기 → 요약 → 답장 초안 작성 → 승인 대기

레슨 4/413개 단계

간단한 이메일 도우미 에이전트 만들기은(는) CoddyKit의 무료 AI Agents 강의입니다. 이것은 4개 중 4번째 강의입니다. 아래에서 전체 강의를 무료로 읽을 수 있으며, 내장 코드 에디터와 24/7 AI 튜터와 함께 브라우저에서 직접 실습할 수 있습니다. 이 강의는 AI Agents 학습 경로의 일부이며, 진행 상황이 웹과 CoddyKit 앱에 동기화됩니다. AI Agents 강의에는 총 4개의 강의가 포함되어 있습니다.

이메일 어시스턴트 에이전트 아키텍처

이메일 어시스턴트 에이전트는 일반적인 처리 흐름을 따릅니다: 가져오기 → 분류 → 결정 → 초안 작성 → 승인 → 전송. 에이전트는 읽지 않은 이메일을 읽고, LLM을 사용하여 답변을 분류하고 작성한 다음, 전송하기 전에 사람의 승인을 기다립니다. 이러한 사람 개입 방식의 설계는 완전히 자율적인 이메일 전송으로 인해 발생하는 큰 실수를 방지합니다.

# 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')

에이전트 도구 정의

명확한 도구 정의로 에이전트를 구성하세요. 각 도구는 특정 책임을 맡은 Python 함수입니다. 이렇게 모듈화하면 에이전트를 테스트하고 디버깅하기 쉬우며, 새로운 기능으로 확장하기도 쉽습니다.

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')

1단계: 이메일 가져오기 및 파싱

첫 번째 도구는 읽지 않은 이메일을 가져오고 LLM에 필요한 핵심 필드인 제목, 보낸 사람, 일반 텍스트 본문을 추출합니다. 본문은 짧게 유지하세요. LLM 컨텍스트 제한을 지키고 API 비용을 줄이려면 처음 2000자로 잘라내세요.

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

2단계: LLM 분류

각 이메일을 분류하도록 LLM에 보냅니다. 모델이 카테고리와 추론을 포함한 JSON 개체를 반환하도록 요청하는 구조화된 프롬프트를 사용하세요. JSON 출력을 명시적으로 요청하는 편이 자유 형식 텍스트를 파싱하는 것보다 안정적입니다.

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'

3단계: 답장 초안 생성

action_needed로 분류된 이메일에는 LLM에 답장 초안을 생성하도록 요청하세요. 에이전트의 역할과 어조에 대한 컨텍스트를 제공하세요. 전문적이면서도 간결한 답장을 요청하고, 사람이 입력해야 하는 내용에는 자리 표시자를 포함하도록 하세요.

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()

4단계: 사람 검토를 위한 초안 저장

AI가 생성한 이메일을 사람의 검토 없이 절대 전송하지 마세요. 이메일을 Gmail 초안으로 저장하면 사람이 Gmail을 열어 검토하고, 필요한 경우 편집한 뒤 전송할 수 있습니다. 초안 생성 API는 전송 API와 동일하며, 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.')

5단계: 처리된 이메일 표시

이메일을 처리한 후(분류하고 초안을 만들었거나 보관한 후) 다시 처리하지 않도록 표시하세요. AgentProcessed와 같은 사용자 지정 라벨을 추가하고 UNREAD 라벨을 제거하세요. 라벨이 없으면 한 번만 만드세요.

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)

전체 에이전트 실행 루프

모든 단계를 하나의 run() 메서드로 연결하세요. 에이전트는 이메일을 차례로 처리하고, 조치가 필요한 항목의 초안을 만들고, 스팸을 보관하며, 모두 처리된 것으로 표시합니다. 마지막에는 요약을 기록하세요.

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

오류 처리 및 복원력

각 이메일 처리 단계를 try/except로 감싸세요. 한 이메일의 실패가 나머지 이메일을 처리하는 에이전트를 중단시켜서는 안 됩니다. 나중에 검토할 수 있도록 이메일 ID와 제목을 함께 오류로 기록하세요.

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'

에이전트 주기적 실행 예약

간단한 루프에서 실행 사이에 time.sleep()을 사용하여 일정에 따라 이메일 에이전트를 실행하세요. 운영 환경에서는 cron, APScheduler 또는 클라우드 함수를 이용하는 작업 스케줄러를 사용하세요. 수신자의 시간대를 고려하여 업무 외 시간에 실행되지 않도록 항상 확인 절차를 추가하세요.

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)

로그 기록 및 감사 추적

에이전트가 처리하는 모든 이메일의 JSON 로그를 유지하세요. 이렇게 하면 에이전트의 동작을 확인하고, 잘못된 분류를 진단하며, 시간이 지남에 따라 프롬프트를 개선할 수 있는 감사 추적 기록이 만들어집니다.

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'))

빠른 확인: 사람 개입 방식

이메일 어시스턴트 설계에 대한 이해도를 확인해 보세요.

인터랙티브 레슨에서 답해보세요

이메일 어시스턴트 에이전트 복습

완전한 이메일 어시스턴트 에이전트 처리 흐름을 만들었습니다.

가져오기: Gmail API 검색 쿼리로 읽지 않은 이메일 가져오기
분류: LLM이 각 이메일을 action_needed/fyi/spam으로 분류하고 JSON으로 출력
초안 작성: LLM이 전문적인 어조의 답장을 생성하고 [FILL IN] 자리 표시자 포함
초안 저장: 사람이 검토할 수 있도록 Gmail에 저장 — 자동 전송은 절대 하지 않음
처리 완료 표시: 다시 처리하지 않도록 사용자 지정 라벨 추가 및 UNREAD 제거
오류 격리: 한 번의 실패가 일괄 처리를 중단하지 않도록 각 이메일을 try/except로 감싸기
감사 로그: 검토 및 개선을 위해 모든 작업을 기록하는 JSONL 파일

무료로 시작

AI 튜터와 함께 AI Agents을(를) 배우세요 — 무료

브라우저에서 실제 코드를 작성하고 실행하며, 24/7 AI 튜터로부터 즉각적인 도움을 받고, 웹이나 앱에서 중단한 부분부터 계속 학습하세요.

코스: 60
레슨: 239

무료로 학습 시작 전체 코스 보기

자주 묻는 질문

“간단한 이메일 도우미 에이전트 만들기” 강의는 무료인가요?

네 — “간단한 이메일 도우미 에이전트 만들기” 전체 내용을 이 웹사이트에서 무료로 읽을 수 있습니다. 인터랙티브하게 실습하려면(내장 코드 에디터와 24/7 AI 튜터), CoddyKit PRO로 업그레이드하면 AI Agents 강의 전체를 잠금 해제할 수 있습니다. AI Agents 강의에는 총 4개의 강의가 포함되어 있습니다.

“간단한 이메일 도우미 에이전트 만들기”에서 뭘 배우나요?

처음부터 끝까지 진행합니다. 받은 편지함 읽기 → 요약 → 답장 초안 작성 → 승인 대기 브라우저에서 직접 실행하는 실습 코드로 AI Agents을(를) 배우며, 24/7 AI 튜터가 강의를 진행하면서 질문에 답변해줍니다.

AI Agents을(를) 시작하는 데 경험이 필요한가요?

사전 경험은 필요하지 않습니다. CoddyKit의 AI Agents은(는) 초급자부터 고급 학습자까지를 위해 구성되어 있으므로, 여기서 시작하거나 처음부터 시작할 수 있으며 자신의 속도대로 진행할 수 있습니다. 이것은 4개 중 4번째 강의입니다.

“간단한 이메일 도우미 에이전트 만들기” 강의는 얼마나 걸리나요?

대부분의 CoddyKit 강의는 약 5~10분이 소요됩니다. 각 강의는 간결하고 인터랙티브하여 꾸준한 진행이 가능하며, 웹과 앱에서 중단한 부분부터 바로 시작할 수 있습니다.

이 AI Agents 강의에서 코드를 작성하고 실행할 수 있나요?

네. 모든 AI Agents 강의에는 내장 코드 에디터가 포함되어 있으므로, 브라우저에서 바로 실제 코드를 작성하고 실행한 후 즉시 AI 피드백을 받을 수 있습니다 — 로컬 설정이 필요 없습니다.

이 강의의 모든 강의

← AI Agents(으)로 돌아가기