AI办公自动化（Excel/邮件/文档）完全教程

1. AI办公自动化概述与应用场景

办公自动化正从"规则驱动"走向"智能驱动"。大语言模型（LLM）的成熟，使得过去需要人工判断的复杂任务——如理解邮件语义、解析非结构化文档、生成动态报表——现在可以通过 API 调用自动完成。

典型应用场景

技术栈选型

• 语言模型：OpenAI GPT-4 / Claude / 本地部署的开源模型
• Python生态：openpyxl（Excel）、python-docx（Word）、python-pptx（PPT）、PyPDF2（PDF）
• 邮件协议：IMAP/SMTP + 专用SDK
• 语音处理：Whisper（语音转文字）

2. Excel智能处理

2.1 自然语言生成公式

用户输入"计算A列中大于100的数的平均值"，AI直接返回可用的Excel公式。

import openai

def nl_to_formula(description: str) -> str:
    """将自然语言描述转换为Excel公式"""
    response = openai.chat.completions.create(
        model="gpt-4",
        messages=[
            {
                "role": "system",
                "content": (
                    "你是Excel公式专家。用户用中文描述需求，"
                    "你只返回一个可用的Excel公式，不加任何解释。"
                    "假设数据从A1开始。"
                )
            },
            {"role": "user", "content": description}
        ],
        temperature=0
    )
    return response.choices[0].message.content.strip()

# 示例调用
formula = nl_to_formula("计算A列中大于100的数的平均值")
print(formula)  # 输出: =AVERAGEIF(A:A,">100")

2.2 自动化数据分析与报告

结合 openpyxl 和 LLM，可以读取 Excel 数据并自动生成分析报告。

from openpyxl import load_workbook
import openai

def analyze_excel(file_path: str, sheet_name: str = None) -> str:
    """读取Excel数据并生成分析报告"""
    wb = load_workbook(file_path, data_only=True)
    ws = wb[sheet_name] if sheet_name else wb.active

    # 提取表头和前50行数据
    headers = [cell.value for cell in ws[1]]
    rows = []
    for row in ws.iter_rows(min_row=2, max_row=51, values_only=True):
        rows.append(dict(zip(headers, row)))

    # 构造数据摘要
    data_summary = f"表头: {headers}\n数据行数: {ws.max_row - 1}\n前5行样本:\n"
    for r in rows[:5]:
        data_summary += f"  {r}\n"

    # 调用LLM生成分析
    response = openai.chat.completions.create(
        model="gpt-4",
        messages=[
            {
                "role": "system",
                "content": "你是数据分析师。根据提供的Excel数据摘要，生成简洁的中文分析报告，包含关键发现和建议。"
            },
            {"role": "user", "content": data_summary}
        ]
    )
    return response.choices[0].message.content

report = analyze_excel("sales_2024.xlsx")
print(report)

2.3 智能图表生成

from openpyxl import load_workbook
from openpyxl.chart import BarChart, Reference

def auto_chart(file_path: str, output_path: str):
    """根据数据自动生成柱状图"""
    wb = load_workbook(file_path)
    ws = wb.active

    chart = BarChart()
    chart.title = "销售数据概览"
    chart.style = 10
    chart.y_axis.title = "销售额"
    chart.x_axis.title = "月份"

    # 假设第一列是标签，第二列是数值
    data = Reference(ws, min_col=2, min_row=1, max_row=ws.max_row)
    cats = Reference(ws, min_col=1, min_row=2, max_row=ws.max_row)
    chart.add_data(data, titles_from_data=True)
    chart.set_categories(cats)
    chart.shape = 4

    ws.add_chart(chart, "D2")
    wb.save(output_path)
    print(f"图表已保存到 {output_path}")

3. 邮件智能分类与自动回复

3.1 邮件读取与分类

import imaplib
import email
from email.header import decode_header
import openai

