企业官网建设流程全解析

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 还是用 SDK

RAGFlow 提供 HTTP API，社区也有非官方的 Python 封装。我建议直接用 HTTP API，原因很实在：版本迭代快，SDK 经常跟不上；HTTP 接口文档清晰，出问题好排查；而且你不需要额外装依赖，requests就够了。

RAGFlow 的核心接口就两个：一个是创建/获取对话会话，一个是发起提问。前者拿到conversation_id，后者带着question和conversation_id去请求，返回答案和引用片段。整个交互非常直白，没有复杂的鉴权流程，一个 API Key 走天下。

2.4 组件版本与依赖清单

我这次用的环境如下，供参考：

Python 依赖就三个：lark-oapi、requests、websockets（SDK 内部会用到）。装的时候注意，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_handler=event_handler, log_level=lark.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": f"Bearer {API_KEY}", "Content-Type": "application/json" } # 创建会话 resp = requests.post( f"{RAGFLOW_BASE}/api/v1/conversation", headers=headers, json={"dialog_id": DIALOG_ID, "name": "feishu_session"} ) conversation_id = resp.json()["data"]["id"] # 提问 resp = requests.post( f"{RAGFLOW_BASE}/api/v1/conversation/completion", headers=headers, json={ "conversation_id": conversation_id, "question": "公司的报销流程是什么", "stream": False } ) print(resp.json())

返回结果里answer是答案，reference是引用的原文片段。把这两个拼起来回给飞书，用户就能看到答案和出处。

4.3 会话映射的设计

飞书的每个会话（单聊或群聊）应该对应 RAGFlow 的一个 conversation。我的做法是用一个字典存映射，key 是飞书的chat_id，value 是 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", headers=headers, json={ "conversation_id": conversation_id, "question": question, "stream": False }, timeout=60 ) 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(..., timeout=60) 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]}") return

5.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(encoding='utf-8')

如果是文件读写乱码，统一用encoding='utf-8'。RAGFlow 返回的 JSON 默认是 UTF-8，requests会自动处理，一般不用手动 decode。

6.4 长连接频繁断开

长连接断开通常是网络不稳定或者心跳没配好。飞书 SDK 默认有心跳机制，但如果你的网络环境有代理或者防火墙，可能会干扰。检查一下有没有设置HTTP_PROXY之类的环境变量，有的话清掉。

另外，如果服务跑在容器里，容器的网络策略可能限制长连接。我试过在 Docker 里跑，需要加--network host才能稳定连接。

6.5 常见问题速查表

7. 性能优化与扩展方向

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 字段就是干这个的，别浪费。