我如何自动化邮件监控 使用 GPT-4o

我每天阅读数百封邮件。实际上不是我 — 是 GPT-4o mini 在读。我只需每天早上在 WhatsApp 上收到一份 10 行的摘要。

Erwan 的邮箱 每天收到大量通讯：行业新闻简报、潜在客户回复、SaaS 工具催促、各类服务通知，全部混在一起，没有明显的优先级。hello@thenocodeguy.com, 每天收到大量通讯：行业新闻简报、潜在客户回复、SaaS 工具催促、各类服务通知，全部混在一起，没有明显的优先级。

经典问题：要么每天花 30 分钟全部读完（巨大的认知负担），要么跳过然后错过重要内容。

我的解决方案：一个自动流水线，读取、分类、总结并传递有用信息 — 无需任何人触碰邮箱。

一览架构

四个步骤，零次人工点击：

1. 1获取 : Microsoft Graph API 获取过去 24 小时的未读邮件
2. 2过滤 : Python 脚本分类：新闻简报、潜客、告警、垃圾邮件
3. 3摘要 : GPT-4o mini 为每封邮件生成摘要 + 相关性评分
4. 4摘要推送 : 每天早上 7:30 发送一条结构化的 WhatsApp 消息

第一步 — Microsoft Graph API

邮箱托管于 Microsoft 365。Graph API 允许通过 OAuth2 访问邮件，无需经过 IMAP。这更快、更稳定，且令牌不会过期（应用凭据，无需刷新令牌）。

脚本首先调用以获取过去 24 小时的邮件：

import httpx

def get_token(tenant_id, client_id, client_secret):
    url = f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token"
    data = {
        "grant_type": "client_credentials",
        "client_id": client_id,
        "client_secret": client_secret,
        "scope": "https://graph.microsoft.com/.default",
    }
    r = httpx.post(url, data=data)
    return r.json()["access_token"]

def fetch_recent_emails(token, user_email, hours=24):
    since = (datetime.utcnow() - timedelta(hours=hours)).isoformat() + "Z"
    url = (
        f"https://graph.microsoft.com/v1.0/users/{user_email}/messages"
        f"?$filter=receivedDateTime ge {since}"
        f"&$select=subject,from,receivedDateTime,bodyPreview,isRead"
        f"&$top=50&$orderby=receivedDateTime desc"
    )
    headers = {"Authorization": f"Bearer {token}"}
    r = httpx.get(url, headers=headers)
    return r.json().get("value", [])

简单。没有复杂的邮件库，没有 MIME 解析。Graph API 直接返回整洁的 JSON，包含发件人、主题和正文预览。

第二步 — 过滤与分类

在调用 GPT 之前先过滤。将 50 封邮件发给大语言模型既昂贵又慢。初始分类通过确定性规则完成：

SPAM_PATTERNS = ["unsubscribe", "se désabonner", "no-reply@", "noreply@"]
PRIORITY_SENDERS = ["@client.com", "erwan@", "hello@thenocodeguy.com"]
NEWSLETTER_KEYWORDS = ["newsletter", "digest", "weekly", "hebdo", "recap"]

def classify_email(email: dict) -> str:
    subject = email["subject"].lower()
    sender = email["from"]["emailAddress"]["address"].lower()
    preview = email["bodyPreview"].lower()
    
    if any(p in sender for p in SPAM_PATTERNS):
        return "spam"
    if any(s in sender for s in PRIORITY_SENDERS):
        return "priority"
    if any(k in subject or k in preview for k in NEWSLETTER_KEYWORDS):
        return "newsletter"
    return "other"

def filter_for_llm(emails):
    return [e for e in emails 
            if classify_email(e) in ("priority", "newsletter")]

结果：通常从 40-50 封邮件缩减到 10-15 封真正值得摘要的。节省 Token，摘要质量更好。

第三步 — GPT-4o mini 摘要与评分

对于每封过滤后的邮件，GPT-4o mini 生成两项内容：1-2 句话的摘要，以及 1 到 5 的相关性评分。提示简短且结构化以强制 JSON 响应：

SYSTEM_PROMPT = """Tu es un assistant de veille email pour un consultant en automatisation IA.
Pour chaque email, génère un JSON avec:
- summary: résumé actionnable en 1-2 phrases max (français)
- score: pertinence de 1 (bruit) à 5 (action requise)
- action: null ou "répondre" | "lire" | "archiver"
"""

def summarize_email(email: dict, client) -> dict:
    content = f"""
Expéditeur: {email['from']['emailAddress']['address']}
Sujet: {email['subject']}
Preview: {email['bodyPreview'][:500]}
"""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": content},
        ],
        response_format={"type": "json_object"},
        max_tokens=150,
    )
    return json.loads(response.choices[0].message.content)

response_format: json_object 至关重要 — 它避免了脆弱的 Markdown 解析。GPT 返回整洁的 JSON，可直接解析。

{
  "summary": "Client Kelly demande un devis pour automatisation CRM. Deadline vendredi.",
  "score": 5,
  "action": "répondre"
}

第四步 — WhatsApp 摘要

摘要按评分降序排列，格式化为可读的 WhatsApp 消息，并通过 OpenClaw 网关发送。整个流程作为 Windmill 定时作业在每天早上 7:30 运行。

def format_digest(summaries: list[dict]) -> str:
    sorted_items = sorted(summaries, key=lambda x: x["score"], reverse=True)
    
    lines = ["📧 *Digest Email — ce matin*\n"]
    
    for item in sorted_items:
        score_emoji = "🔴" if item["score"] >= 4 else "🟡" if item["score"] >= 2 else "⚪"
        action = f" → _{item['action']}_" if item["action"] else ""
        lines.append(f"{score_emoji} {item['subject']}")
        lines.append(f"   {item['summary']}{action}\n")
    
    lines.append(f"_{len(sorted_items)} emails analysés_")
    return "\n".join(lines)

消息在 WhatsApp 中看起来像这样：

Windmill 编排

完整脚本作为调度的 Python 作业在 Windmill 上运行。一些重要的实现细节：

实际改变了什么

在这个流水线之前，Erwan 早上打开邮箱需要花 15-20 分钟来整理、阅读、决策。这是每天的认知成本，看起来很小 — 但乘以 250 个工作日，每年代表 60 到 80 小时。

现在，工作流简化为：花 2 分钟阅读 WhatsApp 摘要，回复 1-2 封标记为「需要处理」的邮件。其余的自动处理。

接下来我会扩展的内容

当前流水线有意保持简单。自然的扩展方向：

• 自动回复潜客 : 检测报价请求 → GPT-4o 生成回复草稿 → 快速确认后发送（WhatsApp 中 1 次点击）。
• 跟进线程 : 跟踪对话线程：如果潜客 3 天内未回复，自动生成跟进消息。
• 洞察提取 : 识别 30 天内的规律：哪些主题反复出现？哪些客户写信最频繁？用于商业战略的数据。

这些扩展中的每一个都是一个额外的 Windmill 脚本。不是架构重构。这就是从一开始就设计良好的流水线的真正价值：可扩展性是轻而易举的。

值得记住的教训

人们经常把「邮件自动化」当作抽象概念来谈。实际上，这是一个 200 行 Python 代码的流水线，每天早上 7:30 运行，每周节省一小时 — 永久如此。

努力与价值的比例是惊人的。初始设置：4-5 小时。持续收益：每周至少 1 小时。1 个月后 ROI 为正。

最好的工作流是那些你忘了它们还在运行的工作流。