def fetch_emails(host: str, user: str, password: str, folder: str = "INBOX", limit: int = 20):
    """通过IMAP获取最新邮件"""
    mail = imaplib.IMAP4_SSL(host)
    mail.login(user, password)
    mail.select(folder)
    _, msg_ids = mail.search(None, "ALL")
    ids = msg_ids[0].split()[-limit:]

    emails = []
    for eid in ids:
        _, data = mail.fetch(eid, "(RFC822)")
        msg = email.message_from_bytes(data[0][1])
        subject = decode_header(msg["Subject"])[0][0]
        if isinstance(subject, bytes):
            subject = subject.decode()
        body = ""
        if msg.is_multipart():
            for part in msg.walk():
                if part.get_content_type() == "text/plain":
                    body = part.get_payload(decode=True).decode(errors="ignore")
                    break
        else:
            body = msg.get_payload(decode=True).decode(errors="ignore")
        emails.append({
            "from": msg["From"],
            "subject": subject,
            "body": body[:500]  # 截取前500字
        })
    mail.logout()
    return emails

def classify_email(subject: str, body: str) -> dict:
    """对邮件进行智能分类"""
    response = openai.chat.completions.create(
        model="gpt-4",
        messages=[
            {
                "role": "system",
                "content": (
                    "将邮件分类为以下类别之一：[工作汇报, 客户询价, 内部通知, "
                    "会议邀请, 垃圾邮件, 紧急事务, 其他]。同时判断紧急程度：高/中/低。\n"
                    '返回JSON: {"category": "...", "urgency": "...", "summary": "一句话摘要"}'
                )
            },
            {"role": "user", "content": f"主题: {subject}\n正文: {body}"}
        ],
        response_format={"type": "json_object"},
        temperature=0
    )
    return eval(response.choices[0].message.content)

3.2 智能回复生成

def generate_reply(subject: str, body: str, tone: str = "professional") -> str:
    """根据邮件内容生成回复草稿"""
    tone_map = {
        "professional": "正式、专业的语气",
        "friendly": "友好、亲切的语气",
        "concise": "简洁、直接的语气"
    }
    response = openai.chat.completions.create(
        model="gpt-4",
        messages=[
            {
                "role": "system",
                "content": f"你是邮件助手。根据收到的邮件内容生成回复草稿，使用{tone_map.get(tone, tone_map['professional'])}。回复要有针对性，不要泛泛而谈。"
            },
            {"role": "user", "content": f"主题: {subject}\n正文:\n{body}"}
        ]
    )
    return response.choices[0].message.content

4. 文档自动生成与格式化

使用 python-docx 生成结构化的 Word 文档。

from docx import Document
from docx.shared import Inches, Pt
from docx.enum.text import WD_ALIGN_PARAGRAPH
import openai

def generate_document(title: str, outline: list[str], output_path: str):
    """根据大纲自动生成Word文档"""
    # 先让AI扩写每个章节
    sections = {}
    for heading in outline:
        response = openai.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": "你是技术文档写手。根据标题写出200-400字的章节内容，专业且易懂。"},
                {"role": "user", "content": heading}
            ]
        )
        sections[heading] = response.choices[0].message.content

    # 生成Word文档
    doc = Document()

    # 标题
    doc_title = doc.add_heading(title, level=0)
    doc_title.alignment = WD_ALIGN_PARAGRAPH.CENTER

    # 目录页（简化版）
    doc.add_heading("目录", level=1)
    for i, heading in enumerate(outline, 1):
        doc.add_paragraph(f"{i}. {heading}", style="List Number")
    doc.add_page_break()

    # 各章节内容
    for i, heading in enumerate(outline, 1):
        doc.add_heading(f"{i}. {heading}", level=1)
        doc.add_paragraph(sections[heading])

    # 设置默认字体
    style = doc.styles['Normal']
    font = style.font
    font.name = 'Microsoft YaHei'
    font.size = Pt(11)

    doc.save(output_path)
    print(f"文档已保存: {output_path}")

# 使用示例
outline = [
    "项目背景与目标",
    "技术方案设计",
    "实施计划与里程碑",
    "风险评估与应对",
    "预算与资源需求"
]
generate_document("2024年AI平台建设方案", outline, "ai_platform_proposal.docx")

5. 会议纪要自动整理

结合语音转录（Whisper）和 LLM 提取会议要点。

import openai
from datetime import datetime

def transcribe_audio(audio_path: str) -> str:
    """使用Whisper转录音频"""
    with open(audio_path, "rb") as f:
        transcript = openai.audio.transcriptions.create(
            model="whisper-1",
            file=f,
            language="zh"
        )
    return transcript.text

