1. 从30条数据的批量处理说起为什么需要智能路由手里有一批数据要处理30条说多不多说少不少。如果纯手工一条条跑半小时就没了而且中间还得盯着万一某条数据触发了模型的限流或者响应超时整条流水线就卡在那里。这种场景在RPA机器人流程自动化的实际落地中太常见了——数据量不大但重复性极高人工做纯属浪费生命。我最初的做法很朴素写一个Python脚本循环30次每次调用同一个模型接口拿到结果写回表格。跑通之后确实省事了但问题很快暴露出来。同一个模型连续调用30次有时候会遇到响应变慢、偶尔报错的情况尤其是当这30条数据的类型差异比较大的时候——有的适合通用模型有的需要更强的推理能力有的只是简单的文本清洗。用一个模型硬扛所有任务就像拿一把锤子拧所有螺丝能拧但效率和质量都打折扣。这时候“智能路由”的概念就派上用场了。所谓智能路由核心思路是根据每条数据的特征动态选择最合适的模型来处理。比如短文本分类用轻量模型就够了长文本摘要可能需要上下文窗口更大的模型涉及逻辑推理的任务则要路由到推理能力更强的模型上。蓝耘智能路由提供的正是这种能力——它对外暴露一个统一的调用入口内部根据你设定的策略自动分发请求到不同的模型后端。把这套机制和RPA结合起来价值就很明显了。RPA负责“手”的部分——打开表格、读取数据、写入结果、处理异常弹窗智能路由负责“脑”的部分——判断每条数据该用哪个模型、自动切换、失败重试。两者配合30条数据的处理就从一个脆弱的单模型循环变成了一个有弹性、有冗余的自动化流水线。这篇文章适合两类人看一类是已经在用RPA做业务流程自动化但遇到模型调用瓶颈的从业者另一类是想了解智能路由在实际项目中怎么落地、怎么和脚本配合的技术人员。我会把整个方案的搭建过程、关键决策点、踩过的坑都摊开来讲代码和配置直接给到能跑的程度。2. 蓝耘智能路由的接入方式与模型切换逻辑2.1 统一入口背后的路由策略配置蓝耘智能路由最核心的价值在于它把多个模型的调用统一成了一个入口。你不需要在代码里维护一堆不同厂商的SDK和API Key只需要面向一个端点发请求路由层帮你做分发。这个设计思路和API网关很像但多了“智能”的部分——它不只是转发还会根据你配置的策略做决策。路由策略的配置通常有几种维度。第一种是按任务类型路由比如你在请求里带一个task_type字段路由层根据这个字段映射到对应的模型。第二种是按数据特征路由比如根据输入文本的长度、语言、是否包含代码块等特征自动判断。第三种是按优先级和成本路由比如优先走便宜的模型失败或置信度低时再升级到更强的模型。我在这个项目里用的是第一种和第二种的结合。30条数据里我预先给每条打了一个简单的标签——simple、standard、complex。simple走轻量模型standard走通用模型complex走推理增强模型。这个标签可以手工标注也可以用规则自动生成比如文本长度超过500字且包含“分析”“推理”等关键词的就标为complex。配置大概长这样routing: rules: - match: task_type: simple target: model-lite - match: task_type: standard target: model-general - match: task_type: complex target: model-reasoning fallback: model-general retry: max_attempts: 3 backoff: exponential这个配置文件放在路由层的管理界面或者通过API下发都可以。关键是fallback字段——当主模型不可用时自动降级到备用模型保证流水线不中断。retry部分定义了重试策略指数退避意味着第一次失败等1秒第二次等2秒第三次等4秒避免短时间内反复冲击同一个后端。2.2 在Python脚本中调用路由接口的完整写法配置好路由策略之后Python脚本这边其实很简单。你只需要把请求发到路由层的统一端点带上必要的元数据剩下的交给路由层处理。下面是我实际用的调用封装import requests import time import json ROUTER_ENDPOINT https://api.lanyun-router.example.com/v1/chat/completions API_KEY your-api-key-here def call_with_routing(prompt, task_typestandard, max_retries3): headers { Authorization: fBearer {API_KEY}, Content-Type: application/json } payload { messages: [{role: user, content: prompt}], task_type: task_type, routing_strategy: auto } for attempt in range(max_retries): try: resp requests.post( ROUTER_ENDPOINT, headersheaders, jsonpayload, timeout60 ) if resp.status_code 200: data resp.json() return { success: True, content: data[choices][0][message][content], model_used: data.get(model, unknown), attempts: attempt 1 } else: print(f请求失败状态码 {resp.status_code}第 {attempt1} 次尝试) except requests.exceptions.Timeout: print(f请求超时第 {attempt1} 次尝试) except requests.exceptions.RequestException as e: print(f请求异常{e}第 {attempt1} 次尝试) if attempt max_retries - 1: wait 2 ** attempt time.sleep(wait) return {success: False, content: None, model_used: None, attempts: max_retries}这段代码有几个细节值得说。task_type字段是路由层做决策的依据你传什么值取决于你在路由策略里怎么定义的。routing_strategy设为auto表示让路由层自己判断如果你有明确的偏好可以改成具体策略名。超时设了60秒因为有些复杂任务确实需要更长的处理时间设太短会导致不必要的重试。返回值里我特意带了model_used字段这样在后续分析时能知道每条数据实际走了哪个模型。这个信息在调试和优化阶段非常有用——如果发现某类数据总是被路由到不合适的模型上就可以回头调整路由规则。2.3 模型自动切换的触发条件与降级机制自动切换模型这件事听起来很智能但实际落地时你需要明确“什么时候切”。我在项目里定义了三种触发条件。第一种是显式路由就是前面说的根据task_type字段直接指定。这是最可控的方式适合你对数据特征有清晰认知的场景。第二种是失败降级。当主模型返回错误码比如429限流、503服务不可用或者超时路由层自动把请求转发到备用模型。这个切换对调用方是透明的脚本层面不需要做额外处理只需要在重试逻辑里等待路由层返回结果。第三种是质量降级。这个稍微复杂一些——当模型返回的内容置信度低于阈值或者格式不符合预期比如要求返回JSON但返回了纯文本路由层可以自动升级到更强的模型重新处理。这个机制需要你在路由层配置质量校验规则比如用正则表达式检查返回格式或者用一个轻量分类器判断返回内容是否合格。注意质量降级会带来额外的调用成本和时间开销建议只在关键任务上启用。对于30条数据这种小批量场景我一般只开前两种第三种留给对准确性要求极高的场景。降级链的设计也有讲究。我通常配三级主模型 → 同级别备用模型 → 降级模型。比如complex任务的主模型是推理增强版备用是同厂商的通用版最后兜底是轻量版。这样即使推理增强版整体不可用任务也不会完全失败只是质量可能略有下降。3. RPA与脚本的配合30条数据流水线的搭建过程3.1 数据读取、任务分发与结果回写的分工RPA和Python脚本在这个方案里的分工需要提前想清楚否则很容易出现“两边都在做同一件事”的混乱。我的划分原则是RPA负责所有和界面、文件、系统交互的操作Python脚本负责所有和模型调用、数据处理相关的逻辑。具体到30条数据的场景RPA做这几件事打开Excel或数据库客户端读取30条待处理数据把数据传递给Python脚本可以通过命令行参数、临时文件或消息队列等待脚本处理完成然后把结果写回表格的对应列最后关闭应用或保存文件。Python脚本做这几件事接收RPA传来的数据根据预设规则给每条数据打task_type标签调用蓝耘智能路由接口处理返回结果把结果整理成RPA能识别的格式返回。这种分工的好处是职责清晰。RPA工具比如影刀RPA擅长的是界面操作和流程编排它的Python集成能力虽然不错但如果你把复杂的模型调用逻辑全塞进RPA的脚本节点里调试和维护都会很痛苦。反过来Python脚本擅长数据处理和API调用但让它去操作Excel界面或者处理弹窗那就是自找麻烦。实际搭建时我用的是“RPA调用Python脚本 临时JSON文件交换数据”的方式。RPA把30条数据写到一个input.json文件里然后执行python process.py --input input.json --output output.json脚本跑完后RPA读取output.json把结果逐条写回表格。这个方式比命令行传参更稳定因为30条数据序列化后体积不小直接传参容易遇到长度限制或转义问题。3.2 用Python处理批量数据的核心循环与异常捕获脚本这边的核心逻辑是一个循环但循环里要做的事情比想象中多。下面是我实际用的处理流程import json import sys import argparse from concurrent.futures import ThreadPoolExecutor, as_completed def classify_task(item): text item.get(content, ) if len(text) 100: return simple elif any(kw in text for kw in [分析, 推理, 对比, 评估]): return complex else: return standard def process_single(item): task_type classify_task(item) prompt build_prompt(item, task_type) result call_with_routing(prompt, task_typetask_type) return { id: item[id], task_type: task_type, result: result[content], model_used: result[model_used], success: result[success] } def main(): parser argparse.ArgumentParser() parser.add_argument(--input, requiredTrue) parser.add_argument(--output, requiredTrue) parser.add_argument(--workers, typeint, default3) args parser.parse_args() with open(args.input, r, encodingutf-8) as f: items json.load(f) results [] with ThreadPoolExecutor(max_workersargs.workers) as executor: futures {executor.submit(process_single, item): item for item in items} for future in as_completed(futures): item futures[future] try: results.append(future.result()) except Exception as e: results.append({ id: item[id], task_type: unknown, result: None, model_used: None, success: False, error: str(e) }) results.sort(keylambda x: x[id]) with open(args.output, w, encodingutf-8) as f: json.dump(results, f, ensure_asciiFalse, indent2) if __name__ __main__: main()这里有几个关键决策。第一用了ThreadPoolExecutor而不是串行循环。30条数据如果串行处理每条平均3秒总共90秒用3个并发理论上30秒就能跑完。但并发数不能设太高因为路由层和后端模型都有速率限制设成3到5比较稳妥。第二异常捕获放在了future.result()那一层。这样即使某条数据处理失败也不会影响其他数据的处理失败的那条会记录错误信息后续可以单独重跑。第三结果按id排序后再写入。并发处理的完成顺序是不确定的如果不排序回写到表格时就会乱序RPA那边匹配起来很麻烦。3.3 并发数、超时时间与重试次数的参数调优参数调优这部分我踩过不少坑。最开始我把并发数设成10想着越快越好结果路由层直接返回429限流一半的请求失败了。后来降到3稳定跑完但速度不够理想。最终我通过几次测试找到了一个平衡点。参数初始值调整后调整理由并发数103避免触发路由层限流单次超时30秒60秒复杂任务需要更长处理时间最大重试5次3次减少无效重试失败快速暴露退避基数1秒2秒给后端足够的恢复时间超时时间的设置需要根据任务类型区分。simple任务一般5秒内返回设30秒绰绰有余complex任务可能需要30到50秒设60秒比较安全。我在代码里没有做区分统一设了60秒因为超时设长了只是浪费等待时间设短了会导致不必要的重试后者代价更大。重试次数从5次降到3次是因为我发现很多失败是“确定性失败”——比如输入格式不对、API Key过期这种失败重试100次也没用。3次足够覆盖偶发的网络抖动和服务端瞬时故障。实操心得在正式跑30条数据之前先用2到3条数据做一次全流程测试确认路由策略、并发参数、结果回写都正常再批量跑。这个习惯帮我省了很多返工时间。4. 实测中遇到的五个典型问题与排查过程4.1 路由策略不生效请求全部走了默认模型第一次跑通脚本后我查看model_used字段发现30条数据全部走了model-general我配置的simple和complex路由完全没生效。这个问题排查了大概二十分钟。第一步检查请求体。确认task_type字段确实传了而且值和配置文件里的match条件一致。打印出来的请求体显示task_type是simple没问题。第二步检查路由配置。登录路由管理界面发现配置确实保存了但状态是“草稿”而不是“已发布”。原来蓝耘智能路由的配置需要显式发布才会生效我改完配置后直接关了页面没有点发布按钮。第三步发布配置后重新跑model_used字段开始出现不同的值了。这个问题说起来很简单但如果不检查路由层的配置状态光看脚本代码是找不到原因的。4.2 并发请求触发限流429错误的处理方式并发数设成10的那次日志里大量出现429状态码。429的意思是“请求过多”说明路由层或后端模型有速率限制。我当时的处理方式是简单重试但效果不好因为重试的请求又加剧了限流。正确的做法是“退避 降并发”。我在代码里加了指数退避同时把并发数从10降到3。另外路由层本身也有队列机制当请求超过速率限制时它会自动排队而不是直接拒绝。但这个队列有容量上限超过之后还是会返回429。所以最根本的解决办法还是控制并发数。后来我总结了一个经验对于30条数据这种规模并发数设3是最稳妥的。如果数据量增加到100条以上可以考虑设5但需要观察路由层的限流阈值。不同模型后端的限流策略不一样轻量模型通常宽松一些推理增强模型则比较严格。4.3 返回格式不一致不同模型的输出结构差异这个问题比较隐蔽。轻量模型和推理增强模型返回的内容格式不完全一样。轻量模型倾向于返回纯文本推理增强模型有时候会返回带Markdown格式的内容比如用**加粗关键词、用-列要点。我的脚本原本假设所有返回都是纯文本直接写回表格结果表格里出现了一堆Markdown符号。解决办法是在脚本里加一个后处理步骤根据model_used字段做格式归一化。如果模型返回了Markdown就用正则表达式去掉格式符号或者用strip()方法清理首尾空白。更稳妥的方式是在prompt里明确要求“返回纯文本不要使用任何Markdown格式”但实际测试下来模型并不总是遵守这个指令所以后处理还是必要的。import re def normalize_output(text): if text is None: return text re.sub(r\*\*(.?)\*\*, r\1, text) text re.sub(r^[-*]\s, , text, flagsre.MULTILINE) text re.sub(r(.?), r\1, text) return text.strip()这个归一化函数不追求完美只处理最常见的几种格式符号。如果你对输出格式要求很严格建议在路由层配置输出校验规则让不符合格式的返回触发质量降级。4.4 脚本在RPA中调用时路径与编码的坑RPA调用Python脚本时工作目录和编码是两个高频问题。我第一次在影刀RPA里调用脚本时报了“文件找不到”的错误。原因是RPA执行脚本时的工作目录不是脚本所在目录而是RPA自己的安装目录。解决办法是在RPA的命令里用绝对路径或者在脚本开头用os.chdir(os.path.dirname(os.path.abspath(__file__)))切换工作目录。编码问题更隐蔽。Windows系统默认编码是GBK而我的JSON文件是UTF-8。RPA写input.json时用了GBK脚本读取时用UTF-8结果中文全部乱码。解决办法是在RPA和脚本两边都显式指定UTF-8编码。RPA那边在写文件时设置编码参数脚本这边在open()里加encodingutf-8。注意如果你的数据里包含emoji或其他特殊字符GBK编码会直接报错。统一用UTF-8是最省心的做法。4.5 处理结果回写表格时的数据错位问题并发处理导致的结果乱序我在前面提过用排序解决。但还有一个更隐蔽的错位问题RPA读取output.json后按顺序逐条写回表格但如果某条数据在脚本里处理失败了output.json里仍然有这条记录只是result字段是null。RPA如果不做判断直接把null写进表格就会覆盖掉原始数据。我的处理方式是在RPA的回写环节加一个条件判断只有当success为true时才写回结果否则在备注列标记“处理失败”保留原始数据不动。这样即使部分数据失败也不会破坏原始表格的完整性。另外回写时的行号匹配也要注意。我用的是id字段做匹配而不是行号。因为RPA读取数据时可能做了筛选或排序行号不一定和原始表格一致。用唯一id匹配是最可靠的方式。5. 从30条到300条这套方案的扩展性与优化方向5.1 批量规模增大后的分片与队列策略30条数据用线程池并发处理没问题但如果数据量增加到300条甚至更多就需要考虑分片和队列了。一次性把300条全部提交到线程池会导致大量请求同时涌向路由层触发限流的概率大大增加。我的做法是把数据分成多个批次每批30到50条批次之间加一个短暂的间隔。这个逻辑可以用一个简单的循环实现def process_in_batches(items, batch_size30, interval5): all_results [] for i in range(0, len(items), batch_size): batch items[i:ibatch_size] batch_results process_batch(batch) all_results.extend(batch_results) if i batch_size len(items): time.sleep(interval) return all_results更优雅的方式是用消息队列比如把每条数据作为一个消息推送到队列里消费者按固定速率从队列取消息处理。这样天然实现了速率控制而且支持断点续传——如果处理到一半脚本崩溃了重启后从队列里继续取就行不需要重新处理已经完成的数据。对于300条这个规模我倾向于用“分片 断点续传”的方案。每处理完一批就把结果写入一个带时间戳的中间文件全部处理完后再合并。这样即使中途出问题也能从上次中断的地方继续。5.2 路由策略的精细化按成本、延迟、质量三维度权衡当数据量增大后路由策略的优化空间就体现出来了。30条数据时成本差异不明显300条数据时不同模型的调用成本差距可能达到数倍。这时候就需要在成本、延迟、质量之间做权衡。我的做法是给每个任务类型定义一个“优先级”。simple任务优先考虑成本和延迟complex任务优先考虑质量standard任务取中间值。路由层支持按优先级选择模型你可以在配置里给每个模型打上成本、延迟、质量的评分然后让路由层根据任务优先级自动选择。任务类型成本权重延迟权重质量权重推荐模型simple0.60.30.1轻量模型standard0.30.30.4通用模型complex0.10.20.7推理增强模型这个权重表不是固定的需要根据实际业务需求调整。如果你的业务对响应时间极其敏感就把延迟权重调高如果对准确性要求极高就把质量权重调高。5.3 日志、监控与失败重跑机制的设计数据量小的时候出错了看一眼日志就能定位。数据量大了之后没有完善的日志和监控排查问题就是大海捞针。我在脚本里加了结构化日志每条数据的处理过程都记录以下信息数据ID、任务类型、使用的模型、请求耗时、重试次数、最终状态。import logging import json logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(process.log, encodingutf-8), logging.StreamHandler() ] ) def log_result(item_id, task_type, model_used, elapsed, attempts, success): logging.info(json.dumps({ id: item_id, task_type: task_type, model: model_used, elapsed_ms: int(elapsed * 1000), attempts: attempts, success: success }, ensure_asciiFalse))有了这些日志你可以很容易地统计出哪个模型的平均耗时最长、哪类任务的重试率最高、哪个时间段失败率集中。这些数据是后续优化的依据。失败重跑机制也很重要。我通常会在全部处理完成后扫描一遍结果把success为false的记录提取出来单独重跑一次。重跑时可以把并发数降到1超时时间延长到120秒给这些“困难户”更宽松的处理条件。如果重跑仍然失败就标记为“需人工介入”不再自动重试。5.4 什么场景适合这套方案什么场景不适合这套“智能路由 RPA Python脚本”的方案最适合的场景是数据量在几十到几百条之间任务类型有差异对处理质量有一定要求但又不值得为此搭建一套完整的微服务架构。它的优势是搭建快、成本低、灵活可调。不适合的场景也很明确。如果数据量在10条以下手工处理可能比写脚本更快没必要上这套方案。如果数据量在几千条以上线程池和文件交换的方式就会成为瓶颈需要考虑消息队列和分布式处理。如果所有数据的类型完全一致不需要路由切换那直接用单一模型调用就行智能路由反而增加了复杂度。还有一个容易被忽略的点这套方案对RPA的稳定性有依赖。如果RPA流程本身不稳定比如界面元素定位经常失败、弹窗处理不完善那么即使模型调用部分做得再好整体流水线也会频繁中断。所以在实施这套方案之前先确保RPA流程本身是稳定的。6. 几个让我少走弯路的实操习惯第一个习惯是“先跑通一条再跑通一批”。不管方案设计得多完美先用一条数据走完全流程确认输入、路由、调用、返回、回写每个环节都正常再扩展到30条。这个习惯帮我发现了不少“设计时没想到、跑起来才暴露”的问题比如编码问题、路径问题、格式问题。第二个习惯是“保留原始数据不动”。处理结果写回表格时永远不要覆盖原始数据列。新增一列写结果再加一列写状态和使用的模型。这样即使处理出错原始数据还在可以随时重跑。第三个习惯是“给每条数据留痕”。model_used、attempts、elapsed_ms这些字段看起来不起眼但在优化阶段非常有用。你能清楚地知道每条数据走了哪个模型、重试了几次、花了多长时间。没有这些数据优化就是凭感觉。第四个习惯是“配置和代码分离”。路由策略、并发数、超时时间这些参数不要硬编码在脚本里放到配置文件或环境变量里。这样调整参数时不需要改代码也不需要重新测试脚本逻辑。第五个习惯是“失败是常态不是异常”。在分布式调用中网络抖动、服务限流、超时都是正常现象。脚本的设计目标不是“永远不失败”而是“失败后能自动恢复恢复不了能快速定位”。把失败处理当成核心逻辑的一部分来写而不是事后补丁。