RPA Coze · 企业级 AI 运维驱动的分布式 RPA 平台

Coze + LangGraph + FastAPI + WebSocket + pywinauto + Playwright + Docker + 阿里云搭的企业级 RPA 平台。聊天框一句话触发月末对账，Master 智能调度多台 Worker 并发执行浪潮 GS ERP 与 Web 自动化，AIOps 子系统 (LangGraph Plan-Execute-Replan + MCP + 双层故障知识库) 自动诊断并闭环修复，结果文件自动回推到对话流。

七件事

一个平台干七件事：

1. 对话式任务触发 — Coze AI Bot 作统一入口，自然语言映射到 40+ RPA 任务（“导出 3 月经销商对账” → task_id + params）。自定义 HTTP Tools 通过 Function Calling 调 Master REST API；Prompt JSON Schema + 后端 task_mapping.json 双层校验拦截 LLM 幻觉；Coze 代理支持 SSE 流式转发、multipart 上传、conversation_id 自动提取保存。

2. 分布式任务调度 — Master-Worker 架构，单 Master 同时维护 N 路 WebSocket 双向 + M 路 SSE 单向连接。SQLAlchemy 2.0 async + asyncpg (PostgreSQL) + Redis (hiredis) 分布式锁，Repository 仓储模式 + Alembic 迁移；运行时自动 fallback 到 SQLite + 内存缓存保障单机兜底。按负载加权择优分配 (心跳活跃度 + 历史任务数)，覆盖幂等去重 / 心跳熔断 / 失败重试 / 任务中止 / 实时广播全套生命周期。

3. 企业 ERP + Web 自动化 — 基于 pywinauto (UIA backend) 对浪潮 GS (.NET WinForms MDI) 实现深度控制；基于 Playwright 实现抖音链客等浏览器场景；pytesseract + Pillow 做图像/OCR 兜底；PyInstaller 打包 exe + PyQt5 沙盘监控悬浮窗 UI。40+ 业务流程：科目余额表、明细账、经销商对账、汇率云下载、跨公司报告、银行对账、费用/凭证批量入账等。

4. 文件全链路回流 — Worker → Master 分块上传 (SHA-256 校验 + 临时文件原子 rename)，Master /files/{id}/download 对外直出，任务完成时遍历 summary.result_files 拼 Markdown 多链接推回 Coze 会话，Nginx 层 client_max_body_size=200M 支持大文件。

5. 全栈监控与可观测性 — 内置暗色主题监控面板 dashboard.html (5 Tab: 概览 / Worker / 历史 / 异常 / 日志)，HTTPBasic 鉴权；Worker 健康检查后台守护线程 (60s 心跳超时熔断 + 任务自动重派)；资源监控 (CPU / 内存 / 磁盘心跳上报)；事件告警 (severity 分级，error/critical 触发 alert)；日志聚合 (任务日志 / 失败步骤 / Worker 事件三源)；SSE 实时推送 + 任务运维 (暂停 / 恢复 / 中止)；零外部依赖 (无 Prometheus / Grafana / ELK)。

6. AI 运维助手 + 自愈测试闭环 — 独立子系统 rpa-aiops/（基于 SuperBizAgent 扩展）：① RAG 知识库问答 (Milvus + qwen-embedding-v4 + langchain-text-splitters)；② AIOps 自动故障诊断 (LangGraph 1.1 Plan-Execute-Replan StateGraph + MemorySaver 多轮隔离)；③ MCP 协议 (fastmcp 3.2 + langchain-mcp-adapters 0.2 + mcp 1.27) 接入 3 个工具服务 (日志 / 监控 / RPA)；④ 双层故障知识库召回 (Planner 起步 hint 主动避坑 + Replanner 失败信号案例召回) 实现"执行→失败→召回→修复→沉淀"自学习闭环；⑤ APScheduler 3.11 定时自动测试 + LangSmith LLM 追踪；⑥ RPA 自动测试 MCP server 可反向驱动 RPA Master 跑 E2E 验证修复效果。

7. Worker OTA 热更新与多节点治理 — Master 侧 update_manager.py + /updates 路由 + updates.html 管理台，Worker 侧 core/update_manager.py 接收升级包 + watchdog.py 看门狗 + 接收 restart 信号自重启；主仓 (Gitee monorepo) → 副本仓 → 阿里云生产节点 → 客户内网节点四地同步 SCP + Git 脚本化，部署 30min → 3min。

系统架构