def extract_meeting_minutes(transcript: str) -> dict:
    """从转录文本中提取结构化会议纪要"""
    response = openai.chat.completions.create(
        model="gpt-4",
        messages=[
            {
                "role": "system",
                "content": (
                    "从会议转录文本中提取结构化纪要，返回JSON格式：\n"
                    '{"title": "会议主题", '
                    '"date": "日期", '
                    '"participants": ["参会人"], '
                    '"key_points": ["要点1", "要点2"], '
                    '"action_items": [{"task": "任务描述", "owner": "负责人", "deadline": "截止日期"}], '
                    '"decisions": ["决议1"]}'
                )
            },
            {"role": "user", "content": transcript}
        ],
        response_format={"type": "json_object"}
    )
    return eval(response.choices[0].message.content)

def format_minutes_markdown(minutes: dict) -> str:
    """将会议纪要格式化为Markdown"""
    md = f"# {minutes['title']}\n\n"
    md += f"**日期**: {minutes['date']}\n"
    md += f"**参会人**: {', '.join(minutes['participants'])}\n\n"
    md += "## 关键要点\n"
    for point in minutes['key_points']:
        md += f"- {point}\n"
    md += "\n## 行动项\n"
    md += "| 任务 | 负责人 | 截止日期 |\n|------|--------|----------|\n"
    for item in minutes['action_items']:
        md += f"| {item['task']} | {item['owner']} | {item['deadline']} |\n"
    md += "\n## 决议\n"
    for d in minutes['decisions']:
        md += f"- {d}\n"
    return md

# 完整流程
transcript = transcribe_audio("meeting_recording.mp3")
minutes = extract_meeting_minutes(transcript)
markdown = format_minutes_markdown(minutes)
with open(f"minutes_{datetime.now().strftime('%Y%m%d')}.md", "w") as f:
    f.write(markdown)

6. 日程管理与智能排程

from datetime import datetime, timedelta
import openai

def smart_schedule(events: list[dict], new_task: dict) -> dict:
    """
    智能排程：在现有日程中找到最佳空闲时段
    events: [{"start": "2024-01-15 09:00", "end": "2024-01-15 10:00", "title": "会议"}]
    new_task: {"title": "代码审查", "duration_minutes": 60, "priority": "high"}
    """
    prompt = f"""现有日程:
{chr(10).join(f"- {e['start']} ~ {e['end']}: {e['title']}" for e in events)}

新任务: {new_task['title']}
时长: {new_task['duration_minutes']}分钟
优先级: {new_task['priority']}

请在工作时间(9:00-18:00)内找到最早的可用时段，返回JSON:
{{"suggested_start": "YYYY-MM-DD HH:MM", "reason": "推荐理由"}}"""

    response = openai.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": "你是日程管理助手。找到最优的时间安排。"},
            {"role": "user", "content": prompt}
        ],
        response_format={"type": "json_object"},
        temperature=0
    )
    return eval(response.choices[0].message.content)

def conflict_check(events: list[dict]) -> list[str]:
    """检查日程冲突"""
    conflicts = []
    sorted_events = sorted(events, key=lambda x: x['start'])
    for i in range(len(sorted_events) - 1):
        current_end = sorted_events[i]['end']
        next_start = sorted_events[i + 1]['start']
        if next_start < current_end:
            conflicts.append(
                f"冲突: [{sorted_events[i]['title']}] 与 [{sorted_events[i+1]['title']}]"
            )
    return conflicts

7. PPT自动生成

使用 python-pptx 生成演示文稿，结合 AI 自动填充内容。

from pptx import Presentation
from pptx.util import Inches, Pt
import openai

def generate_ppt_content(topic: str, slide_count: int = 8) -> list[dict]:
    """用AI生成PPT各页内容"""
    response = openai.chat.completions.create(
        model="gpt-4",
        messages=[
            {
                "role": "system",
                "content": (
                    f"为演示文稿生成{slide_count}页内容，返回JSON数组。\n"
                    "每页: {\"title\": \"标题\", \"bullets\": [\"要点1\", \"要点2\"], \"notes\": \"演讲者备注\"}\n"
                    "第一页是封面(title + subtitle)，最后一页是总结/致谢。"
                )
            },
            {"role": "user", "content": topic}
        ],
        response_format={"type": "json_object"}
    )
    return eval(response.choices[0].message.content)["slides"]

