1. 从一条数据到三十条数据：为什么需要智能路由

做过RPA（机器人流程自动化）的人都有一个共识：单条数据的处理脚本好写，三十条数据的批量处理才是真正的考验。我最早接触RPA是在一个数据录入场景里，当时的需求很简单——把一批表单数据逐条填入某个业务系统。单条跑通花了不到半小时，但当我尝试连续处理三十条时，问题就全冒出来了。

最典型的坑是模型调用。很多RPA脚本在处理每条数据时都需要调用AI模型来做内容理解、分类或者摘要生成。如果三十条数据全部走同一个模型，会遇到两个极端：要么全部用轻量模型，速度快但复杂数据的处理质量堪忧；要么全部用重量级模型，质量上去了但时间和成本成倍增长。更麻烦的是，有些数据本身就简单到不需要模型介入，用规则就能搞定，硬调模型纯属浪费。

这就是智能路由要解决的核心问题。所谓智能路由，说白了就是在脚本运行过程中，根据当前数据的特征，自动判断该走哪条处理路径——是用轻量模型快速过，还是用重量级模型精细处理，或者干脆走本地规则引擎不调模型。整个过程对RPA脚本来说是透明的，脚本只管把数据交给路由层，路由层负责决策和分发。

蓝耘智能路由在这个场景里的定位，就是充当RPA脚本和底层模型之间的调度中间件。它对外暴露统一的调用接口，对内维护一个模型池和一套路由策略。你可以在策略里定义：什么样的数据走什么模型、什么条件下降级到规则处理、并发请求怎么分配。RPA脚本不需要关心背后有几个模型、每个模型的接口差异是什么，只需要按统一格式发请求就行。

这套方案适合谁？我认为三类人最需要：一是正在做RPA批量数据处理但被模型调用效率困扰的开发者；二是需要在成本和质量之间找平衡的团队；三是希望脚本具备一定自适应能力、不想为每种数据类型写一套逻辑的工程师。哪怕你之前没接触过智能路由，只要写过RPA脚本、调过模型API，这篇文章的内容都能直接参考。

2. 整体架构设计：路由层到底怎么嵌入RPA流程

2.1 为什么不在脚本里写if-else

很多人第一反应是：我在RPA脚本里写几个if-else不就行了？数据长度超过多少走大模型，否则走小模型。这个思路在数据量少、规则简单的时候确实能用，但三十条数据一上来，问题就暴露了。

首先是规则维护成本。if-else写多了之后，脚本会变得极其臃肿。你今天加一条“包含表格的数据走大模型”，明天加一条“纯文本且字数少于100的走规则”，后天又要加“带图片的走多模态模型”。每加一条规则，都要改动脚本主体逻辑，测试成本也跟着涨。

其次是并发和容错。三十条数据如果串行处理，每条都等模型返回，总耗时就是三十次调用的累加。如果某条数据的模型调用超时了，脚本是重试还是跳过？重试几次？这些逻辑写在RPA脚本里会让代码变得非常难维护。

智能路由层的价值就在于把这些决策逻辑从脚本里抽出来，放到一个独立的中间层。RPA脚本只负责“取数据、发请求、收结果、写回”，路由层负责“判断走哪条路、调哪个模型、失败了怎么办”。职责分离之后，脚本变得干净，路由策略可以独立调整，互不影响。

2.2 路由层的三个核心模块

我设计的路由层包含三个核心模块，每个模块各司其职。

第一个是特征提取模块。每条数据进来之后，先做一次轻量分析，提取出用于路由决策的特征。这些特征包括：数据长度（字符数或token数）、数据类型（纯文本、结构化字段、混合内容）、是否包含特定关键词、历史处理记录等。特征提取本身不调模型，纯靠本地计算，耗时在毫秒级。

第二个是策略引擎模块。拿到特征之后，策略引擎按照预定义的规则决定路由目标。策略可以用简单的阈值判断，也可以用打分机制。比如给每个模型算一个“适配分”，选分最高的。策略引擎还负责处理降级逻辑：如果目标模型不可用，自动切换到备用模型或规则引擎。

第三个是执行与回写模块。路由决策做完之后，执行模块负责调用对应的模型接口，拿到结果后统一格式化，再返回给RPA脚本。这个模块还要处理超时、重试、并发控制等工程问题。

这三个模块串起来，就形成了一条完整的处理链路。RPA脚本调一次路由接口，路由层内部完成特征提取、策略决策、模型调用、结果回写，脚本拿到的是最终结果，中间过程完全透明。

2.3 模型池的选型与配置

模型池里放哪些模型，直接决定了路由的灵活性和效果。我的建议是至少准备三档：轻量档、标准档、增强档。

