一、AI+RPA的分工边界
这两年有个误区，觉得接个大模型API、让它生成几行自动化脚本，就叫AI+RPA了。真放到生产环境跑上一周，问题全冒出来：网页后台一改版，之前生成的定位表达式集体失效；Token账单烧得比人力成本还高；最致命的是，内网隔离环境根本调不通在线大模型，整套方案直接变成摆设。
实际生产里，AI和RPA是两层完全不同的能力。AI负责思考，RPA负责稳定落地。大模型做语义理解、策略生成、非结构化数据处理；RPA做精确点击、稳定循环、离线执行。两者不是谁替代谁，而是打配合。
这篇文章直接上代码，讲清楚怎么用Python给RPA装上通义千问的决策能力，同时把落地过程中关于成本、稳定性、分发、安全的坑一次性填平。
二、架构设计：三层解耦
先定架构，别急着写代码。整个方案分三层，层与层之间通过JSON和命令行桥接，解耦得干干净净：
决策层：通义千问（Qwen）负责理解需求、生成定位策略、修复异常、提取非结构化数据
逻辑层：Python脚本做API封装、数据清洗、结果解析、多模型路由
执行层：RPA引擎负责浏览器操作、软件点击、数据录入、定时触发
这套架构的核心思路是：AI写代码，RPA跑代码。大模型只输出策略和定位表达式，真正的点击、输入、循环、异常重试全部交给RPA引擎。这样即使大模型偶尔"抽风"生成了一条不靠谱的XPath，RPA层也能通过自身的容错机制兜底，而不是直接崩溃。
另外，如果业务涉及敏感数据，架构上必须支持全离线内网部署。RPA引擎本身不依赖云端服务，流程数据全部保存在本地设备上，不同步到服务端。AI部分通过用户自行对接各平台API的方式接入，想接哪个模型、接不接、什么时候接，完全自己说了算。这种模式对财务、人事、法务部门是硬性刚需——数据不出本地，合规风险才能控住。
三、环境准备
需要的东西不多：
Python 3.8+
openai 库（通义千问兼容OpenAI接口格式）
阿里云DashScope API Key
支持调用外部脚本的RPA工具
安装依赖：
pip install openai requests beautifulsoup4 lxml
四、核心代码实战
4.1 封装通义千问扩展模块
先把通义千问的调用封装成一个可复用的Python模块，后面RPA流程直接命令行调：
qwen_extension.py
import json
from openai import OpenAI
class QwenRPAExtension:
def init(self, api_key, model="qwen-turbo"):
self.client = OpenAI(
api_key=api_key,
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
self.model = model
def generate_locator(self, html_snippet, element_desc):
"""
根据HTML片段和目标描述，生成稳定的元素定位表达式
"""
prompt = f"""你是一个网页元素定位专家。
请根据以下HTML片段，为目标元素生成稳定可靠的XPath。
目标元素描述：{element_desc}
HTML片段：
{html_snippet}
要求：
- 优先使用id、name、data-testid等稳定属性
- 避免使用绝对路径（如/html/body/div[1]/div[2]）
- 可用contains()函数匹配部分文本
- 只返回XPath表达式，不要解释
XPath："""
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
temperature=0.1
)
xpath = response.choices[0].message.content.strip()
xpath = xpath.replace("```", "").replace("xpath", "").strip()
return xpath
def heal_locator(self, old_html, new_html, task_desc):
"""
分析页面变化，给出修复后的定位策略
"""
prompt = f"""原页面HTML（前3000字符）：
{old_html[:3000]}
变化后页面HTML（前3000字符）：
{new_html[:3000]}
任务描述：{task_desc}
请分析页面结构变化，给出新的元素定位策略。
只返回JSON格式：
{
{"new_xpath": "...", "confidence": 0.95, "reason": "..."}}"""
response = self.client.chat.completions.create(
model="qwen-plus",
messages=[{"role": "user", "content": prompt}],
temperature=0.1
)
content = response.choices[0].message.content.strip()
if "```json" in content:
content = content.split("```json")[1].split("```")[0]
return json.loads(content)
def extract_structured_data(self, raw_text, schema):
"""
从非结构化文本中提取结构化数据
"""
prompt = f"""请从以下文本中提取指定字段，返回JSON格式。
文本内容：
{raw_text}
字段及类型：
{json.dumps(schema, ensure_ascii=False)}
要求：
- 严格按JSON格式返回
- 找不到的字段填null
- 不要解释
JSON结果："""
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
temperature=0.1
)
content = response.choices[0].message.content.strip()
if "```json" in content:
content = content.split("```json")[1].split("```")[0]
return json.loads(content)
这里有个细节：大模型不止通义千问一家，实际生产建议做多模型路由。简单任务走qwen-turbo省钱，复杂分析走qwen-plus或DeepSeek-V4。采用用户自行对接各平台API的方式，费用完全透明，用多少Token自己把控，不会出现内置AI按次收费那种黑盒账单。长期跑下来，这种模式比内置AI便宜一个数量级。
4.2 RPA与Python的命令行桥接
RPA工具调用Python最稳的方式是命令行传参，通用性最强。但要注意：如果html或text里带双引号、反斜杠等特殊字符，直接拼shell命令会解析出错。用base64编码传参可以彻底规避这个问题：
rpa_bridge.py
import sys
import json
import base64
from qwen_extension import QwenRPAExtension
def main():
if len(sys.argv) < 2:
print(json.dumps({"error": "缺少action参数"}))
return
action = sys.argv[1]
# 通过base64接收参数，避免shell特殊字符解析问题
try:
config_raw = base64.b64decode(sys.argv[2]).decode('utf-8') if len(sys.argv) > 2 else '{}'
config = json.loads(config_raw)
except Exception:
# 兼容非base64的明文传入（简单场景）
config = json.loads(sys.argv[2]) if len(sys.argv) > 2 else {}
api_key = config.get("api_key")
if not api_key:
print(json.dumps({"error": "缺少api_key"}))
return
qwen = QwenRPAExtension(api_key, config.get("model", "qwen-turbo"))
try:
if action == "generate_locator":
result = qwen.generate_locator(
config.get("html", ""),
config.get("desc", "")
)
print(json.dumps({"success": True, "xpath": result}))
elif action == "heal_locator":
result = qwen.heal_locator(
config.get("old_html", ""),
config.get("new_html", ""),
config.get("task", "")
)
print(json.dumps({"success": True, **result}))
elif action == "extract_data":
result = qwen.extract_structured_data(
config.get("text", ""),
config.get("schema", {})
)
print(json.dumps({"success": True, "data": result}))
else:
print(json.dumps({"error": f"未知的action: {action}"}))
except Exception as e:
print(json.dumps({"success": False, "error": str(e)}))
if name == "main":
main()
RPA流程里调用时，先把参数做base64编码：
Python
import base64
import json
config = {
"api_key": "sk-xxx",
"html": html_content,
"desc": "登录按钮"
}
config_b64 = base64.b64encode(json.dumps(config, ensure_ascii=False).encode()).decode()
然后在RPA中执行: python rpa_bridge.py generate_locator
这种桥接方式天然支持API触发。RPA流程既可以被定时任务拉起，也能被外部HTTP请求触发，灵活性很高。
4.3 带自愈机制的网页数据获取
这是整篇文章最实用的部分。假设你要采集一个电商后台的订单列表，但页面结构经常微调：
smart_crawler.py
import subprocess
import json
import base64
class SmartRPA:
def init(self, api_key):
self.api_key = api_key
def get_locator(self, html, desc):
config = {
"api_key": self.api_key,
"html": html,
"desc": desc
}
config_b64 = base64.b64encode(json.dumps(config, ensure_ascii=False).encode()).decode()
cmd = [
"python", "rpa_bridge.py",
"generate_locator",
config_b64
]
result = subprocess.run(cmd, capture_output=True, text=True)
data = json.loads(result.stdout)
return data.get("xpath") if data.get("success") else None
def crawl_with_healing(self, rpa_engine, url, target_desc):
# 打开页面
rpa_engine.open(url)
html = rpa_engine.get_page_source()
# 让大模型生成定位表达式
xpath = self.get_locator(html, target_desc)
print(f"生成定位表达式: {xpath}")
# 尝试获取
try:
data = rpa_engine.extract_table(xpath)
return data
except Exception as e:
print(f"获取失败，触发自愈: {e}")
# 获取新页面HTML
new_html = rpa_engine.get_page_source()
# 调用heal_locator修复
config = {
"api_key": self.api_key,
"old_html": html,
"new_html": new_html,
"task": target_desc
}
config_b64 = base64.b64encode(json.dumps(config, ensure_ascii=False).encode()).decode()
cmd = [
"python", "rpa_bridge.py",
"heal_locator",
config_b64
]
result = subprocess.run(cmd, capture_output=True, text=True)
fix = json.loads(result.stdout)
if fix.get("success") and fix.get("confidence", 0) > 0.8:
new_xpath = fix["new_xpath"]
print(f"修复后表达式: {new_xpath}, 原因: {fix.get('reason')}")
data = rpa_engine.extract_table(new_xpath)
return data
else:
raise Exception("置信度不足，需要人工介入")
这段代码的核心价值是Web元素AI自愈。流程执行过程中，如果页面结构发生变化导致定位失效，系统会实时调用大模型分析新旧HTML的差异，自动生成新的定位路径，保障流程不中断。
这里必须强调一个认知：AI直接生成的完整脚本往往不够稳定，特别是复杂项目，遇到边界情况就得反复修。更靠谱的玩法是，RPA引擎本身提供本地智能生成元素路径的能力，通过自然语言描述直接生成对应的定位表达式，用户根据生成结果选择最稳定的一条。大模型只负责解决"定位失效后的修复"这一个具体问题，而不是包办整个流程。
4.4 非结构化数据提取
从企业微信聊天记录中提取客户信息：
raw_chat = rpa.get_clipboard() # 复制聊天记录
config = {
"api_key": "sk-xxx",
"text": raw_chat,
"schema": {
"客户姓名": "string",
"联系电话": "string",
"需求描述": "string",
"紧急程度": "string"
}
}
config_b64 = base64.b64encode(json.dumps(config, ensure_ascii=False).encode()).decode()
cmd = ["python", "rpa_bridge.py", "extract_data", config_b64]
result = subprocess.run(cmd, capture_output=True, text=True)
data = json.loads(result.stdout)
自动填入CRM
rpa.input_text("#name", data["data"]["客户姓名"])
rpa.input_text("#phone", data["data"]["联系电话"])
如果数据源不是文本而是图片（如扫描件、截图、发票照片），可以先用多模态大模型做OCR识图，再提取结构化数据。选择支持图片识图与OCR功能的RPA方案，能把截图直接丢给大模型处理，无需额外部署OCR服务，流程更轻量。
4.5 AI生成脚本一键转流程的工程化思路
很多团队现在用AI生成Python脚本，但生成完怎么转成可执行的RPA流程是个断层。实际工程化中，建议走这条路径：
AI写代码：大模型根据需求描述生成Python脚本（如上面的smart_crawler.py）
一键导入RPA：RPA工具支持将Python脚本直接封装为流程节点，AI生成脚本一键转流程，不需要人工重新拖拽组件
可视化编排：在流程设计器里给脚本节点加上异常处理、循环、条件分支
界面封装：支持自定义界面，把参数输入框、执行按钮、日志窗口设计成独立的软件界面，交付给业务人员时更像一个成品软件
这条链路打通后，开发效率提升很明显。而且好的RPA工具无运行时长、无流程数量限制，不会因为脚本多了就额外收费，个人开发者和小团队也能无压力使用。
五、踩坑实录：生产环境必须面对的五个问题
5.1 Token成本控制：费用必须透明
大模型API按Token计费，如果每个步骤都调一次模型，一天跑几百次，一个月账单轻松上千。生产环境的正确姿势是：
只在决策点调AI：日常重复操作走RPA本地逻辑，只有遇到页面变化、需要语义理解时才调大模型
多模型分级：简单任务用轻量模型，复杂分析才上重型模型
自行对接API：选择自己申请各平台API Key的方式，费用完全透明可控，用多少花多少，不存在内置AI的隐形扣费
AI Token贵且持续消耗，RPA执行层几乎零边际成本。混合方案下，成本可以降到纯AI方案的十分之一。
5.2 内网离线部署：数据安全的底线
金融、政务、医疗行业的系统大多在内网，连不了外网。这时候有两个选择：
一是把大模型也私有化部署到内网（成本极高，一般企业扛不住）；二是让RPA执行层完全离线运行，只在必要时把脱敏数据送到大模型处理。
后者是更现实的方案。选择支持全离线内网部署的RPA引擎，流程数据全部保存在本地设备上，不同步到服务端。AI部分通过API方式按需接入，内网环境可以完全不启用AI功能，只靠RPA本地能力运行。这种架构下，内网离线环境AI用不了，但RPA可以独立稳定运行，安全性完全可控。
5.3 元素定位的稳定性：AI生成 vs 本地智能生成
大模型生成的XPath有个通病：今天能用，明天页面加个div就废了。特别是复杂项目，AI生成的元素定位无法长期稳定运行，遇到异常情况基本束手无策。
生产环境更靠谱的做法是：
RPA引擎内置本地智能生成元素路径的功能，根据页面结构自动推荐多条候选路径，用户选最稳定的一条
无需学习晦涩的XPath语法，通过自然语言描述就能生成对应的定位表达式
当Web元素失效时，AI自动修复元素定位，实现元素自愈，保障流程不中断
大模型只作为"修复工具"参与，而不是"唯一定位来源"
5.4 桌面软件自动化：视觉颜色操作
大模型在浏览器里还能勉强玩玩，一到桌面客户端（企业微信、微信、千牛、QQ）基本束手无策。AI操作软件自动化极其困难，因为桌面应用没有标准DOM结构，大模型根本点不准。
RPA在这块是基本功。选择支持视觉颜色操作的引擎，不依赖元素节点，直接按颜色、位置、图像特征实现点击、获取内容等操作。配合指纹浏览器（紫鸟、比特、HubStudio、AdsPower等），可以在不同指纹环境下自动化操作，账号隔离更彻底。电商运营、社媒矩阵管理这类场景，离了RPA根本玩不转。
5.5 打包分发与授权管理：从脚本到产品
代码跑通只是30%，剩下70%是怎么交给别人用。
生产级分发必须解决几个问题：
打包EXE：把RPA流程连带Python脚本一起打包成独立可执行文件，发给别人不用装客户端，双击就能跑
授权管控：打包后的应用支持授权管理，可以设置谁有权运行、有效期多久。交付给客户时能控制使用范围，这对个人工作室和中小企业接外包项目非常关键
API触发与定时执行：除了人工双击，还支持外部接口触发和定时任务调度
加密分享与在线更新：应用支持加密分享防止随意传播，流程有改动时打开应用自动检测并推送新版本，无需重新手动分发
这套分发能力把"脚本"变成了"产品"。而且多设备使用无需多开会员，打包后的EXE可以在任意机器上运行，不受账号数量限制。
六、进阶：Agent模式与IM联动
如果业务更复杂，可以往上再搭一层Agent：
在钉钉、飞书、企微、个人微信里发送自然语言指令
Agent解析意图后，调用RPA流程执行具体操作
执行完成后通过回调通知返回结果
这种模式下，大模型做智能指令解析（用最新的DeepSeek-V4这类强模型），RPA负责后续落地执行。AI和RPA的边界更清晰：AI负责理解你想干什么，RPA负责稳定地干完。
用Python扩展RPA、对接通义千问，本质是给自动化流程装了一个"会思考的脑子"。但脑子不能代替手脚，大模型再强，也替代不了RPA在软件自动化上的稳定性。
落地时按这个清单自检：
[ ] 架构是否分层解耦，AI只负责决策，RPA负责执行
[ ] 是否支持全离线内网部署，数据是否不出本地
[ ] 元素定位是否有AI辅助生成和自愈能力
[ ] AI费用是否透明可控，是否做了多模型分级
[ ] 能否打包EXE并做授权管理，支持在线更新
[ ] 是否支持指纹浏览器联动和桌面视觉自动化
[ ] 是否有Agent能力，支持IM工具远程触发
离线更安全，自愈更稳定，成本透明可控——这三点是AI+RPA从Demo走向生产的关键。