RPA 平台 (多仓库 · 三个独立模块)
│
├── rpa-master/                      # 调度中枢 + 前端入口（阿里云 Docker）
│   ├── master/
│   │   ├── main.py                  # FastAPI + lifespan + CORS + Coze 代理 + 注册路由
│   │   ├── security.py              # HTTPBasic Dashboard 鉴权
│   │   ├── database.py              # legacy SQLite 兼容
│   │   ├── db/                      # SQLAlchemy 2.0 async
│   │   │   ├── session.py           # asyncpg 连接池
│   │   │   ├── models.py            # ORM 模型
│   │   │   ├── repositories/        # 仓储模式 (tasks.py / workers.py)
│   │   │   └── migrations/          # Alembic 迁移
│   │   ├── cache/redis_client.py    # Redis 分布式锁 + 缓存
│   │   ├── routers/                 # REST + WS + SSE 路由
│   │   │   ├── tasks.py             # /tasks/create | /complete | /history | /overview | /errors
│   │   │   ├── workers.py           # /workers + 注册 + 心跳 + /workers/events
│   │   │   ├── files.py             # /files/upload | /files/{id}/download
│   │   │   ├── websocket.py         # /ws/workers/{id} 长连接 + ws_manager
│   │   │   ├── events.py            # /events SSE 广播
│   │   │   └── updates.py           # /updates OTA 包管理
│   │   ├── services/
│   │   │   ├── task_manager.py      # 任务状态机 + Coze 推送 + 择优分配
│   │   │   ├── worker_manager.py    # Worker 注册 / 心跳检查守护线程 / 释放
│   │   │   ├── file_manager.py      # 分块上传 + SHA-256 + 原子 rename
│   │   │   ├── logger_manager.py    # 结构化任务日志
│   │   │   ├── update_manager.py    # OTA 包构建 / 分发 / 回滚
│   │   │   ├── event_hub.py         # SSE 跨线程事件广播 (bind_loop)
│   │   │   └── shared_state.py      # COZE_API_KEY / latest_conversation_id
│   │   ├── models/schemas.py        # Pydantic 请求/响应模型
│   │   ├── static/
│   │   │   ├── dashboard.html       # 监控面板 (暗色主题 5 Tab)
│   │   │   └── updates.html         # OTA 管理台
│   │   ├── task_mapping.json        # 任务 → required_param + default_params Schema
│   │   ├── storage/, uploads/, logs/
│   │   └── requirements.txt
│   ├── shared/
│   │   └── log_config.py            # Master/Worker 共用日志配置
│   ├── coze-chat.html               # 单页聊天 UI (Coze 嵌入 + Worker 选择 + SSE)
│   ├── Dockerfile                   # python:3.11-slim-bullseye + 阿里云 pypi + 非 root rpa
│   ├── docker-compose.yml           # master + nginx + healthcheck + certbot 预留
│   ├── ops/nginx/docker-rpa.conf    # 反代 + WS upgrade + SSE 无缓冲 + 3600s 超时
│   └── .env / .env.example
│
├── rpa-worker/                      # 执行端 (客户 Windows 机器)
│   ├── RPAWorker.py                 # PyInstaller 打包入口 (sys._MEIPASS 感知)
│   ├── RPAWorker.spec               # PyInstaller spec
│   ├── worker_v2/
│   │   ├── main.py                  # argparse + signal handler + 自重启循环
│   │   ├── config/
│   │   │   ├── settings.py          # WorkerConfig.from_env
│   │   │   └── endpoints.py         # Master URL 统一配置
│   │   ├── core/                    # 核心组件
│   │   │   ├── worker_core.py       # 协调器
│   │   │   ├── ws_client.py         # WebSocket 客户端 + 自动重连
│   │   │   ├── task_runner.py       # 任务执行器 (18KB 大模块)
│   │   │   ├── task_context.py      # abort / checkpoint 协议
│   │   │   ├── update_manager.py    # Worker 侧 OTA 接收 + 解包 (19KB)
│   │   │   ├── watchdog.py          # 看门狗自愈
│   │   │   ├── state_store.py       # 本地 SQLite 状态
│   │   │   └── event_queue.py       # Master ← Worker 事件队列
│   │   ├── apps/
│   │   │   ├── base/
│   │   │   │   ├── base_app.py
│   │   │   │   ├── voucher_flow_base.py
│   │   │   │   └── auto_discover.py # 应用自动发现
│   │   │   ├── flows/               # 40+ 业务流程
│   │   │   │   ├── flow_bank_reconciliation.py
│   │   │   │   ├── flow_bank_add_subject.py
│   │   │   │   ├── flow_batch_run.py                        # 批量执行 11 类
│   │   │   │   ├── flow_construction_income/payment/receipt.py
│   │   │   │   ├── flow_contract_payment.py
│   │   │   │   ├── flow_cost_invoice_commission/issue/receive.py
│   │   │   │   ├── flow_data_download_summary.py            # 科目余额批量
│   │   │   │   ├── flow_deposit_payment.py
│   │   │   │   ├── flow_dept_expense_summary.py
│   │   │   │   ├── flow_dept_welfare_expense.py
│   │   │   │   ├── flow_distributor_reconciliation.py       # 经销商对账
│   │   │   │   ├── flow_douyin_linke.py                     # ⭐ Playwright Web 自动化
│   │   │   │   ├── flow_entertainment_expense.py
│   │   │   │   ├── flow_fund_transfer.py
│   │   │   │   ├── flow_huilvyun_download.py                # 汇率云
│   │   │   │   ├── flow_intercompany_report.py              # 跨公司报告
│   │   │   │   ├── flow_mgmt_only_smoke.py                  # 冒烟测试
│   │   │   │   ├── flow_non_contract_payment.py
│   │   │   │   ├── flow_operation_agency/other/payment.py
│   │   │   │   ├── flow_other_expense.py
│   │   │   │   ├── flow_purchase_payment.py
│   │   │   │   ├── flow_quarterly_interest.py               # 季度利息
│   │   │   │   ├── flow_reconciliation_check.py
│   │   │   │   ├── flow_reserve_fund_application/verification.py
│   │   │   │   ├── flow_revenue_cost_report.py
│   │   │   │   ├── flow_salary_voucher.py
│   │   │   │   ├── flow_scenic_spot_voucher.py              # 景区入账
│   │   │   │   ├── flow_social_security_voucher.py          # 社保凭证
│   │   │   │   ├── flow_subsidiary_transfer.py
│   │   │   │   ├── flow_tax_declaration_draft.py            # 纳税申报草稿
│   │   │   │   └── flow_traffic_expense.py
│   │   │   └── flow_batch_runner.py                         # 批量执行器
│   │   ├── components/
│   │   ├── ui/                      # PyQt5 沙盘监控悬浮窗
│   │   └── requirements.txt
│   ├── rpa_core/
│   │   ├── components/langchao/     # 浪潮 GS 控件级原子操作
│   │   │   ├── component_subject_project_balance_export.py  # 陕文投项目余额表
│   │   │   ├── interactive_balance_export.py                # 通用辅助余额表
│   │   │   ├── window_utils.py                              # MDI 子窗 / tab 管理
│   │   │   └── ...
│   │   └── requirements.txt         # pywinauto + playwright + OCR + Excel
│   ├── tests/                       # 17 个测试文件
│   └── installer/                   # NSIS 安装包脚本
│
└── rpa-aiops/                       # ⭐ AI 运维助手子系统 (端口 :9900)
    ├── app/
    │   ├── main.py                  # FastAPI + Milvus 启动自检
    │   ├── config.py                # DashScope + Milvus + MCP (3 server) + RPA Master URL
    │   ├── api/
    │   │   ├── chat.py              # /chat RAG 对话 (流式 SSE)
    │   │   ├── aiops.py             # /aiops 故障诊断 (流式 SSE)
    │   │   ├── file.py              # /file/upload 知识库文档上传
    │   │   ├── fault_kb.py          # ⭐ /fault_kb 故障知识库管理
    │   │   └── schedule.py          # 定时任务管理
    │   ├── agent/
    │   │   ├── mcp_client.py        # MCP 多服务器单例 + 重试拦截器
    │   │   └── aiops/               # Plan-Execute-Replan StateGraph
    │   │       ├── planner.py       # ⭐ 双 RAG (经验文档 + 故障 hint)
    │   │       ├── executor.py      # LangGraph ToolNode 自动执行
    │   │       ├── replanner.py     # continue/replan/respond + 案例召回
    │   │       ├── state.py         # PlanExecuteState TypedDict
    │   │       └── utils.py         # Tools 描述格式化
    │   ├── services/
    │   │   ├── aiops_service.py     # StateGraph 编译 + SSE 事件格式化
    │   │   ├── rag_agent_service.py # create_agent + MemorySaver
    │   │   ├── vector_index_service.py     # 文档切分 + 入库
    │   │   ├── vector_store_manager.py     # Milvus 单例 + Top-K 检索
    │   │   ├── fault_kb_recall.py          # ⭐ 故障知识库双层召回
    │   │   └── scheduler_service.py        # APScheduler 定时任务
    │   ├── tools/
    │   │   ├── knowledge_tool.py    # @tool retrieve_knowledge (RAG 检索)
    │   │   └── time_tool.py         # @tool get_current_time
    │   ├── models/                  # Pydantic 数据模型
    │   ├── core/                    # LLM 工厂 + Milvus 客户端
    │   └── utils/
    ├── mcp_servers/
    │   ├── cls_server.py            # 日志查询 MCP (:8003)
    │   ├── monitor_server.py        # 监控数据 MCP (:8004)
    │   └── rpa_server.py            # ⭐ RPA 自动测试 MCP (:8005, 新增)
    ├── test_cases/
    │   └── safe_subset.yaml         # 定时测试用例
    ├── pyproject.toml               # super-biz-agent-py v1.2.1 (fork)
    ├── requirements.txt             # LangChain 1.2 + LangGraph 1.1 + MCP 1.27
    └── start-windows.bat            # 一键拉 Milvus + 3 MCP + FastAPI

