标签：AI Agent | RPA 智能体 | MCP 协议 | 自动化编排 | 本地大模型 | 认知自动化 | 内网离线 --- 有几年自动化经验的人都会发现，一个越来越明显的趋势：**规则驱动的 RPA 正在遇到天花板**。 传统的 RPA 工具靠录制和固定规则运行，页面结构一变就崩溃，遇到非结构化数据直接抓瞎。去年给一个客户做发片自动录入，系统升级后所有元素路径全变了，维护成本比重新开发还高。更头疼的是，客户是内网环境，传统方案依赖云端控制中心，离线状态下根本跑不起来。 而 AI Agent 的兴起给了 RPA 一条新出路。不是替代，是融合——让 RPA 具备“理解-决策-执行”的闭环能力，从“机械臂”变成“数字员工”。 这就是 **RPA 智能体**（RPA Agent）的核心思路：用 MCP 协议把大模型能力注入 RPA 流程，让自动化流程能看懂文档、理解意图、动态决策，而且全程可以在内网离线环境跑通。 --- ## 二、RPA 智能体的技术架构 ### 2.1 三层架构模型 ``` ┌─────────────────────────────────────────┐ │ 业务层（Workflow） │ │ 发片处理 / 数据填报 / 报表生成 │ ├─────────────────────────────────────────┤ │ 智能层（Agent Core） │ │ 意图识别 → 任务规划 → 工具调用 → 结果校验 │ ├─────────────────────────────────────────┤ │ 执行层（RPA Engine） │ │ 元素捕获 → 模拟操作 → 数据提取 → 异常处理 │ ├─────────────────────────────────────────┤ │ 感知层（MCP Tools） │ │ 本地大模型 / OCR / 文档解析 / 知识库检索 │ └─────────────────────────────────────────┘ ``` ### 2.2 MCP 协议的关键作用 MCP（Model Context Protocol）是 Anthropic 提出的开放标准，目的是让 AI 模型能安全地调用外部工具。对 RPA 来说，这意味着： - **大模型可以直接操控 RPA**：AI 不再只是聊天，而是能打开浏览器、填表单、点按钮 - **工具标准化**：一个 MCP Server 可以被多个 Agent 复用，避免重复开发 - **安全隔离**：敏感操作在本地沙箱执行，数据不出域 --- ## 三、实战：搭建一个“智能发片处理 Agent” ### 3.1 场景描述 需求：自动从邮箱下载 PDF 发片，提取关键信息，录入财务系统，并生成汇总 Excel。 传统 RPA 的痛点： - PDF 版式各异，固定模板匹配失效 - 财务系统偶尔弹窗提示，规则 RPA 不会处理 - 发片信息有误时需要人工判断 - **内网环境无法连接云端，传统方案直接报废** RPA 智能体的解法： - 用本地大模型理解 PDF 内容，动态提取字段 - Agent 自主判断异常，调用 OCR 修正或人工复核 - 全流程无人值守，异常自动分级上报 - **全链路本地部署，内网离线也能稳定运行** ### 3.2 环境准备 ``` # requirements.txt # 核心依赖 mcp>=1.0.0 # MCP 协议客户端 langchain>=0.3.0 # Agent 框架 langchain-ollama>=0.2.0 # Ollama 集成 ollama>=0.3.0 # 本地大模型接口 pdfplumber>=0.11.0 # PDF 解析 openpyxl>=3.1.0 # Excel 生成 paddleocr>=2.7.0 # 离线 OCR（可选） ``` **本地模型部署**（以内网环境为例）： ``` # 使用 Ollama 部署本地 Qwen2.5 模型 ollama pull qwen2.5:14b # 验证模型可用 curl https://localhost:11434/api/generate -d '{ "model": "qwen2.5:14b", "prompt": "提取发片中的购买方名称、金额、税率" }' ``` ### 3.3 MCP Server 定义（RPA 工具集） 创建 `mcp_rpa_server.py`，把 RPA 能力封装成 MCP 工具： ```ja vascript from mcp.server.fastmcp import FastMCP import pdfplumber import json import re from pathlib import Path # 注：以下为示例代称，实际开发中可替换为： # - Playwright / Selenium（网页自动化） # - PyAutoGUI（桌面自动化） # - 或具体 RPA 工具的 Python API import rpa_sdk as rpa mcp = FastMCP("rpa_invoice_agent") @mcp.tool() def launch_browser(url: str) -> str: """启动浏览器并访问指定 URL""" try: rpa.browser.launch(headless=False) rpa.browser.na vigate(url) return f"浏览器已启动，访问: {url}" except Exception as e: return f"启动失败: {str(e)}" @mcp.tool() def extract_pdf_invoice(pdf_path: str) -> str: """从 PDF 发片中提取结构化信息""" result = { "buyer_name": "", "amount": "", "tax_rate": "", "invoice_date": "", "raw_text": "" } path = Path(pdf_path) if not path.exists(): return json.dumps({"error": f"文件不存在: {pdf_path}"}, ensure_ascii=False) with pdfplumber.open(pdf_path) as pdf: for page in pdf.pages: text = page.extract_text() if text: result["raw_text"] += text + "\n" lines = text.split('\n') for line in lines: if "购买方" in line and "名称" in line: result["buyer_name"] = line.split("名称")[-1].strip() if "¥" in line and ("金额" in line or "合计" in line): # 提取数字部分（修正后的正则） nums = re.findall(r'¥?\s*([\d,]+\.?\d*)', line) if nums: result["amount"] = nums[-1].replace(",", "") if "税率" in line: tax_match = re.search(r'(\d+)%', line) if tax_match: result["tax_rate"] = tax_match.group(1) + "%" if "开票日期" in line or "日期" in line: date_match = re.search(r'(\d{4}[年/-]\d{1,2}[月/-]\d{1,2})', line) if date_match: result["invoice_date"] = date_match.group(1) return json.dumps(result, ensure_ascii=False, indent=2) @mcp.tool() def fill_form_field(selector: str, value: str) -> str: """在网页表单中填写指定字段""" try: rpa.element.wait_for(selector, timeout=10) rpa.element.input(selector, value) return f"字段 {selector} 已填写: {value}" except Exception as e: return f"填写失败: {str(e)}。建议检查页面是否加载完成。" @mcp.tool() def click_element(selector: str) -> str: """点击页面元素""" try: rpa.element.click(selector) return f"已点击: {selector}" except Exception as e: # 智能重试：如果元素被遮挡，尝试滚动到可见区域 try: rpa.browser.scroll_to(selector) rpa.element.click(selector) return f"滚动后已点击: {selector}" except Exception as e2: return f"点击失败: {str(e2)}" @mcp.tool() def export_excel(data: str, output_path: str) -> str: """将 JSON 数据导出为 Excel 文件""" import openpyxl from openpyxl.styles import Font, PatternFill # 修正：兼容字符串和对象输入 if isinstance(data, str): records = json.loads(data) else: records = data if not isinstance(records, list): records = [records] wb = openpyxl.Workbook() ws = wb.active ws.title = "发片汇总" # 表头样式 header_fill = PatternFill(start_color="366092", end_color="366092", fill_type="solid") header_font = Font(color="FFFFFF", bold=True) headers = ["购买方", "金额", "税率", "开票日期", "状态"] for col, header in enumerate(headers, 1): cell = ws.cell(row=1, column=col, value=header) cell.fill = header_fill cell.font = header_font # 数据写入 for row_idx, record in enumerate(records, 2): ws.cell(row=row_idx, column=1, value=record.get("buyer_name", "")) ws.cell(row=row_idx, column=2, value=record.get("amount", "")) ws.cell(row=row_idx, column=3, value=record.get("tax_rate", "")) ws.cell(row=row_idx, column=4, value=record.get("invoice_date", "")) ws.cell(row=row_idx, column=5, value=record.get("status", "待处理")) # 修正：确保输出目录存在 Path(output_path).parent.mkdir(parents=True, exist_ok=True) wb.sa ve(output_path) return f"Excel 已导出: {output_path}" if __name__ == "__main__": mcp.run(transport="stdio") ``` ### 3.4 Agent 核心逻辑（任务编排） 创建 `invoice_agent.py`，实现智能决策层： ```ja vascript from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_ollama import ChatOllama from langchain_mcp_adapters.tools import load_mcp_tools from mcp import ClientSession, StdioServerParameters import asyncio import json from pathlib import Path SYSTEM_PROMPT = """你是一个企业发片处理智能体，具备以下能力： 1. 从 PDF 提取发片信息（extract_pdf_invoice） 2. 操作浏览器录入系统（launch_browser, fill_form_field, click_element） 3. 生成汇总报表（export_excel） 工作原则： - 遇到信息缺失时，先尝试 OCR 补充识别，仍缺失则标记"待人工复核" - 金额字段必须校验：含税金额 = 不含税金额 × (1 + 税率) - 每次操作后检查页面反馈，确认成功后再执行下一步 - 异常时自动截图保存，记录错误日志 当前任务：处理发片并录入财务系统，生成日报。 """ async def run_agent(): # 启动 MCP Server server_params = StdioServerParameters( command="python", args=["mcp_rpa_server.py"] ) async with ClientSession(server_params) as session: await session.initialize() # 加载 MCP 工具 tools = await load_mcp_tools(session) # 本地大模型（Qwen2.5 14B，可根据硬件调整为 7B） llm = ChatOllama( model="qwen2.5:14b", temperature=0.1, # 低温度保证确定性 base_url="https://localhost:11434" ) prompt = ChatPromptTemplate.from_messages([ ("system", SYSTEM_PROMPT), ("human", "{input}"), MessagesPlaceholder(variable_name="agent_scratchpad") ]) agent = create_openai_tools_agent(llm, tools, prompt) agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True) # 修正：使用跨平台路径 base_dir = Path(__file__).parent invoice_dir = base_dir / "data" / "invoices" output_dir = base_dir / "output" output_dir.mkdir(parents=True, exist_ok=True) # 执行完整任务 result = await agent_executor.ainvoke({ "input": f""" 任务流程： 1. 提取 {invoice_dir} 目录下所有 PDF 发片信息 2. 打开财务系统 https://finance.company.com/login 3. 逐条录入发片数据（购买方、金额、税率、日期） 4. 生成今日汇总 Excel 到 {output_dir}/invoice_daily.xlsx 5. 校验总金额是否与 PDF 提取一致 """ }) print("执行结果:", result["output"]) if __name__ == "__main__": asyncio.run(run_agent()) ``` ### 3.5 运行与验证 ``` # 1. 确保 Ollama 服务运行 ollama serve # 2. 准备测试数据 mkdir -p data/invoices output # 3. 运行 Agent python invoice_agent.py ``` **预期输出**： ``` [Agent] 正在提取 PDF: invoice_001.pdf [Tool] extract_pdf_invoice 返回: {"buyer_name": "XX科技有限公司", "amount": "12500.00", ...} [Agent] 校验通过，金额 12500.00 符合规则 [Agent] 启动浏览器，访问财务系统... [Tool] launch_browser 返回: 浏览器已启动... [Agent] 正在录入第 1/5 条发片... ... [Agent] 生成 Excel 汇总: output/invoice_daily.xlsx [Agent] 任务完成，共处理 5 张发片，总金额 68300.00 元 ``` --- ## 四、关键设计要点 ### 4.1 本地离线部署方案 企业内网环境无法调用云端 API，这套架构的核心优势是 **全链路本地化**： | 组件 | 本地替代方案 | 说明 | |------|-------------|------| | 大模型 | Ollama + Qwen2.5/DeepSeek | 14B 参数足够处理发片理解 | | 向量库 | Chroma/FAISS | 本地知识库检索 | | OCR | PaddleOCR / Tesseract | 开源方案，离线可用 | | RPA 引擎 | 支持本地运行的 SDK | 无需连接云端控制中心 | ### 4.2 腾讯云部署建议 如果企业已有腾讯云基础设施，可以这样搭建： - **轻量应用服务器**：部署 Ollama 和 RPA 引擎，2核4G 起步，14B 模型流畅运行 - **对象存储 COS**：PDF 发片上传触发云函数，自动调用 Agent 处理 - **云函数 SCF**：无服务器架构运行轻量 Agent 任务，按量计费 - **私有网络 VPC**：全链路内网通信，数据不经过公网 成本估算：轻量服务器（约 80 元/月）+ 云函数（按调用次数）≈ 月均 150 元以内，支撑 50 人团队日常发片处理。 ### 4.3 异常处理机制 RPA 智能体相比传统 RPA 最大的升级是 **自主异常恢复**： ``` # 在 Agent Prompt 中定义异常处理策略 RECOVERY_RULES = """ 异常处理优先级： 1. 元素未找到 → 等待 3 秒重试 → 刷新页面 → 切换备用选择器 2. 弹窗遮挡 → 自动关闭弹窗 → 继续原操作 3. 数据校验失败 → 标记异常 → 截图存档 → 人工复核队列 4. 系统超时 → 重试 2 次 → 记录日志 → 跳过该任务 """ ``` ### 4.4 安全与权限控制 企业级部署必须考虑： - **沙箱执行**：RPA 操作在隔离进程运行，防止恶意脚本 - **操作审计**：每一步都记录日志，支持回溯 - **敏感数据脱敏**：发片中的纳税人识别号等字段自动加密存储 --- ## 五、效果对比：传统 RPA vs RPA 智能体 | 维度 | 传统 RPA | RPA 智能体（本文方案） | |------|----------|-----------------------| | 发片版式适配 | 需为每种版式写模板 | 大模型自动理解，零模板 | | 异常处理 | 中断，等人工介入 | 自主判断，分级处理 | | 系统升级兼容 | 元素路径失效即崩溃 | 智能重试+备用策略 | | 部署方式 | 依赖云端控制中心 | 全本地离线，数据不出域 | | 交付形式 | 脚本分发，环境配置复杂 | 可一键打包为独立程序，即开即用 | | 维护成本 | 高（规则频繁调整） | 低（模型迭代即可） | --- ## 六、扩展场景 这套架构不限于发片处理，可以复用到： - **智能客服填单**：AI 理解客户对话，自动在 CRM 创建工单 - **简历自动筛选**：解析 PDF/Word 简历，匹配岗位 JD，录入 HR 系统 - **舆情监控日报**：抓取多平台数据，AI 生成摘要，自动推送企业微信 - **IT 运维自动化**：根据告警日志，Agent 自动登录服务器执行修复脚本 对于需要分发给非技术人员的场景，可以将整个 Agent 打包为独立可执行文件，脱离开发环境直接运行，大幅降低交付门槛。 核心逻辑不变：**感知（MCP 工具）→ 认知（本地大模型）→ 执行（RPA 引擎）→ 校验（闭环反馈）**。 --- ## 七、总结与展望 RPA 智能体不是概念炒作，而是企业自动化落地的必然方向。MCP 协议的出现让大模型和 RPA 的融合有了标准接口，本地大模型的成熟则解决了数据安全和成本问题。 对于技术团队来说，现在入场的门槛已经很低：一台带 GPU 的开发机、开源的 Ollama、支持 MCP 的 RPA SDK，就能搭出一套可用的认知自动化系统。而且全程内网离线运行，不用担心数据泄露风险。