0Pricing
AI Agents · 课时

构建简单的电子邮件助手代理

端到端流程：读取收件箱 → 生成摘要 → 撰写回复 → 等待批准。

构建简单的电子邮件助手代理 是 CoddyKit 上的免费 AI Agents 课时。 这是第 4 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 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 所需的关键字段：主题、发件人和纯文本正文。请保持正文简短——截取前 2000 个字符，以适应 LLM 的上下文限制并降低 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

第 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 步：生成回复草稿

对于被归类为需要处理的邮件，请让 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

错误处理与韧性

请用异常捕获结构包裹每个邮件处理步骤。某封邮件处理失败时，绝不能阻止代理继续处理其余邮件。记录错误时加入邮件 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 使用 JSON 输出将每封邮件归类为需要处理、供参考或垃圾邮件
  • 起草：LLM 以专业语气生成回复，并加入 [FILL IN] 占位符
  • 保存草稿：将草稿存储在 Gmail 中供人工审核——绝不自动发送
  • 标记已处理：添加自定义标签并移除 UNREAD，防止重复处理
  • 错误隔离：为每封邮件使用异常捕获结构，防止单个错误中断整个批次
  • 审计日志：使用 JSONL 文件记录每项操作，便于审核和改进

常见问题解答

「构建简单的电子邮件助手代理」课时是免费的吗？

是的 — 「构建简单的电子邮件助手代理」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。

「构建简单的电子邮件助手代理」这节课中我会学到什么？

端到端流程：读取收件箱 → 生成摘要 → 撰写回复 → 等待批准。 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗？

无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 4 节课，共 4 节。

「构建简单的电子邮件助手代理」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 AI Agents 课中编写并运行代码吗？

能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

  1. 通过 API 连接 Gmail
  2. 以编程方式读取和发送电子邮件
  3. 创建与查询日历事件
  4. 构建简单的电子邮件助手代理
← 返回 AI Agents