目录


一、真实案例:老王的80万教训

去年认识一个同行老王,在某制造业公司干RPA,月薪12K。他给采购部门搭了个供应商对账机器人,第一周跑了300多单,财务部门挺满意。

第二周翻车了——一个供应商改了银行账号格式(多了个空格),机器人没识别出来,直接按旧账号打款,多转了80多万。幸好追回,但项目被叫停,老王被总监叫去谈话。

那次之后他花了一个月重构:加数据校验、异常分支、人工复核节点、钉钉告警。二次上线后跑了8个月零故障,月入4000+单据。因为这个项目,老王被提拔为自动化小组负责人,薪资涨到18K

去年公司上"智能审单",需要OCR+NLP。老王不懂AI,但他懂业务逻辑、数据流向、异常场景。他牵头做业务架构,AI工程师负责模型,三个月上线,年省200万+。现在老王月薪35K,title是"流程自动化架构师"

他的路径很典型:工具操作工 → 流程设计师 → 业务+技术复合架构师

这里有个问题想问问大家:你们做RPA项目时,异常处理是怎么设计的? 是全部交给机器人自己处理,还是关键节点必留人工复核?评论区聊聊。


二、市场需求:在涨,但结构变了

Gartner预测未来三年全球RPA开发与实施岗位需求年增 25%。国内2025年全行业缺口约 12万人,平均年薪 15-30万,具备AI技能的复合型人才薪资高出 47%

但注意一个细节:需求从"纯开发"转向"流程优化+业务理解"

现在的JD不再问"会不会用UiPath/影刀搭流程",而是问:"对财务共享中心业务逻辑了解多少?""能不能独立做ROI分析?""会不会用Python做数据清洗?"

纯工具操作工的市场在萎缩,懂业务的流程设计师越来越吃香。RPA流程优化师在智能财务领域紧缺度排第二,占比 28%


三、薪资真相:两极分化严重

表格

级别 经验 一线城市 核心能力
初级 1-2年 8K-15K 流程录制、简单调试
中级 3-5年 15K-25K 复杂流程设计、异常处理、业务沟通
高级 5年+ 30K-50K 平台架构、治理规范、技术评估

薪资差距的核心是"复合能力"。只会拖拖拽拽,天花板就是15K。懂Python/SQL、能结合AI、深耕某个行业,溢价明显。


四、技术困境:RPA不是终点,是起点

做这行最焦虑的不是加班,是技术迭代速度

RPA门槛低,非科班培训两三个月就能上手。但低门槛=高竞争。预计2027年前 30%中小RPA厂商被并购,头部企业技术支出占营收22%。

三个明确趋势:

1. RPA+AI深度融合

规则驱动→智能自动化。结合大模型处理非结构化数据,OCR+NLP智能审单。不会点AI的RPA工程师,未来三年竞争力大幅下降

2. 从"搭流程"到"管流程"

企业不再满足单点自动化,要全链路编排、异常自愈、集中监控。需要平台化思维。

3. 安全合规成刚需

自动化触及核心业务后,数据安全、审计合规、权限管理从"可选项"变"必选项"。


五、职业发展:三条realistic出路

路线一:纵向深耕,走流程架构师

流程开发 → 高级流程工程师 → RPA架构师。要求对行业业务逻辑理解深,能设计企业级自动化治理框架。

路线二:横向扩展,转型超自动化

RPA只是超自动化一个组件。补充低代码平台、iPaaS、流程挖掘(Process Mining)。市场需求增长快,天花板更高。

路线三:行业专家路线

选垂直领域(财务/HR/供应链),变成"懂技术的业务专家"。这类人才稀缺,议价能力最强。


六、工具选型:国产化趋势

说到工具,很多人问:学哪个平台最有前途?

国内厂商在崛起,不要只盯着UiPath。国产工具在本土化适配、服务响应、价格上有优势,信创政策推动下,国企央企采购清单里国产比重越来越高。

但选型要看实际场景。最近在一个财务对账项目里,团队对比了几款工具:

工具 适用场景 Excel兼容 中文OCR 部署方式 信创适配 成本
UiPath 复杂企业级 良好 一般 云+本地
影刀 轻量快速 良好 良好 云端
蓝印RPA 财务/对账类 支持复杂公式和宏 发票/票据识别准确率实测可用 私有化

选型没有绝对好坏,只有场景匹配度

工具只是手段,值钱的是你用工具解决业务问题的能力


七、干货:异常处理代码实战(建议收藏)

老王那次80万的教训,核心问题是异常处理缺失。下面是他重构后的核心逻辑,可以直接复用。

环境依赖

# requirements.txt
logging
json
datetime
pathlib

核心代码

import logging
import json
from datetime import datetime
from pathlib import Path

