1. 客户咨询自动拆分的整体设计思路1.1 为什么需要把一条咨询拆成多个任务做过企业微信二次开发的人都有一个共同感受客户发过来的一段话往往不是单一意图。比如客户说“你们这个产品多少钱另外能不能对接我们现有的ERP还有售后响应时间是多长”这一条消息里其实包含了三个完全不同的诉求——报价咨询、技术对接可行性、售后服务政策。如果只用一个接口去处理要么全部丢给人工要么用一个通用回复敷衍过去客户体验很差。我在实际项目中遇到过最典型的情况是一家做SaaS的客户他们的售前咨询量每天大概800到1200条其中超过60%的消息包含两个以上的意图。最初他们用的是一个简单的关键词匹配加人工分派结果就是客服每天要手动复制粘贴拆分需求效率极低而且经常漏掉信息。后来我们做了一套自动拆分加任务分发的机制客服只需要处理系统分派好的独立任务响应时间从平均15分钟降到了3分钟以内。这套方案的核心思路就是把非结构化的客户咨询消息通过意图识别和实体抽取拆分成多个结构化的接口任务每个任务独立路由到对应的处理接口去执行。这样做的好处是每个任务职责单一处理逻辑清晰也方便后续做数据统计和效果追踪。1.2 整体架构分层与数据流向整个系统我把它分成四层从下到上依次是接入层、解析层、调度层、执行层。接入层负责接收企业微信推送过来的消息事件这里通常用企业微信的接收消息回调接口配置好URL和Token之后客户发的每一条消息都会以XML或JSON格式推送到你的服务器。解析层是核心负责对消息内容做意图分类和实体抽取把一段话拆成多个带标签的任务对象。调度层根据任务类型和优先级决定每个任务走哪个接口、什么顺序执行、是否需要并行。执行层就是具体的业务接口比如报价查询接口、库存查询接口、工单创建接口等等。数据流向是这样的企业微信服务器推送消息到接入层接入层做基础校验和去重后把原始消息交给解析层。解析层调用意图识别模型或规则引擎输出一个任务列表每个任务包含任务类型、关键参数、优先级、原始文本片段。调度层拿到任务列表后根据预设的路由规则把每个任务分发到对应的执行接口。执行接口处理完之后把结果汇总回调度层调度层再决定是合并回复还是分条回复给客户。这个架构的关键在于解析层和调度层的解耦。解析层只负责“拆”不关心任务怎么执行调度层只负责“派”不关心任务怎么解析。这样后续要增加新的任务类型只需要在解析层加规则在调度层加路由配置不用动执行层的代码。1.3 方案选型规则引擎还是模型推理这是很多人纠结的地方。我的建议是初期用规则引擎快速跑通中期引入轻量级模型做补充长期看数据量决定是否上大模型。为什么这么说规则引擎的优势是可控、可解释、响应快一条消息进来正则匹配加关键词命中几毫秒就能出结果。但规则引擎的短板也很明显客户说话的方式千变万化你不可能穷举所有表达。我试过纯规则方案一开始命中率能到85%左右但剩下15%的长尾表达需要不断加规则维护成本越来越高。后来我们引入了一个轻量级的文本分类模型用历史工单数据做训练把意图分类的准确率提升到了93%以上。模型负责粗分类规则负责细粒度的实体抽取两者结合效果最好。至于大模型如果你的咨询量每天超过5000条而且预算允许可以考虑用大模型做意图理解和任务拆分。但要注意大模型的响应延迟通常在1到3秒对于实时性要求高的场景需要做异步处理。另外大模型的输出格式需要做严格的校验不能直接信任它的返回结果。2. 核心细节解析与实操要点2.1 企业微信消息接收与去重处理企业微信的消息回调有个坑就是同一条消息可能会推送多次。如果你不做去重客户发一句话系统可能创建了三个重复任务。去重的逻辑很简单企业微信推送的消息体里有一个MsgId字段你把这个MsgId存到Redis里设置一个5分钟的过期时间每次收到消息先查一下这个MsgId是否已经处理过。如果已经存在直接返回成功不再往下走。具体实现上企业微信回调的URL需要配置在企业微信管理后台的应用管理页面。你需要提供一个GET接口用于URL验证一个POST接口用于接收消息。GET接口收到请求后需要把企业微信传来的echostr原样返回同时校验msg_signature。POST接口收到消息后先做签名校验然后解析XML提取MsgId、FromUserName、Content等字段。import hashlib import xml.etree.ElementTree as ET from redis import Redis redis_client Redis(hostlocalhost, port6379, db0) def verify_signature(token, timestamp, nonce, signature): sort_list sorted([token, timestamp, nonce]) sort_str .join(sort_list) hash_str hashlib.sha1(sort_str.encode()).hexdigest() return hash_str signature def handle_message(xml_data): root ET.fromstring(xml_data) msg_id root.find(MsgId).text content root.find(Content).text from_user root.find(FromUserName).text # 去重检查 cache_key fwx_msg:{msg_id} if redis_client.exists(cache_key): return success redis_client.setex(cache_key, 300, 1) # 继续处理 return process_content(content, from_user)注意Redis的过期时间不要设置太长5分钟足够了。设置太长会占用内存设置太短可能导致重复消息漏过去。另外MsgId在企业微信的不同应用之间是全局唯一的所以可以直接用MsgId作为key。2.2 意图识别与实体抽取的工程实现意图识别我建议分成两级第一级是粗分类判断这条消息属于哪个大类比如售前咨询、售后问题、投诉建议、合作洽谈。第二级是细分类在售前咨询下面再分报价咨询、功能咨询、对接咨询等。粗分类可以用关键词加权的方式快速实现细分类用模型或者更复杂的规则。实体抽取的重点是提取任务执行所需的关键参数。比如报价咨询需要提取产品名称、数量、客户等级对接咨询需要提取对方系统名称、对接方式、数据量级。这些参数如果提取不到任务执行接口就没法正常工作。我常用的做法是维护一个实体词典每个实体类型对应一组同义词和正则模式。比如产品名称我会把公司所有产品线名称、简称、常见错别字都列进去。客户说“你们那个CRM系统”我能匹配到“CRM”这个产品客户说“客户管理系统”我也能通过同义词映射到“CRM”。import re ENTITY_PATTERNS { product: { patterns: [rCRM|客户管理系统|客户管理, rERP|进销存|库存管理], mapping: {客户管理系统: CRM, 进销存: ERP} }, quantity: { patterns: [r(\d)\s*(个|套|人|用户|账号)], mapping: {} } } def extract_entities(text): entities {} for entity_type, config in ENTITY_PATTERNS.items(): for pattern in config[patterns]: matches re.findall(pattern, text) if matches: value matches[0] if isinstance(matches[0], str) else matches[0][0] if value in config[mapping]: value config[mapping][value] entities[entity_type] value break return entities实操心得实体词典一定要做成可配置的不要硬编码在代码里。我们后来把词典放在数据库里运营人员可以自己维护新增产品线的时候不用找开发改代码。另外正则匹配要注意贪婪匹配的问题比如“我要买两个CRM系统和一个ERP”如果不加限制数量可能匹配到“两个”也可能匹配到“一个”需要根据业务逻辑做优先级排序。2.3 任务拆分策略与优先级排序任务拆分不是简单地按句子切分。客户说“帮我查一下上个月的订单另外把发票寄到新地址”这里有两个任务订单查询和地址变更。但如果你按句号切分第一句是订单查询第二句是地址变更看起来没问题。但客户如果说“帮我查一下上个月的订单然后寄发票”这是一句话但包含两个任务。我的做法是基于意图标签做拆分而不是基于句子边界。先用意图识别模型对整条消息做多标签分类识别出所有可能的意图然后针对每个意图从原文中抽取对应的文本片段和实体参数。这样即使两个意图混在一句话里也能正确拆分。优先级排序的规则一般是投诉类最高售后问题次之售前咨询再次合作洽谈最低。同一优先级内按消息到达时间排序。为什么要做优先级因为客服资源是有限的如果所有任务都平等对待投诉客户等了半小时没人理售前咨询却秒回这显然不合理。任务类型优先级建议响应时间路由目标投诉建议P01分钟内客服主管售后问题P13分钟内售后专员售前咨询P25分钟内售前客服合作洽谈P330分钟内商务经理其他P42小时内通用客服2.4 接口任务的幂等性与重试机制任务拆分出来之后每个任务都要调用对应的业务接口去执行。这里有一个很容易被忽略的问题接口调用失败怎么办网络抖动、下游服务超时、数据库连接池满了这些情况在生产环境里太常见了。如果不做重试任务就丢了如果无脑重试可能造成重复下单、重复发券等严重后果。我的方案是每个任务生成一个全局唯一的task_id执行接口必须支持幂等。具体来说执行接口收到请求后先查一下这个task_id是否已经执行过如果执行过直接返回上次的结果。这样重试的时候就不会产生副作用。重试策略我一般用指数退避第一次失败等1秒重试第二次等2秒第三次等4秒最多重试3次。3次都失败就写入死信队列人工介入处理。import time import requests def execute_task_with_retry(task, max_retries3): for attempt in range(max_retries): try: response requests.post( task[endpoint], jsontask[payload], headers{X-Task-Id: task[task_id]}, timeout5 ) if response.status_code 200: return response.json() except requests.RequestException as e: if attempt max_retries - 1: send_to_dead_letter(task, str(e)) raise time.sleep(2 ** attempt) return None注意幂等键的设计很关键。不要用业务字段做幂等键比如订单号因为订单号可能还没生成。用task_id最稳妥task_id在任务创建时就生成好贯穿整个执行链路。3. 实操过程与核心环节实现3.1 从零搭建消息接收服务先说一下环境准备。你需要一台公网可访问的服务器因为企业微信的回调必须走公网。服务器上装好Python 3.8以上版本Redis用于去重和任务队列MySQL或者PostgreSQL用于持久化任务数据。如果任务量不大Redis的List结构就够用了如果任务量很大建议上RabbitMQ或者Kafka。第一步是在企业微信管理后台创建应用。进入“应用管理”点击“创建应用”填好应用名称和Logo创建完成后你会拿到AgentId和Secret。然后在“接收消息”区域设置API接收URL填你的服务器地址Token和EncodingAESKey随机生成记下来后面要用。第二步是写URL验证接口。企业微信会发一个GET请求过来带上msg_signature、timestamp、nonce、echostr四个参数。你需要用Token、timestamp、nonce计算签名和msg_signature比对一致的话把echostr解密后返回。from flask import Flask, request import hashlib app Flask(__name__) TOKEN your_token_here app.route(/wechat/callback, methods[GET]) def verify_url(): msg_signature request.args.get(msg_signature) timestamp request.args.get(timestamp) nonce request.args.get(nonce) echostr request.args.get(echostr) sort_list sorted([TOKEN, timestamp, nonce]) sort_str .join(sort_list) hash_str hashlib.sha1(sort_str.encode()).hexdigest() if hash_str msg_signature: return echostr return 验证失败, 403第三步是写消息接收接口。POST请求进来后先做签名校验然后解析XML提取消息内容做去重然后丢到任务队列里异步处理。为什么要异步因为企业微信要求回调接口在5秒内返回如果你在回调里做意图识别、调接口、等结果很容易超时。超时后企业微信会重试导致重复处理。3.2 意图识别模块的落地细节意图识别模块我建议单独部署成一个服务通过HTTP或者gRPC对外提供接口。这样做的好处是可以独立扩缩容而且不同语言的服务都能调用。接口设计很简单输入是文本输出是意图列表和实体列表。// 请求 { text: 你们CRM多少钱另外能对接钉钉吗, user_id: zhangsan } // 响应 { tasks: [ { task_type: price_inquiry, confidence: 0.95, entities: {product: CRM}, text_segment: 你们CRM多少钱 }, { task_type: integration_inquiry, confidence: 0.88, entities: {target_system: 钉钉}, text_segment: 另外能对接钉钉吗 } ] }模型训练方面如果你没有标注数据可以先用手写规则跑一段时间把规则命中的结果作为弱标注数据人工修正一部分后训练模型。我们当时大概积累了5000条标注数据模型准确率就稳定在90%以上了。标注的时候要注意一条消息如果有多个意图要全部标出来不能只标一个。实操心得意图识别的置信度阈值不要设太高。设太高会漏掉很多任务设太低会引入噪音。我的经验是0.7到0.8之间比较合适。低于阈值的任务不要直接丢弃可以标记为“待确认”让客服人工判断一下。这样既不漏也不会给客服增加太多负担。3.3 任务调度与并行执行任务调度层我推荐用Celery或者RQ这样的分布式任务队列。每个任务作为一个独立的job提交到队列里worker从队列里取任务执行。这样做的好处是任务之间互不阻塞一个任务执行慢不会影响其他任务。并行执行的时候要注意资源竞争问题。比如两个任务同时去查同一个客户的订单可能会给数据库造成压力。我的做法是在调度层加一个简单的限流同一个客户的任务串行执行不同客户的任务并行执行。这样既保证了客户维度的顺序性又保证了整体吞吐量。from celery import Celery app Celery(tasks, brokerredis://localhost:6379/0) app.task(bindTrue, max_retries3) def execute_task(self, task): try: result call_business_api(task) save_task_result(task[task_id], result) return result except Exception as exc: raise self.retry(excexc, countdown2 ** self.request.retries)任务执行结果需要汇总。我的做法是每个任务执行完后把结果写入一个结果表同时更新主任务的完成状态。当所有子任务都完成后触发一个汇总回调把结果合并后回复给客户。如果某个子任务超时未完成汇总回调也要能处理不能无限等待。3.4 结果合并与客户回复结果合并的策略取决于任务类型。如果是多个查询类任务直接把结果拼接起来回复就行。如果是查询加操作类任务比如“查一下订单然后取消订单”那就要注意顺序先查再取消不能并行。我在调度层给每个任务加了一个sequence字段同一个客户的任务按sequence排序执行。回复客户的时候企业微信支持文本、图文、模板卡片等多种消息类型。我一般用文本消息做快速回复用模板卡片做结构化信息的展示。比如报价咨询的回复用模板卡片展示产品名称、价格、有效期比纯文本清晰得多。def reply_customer(user_id, results): if len(results) 1: send_text_message(user_id, results[0][text]) else: merged_text \n\n.join([r[text] for r in results]) send_text_message(user_id, merged_text)注意企业微信的客服消息有频率限制同一个客户短时间内不要发太多条消息。如果任务很多建议合并成一条消息发送或者用模板卡片把多个结果放在一张卡片里。4. 常见问题与排查技巧实录4.1 消息重复推送导致任务重复创建这是最常见的问题没有之一。企业微信在回调超时或者网络异常时会重试推送同一条消息。如果你没有做去重客户发一句话系统创建了多个重复任务客服就会收到多条重复的工单。排查方法很简单在消息接收接口里打印MsgId看看同一个MsgId是不是出现了多次。如果是检查你的去重逻辑是不是生效了。常见的原因有三个一是Redis连接失败去重检查直接跳过了二是去重key的过期时间设得太短消息重试的时候key已经过期了三是去重逻辑写在了异步任务里而不是在接收接口里同步执行。解决方案就是前面说的在接收接口里用Redis做同步去重key的过期时间设5分钟。另外要注意企业微信的MsgId在同一个应用内是唯一的但不同应用之间可能重复所以key要加上AgentId前缀。4.2 意图识别准确率低导致任务拆分错误意图识别准确率低的表现是该拆的任务没拆出来或者不该拆的拆了一堆。比如客户说“我要投诉你们客服态度差”结果系统识别成了“售后问题”和“投诉建议”两个任务其实就是一个投诉任务。排查的时候先把识别错误的case收集起来看看是规则问题还是模型问题。如果是规则问题检查关键词列表是不是有遗漏或者冲突。如果是模型问题看看训练数据里这类表达是不是太少。我遇到过一个典型情况客户说“你们这个多少钱”模型识别成了“功能咨询”因为训练数据里“多少钱”的样本都被标成了“功能咨询”。后来补充了报价咨询的样本问题就解决了。避坑技巧意图识别一定要做badcase分析。每周抽100条识别结果人工检查一遍把错误的挑出来分析原因。坚持做一个月准确率至少提升10个百分点。4.3 接口调用超时导致任务卡死接口调用超时是分布式系统的常态。下游服务响应慢、网络抖动、数据库锁表都会导致超时。如果不做超时控制任务就会一直卡在那里占用worker资源。我的做法是给每个接口调用设置明确的超时时间一般3到5秒。超时后触发重试重试3次还失败就写入死信队列。死信队列里的任务需要人工介入但至少不会阻塞正常任务。另外要注意重试的时候要区分可重试错误和不可重试错误。网络超时、连接拒绝这些可以重试参数错误、权限不足这些重试也没用直接失败就行。我在任务对象里加了一个retryable字段调度层根据这个字段决定是否重试。问题现象可能原因排查方法解决方案任务重复创建消息重复推送检查MsgId是否重复Redis同步去重任务拆分错误意图识别不准收集badcase分析补充规则或训练数据任务卡死接口超时查看worker日志设置超时和重试结果丢失汇总逻辑异常检查结果表增加补偿机制回复延迟任务队列积压查看队列长度增加worker数量4.4 企业微信接口频率限制的应对企业微信的接口调用有频率限制比如发送消息接口每个应用每分钟最多调用600次。如果你的任务量很大很容易触发限流。触发限流后接口会返回错误码45009这时候不要无脑重试要等一段时间再试。我的做法是在调度层加一个令牌桶限流器控制发送消息的速率。令牌桶的容量设为500每秒补充10个令牌。这样既能充分利用配额又不会触发限流。另外如果多个任务的结果可以合并成一条消息发送尽量合并减少接口调用次数。import time class TokenBucket: def __init__(self, capacity, rate): self.capacity capacity self.tokens capacity self.rate rate self.last_time time.time() def consume(self, num1): now time.time() self.tokens min(self.capacity, self.tokens (now - self.last_time) * self.rate) self.last_time now if self.tokens num: self.tokens - num return True return False实操心得限流器要放在发送消息之前而不是在任务执行之前。因为任务执行可能很快但发送消息受限于企业微信的配额。把限流器放在发送环节可以最大化任务执行的吞吐量。4.5 数据一致性与补偿机制任务拆分之后主任务和子任务的状态需要保持一致。如果子任务执行成功了但主任务状态没更新就会出现数据不一致。我的做法是用数据库事务保证主任务和子任务的创建是原子的子任务执行结果通过消息队列异步更新主任务状态。如果更新失败需要一个补偿机制。我一般会起一个定时任务每5分钟扫描一次状态为“执行中”但超过10分钟没有更新的主任务重新触发状态同步。这个补偿机制虽然简单但能解决90%以上的数据不一致问题。另外任务结果表要保留至少30天的数据方便排查问题。如果数据量太大可以按月分表。我见过一个客户任务结果表没有分表半年后单表数据量过亿查询慢得没法用。后来做了分表才解决。4.6 多应用场景下的任务隔离很多公司不止一个企业微信应用可能有客服应用、内部审批应用、通知应用等等。如果所有应用的消息都走同一套任务拆分逻辑很容易互相干扰。比如内部审批的消息被识别成了客户咨询创建了不该创建的任务。我的做法是按AgentId做隔离。每个应用有独立的配置包括意图识别规则、任务路由规则、优先级策略。消息进来后先根据AgentId找到对应的配置然后再走后续流程。这样不同应用之间互不影响也方便单独调整某个应用的策略。配置隔离的实现方式很简单用一张配置表字段包括AgentId、config_key、config_value。启动的时候把配置加载到内存里用AgentId作为key。配置变更的时候发一个消息通知各节点刷新缓存。这套系统我们跑了大概一年半中间迭代了十几个版本。最大的体会是不要追求一步到位先跑通最小闭环再逐步优化。一开始规则粗糙一点没关系关键是让整个流程转起来然后在实际运行中不断调整。意图识别准确率从最初的70%提升到现在的95%靠的不是一次性设计得多完美而是持续地badcase分析和规则迭代。