1. 为什么会把机器人做成“中台”而不是随便接个API了事先说说背景。我所在的团队负责公司内部大量的自动化业务流程销售要查订单状态、研发要拉取代码仓库统计、HR要处理入转调离的审批摘要、运营每天要把多维表格里的数据整理成日报……最开始的做法很简单每个部门提需求我们就给飞书群里扔一个机器人转发一下对应的接口。一开始确实爽一个群一个机器人各管各的事。但老实讲这种“野路子”撑不过三个月。四个部门、十几个场景之后问题全部涌出来了机器人账号越来越多、事件回调地址越配越乱、权限全靠“拉人进群”来控制、日志散落在不同服务器上根本没法审计。最致命的是当我们需要把AI能力加进来时——比如让模型自动总结日报、让Codex根据工单生成代码片段——没有一个统一的入口把“人的请求”和“AI的处理”衔接起来。所以后来我下决心做了一套飞书智能交互中台。它的本质不是“一个机器人”而是一个统一的交互层所有飞书事件请求先进中台由中台判断该调用哪个Skill技能单元做什么样的权限校验是否需要多个Skill串联执行然后统一返回结果。这篇文章就是把我实际落地的过程、取舍和踩坑整理出来重点讲三件事多Skill编排、权限管控、可插拔管理。如果你正在做飞书机器人相关的项目或者想把LLM能力接入飞书但不知道怎么组织代码结构这篇文章应该能帮你少走不少弯路。2. 整体架构设计接入层、编排层、技能层要分开2.1 中台的分层思想我在设计时没有把代码写成一坨“接收消息 → 调API → 回复”而是强行分了四层。这个分层是后面一切灵活性尤其是可插拔的基础。接入层只负责与飞书开放平台打交道。接收事件回调、处理URL验证、解密Encrypt Key、解析消息卡片回调。这一层不知道“业务是什么”只把飞书的Event结构体转成平台无关的Request对象。编排层根据Request的意图决定执行哪个Skill、多个Skill的执行顺序、是否需要人工审批等。这一层是“中台”的核心也是“多Skill编排”的主战场。技能层每一个Skill是一个独立的功能单元。比如“订单查询Skill”“日报生成Skill”“表格发送Skill”。它们不关心飞书的消息格式只输入结构化参数、输出结构化结果。基础服务层提供公共能力比如Redis缓存、权限校验客户端、审批流引擎、AI模型网关。Skill如果需要调LLM不直接对接而是走基础服务层的网关这样换模型供应商时不需要改Skill代码。这个分层带来的直接收益是后续加一个新的Skill完全不碰接入层和编排层的代码。只需要在技能层新增一个模块再去配置一个注册项就完事。这就是“可插拔”的起点。2.2 飞书侧的资源模型落地之前要把飞书这边的几个概念理清楚应用App、机器人Bot、事件订阅Event Subscription、消息卡片Message Card、多维表格Bitable、云文档Docs。一个飞书自建应用可以开启“机器人”能力生成一个Bot User。用户机器人或在群里它飞书会向你的回调URL发送事件事件类型包括im.message.receive_v1等。应用可以申请各种API权限比如im:message、bitable:app、docs:document等权限审批是租户管理员在飞书管理后台做的。如果你要发富文本卡片走的是im/v1/messages接口需要构造卡片JSON。我在项目里用的回调模式是“长连接”模式WebSocket而不是“Webhook模式”。原因是Webhook模式要求公网可访问的回调地址本地调试很麻烦而且飞书会频繁校验回调URL的签名和加密。长连接模式通过长连接SDK在本地维持连接不用暴露公网端口安全性和调试体验都好得多。这个选择后面会详细说因为它直接影响了我如何做权限管控和“可插拔”。3. 多 Skill 编排从“单点回复”到“流程组合”3.1 为什么要编排很多初做飞书机器人的朋友逻辑是这样的收到消息 → if 消息包含“订单” → 调订单接口 → 回复。这是典型的“路由”不是“编排”。但真实企业场景里一个请求往往需要多个能力配合。我遇到的一个典型案例运营人员在群里发了一句“帮我把今天多维表格里的销售数据按区域汇总生成表格发给华东群”。这句话要拆成几步解析“多维表格里” → 调用多维表格读取Skill解析“销售数据汇总” → 调用数据聚合Skill生成表格 → 需要调用表格生成Skill生成xlsx或飞书电子表格发给华东群 → 调用消息发送Skill这个过程中每个Skill只做自己的一件事但编排层要把它们串成一个有序的Pipeline。3.2 三种编排模式我在实际项目里整理出三种编排模式分别应对不同场景路由式编排最基础的一层。根据用户请求的意图Intent选择唯一的Skill。我用的方式不是传统的关键词正则太脆而是让LLM作为“意图分类器”把用户消息和Skill清单一起送给LLM让它返回该调用哪个Skill以及抽取出的参数。用户消息查一下订单OD-20250401的物流状态 → 中台调用意图识别模型 → 返回 Skillorder_query, 参数{order_id:OD-20250401} → 编排层路由到订单查询Skill这里的核心是让Skill自己声明“我能处理什么”然后LLM做匹配。只要Skill的声明写得清楚新增Skill后不需要改路由代码。管道式编排当一个请求需要多个Skill按顺序协作时我定义了一个Pipeline对象。每个Pipeline声明了Skill节点列表、每个节点从上游拿到什么参数、是否需要人工确认断点等等。{ pipeline_id: report_send_pipeline, name: 生成日报并发群, nodes: [ {skill: bitable_fetch_skill, input_mapping: {table: user_msg.table}}, {skill: data_aggregate_skill, input_mapping: {source: prev.result}}, {skill: sheet_build_skill, input_mapping: {data: prev.result}}, {skill: message_send_skill, input_mapping: {file: prev.result, chat_id: target_chat}} ] }每个节点执行完后把输出塞进一个上下文对象里下一个节点通过prev.result引用。执行到“给外部群发送文件”这种高危节点前Pipeline会暂停等人点击审批卡片后再继续。这个“暂停恢复”机制是权限管控的重要一环后面细讲。状态机式编排处理多轮对话时用的。比如用户先问“帮我查一下华东区销售数据”机器人需要追问“你指的是本周还是本月”用户回答后继续推进。这种场景需要在Redis里保存会话状态编排层根据状态转移判断下一轮该干嘛。我把它做成了一个通用的SessionManager为每个用户维护一个状态机状态机里每个状态绑定对应的Skill。3.3 我最终选择的编排引擎三种模式听起来很多但实际在我的中台里核心引擎只做一件事根据Skill注册表LLM意图识别结果挑选合适的执行策略。对于“无需多步但需要参数填充”的请求用路由式对于“明确需要多个Skill协作”的场景先判断是否存在对应的Pipeline没有则动态生成编排计划让LLM根据Skill声明自动编排。动态生成这块我做得比较保守只允许LLM组合不超过4个Skill避免模型“放飞自我”。这套设计的核心是Skill本身不感知编排。编排完全是中台调度层的行为每个Skill只关心“给我一个合法参数我返回一个结构化结果”。这样编排策略随时可以调整Skill不需要有任何改动。这就是可插拔和可编排能同时成立的原因。4. 权限管控身份、技能、操作三层模型4.1 不能只靠“在群里”来判断飞书机器人最常见的一个权限漏洞是任何把机器人拉进群、或能在群里机器人的人都能触发机器人的一切功能。这在内部小范围用还凑合一旦涉及“发送文件到外部群”“读取多维表格数据”“调用Codex消耗token生成代码”这类操作就必须有完善的权限管控。我的权限体系分了三层身份层先搞清楚“是谁在发消息”。飞书事件回调里带有open_id和user_id我用open_id作为唯一用户标识。在系统里有一个用户表把open_id映射到内部员工号、部门、角色标签。这个映射表通过飞书通讯录API定期同步contact/user/batch_get保证员工入离职状态及时更新。技能层每一个Skill在注册时要声明允许的“身份范围”。范围可以是某个部门需要对接通讯录API判断用户是否属于该部门、某个群聊判断当前消息来自哪个chat_id、某个角色标签从内部权限系统同步。比如订单查询Skill只允许销售部门代码生成Skill只允许研发Leader角色多维表格读取Skill只允许该表格协作者。我在注册表里加了一个allowed_scopes字段{ skill_id: order_query_skill, name: 订单查询, allowed_scopes: { departments: [sales_dept], chat_ids: [oc_xxx], roles: [sales_manager] } }校验是“或”逻辑满足任一条件即可。这层校验在编排层执行任何一个Skill命中之前都会检查避免绕过编排层直接调用Skill内部的入口。操作层即使身份没问题、技能也开放了某些具体操作仍然需要额外确认。这主要是针对“对外发送”“删除数据”“生成代码并写入仓库”这类不可逆或高风险动作。我的做法是引入“操作审批点”。在Pipeline执行到某个高风险节点时自动在管理群发送一张审批卡片卡片上有“允许执行”和“拒绝拒绝”按钮只有具备审批权限的管理员可以点击。审批通过后编排层才会继续推进Pipeline否则整个流程终止并通知发起人。4.2 权限校验的工程实现权限校验最关键的要求是快不能每次消息进来都去远程调API查一遍部门和角色。我的方案是启动时全量加载用户权限快照到Redis设置10分钟过期。用户在消息进来时先查Redis命中直接用没命中则回源飞书通讯录接口拉取并回填缓存。用户主动触发“刷新我的权限”命令时可以提前刷新缓存。这样做的好处是普通人在群里机器人时响应延迟基本不受权限校验影响而权限变更最多10分钟后生效企业内部不会觉得滞后期无法接受。4.3 一个真实权限事故的教训有一次公司销售总监投诉说有人用机器人调了销售明细数据。排查后发现最开始订单查询Skill只给销售部门开放但后面接数据聚合Skill时Pipeline配置文件里的allowed_scopes被漏掉了导致Pipeline直接调了另一个查询Skill等于“绕过了技能层校验”。这次事故之后我把权限校验逻辑改到了编排层调用链的必经环节上——也就是“执行Pipeline的引擎入口处”再统一校验一次并让测试环境默认拒绝所有Skill执行只有授权后的群聊才能真正跑通。这提醒所有做中台的朋友权限管控点要放在不可绕过的公共链路上而不是每个Skill内部自觉校验。5. 可插拔管理Skill 是“插上去就能用”的插件5.1 设计一个Skill的最小接口“可插拔”是我整个项目里最花心思的部分。要让团队里每个新人都能快速接入一个新功能同时保证老功能不受影响就必须定义一套稳定的Skill接口。我在Python端定义了一个抽象基类# skill_base.py from abc import ABC, abstractmethod from typing import Dict, Any class BaseSkill(ABC): # 技能元信息注册时读取 skill_id: str name: str description: str input_schema: Dict[str, Any] {} output_schema: Dict[str, Any] {} abstractmethod def execute(self, params: Dict[str, Any], context: Dict[str, Any]) - Dict[str, Any]: 执行技能核心逻辑返回结构化结果 pass def pre_check(self, user, chat_id, params) - Dict[str, Any]: 技能级别的自定义权限/参数预检返回(是否通过, 错误信息) return {ok: True, message: }每个Skill只需要继承BaseSkill实现execute方法再写一个skill_manifest.json做注册声明{ skill_id: bitable_fetch_skill, name: 多维表格数据读取, description: 用户提供多维表格token和视图名读取指定视图的记录数据, version: 1.2.0, owner: data_team, allowed_scopes: { departments: [data_team, ops_team] }, entry: skills.bitable_fetch_skill:BitTableFetchSkill }5.2 注册中心与热加载系统中维护了一个“技能注册表”本质上是一个SQLite表Redis索引。启动时扫描所有skills/目录下的manifest文件插入注册表并记录下来每个Skill对应的模块导入路径。为了让“插拔”做到不用重启服务我用importlib来实现模块动态加载# registry.py import importlib class SkillRegistry: def __init__(self): self._skills {} def register_from_manifest(self, manifest_path: str): manifest json.load(open(manifest_path, encodingutf-8)) module_path, class_name manifest[entry].split(:, 1) module importlib.import_module(module_path) skill_cls getattr(module, class_name) skill_instance skill_cls() skill_instance.manifest manifest self._skills[manifest[skill_id]] skill_instance return skill_instance def reload_skill(self, skill_id: str): # 按skill_id找到旧的模块信息移除缓存后重新加载 pass新开发一个Skill时开发者只需要在skills/下新建目录放代码和manifest然后运行一个python manage.py reload skill_id命令。热加载成功后编排层马上能看到新Skill的声明。无需重启中台进程这在联调阶段效率极高。5.3 灰度与下线可插拔不只是“能加”还要能“灵活动态地减”。我在注册表里加了三个状态active、gray、disabled。active所有流量正常进入。gray只对白名单内的用户/群聊生效用于新版本Skill的小范围验证。disabled彻底不接收流量已有会话直接拒绝执行。灰度发布时我在编排层加了一个流量判断如果Skill标记为gray则检查当前请求的open_id是否在灰名单中。否则走旧版本逻辑。这个灰名单在管理后台配置不需要改代码。下线一个Skill时我会先把它置为disabled观察一周日志确认无新请求再删除代码目录。因为注册表里还保留着entry字符串编排请求如果命中了disabled的Skill会返回“该功能已下线”的提示而不是直接报错用户侧体验更平滑。5.4 可插拔带来的一个额外收益可插拔设计让我能直接给“普通同事”放权。以前每个新需求都要我来改代码、部署、测试现在我把“接入新Skill”的流程做成了文档化模板数据分析团队的同学按照模板写一个Python文件manifest跑一遍自测命令然后在管理后台提交注册申请等代码Review通过后热加载即可上线。后来我们团队把“申请新Skill”的流程也接进了飞书审批。开发者在多维表格里登记申请审批机器人自动在管理群发卡片点通过后自动触发部署流水线。一个简单的数据查询Skill从提出需求到上线最快半天搞定。6. 实际操作中踩过的坑与排查链路6.1 Codex接入飞书的连接方式标题里提到的Codex其实是把OpenAI的Codex能力封装成一个“代码生成Skill”。踩坑的点在于飞书的事件回调机制和Codex的流式输出不匹配。飞书的im.message.receive_v1事件要求你在3秒内响应回调否则飞书会视为超时并重试。而Codex处理一个代码生成请求往往要几十秒甚至更久。如果直接在事件回调里同步调用Codex必炸。我的解决方案是异步任务主动推送回调收到消息后立刻返回HTTP 200空响应回执给飞书把真正任务丢进Celery队列。任务执行期间先在群里发一张“正在处理”的普通文本消息或卡片。Codex生成完成后用消息更新接口im/v1/messages/{message_id}把卡片内容更新为最终结果如果是代码内容用im/v1/messages?receive_id_typechat_id创建富文本消息推送代码块。这样做的另外一个好处是用户看到“正在处理”卡片后知道机器人没有卡死。新版飞书消息卡片支持msg_typeinteractive可以在卡片里放“重新生成”按钮触发另一个Skill重新跑任务。这个交互闭环用户反馈非常好。6.2 机器人发送表格的多维编码问题“飞书机器人发送表格”是群里最常见的需求。这里最容易踩坑的是你以为发送的是Excel文件实际上飞书有两种“表格”完全不是一回事。一种是上传一个真正的xlsx文件走im/v1/files接口以file_typexlsx发送。这种适合发送给用户下载编辑。另一种是消息卡片内嵌表格走interactive卡片JSON字段是table。卡片内表格只能展示不能下载数据量小时展示效果好。我一开始没分清楚写了个通用发送函数结果用户说“收到的表格打开是乱的”。排查后发现xlsx文件上传用的是二进制multipart表单飞书那边要求文件名不能包含非UTF-8字符且文件大小不能超过30MB而卡片内嵌表格则要求每个单元格内容不超过2000字符且不能出现\n换行符需要替换成br。最终的发送函数做了两层判断数据量大或需要导出编辑时走xlsx上传数据量小比如50行以内则生成卡片表格。一句话总结先问用户要“能下载的文件”还是“能直接看的卡片”别自作主张。6.3 多维表格API的速率限制与字段类型坑用Bitable多维表格做数据读取时最烦的不是授权而是API的速率限制和字段类型多样性。飞书多维表格的API按应用维度限流默认大约是每秒10次请求。问题在于当你一次性读取一个超大视图比如1万行记录时必须分页拉取每页最大500条。如果你为了赶时间并发拉取多个页很容易触发限流返回429。我的处理方式是写了一个带限流器的Bitable客户端采用令牌桶算法每秒最多8个请求所有Skill都走这一个客户端。宁可慢一点也不要被限流断了任务。字段类型是另一个大坑。多维表格的字段可以是文本、数字、日期、单选、多选、人员、附件、公式、关联等API返回的字段值格式差异非常巨大。人员字段返回的是数组每个元素是open_id日期字段可能是时间戳或格式化字符串公式字段可能是计算结果或错误信息。我在数据聚合Skill里加了一个字段类型归一化层所有字段读取后统一转为字符串或数字并额外记录一个字段类型标签供下游处理。否则用LLM做数据摘要时模型会把“人员字段的open_id数组”当成普通文本生成完全没意义的总结。6.4 消息事件回调和重试的重复处理飞书的事件订阅有一个机制如果回调地址没有在限定时间内返回成功飞书会重试推送事件。重试间隔一般是3秒、30秒、5分钟等递增。如果你的接入层没有做幂等处理重试就会导致同一个用户请求被多次执行。最典型的场景是中台执行一个耗时任务时飞书认为回调超时了自动重试于是任务被重复触发。我在这块吃过亏有次用户一句话“生成周报”被重复执行了三次发出三份周报到群里。解决方案是给每个事件生成一个唯一的事件ID飞书回调会带header.event_id在Redis里记录processed_events设置TTL 24小时。处理前先判断event_id是否已存在存在则直接忽略。后来扩展到所有回调处理前统一做幂等判断。这个经验值得所有做飞书机器人的朋友注意飞书不会再发一次不代表它不会重试。6.5 如何把云文档内容嵌入到自己的网站这个需求来自团队内部的Wiki整理想把飞书云文档里的内容同步到我们的技术博客站避免两处维护。最靠谱的方式是用飞书开放平台的文档块接口docx/v1/documents/{document_id}/blocks拉取文档结构按块类型标题、正文、列表、表格、引用转成HTML或Markdown然后走站点发布流程。这里有几个注意点文档需要给应用授docx:document:readonly权限且文档对应用可见要么添加协作者要么使用租户内所有文档权限。图片块拉取到的还是media_id需要再调用drive/v1/medias/{file_token}/download拿到真实图片URL。表格块转换时列宽可能丢失最好在站点侧重新做响应式布局。也有同事提到用lark sync去同步到Obsidian这适合个人知识库但做网站内容嵌入还是直接走开放平台API可控性最高因为你能控制文档转HTML的每一步样式。7. 踩过坑之后我对这套架构的反思整个项目上线至今运行已经有小半年期间迭代了二十几个Skill模块化架构给我带来了巨大的维护红利。最早的两个机器人发布机已经可以退役所有请求都收拢到中台统一处理。现在我最深的体会是不要把权限管控寄托在“每个Skill自己想起来校验”上一定要把校验放到编排层的公共路径。可插拔的意义不只是“方便加功能”更重要的是“危险的旧功能可以被优雅降级而不是紧急删除代码”。与AI模型相关的Skill一定要设计成异步卡片更新模式否则飞书事件回调超时重试会把你折磨到疯。如果以后有朋友也想做类似的企业内机器人中台我会建议他先想清楚三层模型接入层只管协议编排层只管调度技能层只做业务。这三层中间不要互相渗透。哪怕前期多写一点胶水代码也比后期把业务逻辑和机器人逻辑绞在一起强得多。最后分享一个小技巧团队里新同事接手的第一个小任务我通常会安排他写一个新的Skill比如“汇率查询”从写manifest到热加载到群里测试完整走一遍。这个流程走完他对整个中台的“插拔”机制基本就无师自通了。