Tworzenie prostego agenta asystenta e-mail
Od początku do końca: odczytanie skrzynki → podsumowanie → przygotowanie odpowiedzi → oczekiwanie na zatwierdzenie.
Tworzenie prostego agenta asystenta e-mail to bezpłatna lekcja AI Agents na CoddyKit. To lekcja 4 z 4. Możesz przeczytać całą lekcję poniżej za darmo — a potem ćwiczyć ją interaktywnie w przeglądarce z wbudowanym edytorem kodu i tutorem AI dostępnym 24/7. To część ścieżki edukacyjnej AI Agents, a Twój postęp synchronizuje się między webem a aplikacją CoddyKit. Kurs AI Agents zawiera 4 lekcji w sumie.
Architektura agenta asystenta poczty e-mail
Agent asystenta poczty e-mail działa według standardowego potoku: Pobierz → Klasyfikuj → Podejmij decyzję → Przygotuj wersję roboczą → Zatwierdź → Wyślij. Agent odczytuje nieprzeczytane wiadomości, korzysta z LLM do ich klasyfikowania i przygotowywania odpowiedzi, a następnie czeka na zatwierdzenie przez człowieka przed wysłaniem. Taka konstrukcja z udziałem człowieka zapobiega kosztownym błędom, które mogłyby wystąpić przy całkowicie autonomicznym wysyłaniu wiadomości.
# 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')Definiowanie narzędzi agenta
Agenta należy zbudować z jasno zdefiniowanych narzędzi. Każde narzędzie jest funkcją Pythona o konkretnym zadaniu. Taka modułowa struktura ułatwia testowanie i debugowanie agenta oraz jego rozbudowę o nowe możliwości.
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')Krok 1: Pobieranie i analizowanie wiadomości e-mail
Pierwsze narzędzie pobiera nieprzeczytane wiadomości e-mail i wyodrębnia kluczowe pola potrzebne LLM: temat, nadawcę i treść w postaci zwykłego tekstu. Treść powinna być krótka — należy ograniczyć ją do pierwszych 2000 znaków, aby zmieścić się w limitach kontekstu LLM i zmniejszyć koszty 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 emailsKrok 2: Klasyfikacja przez LLM
Każdą wiadomość e-mail należy wysłać do LLM w celu jej sklasyfikowania. Należy użyć ustrukturyzowanego promptu, który poprosi model o zwrócenie obiektu JSON zawierającego kategorię i uzasadnienie. Należy wyraźnie zażądać wyniku w formacie JSON — jest to bardziej niezawodne niż analizowanie swobodnego tekstu.
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'Krok 3: Generowanie wersji roboczych odpowiedzi
W przypadku wiadomości sklasyfikowanych jako action_needed należy poprosić LLM o wygenerowanie wersji roboczej odpowiedzi. Należy przekazać kontekst dotyczący roli agenta i tonu wypowiedzi. Warto poprosić o profesjonalną, ale zwięzłą odpowiedź oraz umieścić symbol zastępczy w każdym miejscu, które człowiek musi uzupełnić.
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()Krok 4: Zapisywanie wersji roboczych do weryfikacji przez człowieka
Nie należy nigdy wysyłać wiadomości e-mail wygenerowanych przez AI bez weryfikacji przez człowieka. Należy zapisać je jako wersje robocze Gmaila, aby człowiek mógł otworzyć Gmaila, sprawdzić wiadomość, w razie potrzeby ją edytować i wysłać. API tworzenia wersji roboczej jest takie samo jak API wysyłania — należy użyć 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.')
Krok 5: Oznaczanie przetworzonych wiadomości
Po przetworzeniu wiadomości — sklasyfikowaniu i utworzeniu wersji roboczej lub zarchiwizowaniu — należy ją oznaczyć, aby zapobiec ponownemu przetwarzaniu. Należy dodać niestandardową etykietę, taką jak AgentProcessed, i usunąć etykietę UNREAD. Jeśli etykieta nie istnieje, należy utworzyć ją tylko raz.
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)
Pełna pętla działania agenta
Wszystkie kroki należy połączyć w jednej metodzie run(). Agent przechodzi przez wiadomości e-mail, klasyfikuje każdą z nich, tworzy wersje robocze dla wiadomości wymagających działania, archiwizuje spam i oznacza wszystkie wiadomości jako przetworzone. Na końcu należy zapisać podsumowanie w dzienniku.
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 summaryObsługa błędów i odporność
Każdy etap przetwarzania wiadomości e-mail należy objąć konstrukcją try/except. Błąd dotyczący jednej wiadomości nigdy nie powinien zatrzymać przetwarzania pozostałych. Błędy należy zapisywać wraz z identyfikatorem i tematem wiadomości, aby można było przejrzeć je później.
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'Okresowe uruchamianie agenta według harmonogramu
Agenta pocztowego można uruchamiać według harmonogramu za pomocą prostej pętli z wywołaniem time.sleep() pomiędzy kolejnymi uruchomieniami. W środowisku produkcyjnym należy użyć harmonogramu zadań, takiego jak cron, APScheduler lub funkcja chmurowa. Należy zawsze dodać sprawdzanie, które zapobiega uruchamianiu poza godzinami pracy, aby respektować strefy czasowe odbiorców.
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)Rejestrowanie i ślad audytowy
Należy prowadzić dziennik JSON każdej wiadomości e-mail przetwarzanej przez agenta. Tworzy to ślad audytowy, który można przejrzeć w celu weryfikacji działania agenta, diagnozowania błędnych klasyfikacji i stopniowego ulepszania promptów.
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'))
Szybki test: proces z udziałem człowieka
Proszę sprawdzić znajomość projektu asystenta poczty e-mail.
Podsumowanie agenta asystenta poczty e-mail
Utworzyli Państwo kompletny potok agenta asystenta poczty e-mail:
- Pobieranie: pobieranie nieprzeczytanych wiadomości za pomocą zapytań wyszukiwania Gmail API
- Klasyfikacja: LLM przypisuje każdej wiadomości kategorię action_needed/fyi/spam i zwraca wynik w formacie JSON
- Wersja robocza: LLM generuje odpowiedź w profesjonalnym tonie z symbolami zastępczymi [FILL IN]
- Zapisywanie wersji roboczej: przechowywanie w Gmailu do weryfikacji przez człowieka — nigdy nie należy wysyłać automatycznie
- Oznaczanie jako przetworzone: dodanie niestandardowej etykiety i usunięcie UNREAD, aby zapobiec ponownemu przetwarzaniu
- Izolowanie błędów: objęcie każdej wiadomości konstrukcją try/except, aby pojedynczy błąd nie zatrzymał całej partii
- Dziennik audytowy: plik JSONL rejestrujący każde działanie na potrzeby weryfikacji i ulepszania
Często zadawane pytania
Czy lekcja „Tworzenie prostego agenta asystenta e-mail” jest bezpłatna?
Tak — pełny tekst „Tworzenie prostego agenta asystenta e-mail” jest dostępny za darmo tutaj w sieci. Aby ćwiczyć ją interaktywnie (wbudowany edytor kodu i tutor AI dostępny 24/7) i odblokować resztę kursu AI Agents, przejdź na CoddyKit PRO. Kurs AI Agents zawiera 4 lekcji w sumie.
Co nauczysz się w „Tworzenie prostego agenta asystenta e-mail”?
Od początku do końca: odczytanie skrzynki → podsumowanie → przygotowanie odpowiedzi → oczekiwanie na zatwierdzenie. Ćwiczysz AI Agents z praktycznym kodem, który uruchamiasz bezpośrednio w przeglądarce, a tutor AI dostępny 24/7 odpowiada na Twoje pytania podczas pracy nad lekcją.
Czy potrzebuję doświadczenia, aby zacząć AI Agents?
Nie wymagamy żadnego doświadczenia. AI Agents w CoddyKit jest strukturyzowany dla początkujących i zaawansowanych użytkowników, więc możesz zacząć tutaj lub od początku i uczyć się w swoim tempie. To lekcja 4 z 4.
Ile czasu zajmuje lekcja „Tworzenie prostego agenta asystenta e-mail”?
Większość lekcji CoddyKit trwa około 5–10 minut. Każda lekcja to mały, interaktywny krok, dzięki czemu robisz systematyczne postępy i zawsze wracasz dokładnie do tego samego miejsca — na webie i w aplikacji.
Czy mogę pisać i uruchamiać kod w tej lekcji AI Agents?
Tak. Każda lekcja AI Agents zawiera wbudowany edytor kodu, więc piszesz i uruchamiasz prawdziwy kod bezpośrednio w przeglądarce i od razu otrzymujesz sprzężenie zwrotne od AI — bez konfiguracji na komputerze.
Wszystkie lekcje w tym kursie
- Łączenie z Gmailem przez API
- Programowe odczytywanie i wysyłanie e-maili
- Tworzenie i wyszukiwanie wydarzeń w kalendarzu
- Tworzenie prostego agenta asystenta e-mail