第 27 章 案例三 自动化工作流 Agent

本章要解决的问题

每周要花 3 小时的报表生成，交给 Agent 自动跑——规划、多智能体、成本优化怎么配合？

章节大纲

• 27.1 多步骤任务拆解（规划 + 多智能体）
• 27.2 MCP 集成办公套件（豆包）
• 27.3 成本优化与异常恢复
• 27.4 全流程编排与监控
• 🛠 解决方案：自动化任务失败自愈机制 + 成本预算控制

27.1 场景与任务拆解

27.1.1 场景：周报自动生成

业务：某运营团队每周要生成一份经营周报，人工耗时 3 小时：

拉数据（订单/流量/转化）→ 清洗 → 分析趋势 → 撰写报告 → 发邮件/同步群

交给 Agent 的目标：全自动跑，出错自愈，成本可控。

27.1.2 任务拆解（第 15 章规划）

[<span>Planner</span>] 生成周报任务计划
  ├─ 步骤<span>1</span>：拉取本周订单数据（工具：数据查询）
  ├─ 步骤<span>2</span>：拉取本周流量数据（工具：数据查询）
  ├─ 步骤<span>3</span>：分析销售趋势（步骤<span>1</span>/<span>2</span> 后）
  ├─ 步骤<span>4</span>：分析渠道表现（步骤<span>1</span>/<span>2</span> 后）
  ├─ 步骤<span>5</span>：撰写周报（步骤<span>3</span>/<span>4</span> 后）
  └─ 步骤<span>6</span>：发送周报（步骤<span>5</span> 后）

依赖关系：<span>1</span>,<span>2</span> → <span>3</span>,<span>4</span> → <span>5</span> → <span>6</span>

图 1：周报任务依赖图

两个独立分支（步骤 3 和 4 都依赖 1、2 但不互相依赖）→ 可并行（第 12 章）。

27.1.3 为什么用规划 + 多智能体，而不是一条提示链

本案例用"规划 + 并行 + 主管编排"：Planner 出计划 → 主管派活 → 分析类任务并行 → 汇总质检 → 发送。

27.2 MCP 集成办公套件

27.2.1 需要的工具集

27.2.2 MCP Server：办公套件接入（第 7 章）

示例代码：以下代码演示核心结构，省略了异常处理、日志和完整 import。

<span>from</span> fastmcp <span>import</span> FastMCP

mcp = FastMCP(<span>"office-suite"</span>)

<span>@mcp.tool()</span>
<span>def</span> <span>create_report_doc</span>(<span>title: <span>str</span>, content: <span>str</span></span>) -> <span>str</span>:
    <span>"""创建文档并返回链接。生成报告/文档时使用。"""</span>
    doc_id = office_api.create_doc(title, content)
    <span>return</span> <span>f"文档已创建：https://docs.internal/<span>{doc_id}</span>"</span>

<span>@mcp.tool()</span>
<span>def</span> <span>send_wecom_message</span>(<span>channel: <span>str</span>, content: <span>str</span></span>) -> <span>dict</span>:
    <span>"""发送企业微信消息。任务完成/异常通知时使用。"""</span>
    <span>return</span> wecom_api.send(channel, content)

<span>@mcp.tool()</span>
<span>def</span> <span>send_email</span>(<span>to: <span>str</span>, subject: <span>str</span>, body: <span>str</span></span>) -> <span>dict</span>:
    <span>"""发送邮件。需要正式通知时使用。注意：敏感操作，需确认。"""</span>
    <span>return</span> email_api.send(to, subject, body)

注意：send_email 是敏感操作（第 5/22 章）——MCP Server 内部要做确认机制。

27.3 成本优化与异常恢复

27.3.1 成本优化（第 23 章五个金矿）

27.3.2 异常恢复：失败自愈（第 21/23 章）

自动化任务最怕"半夜挂了没人管"。设计三级自愈：

图 2：三级自愈机制

<span>def</span> <span>run_with_recovery</span>(<span>plan, budget</span>):
    <span>for</span> step <span>in</span> plan[<span>"steps"</span>]:
        <span>try</span>:
            result = execute_step(step)
        <span>except</span> DataSourceError:
            <span># 一级：重试（指数退避）</span>
            result = retry(step, times=<span>2</span>, backoff=[<span>5</span>, <span>15</span>])
        <span>except</span> DataMissingError <span>as</span> e:
            <span># 二级：跳过并记录（该数据源本周缺失）</span>
            result = {<span>"skipped"</span>: <span>True</span>, <span>"reason"</span>: <span>str</span>(e)}
            notify(<span>"周报数据缺失"</span>, <span>f"<span>{step}</span> 数据源不可用，已跳过"</span>)
        <span>except</span> Exception <span>as</span> e:
            <span># 三级：降级（用缓存数据/标记待人工）</span>
            result = fallback(step, e)
            escalate_to_human(step, e)      <span># 转人工</span>
        results.append(result)
    <span>return</span> results

