面向非技术用户的端到端日报自动化系统 —— 丢入数据源,自动生成日报,飞书推送到群。
这是一个 AI Agent + Workflow 混合架构 的日报自动化框架。它通过 LLM 驱动的数据分析师 Agent 理解 Excel 面板结构,利用插件化管线自动完成数据粘贴、公式刷新和日报生成,最后通过飞书 Webhook 推送分析结果。
日常使用只需两步:把 CSV 数据源丢进文件夹 → 双击运行 → 确认 → 完成。
| 场景 | 入口 | 频率 | 说明 |
|---|---|---|---|
| AI 入职培训 | onboard.py |
一次性 | 自动扫描 Excel 结构 → Agent 理解面板逻辑 → 对话确认 → 生成管线配置 |
| 日常工作流 | main.py |
每天 | 沙箱安全预览 → 自动诊断 → 确认后正式运行 → 飞书通知 |
| Chat BI | chat.py |
随时 | 自然语言数据查询,Agent 自动理解意图并分析数据 |
- Agent + Workflow 混合架构:ReAct 循环驱动的智能决策 + 固定管线的高效执行,通过文件系统握手解耦
- 沙箱安全机制:正式运行前先在临时副本上试跑,AI 分析和飞书发送均被拦截,确认后才操作真实文件
- 智能粘贴匹配:三层匹配策略(表头 Jaccard 相似度 → 文件名 glob → 数据形态),自动将 CSV 数据写入 Excel 对应 Sheet
- Agent 自诊断:管线出错时自动收集诊断信息,Agent 交叉排查配置、数据源和 Excel 结构,定位根因并修复
- 插件化设计:管线步骤可插拔,只需继承
BasePlugin并实现run()方法 - 多面板独立管理:每个面板独立目录、独立管线、独立配置,互不干扰
- 支持 DeepSeek / 阿里百炼双 LLM 提供商,自动检测切换
- win32com 真实刷新 Excel 公式 + openpyxl 读取数值(解决公式不计算问题)
- 三层记忆管理(会话热记忆 / 面板温记忆 / 运行冷记忆),不依赖向量数据库
- 全中文交互,Windows 终端 UTF-8 兼容
新面板首次接入时运行,只需一次。用户将 Excel 和数据源样表放入 training_incoming/<面板名>/,运行后 Agent 自动理解面板结构并生成配置。
Phase 1 — 自动扫描(零交互)
系统自动打开 Excel,完成以下扫描:
| 扫描项 | 内容 |
|---|---|
| Sheet 结构 | 所有 Sheet 的名称、行列数 |
| 公式解析 | INDEX / MATCH / VLOOKUP / SUM / IF 等公式类型识别 |
| 公式链拓扑 | Sheet 间引用关系图(谁引用谁) |
| 列级映射 | 每列引用了哪个 Sheet 的哪列 |
| 表头提取 | 每个 Sheet 前 3 行的文字内容 |
| 公式密度 | 每个 Sheet 的公式占非空单元格比例 |
| 自动检查 | #REF! 错误、空 Sheet、孤立 Sheet、分类矛盾 |
Phase 1.5 — 用户选表
系统列出候选监控表(公式密度 > 5% 或有引用关系的 Sheet),按公式密度降序排列。用户输入编号选择关注的监控表后,系统自动追溯公式链,过滤出相关 Sheet,忽略无关的旧版或废弃 Sheet。
[1] 门店目标达成(总) | 200行x15列 | 公式密度 42% | 引用 3 个数据源
[2] 管理层绩效看板(南区) | 150行x12列 | 公式密度 35% | 引用 2 个数据源
请选择监控表(多选用逗号分隔,如 1,2): 1
Phase 2 — Agent 理解与对话
Agent 获取过滤后的数据,通过 ReAct 循环逐步理解每个 Sheet 的角色:
Agent: 分析完成,以下是我的理解:
[监控表] 门店目标达成(总) — 公式密度 42%,引用 3 个数据源,表头含"目标""实际""达成率"
[数据源] 每日运营数据源 — 公式密度 2%,被监控表引用,表头为原始业务字段
[数据源] LeadsBy店 — 公式密度 1%,被监控表引用
是否有需要调整的?
用户: LeadsBy店 实际也是数据源吗?
Agent: 是的,它公式密度极低(1%),被监控表引用,表头是原始线索字段 → 确认为数据源 ✓
用户: 好,没问题了,保存吧
Agent: [调用 write_file] pipeline.json 已保存 ✓
understanding_report.md 已保存 ✓
入职培训完成!
产出物:
experiments/<面板名>/pipeline.json— 3 步管线配置experiments/<面板名>/understanding_report.md— 面板理解报告
每天将新数据 CSV 丢入 incoming/,运行 python main.py。
Step 0: 门禁检查
├── 检查 pipeline.json 是否存在(不存在 → 提示先运行入职培训 → 退出)
├── 校验管线结构完整性
└── 检查 监控面板.xlsx + incoming/ 目录存在
Step 1: 沙箱预演
├── 复制 监控面板.xlsx 到临时目录 sandbox_work/
├── 执行 3 步管线(全部操作临时副本):
│ ① excel_paster: 扫描 incoming/ → 三层匹配 → 写入沙箱副本
│ ② excel_refresh_reader: win32com 刷新公式 → openpyxl 读取数值
│ ③ ai_analysis_sender: 生成 mock 摘要 + 拦截飞书发送
├── 导出 _context.json(每步状态 + 耗时 + 中间数据)
└── 展示沙箱摘要(粘贴结果 + AI 分析预览 + 飞书消息预览)
Step 2: 确认正式运行
├── 展示三维度摘要
├── 用户输入 y → 进入 Step 3
├── 用户输入 n → 退出(不写任何真实文件)
└── 用户描述不满 → Agent 对话调整配置 → 重新沙箱
Step 3: 正式运行
├── 操作真实 Excel 文件
├── 真实 LLM 分析 + 飞书发送
└── 展示最终结果
沙箱 vs 正式对比:
| 沙箱模式 | 正式模式 | |
|---|---|---|
| 操作文件 | 临时副本(sandbox_work/) | 监控面板.xlsx(真实文件) |
| AI 分析 | mock 摘要(不调 LLM) | 真实 LLM 调用 |
| 飞书发送 | 拦截,记录但不发 | 真实发送 |
| 沙箱副本 | y 确认后自动删除 | 无 |
| 用户确认 | 需要(Step 2) | 无需(已确认过) |
三种运行模式:
| 命令 | 行为 |
|---|---|
python main.py |
列出所有面板 → 交互选择 → 沙箱 → 确认 → 正式 |
python main.py pipeline.json |
指定面板 → 沙箱 → 确认 → 正式 |
python main.py pipeline.json --live |
跳过沙箱,直接正式运行(定时任务用) |
随时通过自然语言查询面板数据。
启动: python chat.py <面板名>
Agent 自动加载上下文:
├── understanding_report.md → Sheet 角色和指标含义
├── pipeline.json → 数据源 ↔ Sheet 映射关系
├── 监控面板.xlsx 基本结构 → 有哪些 Sheet
└── incoming/ 文件列表 → 原始数据源一览
对话示例:
用户: 广州南区新机会量前 3 的店铺
Agent: [理解:指标=新机会量,维度=广州南区,排序=降序取前3]
[调用 read_excel(监控面板.xlsx, sheet="运营总览By店")]
广州南区新机会量前 3 的店铺:
| 店铺 | 新机会量 |
|--------|---------|
| 白云店 | 2,156 |
| 天河店 | 1,234 |
| 越秀店 | 987 |
用户: 白云店的线索转化率呢?
Agent: [理解:它=白云店,上一轮上下文连贯]
[计算:线索转化率 = 试驾数 / 线索量]
...
白云店线索转化率:34.2%(线索量 6,305 → 试驾数 2,156)
约束:只读,不写任何文件,不执行管线,不发飞书。
- Python 3.10+
- Windows(win32com 依赖 Microsoft Excel)
- LLM API Key(DeepSeek 或阿里百炼,二选一)
cd daily_report_automation
pip install -r requirements.txt在 daily_report_automation/.env 中配置:
# LLM(二选一)
DEEPSEEK_API_KEY=sk-xxxxx # DeepSeek 直连
# 或
DASHSCOPE_API_KEY=sk-xxxxx # 阿里百炼
# 通知
FEISHU_WEBHOOK_URL=https://open.feishu.cn/open-apis/bot/v2/hook/xxxxx# 第一步:入职培训(新面板,仅需一次)
python onboard.py
# 第二步:日常工作流(每天)
python main.py # 交互选择面板
python main.py pipeline.json # 指定面板
python main.py pipeline.json --live # 定时任务模式
# 随时:数据查询
python chat.py <面板名>
# 辅助:诊断修复
python -m framework.diagnose <面板名>也可以双击根目录的 .bat 文件一键启动。
AI自动化工作流/
├── README.md
├── .gitignore
├── AI入职培训.bat # 一键启动入职培训
├── ChatBI.bat # 一键启动 Chat BI
├── 工作流启动.bat # 一键启动日常管线
│
└── daily_report_automation/
├── requirements.txt
├── onboard.py # 场景 A:AI 入职培训
├── main.py # 场景 B:日常工作流
├── chat.py # 场景 C:Chat BI
│
├── framework/ # 核心框架
│ ├── engine.py # PipelineEngine:管线执行引擎
│ ├── agent.py # UniversalAgent:ReAct 工具调用循环
│ ├── llm_client.py # LLM 统一调用入口
│ ├── plugin_base.py # BasePlugin + StepResult
│ ├── sandbox.py # 沙箱安全机制
│ ├── config.py # 管线加载与校验
│ ├── tools.py # Agent 工具集(6 个工具)
│ ├── prompts.py # Prompt 工程(角色底座 + 场景指令)
│ ├── scan_excel.py # Excel 结构扫描
│ ├── diagnose.py # Agent 诊断修复入口
│ └── logger.py # 结构化日志 + 自动清理
│
├── plugins/ # 管线插件
│ ├── excel_paster.py # ① 扫描 + 匹配 + 粘贴
│ ├── excel_refresh_reader.py # ② win32com 刷新 + 读取
│ └── ai_analysis_sender.py # ③ AI 分析 + 飞书发送
│
├── onboarding/ # 入职培训交互
│ ├── display.py # 终端 UI 渲染
│ └── verify.py # 7 项自动检查
│
├── tests/ # 单元测试
│
├── experiments/ # 已配置的面板(场景 A 产出)
├── test_data/ # 测试数据
└── training_incoming/ # 新面板入职材料
每个面板一个独立目录:
experiments/<面板名>/
├── 监控面板.xlsx ← Excel 模板(含公式和格式)
├── pipeline.json ← 3 步管线配置
├── incoming/ ← 每日数据源投放目录
├── understanding_report.md ← AI 理解报告
└── logs/ ← 运行日志 + context 导出
{
"name": "<面板名>日报",
"version": 2,
"excel_path": "experiments/<面板名>/监控面板.xlsx",
"steps": [
{
"id": "step1_paste",
"name": "粘贴数据",
"plugin": "excel_paster",
"config": {
"excel_path": "...", // Excel 路径
"source_dir": "experiments/<面板名>/incoming/",
"match_by_header": true, // 启用表头匹配
"min_similarity": 0.4, // Jaccard 最低相似度
"source_sheets": ["Sheet1"], // 数据源 Sheet 列表
"patterns": { // 文件名 glob 规则
"Sheet1": {"match": ["*关键词*"]}
},
"paste_regions": { // 各 Sheet 粘贴区域
"Sheet1": {
"sheet_name": "Sheet1",
"start_row": 2, // 数据起始行
"start_col": 1, // 数据起始列
"preserve_formula_cols": [], // 保留公式的列
"skip_header_rows": 1, // 跳过表头行数
"clear_before": true // 粘贴前清空旧数据
}
}
}
},
{
"id": "step2_refresh_read",
"name": "刷新公式并读取",
"plugin": "excel_refresh_reader",
"config": {
"excel_path": "...",
"timeout_seconds": 30, // 刷新超时
"sheets_to_read": [
{"sheet_name": "监控表名", "range": "A1:Z99"}
]
}
},
{
"id": "step3_analyze_send",
"name": "AI 分析并发送",
"plugin": "ai_analysis_sender",
"config": {
"model": "deepseek-v3", // LLM 模型
"focus_metrics": ["指标1","指标2"], // 关注的核心指标
"webhook_url_env": "FEISHU_WEBHOOK_URL",
"title": "<面板名>日报"
}
}
],
"error_handling": {
"retry": {
"max_retries": 2, // 全局重试次数
"interval_seconds": 30 // 重试间隔
},
"skip_and_alert": ["step1_paste"], // 出错不重试、记录后继续
"pause_and_notify": [] // 出错立即停止管线
}
}
step1_paste建议列入skip_and_alert:粘贴失败不阻塞后续刷新和分析步骤。
Agent 通过 ReAct 循环自主决定何时调用哪个工具,最多 10 轮。6 个工具如下:
| 工具 | 入职 | 诊断 | Chat BI | 作用 | 限制 |
|---|---|---|---|---|---|
scan_excel |
✅ | ✅ | ❌ | 扫描 Excel 结构(Sheet 分类 + 公式链 + 列映射 + 问题检测) | 仅结构,不读数据 |
read_excel |
✅ | ✅ | ✅ | 读取 Excel 指定 Sheet 和范围的单元格数据 | 结果超 4000 字符截断 |
read_file |
✅ | ✅ | ✅ | 读取任意文本文件(CSV / JSON / MD / 配置) | 结果超 4000 字符截断 |
write_file |
✅ | ✅ | ❌ | 写入或覆盖文件,保存 pipeline.json 和 understanding_report.md | 路径限制在项目目录内,不能写 .env |
Bash(sandbox) |
❌ | ✅ | ❌ | 沙箱验证修复,运行 python main.py 沙箱模式 |
不可加 --live,超时 120s |
search_files |
✅ | ✅ | ❌ | 在项目文件中搜索关键词 | 只读 |
管线步骤由插件实现。所有插件继承 BasePlugin,位于 plugins/ 目录。
BasePlugin 接口:
class BasePlugin:
def __init__(self, config: dict):
self.config = config
def run(self, context: dict) -> StepResult:
"""正式模式执行。必须实现。"""
raise NotImplementedError
def dry_run(self, context: dict) -> StepResult:
"""沙箱模式执行。默认等同 run(),有副作用的插件应覆写。"""
return self.run(context)StepResult 结构:
@dataclass
class StepResult:
step_id: str # 步骤 ID(如 "step1_paste")
step_name: str # 步骤名称
status: str # "ok" | "skipped" | "error"
data: dict # 传给下游的数据
error: str # 失败原因(status=error 时)
duration_ms: int # 执行耗时(引擎自动填充)
meta: dict # 额外元信息最小示例 — 创建一个新插件:
from framework.plugin_base import BasePlugin, StepResult
class MyPlugin(BasePlugin):
def run(self, context: dict) -> StepResult:
# 从 context 取上游数据
upstream = context.get("step1_paste", {})
# 执行逻辑
result = do_something(upstream)
return StepResult(
step_id="my_step",
step_name="我的插件",
status="ok",
data={"output": result},
)
def dry_run(self, context: dict) -> StepResult:
# 沙箱模式:跳过副作用(如发送通知)
return StepResult(
step_id="my_step",
step_name="我的插件",
status="ok",
data={"output": "[沙箱] 已拦截"},
meta={"sandbox": True},
)注册插件(在 main.py 的 PLUGIN_REGISTRY 中):
from plugins.my_plugin import MyPlugin
PLUGIN_REGISTRY = {
"excel_paster": ExcelPaster,
"excel_refresh_reader": ExcelRefreshReader,
"ai_analysis_sender": AIAnalysisSender,
"my_plugin": MyPlugin, # 新增
}context 数据传递约定:
context["step_id"]和context["plugin_name"]均可取上游输出- 引擎自动注入
context["sandbox_mode"](bool)和context["pipeline_name"] - 插件自行约定输出 key,下游按 key 读取
两个引擎互不调用、互不依赖,通过文件系统握手机制解耦:
Agent 引擎 Workflow 引擎
┌──────────────┐ ┌──────────────┐
│ ReAct 循环 │ │ PipelineEngine│
│ LLM 决策驱动 │ │ 代码步骤驱动 │
│ 灵活·有上下文 │ │ 固定·可重复 │
└──────┬───────┘ └──────┬───────┘
│ │
│ 写入文件 │ 写入文件
│ pipeline.json │ *_context.json
│ understanding_report.md │ 监控面板.xlsx(更新后)
│ │
└──────── 文件系统(互相读取)────────┘
| Agent 引擎 | Workflow 引擎 | |
|---|---|---|
| 执行模型 | ReAct 循环(思考→行动→观察→再思考) | 按序执行(遍历 steps 数组) |
| 决策方式 | LLM 驱动,自主选择工具 | 代码固定逻辑 |
| 适用场景 | 入职理解、诊断修复、数据问答 | 数据粘贴、公式刷新、日报生成 |
| LLM 调用 | 多轮(最多 10 轮),tools 模式 | 单次(system + user),纯文本 |
| temperature | 0.1(推理稳定) | 0.3(生成变化) |
Agent ReAct 循环核心逻辑(framework/agent.py):
Agent.chat(user_message):
① messages.append(user_message)
② while 未到上限(10轮):
response = call_llm(messages, tools=TOOL_DEF)
if 是文本回复 → 返回给用户
if 是工具调用 → 执行工具 → 结果追加到 messages → 继续循环
③ return "达到最大轮次"
常见轮次:Chat BI 1-3 轮,诊断 2-5 轮,入职 3-6 轮。
沙箱通过三步保证"先试跑、再执行":
Sandbox.setup()
→ 复制 监控面板.xlsx 到 sandbox_work/<uuid>/
→ 替换各步骤 config 中的 excel_path 为副本路径
PipelineEngine.run(steps, sandbox_mode=True)
→ 每个步骤调用 plugin.dry_run() 而非 plugin.run()
→ excel_paster: 写入沙箱副本
→ excel_refresh_reader: 刷新沙箱副本
→ ai_analysis_sender: 返回 mock 摘要 + 拦截飞书(meta.sandbox=True)
用户确认 y
→ Sandbox.cleanup() 删除副本
→ PipelineEngine.run(steps, sandbox_mode=False) 操作真实文件
| 文件 | 写入方 | 读取方 | 说明 |
|---|---|---|---|
pipeline.json |
Agent(入职时) | Workflow(每次运行)、Agent(诊断/ChatBI 时) | 配置快照,Agent 产出 → Workflow 消费 |
understanding_report.md |
Agent(入职时) | Agent(诊断/ChatBI 时) | 面板知识,Agent 自己消费 |
*_context.json |
Workflow(执行导出) | Agent(诊断时) | 运行记录,Workflow 产出 → Agent 消费 |
diagnostic_package.json |
Workflow(出错时) | Agent(诊断时) | 诊断包 |
监控面板.xlsx |
Workflow(更新数据) | Agent(ChatBI 查询时) | Workflow 更新 → Agent 读取 |
incoming/ 数据源 |
用户(投放) | Workflow(粘贴)、Agent(ChatBI 查询) | 共享输入 |
fixes.jsonl |
Agent(修复时) | Agent(诊断时) | 修复历史 |
不依赖向量数据库,纯文件系统管理:
第 1 层:会话内(热记忆)
存储内容: messages 列表(当前对话完整历史)
存储位置: 内存
生命周期: 会话结束即丢弃
使用场景: Agent 多轮对话上下文连贯
第 2 层:面板级(温记忆)
存储内容: understanding_report.md + pipeline.json
存储位置: experiments/<面板名>/
生命周期: 永久
使用场景: Agent 启动时自动读取,了解面板知识和配置
第 3 层:运行级(冷记忆)
存储内容: *_context.json + diagnostic_package.json + fixes.jsonl
存储位置: experiments/<面板名>/logs/
生命周期: 日志保留最近 20 个,自动清理旧日志
使用场景: Agent 诊断时注入摘要,不作为 messages 注入
┌─────────────────────────────────────┐
│ 第一层:角色底座(共享,约 200 token)│
├─────────────────────────────────────┤
│ 第二层:场景指令(入口装配,三选一) │
│ 含:流程引导 + CoT + Few-shot + 约束 │
├─────────────────────────────────────┤
│ 第三层:上下文注入(动态,从磁盘读取)│
└─────────────────────────────────────┘
| 场景 | 角色底座 | CoT | Few-shot | ToT(思维树) | 硬性束缚 |
|---|---|---|---|---|---|
| 入职培训 | ✅ | ✅ 引导 | ✅ 2 例 | — | ❌ |
| 诊断修复 | ✅ | ✅ 引导 | — | ✅ 三路径并行 | ❌ |
| Chat BI | ✅ | ✅ 引导 | ✅ 1 例 | — | ❌ |
| 日报生成 | — | — | — | — | ❌ |
| 操作 | 限制 |
|---|---|
write_file 工具 |
路径限制在项目根目录内,禁止写入 .env 和系统文件 |
Bash(sandbox) 工具 |
只能运行沙箱模式,禁止 --live,超时 120s |
| 只读工具 | 结果超 4000 字符自动截断,防止 token 爆炸 |
| 管线正式运行 | 必须经过沙箱预览 → 用户确认后才能执行 |
| 策略 | 配置项 | 触发条件 | 行为 |
|---|---|---|---|
| 自动重试 | retry |
全局,所有步骤出错 | 等 N 秒后重试,最多 M 次 |
| 跳过继续 | skip_and_alert |
列表中的 step 出错 | 不重试,记录 error,继续执行后续步骤 |
| 暂停管线 | pause_and_notify |
列表中的 step 出错 | 立即 raise RuntimeError,管线停止 |
配置示例:
"error_handling": {
"retry": { "max_retries": 2, "interval_seconds": 30 },
"skip_and_alert": ["step1_paste"],
"pause_and_notify": []
}建议将
step1_paste列入skip_and_alert:数据源为空或匹配失败时,粘贴出错不应阻塞后续的刷新和分析。
管线沙箱出错时,Agent 自动介入,采用思维树(ToT)三路径并行快速评估:
诊断流程:
① 快速评估(三路径并行判断优先级)
A. 配置写错了?
快速检查: pipeline.json 与 understanding_report.md 一致吗?
→ 不一致 → 高优先级
B. 数据源变了?
快速检查: incoming/ 文件名能匹配 patterns 吗?表头对得上吗?
→ 匹配不上 → 高优先级
C. Excel 结构被改了?
快速检查: scan_excel 结果和 understanding_report.md 一致吗?
→ 一致 → 基本排除
② 深入排查
选择高优先级路径深入,不需要三条全跑
③ 输出诊断报告
1. 发现了哪些问题
2. 每个问题的根因
3. 具体修复方案(哪个字段、当前值→新值)
④ 用户确认修复 → Agent 执行 → 重新沙箱验证
诊断入口:
# 交互模式(沙箱出错时 main.py 内触发)
python -m framework.diagnose <面板名>
# 也可以独立运行,对已有面板进行诊断| 症状 | 可能原因 | 检查方法 |
|---|---|---|
| 粘贴步骤报错 | 数据源文件名变了,pattern 匹配不上 | 查看 incoming/ 文件名,对比 pipeline.json 的 patterns 配置 |
| 粘贴步骤报错 | 表头变了,Jaccard 相似度低于阈值 | 检查 CSV 表头是否与 Excel 对应 Sheet 表头匹配 |
| 读取结果为空 | Excel 公式缓存过期 | 确保 excel_refresh_reader 步骤在粘贴之后执行 |
| AI 分析不准确 | 指标配置有误 | 检查 focus_metrics 是否包含正确的指标名 |
| Chat BI 查询不到数据 | incoming/ 为空或 监控面板.xlsx 未更新 | 先运行一次 main.py 更新面板数据 |
| 启动报错"请先运行入职培训" | pipeline.json 不存在 | 先运行 python onboard.py 完成入职培训 |
系统设计了严格的运行顺序,确保数据流完整:
入职培训 (onboard.py)
↓ 产出 pipeline.json + understanding_report.md
↓
日常管线 (main.py) ← 门禁检查 pipeline.json 存在
Chat BI (chat.py) ← 门禁检查 pipeline.json 存在
如果跳过入职培训直接运行 main.py 或 chat.py,系统会提示先运行入职培训并退出。
- 语言: Python 3.10+
- LLM: DeepSeek / 阿里百炼(OpenAI 兼容 API)
- Excel: openpyxl(读写)+ win32com(公式刷新)
- 通知: 飞书 Webhook
- 架构: ReAct Agent + Pipeline Workflow 混合模式
MIT