搭一个会分拣邮件、排日程、自动跟进的个人 AI 助理

这套东西最终能跑出什么

先描述跑通之后的日常，再决定要不要动手。

早上 8 点，你收到一份"晨报"：

• 昨天夜里进来的 62 封邮件被分成了 6 类，其中 4 封被标成「今天必须处理」；
• 3 封是会议邀约，助理已经把它们转成了日程草稿，只等你点确认；
• 17 封订阅邮件被自动归档到「订阅」文件夹，不再占用未读；
• 5 封是需要你回复的，每封都附了一段中文回信草稿；
• 有 2 个线程挂了 4 天没人回，出现在「待跟进」清单里。

晚上 6 点，助理再跑一次，把当天仍然没有回复的线程重新列出来，并按你设定的天数往后顺延提醒。

整个过程里，助理只做三件事：分类、建草稿、提醒，真正的"发送"和"确认"仍然握在你手里。这个边界一开始就要划清楚，后面会反复用到。

前置条件清单

动手之前，先把这些准备好：

1. 一个常用邮箱，并且知道它的 IMAP / SMTP 服务器地址和端口。Gmail 走 Google API 或 IMAP 都可以；国内主流邮箱一般需要在网页端设置里手动开启 IMAP/SMTP 服务并生成「授权码 / 应用专用密码」，具体路径以各邮箱官方帮助页为准。

2. 一个日历。Google Calendar、Outlook Calendar、飞书日历、钉钉日历都行。本教程默认先用 .ics 文件 + 日历草稿的方式，因为它不依赖任何 SDK，最不容易卡住。

3. 一台长期开机的机器。一台小 VPS、家里的迷你主机、树莓派都够用。它不需要很强，因为重活都交给模型 API 了。

4. Python 环境，建议 3.10 以上，具体以 Python 官方文档当前版本为准。

5. 一个大模型 API Key。多数厂商都提供 OpenAI 兼容的接口，选一个你已经在用的即可，模型名以厂商官方文档为准。

6. 一个本地数据库：SQLite 足够，不需要额外安装服务。

7. 一个调度器：Linux 上直接用 cron，Windows 上用「任务计划程序」。

8. 时间预算：跑通最小闭环大约半天到一天。

还需要一个观念准备：先用只读模式跑两周。只看不写、只生成草稿不发送，观察分类准确率，再逐步放开权限。

分步骤

第 1 步：先定义"行动"，别急着写代码

很多人一上来就写 IMAP 连接，结果写到一半发现不知道该往哪分流。先花十分钟把输出结构定下来。建议先只用 6 个分类：

配套定一个 JSON 结构：

```json

{

"category": "ACTION_TODAY",

"summary": "一句话中文摘要",

"needs_reply": true,

"deadline": "YYYY-MM-DDTHH:MM:SS+08:00 或 null",

"meeting": {

"title": "会议标题",

"start": "YYYY-MM-DDTHH:MM:SS+08:00",

"end": "YYYY-MM-DDTHH:MM:SS+08:00",

"attendees": ["someone@example.com"]

},

"followup_after_days": 3,

"draft_reply": "中文回信草稿，不需要回复时为空字符串"

}

```

字段固定下来，后面所有代码都围绕它转。

第 2 步：把邮箱接进来（只读）

先写一个能把未读邮件捞出来的脚本。这里的密码请用邮箱的「授权码 / 应用专用密码」，不要用你登录网页的主密码。

```bash

.env

IMAP_HOST=imap.example.com

IMAP_PORT=993

IMAP_USER=you@example.com

IMAP_PASS=这里填授权码或应用专用密码

LLM_API_KEY=你的模型厂商 Key

LLM_BASE_URL=https://api.example.com/v1

LLM_MODEL=按厂商官方文档填写

```

```python

inbox.py

import imaplib, email, os

from email.header import decode_header

from dotenv import load_dotenv

load_dotenv()

def connect():

m = imaplib.IMAP4_SSL(os.environ["IMAP_HOST"], int(os.environ.get("IMAP_PORT", 993)))

m.login(os.environ["IMAP_USER"], os.environ["IMAP_PASS"])

return m

def fetch_unseen(limit=20):

m = connect()

m.select("INBOX")

typ, data = m.search(None, "UNSEEN")

uids = data[0].split()[-limit:]

for uid in uids:

typ, msg_data = m.fetch(uid, "(RFC822)")

if msg_data and msg_data[0]:

yield uid, email.message_from_bytes(msg_data[0][1])

m.logout()

def dec(raw):

if not raw:

return ""

out = []

for text, enc in decode_header(raw):

if isinstance(text, bytes):

out.append(text.decode(enc or "utf-8", errors="replace"))

else:

out.append(text)

return "".join(out)

def get_body(msg):

if msg.is_multipart():

for part in msg.walk():

ctype = part.get_content_type()

disp = str(part.get("Content-Disposition") or "")

if ctype == "text/plain" and "attachment" not in disp:

payload = part.get_payload(decode=True) or b""

return payload.decode(part.get_content_charset() or "utf-8", errors="replace")

return ""

payload = msg.get_payload(decode=True) or b""

return payload.decode(msg.get_content_charset() or "utf-8", errors="replace")

```