def build_ppt(slides_data: list[dict], output_path: str):
    """根据数据生成PPT文件"""
    prs = Presentation()

    for i, slide_data in enumerate(slides_data):
        if i == 0:
            # 封面页
            layout = prs.slide_layouts[0]  # Title Slide
            slide = prs.slides.add_slide(layout)
            slide.shapes.title.text = slide_data["title"]
            if len(slide_data.get("bullets", [])) > 0:
                slide.placeholders[1].text = slide_data["bullets"][0]
        else:
            # 内容页
            layout = prs.slide_layouts[1]  # Title and Content
            slide = prs.slides.add_slide(layout)
            slide.shapes.title.text = slide_data["title"]
            body = slide.placeholders[1]
            tf = body.text_frame
            for j, bullet in enumerate(slide_data.get("bullets", [])):
                if j == 0:
                    tf.text = bullet
                else:
                    tf.add_paragraph().text = bullet

        # 演讲者备注
        if "notes" in slide_data:
            slide.notes_slide.notes_text_frame.text = slide_data["notes"]

    prs.save(output_path)
    print(f"PPT已生成: {output_path}")

# 使用示例
slides = generate_ppt_content("2024年Q4产品路线图", slide_count=10)
build_ppt(slides, "product_roadmap_q4.pptx")

8. PDF智能解析与提取

import PyPDF2
import openai

def extract_pdf_text(pdf_path: str) -> str:
    """提取PDF全文"""
    reader = PyPDF2.PdfReader(pdf_path)
    text = ""
    for page in reader.pages:
        text += page.extract_text() + "\n"
    return text

def summarize_pdf(pdf_path: str, max_words: int = 500) -> str:
    """对PDF内容进行智能摘要"""
    full_text = extract_pdf_text(pdf_path)
    # 对于长文档，截取前面部分（生产环境可用分块策略）
    truncated = full_text[:8000]

    response = openai.chat.completions.create(
        model="gpt-4",
        messages=[
            {
                "role": "system",
                "content": f"将以下PDF内容总结为不超过{max_words}字的中文摘要，突出核心观点和关键数据。"
            },
            {"role": "user", "content": truncated}
        ]
    )
    return response.choices[0].message.content

def extract_structured_data(pdf_path: str, schema: dict) -> dict:
    """从PDF中提取结构化数据"""
    full_text = extract_pdf_text(pdf_path)
    response = openai.chat.completions.create(
        model="gpt-4",
        messages=[
            {
                "role": "system",
                "content": (
                    f"从文档中提取以下字段，返回JSON:\n{schema}\n"
                    "如果某字段找不到，填null。"
                )
            },
            {"role": "user", "content": full_text[:6000]}
        ],
        response_format={"type": "json_object"}
    )
    return eval(response.choices[0].message.content)

# 使用示例
schema = {
    "company_name": "公司名称",
    "report_date": "报告日期",
    "revenue": "营收数据",
    "key_risks": "主要风险"
}
data = extract_structured_data("annual_report.pdf", schema)
print(data)

9. 多文档摘要与对比

import openai

def compare_documents(doc_texts: dict[str, str]) -> str:
    """
    对比多份文档，找出异同点
    doc_texts: {"文档A": "内容...", "文档B": "内容..."}
    """
    docs_formatted = "\n\n".join(
        f"=== {name} ===\n{text[:3000]}" for name, text in doc_texts.items()
    )

    response = openai.chat.completions.create(
        model="gpt-4",
        messages=[
            {
                "role": "system",
                "content": (
                    "对比以下多份文档，输出：\n"
                    "1. 各文档核心观点摘要（各3句话）\n"
                    "2. 共同点列表\n"
                    "3. 差异点列表（标注来源文档）\n"
                    "4. 综合建议"
                )
            },
            {"role": "user", "content": docs_formatted}
        ]
    )
    return response.choices[0].message.content