自愈分级原则：可重试的重试 → 可跳过的跳过（带通知）→ 兜不住的转人工。自动化不等于无人值守，而是"能自愈的自愈，自愈不了的通知人"。

27.3.3 预算控制（第 23 章三层熔断）

BUDGET = {
    <span>"per_task"</span>: {<span>"max_cost"</span>: <span>0.5</span>, <span>"max_steps"</span>: <span>20</span>},
    <span>"per_month"</span>: {<span>"max_cost"</span>: <span>30</span>, <span>"alarm"</span>: <span>0.8</span>},
}
<span># 单任务费用超限 → 终止并通知；月预算 80% → 告警</span>

27.4 全流程编排与监控

27.4.1 编排器：定时触发 + 状态管理

<span>from</span> apscheduler.schedulers.blocking <span>import</span> BlockingScheduler

<span>def</span> <span>weekly_report_job</span>():
    <span># 1. 生成计划（第15章 Planner）</span>
    plan = planner.plan(<span>"生成本周经营周报"</span>)
    <span># 2. 并行执行独立分支（第12章），生产环境需加信号量限流</span>
    results = run_branches(plan, parallel_groups=[[<span>"分析趋势"</span>,<span>"分析渠道"</span>]])
    <span># 3. 汇总质检（第17章）</span>
    report = synthesize(results)
    ok, issues = quality_check(report)
    report = report <span>if</span> ok <span>else</span> revise(report, issues)
    <span># 4. 交付（MCP 工具）</span>
    doc_url = create_report_doc(<span>"第X周经营周报"</span>, report)
    send_wecom_message(<span>"周报群"</span>, <span>f"周报已生成：<span>{doc_url}</span>"</span>)
    <span># 5. 留痕（第21章）</span>
    save_trace(<span>"weekly_report"</span>, results, doc_url)

scheduler = BlockingScheduler()
scheduler.add_job(weekly_report_job, <span>"cron"</span>, day_of_week=<span>"mon"</span>, hour=<span>"9"</span>)
scheduler.start()

图 3：自动化编排监控

27.4.2 监控面板（第 20/21 章）

27.4.3 数据质量校验（防"自动生成垃圾"）

自动化最大的隐藏风险：流程跑通了，但数据错了，还自动发出去了。必须在发送前加数据校验：

图 4：发送前数据校验

<span>def</span> <span>data_sanity_check</span>(<span>report</span>):
    checks = [
        report[<span>"revenue"</span>] >= <span>0</span>,                        <span># 金额非负</span>
        <span>abs</span>(report[<span>"revenue"</span>] - <span>sum</span>(report[<span>"items"</span>])) < <span>0.01</span>,  <span># 汇总一致</span>
        report[<span>"orders"</span>] > <span>0</span>,                          <span># 订单数合理</span>
        <span>len</span>(report[<span>"channels"</span>]) >= <span>3</span>,                  <span># 渠道完整</span>
    ]
    <span>return</span> <span>all</span>(checks)
<span># 校验不过 → 不发送，转人工审核</span>

铁律：自动化任务发送给"人看"的内容，发送前必须过数据校验 + 质检（第 17 章）——宁可拦下误报，不可发出错误。

🛠 解决方案：自动化任务失败自愈机制 + 成本预算控制

常见问题

1. "半夜任务挂了没人管"：无自愈。对策：三级自愈（重试→跳过带通知→转人工，27.3.2）。
2. "自动发出了错误报表"：无数据校验。对策：发送前 data_sanity_check + 质检（27.4.3）。
3. "成本一个月比一个月高"：无预算熔断。对策：三层熔断 + 成本监控（27.3.3）。
4. "任务偶尔超时"：数据量大。对策：并行化 + 大任务拆分 + 超时分层。
5. "改了一版，流程直接跑不通"：无回归。对策：每周跑一次"干跑"验证（不真发），用评测集回归（第 20 章）。

解决方案速查表

实战提示

1. 自动化 ≠ 无人值守：能自愈的自愈，自愈不了的通知人。
2. 发送前必须校验：给"人看"的内容，数据校验 + 质检缺一不可。
3. 三级自愈是标配：重试 → 跳过（通知）→ 转人工。
4. 预算熔断必须有：自动化任务没有预算上限就是定时炸弹。
5. 干跑回归：改流程后先"干跑"（不真发）验证，再让它自动跑。