轻量档选响应速度最快、成本最低的模型，适合处理短文本分类、简单抽取这类任务。标准档选综合能力均衡的模型，适合大多数常规数据处理。增强档选能力最强的模型，用于处理复杂推理、长文本理解、多步骤任务。

除了模型档位，还要考虑规则引擎作为零成本路径。有些数据根本不需要模型，比如格式固定的字段提取、简单的关键词匹配，用正则或字符串处理就能搞定。把这类数据路由到规则引擎，既省时间又省成本。

配置的时候要注意：每个模型都要配好超时时间、最大重试次数、并发上限。超时时间根据模型的实际响应速度来定，一般轻量档设短一些（比如5秒），增强档设长一些（比如30秒）。并发上限要根据你的账户配额来设，避免触发限流。

3. 核心细节解析：特征提取与策略引擎的实现要点

3.1 特征提取：哪些信号真正影响路由决策

特征提取是路由的起点，提取哪些特征直接决定了路由的准确性。我踩过的坑是：一开始提取了太多特征，结果策略引擎变得极其复杂，维护成本飙升。后来精简到四个核心特征，效果反而更好。

第一个特征是数据长度。这是最直观也最有效的信号。我的经验值是：少于50个字符的数据，大概率可以用规则引擎处理；50到500字符之间的，走轻量档模型；超过500字符的，考虑标准档或增强档。当然这个阈值不是固定的，要根据你的实际数据分布来调整。

第二个特征是内容复杂度。怎么判断复杂度？我用的方法很简单：看数据里是否包含嵌套结构、是否有多层条件、是否需要跨字段推理。比如一条数据只是“姓名：张三，年龄：30”，复杂度极低；但如果数据是“根据以下三条记录判断该用户的风险等级”，复杂度就高了。复杂度高的数据直接路由到增强档。

第三个特征是历史处理结果。如果同类型的数据之前用轻量档处理过，但人工复核发现质量不达标，那这类数据下次就应该路由到更高档位。这个特征需要路由层维护一个简单的反馈记录表，记录每类数据的处理质量和路由档位。

第四个特征是实时负载。如果增强档模型当前并发已满，新来的数据要么排队，要么降级到标准档。实时负载特征让路由层具备一定的弹性，避免在高峰期把所有请求都压到同一个模型上。

这四个特征提取出来之后，组成一个特征向量，传给策略引擎做决策。特征提取的代码不复杂，但要注意：提取过程本身不能太慢，否则会成为整个链路的瓶颈。我的做法是用轻量级的字符串操作和正则，避免在特征提取阶段做复杂的NLP处理。

3.2 策略引擎：阈值判断还是打分机制

策略引擎的实现方式有两种主流选择：阈值判断和打分机制。

阈值判断就是写一系列if-else规则，比如“如果长度大于500且复杂度为高，走增强档”。这种方式实现简单、调试直观，适合规则明确的场景。缺点是规则之间的优先级和边界情况需要仔细处理，规则多了之后容易冲突。

打分机制是给每个模型算一个适配分，选分最高的。比如轻量档的基础分是60，如果数据长度小于100加20分，如果复杂度低加15分，最终得分95；增强档基础分50，长度大于500加30分，复杂度高加25分，最终得分105，所以选增强档。打分机制的好处是规则之间不会硬冲突，调整权重就能改变路由倾向。

我最终选的是打分机制为主、阈值判断为辅的混合方案。大部分情况用打分，但在一些明确的边界条件下用阈值直接短路。比如数据长度超过2000字符，直接走增强档，不参与打分，因为这种数据用轻量档基本不可能处理好。

策略引擎还有一个关键设计：可配置化。路由策略不应该硬编码在代码里，而应该放在配置文件或数据库里。这样调整策略不需要改代码、不需要重新部署，改配置就能生效。我用的是一个简单的JSON配置文件，里面定义了每个模型的权重、各个特征的加分规则、阈值短路的条件。

3.3 降级与容错：模型不可用时怎么办

三十条数据批量处理，最怕的就是某条数据卡住导致整个流程停滞。降级与容错机制是保证流程稳定运行的关键。

我的降级策略分三层。第一层是同档位切换：如果增强档的某个模型超时或返回错误，自动切换到同档位的另一个模型。比如增强档配了两个模型，A不可用就走B。第二层是降档处理：如果同档位都没有可用模型，降级到下一档。增强档不可用就降标准档，标准档不可用就降轻量档。第三层是规则兜底：如果所有模型都不可用，走本地规则引擎做最基础的处理，至少保证流程不中断。

容错方面，每条数据的处理都要有独立的超时控制和重试计数。重试次数建议设为2到3次，每次重试之间加一个短暂的等待（比如1秒），避免瞬间重试导致连续失败。如果重试次数用尽仍然失败，记录错误日志，跳过这条数据继续处理下一条，最后统一汇总失败列表。

