第 27 章 案例三 自动化工作流 Agent
英文标题：第 27 章 案例三 自动化工作流 Agent
第 27 章 案例三：自动化工作流 Agent 本章要解决的问题 章节大纲 27.1 多步骤任务拆解（规划 + 多智能体） 27.2 MCP 集成办公套件（豆包） 27.3 成本优化与异常恢复 27.
打开原文相关 8 · 机会 2
3.2 KB
589 B
25.9 KB
16.3 KB
17.1 KB
18.2 KB
5.7 KB
中文
第 27 章 案例三：自动化工作流 Agent
本章要解决的问题
每周要花 3 小时的报表生成，交给 Agent 自动跑——规划、多智能体、成本优化怎么配合？
章节大纲
- 27.1 多步骤任务拆解（规划 + 多智能体）
- 27.2 MCP 集成办公套件（豆包）
- 27.3 成本优化与异常恢复
- 27.4 全流程编排与监控
- 🛠 解决方案：自动化任务失败自愈机制 + 成本预算控制
27.1 场景与任务拆解
27.1.1 场景：周报自动生成
业务 ：某运营团队每周要生成一份经营周报，人工耗时 3 小时：
拉数据（订单/流量/转化）→ 清洗 → 分析趋势 → 撰写报告 → 发邮件/同步群
交给 Agent 的目标 ：全自动跑，出错自愈，成本可控。
27.1.2 任务拆解（第 15 章规划）
[Planner] 生成周报任务计划
├─ 步骤1：拉取本周订单数据（工具：数据查询）
├─ 步骤2：拉取本周流量数据（工具：数据查询）
├─ 步骤3：分析销售趋势（步骤1/2 后）
├─ 步骤4：分析渠道表现（步骤1/2 后）
├─ 步骤5：撰写周报（步骤3/4 后）
└─ 步骤6：发送周报（步骤5 后）
依赖关系：1,2 → 3,4 → 5 → 6
图 1：周报任务依赖图
两个独立分支 （步骤 3 和 4 都依赖 1、2 但不互相依赖）→ 可并行 （第 12 章）。
27.1.3 为什么用规划 + 多智能体，而不是一条提示链
| 需求 | 方案 | 理由 |
|---|---|---|
| 任务结构固定 | 提示链即可（第 10 章） | 步骤写死 |
| 但数据源可能变 | 需要规划的动态性（第 15 章） | 本周数据源缺一个要重排 |
| 分析任务较独立 | 可并行 + 角色分工 | 多智能体（第 16 章） |
本案例用"规划 + 并行 + 主管编排" ：Planner 出计划 → 主管派活 → 分析类任务并行 → 汇总质检 → 发送。
27.2 MCP 集成办公套件
27.2.1 需要的工具集
| 工具 | 用途 | 来源 |
|---|---|---|
| 查订单数据 | 拉取销售数据 | 内部数据 API |
| 查流量数据 | 拉取流量数据 | 内部数据 API |
| 写文档 | 生成周报文档 | 办公套件 MCP |
| 发消息 | 同步到群 | 办公套件 MCP |
| 发邮件 | 邮件通知 | 办公套件 MCP |
27.2.2 MCP Server：办公套件接入（第 7 章）
示例代码 ：以下代码演示核心结构，省略了异常处理、日志和完整 import。
from fastmcp import FastMCP
mcp = FastMCP("office-suite")
@mcp.tool()
def create_report_doc(title: str, content: str) -> str:
"""创建文档并返回链接。生成报告/文档时使用。"""
doc_id = office_api.create_doc(title, content)
return f"文档已创建：https://docs.internal/{doc_id}"
@mcp.tool()
def send_wecom_message(channel: str, content: str) -> dict:
"""发送企业微信消息。任务完成/异常通知时使用。"""
return wecom_api.send(channel, content)
@mcp.tool()
def send_email(to: str, subject: str, body: str) -> dict:
"""发送邮件。需要正式通知时使用。注意：敏感操作，需确认。"""
return email_api.send(to, subject, body)
注意 ：send_email 是敏感操作（第 5/22 章）——MCP Server 内部要做确认机制。
27.3 成本优化与异常恢复
27.3.1 成本优化（第 23 章五个金矿）
| 优化 | 做法 | 收益 |
|---|---|---|
| 结果缓存 | 相同数据查询命中缓存 | 数据拉取零成本 |
| 小模型路由 | 数据清洗/格式整理走小模型 | 简单步骤降本 60% |
| 并行分析 | 趋势/渠道分析并行 | 时间减半 |
| 上下文精简 | 分析只传必要指标 | 省输入 token |
| 预算封顶 | 单任务费用上限 | 防失控 |
27.3.2 异常恢复：失败自愈（第 21/23 章）
自动化任务最怕"半夜挂了没人管"。设计三级自愈：
图 2：三级自愈机制
def run_with_recovery(plan, budget):
for step in plan["steps"]:
try:
result = execute_step(step)
except DataSourceError:
result = retry(step, times=2, backoff=[5, 15])
except DataMissingError as e:
result = {"skipped": True, "reason": str(e)}
notify("周报数据缺失", f"{step} 数据源不可用，已跳过")
except Exception as e:
result = fallback(step, e)
escalate_to_human(step, e)
results.append(result)
return results
自愈分级原则 ：可重试的重试 → 可跳过的跳过（带通知）→ 兜不住的转人工。自动化不等于无人值守，而是"能自愈的自愈，自愈不了的通知人"。
27.3.3 预算控制（第 23 章三层熔断）
BUDGET = {
"per_task": {"max_cost": 0.5, "max_steps": 20},
"per_month": {"max_cost": 30, "alarm": 0.8},
}
27.4 全流程编排与监控
27.4.1 编排器：定时触发 + 状态管理
from apscheduler.schedulers.blocking import BlockingScheduler
def weekly_report_job():
plan = planner.plan("生成本周经营周报")
results = run_branches(plan, parallel_groups=[["分析趋势","分析渠道"]])
report = synthesize(results)
ok, issues = quality_check(report)
report = report if ok else revise(report, issues)
doc_url = create_report_doc("第X周经营周报", report)
send_wecom_message("周报群", f"周报已生成：{doc_url}")
save_trace("weekly_report", results, doc_url)
scheduler = BlockingScheduler()
scheduler.add_job(weekly_report_job, "cron", day_of_week="mon", hour="9")
scheduler.start()
图 3：自动化编排监控
27.4.2 监控面板（第 20/21 章）
| 指标 | 告警阈值 | 说明 |
|---|---|---|
| 任务成功率 | < 90% | 周报是否稳定跑通 |
| 平均成本 | > 0.3 元/次 | 成本是否异常 |
| 失败原因 | 同一原因 > 2 次 | 需要修复而非重试 |
| 生成报告质量 | 抽评 < 4 分 | LLM-as-Judge |
27.4.3 数据质量校验（防"自动生成垃圾"）
自动化最大的隐藏风险：流程跑通了，但数据错了，还自动发出去了 。必须在发送前加数据校验：
图 4：发送前数据校验
def data_sanity_check(report):
checks = [
report["revenue"] >= 0,
abs(report["revenue"] - sum(report["items"])) < 0.01,
report["orders"] > 0,
len(report["channels"]) >= 3,
]
return all(checks)
铁律：自动化任务发送给"人看"的内容，发送前必须过数据校验 + 质检（第 17 章）——宁可拦下误报，不可发出错误。
🛠 解决方案：自动化任务失败自愈机制 + 成本预算控制
常见问题
- "半夜任务挂了没人管" ：无自愈。对策：三级自愈（重试→跳过带通知→转人工，27.3.2）。
- "自动发出了错误报表" ：无数据校验。对策：发送前 data_sanity_check + 质检（27.4.3）。
- "成本一个月比一个月高" ：无预算熔断。对策：三层熔断 + 成本监控（27.3.3）。
- "任务偶尔超时" ：数据量大。对策：并行化 + 大任务拆分 + 超时分层。
- "改了一版，流程直接跑不通" ：无回归。对策：每周跑一次"干跑"验证（不真发），用评测集回归（第 20 章）。
解决方案速查表
| 现象 | 根因 | 解决方案 |
|---|---|---|
| 挂了没人管 | 无自愈 | 三级自愈 |
| 发出错误数据 | 无校验 | 发送前校验 + 质检 |
| 成本失控 | 无预算 | 三层熔断 |
| 任务超时 | 串行/超量 | 并行 + 拆分 |
| 改动跑不通 | 无回归 | 干跑验证 |
实战提示
- 自动化 ≠ 无人值守 ：能自愈的自愈，自愈不了的通知人。
- 发送前必须校验 ：给"人看"的内容，数据校验 + 质检缺一不可。
- 三级自愈是标配 ：重试 → 跳过（通知）→ 转人工。
- 预算熔断必须有 ：自动化任务没有预算上限就是定时炸弹。
- 干跑回归 ：改流程后先"干跑"（不真发）验证，再让它自动跑。
English / 原文
第 27 章 案例三：自动化工作流 Agent
本章要解决的问题
每周要花 3 小时的报表生成，交给 Agent 自动跑——规划、多智能体、成本优化怎么配合？
章节大纲
- 27.1 多步骤任务拆解（规划 + 多智能体）
- 27.2 MCP 集成办公套件（豆包）
- 27.3 成本优化与异常恢复
- 27.4 全流程编排与监控
- 🛠 解决方案：自动化任务失败自愈机制 + 成本预算控制
27.1 场景与任务拆解
27.1.1 场景：周报自动生成
业务 ：某运营团队每周要生成一份经营周报，人工耗时 3 小时：
拉数据（订单/流量/转化）→ 清洗 → 分析趋势 → 撰写报告 → 发邮件/同步群
交给 Agent 的目标 ：全自动跑，出错自愈，成本可控。
27.1.2 任务拆解（第 15 章规划）
[Planner] 生成周报任务计划
├─ 步骤1：拉取本周订单数据（工具：数据查询）
├─ 步骤2：拉取本周流量数据（工具：数据查询）
├─ 步骤3：分析销售趋势（步骤1/2 后）
├─ 步骤4：分析渠道表现（步骤1/2 后）
├─ 步骤5：撰写周报（步骤3/4 后）
└─ 步骤6：发送周报（步骤5 后）
依赖关系：1,2 → 3,4 → 5 → 6
图 1：周报任务依赖图
两个独立分支 （步骤 3 和 4 都依赖 1、2 但不互相依赖）→ 可并行 （第 12 章）。
27.1.3 为什么用规划 + 多智能体，而不是一条提示链
| 需求 | 方案 | 理由 |
|---|---|---|
| 任务结构固定 | 提示链即可（第 10 章） | 步骤写死 |
| 但数据源可能变 | 需要规划的动态性（第 15 章） | 本周数据源缺一个要重排 |
| 分析任务较独立 | 可并行 + 角色分工 | 多智能体（第 16 章） |
本案例用"规划 + 并行 + 主管编排" ：Planner 出计划 → 主管派活 → 分析类任务并行 → 汇总质检 → 发送。
27.2 MCP 集成办公套件
27.2.1 需要的工具集
| 工具 | 用途 | 来源 |
|---|---|---|
| 查订单数据 | 拉取销售数据 | 内部数据 API |
| 查流量数据 | 拉取流量数据 | 内部数据 API |
| 写文档 | 生成周报文档 | 办公套件 MCP |
| 发消息 | 同步到群 | 办公套件 MCP |
| 发邮件 | 邮件通知 | 办公套件 MCP |
27.2.2 MCP Server：办公套件接入（第 7 章）
示例代码 ：以下代码演示核心结构，省略了异常处理、日志和完整 import。
from fastmcp import FastMCP
mcp = FastMCP("office-suite")
@mcp.tool()
def create_report_doc(title: str, content: str) -> str:
"""创建文档并返回链接。生成报告/文档时使用。"""
doc_id = office_api.create_doc(title, content)
return f"文档已创建：https://docs.internal/{doc_id}"
@mcp.tool()
def send_wecom_message(channel: str, content: str) -> dict:
"""发送企业微信消息。任务完成/异常通知时使用。"""
return wecom_api.send(channel, content)
@mcp.tool()
def send_email(to: str, subject: str, body: str) -> dict:
"""发送邮件。需要正式通知时使用。注意：敏感操作，需确认。"""
return email_api.send(to, subject, body)
注意 ：send_email 是敏感操作（第 5/22 章）——MCP Server 内部要做确认机制。
27.3 成本优化与异常恢复
27.3.1 成本优化（第 23 章五个金矿）
| 优化 | 做法 | 收益 |
|---|---|---|
| 结果缓存 | 相同数据查询命中缓存 | 数据拉取零成本 |
| 小模型路由 | 数据清洗/格式整理走小模型 | 简单步骤降本 60% |
| 并行分析 | 趋势/渠道分析并行 | 时间减半 |
| 上下文精简 | 分析只传必要指标 | 省输入 token |
| 预算封顶 | 单任务费用上限 | 防失控 |
27.3.2 异常恢复：失败自愈（第 21/23 章）
自动化任务最怕"半夜挂了没人管"。设计三级自愈：
图 2：三级自愈机制
def run_with_recovery(plan, budget):
for step in plan["steps"]:
try:
result = execute_step(step)
except DataSourceError:
result = retry(step, times=2, backoff=[5, 15])
except DataMissingError as e:
result = {"skipped": True, "reason": str(e)}
notify("周报数据缺失", f"{step} 数据源不可用，已跳过")
except Exception as e:
result = fallback(step, e)
escalate_to_human(step, e)
results.append(result)
return results
自愈分级原则 ：可重试的重试 → 可跳过的跳过（带通知）→ 兜不住的转人工。自动化不等于无人值守，而是"能自愈的自愈，自愈不了的通知人"。
27.3.3 预算控制（第 23 章三层熔断）
BUDGET = {
"per_task": {"max_cost": 0.5, "max_steps": 20},
"per_month": {"max_cost": 30, "alarm": 0.8},
}
27.4 全流程编排与监控
27.4.1 编排器：定时触发 + 状态管理
from apscheduler.schedulers.blocking import BlockingScheduler
def weekly_report_job():
plan = planner.plan("生成本周经营周报")
results = run_branches(plan, parallel_groups=[["分析趋势","分析渠道"]])
report = synthesize(results)
ok, issues = quality_check(report)
report = report if ok else revise(report, issues)
doc_url = create_report_doc("第X周经营周报", report)
send_wecom_message("周报群", f"周报已生成：{doc_url}")
save_trace("weekly_report", results, doc_url)
scheduler = BlockingScheduler()
scheduler.add_job(weekly_report_job, "cron", day_of_week="mon", hour="9")
scheduler.start()
图 3：自动化编排监控
27.4.2 监控面板（第 20/21 章）
| 指标 | 告警阈值 | 说明 |
|---|---|---|
| 任务成功率 | < 90% | 周报是否稳定跑通 |
| 平均成本 | > 0.3 元/次 | 成本是否异常 |
| 失败原因 | 同一原因 > 2 次 | 需要修复而非重试 |
| 生成报告质量 | 抽评 < 4 分 | LLM-as-Judge |
27.4.3 数据质量校验（防"自动生成垃圾"）
自动化最大的隐藏风险：流程跑通了，但数据错了，还自动发出去了 。必须在发送前加数据校验：
图 4：发送前数据校验
def data_sanity_check(report):
checks = [
report["revenue"] >= 0,
abs(report["revenue"] - sum(report["items"])) < 0.01,
report["orders"] > 0,
len(report["channels"]) >= 3,
]
return all(checks)
铁律：自动化任务发送给"人看"的内容，发送前必须过数据校验 + 质检（第 17 章）——宁可拦下误报，不可发出错误。
🛠 解决方案：自动化任务失败自愈机制 + 成本预算控制
常见问题
- "半夜任务挂了没人管" ：无自愈。对策：三级自愈（重试→跳过带通知→转人工，27.3.2）。
- "自动发出了错误报表" ：无数据校验。对策：发送前 data_sanity_check + 质检（27.4.3）。
- "成本一个月比一个月高" ：无预算熔断。对策：三层熔断 + 成本监控（27.3.3）。
- "任务偶尔超时" ：数据量大。对策：并行化 + 大任务拆分 + 超时分层。
- "改了一版，流程直接跑不通" ：无回归。对策：每周跑一次"干跑"验证（不真发），用评测集回归（第 20 章）。
解决方案速查表
| 现象 | 根因 | 解决方案 |
|---|---|---|
| 挂了没人管 | 无自愈 | 三级自愈 |
| 发出错误数据 | 无校验 | 发送前校验 + 质检 |
| 成本失控 | 无预算 | 三层熔断 |
| 任务超时 | 串行/超量 | 并行 + 拆分 |
| 改动跑不通 | 无回归 | 干跑验证 |
实战提示
- 自动化 ≠ 无人值守 ：能自愈的自愈，自愈不了的通知人。
- 发送前必须校验 ：给"人看"的内容，数据校验 + 质检缺一不可。
- 三级自愈是标配 ：重试 → 跳过（通知）→ 转人工。
- 预算熔断必须有 ：自动化任务没有预算上限就是定时炸弹。
- 干跑回归 ：改流程后先"干跑"（不真发）验证，再让它自动跑。