部署拓扑

                               公网 ingress
                                    │
                              ┌─────┴──────┐
                              │  阿里云 ECS │
                              │ 123.56.118.83│
                              └─────┬──────┘
                                    │
                     ┌──────────────┴──────────────┐
                     │  Docker Compose              │
                     │  ┌────────────┐  ┌────────┐ │
                     │  │ nginx 1.25 │─►│ master │ │
                     │  │  :80 / :443│  │ :9000  │ │
                     │  │  upstream  │  │ python │ │
                     │  └────────────┘  │  3.11  │ │
                     │    │             └────────┘ │
                     │    └── /ws/  → WS upgrade   │
                     │        /events/ → SSE       │
                     │        /  → HTTP            │
                     │  私网: rpa-net (bridge)     │
                     └─────────────────────────────┘
                             ▲              ▲
                             │              │
                    WS(/ws/workers/{id})   HTTP/SSE
                             │              │
            ┌────────────────┴──┐    ┌──────┴──────┐
            │                   │    │             │
       ┌────┴─────┐       ┌─────┴────┐   ┌─────────┴──┐
       │ Worker 1 │       │ Worker N │   │ Frontend   │
       │ Win 10/11│  ...  │ Win 10/11│   │ 浏览器     │
       │ pywinauto│       │ pywinauto│   │ coze-chat  │
       │ Playwright│      │          │   │ dashboard  │
       │ PyQt5 UI │       │          │   └────────────┘
       └──────────┘       └──────────┘

  单独部署:
       ┌──────────────────────────┐
       │  rpa-aiops 独立节点       │
       │  FastAPI :9900           │
       │  Milvus Docker :19530    │
       │  MCP ×3  :8003/4/5       │
       │  HTTP 调 Master API      │
       │  (rpa_master_url 配置)    │
       └──────────────────────────┘