def batch_summarize(file_paths: list[str]) -> dict[str, str]:
    """批量摘要多个文档"""
    results = {}
    for path in file_paths:
        if path.endswith('.pdf'):
            text = extract_pdf_text(path)
        elif path.endswith('.docx'):
            from docx import Document
            doc = Document(path)
            text = "\n".join(p.text for p in doc.paragraphs)
        else:
            with open(path, 'r', encoding='utf-8') as f:
                text = f.read()

        response = openai.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": "用3-5句话总结以下文档的核心内容。"},
                {"role": "user", "content": text[:5000]}
            ]
        )
        results[path] = response.choices[0].message.content
    return results

10. 实战案例：企业办公自动化系统

将上述模块整合为一个统一的办公自动化服务。

"""
企业办公自动化系统 - 核心架构
"""
from dataclasses import dataclass
from enum import Enum
from typing import Callable
import json

class TaskType(Enum):
    EMAIL_CLASSIFY = "email_classify"
    EXCEL_ANALYZE = "excel_analyze"
    DOC_GENERATE = "doc_generate"
    MEETING_MINUTES = "meeting_minutes"
    PPT_GENERATE = "ppt_generate"
    PDF_PARSE = "pdf_parse"

@dataclass
class AutomationTask:
    task_type: TaskType
    input_data: dict
    callback: Callable = None

class OfficeAutomationEngine:
    """办公自动化调度引擎"""

    def __init__(self):
        self.handlers = {
            TaskType.EMAIL_CLASSIFY: self._handle_email,
            TaskType.EXCEL_ANALYZE: self._handle_excel,
            TaskType.DOC_GENERATE: self._handle_document,
            TaskType.MEETING_MINUTES: self._handle_meeting,
            TaskType.PPT_GENERATE: self._handle_ppt,
            TaskType.PDF_PARSE: self._handle_pdf,
        }
        self.task_log = []

    def submit(self, task: AutomationTask) -> dict:
        """提交并执行自动化任务"""
        handler = self.handlers.get(task.task_type)
        if not handler:
            return {"error": f"不支持的任务类型: {task.task_type}"}

        try:
            result = handler(task.input_data)
            self.task_log.append({
                "type": task.task_type.value,
                "status": "success",
                "output": result
            })
            return {"status": "success", "data": result}
        except Exception as e:
            self.task_log.append({
                "type": task.task_type.value,
                "status": "failed",
                "error": str(e)
            })
            return {"status": "error", "message": str(e)}

    def _handle_email(self, data: dict) -> dict:
        emails = fetch_emails(data["host"], data["user"], data["password"])
        classified = [classify_email(e["subject"], e["body"]) for e in emails]
        return {"total": len(emails), "classified": classified}

    def _handle_excel(self, data: dict) -> dict:
        return {"report": analyze_excel(data["file_path"])}

    def _handle_document(self, data: dict) -> dict:
        generate_document(data["title"], data["outline"], data["output_path"])
        return {"file": data["output_path"]}

    def _handle_meeting(self, data: dict) -> dict:
        transcript = transcribe_audio(data["audio_path"])
        minutes = extract_meeting_minutes(transcript)
        return {"minutes": minutes}

    def _handle_ppt(self, data: dict) -> dict:
        slides = generate_ppt_content(data["topic"], data.get("slide_count", 8))
        build_ppt(slides, data["output_path"])
        return {"file": data["output_path"]}

    def _handle_pdf(self, data: dict) -> dict:
        if data.get("schema"):
            return extract_structured_data(data["file_path"], data["schema"])
        return {"summary": summarize_pdf(data["file_path"])}

    def get_report(self) -> str:
        """生成任务执行报告"""
        success = sum(1 for t in self.task_log if t["status"] == "success")
        failed = sum(1 for t in self.task_log if t["status"] == "failed")
        return json.dumps({
            "total_tasks": len(self.task_log),
            "success": success,
            "failed": failed,
            "success_rate": f"{success/len(self.task_log)*100:.1f}%" if self.task_log else "N/A"
        }, ensure_ascii=False, indent=2)

# 使用示例
engine = OfficeAutomationEngine()

# 批量处理邮件
engine.submit(AutomationTask(
    task_type=TaskType.EMAIL_CLASSIFY,
    input_data={"host": "imap.company.com", "user": "user", "password": "pass"}
))

