ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

飞书机器人接入实战:事件订阅、大模型对话与多维表格落库全攻略

飞书机器人接入实战:事件订阅、大模型对话与多维表格落库全攻略 做飞书接入之前我先说一个真实场景去年我帮团队搭过一个会自己干活的内部工具不是那种只会群发提醒的机器人而是真正能在群里接需求、查数据、回状态、做记录的助理。用下来最大的感受是飞书这套开放能力比大多数人想象中要成熟得多——从前台的消息收发、后台的权限管理到多维表格当数据库几乎每个环节都有现成的接口。这个项目做下来我踩了不少坑也摸清了一条从零到上线的最短路径所以打算写成一篇带实操代码的完整记录给也想在飞书里搭全天候客服/助理的朋友做个参考。这篇文章适合你有这么几种情况想在飞书群里放一个能自动回复的机器人想把飞书消息接到大模型做智能问答想把客服记录、工单状态直接落到多维表格里做统计甚至只是想搞清楚飞书开放平台的接入流程到底是怎样的。我会从整体架构讲到后台配置从消息收发代码讲到大模型对接再到多维表格落库最后把常见的坑都列出来。每一段都会附上能直接跑通的核心代码目标是你照着敲一遍就能拥有一个自己的飞书助理。1. 搭建前先想清楚这套全天候客服到底分几层很多人一上来就打开飞书开放平台创建应用然后就开始写代码结果写到一半发现权限不够、事件收不到、消息发不出去回头再查文档来回折腾好几天。其实飞书这种平台型接入项目动手前最该做的是把架构想明白想明白之后后面的每一步都只是照做而已。1.1 整体调用链路的四个角色以我搭的这套系统为例整个链路涉及四个角色飞书客户端员工在群里发消息、机器人或者直接私聊机器人。飞书开放平台负责把消息事件推送给你的服务端也负责校验你的API调用凭证。你的服务端接收飞书推送的事件做业务判断决定是直接回复还是调大模型、查多维表格。外部能力层大模型API、多维表格API、企业内部的查询接口等等。这四个角色之间通过两种方式通信一种是飞书主动把事件推给你消息回调、机器人被等另一种是你的服务端主动调用飞书OpenAPI发消息、查用户、读写表格。理解了这两条线后面看文档就不会迷路。1.2 为什么建议采用事件订阅 Webhook而不是轮询飞书开放平台提供了事件订阅机制也就是在群里有人发消息、机器人、加好友、进群退群这些动作发生时飞书会实时把事件数据POST到你配置的回调地址上。这比起你每隔几秒主动调用API去拉新消息有几个明显优势实时性高消息到的瞬间就能响应。服务端压力小不用一直空转轮询。飞书侧做了重试机制你的服务临时不可用它还会尝试重新推送。当然了有些场景确实需要主动拉取比如做历史消息的批量导入这时候用im/v1/messages这类查询接口就行。但客服这个场景事件订阅是绝对的主流我在搭建时选的就是这条路线。1.3 服务端选型长短连接、框架与部署机的最小可行方案服务端这块我用的是Python FastAPI原因很简单Python生态里调用大模型、处理JSON、写脚本都方便FastAPI又是轻量异步框架起个Webhook服务几行代码就搞定。项目结构长这样feishu-assistant/ ├── app.py # FastAPI入口 ├── feishu_client.py # 飞书API封装 ├── llm_client.py # 大模型调用 ├── bitable_client.py # 多维表格操作 ├── config.py # 配置项 └── requirements.txt部署上如果你的服务器有公网IP直接用Webhook方式暴露一个HTTPS地址就行。如果没有公网IP飞书还提供了长连接模式WebSocket服务端主动跟飞书建立连接消息直接推到这个长连接上不需要暴露公网端口。这对我这种经常在个人服务器上搭东西的人来说很实用后面细讲。2. 飞书开放平台的后台配置权限、事件订阅与常见错误码这章是整个接入最容易出问题的地方。大部分第一次接触飞书开放平台的人都会在应用后台到底要配哪些东西上懵掉。我按创建应用到发布上线的完整顺序来讲。2.1 创建企业自建应用的最低配置路径在飞书开放平台open.feishu.cn用管理员账号登录后进入开发者后台点击创建企业自建应用填应用名称和描述。创建好之后重点要配这么几块应用能力添加机器人能力。这一步会在应用下生成一个机器人后面群里提到的就是它。权限管理在权限列表里搜索并开通以下权限点im:message读取用户发给机器人的消息。im:message:send_as_bot以机器人的身份发送消息。im:chat:readonly读取群信息用于判断消息来自哪个群。contact:user.base:readonly读取用户基本信息用于识别发送者是谁。如果要读多维表格还需要bitable:app。事件订阅配置回调地址并订阅你需要的事件。版本发布以上都配好后创建版本并发布。这里特别提醒很多人配完权限发现还是不生效原因就是没发布版本。自建应用需要发布后权限和事件配置才会真正生效。我第一回搭建时还犯过一个低级错误权限开通了但是发布的是测试版同事那边根本不是这个版本结果消息一直推不过来。后来学乖了每次改配置先发一个版本再在群里真实验证一次。2.2 事件订阅里最难的一步URL验证与加密解密在事件订阅后台你需要填一个回调URL。填完保存时飞书会立刻向这个URL发送一个验证请求如果没开启加密飞书会POST一个{challenge: xxxx}的JSON你的服务端需要原样把challenge字段返回。如果开启了加密通常建议开启飞书POST请求的body里会带一个encrypt字段你需要用配置的Encrypt Key做AES解密拿到明文后再处理challenge。很多人在这一步一直报验证失败多半就是加解密没对上。推荐直接用官方SDKSDK里自带解密逻辑不需要手写AES。以Python为例用lark_oapi的EventDispatcherHandler会帮你处理掉解密和challenge响应import lark_oapi as lark from lark_oapi.api.im.v1 import P2ImMessageReceiveV1 def do_message_receive(data: P2ImMessageReceiveV1) - None: # 处理消息事件 pass event_handler ( lark.EventDispatcherHandler.builder(encrypt_key, verification_token) .register_p2_im_message_receive_v1(do_message_receive) .build() )如果你坚持用原生Web框架自己实现也可以但一定要确认加密模式和填充方式跟飞书文档保持一致这个踩坑率极高。我在社群里看到很多人问飞书错误代码2700002我遇到的情况基本都是事件订阅回调校验不过、加解密没对上或者URL不可达按这个方向排查基本都能解决。2.3 订阅哪些事件才能满足客服/助理需求事件订阅里可选的事件非常多但对客服/助理这个场景来说最核心的就是这么几个im.message.receive_v1收到消息时触发。这是最关键的群里机器人、私聊机器人都是这个事件。im.chat.member.added_v1机器人被拉进群时触发可以做欢迎语。contact.user.updated_v2用户资料变更如果助理要维护通讯录相关功能可以订阅。实际开发中我基本上只订阅了im.message.receive_v1一个事件其他都靠主动调API处理。订阅太多反而会增加服务端的处理噪音。3. 服务端接入消息收发与事件处理的完整代码后台配置完成后就到了最核心的代码部分。这章我会讲清楚两件事怎么收到消息怎么发消息。顺便把避免机器人自己回复自己和只响应自己的消息这类逻辑也一起讲掉。3.1 事件回调代码FastAPI版最小实现不管后台的加密怎么配到了服务端核心逻辑都一样接收POST请求解析请求体里的消息事件提取消息文本、发送者、群ID然后走业务逻辑。我用FastAPI写了一个可以直接跑的最小版本from fastapi import FastAPI, Request import json app FastAPI() app.post(/webhook/event) async def receive_event(request: Request): data await request.json() # 处理URL验证 if challenge in data: return {challenge: data[challenge]} event data.get(event, {}) message event.get(message, {}) chat_id message.get(chat_id) message_id message.get(message_id) msg_type message.get(message_type) # 只处理文本消息 if msg_type ! text: return {code: 0} # 消息文本在 content 里是JSON字符串 content json.loads(message.get(content, {})) text content.get(text, ) # 把事件交给业务处理 await handle_message(text, chat_id, message_id) return {code: 0}这里的handle_message就是你的业务入口。如果想用官方SDK替代手写解析也能做到而且官方SDK会自动处理加密、重试、长连接后面我会单独介绍。3.2 发消息tenant_access_token申请与消息发送服务端要主动发消息不是直接用App ID和App Secret调接口而是先要换取一个tenant_access_token。这个token的有效期通常是2小时建议做缓存避免每次都请求换取。用纯requests实现大概是这样的import requests import time import json APP_ID your_app_id APP_SECRET your_app_secret _token_cache {token: , expire_at: 0} def get_tenant_access_token(): if _token_cache[token] and _token_cache[expire_at] time.time(): return _token_cache[token] resp requests.post( https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal, json{app_id: APP_ID, app_secret: APP_SECRET}, timeout5, ) data resp.json() if data.get(code) 0: _token_cache[token] data[tenant_access_token] _token_cache[expire_at] time.time() data[expire] - 60 return _token_cache[token] else: raise Exception(f获取token失败: {data}) def send_text(chat_id: str, text: str): token get_tenant_access_token() resp requests.post( https://open.feishu.cn/open-apis/im/v1/messages?receive_id_typechat_id, headers{ Authorization: fBearer {token}, Content-Type: application/json, }, json{ receive_id: chat_id, msg_type: text, content: json.dumps({text: text}), }, timeout10, ) return resp.json()这里要特别提醒的是receive_id_type参数。如果你传的是chat_id那receive_id字段就填群ID如果你传的是open_id那receive_id就填用户ID。传错类型接口会一直报参数错误。3.3 官方SDK与长连接模式没有公网IP也能接入如果你不想自己处理加解密、不想申请公网回调地址那强烈建议用官方Python SDK它内置了长连接WebSocket模式。我后来就把服务切到了长连接省掉了所有回调配置的麻烦。长连接模式的核心代码import lark_oapi as lark from lark_oapi.api.im.v1 import ( P2ImMessageReceiveV1, ReplyMessageRequest, ReplyMessageRequestBody, ) app_id your_app_id app_secret your_app_secret def handle_message(data: P2ImMessageReceiveV1) - None: message_id data.event.message.message_id content json.loads(data.event.message.content) text content.get(text, ) # 业务逻辑 reply_text process(text) # 回复消息 request ReplyMessageRequest.builder() \ .message_id(message_id) \ .request_body( ReplyMessageRequestBody.builder() .content(json.dumps({text: reply_text})) .msg_type(text) .build() ) \ .build() client.im.v1.message.reply(request) event_handler ( lark.EventDispatcherHandler.builder(, ) .register_p2_im_message_receive_v1(handle_message) .build() ) client lark.Client.builder() \ .app_id(app_id) \ .app_secret(app_secret) \ .log_level(lark.LogLevel.INFO) \ .build() ws_client lark.ws.Client.builder(, ) \ .event_handler(event_handler) \ .build() ws_client.start()这段代码跑起来后你的服务会以WebSocket方式连到飞书之后所有消息事件都会从这个长连接里推送过来。不需要配回调URL不需要有公网IP对个人开发者来说太友好了。3.4 必踩的坑机器人自己回复自己、风暴刷屏接入消息收发后第一个会遇到的诡异问题就是机器人回复的消息又触发了im.message.receive_v1事件于是机器人回复自己的回复陷入死循环。解决思路是判断消息发送者是不是机器人自己。飞书的消息事件里有个sender.sender_type和sender.id字段你只需要在事件处理逻辑开头加一行判断sender_id data.event.sender.sender_id.open_id if sender_id bot_open_id: returnbot_open_id可以通过调用/open-apis/bot/v3/info接口拿到。更稳妥的做法是只响应群里机器人的消息不响应普通群消息。群里普通消息都去回复的话会打扰所有人而且很容易触发频率限制。判断是否了机器人需要解析消息里的mentions数组mentions data.event.message.mentions or [] if not any(m.get(id, {}).get(open_id) bot_open_id for m in mentions): return这样处理之后机器人在群里就只会在被点名时出现平时完全隐身体验上更符合客服/助理的定位。4. 让机器人真正会说话接入大模型并设计对话边界一个只会回你好我是机器人的飞书助手说实话意义不大。要把全天候助理做起来大模型这层是关键。我接的是DeepSeek的API因为它在中文对话、成本、响应速度上都很均衡。而且它兼容OpenAI的接口格式这意味着我以后想换GPT、Kimi或者其他模型的API改动量非常小。4.1 大模型API接入OpenAI兼容格式的通用写法DeepSeek的API地址是https://api.deepseek.com用OpenAI SDK可以直接调用from openai import OpenAI client OpenAI( api_keyyour_deepseek_api_key, base_urlhttps://api.deepseek.com/v1, ) def call_llm(prompt: str, history: list None) - str: messages [{role: system, content: system_prompt()}] if history: messages.extend(history) messages.append({role: user, content: prompt}) resp client.chat.completions.create( modeldeepseek-chat, messagesmessages, temperature0.7, max_tokens1024, ) return resp.choices[0].message.content这里的base_url改成其他厂商的地址就能快速迁移。代码里我故意在max_tokens这里设置了1024而不是不设上限——因为飞书对机器人回复消息的调用时间有要求如果大模型一次生成太长可能会导致回复超时所以这里宁可截断一点也要保证响应速度。4.2 设计System Prompt让助理知道什么该答、什么不该答系统提示词是决定助理质量的关键。我见过很多人直接让大模型你是一个客服这样出来的回答往往又空又格式化。我的做法是在System Prompt里给机器人设好身份边界、回复风格和禁用范围。下面这个是我在项目里实际用的简化版你是公司的内部客服助理名叫小飞。 你的职责是回答员工关于公司制度、IT支持、日常流程的问题。 回复要求 1. 简洁、口语化控制在200字以内。 2. 如果问题涉及公司机密或你无法确认的信息明确说这个我需要查一下。 3. 不要编造具体的人名、数据、日期。 4. 涉及需要人工处理的问题引导用户在群里IT值班同事。这样设置之后机器人的回复明显收敛了很多不会动不动就长篇大论也不会一本正经地编造事实。这里我特别想强调一点大模型本身没有边界意识边界完全靠System Prompt和代码逻辑双重控制。4.3 多轮上下文如何在无状态机器人里记住刚才聊了什么飞书消息回调每次都是独立的HTTP请求服务端如果不做存储机器人是记不住上文聊了什么的。要让助理连续对话最简单的方案是把历史对话放到多维表格或者本地缓存里。我用的策略是按会话ID存缓存会话ID就用私聊里的open_id或者群聊里的chat_idimport time from collections import defaultdict session_cache defaultdict(list) MAX_HISTORY 10 def get_session_history(session_id: str) - list: return session_cache[session_id][-MAX_HISTORY:] def append_session(session_id: str, role: str, content: str): session_cache[session_id].append({ role: role, content: content, time: time.time(), }) if len(session_cache[session_id]) MAX_HISTORY: session_cache[session_id] session_cache[session_id][-MAX_HISTORY:]这里我用的是内存缓存如果服务重启历史就没了。生产环境建议换成Redis但原理是一样的。多维表格也能做上下文存储但每次查表会多几十毫秒延迟对实时对话这个场景来说不是最优解。多维表格我更推荐用来做结构化记录这个下一章详细说。4.4 超时与异常处理大模型挂了机器人不能跟着挂大模型API是有可能超时、报错的。如果大模型请求超时飞书那边还在等回复用户看到的就是机器人已读不回。所以我在调用大模型时做了两件事一是设置请求超时二是异常兜底。def safe_llm_call(prompt: str) - str: try: return call_llm(prompt) except Exception as e: # 打日志 返回兜底文案 print(f[LLM ERROR] {e}) return 抱歉我当前有点卡顿请稍后再试。5. 多维表格当底座工单、记忆与每日小结机器人能聊天之后你会发现它缺一个记忆仓库。客户问过什么问题、哪些问题没有解决、今天总共处理了多少条请求这些数据如果不落库就只是一个聊完即焚的玩具。多维表格在这里就是个很好的落库方案因为飞书自带的多维表格有现成的API而且表格本身就能做筛选、分组、看板业务人员不用写代码也能直接查看数据。5.1 用多维表格建一张客服工单表在建表之前先在飞书里手动创建一个多维表格里面至少要有这几个字段问题文本回答文本状态单选待处理/已解决/已升级来源文本记录是哪个群/哪个用户问的创建时间创建时间字段类型自动生成然后用开放API往表格里写记录。调用方式并不复杂HTTP接口如下APP_TOKEN your_app_token TABLE_ID your_table_id def add_record(fields: dict): token get_tenant_access_token() resp requests.post( fhttps://open.feishu.cn/open-apis/bitable/v1/apps/{APP_TOKEN}/tables/{TABLE_ID}/records, headers{ Authorization: fBearer {token}, Content-Type: application/json, }, json{fields: fields}, timeout10, ) return resp.json()调用时把问题、回答、状态等字段塞进fields字典即可。这里需要注意多维表格的字段类型和API传参类型是对应的文本字段传字符串单选字段也传字符串但实际上单选字段传的是选项名。创建时间字段不用你传表格会自动生成。5.2 从普通回复到自动登记让每条问答都留下痕迹我把这个落库逻辑加到了机器人处理消息的流程里用户问了一个问题机器人回复完之后自动把问题 回答 状态写到多维表格。这样我每天晚上打开表格就能看到当天所有的问答记录。实际效果是团队里很多重复性问题我直接通过表格筛选功能就能统计出频率最高的前几个然后针对性优化系统提示词。这个过程完全是数据驱动的比我拍脑袋改Prompt靠谱太多。5.3 查询历史记录与工单状态更新除了写入多维表格也支持查询。比如我的历史工单这种需求可以通过查询接口检索def search_records(condition_field: str, condition_value: str): token get_tenant_access_token() resp requests.post( fhttps://open.feishu.cn/open-apis/bitable/v1/apps/{APP_TOKEN}/tables/{TABLE_ID}/records/search, headers{ Authorization: fBearer {token}, Content-Type: application/json, }, json{ filter: { conjunction: and, conditions: [ { field_name: condition_field, operator: is, value: [condition_value], } ], } }, timeout10, ) return resp.json().get(items, [])这个接口在做查进度查历史这类对话时非常有用。比如用户问我刚才提交的那个问题处理好了吗机器人就先按用户ID去表格里查记录查到之后再针对状态字段做回复。5.4 用多维表格做定时日报和统计思路多维表格另一个我很常用的玩法是定时统计。通过records/search把当天记录拉出来然后在大模型里做一个简单的汇总Prompt就能在每天下班前自动发一条今日客服小结到群里。内容大概包括今天处理了多少条问题、高频问题有哪些、有多少待处理。这个功能跑起来之后团队对助理的价值感知会一下子提升很多。6. 上线一周后遇到的坑与收尾建议代码写完了接入也验证通过了不等于事情结束。真正上线跑起来遇到的各种问题才会浮出水面。这一章我把这一周里遇到的真实问题和排查思路列出来希望能让你少走弯路。6.1 token缓存失效与并发请求导致凭证错误我在第3章写了token缓存但上线后第一天就发现一个问题多个事件同时进来时如果token刚好过期多个请求同时去获取新token飞书侧会短暂出现频率限制导致部分请求失败。解决方案是给token获取加一个互斥锁用Python的话就是threading.Lock()import threading _token_lock threading.Lock() def get_tenant_access_token(): with _token_lock: # 检查缓存并刷新 ...这种锁 缓存的组合在并发场景下是标准的处理方式。另一个值得一提的点是token要提前60秒刷新代码里的-60就是为了这个千万不要刚好卡在过期时刻去换。6.2 事件重复推送幂等处理不能偷懒飞书为了保证事件不丢失会做重试推送。如果你的服务端处理超时或者返回了非2xx响应飞书会重新推送同一条事件。这样就会出现同一个用户消息被处理两次、机器人回复两次的情况。我的做法是在Redis里记录消息ID处理前先判断是否已经处理过def is_duplicate(message_id: str) - bool: key ffeishu_msg:{message_id} if redis.set(key, 1, nxTrue, ex60): return False return True如果服务重启、Redis没配也可以用多维表格的查找接口来判断但性能会差一些。这个幂等处理属于必须做的否则用户会看到机器人偶尔发重复消息观感很差。6.3 大模型回复里的特殊字符和格式问题大模型返回的内容有时候会带符号、Markdown标记、甚至[]()这类特殊字符。如果直接通过飞书文本消息发出去可能会被飞书解析成某种格式或者出现奇怪的展示。我在代码里加了清理逻辑把Markdown的加粗、链接等语法去掉只保留纯文本。import re def clean_text(text: str) - str: # 去掉 某人的格式 text re.sub(r_user_\d, , text) # 去掉 Markdown 链接语法 text re.sub(r\[([^\]])\]\([^)]*\), r\1, text) return text.strip()还有个很隐蔽的问题大模型输出里如果带了{或者}直接作为JSON字符串发送可能没问题但如果拼接飞书卡片card消息结构可能被破坏。所以我跟大模型约定不要输出JSON不要输出复杂格式只用纯文本。6.4 监控告警用Uptime Kuma盯住机器人的健康状态机器人也是服务服务就会挂。我对服务健康最直接的需求是当服务不可用时第一时间在飞书群里收到告警。这个需求用自托管监控工具Uptime Kuma就能实现它内置了飞书Webhook通知方式。配置非常简单在Uptime Kuma里添加监控项目标地址填你的机器人服务健康检查接口比如/healthz然后在通知设置里添加一个飞书机器人Webhook。飞书群里建一个自定义机器人拿到Webhook地址填进去当Uptime Kuma检测到服务挂了就会往群里推告警。这个联动非常实用尤其是全天候这三个字意味着没有人盯着服务只有告警才能确保出问题时你能及时知道。我把它理解为整个系统最后一道保险丝。6.5 扩展思路定时任务、人为升级、语音记录上线稳定之后我还做了一些小扩展这里简单提两个方向定时任务每天早上9点把多维表格里待处理的任务汇总后发到群里。用APScheduler或者系统的cron请求你的一个内部接口都能实现。人工升级当大模型对某个问题的置信度过低或者问题里有人工转接等关键词时机器人自动IT值班同事并附上上下文。这两个扩展都建立在前面已打通的消息收发、多维表格、大模型三块能力上属于加个分支就完事的需求。整个飞书接入项目走到这一步你拥有的其实已经不仅仅是一个客服机器人了而是一个可以不断往上堆功能的自动化底座。我在实际使用中最深的体会是飞书把办公场景里最常用的消息和数据都变成了API你只需要把这两条线串起来剩下的想象力就是你的了。
返回列表