# 配置日志:按天归档,便于审计
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - [%(filename)s:%(lineno)d] - %(message)s',
    handlers=[
        logging.FileHandler('rpa_process.log', encoding='utf-8'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

class PaymentProcessor:
    def __init__(self):
        self.error_log = []
    
    def validate_account(self, account: str) -> bool:
        """
        账号格式校验:
        1. 去除空格和制表符
        2. 检查长度(19位)
        3. 检查是否全为数字
        """
        cleaned = account.replace(" ", "").replace("\t", "")
        if len(cleaned) != 19 or not cleaned.isdigit():
            logger.error(f"账号格式异常: '{account}' -> 清洗后: '{cleaned}'")
            return False
        return True
    
    def check_amount_threshold(self, amount: float) -> bool:
        """
        金额阈值检查:
        - 超5万触发人工复核
        - 记录预警日志
        """
        if amount > 50000:
            logger.warning(f"金额超限需复核: {amount}元")
            return False
        return True
    
    def process_payment(self, data: dict) -> dict:
        """
        核心处理逻辑
        返回结构:{"status": "success|failed|review_required", ...}
        """
        result = {
            "status": "pending",
            "timestamp": datetime.now().isoformat(),
            "original_data": data
        }
        
        try:
            # 1. 数据格式校验
            if not self.validate_account(data.get('account', '')):
                raise ValueError("账号格式校验失败")
            
            # 2. 金额阈值复核
            if not self.check_amount_threshold(data.get('amount', 0)):
                result["status"] = "review_required"
                self.send_alert(f"需人工复核: {data}")
                return result
            
            # 3. 执行转账(模拟)
            transfer_result = self.execute_transfer(data)
            result["status"] = "success"
            result["transfer_id"] = transfer_result.get('id')
            logger.info(f"转账成功: {result}")
            
        except ValueError as e:
            # 可预期的业务异常
            result["status"] = "validation_failed"
            result["error"] = str(e)
            logger.error(f"校验失败: {e}")
            
        except Exception as e:
            # 未预期的系统异常
            result["status"] = "failed"
            result["error"] = str(e)
            logger.error(f"未预期异常: {e}", exc_info=True)
            self.send_alert(f"流程异常: {str(e)}")
            
        finally:
            # 无论成功失败,记录全链路日志
            self.log_transaction(result)
        
        return result
    
    def execute_transfer(self, data: dict) -> dict:
        """模拟转账API调用"""
        return {"id": f"TXN{datetime.now().strftime('%Y%m%d%H%M%S')}"}
    
    def send_alert(self, message: str):
        """发送告警(钉钉/企业微信)"""
        logger.info(f"[ALERT] {message}")
        # 实际接入钉钉机器人API
    
    def log_transaction(self, result: dict):
        """全链路日志记录:按天归档"""
        log_file = Path(f"transactions/{datetime.now().strftime('%Y%m%d')}.json")
        log_file.parent.mkdir(exist_ok=True)
        
        with open(log_file, 'a', encoding='utf-8') as f:
            f.write(json.dumps(result, ensure_ascii=False) + '\n')

# ============ 测试用例 ============

if __name__ == "__main__":
    processor = PaymentProcessor()
    
    # 用例1:正常数据
    normal_data = {
        "account": "6222 0222 0000 1234 567",
        "amount": 30000,
        "vendor": "供应商A"
    }
    print("【正常数据】", processor.process_payment(normal_data))
    
    # 用例2:异常数据 - 格式错误(少一位)
    bad_data = {
        "account": "6222 0222 0000 1234",
        "amount": 30000,
        "vendor": "供应商B"
    }
    print("【格式错误】", processor.process_payment(bad_data))
    
    # 用例3:超限数据 - 需人工复核
    large_data = {
        "account": "6222 0222 0000 1234 567",
        "amount": 80000,
        "vendor": "供应商C"
    }
    print("【金额超限】", processor.process_payment(large_data))
    
    # 用例4:边界情况 - 空账号
    empty_data = {
        "account": "",
        "amount": 10000,
        "vendor": "供应商D"
    }
    print("【空账号】", processor.process_payment(empty_data))
    
    # 用例5:边界情况 - 负数金额
    negative_data = {
        "account": "6222 0222 0000 1234 567",
        "amount": -5000,
        "vendor": "供应商E"
    }
    print("【负数金额】", processor.process_payment(negative_data))

运行效果

2026-04-23 10:15:30,123 - ERROR - [rpa_demo.py:35] - 账号格式异常: '6222 0222 0000 1234' -> 清洗后: '6222022200001234'
2026-04-23 10:15:30,124 - ERROR - [rpa_demo.py:75] - 校验失败: 账号格式校验失败
【格式错误】 {'status': 'validation_failed', 'error': '账号格式校验失败', ...}

2026-04-23 10:15:30,125 - WARNING - [rpa_demo.py:45] - 金额超限需复核: 80000元
【金额超限】 {'status': 'review_required', ...}

这段代码的核心设计点

  1. 分层异常捕获:先具体异常(ValueError),再通用兜底(Exception)

  2. 全链路日志:无论成功失败,都记录完整上下文

  3. 阈值人工复核:关键节点保留人工干预入口

  4. 按天归档:便于审计和排查


八、避坑清单(建议收藏)

  1. 异常处理不是可选项——任何涉及资金/核心数据的流程,必须设计异常分支+人工复核

  2. 日志要全链路——不只是"成功/失败",要记录输入参数、中间状态、输出结果

  3. 别相信"100%自动化"——关键节点留人工干预入口,这是保命设计

  4. 上线前做压力测试——Excel 100条和10万条的处理逻辑可能完全不同

  5. 文档同步更新——流程改了,文档必须改,否则三个月后你自己都看不懂

最后说点实际的:RPA这行不会消失,但"只会拖拖拽拽"的人会被淘汰。如果你现在还在做纯流程录制,建议尽快补Python、补业务、补AI。技术迭代不等人。

Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