注意：降级处理的结果质量可能不如预期，所以降级发生时一定要记录日志。后续可以针对降级处理的数据做人工复核，判断是否需要调整路由策略。

4. 实操过程：三十条数据的完整处理流程

4.1 环境准备与依赖安装

先说一下我的运行环境：Python 3.10，主要依赖三个库——requests用于调用模型接口，pyyaml用于读取配置文件，concurrent.futures用于并发控制。RPA部分我用的是一个常见的自动化工具，通过它的Python脚本节点来调用路由层。

安装依赖很简单：

pip install requests pyyaml

路由层的代码结构我分成了四个文件：feature_extractor.py负责特征提取，strategy_engine.py负责策略决策，model_caller.py负责模型调用，router.py是对外暴露的统一入口。RPA脚本只需要importrouter，调用它的process方法就行。

配置文件router_config.yaml里定义了模型池和策略参数。模型池部分列出每个模型的名称、接口地址、档位、超时时间、并发上限。策略部分定义打分规则和阈值短路条件。这个文件是路由层的大脑，调整路由行为只需要改这个文件。

4.2 特征提取模块的代码实现

特征提取模块的核心是一个函数，输入是原始数据字符串，输出是一个包含四个特征的字典。代码不复杂，但有几个细节要注意。

import re def extract_features(data: str) -> dict: length = len(data) # 复杂度判断：包含嵌套结构或多层条件 complexity_score = 0 if re.search(r'[{}[\]()]', data): complexity_score += 1 if data.count('如果') + data.count('条件') > 2: complexity_score += 1 if re.search(r'\d+[.、]\s*.+\n\d+[.、]', data): complexity_score += 1 # 历史质量反馈（从缓存读取，这里简化为默认值） history_quality = get_history_quality(data[:20]) # 实时负载（从模型调用模块获取） current_load = get_current_load() return { 'length': length, 'complexity': complexity_score, 'history_quality': history_quality, 'load': current_load }

这里有个细节：get_history_quality是根据数据的前20个字符做模糊匹配，查历史记录里同类数据的处理质量。如果之前同类数据用轻量档处理过且质量评分低于阈值，这次就会倾向于路由到更高档位。这个机制让路由层具备了一定的学习能力。

4.3 策略引擎的打分与决策逻辑

策略引擎拿到特征字典后，对每个候选模型算一个适配分。我定义的打分规则是这样的：

每个模型的基础分都是50，加上特征加分后得到最终得分。得分最高的模型就是路由目标。如果最高分低于一个阈值（比如40分），说明所有模型都不太合适，走规则引擎兜底。

阈值短路条件单独处理：长度超过2000字符直接走增强档，长度小于20字符且复杂度为0直接走规则引擎。这两个条件在打分之前判断，命中就直接返回，不参与打分。

4.4 模型调用与并发控制

模型调用模块负责实际发起请求。三十条数据如果串行调用，总耗时可能是几分钟甚至十几分钟。我用concurrent.futures.ThreadPoolExecutor做并发，最大并发数设为5。这个数字不是随便定的：并发太高容易触发模型接口的限流，并发太低又发挥不出批量处理的优势。5是一个比较稳妥的起点，你可以根据实际配额调整。

from concurrent.futures import ThreadPoolExecutor, as_completed def process_batch(data_list, max_workers=5): results = [] with ThreadPoolExecutor(max_workers=max_workers) as executor: future_to_data = { executor.submit(process_single, data): data for data in data_list } for future in as_completed(future_to_data): data = future_to_data[future] try: result = future.result(timeout=60) results.append(result) except Exception as e: results.append({'error': str(e), 'data': data}) return results

process_single函数内部就是完整的路由流程：提取特征、策略决策、调用模型、返回结果。每条数据的处理是独立的，一条失败不影响其他条。

4.5 完整运行记录与结果分析

我用三十条模拟数据跑了一次完整流程。数据分布是这样的：10条短文本（长度小于50），15条中等长度文本（100到400字符），5条长文本（超过800字符且包含嵌套结构）。

路由结果：10条短文本中有8条走了规则引擎，2条因为包含特定关键词走了轻量档；15条中等长度文本全部走了轻量档；5条长文本全部走了增强档。总耗时约12秒，其中增强档的5条数据耗时占了8秒左右。

质量方面，规则引擎处理的8条数据全部正确；轻量档处理的17条数据中有1条质量不达标，人工复核后发现是数据本身有歧义，跟模型选择无关；增强档处理的5条数据全部正确。整体准确率超过96%。

成本方面，如果三十条全部走增强档，成本大约是现在的4倍。智能路由把大部分简单数据分流到了低成本路径，省下来的成本相当可观。

5. 常见问题与排查技巧实录

5.1 路由决策不符合预期怎么排查