任务生命周期 (状态机)

                用户聊天框输入自然语言
                         │
                         ▼
                ┌──────────────────┐
                │ Coze Tool 调用    │
                │ /tasks/create    │
                └────────┬─────────┘
                         │ task_id 幂等检查 + Schema 校验
                         ▼
                ┌──────────────────┐
                │   QUEUED         │ DB 唯一索引 + asyncio.Lock 双保护
                └────────┬─────────┘
                         │ 择优 Worker (心跳活跃 + 负载最低)
                         ▼
                ┌──────────────────┐
                │   ASSIGNED       │ WS 下发 + Worker 状态 BUSY
                └────────┬─────────┘
                         │ Worker ack
                         ▼
                ┌──────────────────┐
                │   RUNNING        │ pywinauto/Playwright 执行 + 心跳保活
                └────┬─────────┬───┘
                     │         │ 心跳 60s 超时
                     │         ▼
                     │  ┌──────────────┐
                     │  │   TIMEOUT    │ → Worker OFFLINE 软下线
                     │  └──────┬───────┘   任务回 QUEUED 重新分配
                     │         │
                     │  ┌──────────────┐
                     │  │   ABORTED    │ ← 用户中止 / 任务取消
                     │  └──────────────┘
                     ▼
            ┌────────────┬──────────────┐
            ▼            ▼              ▼
       ┌─────────┐ ┌────────────┐ ┌───────────┐
       │  DONE   │ │   FAILED   │ │   RETRY   │
       └────┬────┘ └─────┬──────┘ └─────┬─────┘
            │            │              │
            │ summary.result_files      │ 失败次数 < N
            ▼            ▼              │
   ┌─────────────────────────────┐     │
   │ Coze /v3/chat/messages       │ ◄──┘ 回 QUEUED
   │ Markdown 多文件下载链接       │
   └─────────────────────────────┘

AI 对话流 (Tool-use 编排)

用户: "帮我导出 3 月份经销商对账"
    │
    ▼
Coze Bot (底座 DeepSeek / Doubao / 豆包)
    │ ① NLU 意图识别 → distributor_reconciliation
    │ ② Slot Filling  → start_date=2026-03-01, end_date=2026-03-31
    │ ③ Tool 选择     → create_task
    ▼
自定义 HTTP Tool (Coze Plugin)
    │ POST /tasks/create
    │ { task_id, user_id, params: {...} }
    ▼
Master 双层 Schema 校验
    │ ① Coze 端 JSON Schema (前置约束)
    │ ② 后端 task_mapping.json required_param 兜底
    │ → 拦截 LLM 幻觉/缺参/枚举越界请求
    ▼
入调度引擎 → Worker 执行 → 产出文件
    │
    ▼ 任务完成
Master → Coze /v3/chat/messages (异步主动推送)
    │ 遍历 summary.result_files
    │ 拼装 Markdown:
    │   📎 结果文件（3 个）：
    │     • 对账_2026-03_完整.xlsx：http://.../files/abc/download
    │     • 对账_2026-03_异常.xlsx：http://.../files/def/download
    │     • 对账_2026-03_汇总.csv ：http://.../files/ghi/download
    ▼
用户在聊天框直接点链接下载（无需登录任何系统）

本质是 Single-step Tool-use Agent，Coze 负责 NLU + Tool 路由 + Memory，Master 负责后端任务级编排。MVP 阶段省 80% Agent 框架开发量，后期可平滑迁移 LangGraph（AIOps 子系统已经在用）。

实时通信子系统

                    Master (FastAPI + asyncio 单实例)
                    ┌──────────────────────────────┐
                    │  lifespan async context mgr   │
                    │  ┌────────────────────────┐   │
                    │  │ httpx.AsyncClient      │   │ (5min timeout, connect 10s)
                    │  │ (全局 Coze/HTTP 客户端) │   │
                    │  ├────────────────────────┤   │
                    │  │ WS Connection Manager  │   │ ←── N 路 Worker (双向)
                    │  │  (心跳 + 重连)          │   │
                    │  ├────────────────────────┤   │
                    │  │ SSE Broadcaster        │   │ ←── M 路前端 (单向)
                    │  │  (event_hub.bind_loop) │   │
                    │  │  跨线程 asyncio.run_   │   │
                    │  │  coroutine_threadsafe  │   │
                    │  ├────────────────────────┤   │
                    │  │ Task Queue             │   │
                    │  │ (asyncio.Queue + Lock) │   │
                    │  └────────────────────────┘   │
                    └──────────────────────────────┘
                          ▲              ▲
                          │              │
          ┌───────────────┘              └────────────┐
          │                                            │
   Nginx: /ws/ upgrade                          Nginx: /events/ no-buffer
   proxy_read_timeout 3600s                     chunked_transfer_encoding on
          │                                            │
   ┌──────┴──────┐                              ┌──────┴──────┐
   │ Worker × N  │                              │ Frontend ×M │
   │ (Windows)   │                              │ (浏览器)    │
   │ ws_client.py│                              │ EventSource │
   │ watchdog.py │                              │ (auto retry)│
   └─────────────┘                              └─────────────┘

