1. 为什么我要把飞书机器人和本地知识库接起来
公司内部的知识散落在飞书文档、Confluence、本地 Markdown 和一堆 PDF 里，每次有人问“报销标准是什么”“部署流程在哪”，都得靠人肉翻文档。我一开始想的是直接买个 SaaS 知识库，但数据合规过不了，而且我们很多技术文档是内部格式，上传到第三方平台不放心。于是决定走本地化路线：用 RAGFlow 在本地搭一套检索增强生成的知识库，再通过飞书机器人把问答入口放到大家每天都会用的聊天窗口里。
这条链路的核心价值在于：数据不出内网，问答入口零学习成本。飞书机器人负责接收用户消息、把问题转发给本地服务；本地服务调用 RAGFlow 的检索和对话接口，拿到答案后再回传给飞书。整个过程中，AI 智能体扮演的是“调度员”角色——它不直接生成答案，而是协调检索、重排、生成这几个环节，确保回答有据可查。
适合谁来参考这篇内容？如果你已经有一点 Python 基础，了解 HTTP 请求和 WebSocket 的基本概念，并且手头有一台能跑 Docker 的机器，那就可以跟着走一遍。完全没接触过 RAGFlow 也没关系，我会把关键配置和踩坑点都标出来。整条链路我实测跑了三周，日均处理 200+ 次问答请求，稳定性可以接受，但中间踩的坑确实不少，下面逐一拆解。
2. 链路拆解：消息从飞书到 RAGFlow 再回来，中间到底经过了什么
2.1 飞书机器人的两种消息接收模式：Webhook 与 WebSocket
飞书机器人接收消息有两种方式。第一种是Webhook 模式：你在飞书开放平台配置一个公网可访问的回调地址，用户发消息后飞书服务器主动 POST 过来。这种方式要求你的服务有公网 IP 或者内网穿透，对于纯内网环境不太友好。第二种是WebSocket 长连接模式：你的服务主动向飞书建立一条长连接，消息通过这条连接推送过来，不需要公网入口。我选的是 WebSocket 模式，因为我们的 RAGFlow 跑在内网机器上，没有对外暴露端口。
WebSocket 模式的关键在于心跳机制。飞书服务端和你的客户端之间需要定期交换心跳包来维持连接。如果心跳断了，飞书会认为你的服务不可用，消息就推不过来了。我在实现时用的是飞书官方 Python SDK 里的lark-oapi库，它内部封装了心跳逻辑，但你需要处理好断线重连。实测下来，网络抖动导致的断连大概每天会出现 1-2 次，所以重连逻辑必须写健壮。
2.2 RAGFlow 的检索与对话接口：不是简单的“一问一答”
RAGFlow 对外提供的是 RESTful API，核心有两个端点：一个是检索接口，传入问题返回相关文档片段；另一个是对话接口，传入问题和检索结果，返回生成的答案。很多人以为直接调对话接口就行了，但实际上检索质量决定了最终答案的上限。如果检索出来的片段不相关，再强的生成模型也救不回来。
我在配置 RAGFlow 时重点调了三个参数：similarity_threshold（相似度阈值）、top_k（返回片段数）和rerank（是否启用重排序）。相似度阈值设太低会引入噪声，设太高会漏掉关键信息。经过几轮测试，我把阈值定在 0.35，top_k 设为 5，并开启了重排序模型。这样既能保证召回率，又能把最相关的片段排到前面。
2.3 AI 智能体在链路中的角色：调度而非生成
这里的 AI 智能体不是指某个具体的大模型，而是一段调度逻辑。它做的事情包括：判断用户消息是普通提问还是指令、决定要不要走检索、把检索结果拼装成合适的 prompt、调用生成模型、最后把答案格式化后回传飞书。我用的是 ReAct 模式的思想——先推理当前该做什么，再执行动作，观察结果后继续推理下一步。比如用户问“帮我查一下上周的部署记录”，智能体会先识别出这是一个检索请求，然后调用 RAGFlow 检索接口，拿到结果后再决定是直接返回还是需要进一步追问。
这种调度逻辑的好处是可解释、可干预。如果某类问题回答不好，我可以直接看日志里智能体的推理步骤，定位是检索没召回还是生成跑偏了。
3. 环境准备：从零把 RAGFlow 和飞书 SDK 跑起来
3.1 RAGFlow 的 Docker 部署与模型配置
RAGFlow 官方推荐用 Docker Compose 部署。我用的是一台 16 核 32G 内存的 Ubuntu 22.04 机器，没有 GPU，纯 CPU 推理。拉取镜像后，docker compose up -d启动，默认端口是 9380。第一次启动会比较慢，因为要下载嵌入模型和重排序模型，大概等了 15 分钟。
启动完成后，访问http://localhost:9380进入管理界面。这里有个坑：默认的嵌入模型是 BAAI/bge-large-zh-v1.5，如果你的文档里有大量英文技术术语，建议换成 bge-m3，它对中英文混合场景更友好。我在初期用默认模型时，英文 API 文档的检索效果很差，换了 bge-m3 之后明显改善。
模型配置在docker/.env文件里，修改EMBEDDING_MODEL和RERANK_MODEL两个变量即可。改完后需要重启容器，并且重新解析已有文档，因为嵌入向量变了。
3.2 飞书开放平台的应用创建与权限配置
在飞书开放平台创建企业自建应用，拿到App ID和App Secret。然后开启机器人能力，在“事件订阅”里选择“使用长连接接收事件”。权限方面，至少需要开通：im:message（接收和发送消息）、im:message.group_at_msg（群里@机器人）、im:resource（上传下载文件）。如果你想让机器人能发表格，还需要sheets:spreadsheet权限。
配置完成后，把应用发布到企业内，只有发布后机器人才能被搜索到。我第一次忘了发布，在飞书里搜不到机器人，折腾了半天才发现。
3.3 Python 依赖安装与虚拟环境隔离
我用的 Python 3.10，创建虚拟环境后安装以下依赖：
python -m venv venv source venv/bin/activate pip install lark-oapi==1.2.0 pip install requests==2.31.0 pip install python-dotenv==1.0.0 pip install loguru==0.7.2lark-oapi是飞书官方 SDK，封装了 WebSocket 连接和消息解析。loguru用来打日志，比标准库的 logging 好用很多。注意版本号，我试过lark-oapi1.3.x 的某个版本有内存泄漏问题，跑一天后内存涨到 4G，换回 1.2.0 就正常了。
4. 核心代码实现：消息接收、检索调度与结果回传
4.1 WebSocket 长连接的建立与心跳保活
飞书 SDK 建立长连接的代码很简洁：
import lark_oapi as lark from lark_oapi.api.im.v1 import * def do_message_event(data: P2ImMessageReceiveV1) -> None: # 处理消息 pass event_handler = lark.EventDispatcherHandler.builder("", "") \ .register_p2_im_message_receive_v1(do_message_event) \ .build() cli = lark.ws.Client( app_id="your_app_id", app_secret="your_app_secret", event_handler=event_handler, log_level=lark.LogLevel.INFO ) cli.start()cli.start()是阻塞的，内部会自动处理心跳和重连。但要注意：如果主线程被其他任务阻塞，心跳可能会延迟。我一开始把 RAGFlow 的调用放在主线程里同步执行，结果处理一个耗时 10 秒的请求时，心跳超时导致连接断开。后来改成用线程池异步处理消息，主线程只负责接收和分发，问题就解决了。
4.2 消息解析与用户意图识别
飞书推送过来的消息是 JSON 格式，SDK 已经帮你解析成了对象。关键字段包括message.content（消息内容）、message.message_type（消息类型）、sender.sender_id（发送者 ID）。消息内容是一个 JSON 字符串，需要二次解析：
import json content = json.loads(data.event.message.content) user_text = content.get("text", "").strip()用户意图识别我做得比较简单：如果消息以“/”开头，当作指令处理（比如/help、/clear）；否则当作普通提问。更复杂的意图识别可以接一个小模型做分类，但实测下来，对于知识库问答场景，直接走检索就够了，没必要过度设计。
4.3 调用 RAGFlow 检索接口并组装 Prompt
RAGFlow 的检索接口是POST /api/v1/retrieval，请求体如下：
import requests def retrieve_from_ragflow(question: str) -> list: url = "http://localhost:9380/api/v1/retrieval" payload = { "question": question, "dataset_ids": ["your_dataset_id"], "top_k": 5, "similarity_threshold": 0.35, "rerank": True } headers = { "Content-Type": "application/json", "Authorization": "Bearer your_api_key" } resp = requests.post(url, json=payload, headers=headers, timeout=30) resp.raise_for_status() return resp.json().get("data", {}).get("chunks", [])拿到 chunks 后，把它们拼成 prompt：
def build_prompt(question: str, chunks: list) -> str: context = "\n\n".join([c["content"] for c in chunks]) prompt = f"""基于以下参考资料回答问题。如果资料中没有相关信息，直接说“根据现有资料无法回答”。 参考资料： {context} 问题：{question} 答案：""" return prompt这里有个细节：chunks 的顺序很重要。RAGFlow 返回的结果已经按相关性排序了，但如果你自己做了二次处理，一定要保持顺序。我试过按文档 ID 排序，结果最相关的片段被排到了后面，生成质量明显下降。
4.4 把答案回传给飞书：文本、富文本还是表格
飞书机器人支持发送纯文本、富文本（post）、卡片（interactive）和表格。纯文本最简单，但格式受限；富文本可以加粗、换行、加链接；卡片可以带按钮和交互。我一开始用纯文本，但答案里如果有代码块或列表，显示效果很差。后来改用富文本，代码块用等宽字体，列表用缩进，可读性好了很多。
发送消息的代码：
from lark_oapi.api.im.v1 import * def send_message(chat_id: str, text: str): client = lark.Client.builder() \ .app_id("your_app_id") \ .app_secret("your_app_secret") \ .build() request = CreateMessageRequest.builder() \ .receive_id_type("chat_id") \ .request_body( CreateMessageRequestBody.builder() .receive_id(chat_id) .msg_type("text") .content(json.dumps({"text": text})) .build() ).build() response = client.im.v1.message.create(request) if not response.success(): logger.error(f"发送失败: {response.code}, {response.msg}")如果要发表格，需要先创建飞书表格，再把表格链接发出来。这个流程比较绕，后面单独说。
5. 踩坑实录：那些让我熬夜排查的报错与异常
5.1 WebSocket 频繁断连：心跳超时与线程阻塞
前面提到过，主线程阻塞会导致心跳超时。但还有一个更隐蔽的原因：网络代理。如果你的机器配置了 HTTP 代理，lark-oapi的 WebSocket 连接可能会走代理，导致连接不稳定。我一开始没注意，后来在日志里看到连接建立后几秒就断开，排查了半天才发现是环境变量http_proxy在作怪。解决办法是在代码里显式设置no_proxy，或者直接取消代理配置。
另一个坑是飞书服务端的限流。如果你的机器人短时间内收到大量消息，飞书可能会主动断开连接。我在压测时模拟了 100 个并发用户，连接断了三次。后来加了消息队列，把并发请求排队处理，就稳定了。
5.2 RAGFlow 检索结果为空：相似度阈值与文档解析质量
检索结果为空是最常见的问题。原因通常有两个：一是相似度阈值设太高，二是文档解析质量差。我有一份 PDF 文档，里面是扫描件，RAGFlow 默认的 OCR 没识别出来，导致整个文档的嵌入向量都是乱的。后来换了 OCR 引擎，重新解析后才正常。
排查方法：先用一个你知道答案的问题去检索，看返回的 chunks 里有没有相关内容。如果没有，逐步降低相似度阈值，直到有结果为止。但注意，阈值太低会引入噪声，需要权衡。
5.3 消息重复消费：事件去重与幂等处理
飞书的事件推送可能会重复。如果你的服务处理慢，飞书没收到确认，就会重推。我遇到过同一条消息被处理了三次，用户收到了三条一样的回答。解决办法是做事件去重：用message_id作为唯一键，处理前先查一下是否已经处理过。可以用 Redis 存已处理的 message_id，设置 5 分钟过期。
import redis r = redis.Redis(host='localhost', port=6379, db=0) def is_duplicate(message_id: str) -> bool: key = f"feishu:msg:{message_id}" if r.exists(key): return True r.setex(key, 300, "1") return False5.4 长文本截断与分片策略
RAGFlow 对单次检索的文本长度有限制，如果用户问题很长，或者检索出来的 chunks 总长度超过模型上下文窗口，就会被截断。我一开始没注意，导致一些复杂问题的答案不完整。后来加了分片策略：如果 chunks 总长度超过 3000 字符，就只取前 3 个最相关的，并在 prompt 里说明“仅基于前三个片段回答”。
6. 进阶玩法：让机器人支持表格发送与批量文件解析
6.1 飞书机器人发送表格的完整流程
飞书机器人不能直接发一个表格文件，但可以发一个表格链接。流程是：先调用飞书表格 API 创建一个表格，写入数据，然后把表格的 URL 发给用户。创建表格的接口是POST /open-apis/sheets/v3/spreadsheets，写入数据用POST /open-apis/sheets/v2/spreadsheets/{spreadsheetToken}/values。
这个流程比较繁琐，我封装了一个函数：
def send_table(chat_id: str, headers: list, rows: list): # 1. 创建表格 # 2. 写入表头和数据 # 3. 发送表格链接 pass注意：创建表格需要sheets:spreadsheet权限，而且表格默认在应用的空间里，需要把链接权限设置为“企业内可阅读”。
6.2 RAGFlow 批量处理文件的技巧
RAGFlow 支持批量上传文件，但如果你有几百个文件，一个个上传很慢。我写了一个脚本，用requests批量调用上传接口，并发数控制在 5 左右，太快会被限流。另外，文件命名很重要：RAGFlow 会用文件名作为文档标题，如果文件名是乱码或编号，检索时很难定位。建议文件名包含关键信息，比如部署流程_v2.3_20260101.pdf。
6.3 用 AI 智能体做多轮对话与上下文管理
单轮问答够用，但有些问题需要多轮澄清。比如用户问“这个怎么配置”，机器人需要反问“你指的是哪个模块”。实现多轮对话的关键是维护会话上下文。我用 Redis 存每个用户的最近 5 轮对话，每次请求时把历史对话拼到 prompt 里。但要注意，上下文太长会挤占检索结果的空间，所以历史对话只保留摘要，不保留原文。
7. 稳定性与性能：我实测下来的调优经验
7.1 并发处理：线程池与异步 IO 的选择
飞书消息是异步推送的，但 RAGFlow 的调用是同步 HTTP 请求。如果串行处理，一个慢请求会阻塞后面的消息。我用concurrent.futures.ThreadPoolExecutor开了 10 个线程，每个线程独立处理一条消息。实测下来，10 个线程足够应对日均 200+ 的请求量。如果请求量更大，可以考虑用aiohttp做异步调用，但改造工作量不小。
7.2 缓存策略：减少重复检索
很多用户会问重复的问题，比如“报销标准”。如果每次都走一遍检索和生成，浪费资源。我在 Redis 里缓存了问题和答案的映射，key 是问题的 MD5，过期时间 1 小时。命中缓存时直接返回，响应时间从 3 秒降到 50 毫秒。但要注意，知识库更新后要主动清缓存，否则用户会拿到旧答案。
7.3 日志与监控：出问题时怎么快速定位
日志我分了三个级别：INFO 记录每条消息的收发和检索结果数量，WARNING 记录检索为空或生成超时，ERROR 记录接口报错和连接断开。用loguru按天切割日志文件，保留 7 天。另外，我加了一个简单的健康检查接口，每分钟检查一次 WebSocket 连接状态和 RAGFlow 可用性，异常时发飞书告警。
8. 一些让我少走弯路的实操心得
第一，不要一上来就追求完美。我一开始想把意图识别、多轮对话、表格发送全做完，结果卡了两周。后来先跑通“接收消息→检索→回传”这条最小链路，再逐步加功能，效率高很多。
第二，RAGFlow 的文档解析质量比模型选择更重要。我试过换各种嵌入模型，效果提升都不如把 PDF 解析质量搞好来得明显。扫描件一定要走 OCR，表格一定要用表格解析模式，代码块要保留格式。
第三，飞书机器人的消息长度有限制。单条消息最多 30KB，如果答案太长，需要分多条发送。我写了一个分片函数，按段落切分，每片不超过 2000 字符，依次发送。
第四，测试时用真实问题。我一开始用“测试问题 1”“测试问题 2”去测，检索效果很好。后来换成真实用户的问题，发现很多口语化表达检索不到。建议收集 50 个真实问题做回归测试。
第五，WebSocket 连接不要放在主线程。如果你还有其他任务要跑，比如定时同步文档，一定要把 WebSocket 放在独立线程或进程中，否则互相阻塞。
这套链路我跑了三周，中间重启过两次服务，都是因为内存泄漏（后来定位到是某个依赖库的问题）。整体来说，对于内部知识库问答场景，这套方案够用且可控。如果你也在做类似的事情，希望这些踩坑记录能帮你省下几个熬夜的晚上。
