1. 这套链路到底解决了什么问题先把场景说清楚。公司内部有一堆制度文档、产品手册、运维规范平时散落在各个角落同事想查个东西要么在群里问要么翻半天文件夹。飞书是大家每天必开的工具RAGFlow 是这两年本地化知识库问答里比较能打的开源方案把这两头接起来就能实现「在飞书里 一下机器人直接问知识库答案带着原文出处回来」。我这次搭的链路核心就三段飞书机器人负责接收消息和回传答案中间一个 Python 服务做协议转换和业务编排本地 RAGFlow 负责检索和生成。听起来简单但真动手会发现坑集中在三个地方——飞书的回调验签和消息去重、RAGFlow 的 API 调用姿势、以及长连接和超时的处理。这篇就把这三块掰开揉碎讲从零到跑通包括我踩过的每一个坑。适合谁看如果你手上有本地部署的 RAGFlow想让它在飞书里变成一个能用的问答入口或者你正在做类似的「IM 知识库」集成这篇可以直接抄作业。不需要你是飞书开放平台老手但至少得会装 Python、会看日志、能改配置文件。先说结论性的架构选择后面再展开为什么。整体走的是飞书事件订阅长连接模式 本地 Python 中转服务 RAGFlow HTTP API的组合。没有用公网 IP没有配内网穿透没有搞复杂的网关一台能跑 RAGFlow 的机器上再起一个 Python 进程就够了。这个选择对中小团队特别友好因为省掉了域名、证书、公网暴露这一整套运维负担。提示本文所有操作均在本地内网环境完成不涉及任何公网暴露配置。如果你的飞书应用需要外网访问请自行评估安全策略本文不展开这部分。2. 整体架构设计与选型思路2.1 为什么是长连接而不是 Webhook飞书机器人接收消息有两条路一是 Webhook 回调飞书把事件 POST 到你配置的公网地址二是长连接WebSocket你的服务主动连飞书的网关事件通过这条连接推过来。Webhook 的问题在于你必须有一个公网可达的 HTTPS 地址还得处理证书、域名、防火墙。对本地部署场景来说这基本等于要额外维护一套反向代理。长连接就绕开了这个问题——你的服务主动往外连飞书通过已有连接推事件本地机器不需要任何入站端口。代价是长连接需要自己维护心跳和重连。飞书官方 SDK 已经把这块封装好了你只要调用ws.Client启动就行断线它会自动重连。我实测下来连续跑一周没有出现掉线不恢复的情况稳定性够用。2.2 为什么中间要加一层 Python 服务有人会问飞书机器人能不能直接调 RAGFlow 的 API技术上可以但实际不行。原因有三个第一飞书的事件格式和 RAGFlow 的请求格式完全对不上中间必须做字段映射和消息组装。第二RAGFlow 的对话接口是有状态的需要维护 session 和 conversation 的对应关系这个映射逻辑得有个地方存。第三你需要做消息去重、超时控制、错误兜底这些都不适合塞进飞书的事件处理里。所以中间这层 Python 服务本质是个适配器 状态管理器。它对外接飞书的长连接对内调 RAGFlow 的 HTTP 接口中间维护一张「飞书会话 → RAGFlow 会话」的映射表。这个设计的好处是两边解耦以后换 IM 或者换知识库只改一边就行。2.3 RAGFlow 用 API 还是用 SDKRAGFlow 提供 HTTP API社区也有非官方的 Python 封装。我建议直接用 HTTP API原因很实在版本迭代快SDK 经常跟不上HTTP 接口文档清晰出问题好排查而且你不需要额外装依赖requests就够了。RAGFlow 的核心接口就两个一个是创建/获取对话会话一个是发起提问。前者拿到conversation_id后者带着question和conversation_id去请求返回答案和引用片段。整个交互非常直白没有复杂的鉴权流程一个 API Key 走天下。2.4 组件版本与依赖清单我这次用的环境如下供参考组件版本说明操作系统Windows 11 / Ubuntu 22.04两个环境都测过Python3.103.9 以下部分库不兼容RAGFlow本地 Docker 部署默认 9380 端口飞书 SDKlark-oapi 最新版官方 Python SDK网络库requests调 RAGFlow 用Python 依赖就三个lark-oapi、requests、websocketsSDK 内部会用到。装的时候注意lark-oapi对 Python 版本有要求3.8 以下会报语法错误建议直接上 3.10。3. 飞书机器人配置的完整流程3.1 创建应用与获取凭证进飞书开放平台创建一个「企业自建应用」。创建完你会拿到两个关键东西App ID和App Secret。这两个是后面 Python 服务连接飞书的凭证相当于账号密码别泄露。然后去「权限管理」里开权限。机器人要能收消息、发消息至少需要这几个权限接收消息、发送消息、获取与发送单聊消息、以应用身份发消息。权限开完记得发布版本不发布权限不生效这一步很多人会漏。接着去「事件订阅」页面选择「使用长连接接收事件」然后添加事件「接收消息」。这里有个细节长连接模式下不需要填请求地址飞书会通过你建立的连接推事件。如果你看到页面还要求填 URL说明你选错了订阅方式。3.2 机器人能力与可见范围在「应用功能」里开启「机器人」能力。开启后可以设置机器人名称、头像、描述。可见范围建议先设成「仅自己可见」做测试跑通后再放开给全员。有个坑要注意机器人默认只能在被 的时候收到消息。如果你想让它在单聊里直接回复需要在事件里判断消息类型。群聊里必须 机器人这是飞书的机制改不了。3.3 长连接模式的连接验证配置完成后写个最小脚本验证连接能不能建立。核心代码就几行import lark_oapi as lark def do_message_receive(data): print(收到消息:, data) event_handler lark.EventDispatcherHandler.builder(, ) \ .register_p2_im_message_receive_v1(do_message_receive) \ .build() ws_client lark.ws.Client( app_id你的APP_ID, app_secret你的APP_SECRET, event_handlerevent_handler, log_levellark.LogLevel.DEBUG ) ws_client.start()跑起来后在飞书里给机器人发条消息控制台应该能打印出事件内容。如果没反应先看日志里有没有连接成功的提示再看权限和事件订阅有没有配对。注意EventDispatcherHandler.builder的两个参数是加密 key 和验证 token长连接模式下可以留空但如果你在开放平台配了就得填上否则验签会失败。4. RAGFlow 本地知识库的准备与调用4.1 知识库创建与文件解析RAGFlow 部署好之后第一件事是建知识库、传文档。这里有个经验文档解析质量直接决定问答质量。RAGFlow 支持 PDF、Word、Markdown、TXT 等格式但 PDF 里的表格和扫描件解析效果参差不齐。我的做法是能转 Markdown 的先转 Markdown 再传表格单独整理成结构化文本。RAGFlow 的解析配置里有个「分块大小」和「重叠长度」默认值对中文文档偏大建议把分块调到 300-500 字符重叠 50-80 字符这样检索粒度更细召回更准。解析完成后知识库会显示每个文档的分块数量。如果某个文档分块数是 0说明解析失败点进去看日志通常是编码问题或者文件损坏。4.2 获取 API Key 与对话 ID在 RAGFlow 的「API」页面生成一个 API Key。然后在「对话」页面创建一个助手Assistant绑定你的知识库。创建完进入助手详情URL 里会有一个dialog_id这个后面要用。RAGFlow 的对话接口大致是这样import requests RAGFLOW_BASE http://127.0.0.1:9380 API_KEY 你的API_KEY DIALOG_ID 你的对话ID headers { Authorization: fBearer {API_KEY}, Content-Type: application/json } # 创建会话 resp requests.post( f{RAGFLOW_BASE}/api/v1/conversation, headersheaders, json{dialog_id: DIALOG_ID, name: feishu_session} ) conversation_id resp.json()[data][id] # 提问 resp requests.post( f{RAGFLOW_BASE}/api/v1/conversation/completion, headersheaders, json{ conversation_id: conversation_id, question: 公司的报销流程是什么, stream: False } ) print(resp.json())返回结果里answer是答案reference是引用的原文片段。把这两个拼起来回给飞书用户就能看到答案和出处。4.3 会话映射的设计飞书的每个会话单聊或群聊应该对应 RAGFlow 的一个 conversation。我的做法是用一个字典存映射key 是飞书的chat_idvalue 是 RAGFlow 的conversation_id。第一次收到某个 chat 的消息时创建会话之后复用。这个映射要不要持久化看你的需求。如果服务重启后可以接受重新开始对话内存字典就够了。如果要保留上下文就存到 SQLite 或 Redis。我图省事用的内存字典重启后对话历史丢失但知识库问答本身不依赖历史影响不大。5. 核心代码实现与关键细节5.1 消息接收与去重飞书的事件推送有个特性同一条消息可能推送多次。如果你不做去重用户问一句机器人可能回三遍。去重的办法是用message_id做幂等收到消息先查这个 id 处理过没有。processed_ids set() def do_message_receive(data): event data.event msg event.message msg_id msg.message_id if msg_id in processed_ids: return processed_ids.add(msg_id) # 只处理文本消息 if msg.message_type ! text: return content json.loads(msg.content) question content.get(text, ).strip() # 去掉 机器人 的部分 question re.sub(r\S\s*, , question) chat_id msg.chat_id reply(chat_id, question)processed_ids用 set 存会有内存泄漏风险长期跑建议换成带过期时间的缓存或者定期清理。我跑了一周没清理内存占用可以忽略但生产环境还是规范点好。5.2 调用 RAGFlow 并组装回复回复逻辑分两步先调 RAGFlow 拿答案再把答案发回飞书。发消息用飞书 SDK 的im.v1.message.create接口。def reply(chat_id, question): conversation_id get_or_create_conversation(chat_id) resp requests.post( f{RAGFLOW_BASE}/api/v1/conversation/completion, headersheaders, json{ conversation_id: conversation_id, question: question, stream: False }, timeout60 ) result resp.json() answer result[data][answer] references result[data].get(reference, []) # 组装回复文本 text answer if references: text \n\n---\n参考来源\n for i, ref in enumerate(references[:3], 1): text f{i}. {ref.get(content, )[:100]}...\n send_message(chat_id, text)这里有个细节RAGFlow 返回的reference结构可能因版本不同而有差异有的版本是chunks有的是reference。建议先打印一次完整返回看清楚字段名再写代码。5.3 超时与异常处理RAGFlow 生成答案可能比较慢尤其是文档多、模型大的时候。我设了 60 秒超时超过就返回「知识库响应超时请稍后再试」。如果不设超时飞书那边可能先超时用户看到的是机器人没反应。异常处理要覆盖三种情况RAGFlow 连不上、返回格式异常、飞书发送失败。每种都要有兜底回复不能让用户干等。try: resp requests.post(..., timeout60) resp.raise_for_status() result resp.json() except requests.Timeout: send_message(chat_id, 知识库响应超时请稍后再试) return except Exception as e: send_message(chat_id, f处理出错{str(e)[:50]}) return5.4 长文本的分段发送飞书单条消息有长度限制RAGFlow 的答案加上引用很容易超。我的做法是超过 2000 字符就分段发或者只发答案的前 1500 字符剩下的让用户点「查看详情」。分段发送要注意顺序飞书消息是异步的连续发多条可能乱序。稳妥的做法是加个短延迟或者用飞书的批量发送接口。6. 踩坑实录与排查技巧6.1 机器人收不到消息这是最高频的问题。排查顺序先看长连接有没有建立成功日志里有connected字样再看权限有没有发布最后看事件订阅里「接收消息」有没有添加。我遇到过一次权限开了但没发布版本折腾了半小时才发现。飞书开放平台的权限修改后必须重新发布应用版本才生效这个设计很容易让人踩坑。还有一种情况是机器人被拉进群了但群里 它没反应。检查一下机器人的可见范围如果群不在可见范围内消息不会推过来。6.2 RAGFlow 返回空答案RAGFlow 返回空答案通常是两个原因知识库里没有相关内容或者检索阈值设太高。RAGFlow 的助手配置里有个「相似度阈值」默认 0.2如果设成 0.8很多相关问题会被过滤掉。我的建议是先把阈值调到 0.1 做测试确认链路通了再慢慢往上调。另外如果知识库刚上传文档还没解析完检索也是空的等解析进度到 100% 再试。6.3 中文乱码与编码问题Windows 环境下跑 Python控制台输出中文经常乱码。解决办法是在脚本开头加import sys sys.stdout.reconfigure(encodingutf-8)如果是文件读写乱码统一用encodingutf-8。RAGFlow 返回的 JSON 默认是 UTF-8requests会自动处理一般不用手动 decode。6.4 长连接频繁断开长连接断开通常是网络不稳定或者心跳没配好。飞书 SDK 默认有心跳机制但如果你的网络环境有代理或者防火墙可能会干扰。检查一下有没有设置HTTP_PROXY之类的环境变量有的话清掉。另外如果服务跑在容器里容器的网络策略可能限制长连接。我试过在 Docker 里跑需要加--network host才能稳定连接。6.5 常见问题速查表现象可能原因排查方法机器人无响应长连接未建立看日志有无 connected权限报错权限未发布重新发布应用版本答案为空阈值过高/未解析完调低阈值检查解析进度回复重复消息未去重用 message_id 做幂等超时无回复未设超时兜底加 timeout 和异常处理中文乱码编码未指定统一 utf-87. 性能优化与扩展方向7.1 流式输出提升体验RAGFlow 支持流式返回stream: True答案会一段段吐出来。飞书这边可以用「更新消息」的方式实现打字机效果先发一条「正在思考...」然后不断更新这条消息的内容。这个体验提升很明显用户不用干等十几秒。实现上稍微复杂一点需要维护消息 id 和流式内容的对应关系。如果追求简单非流式也够用。7.2 多知识库路由如果公司有多个知识库比如制度库、产品库、技术库可以根据问题内容路由到不同的助手。简单做法是关键词匹配复杂点可以用一个小模型做意图分类。RAGFlow 本身支持多知识库也可以在助手层面配置。7.3 加缓存减少重复调用同样的问题反复问每次都调 RAGFlow 很浪费。可以在中间层加一层缓存key 是问题的哈希value 是答案。缓存有效期设个几小时既能减少调用又能保证答案不太旧。7.4 日志与监控生产环境一定要打日志记录每次请求的问题、耗时、是否命中缓存、RAGFlow 返回状态。出问题的时候日志是唯一的线索。我用的是 Python 标准 logging输出到文件按天切割。监控方面可以统计每天的提问量、平均响应时间、错误率。这些数据能帮你判断知识库质量和服务稳定性。8. 一些实操心得搭这套东西最耗时的不是写代码而是配置和调试。飞书的权限体系、RAGFlow 的解析配置每个环节都有细节。我的建议是分步验证先单独验证飞书长连接能收到消息再单独验证 RAGFlow API 能返回答案最后把两边接起来。不要一上来就写完整逻辑出了问题根本不知道是哪一环。另外RAGFlow 的文档解析质量真的决定一切。我见过太多人抱怨问答不准结果一看知识库里全是扫描件 PDF解析出来一堆乱码。花时间把文档整理好比调任何参数都管用。最后说个细节飞书机器人的回复最好带上「参考来源」这样用户能自己判断答案可不可信。RAGFlow 返回的 reference 字段就是干这个的别浪费。