做同城配送这两年最磨人的不是路上堵车而是每天几百条散落在微信群里的订单消息。客户不会乖乖按格式填单他们习惯发“明天早上送三十箱水到XX店到了打这个电话139xxxx”这种口语化消息调度员得自己补全地址、算时间、找车。后来我干脆写了个机器人起名叫 CloddsBot全称比较装叫 Cloud Logistics Ordered Data Distribution System核心就干一件事把聊天消息里的订单信息自动抽出来转成结构化数据分派给合适的人再把状态同步给客户。这篇文章从架构、消息接入、字段解析、状态机、调度队列、数据表设计到上线后的坑完整过一遍想自己做“聊天转工单、转订单”系统的朋友可以直接参考。1. CloddsBot 解决的问题是业务上的“信息断点”而不是技术难题1.1 需求全在聊天框里系统里什么都没有我们团队做同城配送客户主要是一些商超门店和批发商他们下货的方式非常原始给自己的对接人发微信、发企业微信群。一天下来调度群里的消息是这样的上午9点从A仓库送50箱农夫山泉到解放路店联系人张姐 138xxxx 下午两点B区三店缺12箱方便面急 明天能不能安排车去物流园拉100件饮料收货人王老板这些消息信息密度高、语义散、格式乱但人一眼能看懂。问题在于调度员看懂了之后还得手动往后台系统里录入一遍然后打电话或者发消息联系司机最后再回客户一句“已安排”。这个链路里录入靠人、分配靠人、回执靠人任何一环慢了或者漏了就是一次客诉。CloddsBot 的核心价值就是把这个链路里的“人工转写”和“人工分派”四个字拿掉。客户在群里发完消息机器人自动解析、自动建档、自动分给负载最低的司机然后自动在群里回一句“单据已生成司机王师傅预计10分钟后联系您”。1.2 定位它不是 CRM也不是 TMS而是消息网关市面上有各种运输管理系统TMS、客户管理系统CRM它们都假设数据已经结构化地进了系统——有标准字段、有订单号、有商品编码。但现实是第一个入口就是一段混乱的聊天文本连字段都没有系统再强大也无从下手。CloddsBot 的最大定位是“把非结构化聊天消息翻译成结构化业务数据”。它在业务系统前面加一层负责听、译、派、回这四个动作听接住企业微信群聊天回调消息译把口语化内容解析成结构化的订单实体派按规则把订单分配给执行人回把处理结果自动发回聊天群这样做的好处是底层业务系统不用改CloddsBot 跟现有 ERP、TMS 之间只需要一条 API 对接把解析完的订单推送过去。即便没有下游系统只把它当一个自动记录加提醒的工具也能省掉调度员一半的重复劳动。1.3 最小可用版本到底做了多少功能第一版我只要求它做三件事识别订单类型、抽出关键字段、写入数据库并通知司机。没有做多轮对话、没有做复杂权限、没有做人员和车辆的实时位置调度。把这三件事跑通就已经能覆盖 60% 以上的高频场景了后面再慢慢把补全话术、优先级插队这些东西补进来。2. 消息接入层企业微信回调 API 的接入与坑2.1 为什么选了企业微信而非钉钉、飞书群里的客户大多数用企业微信原因很简单外部联系人可以通过微信直接跟企业内部员工的企业微信账号聊天客户那边不需要装任何额外 App。这是企业微信相比钉钉和飞书最大的一个优势。CloddsBot 作为应用接入后可以直接被拉进企业内部群也能接收客户与员工单聊的消息。另外企业微信的服务端 API 提供了一套主动推送消息的回调机制当有人发消息时企业微信服务器会向我们的回调地址发一个 HTTP POST 请求。这个机制让我们不需要维护任何长连接也不需要自己写 WebSocket 客户端只要有一个公网可访问的 HTTPS 接口就行。接入方式有三种我实际对比过接入方式维护成本实时性适用场景HTTP 回调企业微信标准最低需要公网 HTTPS高秒级最常用CloddsBot 采用自建 WebSocket 长连接较高需保活、断线重连最高毫秒级需要极低延迟的大规模场景轮询拉取消息中有延迟低分钟级别不推荐回调不可用时兜底用2.2 回调 URL 验证与加解密流程企业微信回调配置里需要填一个 URL、一个 Token、一个 EncodingAESKey。配置保存时企业微信服务器会 GET 这个 URL带上是msg_signature、timestamp、nonce、echostr四个参数我们需要对echostr做解密并原样返回验证通过才算接入成功。解密时有个容易忽略的点echostr不是直接用 AESKey 解而是要结合msg_signature校验签名再把密文按 AES-256-CBC 解密密钥是 EncodingAESKey 经过 Base64 解码后得到的 32 字节。我第一版偷懒直接用解出来的字符串返回结果发现接口超时——后来才意识到企业微信要求解密后的 JSON 里有个Encrypt字段回调消息和验证的包结构不太一样。下面是 FastAPI 里验证 URL 和接收消息的核心代码框架from fastapi import FastAPI, Request, Response from wechatpy.enterprise.crypto import WeChatCrypto from wechatpy.exceptions import InvalidSignatureException import xmltodict app FastAPI() TOKEN your_token ENCODING_AES_KEY your_encoding_aes_key CORP_ID your_corp_id crypto WeChatCrypto(TOKEN, ENCODING_AES_KEY, CORP_ID) app.api_route(/wechat/callback, methods[GET, POST]) async def wechat_callback(request: Request): query dict(request.query_params) if request.method GET: # URL 验证解出 echostr 并返回 try: echostr crypto.check_signature( query.get(msg_signature, ), query.get(timestamp, ), query.get(nonce, ), query.get(echostr, ) ) return Response(contentechostr) except InvalidSignatureException: return Response(contentinvalid signature, status_code403) # POST 消息回调先解密再解析 raw_body await request.body() try: msg crypto.decrypt_message( raw_body.decode(utf-8), query.get(msg_signature, ), query.get(timestamp, ), query.get(nonce, ) ) data xmltodict.parse(msg)[xml] # 这里拿到消息内容丢给后续处理 process_message(data) return Response(contentsuccess) except Exception as e: logger.exception(callback error: %s, e) return Response(contenterror, status_code500)验证时需要注意一个细节回调要在 5 秒内返回否则企业微信会认为超时并重试推送重试会导致同一个消息收到多遍处理时必须按MsgId做去重。2.3 消息推送的幂等与限流设计企业微信回调的机制是“推送-确认”模式如果我们的服务返回非 200 或者超时它会隔一段时间重推最长可能重试三天。这意味着我们绝不能每收到一次回调就落一条数据必须用MsgId做幂等控制。我的做法是Redis 里放一个bot:idempotent:{msgid}的字符串键值存当前状态过期时间设为 24 小时。每次收到消息先尝试用SETNX写入如果已经存在直接丢弃不往下走。数据库里也给消息源的msg_id建了唯一索引双保险。限流那块主要针对回复消息。企业微信对主动发消息有频控短时间内发太多会返回45009错误码对应“接口调用超过频率限制”。CloddsBot 处理方案是拿 Redis 做令牌桶每个会话每秒只允许发 1 条消息如果某次推送因为频控失败把消息塞回重试队列延迟 10 秒再发。3. 从“一句口语”到“结构化订单”规则解析为主、模型兜底为辅3.1 为什么第一版不直接上大模型CloddsBot 立项时纠结过一个问题要不要把消息直接丢给大模型去解析最后我的结论是不需要而且第一版不该这么干。原因有三一是行业异步客户发的内容相对固定翻来覆去就是“时间地点物品数量联系方式”这几件事规则引擎能覆盖大多数二是成本每天几千条消息全走模型推理是一笔持续支出三是可控性规则出错你知道怎么改模型出错你只能干瞪眼。所以第一版用了“关键词词典 正则模板”的组合。针对我们业务场景建立了一个词库物品名称、数量单位、城市区域、仓库和门店别名都收进去。比如“解放路店”在词典里对应store_001这样解析出的结果可以直接落到业务表的外键上。3.2 字段抽取的正则模板设计一条典型消息是明天上午9点从A仓送50箱农夫山泉到解放路店联系人张姐 138xxxx我需要从中抽出六个字段送达时间、出发仓库、物品、数量、目的地、联系人电话。对应的正则拆成多个小模式分别匹配这比一个大正则要好维护得多import re from datetime import datetime, timedelta TIME_PATTERNS [ re.compile(r(?Pday明天|后天|今天)?(?Phour\d{1,2})[点时](?Pminute\d{0,2})分?), re.compile(r(?Pday明天|后天|今天)?(?Pperiod上午|下午|中午)(?Phour\d{0,2})[点时]), re.compile(r(?Pday明天|后天|今天)?(?Pperiod上午|下午|中午)), ] QUANTITY_PATTERNS [ re.compile(r(?Pqty\d)\s*(?Punit箱|件|桶|瓶|袋|车)), ] ITEM_ALIASES { 农夫山泉: nongfu_spring, 矿泉水: nongfu_spring, 方便面: instant_noodle, 饮料: beverage, 可乐: coke, } WAREHOUSE_ALIASES { A仓: wh_a, a仓: wh_a, B仓: wh_b, 物流园: wh_c, } def parse_time(text: str, now: datetime) - datetime | None: for pat in TIME_PATTERNS: m pat.search(text) if not m: continue hour int(m.group(hour) or 9) minute int(m.group(minute) or 0) if m.group(minute) else 0 day_offset {明天: 1, 后天: 2, 今天: 0, : 0}.get(m.group(day) or , 0) dt now timedelta(daysday_offset) return dt.replace(hourhour, minuteminute, second0, microsecond0) return None解析结果是一堆散字段然后再通过“必填字段完整性”算法计算置信度。如果时间、地点、物品、数量四个必填字段都齐了置信度为 1.0缺一个字段降到 0.75。低于 0.7 的消息我会让机器人不是直接建单而是回一句“信息还缺送货时间麻烦补一下”进入待补全状态。3.3 多轮补全机器人不是一次性耗材客户发消息通常不会一条说全。比如有人说“帮我拉三十箱可乐到解放路店”但没说时间。这在真人对话里不是问题调度员会追问一句“什么时候要”但机器人如果直接拒绝或者建一个缺失字段的单都很蠢。CloddsBot 引入了简单的多轮会话状态管理。每条会话在 Redis 里有一个上下文键bot:session:{conversation_id}值是一个 JSON存着已解析的字段和缺失字段列表。当解析出的必填字段有缺失时机器人从缺失列表里挑一个生成追问话术缺时间 → “预计什么时候送到”缺目的地 → “送到哪个仓/哪个门店”缺联系方式 → “方便留个收货人电话吗”客户补一句解析器会在当前上下文基础上合并新字段直到所有必填字段都齐了才生成订单。会话上下文 30 分钟过期避免旧消息残留导致记忆错乱。这里有一个小技巧每个字段的解析结果要带上“来源消息ID”这样如果客户中途改口比如先说“明天上午”又说“改成下午三点”我们能用时间戳更新的字段覆盖旧值而不是简单拼接。3.4 低置信度订单的人工兜底通道规则解析不可能覆盖 100% 的场景。遇到地址没有收录、数量单位是“一堆”“若干”这种模糊表达CloddsBot 会把订单打上uncertain标记进入人工确认列表。调度员在后台确认页看到的是解析前的原始文本和散列出的字段改完字段点确认订单进入正常分发流程。上线两周后统计人工兜底占比从最初的 18% 降到了 6%大部分是因为词典里的门店别名没有收录补充收录后自动解析成功率明显提升。4. 调度核心状态机与基于 Redis Stream 的派单队列4.1 用状态机把整个订单生命周期管起来CloddsBot 里的订单状态我用了九宫格式的有限状态机每个状态和迁移都是显式定义的不允许任何“非法跳跃”INIT初始 → PARSING解析中 PARSING → PENDING_CONFIRM待确认 PENDING_CONFIRM → DISPATCHED已派单 DISPATCHED → EXECUTING执行中司机已接单 EXECUTING → COMPLETED已完成 PENDING_CONFIRM → CANCELLED已取消客户取消 DISPATCHED → CANCELLED已取消无人接单/超时 EXECUTING → EXCEPTION异常货物破损/迟到等状态机的好处是每个操作都要校验“当前状态是否允许迁移”这个约束写死了就不会出现“订单都完成了系统还给它派司机”这种逻辑漏洞。4.2 为什么选 Redis Stream 而不是 RabbitMQ/Kafka分派订单本质上就是一个消息队列订单确认后丢进队列消费者把订单分配给司机。一开始团队有人建议用 RabbitMQ理由是功能成熟、可靠。但被我否了CloddsBot 整套服务跑在一台 4C8G 的云服务器上Redis 本来就在用再加一个 RabbitMQ 就白白多一个维护项而且这个场景根本用不上 RabbitMQ 的高级特性。Redis Stream 是 Redis 5.0 引入的原生消息队列支持消费组、支持 ACK 确认对 CloddsBot 这种量级日均几千条完全够用部署上还不用多养一个进程。往队列里推一条待派送订单XADD bot:dispatch_queue * order_id 20250115001 priority 1消费者通过XREADGROUP拉取任务处理完成后XACK确认如果处理中途崩溃消息不会被确认等超时后重新进入 PELPending Entries List别的消费者可以继续处理。4.3 消费端派单逻辑与并发控制真正“给哪个司机”的逻辑是消费端最核心的部分。CloddsBot 早期用最简单的轮询后来发现司机的负载差别很大——有人一天八单跑不过来有人闲得发慌于是改成“最少未完成订单优先”策略def select_dispatcher(order, dispatchers): # 传入该区域所有可用司机选出未完成订单最少的 best None best_load float(inf) for d in dispatchers: load get_driver_pending_count(d[id]) if load best_load: best d best_load load if load 0: break return best如果再细一点还可以叠加“区域匹配”维度把司机负责的区域和订单目的地做哈希匹配优先选同区域的人这个按实际业务决定。并发控制上消费者数量不能开太多否则下游的司机端 App 会被同时弹单弹爆。我按min(10, 可用司机数)设置消费者并发度并且每个消费者处理完一条任务后稍等一下避免瞬时请求尖峰。实际压测时10 个消费者同时跑每秒能处理 50 笔派单任务远超业务峰值。5. 数据表设计与“状态更新丢行”问题5.1 三张核心表结构CloddsBot 的数据层并不复杂三张表就够用orders表存订单主体信息CREATE TABLE orders ( id BIGSERIAL PRIMARY KEY, order_no VARCHAR(32) UNIQUE NOT NULL, source_chat_id VARCHAR(64) NOT NULL, source_msg_id VARCHAR(64) UNIQUE NOT NULL, item_name VARCHAR(64) NOT NULL, item_code VARCHAR(32), quantity INT NOT NULL, unit VARCHAR(16) NOT NULL, origin_location VARCHAR(128), dest_location VARCHAR(128) NOT NULL, contact_name VARCHAR(32), contact_phone VARCHAR(20), expect_time TIMESTAMPTZ NOT NULL, priority SMALLINT DEFAULT 1, state VARCHAR(20) NOT NULL DEFAULT INIT, dispatcher_id BIGINT, created_at TIMESTAMPTZ DEFAULT now(), updated_at TIMESTAMPTZ DEFAULT now() );order_events表记录每一次状态变更属于一条“审计轨迹”CREATE TABLE order_events ( id BIGSERIAL PRIMARY KEY, order_id BIGINT NOT NULL, from_state VARCHAR(20), to_state VARCHAR(20) NOT NULL, operator_type VARCHAR(20), -- robot/user/system operator_id BIGINT, remark TEXT, created_at TIMESTAMPTZ DEFAULT now() );drivers表就是执行人基础信息包含姓名、电话、所属区域、当前状态空闲/忙碌/离线。顺便说一句source_msg_id一定要建唯一索引这比 Redis 幂等更可靠因为 Redis 数据可能因为重启/淘汰策略丢数据库的唯一约束才是最终底线。5.2 乐观锁更新用 WHERE 条件而不是 SELECT FOR UPDATE状态迁移时最容易踩的一个坑是并发更新导致状态被覆盖。比如同一个订单司机点了“接单”同时客户取消了订单两个请求同时到达如果代码先查状态再 update很可能后到的那个请求把已取消的订单改成“执行中”。正确做法是把状态作为更新条件UPDATE orders SET state EXECUTING, updated_at now() WHERE id :order_id AND state DISPATCHED;受影响行数是 1说明更新成功是 0说明状态已经不是DISPATCHED程序要重新拉取订单状态再决定下一步。这样就不需要显式加行锁了。5.3 Redis 里到底存了哪些数据Redis 在 CloddsBot 里承载了几类职责幂等、会话、限流、队列、缓存。我整理了一张清单Redis Key 模式类型用途bot:idempotent:{msg_id}STRING消息去重24h 过期bot:session:{chat_id}HASH多轮解析的上下文bot:ratelimit:{chat_id}STRING主动回复消息的令牌桶bot:dispatch_queueSTREAM待派单队列driver:load:{driver_id}STRING司机当前未完成单数dict:item:aliasHASH物品别名词典上线半年后我又加了个简单的缓存层把门店/仓库的坐标信息放进 Redis这样调度时不需要每次都查数据库。6. 上线后实测性能数字和几个让人头疼的细节6.1 压测数据与真实表现CloddsBot 跑在腾讯云一台 4C8G 的轻量服务器上部署结构是 Nginx Gunicorn(4 workers) FastAPI PostgreSQL 13 Redis 6。用 locust 压过一轮接口层在并发 200 的情况下回调接口平均响应时间 120msP99 是 380ms单机每秒能处理大概 600 次回调请求这个量级对业务来说绰绰有余——企业微信那边的推送频率根本打不到这个数。实际运行两周后我统计了一下机器人日均处理消息 2600 条左右自动识别建单成功率 94%从客户发消息到司机收到派单通知的平均耗时 18 秒其中主要的延迟在司机端 App 的推送渠道不在 CloddsBot 这边。6.2 上线初期踩的几个典型问题第一个坑是企业微信的“全程加密”模式。刚开始我只配置了明文模式后来企业微信强制要求使用加密模式导致一批历史回调地址直接失效。推消息和收消息的加解密方式不太一样收消息是解密Encrypt字段发消息是加密content字段两边都要改最稳妥的做法是封装统一的加解密模块而不是各写各的。第二个坑是回调超时重试导致的重复处理。某次数据库抖动回调处理超过 5 秒企业微信连推了三次同一条消息虽然我有 Redis 幂等但第一次请求还没跑完第二次就进来了两个并发进程同时查 Redis 都发现键不存在同时往库里插单最后靠数据库唯一索引兜住了避免了重复订单。这个教训让我把“Redis 判断 数据库唯一约束”两道幂等防线当成标配。第三个坑是时区。服务器是 UTC 时区客户说的是“上午 9 点”如果直接按服务器时间解析会出现“上午 9 点的订单存成下午 5 点”这种离谱问题。所有时间解析统一用东八区处理存数据库用timestamptz取出来展示前再转回东八区。现在代码里强制约定所有外部输入的时间文本一律在解析时加上Asia/Shanghai时区。第四个坑和编码有关。企业微信回调传过来的 XML 里中文是正常的 UTF-8但有些人从微信端转发过来的文本里夹杂了特殊字符全角空格、不换行空格导致正则匹配不到或者文字出现“乱码偏移”。处理方式是解析前先做一次字符清洗def clean_text(text: str) - str: # 去掉零宽字符统一全角空格 text text.replace(\u200b, ).replace(\u200d, ) text text.replace(\u3000, ) return text.strip()6.3 运营几个月后的持续优化CloddsBot 上线的头几个月词典基本靠人工补每次遇到新门店、新产品名就要去后台加一条别名映射。后来我加了个专门的“教机器人”入口调度员在确认页录入新词直接写进 Redis 词典并同步到 PostgreSQL不用再改代码。这个改动让自动解析成功率又往上提了几个点。分派策略也迭代过两版。最初是“最少未完成单优先”但发现有些司机长期不动弹未完成单是很少但接单也不积极。后面加了“司机活跃度”权重只统计过去 15 分钟内有位置心跳的司机不活跃的司机直接过滤掉派出去的单子接单率明显改善。还有一点比较重要CloddsBot 虽然叫 Bot但它是业务系统的一部分不是玩具。我建议所有准备做类似机器人的人一开始就规划好状态机、幂等、审计日志这三件套。聊天解析只是入口的能力如果背后的数据模型和状态流转设计不扎实后面接谁都是一堆破事儿。现在这套系统还在跑我有时候会盯着那群里的消息流看机器人先收到一句乱七八糟的“明天下午送二十桶水到XX路”然后几秒钟后回一句“好的已生成订单配送员张师傅会在下午两点前联系您”。想想以前调度员每天对着手机戳半天这个变化还是挺大的。下一步我打算把大模型作为兜底解析器接进来规则引擎没把握的消息再走模型的 few-shot 抽取成本和覆盖率的平衡估计还得再调一阵。