Nginx 关键配置:
  - /ws/   : proxy_http_version 1.1 + Upgrade + Connection upgrade + 3600s
  - /events/ : proxy_buffering off + proxy_cache off + chunked_transfer_encoding on
  - /    : 默认 HTTP 代理, client_max_body_size 200M
  - keepalive 32 (upstream 连接复用)

容灾覆盖：
  - Worker 离线        → 心跳 60s 超时 → 软下线 + 任务回队列
  - 客户端断连          → SSE 自动重连 + last-event-id 续传
  - Master 滚动重启     → Worker WS 自动重连 + 任务状态从 DB 恢复
  - Coze API 超时/限流   → token 守卫 + 503 友好降级 + 不阻塞业务

监控与可观测性

完整的"指标采集 → 健康检查 → 事件告警 → 可视化"闭环，全自研零外部依赖：

监控面板 (/dashboard, HTTPBasic 鉴权)

┌───────────────────────────────────────────────────────────────────┐
│  ● RPA Master 监控面板                     更新于 12:34:56  [刷新] │
├───────────────────────────────────────────────────────────────────┤
│  ┌──────┐ ┌──────┐ ┌──────────┐ ┌──────────┐ ┌──────┐ ┌──────┐    │
│  │等待中│ │执行中│ │ 今日完成 │ │在线 Worker│ │异常任务│ │ 存储 │    │
│  │  3   │ │  2   │ │   18     │ │   5/7    │ │  1   │ │ 2.3G │    │
│  └──────┘ └──────┘ └──────────┘ └──────────┘ └──────┘ └──────┘    │
│                                                                   │
│  [概览] [Worker] [历史] [异常] [日志]                               │
│                                                                   │
│  执行中的任务                                                      │
│  ┌─────────┬──────────────┬────────┬────────┬──────┬──────┬────┐ │
│  │ 执行ID  │ 任务         │ Worker │当前步骤 │ 进度 │ 耗时 │操作 │ │
│  │ a3f2... │ 经销商对账    │ FIN-01 │ 导出 3/5│ ███60%│02:13│⏸✕ │ │
│  └─────────┴──────────────┴────────┴────────┴──────┴──────┴────┘ │
└───────────────────────────────────────────────────────────────────┘

• 5 Tab：概览 (执行中 + 队列) / Worker (卡片网格) / 历史 / 异常 / 日志
• 5s 自动刷新 (setInterval) + 手动刷新
• 任务运维：/tasks/{id}/pause|resume|abort 一键操作
• 下载：/files/{id}/download 直达

Worker 健康检查 (daemon thread)

     每 30s 一轮 (daemon, 启动时自启)
             │
             ▼
 ┌──────────────────────────┐
 │ 遍历所有非 OFFLINE Worker │
 └──────────────┬───────────┘
                │
                ▼
 ┌──────────────────────────────────┐
 │ now - last_heartbeat > 60s ?     │
 └──────────────┬───────────────────┘
      │YES                  │NO
      ▼                      ▼
 ┌────────────────┐   ┌───────────────┐
 │ 标 OFFLINE     │   │ 检查空闲 Worker│
 │ 记录持有任务   │   │ 是否有待派任务 │
 └───────┬────────┘   └───────┬───────┘
         │                     │
         ▼                     ▼
 ┌────────────────┐   ┌───────────────┐
 │ 自动 complete  │   │ 触发 _try_    │
 │ task = error + │   │ assign_pending│
 │ "机器人好像断了"│   │ _tasks()      │
 └────────────────┘   └───────────────┘

心跳协议 (资源监控)

POST /workers/{worker_id}/heartbeat
{
  "status":         "idle" | "busy" | "paused" | "error",
  "current_task":   "execution_id_or_null",
  "cpu_percent":    23.5,        # psutil
  "memory_percent": 41.2,
  "disk_free_gb":   128.7
}

事件告警

POST /workers/events
{
  "worker_id": "FIN-01",
  "event_type": "login_failed",
  "severity": "error",     # info | warning | error | critical
  "data": { "retry_count": 3, "screenshot": "..." }
}
        │
        ▼
 全部入库 → severity ∈ {error,critical} → _trigger_alert()

日志聚合 (三源统一)

• 任务日志 GET /logs/tasks?execution_id=xxx&limit=100
• 失败步骤 GET /logs/tasks/failed?limit=100
• Worker 事件 GET /workers/events?limit=100

→ 全链路 trace：一个 execution_id 串起 task → step → log → event

AIOps 自动运维子系统 (rpa-aiops 端口 :9900)

把 SuperBizAgent 形态的"RAG + AIOps"内嵌到 RPA 平台，运维场景与业务执行打通。

