1. 为什么我要折腾这条链路先说结论这套东西的核心价值是把「飞书里随口一问」和「本地知识库里躺着的一堆文档」接起来让同事在飞书群里 一下机器人就能拿到基于内部资料的准确回答而不是去翻十几个 PDF。我在一家做企业服务的公司负责内部工具日常最头疼的就是制度、流程、产品手册散落在各种地方。飞书是大家每天必开的工具RAGFlow 是我实测下来本地化部署体验比较顺的 RAG 引擎——它能解析 PDF、Word、Excel、PPT还能做向量检索和重排。把这两个东西接起来中间需要一个「AI 智能体」来调度接收飞书的消息事件去 RAGFlow 检索把结果整理成人话再回给飞书。整条链路涉及几个关键技术点飞书机器人的事件订阅WebSocket 长连接模式、RAGFlow的本地 Docker 部署和 API 调用、Python写的中间层服务、以及WebSocket心跳保活机制。我踩过的坑包括RAGFlow 解析中文 PDF 时表格错乱、飞书事件重复推送导致机器人复读、WebSocket 连接假死但进程不报错。下面把完整流程和这些坑一个个拆开讲。这篇文章适合谁看如果你手上有本地文档想做成问答机器人或者你已经在用飞书但不知道怎么接自建服务再或者你只是想搞明白 RAGFlow 到底怎么调 API那这篇应该能帮你省掉至少两个周末的试错时间。我假设你有基本的 Python 能力能看懂 Docker 命令剩下的细节我都会补上。2. 整体架构设计与选型考量2.1 为什么是「飞书 RAGFlow Python 中间层」这个组合市面上做知识库问答的方案不少我选这套组合的逻辑是这样的飞书作为入口是因为团队已经在用不需要额外推广成本。飞书开放平台对自建应用的支持比较完整机器人可以发消息、收消息、发卡片、发表格权限粒度也够细。相比自己搭一个 Web 聊天界面直接用飞书省掉了前端开发和用户教育。RAGFlow 作为检索引擎核心原因是它的文档解析能力。我对比过几个开源方案RAGFlow 对复杂版式 PDF 的处理明显更细——它会做版面分析把表格、标题、正文分开处理这对制度类文档特别重要。而且它自带 Web UI上传文档、看解析结果、调检索参数都很直观不用写代码就能验证效果。Python 作为中间层是因为飞书 SDK 和 RAGFlow 的 API 都是 HTTP 接口Python 的 requests 和 websockets 库能快速把两边串起来。中间层要做的事情其实不复杂收飞书事件、调 RAGFlow 检索、调大模型生成回答、回飞书消息。用 Python 写一个单文件服务就能跑起来不需要上框架。WebSocket 长连接模式这是飞书事件订阅的两种方式之一。另一种是 Webhook 回调需要你的服务有公网地址。我选长连接是因为本地部署环境没有公网 IP长连接由飞书 SDK 主动连出去不需要暴露端口省掉了内网穿透的麻烦。2.2 数据流向拆解整条链路的数据流是这样的用户在飞书群里 机器人发一条消息飞书服务器把消息事件通过 WebSocket 推给本地服务本地服务解析事件提取用户问题和会话 ID本地服务调 RAGFlow 的检索 API传入问题拿到相关文档片段本地服务把文档片段和问题拼成 Prompt调大模型 API 生成回答本地服务调飞书发消息 API把回答发回群里飞书服务器把消息推给用户这里面有两个容易出问题的地方一是第 2 步的事件推送可能重复飞书在没收到 ACK 时会重推二是第 4 步的检索质量直接决定回答质量RAGFlow 的解析配置很关键。2.3 部署形态选择本地 vs 云端RAGFlow 可以本地 Docker 部署也可以用云服务。我选本地部署的原因一是数据不出内网制度文档比较敏感二是本地部署没有调用次数限制调试阶段可以随便试三是 RAGFlow 的 Docker 镜像做得比较完善docker compose up就能跑起来。本地部署的代价是需要一台配置还行的机器。我的测试机是 Win11 WSL216G 内存跑 RAGFlow 的默认配置含 Elasticsearch、MySQL、Redis、MinIO勉强够用但解析大文件时会卡。后来换到一台 32G 的 Linux 服务器上就顺畅多了。如果你只是试用8G 内存也能跑起来但别指望同时解析多个大文件。3. 环境准备与 RAGFlow 本地部署3.1 硬件与系统要求先说硬性条件避免你装到一半发现跑不起来项目最低配置推荐配置说明内存8GB16GBRAGFlow 全家桶比较吃内存磁盘50GB100GB镜像 文档 向量数据CPU4 核8 核解析文档时吃 CPU系统Linux / Win11WSL2Ubuntu 22.04Win 下建议用 WSL2我实测下来Win11 直接跑 Docker Desktop 也能用但文件挂载的性能不如 WSL2。如果你在 Win11 上折腾建议把项目放在 WSL2 的文件系统里不要放在 Windows 盘符下否则解析速度会慢很多。3.2 RAGFlow Docker 部署实操RAGFlow 官方提供了 docker-compose 配置部署流程如下# 1. 克隆仓库 git clone https://github.com/infiniflow/ragflow.git cd ragflow/docker # 2. 修改 .env 文件设置 RAGFlow 版本 # 默认可能是 latest建议固定一个版本避免更新导致的不兼容 # 编辑 .env找到 RAGFLOW_IMAGE 这一行 # 3. 启动服务 docker compose -f docker-compose.yml up -d # 4. 查看启动状态 docker compose logs -f ragflow-server启动完成后访问http://localhost:80就能看到 RAGFlow 的 Web 界面。默认账号是admin密码在.env文件里可以查到首次登录后建议改掉。这里有个坑RAGFlow 启动时会初始化 Elasticsearch 的索引这个过程可能要几分钟。如果你看到 ragflow-server 容器一直重启先别急等 3-5 分钟再看。如果超过 10 分钟还在重启检查内存是不是不够——Elasticsearch 默认要 2G 堆内存内存不足会直接 OOM。3.3 知识库创建与文档解析技巧RAGFlow 跑起来后第一步是创建知识库。在 Web 界面点「知识库」→「创建知识库」填名称、选嵌入模型。嵌入模型我建议用BAAI/bge-large-zh-v1.5中文效果比默认的英文模型好很多。上传文档后RAGFlow 会自动解析。这里是我踩坑最多的地方PDF 解析RAGFlow 默认用 DeepDoc 解析器对扫描版 PDF 效果一般。如果你的 PDF 是图片扫描的需要先做 OCR。RAGFlow 支持配置 OCR 模型但配置起来比较麻烦。我的做法是先用其他工具把扫描 PDF 转成文字版再上传。表格处理制度文档里经常有表格RAGFlow 的表格解析有时候会把跨页表格拆成两个。解决办法是在解析配置里把「表格识别」打开并且把 chunk 大小调大一点让表格尽量在一个 chunk 里。分块策略RAGFlow 默认按 512 token 分块对制度类文档偏小。我建议调到 1024并且开启「标题层级识别」这样检索时能保留上下文。实测下来分块太小会导致检索到的片段不完整回答时缺关键信息。解析后验证上传文档后一定要点进「解析结果」看一眼。我遇到过解析出来的文本全是乱码的情况原因是 PDF 编码问题。这种文档需要先用工具转一遍编码再上传。3.4 获取 RAGFlow API KeyRAGFlow 的 API 调用需要 API Key。在 Web 界面右上角点用户头像 →「API」→「创建 API Key」复制保存。这个 Key 后面在 Python 代码里要用。RAGFlow 的检索 API 端点是/api/v1/retrieval请求体大概长这样{ question: 员工年假怎么算, dataset_ids: [你的知识库ID], top_k: 5, similarity_threshold: 0.2 }top_k是返回的片段数量similarity_threshold是相似度阈值。这两个参数需要根据你的文档调。我一般先用 top_k5、threshold0.2 试如果回答不准确就调大 top_k如果检索到无关内容就调高 threshold。4. 飞书机器人配置与 WebSocket 接入4.1 创建飞书自建应用飞书机器人的配置在飞书开放平台完成。流程是登录飞书开放平台点「创建企业自建应用」填应用名称、描述、图标在「权限管理」里开通需要的权限im:message收发消息、im:message.group_at_msg接收群聊中机器人消息、im:message.p2p_msg接收单聊消息在「事件订阅」里选择「使用长连接接收事件」添加事件im.message.receive_v1在「机器人」页面启用机器人发布应用等管理员审核这里有个关键点长连接模式不需要配置请求网址这是它比 Webhook 方便的地方。但长连接模式要求你的服务能主动连到飞书服务器所以本地服务需要能访问外网。4.2 获取应用凭证在「凭证与基础信息」页面能拿到App ID和App Secret这两个后面 Python 代码要用。注意 App Secret 不要泄露建议放在环境变量里而不是硬编码。4.3 WebSocket 长连接的工作原理飞书的长连接用的是 WebSocket 协议。你的服务启动后用 App ID 和 App Secret 换一个临时 token然后用这个 token 建立 WebSocket 连接。飞书服务器有事件时通过这个连接推给你。这里要理解一个概念WebSocket 是双向的但飞书的事件推送是单向的。你的服务只需要接收事件不需要通过这个连接发消息。发消息走的是另一个 HTTP API。长连接有个心跳机制飞书服务器会定期发 ping你的服务要回 pong。如果一段时间没收到 pong飞书会断开连接。飞书 SDK 一般会自动处理心跳但如果你自己实现 WebSocket 客户端就要注意这一点。4.4 Python 接入飞书 SDK飞书官方提供了 Python SDK叫lark-oapi。安装pip install lark-oapi用 SDK 建立长连接的代码大概是这样import lark_oapi as lark from lark_oapi.api.im.v1 import * # 创建客户端 client lark.Client.builder() \ .app_id(your_app_id) \ .app_secret(your_app_secret) \ .log_level(lark.LogLevel.DEBUG) \ .build() # 注册事件处理器 def do_message_receive(data: P2ImMessageReceiveV1) - None: # 处理消息 print(data) event_handler lark.EventDispatcherHandler.builder(, ) \ .register_p2_im_message_receive_v1(do_message_receive) \ .build() # 建立长连接 ws_client lark.ws.Client( your_app_id, your_app_secret, event_handlerevent_handler, log_levellark.LogLevel.DEBUG ) ws_client.start()这段代码跑起来后你的服务就会连到飞书的长连接服务器。在飞书里 机器人发消息do_message_receive就会被调用。5. 中间层服务核心逻辑实现5.1 消息去重解决机器人复读问题飞书的事件推送是「至少一次」语义也就是说同一条消息可能推多次。如果你不做去重机器人就会复读。我一开始没做去重结果群里 一次机器人它回了三条一样的消息非常尴尬。去重的思路是记录已处理的消息 ID。飞书的消息事件里有个message_id全局唯一。用一个集合或者 Redis 记录最近处理过的 message_id处理前先查一下处理过就跳过。processed_messages set() def do_message_receive(data): message_id data.event.message.message_id if message_id in processed_messages: return processed_messages.add(message_id) # 限制集合大小避免内存泄漏 if len(processed_messages) 10000: processed_messages.clear() # 处理消息...这个方案在单进程下够用。如果你部署了多个实例需要用 Redis 做共享存储。5.2 调用 RAGFlow 检索拿到用户问题后调 RAGFlow 的检索 APIimport requests def retrieve_from_ragflow(question, dataset_id, api_key): url http://localhost/api/v1/retrieval headers { Authorization: fBearer {api_key}, Content-Type: application/json } payload { question: question, dataset_ids: [dataset_id], top_k: 5, similarity_threshold: 0.2 } resp requests.post(url, jsonpayload, headersheaders, timeout30) resp.raise_for_status() return resp.json()返回结果里有个chunks数组每个元素包含content文档片段和similarity相似度。我一般会把 similarity 低于 0.3 的片段过滤掉避免无关内容干扰回答。5.3 拼接 Prompt 调大模型检索到的片段不能直接发给用户需要让大模型整理成通顺的回答。Prompt 的设计很关键我用的模板是你是一个内部知识助手。请根据以下参考资料回答用户问题。 如果参考资料中没有相关信息请直接说「知识库中没有找到相关内容」不要编造。 参考资料 {context} 用户问题{question} 请用简洁的中文回答不要重复问题。context是把检索到的片段拼起来每个片段前面加上来源标记。这样大模型回答时能引用来源用户也能判断可信度。调大模型 API 的部分我用的是兼容 OpenAI 格式的接口代码很简单def generate_answer(context, question, llm_api_key, llm_base_url): from openai import OpenAI client OpenAI(api_keyllm_api_key, base_urlllm_base_url) resp client.chat.completions.create( modelyour-model-name, messages[ {role: system, content: 你是一个内部知识助手。}, {role: user, content: f参考资料\n{context}\n\n问题{question}} ], temperature0.3 ) return resp.choices[0].message.contenttemperature设 0.3 是为了让回答稳定一点不要天马行空。5.4 回复飞书消息拿到回答后调飞书发消息 API 回给用户。飞书的消息 API 支持文本、富文本、卡片等多种格式。我一般用卡片因为可以加标题和分割线看起来更清晰。def reply_to_feishu(message_id, content): # 用 message_id 回复飞书会自动关联到原消息 url fhttps://open.feishu.cn/open-apis/im/v1/messages/{message_id}/reply headers { Authorization: fBearer {tenant_access_token}, Content-Type: application/json } payload { content: json.dumps({text: content}), msg_type: text } requests.post(url, jsonpayload, headersheaders)tenant_access_token需要用 App ID 和 App Secret 换有效期 2 小时需要缓存和自动刷新。5.5 完整流程串联把上面几步串起来主流程就是收到飞书消息事件去重检查提取问题文本调 RAGFlow 检索过滤低相似度片段拼接 Prompt 调大模型回复飞书消息整个流程是同步的但飞书的事件处理有超时限制大概 3 秒如果 RAGFlow 检索或大模型生成太慢飞书会认为你没收到然后重推。解决办法是把处理逻辑放到后台线程先快速返回 ACK再异步处理。import threading def do_message_receive(data): # 快速返回 threading.Thread(targetprocess_message, args(data,)).start()这样飞书收到 ACK 后就不会重推了。6. 常见问题与排查技巧实录6.1 飞书机器人不回复这是最常见的问题排查顺序如下现象可能原因排查方法完全没反应长连接没建立看日志有没有 WebSocket 连接成功有日志但没回复权限不足检查应用是否开通 im:message 权限回复超时处理逻辑太慢看 RAGFlow 和大模型调用耗时回复报错token 过期检查 tenant_access_token 刷新逻辑我遇到过一次长连接建立成功但收不到事件的情况原因是事件订阅里没添加im.message.receive_v1。这个事件必须手动添加不会自动订阅。6.2 RAGFlow 检索结果不准确检索质量差通常有三个原因文档解析质量差去 RAGFlow 的解析结果页面看如果文本是乱的检索肯定不准。解决办法是换解析器或预处理文档。分块策略不合适分块太大导致检索到的片段包含太多无关内容分块太小导致信息不完整。建议根据文档类型调制度类文档用 1024 token技术文档用 512。相似度阈值设置不当阈值太高会漏掉相关内容太低会引入无关内容。建议先用 0.2 试看检索结果再调。6.3 WebSocket 连接假死长连接跑一段时间后可能假死——进程还在但收不到事件。原因是网络波动导致连接断了但 SDK 没检测到。解决办法是加心跳检测。飞书 SDK 一般会自动重连但有时候重连失败会静默。我的做法是加一个定时任务每隔几分钟检查一次连接状态如果超过一定时间没收到任何事件就主动重连。import time import threading last_event_time time.time() def check_connection(): while True: time.sleep(60) if time.time() - last_event_time 300: # 5 分钟没收到事件重连 ws_client.stop() ws_client.start()这个方案有点粗暴但实测有效。更好的做法是用 SDK 提供的连接状态回调但飞书 SDK 这块文档不太全我还没找到稳定的方法。6.4 大模型回答编造内容这是 RAG 系统的通病。大模型看到参考资料里没有答案但为了「帮忙」会自己编。解决办法是在 Prompt 里明确要求「不知道就说不知道」并且把 temperature 调低。我还在 Prompt 里加了一句「回答末尾标注引用的参考资料编号」这样用户能判断回答是否可信。如果大模型编造往往引用编号也是错的用户一眼能看出来。6.5 内存泄漏Python 服务跑久了内存会涨常见原因是消息去重的集合没清理、HTTP 连接没关闭、大模型返回的对象没释放。我的做法是去重集合限制大小超过就清空用requests.Session复用连接定期重启服务简单粗暴但有效如果内存涨得特别快用tracemalloc排查一般能找到泄漏点。7. 一些实操心得与扩展思路7.1 关于 RAGFlow 的解析配置RAGFlow 的解析配置里有个「chunk 方法」选项默认是「Naive」还有「Manual」「QA」「Table」等。我实测下来制度类文档用「Naive」 标题层级识别效果最好技术文档用「Manual」手动分块更准。另外RAGFlow 支持「RAPTOR」摘要索引开启后会对文档做层次化摘要检索时能召回更宏观的内容。这个功能对长文档特别有用但会消耗更多计算资源。如果你的文档都是几十页的长文档建议开启。7.2 关于飞书消息格式飞书的消息卡片支持 Markdown 子集但和标准 Markdown 有差异。比如表格要用飞书自己的格式代码块要用**包裹。我建议先用纯文本跑通再慢慢换成卡片。飞书还支持「发送表格」消息可以把检索结果以表格形式展示。这个功能在展示多条检索结果时很有用但配置起来比文本复杂需要构造特定的 JSON 结构。7.3 关于性能优化如果并发量不大比如几十人用单进程 Python 服务完全够用。但如果要支持更多人需要考虑RAGFlow 检索加缓存相同问题直接返回缓存结果大模型调用加并发限制避免打爆 API飞书消息回复异步化不阻塞事件处理我实测下来单次问答的耗时大概是RAGFlow 检索 1-2 秒大模型生成 3-5 秒总共 5-7 秒。这个延迟在飞书里可以接受但如果用户问得频繁建议加个「正在思考」的中间状态提示。7.4 后续可以扩展的方向这套链路跑通后可以扩展的方向不少多知识库路由根据问题类型自动选择不同的知识库比如问人事的走人事库问技术的走技术库。RAGFlow 的检索 API 支持传多个 dataset_ids可以在中间层做路由。对话历史把用户的历史问答存起来下次提问时带上历史让大模型能理解上下文。这个需要在中间层维护会话状态。反馈机制在飞书消息里加「有用/没用」按钮收集用户反馈用来优化检索和 Prompt。定时同步如果知识库文档会更新可以写个定时任务自动同步新文档到 RAGFlow。我个人在实际操作中的体会是这套东西最难的不是写代码而是调优。RAGFlow 的解析参数、检索阈值、Prompt 模板每一个都需要根据你的文档特点反复试。我建议先用小批量文档跑通流程再逐步扩大知识库规模这样问题容易定位。最后再分享一个小技巧调试阶段把 RAGFlow 的检索结果和大模型的原始输出都打到日志里出问题时能快速定位是检索不准还是生成不准。这个日志在正式上线后也建议保留方便排查用户反馈的问题。