1. 为什么我要折腾这条链路先说背景。我在团队里一直负责内部工具和效率系统这块日常最头疼的事情之一就是知识散落在各处。飞书群里聊过的技术方案、文档库里存着的部署手册、本地跑的一堆实验记录想找的时候翻半天找不到。后来我们本地部署了一套 RAGFlow 做知识库把文档、PDF、Markdown 全喂进去检索效果确实不错。但问题是每次要用都得打开浏览器、登录、手动输入问题团队里非技术同学根本不愿意用。飞书是大家每天都在用的工具如果能把 RAGFlow 的问答能力直接接到飞书机器人上在群里 一下就能问知识库那使用门槛就降到几乎为零了。这个想法听起来简单但真正动手做的时候从飞书机器人配置、事件订阅、WebSocket 长连接、到 RAGFlow 的 API 调用和流式返回处理中间踩了不少坑。这篇文章就是把这整条链路从头到尾讲清楚。适合谁看如果你手上有本地部署的 RAGFlow想让它在飞书里变成一个能问答的机器人或者你只是想学一下飞书机器人的事件订阅机制和 Python 后端的对接方式这篇都能直接抄作业。我会把每一步的操作、参数、代码都写出来包括我踩过的那些坑。整条链路的核心逻辑其实不复杂飞书用户发消息 → 飞书开放平台推送事件 → 我的 Python 服务接收事件 → 调用 RAGFlow 的对话 API → 拿到回答 → 通过飞书 API 把回答发回群里。难点在于飞书的事件订阅方式选择、RAGFlow API 的鉴权和参数格式、以及消息格式的转换。2. 整体架构设计与技术选型2.1 链路全貌消息是怎么从飞书走到 RAGFlow 再走回来的先把整条链路画清楚后面每一步操作你才知道自己在干什么。用户侧的动作很简单在飞书群里 机器人 或者私聊机器人发一个问题。飞书开放平台收到消息后会通过你配置的事件订阅方式把消息推送到你的服务端。你的服务端解析出问题文本调用 RAGFlow 的对话接口RAGFlow 在本地知识库里检索、生成回答把结果返回给你的服务端。服务端再把回答通过飞书的消息发送接口推回给用户。这里面有几个关键决策点飞书事件订阅用 WebSocket 长连接还是 Webhook 回调RAGFlow 用哪个 API 接口消息格式怎么处理下面逐个说。2.2 为什么选 WebSocket 长连接而不是 Webhook飞书开放平台的事件订阅有两种方式一种是 Webhook你需要有一个公网可访问的 HTTPS 地址飞书把事件 POST 过来另一种是 WebSocket 长连接你的服务主动连到飞书的服务器事件通过这条连接推过来。我选 WebSocket原因很实际不需要公网 IP 和域名。我的 RAGFlow 和这个中间服务都跑在内网机器上没有公网入口。用 Webhook 就得搞内网穿透或者反向代理多一层复杂度。不需要处理签名验证和加密解密。Webhook 模式下飞书会对推送内容做加密你需要处理解密逻辑。WebSocket 模式下 SDK 帮你处理了这些。连接维护简单。飞书的 Python SDK 提供了长连接客户端几行代码就能建立连接并注册事件处理器。当然 WebSocket 也有代价你需要维护长连接的稳定性断线要重连。但飞书 SDK 内部已经做了自动重连实际用下来很稳。注意WebSocket 长连接方式下飞书要求你的应用开启“长连接”模式这个在开发者后台的事件订阅配置里选。选错了模式事件推不过来。2.3 RAGFlow 的 API 选型对话接口 vs 检索接口RAGFlow 提供了好几类 API我一开始看文档的时候有点懵。简单梳理一下对话助手接口Chat Assistant这是最上层的能力你创建一个对话助手绑定一个知识库然后通过 API 发问题它自动完成检索和生成返回最终回答。适合我们这种“问答机器人”场景。检索接口Retrieval只做检索返回匹配的文本块不生成回答。适合你需要自己控制生成逻辑的场景。数据集管理接口用来上传文档、管理知识库的跟问答链路无关。我选的是对话助手接口因为它把检索和生成都封装好了我只需要传问题、拿回答。而且它支持流式返回对于长回答体验更好。2.4 技术栈清单把用到的东西列一下方便你对照准备组件用途版本参考Python后端服务语言3.10飞书 Python SDK处理飞书事件和消息lark-oapi 最新版requests调用 RAGFlow HTTP API最新版RAGFlow本地知识库0.15飞书开放平台应用机器人载体企业自建应用Python 环境建议用 3.10 以上因为飞书 SDK 和一些依赖对版本有要求。如果你还没装 Python去官网下载安装包安装时记得勾选“Add to PATH”不然后面命令行里调不到 python 命令。3. 飞书机器人配置全流程3.1 创建企业自建应用登录飞书开放平台进入开发者后台创建一个“企业自建应用”。名字随便起比如“知识库助手”。创建完之后你会拿到两个关键凭证App ID 和 App Secret。这两个东西后面代码里要用先记下来。这里有个容易忽略的点应用创建后默认是“开发中”状态只有你自己能用。要让团队其他人也能用需要发布版本并等待管理员审核。测试阶段可以先不发布用测试企业或者把自己加为可用成员就行。3.2 开启机器人能力在应用的功能配置里找到“机器人”这一项开启它。开启之后你可以设置机器人的名字、头像、描述。这些是用户在飞书里看到的样子建议设置得清楚一点比如描述写“基于本地知识库的智能问答助手”。开启机器人后还需要配置机器人的权限范围。至少要开通这几个权限接收消息im:message发送消息im:message:send_as_bot获取群组信息im:chat:readonly读取用户发给机器人的单聊消息im:message.p2p_msg:readonly权限没开够的话后面会出现“能收到事件但发不出消息”或者“收不到群消息”的情况。3.3 配置事件订阅为长连接模式这是最关键的一步。在“事件与回调”配置页面选择订阅方式。这里要选“使用长连接接收事件”而不是“将事件发送至开发者服务器”。选完之后在事件列表里添加你需要订阅的事件。核心是这两个接收消息 v2.0im.message.receive_v1用户发消息给机器人时触发。机器人进群im.chat.member.bot.added_v1机器人被拉进群时触发可选。添加事件后不需要填回调地址因为长连接模式下是你的服务主动连上去的。注意事件订阅的配置修改后需要发布版本才能生效。如果你在开发中状态测试确保用的是测试版本。3.4 发布与权限审批配置完成后进入“版本管理与发布”创建一个版本填写更新说明提交发布。企业管理员审核通过后应用就对全员可用了。如果只是自己测试可以在“测试企业与人员”里把自己加进去这样不用审核就能用。发布之后在飞书里搜索你的机器人名字应该能找到它。私聊发个消息试试如果服务端还没跑起来消息会显示发送失败或者没反应这是正常的。4. RAGFlow 本地部署与知识库准备4.1 Docker 部署 RAGFlow 的实操要点RAGFlow 官方推荐用 Docker Compose 部署。我假设你已经装好了 Docker 和 Docker Compose直接说几个关键点。拉取代码后进入 docker 目录先别急着docker compose up。有几个配置需要改.env 文件里面有个RAGFLOW_IMAGE变量指定镜像版本。建议用固定版本号别用 latest不然哪天自动更新了可能出兼容问题。端口映射默认是 80 端口映射到宿主机的 80。如果宿主机 80 被占了改成别的比如8080:80。资源限制RAGFlow 的解析和嵌入模型比较吃内存建议给 Docker 至少 8GB 内存。如果机器内存小可以在 compose 文件里限制各个服务的资源。启动命令就是docker compose -f docker-compose.yml up -d。第一次启动会拉镜像比较慢耐心等。启动后用docker compose logs -f看日志等到所有服务都 ready 了再访问。访问地址是http://你的机器IP:端口。默认账号是admin密码在 .env 文件里或者首次登录时设置。4.2 创建知识库并上传文档登录 RAGFlow 后先创建一个知识库。知识库的配置里有个关键选项嵌入模型Embedding Model。这个决定了文档怎么被向量化。RAGFlow 内置了一些模型也可以接外部的。我用的内置的够用。创建完知识库后上传文档。支持 PDF、Word、Markdown、TXT 等格式。上传后需要点“解析”RAGFlow 会对文档做分块和向量化。解析时间取决于文档数量和大小大文档可能要几分钟。实操心得文档解析质量直接影响问答效果。如果 PDF 是扫描版的RAGFlow 的 OCR 可能识别不准建议先用其他工具转成文字版再上传。另外文档分块的大小可以在知识库配置里调默认是 512 token如果文档里有很多长段落可以适当调大。4.3 创建对话助手并绑定知识库知识库准备好后创建一个“对话助手”。在助手配置里绑定你刚创建的知识库。还可以配置一些参数相似度阈值低于这个值的检索结果会被过滤掉。设太高会漏掉相关内容设太低会引入噪音。我一般设 0.2 到 0.3 之间。Top N检索返回的最大块数。默认 6 左右文档多的话可以调大。温度控制生成回答的随机性。问答场景建议设低一点0.1 到 0.3。创建完助手后你会得到一个Assistant ID这个后面调 API 要用。4.4 获取 API Key 和 Assistant ID在 RAGFlow 的右上角头像菜单里找到“API”或者“API Key”选项生成一个 API Key。这个 Key 是调用所有 RAGFlow API 的凭证。Assistant ID 在对话助手的详情页或者 URL 里能找到。URL 格式大概是http://你的地址/chat/assistant/xxxxxxxx那串 xxxxxxxx 就是 Assistant ID。把 API Key 和 Assistant ID 记下来后面代码里要用。5. Python 服务端核心实现5.1 环境准备与依赖安装先建一个项目目录然后创建虚拟环境。用虚拟环境的好处是依赖隔离不会跟系统里的其他 Python 包冲突。python -m venv venv source venv/bin/activate # Linux/Mac # 或者 venv\Scripts\activate # Windows然后安装依赖pip install lark-oapi requestslark-oapi是飞书官方 Python SDK封装了长连接、事件处理、消息发送等能力。requests用来调 RAGFlow 的 HTTP API。如果你 pip 安装慢可以换国内镜像源比如清华的源。这个网上教程很多不展开。5.2 飞书长连接客户端的初始化飞书 SDK 的长连接客户端初始化很简单核心是传入 App ID 和 App Secret然后注册事件处理器。import lark_oapi as lark from lark_oapi.api.im.v1 import * APP_ID 你的App ID APP_SECRET 你的App Secret # 创建事件处理器 event_handler lark.EventDispatcherHandler.builder(, ) \ .register_p2_im_message_receive_v1(handle_message) \ .build() # 创建长连接客户端 cli lark.ws.Client(APP_ID, APP_SECRET, event_handlerevent_handler, log_levellark.LogLevel.INFO) # 启动 cli.start()这里的handle_message是你自己定义的函数用来处理收到的消息。register_p2_im_message_receive_v1注册的是“接收消息 v2.0”事件。注意EventDispatcherHandler.builder(, )这两个空字符串是给 Webhook 模式用的加密 key 和 verification token。长连接模式下不需要传空就行。5.3 消息事件的解析与去重飞书推过来的消息事件是一个嵌套比较深的结构。核心字段在event.message里message_id消息的唯一 ID用来去重。chat_id会话 ID回复消息时要用。message_type消息类型文本是text。content消息内容是一个 JSON 字符串文本消息里是{text: 用户发的内容}。解析代码大概长这样import json def handle_message(data): event data.event message event.message # 只处理文本消息 if message.message_type ! text: return # 解析内容 content json.loads(message.content) user_text content.get(text, ).strip() # 去掉 机器人 的部分 user_text user_text.replace(_user_1, ).strip() if not user_text: return # 调用 RAGFlow 获取回答 answer ask_ragflow(user_text) # 回复消息 reply_message(message.chat_id, answer)去重这块飞书可能会重复推送同一个事件比如网络抖动时。建议用一个内存集合或者 Redis 记录处理过的message_id处理前先检查。简单场景下用内存集合就行重启后丢失也无所谓。5.4 调用 RAGFlow 对话 APIRAGFlow 的对话 API 是 OpenAI 兼容格式的这降低了对接成本。接口地址是http://你的RAGFlow地址/api/v1/chats/{assistant_id}/completions。请求体大概是这样import requests RAGFLOW_BASE http://你的RAGFlow地址 API_KEY 你的API Key ASSISTANT_ID 你的Assistant ID def ask_ragflow(question): url f{RAGFLOW_BASE}/api/v1/chats/{ASSISTANT_ID}/completions headers { Authorization: fBearer {API_KEY}, Content-Type: application/json } payload { question: question, stream: False, session_id: None # 不传则每次新会话 } resp requests.post(url, headersheaders, jsonpayload, timeout60) resp.raise_for_status() data resp.json() # 提取回答 answer data.get(data, {}).get(answer, 抱歉我没有找到相关信息。) return answer这里有几个参数要注意stream设 False 表示一次性返回完整回答。设 True 是流式返回需要处理 SSE。飞书消息发送不支持流式更新除非用卡片所以简单场景用 False。session_id传 None 表示每次都是新会话不保留上下文。如果你想让机器人记住对话历史需要维护 session_id 的映射。简单场景下不传就行。踩坑记录RAGFlow 的 API 返回结构里回答在data.answer字段。但如果你开了流式返回的是一行行 SSE 数据每行是data: {...}需要逐行解析。我一开始没注意直接resp.json()报错了。5.5 回复消息到飞书回复消息用飞书 SDK 的消息发送接口。核心是构造一个CreateMessageRequest指定接收者chat_id、消息类型和内容。def reply_message(chat_id, text): client lark.Client.builder() \ .app_id(APP_ID) \ .app_secret(APP_SECRET) \ .build() # 构造消息内容 content json.dumps({text: text}) request CreateMessageRequest.builder() \ .receive_id_type(chat_id) \ .request_body( CreateMessageRequestBody.builder() .receive_id(chat_id) .msg_type(text) .content(content) .build() ) \ .build() response client.im.v1.message.create(request) if not response.success(): print(f发送失败: {response.code} {response.msg})注意receive_id_type要跟receive_id的类型匹配。用 chat_id 就传chat_id用 open_id 就传open_id。实操心得飞书消息有长度限制文本消息最多 150KB 左右。RAGFlow 的回答一般不会超但如果知识库返回的内容特别长建议截断或者分段发送。另外飞书对消息发送有频率限制同一个机器人每秒最多发几十条正常问答场景不会触发。6. 常见问题与排查技巧实录6.1 事件收不到从配置到代码的排查顺序这是最常见的问题。机器人配置好了消息也发了但服务端就是没反应。排查顺序建议这样检查事件订阅模式。确认开发者后台选的是“长连接”而不是“Webhook”。选错了事件根本不会推过来。检查事件是否添加。在事件列表里确认“接收消息 v2.0”已经添加。没添加的话飞书不会推。检查版本是否发布。配置修改后必须发布新版本才生效。开发中状态用测试版本。检查服务端日志。长连接建立成功的话SDK 会打印连接日志。如果连日志都没有说明 App ID 或 Secret 错了。检查权限。接收消息需要im:message权限没开的话事件推不过来。我遇到过一次配置都对但就是收不到最后发现是应用版本没发布一直在用旧版本。这个坑很隐蔽因为后台看起来配置都改了。6.2 消息发不出去权限与 ID 类型问题能收到事件但回复失败通常是这几个原因发送消息权限没开。需要im:message:send_as_bot权限。receive_id_type 不匹配。用 chat_id 回复群消息用 open_id 回复私聊。混用会报错。机器人不在群里。如果机器人被移出群了发消息会失败。消息内容格式错误。文本消息的 content 必须是{text: ...}的 JSON 字符串直接传字符串会报错。排查方法就是看 SDK 返回的 response里面有 error code 和 message对照飞书开放平台的错误码文档查。6.3 RAGFlow 返回空或超时RAGFlow 调用失败一般有几种情况API Key 错误返回 401。检查 Key 是否复制完整有没有多余空格。Assistant ID 错误返回 404。检查 ID 是否正确助手是否被删了。知识库没解析完如果文档还在解析中检索可能返回空。等解析完成再试。超时RAGFlow 生成回答可能比较慢尤其是文档多、模型大的时候。建议把 requests 的 timeout 设长一点比如 60 秒。如果经常超时考虑用流式返回边生成边处理。踩坑记录我有一次调 API 一直返回空查了半天发现是知识库的相似度阈值设太高了0.8检索结果全被过滤了。调到 0.2 之后正常。这个参数没有标准值要根据你的文档特点调。6.4 长连接断线重连的处理飞书 SDK 的长连接客户端内部有自动重连机制但如果你发现服务运行一段时间后收不到消息了可能是连接断了没重连上。建议加一个心跳检测或者定时重启。简单做法是用 systemd 或者 supervisor 守护进程挂了自动拉起。或者在代码里捕获异常出错后重新cli.start()。另外长连接对网络稳定性有要求。如果服务器网络抖动频繁连接可能会断。这种情况可以考虑加一个重试逻辑或者换 Webhook 模式如果有公网地址的话。6.5 常见问题速查表现象可能原因解决方法收不到事件订阅模式选错改为长连接模式收不到事件事件未添加添加 im.message.receive_v1收不到事件版本未发布发布新版本发不出消息权限不足开通 im:message:send_as_bot发不出消息ID 类型不匹配检查 receive_id_typeRAGFlow 返回 401API Key 错误重新生成 KeyRAGFlow 返回空阈值太高调低相似度阈值RAGFlow 超时生成太慢加大 timeout 或用流式连接断开网络抖动加重连或守护进程7. 进阶优化与扩展思路7.1 支持富文本和表格消息纯文本消息能覆盖大部分场景但有时候 RAGFlow 返回的内容里有表格或者代码块纯文本展示效果很差。飞书支持富文本消息post 类型和卡片消息interactive 类型。富文本消息的 content 结构比较复杂是一个嵌套的 JSON。简单做法是把 Markdown 转成飞书的富文本格式。飞书也支持直接发 Markdown 卡片用 interactive 类型里面放一个 markdown 元素。实操心得如果你的知识库回答里经常有表格建议用卡片消息。卡片消息的 markdown 支持表格渲染展示效果好很多。但卡片消息的构造比文本消息复杂需要多写一些代码。7.2 多轮对话与上下文管理默认情况下每次提问都是独立的机器人不记得上一轮说了什么。如果要支持多轮对话需要维护 session_id。RAGFlow 的对话 API 支持传 session_id。你可以在第一次调用时拿到返回的 session_id存起来比如用 chat_id 做 key下次同一个会话的提问带上这个 session_idRAGFlow 就会保留上下文。但要注意session_id 有有效期过期了要重新创建。另外多轮对话会消耗更多 token如果知识库文档多成本会上升。7.3 流式返回与打字机效果RAGFlow 支持流式返回回答是一个字一个字生成的。如果能在飞书里实现打字机效果体验会好很多。但飞书的消息发送接口不支持更新已发送的消息除非用卡片的消息更新接口。实现思路是先发一条“正在思考...”的消息拿到 message_id然后用流式接口逐段获取回答每获取一段就更新那条消息。飞书的卡片消息支持更新所以需要用卡片消息。这个方案复杂度较高但效果确实好。如果团队对体验要求高值得投入。7.4 知识库文档的自动同步如果知识库的文档经常更新手动上传很麻烦。RAGFlow 提供了数据集管理的 API可以用脚本定期扫描某个目录把新文档自动上传并解析。这个脚本可以做成定时任务比如每天凌晨跑一次。扫描目录里比上次新增或修改的文件调 RAGFlow 的上传接口然后触发解析。这样就实现了知识库的自动更新。实操心得自动同步的时候要注意去重。RAGFlow 上传文档时会根据文件名判断同名文件会覆盖。如果你的文档命名不规范可能会误覆盖。建议在文件名里带上版本号或者日期。7.5 监控与日志服务跑起来之后建议加一些监控和日志。至少记录这几类信息收到的消息内容和用户 ID调用 RAGFlow 的耗时和返回状态发送消息的结果异常和错误堆栈日志可以用 Python 的 logging 模块输出到文件配合 logrotate 做切割。如果团队有 ELK 或者类似日志系统可以对接进去。监控方面可以定期检查长连接是否存活RAGFlow 是否可访问。如果发现异常通过飞书或者其他渠道告警。8. 一些掏心窝子的经验这套东西我从零搭到稳定运行大概花了两三天其中大部分时间是在踩坑和调试。回头看有几个点如果一开始就知道能省不少时间。第一飞书的事件订阅模式一定要先确认。我一开始选了 Webhook折腾了半天内网穿透后来发现长连接模式根本不需要公网地址直接换过来十分钟就通了。所以动手之前先把文档看仔细选对模式。第二RAGFlow 的 API 文档不算特别详细有些参数的含义要自己试。比如 session_id 传 None 和传空字符串的区别相似度阈值的实际效果这些文档里没写清楚得自己调。建议先用 curl 或者 Postman 把 API 调通再写代码集成。第三长连接的稳定性比想象中好但也不是万无一失。我遇到过运行几天后连接断了没重连的情况后来加了守护进程才稳定。如果你的服务要长期跑一定要考虑进程守护和自动重启。第四知识库的质量决定问答的质量。RAGFlow 只是个工具它检索和生成的效果很大程度上取决于你喂进去的文档。文档结构清晰、内容准确、分块合理问答效果就好。反之如果文档乱七八糟再好的模型也救不了。最后分享一个小技巧调试的时候把 RAGFlow 的返回原始 JSON 打印出来看看检索到了哪些块、相似度是多少。这样你能直观地判断是检索问题还是生成问题调优的时候有方向。我一开始只看最终回答调了半天不知道问题出在哪后来打印了原始返回一眼就看出是阈值设太高了。这个链路后续还可以扩展很多方向比如接入多个知识库根据问题自动路由、支持图片和文件消息、把问答记录存下来做分析等等。但核心链路跑通之后这些都是锦上添花的事情了。