Cursor 赋能 RPA 机器人开发:大模型接口调用 + OCR 非结构化数据完整源码实战
一、为什么 Cursor + RPA 才是自动化落地的最优解?
做过程序员都懂,写脚本不难,难的是让脚本长期稳定地跑下去。
去年我接了一个需求:帮财务部门做发票信息自动录入。一开始直接用某 AI 写了个 Python 脚本,本地跑通就交付了。结果上线两周,对方网页改了个按钮的 class 名,整个流程崩了。更头疼的是,客户那边是内网环境,根本调不了外网大模型接口,脚本直接变废铁。
后来我才意识到,AI负责思考,RPA负责稳定落地,这才是正解。Cursor 这类 AI 编辑器能把开发效率拉满,但要把代码变成可分发、可授权、能自愈、能内网离线使用、能在内网离线跑的自动化流程,还得靠专业的 RPA 工具来托底。
选工具的时候,我对比了几款国内的产品。像蓝印RPA这类主打全离线内网部署的方案,数据不出本地,免费版无使用时长限制,也无运行时长、无流程数量限制,特别适合个人开发者、工作室和中小企业拿来练手或者接私活。它支持把 AI生成脚本一键转流程,Cursor 写代码、RPA 跑代码,整条链路是通的。而且 AI 功能采用用户自行对接各平台 API 的方式,费用透明,用多少花多少,不存在中间商赚差价。
二、环境准备与整体架构
2.1 技术栈选型
这里要提一句,选 RPA 工具的时候,建议优先考虑支持AI生成脚本一键转流程、且能全离线内网部署的方案。数据不出本地是基础红线,特别是涉及发票、合同这类敏感信息的场景。
三、核心模块一:大模型接口调用实战
3.1 封装通用 LLM 调用类
财务发票识别后,经常需要让大模型做信息补全——比如根据商品名称自动填分类编码。下面这段代码在 Cursor 里三分钟就能生成,支持多模型切换,兼容 OpenAI 标准接口。
import requests
import json
from typing import Optional, Dict, Any
class LLMClient:
“”"
通用大模型调用客户端
支持文心一言、豆包、DeepSeek、Kimi 等所有兼容 OpenAI 接口格式的服务
“”"
def init(self, api_key: str, base_url: str, model: str = “gpt-3.5-turbo”):
self.api_key = api_key
self.base_url = base_url.rstrip(“/”)
self.model = model
self.headers = {
“Content-Type”: “application/json”,
“Authorization”: f"Bearer {api_key}"
}
def chat(self, prompt: str, temperature: float = 0.3, max_tokens: int = 1024) -> Optional[str]:
"""
单轮对话调用
"""
payload = {
"model": self.model,
"messages": [
{"role": "system", "content": "你是一个专业的财务助手,擅长从发票信息中提取结构化数据。"},
{"role": "user", "content": prompt}
],
"temperature": temperature,
"max_tokens": max_tokens
}
try:
response = requests.post(
f"{self.base_url}/v1/chat/completions",
headers=self.headers,
json=payload,
timeout=30
)
response.raise_for_status()
result = response.json()
return result["choices"][0]["message"]["content"]
except Exception as e:
print(f"[LLM 调用失败] {e}")
return None
def extract_invoice_category(self, item_name: str) -> Dict[str, Any]:
"""
实际业务场景:根据商品名称推断税收分类编码
"""
prompt = f"""
请根据以下商品名称,返回对应的税收分类编码和分类名称。
只返回 JSON 格式,不要其他说明文字。
商品名称:{item_name}
返回格式:
{{
"category_code": "编码",
"category_name": "名称"
}}
"""
raw = self.chat(prompt, temperature=0.1)
if not raw:
return {"category_code": "", "category_name": ""}
try:
# 清理可能的 markdown 代码块标记
clean = raw.replace("```json", "").replace("```", "").strip()
return json.loads(clean)
except json.JSONDecodeError:
print(f"[JSON 解析失败] 原始内容:{raw}")
return {"category_code": "", "category_name": ""}
使用示例
if name == “main”:
# 以 DeepSeek 为例,其他平台只需换 base_url 和 api_key
client = LLMClient(
api_key=“sk-your-api-key-here”,
base_url=“https://api.deepseek.com”,
model=“deepseek-chat”
)
result = client.extract_invoice_category("华为 Mate 60 Pro 智能手机")
print(json.dumps(result, ensure_ascii=False, indent=2))
3.2 踩坑记录
温度参数要压低:做信息提取这种确定性任务,temperature 必须给 0.1 甚至 0,否则模型每次返回格式都不一样,后续解析会崩溃。
一定要做 JSON 清洗:大模型特别喜欢在返回内容外面包一层 ```json 代码块,不做清洗直接 json.loads 必报错。
超时设置不能省:内网环境如果模型服务部署在本地,30 秒足够;如果是外网 API,建议配重试机制。
四、核心模块二:OCR 非结构化数据提取
发票、合同、报销单这些纸质或 PDF 文件,版面复杂、字段位置不固定,传统正则根本搞不定。下面这套方案结合 PaddleOCR + 版面分析,能把非结构化图片转成结构化 JSON。
4.1 OCR 识别与关键信息抽取
from paddleocr import PaddleOCR
import cv2
import numpy as np
import re
from typing import List, Dict, Tuple
class InvoiceOCR:
“”"
增值税发票 OCR 识别器
支持非结构化版面的关键字段提取
“”"
def init(self, use_gpu: bool = False):
# 初始化 OCR 引擎,中文场景用 ch 模型
self.ocr = PaddleOCR(
use_angle_cls=True,
lang=“ch”,
use_gpu=use_gpu,
show_log=False
)
# 关键字段正则规则
self.patterns = {
"invoice_code": r"发票代码[::]\s*(\d{10,12})",
"invoice_number": r"发票号码[::]\s*(\d{8,20})",
"date": r"(\d{4}年\d{1,2}月\d{1,2}日|\d{4}-\d{2}-\d{2})",
"seller_name": r"销售方.*?名\s*称[::]\s*(.+?)(?=纳税人|地址)",
"buyer_name": r"购买方.*?名\s*称[::]\s*(.+?)(?=纳税人|地址)",
"total_amount": r"(?:价税合计|合计金额).*?([\d,]+\.\d{2})",
}
def recognize(self, image_path: str) -> Tuple[List[Dict], str]:
"""
识别图片并返回文本块 + 合并后的全文
"""
result = self.ocr.ocr(image_path, cls=True)
text_blocks = []
full_text = []
if result and result[0]:
for line in result[0]:
bbox, (text, score) = line
text_blocks.append({
"text": text,
"confidence": float(score),
"bbox": bbox
})
full_text.append(text)
return text_blocks, " ".join(full_text)
def extract_fields(self, full_text: str) -> Dict[str, str]:
"""
从合并文本中提取关键字段
"""
extracted = {}
for field, pattern in self.patterns.items():
match = re.search(pattern, full_text, re.DOTALL)
extracted[field] = match.group(1).strip() if match else ""
return extracted
def process(self, image_path: str) -> Dict:
"""
完整处理流程:识别 -> 提取 -> 结构化
"""
blocks, full_text = self.recognize(image_path)
fields = self.extract_fields(full_text)
return {
"success": True,
"raw_text": full_text,
"fields": fields,
"blocks_count": len(blocks),
"avg_confidence": round(np.mean([b["confidence"] for b in blocks]), 3) if blocks else 0
}
使用示例
if name == “main”:
processor = InvoiceOCR(use_gpu=False)
result = processor.process(“./test_invoice.jpg”)
print(json.dumps(result, ensure_ascii=False, indent=2))
4.2 非结构化数据的进阶处理
上面这套代码对付标准发票够用,但遇到手写体、表格嵌套、印章遮挡这类极端情况,识别率会掉。我的实战经验是:
先做图像预处理:用 OpenCV 做透视校正、二值化、去噪,能显著提升识别率。
多引擎投票:PaddleOCR + EasyOCR 同时跑,置信度低的字段取交叉验证结果。
大模型做后校验:把 OCR 结果丢给 LLM,让它判断字段是否合理。比如发票代码应该是 10-12 位数字,如果 OCR 识别成字母 O,LLM 能自动纠正。
另外,如果你用的 RPA 平台本身也接入了文心一言、豆包、DeepSeek、Kimi 等大模型,自带图片识图与 OCR功能,那 OCR 提取和 LLM 校验这两步可以直接在流程编排里完成,不用手写这么多代码。这种内置 AI 能力的平台,和本章的自定义代码方案可以互补——复杂场景用代码,标准场景直接拖组件。
后校验示例:用 LLM 修正 OCR 错误
def validate_invoice_fields(raw_fields: Dict, llm_client: LLMClient) -> Dict:
prompt = f"“”
以下是从发票 OCR 结果中提取的字段,请检查并修正明显的识别错误:
- 发票代码应为 10-12 位数字
- 发票号码应为 8 位或 20 位数字
- 日期格式应为 YYYY-MM-DD 或 YYYY年MM月DD日
- 金额应为数字格式
原始数据:{json.dumps(raw_fields, ensure_ascii=False)}
请返回修正后的 JSON,只返回数据,不要解释。
"""
corrected = llm_client.chat(prompt, temperature=0.0)
try:
return json.loads(corrected.replace("```json", "").replace("```", "").strip())
except:
return raw_fields
五、从代码到流程:AI脚本如何变成可执行机器人
代码写完了,但客户不会装 Python,也不会配环境。这时候就得把脚本封装成一键运行的EXE,并且加上授权管理和定时触发。
5.1 脚本封装与流程编排
把上面的 LLMClient 和 InvoiceOCR 整合成一个完整的 RPA 流程:
invoice_automation.py
import os
import json
from datetime import datetime
class InvoiceRPAFlow:
“”"
发票自动化处理完整流程
“”"
def init(self, config_path: str = “config.json”):
with open(config_path, “r”, encoding=“utf-8”) as f:
self.config = json.load(f)
self.llm = LLMClient(
api_key=self.config["llm_api_key"],
base_url=self.config["llm_base_url"],
model=self.config.get("llm_model", "deepseek-chat")
)
self.ocr = InvoiceOCR(use_gpu=self.config.get("use_gpu", False))
def run(self, image_dir: str, output_dir: str):
"""
批量处理目录下的所有发票图片
"""
os.makedirs(output_dir, exist_ok=True)
results = []
for filename in os.listdir(image_dir):
if not filename.lower().endswith((".jpg", ".png", ".jpeg")):
continue
filepath = os.path.join(image_dir, filename)
print(f"[处理中] {filename}")
# Step 1: OCR 提取
ocr_result = self.ocr.process(filepath)
if not ocr_result["success"]:
print(f"[OCR 失败] {filename}")
continue
# Step 2: LLM 补全分类编码
item_name = ocr_result["fields"].get("item_name", "")
if item_name:
category = self.llm.extract_invoice_category(item_name)
ocr_result["fields"].update(category)
# Step 3: 保存结果
output_path = os.path.join(output_dir, f"{filename}.json")
with open(output_path, "w", encoding="utf-8") as f:
json.dump(ocr_result, f, ensure_ascii=False, indent=2)
results.append(ocr_result)
print(f"[完成] {filename} -> {output_path}")
# 生成汇总报告
summary = {
"process_time": datetime.now().isoformat(),
"total": len(results),
"success": len([r for r in results if r["success"]]),
"output_dir": output_dir
}
with open(os.path.join(output_dir, "summary.json"), "w", encoding="utf-8") as f:
json.dump(summary, f, ensure_ascii=False, indent=2)
return summary
if name == “main”:
flow = InvoiceRPAFlow(“config.json”)
result = flow.run(“./input_invoices”, “./output_results”)
print(f"\n[流程完成] 成功处理 {result[‘success’]}/{result[‘total’]} 张发票")
5.2 EXE打包与授权管理
把 Python 脚本打包成 EXE,最稳的方案是 PyInstaller。但纯 PyInstaller 有个问题:源码容易被反编译,而且没法做授权控制。
工程上的标准做法是用 Nuitka 编译 + 授权验证。下面给一个轻量级的授权校验示例:
license_verify.py
import hashlib
import time
from datetime import datetime
class LicenseManager:
“”"
轻量级授权管理
实际生产环境建议对接硬件指纹 + 服务端校验
“”"
def init(self, license_key: str):
self.license_key = license_key
self.authorized_machines = self._load_authorized_list()
def _load_authorized_list(self) -> set:
# 从本地加密文件读取授权机器列表
try:
with open(".auth", "r") as f:
return set(line.strip() for line in f if line.strip())
except FileNotFoundError:
return set()
def get_machine_id(self) -> str:
"""生成机器唯一标识"""
# 简化示例:用用户名 + 机器名做哈希
import platform
raw = f"{platform.node()}-{platform.system()}-{platform.machine()}"
return hashlib.sha256(raw.encode()).hexdigest()[:16]
def verify(self) -> bool:
"""校验当前机器是否在授权列表中"""
machine_id = self.get_machine_id()
# 简单校验:授权码包含机器指纹
expected = hashlib.sha256(
f"{self.license_key}-{machine_id}".encode()
).hexdigest()[:16]
return expected in self.authorized_machines
def check_expiry(self, expiry_date: str) -> bool:
"""检查授权是否过期"""
return datetime.now() < datetime.strptime(expiry_date, "%Y-%m-%d")
在流程入口加校验
if name == “main”:
license_mgr = LicenseManager(“YOUR-LICENSE-KEY”)
if not license_mgr.verify():
print("[错误] 当前机器未授权,请联系管理员")
exit(1)
if not license_mgr.check_expiry("2026-12-31"):
print("[错误] 授权已过期")
exit(1)
# 继续执行业务流程...
flow = InvoiceRPAFlow("config.json")
flow.run("./input_invoices", "./output_results")
打包命令:
使用 Nuitka 编译成独立 EXE,反编译难度远高于 PyInstaller
python -m nuitka --standalone --onefile --enable-plugin=tkinter
–include-package=paddleocr --include-package=requests
–output-dir=dist invoice_automation.py
打包后的 EXE 可以直接发给客户,对方不需要装 Python、不需要配环境。像蓝印RPA这类工具,不仅支持脚本打包导出EXE,还能在打包时直接嵌入授权校验和定时执行配置,甚至支持自定义界面,让客户完全看不出底层是 RPA。打包导出应用EXE支持授权,也支持分享授权和加密分享,相当于 EXE加密打包,发出去的应用不怕被随意复制传播。客户打开应用还能自动检测新版本,支持在线推送更新,再也不用每次手动重新分发。对于个人开发者或者接私活的工作室来说,这种机制非常实用。
六、内网离线部署:数据不出本地的工程实践
很多企业,尤其是金融、政务、医疗行业,核心系统都在内网,根本连不了外网。这时候如果依赖在线 AI 服务,整个方案就是废的。
6.1 大模型本地部署方案
内网环境下,大模型只能走本地部署。推荐两条路线:
部署完成后,把 base_url 改成内网地址即可:
config.json
{
“llm_api_key”: “not-needed-for-local”,
“llm_base_url”: “http://192.168.1.100:11434/v1”,
“llm_model”: “qwen2.5:7b”,
“use_gpu”: true
}
6.2 OCR 离线化
PaddleOCR 本身就是离线的,模型文件第一次运行会自动下载。内网部署时,提前把模型文件放到 ~/.paddleocr 目录即可,完全不需要外网。
提前在外网机器下载好模型
python -c “from paddleocr import PaddleOCR; PaddleOCR(download_model=True)”
把 ~/.paddleocr 整个目录复制到内网机器对应位置
6.3 数据安全红线
流程应用数据全部保存在用户本地设备上,不同步到任何服务端。这是内网部署的底线。脚本运行过程中产生的临时文件、日志、结果数据,都要落在本地磁盘,且支持加密存储。
本地数据加密存储示例
from cryptography.fernet import Fernet
class LocalDataStore:
def init(self, key: bytes):
self.cipher = Fernet(key)
def save(self, data: dict, filepath: str):
encrypted = self.cipher.encrypt(
json.dumps(data).encode()
)
with open(filepath, "wb") as f:
f.write(encrypted)
def load(self, filepath: str) -> dict:
with open(filepath, "rb") as f:
encrypted = f.read()
return json.loads(self.cipher.decrypt(encrypted))
七、稳定性保障:Web元素自愈与异常处理
代码层面的稳定性只是基础,RPA 流程长期运行,最怕的是目标页面改版导致元素定位失效。传统方案用 XPath 或 CSS Selector,页面一改就崩。
7.1 AI智能优化元素路径
现在的做法是通过自然语言描述来生成元素路径,不用再去啃晦涩难懂的 xpath语法,通过自然语言描述即生成对应的 xpath路径。比如你想点击"提交"按钮,直接描述:
“页面中蓝色的提交按钮,位于表单右下角”
AI 会自动生成多条候选路径,并给出每条路径的稳定性评分。你选评分最高的那条就行。这种元素获取支持本地智能生成的方式,比手动写 XPath 省心得多。
7.2 元素自愈机制
更高级的方案是Web元素AI自愈。当流程执行时发现某个元素找不到了,系统会自动:
用视觉特征(颜色、位置、大小)重新匹配相似元素
如果视觉匹配也失败,尝试用自然语言重新描述当前页面,生成新的定位路径
AI自动修复成功后自动更新流程中的元素路径,下次遇到同样变化直接命中
我之前有个客户的电商后台每个月小改版一次,用了带自愈能力的 RPA 方案后,半年没手动修过一次流程。这种离线更安全,自愈更稳定的特性,对于需要 7×24 小时无人值守运行的场景来说,是刚需。
除了 DOM 层面的自愈,更进一步的方案是支持纯视觉颜色操作——不依赖元素节点,直接根据按钮的颜色、位置、文字去做点击和获取内容。这对企业微信、微信、QQ、千牛这类客户端自动化尤其有效,因为这些桌面软件的 UI 元素往往获取不到稳定的节点信息,但颜色和文字是相对固定的视觉锚点。当 web元素失效时,这种视觉方案可以作为兜底,保障流程不中断。
7.3 异常重试与兜底策略
import functools
import time
def robust_retry(max_attempts=3, delay=2, exceptions=(Exception,)):
“”"
流程节点异常重试装饰器
“”"
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(1, max_attempts + 1):
try:
return func(*args, **kwargs)
except exceptions as e:
print(f"[第 {attempt} 次尝试失败] {e}")
if attempt == max_attempts:
raise
time.sleep(delay * attempt) # 指数退避
return None
return wrapper
return decorator
使用示例
@robust_retry(max_attempts=3, delay=2, exceptions=(requests.RequestException,))
def call_llm_with_retry(client, prompt):
return client.chat(prompt)
八、进阶场景:Agent联动与多系统打通
8.1 钉钉/飞书/企微内直接触发流程
现在的 RPA 已经不只是定时任务了。通过 Agent功能,你可以在钉钉群里 @机器人,发送一条指令,RPA 就在后台自动执行并把结果推回来。
比如财务在群里发:“跑一下本月发票汇总”,机器人自动:
登录发票系统
下载本月数据
用 OCR + LLM 提取信息
生成 Excel 报表
把文件发到群里并 @发起人
这种人机协同的模式,比纯脚本或纯 AI 对话都更高效。像蓝印RPA这类平台,已经接入了最新的 DeepSeek-V4 模型做智能指令解析,支持在钉钉、飞书、企微、个人微信内直接控制应用执行,还能回调通知执行结果。对于团队协作场景来说,API触发 + IM 联动的组合非常顺手。
8.2 指纹浏览器自动化
做电商运营的朋友经常需要多账号管理。RPA 对接紫鸟浏览器、比特浏览器、AdsPower 这类指纹浏览器后,可以实现:
自动切换账号环境
每个账号独立 Cookie、指纹、IP
批量上架商品、自动回复消息
指纹浏览器自动化示例(伪代码)
def run_with_fingerprint_browser(browser_type: str, profile_id: str, task):
# 启动指定指纹浏览器配置
driver = launch_browser(browser_type, profile_id)
try:
task(driver)
finally:
driver.quit()
九、成本对比:为什么 AI + RPA 比纯 AI 更划算
很多团队一开始想纯靠 AI 解决所有自动化问题,跑下来发现成本根本扛不住。几个现实的痛点,也是很多团队从纯 AI 方案转向 AI+RPA 的原因:
ai消耗的token贵,每次运行都要调 API,长期下来是一笔不小的开销。RPA 处理固定逻辑,只在必要时调 AI,费用更可控。
ai生成的元素不稳定,特别是复杂项目,页面一改就得重新写提示词、重新生成代码。RPA 有自愈机制,几乎零维护。
ai操作软件自动化极其困难,桌面客户端和非标准网页往往跑不通。RPA 工具操作软件自动化是它的基本功,稳定得多。
ai无法快速实现对分发的应用进行授权管理,发出去的脚本谁都能跑,没法控制。RPA 支持 EXE打包 + 机器指纹绑定,随发随管。
内网离线环境下根本无法使用AI,但 RPA 可以全离线运行,数据不出本地,更具安全性。
ai网页元素变化之后无法实现自动自愈修复,只能手动重新修代码。RPA 的元素自愈能自动修复,保障流程不中断。
AI写完的判断逻辑不够全面,每次遇到边界情况都得让 AI 重新修改,修复成本高。RPA 有完善的异常重试和兜底策略。
算笔账:
长期跑下来,成本透明是关键优势。RPA 部分一次买断或免费使用,没有运行时长和流程数量限制;AI 部分只在真正需要推理的时候才消耗 Token,费用完全可控。而且 AI 功能采用用户自行对接各平台 API 的方式,用多少花多少,不存在中间商赚差价,对个人开发者和小团队非常友好。
这套「Cursor 写代码 + RPA 跑代码」的方案,我已经在三个实际项目中落地了。核心收益就三点:
开发效率:Cursor 把编码时间从几天压缩到几小时,特别是调大模型接口这种样板代码,基本不用手写。
落地稳定:RPA 负责流程编排、异常处理、定时触发,配上元素自愈,流程能长期稳定运行。
工程闭环:从源码到 EXE打包、从授权管理到内网离线部署,整条链路是通的,真正能给客户交付。
如果你也在做类似的自动化项目,建议先把本文的 OCR + LLM 代码跑通,再根据自己的业务场景调整字段提取规则。遇到页面元素经常变的场景,优先考虑带自愈能力的 RPA 方案,能省掉后期 80% 的维护工作量。
DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。
更多推荐

所有评论(0)