这是最常见的问题。你明明觉得某条数据应该走增强档，结果路由到了轻量档。排查思路分三步。

第一步，看特征提取结果。在路由层加一个调试日志，把每条数据的特征字典打印出来。很多时候是特征提取出了问题，比如长度计算把空格也算进去了，或者复杂度判断的正则没匹配上。

第二步，看打分详情。把每个模型的各项加分和最终得分都打出来。这样你能清楚看到是哪个特征的加分导致了决策偏移。常见原因是某个特征的权重设得过高，压过了其他特征。

第三步，检查阈值短路。如果特征和打分都正常，但决策还是不对，看看是不是命中了阈值短路条件。阈值短路的优先级最高，一旦命中就不参与打分。

5.2 模型调用超时和限流的处理

超时和限流是批量处理中最常见的工程问题。我的处理原则是：超时必重试，限流必等待。

超时重试的策略是：第一次超时后等1秒重试，第二次超时后等3秒重试，第三次超时后放弃并降级。重试次数不要超过3次，否则单条数据的处理时间会拖得很长。

限流的处理稍微不同。限流通常返回一个特定的状态码或错误信息，收到限流信号后不能立即重试，而是要等待一段时间。我用的策略是：收到限流后等待5秒，然后重试；如果再次限流，等待时间翻倍（10秒、20秒），最多等3次。同时，限流发生后要把并发数临时调低，避免继续触发限流。

提示：限流等待期间，其他数据的处理不受影响，因为每条数据是独立线程处理的。但要注意线程池的总并发数，避免等待的线程占满了池子导致新任务无法提交。

5.3 常见问题速查表

5.4 几个我踩过的坑

第一个坑：特征提取太慢。我一开始在特征提取里加了一个调用外部接口获取历史质量的操作，结果每条数据的特征提取就要花2秒。后来改成从本地缓存读取，耗时降到毫秒级。特征提取一定要用本地计算，不要引入外部依赖。

第二个坑：并发数设得过高。我试过把并发数设到20，结果模型接口直接限流，三十条数据有一半失败了。后来降到5，稳定运行。并发数不是越高越好，要根据接口的承受能力来定。

第三个坑：降级后没有记录。早期版本降级处理之后没有记日志，导致我根本不知道哪些数据走了降级路径。后来加了降级日志，才发现增强档在高峰期经常不可用，降级率高达30%。根据这个数据我调整了增强档的并发配额，降级率降到了5%以下。

第四个坑：配置文件没有版本管理。路由策略调整了好几次，每次改完没有记录改了什么，后来出了问题想回滚都找不到之前的配置。现在我把配置文件也纳入版本管理，每次调整都记清楚改了什么、为什么改。

6. 路由策略的调优与扩展思路

6.1 根据实际数据分布调整阈值

路由策略里的阈值不是拍脑袋定的，要根据实际数据的分布来调。我的做法是：先跑一批数据，统计长度分布和复杂度分布，然后根据分布来定阈值。

比如你的数据里80%都是短文本，那轻量档和规则引擎的阈值就应该设得宽一些，让大部分数据走低成本路径。如果数据里大部分是长文本，那增强档的权重就要提高，避免大量数据在轻量档处理不好再返工。

调整阈值的频率不用太高，建议每处理完一批新数据后review一次。如果发现某类数据的路由结果和预期偏差较大，就针对性调整。

6.2 引入反馈闭环让路由自我优化

路由层目前是根据预设规则做决策，还没有自我学习能力。一个可行的扩展方向是引入反馈闭环：每次处理完数据后，记录路由决策和最终质量评分，定期分析哪些决策是好的、哪些是坏的，自动调整打分权重。

实现方式可以很简单：维护一个决策记录表，每条记录包含特征、路由目标、质量评分。每周跑一次分析，统计每个特征条件下各档位的平均质量分，把质量分高的档位权重调高。这样路由策略会随着数据积累越来越准。

6.3 扩展到更多模型和更复杂的场景

目前模型池只有三档，未来可以扩展到更多档位，比如增加一个“多模态档”处理带图片的数据，或者增加一个“代码专用档”处理代码相关的内容。路由策略也可以从单目标路由扩展到多目标路由，比如同时考虑质量和成本，在满足质量要求的前提下选成本最低的模型。

场景方面，这套路由架构不仅适用于RPA数据处理，还可以用在内容审核、智能客服、文档摘要等场景。核心思路是一样的：根据输入特征自动选择最合适的处理路径。只要把特征提取和策略引擎适配到具体场景，路由层就能复用。

我在实际使用中最大的体会是：智能路由的价值不在于用了多复杂的算法，而在于把“选择”这件事从业务脚本里抽出来，变成一个可配置、可观测、可优化的独立层。三十条数据的批量处理只是一个起点，当数据量到三百条、三千条的时候，这套架构的优势会更加明显。