# 生成项目报告PPT
engine.submit(AutomationTask(
    task_type=TaskType.PPT_GENERATE,
    input_data={
        "topic": "2024年度技术团队工作汇报",
        "slide_count": 12,
        "output_path": "tech_annual_report.pptx"
    }
))

print(engine.get_report())

调度与集成方式

实际部署时，可以通过以下方式触发自动化流程：

• 定时任务：每日早8点自动分类邮件、生成日报
• Webhook触发：收到新邮件时自动分类并推送到Slack/飞书
• 文件监控：监控特定文件夹，新文件入库时自动解析和归档
• API接口：对外暴露REST API，供其他系统调用

# FastAPI 接口示例
from fastapi import FastAPI, UploadFile

app = FastAPI(title="办公自动化API")
engine = OfficeAutomationEngine()

@app.post("/api/email/classify")
async def classify_emails(host: str, user: str, password: str):
    task = AutomationTask(
        task_type=TaskType.EMAIL_CLASSIFY,
        input_data={"host": host, "user": user, "password": password}
    )
    return engine.submit(task)

@app.post("/api/pdf/parse")
async def parse_pdf(file: UploadFile, schema: str = None):
    content = await file.read()
    temp_path = f"/tmp/{file.filename}"
    with open(temp_path, "wb") as f:
        f.write(content)
    task = AutomationTask(
        task_type=TaskType.PDF_PARSE,
        input_data={"file_path": temp_path, "schema": eval(schema) if schema else None}
    )
    return engine.submit(task)

11. 安全合规与权限管理

办公自动化系统处理大量敏感数据，安全设计是底线。

11.1 数据安全原则

• 最小权限原则：每个模块只访问它需要的数据
• 数据脱敏：发送给LLM的内容去除个人身份信息（PII）
• 传输加密：所有API通信使用TLS，邮件连接使用SSL
• 日志审计：记录所有自动化操作，便于追溯

11.2 PII脱敏示例

import re

def sanitize_for_llm(text: str) -> str:
    """在发送给LLM前脱敏个人信息"""
    # 手机号
    text = re.sub(r'1[3-9]\d{9}', '[手机号]', text)
    # 身份证号
    text = re.sub(r'\d{17}[\dXx]', '[身份证号]', text)
    # 邮箱
    text = re.sub(r'[\w.+-]+@[\w-]+\.[\w.]+', '[邮箱]', text)
    # 银行卡号
    text = re.sub(r'\d{16,19}', '[银行卡号]', text)
    return text

11.3 权限控制

from functools import wraps

class Permission:
    EMAIL_READ = "email:read"
    EMAIL_SEND = "email:send"
    FILE_READ = "file:read"
    FILE_WRITE = "file:write"
    LLM_CALL = "llm:call"

def require_permission(*permissions):
    """权限装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            user_perms = get_current_user_permissions()  # 实现从session/JWT获取
            for perm in permissions:
                if perm not in user_perms:
                    raise PermissionError(f"缺少权限: {perm}")
            return func(*args, **kwargs)
        return wrapper
    return decorator

@require_permission(Permission.EMAIL_READ, Permission.LLM_CALL)
def classify_emails_safe(host, user, password):
    """带权限检查的邮件分类"""
    return fetch_emails(host, user, password)

11.4 生产环境检查清单

• LLM API密钥通过环境变量或密钥管理服务注入，不硬编码
• 所有用户输入在发送给LLM前进行长度限制和内容过滤
• 邮件发送功能设置人工审核环节（至少在初期）
• 文件上传限制大小和类型，防止恶意文件
• 定期轮换IMAP/SMTP密码和API密钥
• 建立异常告警机制，自动化任务失败时及时通知

总结

AI办公自动化的核心思路是：将重复性、规则性的办公任务交给AI处理，人专注于决策和创造性工作。技术实现上，Python生态提供了完善的文档处理库，而LLM负责"理解"和"生成"这两个过去无法自动化的环节。

从单点能力（如邮件分类、PDF提取）开始，逐步整合为完整的自动化流水线，是落地的最佳路径。始终把安全合规放在第一位——自动化的效率不应以数据安全为代价。