短信、通知类邮件正文很短，截断到 6000 字符足够判断意图，也能省 token。

第 3 步：让模型输出结构化结果

把邮件正文拼成一段文本交给模型，要求它只返回 JSON。

```python

classify.py

import json, os

from openai import OpenAI # 多数厂商提供 OpenAI 兼容接口

client = OpenAI(api_key=os.environ["LLM_API_KEY"], base_url=os.environ["LLM_BASE_URL"])

SYSTEM = """你是一个邮件分拣助手。

只输出 JSON，不要输出解释、不要用 markdown 代码块包裹。

分类只能是：ACTION_TODAY / SCHEDULE / WAITING / FYI / NEWSLETTER / SPAM。

如果邮件内容不足以判断，分类给 FYI，并在 summary 里说明。

draft_reply 用中文，语气礼貌简洁，不超过 150 字。"""

def build_prompt(sender, subject, date, body):

return f"""发件人：{sender}

主题：{subject}

时间：{date}

正文：

{body[:6000]}"""

def classify(sender, subject, date, body):

resp = client.chat.completions.create(

model=os.environ["LLM_MODEL"],

temperature=0,

response_format={"type": "json_object"},

messages=[

{"role": "system", "content": SYSTEM},

{"role": "user", "content": build_prompt(sender, subject, date, body)},

],

)

text = resp.choices[0].message.content

try:

return json.loads(text)

except json.JSONDecodeError:

兜底：截取第一个 { 到最后一个 }

start, end = text.find("{"), text.rfind("}")

return json.loads(text[start:end + 1]) if start >= 0 else {}

```

response_format 不是所有厂商都支持，遇到报错就把它删掉，靠AI 词典：系统提示词">系统提示词约束 + 上面的兜底解析。

第 4 步：把结果落到 SQLite

```python

store.py

import sqlite3

SCHEMA = """

CREATE TABLE IF NOT EXISTS mail (

message_id TEXT PRIMARY KEY,

received_at TEXT,

sender TEXT,

subject TEXT,

category TEXT,

summary TEXT,

needs_reply INTEGER DEFAULT 0,

draft_reply TEXT,

meeting_json TEXT,

created_at TEXT DEFAULT (datetime('now'))

);

CREATE TABLE IF NOT EXISTS followup (

id INTEGER PRIMARY KEY AUTOINCREMENT,

message_id TEXT UNIQUE,

due_at TEXT,

status TEXT DEFAULT 'open'

);

"""

def init(db_path="assistant.db"):

conn = sqlite3.connect(db_path)

conn.executescript(SCHEMA)

conn.commit()

return conn

```

message_id 做主键，这样同一封邮件重复拉取也不会重复处理——这是整个流程里最重要的一行设计。

第 5 步：生成日程草稿，而不是直接写日历

先落地成 .ics 文件，你双击导入或拖进日历确认。等跑顺了再换成日历 API 直接创建。

```python

calendar_draft.py

import datetime

def make_ics(uid, title, start_utc, end_utc, organizer, attendee=None):

fmt = "%Y%m%dT%H%M%SZ"

lines = [

"BEGIN:VCALENDAR",

"VERSION:2.0",

"PRODID:-//personal-assistant//CN",

"BEGIN:VEVENT",

f"UID:{uid}@assistant.local",

f"DTSTAMP:{datetime.datetime.now(datetime.timezone.utc).strftime(fmt)}",

f"DTSTART:{start_utc.strftime(fmt)}",

f"DTEND:{end_utc.strftime(fmt)}",

f"SUMMARY:{title}",

f"ORGANIZER:mailto:{organizer}",

]

if attendee:

lines.append(f"ATTENDEE:mailto:{attendee}")

lines += ["END:VEVENT", "END:VCALENDAR"]

return "\r\n".join(lines)

```

建议单独建一个叫「AI 助理」的日历，草稿全丢在里面，确认无误后再手动挪进主日历。这样即使模型判断错了，也不会污染你的正式日程。