双能力定位

┌──────────────────────────────────────────┐
│   POST /chat         →   RAG 对话问答     │
│   POST /aiops        →   AIOps 自动诊断   │
│   POST /file/upload  →   知识库文档入库   │
│   /fault_kb/*        →   故障知识库管理   │
│   /schedule/*        →   定时任务管理     │
└──────────────────────────────────────────┘
                共享:
   Milvus (向量库) + DashScope (qwen-max LLM)
   + MCP 工具 (日志/监控/RPA) + 故障知识库

LangGraph 状态机：Plan-Execute-Replan

                  ┌──────────────┐
                  │   START      │
                  └──────┬───────┘
                         ▼
              ┌──────────────────────┐
              │      Planner         │ ◄─── ⭐ 双 RAG 召回
              │  • 经验文档 (RAG)     │      (任务描述 + 故障 hint)
              │  • 故障 hint (KB)    │
              │  → 4-6 步骤 plan     │
              └──────────┬───────────┘
                         ▼
              ┌──────────────────────┐
              │      Executor        │ ◄─── 调 MCP 工具
              │  ToolNode 自动执行    │      (cls/monitor/rpa)
              │  • plan[0] 出队       │
              │  • past_steps 入队   │
              └──────────┬───────────┘
                         ▼
              ┌──────────────────────┐
              │     Replanner        │ ◄─── ⭐ 故障案例召回
              │  评估 → 三选一:       │      (失败信号 → KB)
              │  • continue          │
              │  • replan (新步骤)    │
              │  • respond (报告)     │
              └──┬──────────────┬────┘
                 │              │
        continue/replan       respond
                 ▼              ▼
             Executor       ┌────────┐
              (循环)         │  END   │ → SSE 推送最终报告
                            └────────┘

全局护栏:
  - past_steps >= 8 → 强制 respond，避免无限循环
  - past_steps >= 5 → 禁止 replan，只能 respond
  - replan 新步骤数 ≤ 当前剩余步骤数
  - plan 失败/异常 → 安全降级到默认 3 步计划

MCP (Model Context Protocol) 工具层

MultiServerMCPClient (fastmcp 3.2 + langchain-mcp-adapters 0.2 + mcp 1.27)
   ├── cls server      → 日志查询          (:8003)
   ├── monitor server  → 监控数据          (:8004)
   └── rpa server      → ⭐ RPA 自动测试    (:8005, 新增)

- streamable-http transport
- 每个 server 独立进程，可按 enable 开关组装
- retry_interceptor 全局拦截，失败自动重试
- LangGraph ToolNode 自动把工具结果回喂 LLM

⭐ 双层故障知识库召回（本项目对 SuperBizAgent 的扩展）

┌─────────────────────────────────────────────────────────┐
│ Phase 3-d: Planner 起步阶段 — "潜在风险案例"召回           │
│   query = 用户任务描述本身                                 │
│   recall_by_query() top_k=2                              │
│   → 让 LLM 制定 plan 时主动避坑                           │
│   → 写入 state.fault_kb_hints                            │
└─────────────────────────────────────────────────────────┘
                              ▼
                            Executor
                              ▼
┌─────────────────────────────────────────────────────────┐
│ Phase 3-b: Replanner 失败评估阶段 — "历史失败案例"召回     │
│   query = past_steps 末尾的失败信号                        │
│   recall_fault_cases()                                   │
│   → 给 LLM 看相似失败的修复经验                           │
│   → 写入 state.fault_kb_recalled                         │
└─────────────────────────────────────────────────────────┘

自学习闭环:
 执行 → 失败 → 沉淀为案例 → 下次类似任务 Planner 起步就召回 → 主动规避

容错: 任一召回异常都安全降级为 []，不阻断主流程

Phase 3-c：APScheduler 定时自动测试

APScheduler 3.11 cron (默认 0 3 * * * Asia/Shanghai)
     │
     ▼ 拉 test_cases/safe_subset.yaml
     │
     ▼ AIOps Agent 跑测试用例
     │
     ├─ 巡检验证: 验证 MCP / LLM 可用性
     ├─ 修复回归: 验证诊断建议修复效果
     └─ 全链路冒烟: 通过 mcp_rpa 触发 RPA Master 跑真任务

与 RPA 主平台的集成（闭环关键）

① config.py: rpa_master_url = http://123.56.118.83
   AIOps 直接 HTTP 调 RPA Master /tasks/history, /tasks/errors
   → 故障知识库"案例素材"来自 RPA 真实失败任务记录

② mcp_rpa_url = http://localhost:8005/mcp
   AIOps LLM 可以"调一个 MCP 工具就触发 RPA 任务"
   → 闭环: 诊断完 → 调 RPA → 验证修复 → 沉淀新案例

③ 共享 RAG 知识库
   RPA 业务文档 (科目余额表 SOP / 浪潮 GS 操作手册) 入 Milvus
   → 财务用户"3 月对账怎么操作" 直接命中知识库回答
   → 同一份知识库同时被业务问答和故障诊断复用

OTA 更新与多节点治理

        ┌──────────────────────────────────────┐
        │  开发机 (本地)                        │
        │  git commit → Gitee monorepo (主仓) │
        └────────────────┬─────────────────────┘
                         │
                         ▼
        ┌──────────────────────────────────────┐
        │  Gitee 副本仓 (备份 / 降级用)         │
        └────────────────┬─────────────────────┘
                         │
                         ▼  git pull (两路径)
        ┌────────────────┴────────────────────┐
        │                                     │
        ▼                                     ▼
 ┌─────────────┐                       ┌─────────────┐
 │ 阿里云生产   │                       │ 客户内网节点 │
 │ docker      │                       │ (裸跑)       │
 │ compose up  │                       │ RDP 人工    │
 │ --build     │                       │ 拉包        │
 └──────┬──────┘                       └──────┬──────┘
        │                                      │
        │ Master API /updates 分发             │
        ▼                                      ▼
 ┌───────────────────────────────────────────────────┐
 │ Worker × N (Windows 业务机)                       │
 │   • core/update_manager.py 接收 zip 升级包         │
 │   • 校验哈希 + 解压到 update/ 目录                 │
 │   • watchdog.py 看守 + 请求重启                    │
 │   • RPAWorker.py 捕获 restart_request 自重启       │
 │   • PyInstaller 打包新版 exe 替换老版             │
 └───────────────────────────────────────────────────┘

部署效率提升: 30 min 手工 → 3 min 脚本化

ERP + Web 自动化关键技术

浪潮 GS 是 .NET WinForms MDI 应用；抖音链客是纯 Web 应用。两套栈共存的实战坑：

浪潮 GS (pywinauto UIA backend)

• MDI 子窗作用域隔离 — “科目余额表-结果页” / “辅助余额表-结果页” 可能共存，必须精确锁定子窗搜 DataItem；Pass 1 精确匹配关键字，Pass 2 回退到任意结果页
• Runtime 空表检测 — 进"明细账"前先查 row 0 的"本期借/贷方"是否有非零值，无值则跳过明细避免 _right_click_first_data_row 报错
• 多路读取容错 — 同一 cell 同时尝试 window_text / legacy.Value / children.text / iface_value 四路，任一非零即认定有值
• 业务异常 vs 系统异常解耦 — 自定义 _SubjectNoDataInBalance 表示"本月无数据"，主循环捕获作为 skip 不污染失败统计；区别于 _SubjectExportAborted 中止信号
• 黑名单 + Runtime 双层防御 — skip_detail_codes 配置兜底 runtime 检测漏网场景
• 嵌套 tab 精确关闭 — _close_nested_result_tabs(levels=N) 按层数精确关闭
• 可中止协议 — 每个原子操作前 _check_abort(stop_check)，abort 事件经 WS 下发秒级响应

抖音链客 (Playwright)

• 无头浏览器隔离 — 每个 Worker 独立 Playwright context，cookie / localStorage 隔离
• 登录态持久化 — 首次扫码登录后 storage_state 持久化，后续任务复用
• 网络层拦截 — Playwright route() 拦截 API 请求，绕过 UI 翻页直接抓数据

OCR 兜底

• pytesseract + Pillow — pywinauto 读不到的老控件（如 GDI 绘制的非标控件），截屏走 OCR 识别

完整技术栈清单（版本 pinned）

rpa-master (Python 3.11)

类别	依赖	用途
Web 框架	fastapi ≥ 0.115, uvicorn[standard] ≥ 0.32	异步 Web + WS + SSE
数据模型	pydantic ≥ 2.10	请求/响应校验
数据库	SQLAlchemy[asyncio] ≥ 2.0.30, asyncpg ≥ 0.29, alembic ≥ 1.13	PG 异步 + 迁移
缓存	redis[hiredis] ≥ 5.0	分布式锁 + 缓存
HTTP	httpx ≥ 0.24, requests ≥ 2.31	Coze 代理 + 外部 API
文件上传	python-multipart ≥ 0.0.6	Worker 上传
配置	python-dotenv ≥ 1.0	.env 加载
时区	python-dateutil ≥ 2.8	ISO 时间处理

rpa-worker (worker_v2 + rpa_core)

类别	依赖	用途
Master 通信	websocket-client ≥ 1.6, requests ≥ 2.31	WS 长连接 + HTTP fallback
进程/资源	psutil ≥ 5.9	CPU/内存/磁盘心跳
Worker GUI	PyQt5 ≥ 5.15	沙盘监控悬浮窗
Excel 处理	openpyxl ≥ 3.1, xlrd ≥ 2.0, xlsxwriter ≥ 3.1, pandas ≥ 2.0	结果文件生成
桌面自动化	pywinauto ≥ 0.6.8, pyautogui, pywin32, pyperclip	浪潮 GS UIA
浏览器自动化	playwright ≥ 1.40	抖音链客等 Web
图像/OCR	pillow ≥ 9.0, pytesseract ≥ 0.3.10	非标控件识别
打包	PyInstaller	Worker exe

rpa-aiops (Python ≥ 3.11, <3.14)

类别	依赖 (pinned)	用途
Web	fastapi 0.136, uvicorn 0.46, sse-starlette 3.4	流式 SSE
LLM 框架	langchain 1.2.17, langchain-core 1.3.2, langgraph 1.1.10, langgraph-checkpoint 4.0.3, langgraph-prebuilt 1.0.13	状态机编排
LLM 底座	langchain-qwq 0.3.4, dashscope 1.25.17, openai 2.33	通义千问 qwen-max
向量库	pymilvus 2.6.12, langchain-milvus 0.3.3	RAG 检索
文档切分	langchain-text-splitters 1.1	知识库入库
工具协议	fastmcp 3.2.4, langchain-mcp-adapters 0.2.2, mcp 1.27.0	MCP server 对接
定时任务	APScheduler 3.11.2	定时自动测试
认证	Authlib 1.7, PyJWT 2.12, cryptography 47	OAuth / JWT
追踪	langsmith 0.8, loguru 0.7	LLM trace + 结构化日志
Token	tiktoken 0.12	token 计算
DB	SQLAlchemy 2.0.49	故障知识库存储
测试用例	PyYAML 6.0	safe_subset.yaml 加载
Excel	xlrd 2.0	.xls 读取 (verify_mgmt_reports)
Fork	super-biz-agent-py 1.2.1	基础 Agent 框架

部署与运维

类别	工具	说明
容器	Docker 20.10+, docker compose v2	master + nginx
基础镜像	python:3.11-slim-bullseye	491MB
Web 服务器	nginx 1.25-alpine	反代 + WS upgrade + SSE no-buffer
Pypi 源	阿里云 (主) + 清华 (fallback)	中国境内稳定
Debian 源	清华	APT 下载加速
云服务	阿里云 ECS	123.56.118.83
证书	certbot (预留)	HTTPS 启用后用
版本控制	Gitee (主仓 + 副本仓)	四节点同步

一键启动

Master 端（阿里云 / 自建服务器）

cd rpa-master
docker compose up -d              # 起 master + nginx，30s healthcheck
docker compose logs -f master     # 看启动日志
curl http://localhost/health      # {"status":"ok","service":"rpa-master"}

Worker 端（Windows 业务机）

# 开发模式（直跑 Python）
python rpa-worker\RPAWorker.py --master-host 123.56.118.83 --worker-id win-fin-01

# 生产模式（PyInstaller 打包后免装 Python）
.\RPAWorker.exe --worker-id win-fin-01

# 无头服务器（不起 PyQt5 悬浮窗）
.\RPAWorker.exe --worker-id headless-01 --no-ui

前端入口

• 聊天 UI：http://123.56.118.83/chat
• 任务面板：http://123.56.118.83/dashboard (HTTPBasic)
• OTA 管理：http://123.56.118.83/updates-ui (HTTPBasic)
• OpenAPI：http://123.56.118.83/docs

服务端口：:80（Nginx 反代）+ :9000（Master 直连，兼容硬编码 v1 Worker）

AIOps 子系统

cd rpa-aiops
.\start-windows.bat     # 自动拉 Milvus Docker + 3 MCP + FastAPI + 文档入库

# 验证
curl http://localhost:9900/health

# RAG 对话 (流式)
curl -N -X POST http://localhost:9900/api/chat \
  -H "Content-Type: application/json" \
  -d '{"message":"经销商对账怎么操作", "session_id":"test"}'

# AIOps 自动诊断 (流式 SSE)
curl -N -X POST http://localhost:9900/api/aiops \
  -H "Content-Type: application/json" \
  -d '{"session_id":"diag-001"}'

AIOps 端口：:9900 (FastAPI) + :19530 (Milvus) + :8003/8004/8005 (3 个 MCP server)

踩过的坑与解决方案

坑	根因	解决
Docker apt update 失败	Docker 20.10.9 + debian bookworm overlay2 bug	换 bullseye + APT::Update::Post-Invoke-Success::=""
镜像 pypi 超时	清华源对 mako/alembic 偶尔 timeout	阿里云主 + 清华 extra-index-url
Docker compose v2 缺失	系统 docker 自带版不支持 plugin	通过 ghfast.top 下载 v2 二进制到 ~/.docker/cli-plugins/
Coze API 空 token 异常	httpx 对 "Bearer " 抛 illegal header	/coze-proxy 入口加 COZE_API_KEY 空值守卫 → 503
Worker 任务卡死	pywinauto MDI 多 tab 跨页串扰	子窗严格 scope 隔离 + 精确关键字锁定
陕文投明细页崩溃	结果页 row 0 无值但强行进明细	Runtime 检测本期借/贷方 + 跳过明细
LLM Plan 无限循环	Replanner 不断 replan	全局护栏: past_steps >= 8 强制 respond
前端 offline Worker 展示	副本 HTML 未同步	bind-mount + 多地同步脚本
ExceptionGroup 吞异常	Python 3.11 MCP 客户端异常组	_log_exception_details 展开 exc.exceptions

业务价值

• 月末关账：财务月末 ERP 手工操作 X 小时 → 自动化 Y 分钟
• 7×24 无人值守：员工下班/离职不影响任务继续运行
• 对话降门槛：非技术人员在聊天框一句话触发，无需懂 ERP 菜单
• 故障自愈：AIOps 自动诊断 + 沉淀知识库，减少运维人工介入
• 规模：N 台 Worker / M 类业务流程 / K 家公司覆盖
• 可靠性：任务成功率 > 99.X%，端到端 P99 延迟 < Y s

— 本项目介绍结束 —