第 6 步：跟进提醒，靠定时扫描

```python

followup.py

import datetime, sqlite3

def due_followups(conn):

now = datetime.datetime.now(datetime.timezone.utc).isoformat()

return conn.execute(

"""SELECT m.message_id, m.subject, m.sender, f.due_at

FROM followup f JOIN mail m ON m.message_id = f.message_id

WHERE f.status = 'open' AND f.due_at <= ?

ORDER BY f.due_at""",

(now,),

).fetchall()

def close_followup(conn, message_id):

conn.execute("UPDATE followup SET status='done' WHERE message_id=?", (message_id,))

conn.commit()

```

当你在邮箱里真的回复了对方，下一轮扫描时该线程会出现新的发件邮件，此时把 followup 的状态置为 done 即可。判断"是否已回复"最简单的办法是：检查该会话里是否存在由你自己发出的邮件。

第 7 步：用 cron 串成闭环

```cron

*/15 * * * * cd /opt/assistant && /usr/bin/python3 run_ingest.py >> logs/ingest.log 2>&1

0 8 * * 1-5 cd /opt/assistant && /usr/bin/python3 run_digest.py >> logs/digest.log 2>&1

0 18 * * * cd /opt/assistant && /usr/bin/python3 run_followup.py >> logs/followup.log 2>&1

```

• run_ingest.py：拉未读 → 分类 → 入库 → 生成 ics 草稿。
• run_digest.py：汇总过去 24 小时，生成晨报邮件发给自己。
• run_followup.py：扫描到期跟进项，能自己回的就标记完成，其余的重新提醒。

晨报邮件本身可以用 SMTP 发，也可以用企业微信 / 飞书 / Telegram 的机器人 webhook，发一条纯文本消息更省事。

第 8 步：划清自动化的边界

这张表建议照抄，边界模糊的地方一律保守：

常见坑与排错

1. IMAP 登录失败 AUTHENTICATIONFAILED

九成是用了网页登录密码。需要在邮箱设置里开启 IMAP/SMTP 并生成授权码，把端口和加密方式一起核对一遍。

2. 中文主题变乱码

decode_header 返回的是 (bytes, encoding) 元组，必须逐个解码再拼接，直接 str() 会得到 =?UTF-8?B?...?= 这种原文。

3. 正文取出来是空的

典型的多部分邮件（multipart/alternative）。用 msg.walk() 递归，优先取 text/plain；只有 HTML 时可以把标签粗略去掉再喂给模型。

4. 模型返回的不是合法 JSON

先删掉 response_format 试试，再检查系统提示词里有没有明确写"只输出 JSON"。代码里保留兜底解析，坏数据不要写库。

5. 同一封邮件被处理多次

用 Message-ID 做唯一键，INSERT OR IGNORE。不要用「主题 + 时间」当键，回复线程会撞车。

6. 时间全错了一个时区

数据库里统一存 UTC，展示和生成 ics 时再转本地时区。模型返回的时间字符串要求带偏移量，如 +08:00。

7. 规则写太细，两周后自己都忘了

一开始只保留 6 个分类和 3 条硬规则，跑够两周再看日志调。日志里记录每封邮件的分类理由，回看时很有用。

8. 隐私没想清楚

邮件正文会离开本机。可以在入库前加一层过滤：发件人域名白名单之外的邮件、正文含特定关键词的邮件，只做本地规则分类，不调用模型。

9. 触发频率限制

给 API 调用加上指数退避重试，单次处理失败不要让整批任务挂掉，把失败的 message_id 记下来下次重试。

10. 一上来就自动发送

这是最容易踩的坑。先用草稿模式跑满一个月，等分类准确率稳定了，再考虑对极少数场景开放自动回复。

下一步建议

• 加一份周报：每周一汇总上周所有 WAITING 状态的线程，一眼看清欠了谁。
• 把"创建日程"升级成"找空闲时间"：接入日历 API 读取忙闲，让模型在候选时段里挑一个，而不是被动接受对方给的时间。
• 加人工反馈按钮：晨报里每条分类后面放「对 / 不对」，把结果写回数据库，作为后续调整提示词的依据。
• 迁移到低代码编排：n8n、Dify 这类工具自带定时触发和可视化调试，流程稳定后可以搬过去，改规则不用动代码。
• 补上历史上下文：把过去几个月的邮件做向量化检索，生成回信草稿时带上与同一发件人的历史往来，草稿质量会明显不同。

从最小闭环开始，先让它在只读模式下安静跑两周，你很快就能判断出哪些规则该收紧、哪些可以放手。