企业官网建设流程全解析

1. 为什么我要把飞书机器人和本地 RAGFlow 接在一起

先说清楚这套东西到底解决什么问题。公司内部有一堆制度文档、产品手册、运维手册，散落在各个共享盘里，同事想查个报销标准、查个接口规范，要么翻半天找不到，要么找到的是过期版本。市面上的在线知识库工具不是不能用，但数据要传到别人服务器上，很多团队是不接受的。所以我的目标很明确：文档全部留在本地，问答入口放在大家每天都在用的飞书里。

RAGFlow 这个开源知识库引擎正好卡在这个位置上。它做的是 RAG（检索增强生成）里最脏最累的那部分活：文档解析、切块、向量化、混合检索、重排。你把手册丢进去，它能把 PDF 里的表格、扫描件里的文字都抠出来，检索的时候还能同时走关键词和向量两条路。而飞书机器人负责的是"最后一公里"——同事不用装任何新软件，在群里 @ 一下机器人就能问。

中间缺的那一环，就是 AI 智能体。它要干三件事：接住飞书推过来的消息、去 RAGFlow 里检索出相关片段、把片段和问题一起交给大模型组织成人话再回给飞书。听起来简单，但真动手你会发现坑全在细节里：飞书的事件回调有签名校验和 URL 验证、RAGFlow 的 API 返回结构跟文档写的不完全一样、长连接和 Webhook 两种模式选错了后面全是麻烦。

这篇东西我按实际搭建顺序写，从环境准备到联调排错，每一步都写清楚"为什么这么做"。适合两类人看：一类是想给自己团队搭内部问答的运维或后端，另一类是已经跑通了 RAGFlow 但不知道怎么接到 IM 里的开发者。全程 Python，不需要前端基础。

提示：本文所有操作都在本地或内网环境完成，涉及的公网暴露部分请结合自己团队的网络安全规范评估，不要直接把调试端口开到公网。

2. 动手之前先把三块拼图的关系理清楚

2.1 飞书机器人、RAGFlow、智能体各自负责什么

很多人一上来就写代码，结果写到一半发现职责没分清，改起来牵一发动全身。我建议先把三个角色的边界画出来。

飞书机器人是消息的入口和出口。它有两种接消息的方式：一种是 Webhook，飞书把用户消息 POST 到你配置的一个公网 URL；另一种是长连接（WebSocket），你的程序主动连到飞书服务器，消息通过这条长连接推过来。这个选择非常关键，后面会专门讲。

RAGFlow是知识库的大脑。它对外提供 HTTP API，核心就两个动作：把文档传进去建知识库（dataset），以及拿一个问题去检索（retrieval）。检索返回的是一堆文本块（chunk）和它们的相似度分数，不是最终答案。

AI 智能体是中间那层胶水。它接收飞书的消息，调用 RAGFlow 检索，把检索结果拼成 prompt 交给大模型，再把大模型的回答通过飞书发回去。所谓"智能体"，在这个场景里其实就是一套带状态管理的消息处理流程，别被这个词吓到。

三者的数据流是这样的：

2.2 为什么我最终选了长连接而不是 Webhook

这是整个项目里我改动最大的一次决策，值得单独说。

一开始我用的是 Webhook 模式，因为飞书文档里写得很清楚，配个 URL 就行。但实际跑起来问题一堆：本地开发时你得用内网穿透工具把本地端口暴露出去，每次重启地址就变，飞书后台要重新配；生产环境你得有公网 IP 和 HTTPS 证书，还要处理飞书的 URL 验证挑战（challenge）；更麻烦的是，飞书要求你在 3 秒内返回响应，否则会重试，而 RAGFlow 检索加大模型生成动辄十几秒，根本来不及。

长连接模式（飞书官方叫"长连接"或"WebSocket 模式"）把这些全解决了。你的程序主动连飞书，不需要公网地址，不需要证书，消息推过来之后你可以慢慢处理，处理完再调发送接口回复。代价是你要维护一个常驻进程和断线重连逻辑，但这个成本比搞公网暴露低多了。

注意：长连接模式下，飞书推送的是事件，你回复消息是另外调一次发送 API，不是直接在事件响应里返回。这个和 Webhook 的同步回复逻辑完全不同，代码结构要按异步来设计。

2.3 环境准备：Python 版本、依赖和 RAGFlow 部署方式

Python 我用的 3.10，这个版本在依赖兼容性上最省心。3.12 有些库的 wheel 还没跟上，3.8 又太老，很多新库不支持。装依赖之前先建虚拟环境，别污染系统 Python：

python -m venv venv # Windows venv\Scripts\activate # Linux / macOS source venv/bin/activate

核心依赖就几个：

pip install lark-oapi requests openai python-dotenv

• lark-oapi是飞书官方 Python SDK，长连接、事件处理、消息发送都靠它。
• requests用来调 RAGFlow 的 HTTP API。
• openai用来调大模型，如果你用的是兼容 OpenAI 协议的模型服务，这个库直接能用。
• python-dotenv管理密钥，别把 token 硬编码在代码里。

RAGFlow 的部署我用的是 Docker Compose，官方仓库里有现成的docker-compose.yml。这里有个坑：RAGFlow 对内存要求不低，官方建议至少 16GB，我实测 8GB 能跑起来但解析大 PDF 时会 OOM。如果你在 Windows 11 上用 Docker Desktop，记得在设置里把 WSL2 的内存上限调高，默认它可能只给一半。

git clone https://github.com/infiniflow/ragflow.git cd ragflow/docker docker compose -f docker-compose.yml up -d

起来之后访问http://localhost:9380，默认账号在日志里能找到。进去第一件事是建知识库、传文档、等解析完成。解析状态在文档列表里能看到，从"未解析"到"解析完成"这个过程，小文件几秒，几百页的 PDF 可能要几分钟。

3. RAGFlow 侧：知识库建好只是开始，检索参数才是关键

3.1 建知识库和传文档时最容易忽略的设置

在 RAGFlow 界面里点"新建知识库"，会让你选嵌入模型（embedding model）和解析方法。嵌入模型决定了文本怎么变成向量，这个选错了后面检索质量直接崩。我建议中文场景优先选bge-large-zh系列，它对中文语义的捕捉明显好于通用多语言模型。

解析方法（chunk method）有几个选项：General、Q&A、Resume、Manual 等。制度文档、手册这类结构化文本用 General 就行；如果是问答对形式的 FAQ，用 Q&A 能让切块更精准。切块大小（chunk size）默认 512 token，我一般调到 256 到 384 之间，因为制度条文往往一句话就是一个完整语义单元，切太大反而把不相关的内容混进同一个块里。

传文档的时候有个细节：RAGFlow 支持批量上传，但解析是排队进行的。如果你一次传几十个文件，前面几个解析完了后面还在排队，这时候去检索是搜不到后面那些文档的。我踩过这个坑，以为文档传上去就能用，结果检索一直返回空，查了半天才发现是解析没完成。

3.2 用 API 检索而不是靠界面测试

界面上的检索测试只能验证"能不能搜到"，真正接入程序必须走 API。RAGFlow 的检索接口是POST /api/v1/retrieval，请求体大概长这样：

import requests def retrieve(question, dataset_ids, top_k=5): url = "http://localhost:9380/api/v1/retrieval" headers = { "Authorization": f"Bearer {RAGFLOW_API_KEY}", "Content-Type": "application/json" } payload = { "question": question, "dataset_ids": dataset_ids, "top_k": top_k, "similarity_threshold": 0.2, "vector_similarity_weight": 0.3 } resp = requests.post(url, json=payload, headers=headers, timeout=30) return resp.json()

这里有几个参数值得掰开说。similarity_threshold是相似度阈值，低于这个分数的块会被过滤掉。设太高会漏掉相关内容，设太低会引入噪音。我一般从 0.2 开始调，根据实际召回效果微调。vector_similarity_weight控制向量检索和关键词检索的权重，0.3 意味着更偏向关键词匹配。制度类文档里专有名词多，关键词匹配往往比纯语义更准，所以这个值我压得比较低。

返回结构里，data.chunks是检索到的文本块列表，每个块有content、similarity、document_keyword等字段。document_keyword是来源文档名，这个一定要留着，回复的时候带上出处，用户才信得过。

3.3 检索结果怎么拼成给大模型的 prompt

检索回来的是一堆碎片，直接丢给大模型它会懵。我用的模板是这样的：

def build_prompt(question, chunks): context = "\n\n".join([ f"[来源：{c['document_keyword']}]\n{c['content']}" for c in chunks ]) return f"""你是一个内部知识库助手。请严格根据下面提供的资料回答问题。 如果资料中没有相关信息，直接说"资料中未找到相关内容"，不要编造。 资料： {context} 问题：{question} 回答要求： 1. 用简洁的中文回答 2. 如果引用了资料，在句末标注来源文档名 3. 不要输出与问题无关的内容 """

这个 prompt 里有三个关键约束：限定只用资料回答（防止大模型自由发挥）、要求标注来源（可追溯）、明确拒答话术（避免幻觉）。实测下来，加了"资料中未找到相关内容"这句之后，模型胡编的概率明显下降。

4. 飞书侧：长连接接入的完整流程和签名那些事

4.1 创建应用、开权限、拿凭证

去飞书开放平台建一个企业自建应用。建完之后要做几件事：

第一，在"凭证与基础信息"里拿到App ID和App Secret，这两个是程序的身份证明。

第二，在"权限管理"里开通需要的权限。最少要开这几个：im:message（收发消息）、im:message.group_at_msg（接收群里 @ 机器人的消息）、im:message.p2p_msg（接收单聊消息）。权限开完要发布版本，不发布不生效，这个坑我踩过，代码没问题但一直收不到消息，就是权限没发布。

第三，在"事件与回调"里配置事件订阅方式，选"使用长连接接收事件"。然后添加事件：im.message.receive_v1。这个事件就是用户发消息时触发的。

4.2 长连接的建立和事件分发

lark-oapi这个 SDK 把长连接的复杂度封装得不错。核心代码大概是这样：

import lark_oapi as lark from lark_oapi.api.im.v1 import * def do_message_receive(data: P2ImMessageReceiveV1) -> None: # 这里处理收到的消息 handle_message(data) event_handler = lark.EventDispatcherHandler.builder("", "") \ .register_p2_im_message_receive_v1(do_message_receive) \ .build() cli = lark.ws.Client( APP_ID, APP_SECRET, event_handler=event_handler, log_level=lark.LogLevel.INFO ) cli.start()

cli.start()是阻塞的，它会一直维持长连接。SDK 内部有重连机制，网络抖动断了会自动重连，这点比自己手写 WebSocket 省心很多。

事件回调里拿到的data结构比较深，消息内容在data.event.message.content里，是个 JSON 字符串，文本消息的话解析出来是{"text": "用户输入的内容"}。注意群里 @ 机器人的消息，文本里会带一个@_user_1这样的占位符，要把它去掉再送去检索，否则会污染查询。

4.3 消息去重和异步处理

飞书的事件推送有重试机制，如果它没收到你的确认，会重复推同一条消息。虽然长连接模式下重试概率低，但还是要做去重。我用的是message_id做幂等，处理过的 ID 存一个集合，重复的直接跳过。

另一个必须做的是异步处理。事件回调函数如果阻塞太久，会影响长连接的心跳。我的做法是收到消息后立刻丢进一个队列，用单独的线程池去处理检索和生成，回调函数本身秒回。

from concurrent.futures import ThreadPoolExecutor import threading executor = ThreadPoolExecutor(max_workers=4) processed_ids = set() lock = threading.Lock() def do_message_receive(data): msg_id = data.event.message.message_id with lock: if msg_id in processed_ids: return processed_ids.add(msg_id) executor.submit(handle_message, data)

提示：processed_ids这个集合会一直增长，长期运行要加个清理策略，比如只保留最近一小时的 ID，或者用带过期时间的缓存。

5. 智能体核心逻辑：从收到消息到发出回答

5.1 消息解析：把飞书事件变成干净的查询语句

飞书推过来的消息不是纯文本，得先洗干净。文本消息的 content 是 JSON 字符串，解析后取text字段。群里 @ 机器人的消息，文本里会有@_user_1这种标记，用正则去掉。还要处理换行、多余空格。

import json import re def extract_query(message): content = json.loads(message.content) text = content.get("text", "") # 去掉 @ 占位符 text = re.sub(r"@_user_\d+", "", text).strip() return text

如果用户发的是富文本（post 类型）或者图片，content 结构又不一样。我目前只处理文本，其他类型统一回复"暂时只支持文字提问"。这个取舍是有意的，先把主流程跑稳，别一上来就追求全类型支持。

5.2 检索、生成、回复的完整链路

处理函数的主干逻辑：

def handle_message(data): message = data.event.message chat_id = message.chat_id query = extract_query(message) if not query: return # 1. 检索 chunks = retrieve(query, DATASET_IDS) if not chunks: reply_text(chat_id, "资料中未找到相关内容，请换个说法试试。") return # 2. 生成 prompt = build_prompt(query, chunks) answer = call_llm(prompt) # 3. 回复 reply_text(chat_id, answer)

reply_text调的是飞书的发送消息接口：

def reply_text(chat_id, text): client = lark.Client.builder() \ .app_id(APP_ID).app_secret(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() client.im.v1.message.create(request)

注意content必须是 JSON 字符串，不是 Python dict，这个类型错误很隐蔽，报错信息也不直观。

5.3 大模型调用：流式还是非流式

飞书的消息发送接口不支持流式输出，你只能一次性发完整消息。所以大模型这边用非流式就行，等生成完再发。但这里有个体验问题：RAGFlow 检索加大模型生成，用户可能要等十几秒，期间没有任何反馈。

我的做法是先发一条"正在查询..."的提示消息，拿到答案后再更新这条消息（飞书支持更新消息卡片）。或者简单点，直接发一条提示，答案生成后再发一条新消息。前者体验更好但代码复杂，后者简单粗暴但会刷屏。我选了前者，用消息卡片的更新接口实现。

# 先发占位消息，拿到 message_id placeholder_id = send_placeholder(chat_id, "正在查询知识库...") # 生成完成后更新 update_message(placeholder_id, answer)

6. 联调阶段踩过的坑和排查思路

6.1 消息收不到：从权限到事件订阅逐层排查

第一次跑的时候，程序日志显示长连接建立成功，但发消息完全没反应。排查顺序是这样的：

先看飞书开放平台的事件订阅里，im.message.receive_v1有没有加上。我一开始只配了长连接方式，忘了加具体事件，等于开了门但没告诉飞书往哪推。

再看权限有没有发布。权限管理里勾了权限，但没点"创建版本并发布"，权限是不生效的。这个最坑，因为界面上权限显示是"已开通"，但实际没发布。

最后看机器人有没有被拉进群。单聊要用户主动发起，群聊要把机器人加进群，而且群里必须 @ 机器人才会触发事件（除非你开了接收群内所有消息的权限，但那个权限审核很严）。

6.2 RAGFlow 检索返回空：解析状态和阈值双重检查

检索一直返回空列表，我查了两个地方。一是文档解析状态，在 RAGFlow 界面看文档列表，如果还是"解析中"，那检索不到是正常的。二是similarity_threshold设太高，我一开始设了 0.5，结果大部分查询都被过滤了。降到 0.2 之后召回正常。

还有个隐蔽问题：dataset_ids传错了。RAGFlow 的知识库 ID 和显示名称是两回事，API 要的是 ID。在知识库列表页的 URL 里能看到 ID，或者调列表接口拿。

6.3 长连接频繁断开：心跳和网络环境

长连接跑一段时间就断，日志里能看到重连记录。原因通常是网络环境不稳定，或者中间有设备把空闲连接掐了。lark-oapi内部有心跳机制，但如果你在容器里跑，容器的网络策略可能影响长连接。

我的解决办法是把重连日志级别调高，观察断开频率。如果几分钟断一次，基本是网络问题；如果几小时断一次，属于正常波动，SDK 会自动重连，不用管。另外确保程序不要被系统的休眠策略影响，服务器上跑的话关掉休眠。

6.4 回答质量差：prompt 和检索参数要一起调

一开始回答经常答非所问，我以为是模型不行，换了个更大的模型还是不行。后来发现是检索环节的问题：召回的内容本身就不相关，模型再强也白搭。

调优的顺序应该是先调检索，再调 prompt。检索这边调top_k（召回数量）、similarity_threshold（阈值）、vector_similarity_weight（权重）。prompt 这边主要是把约束写清楚，尤其是拒答逻辑。我最终的组合是top_k=5、threshold=0.2、weight=0.3，配合前面那个带来源标注的 prompt，回答准确率明显上来了。

7. 上线前值得再花时间做的几件事

跑通不等于能用。我在正式给团队用之前，又补了几个东西。

加日志和可观测性。每条消息的 query、召回的 chunk、最终回答都记下来，出问题能回溯。我用的是简单的文件日志，按天切分。这个在调优阶段价值极大，你能看到哪些问题回答得好、哪些答得差，针对性改进。

处理超长消息。飞书单条消息有长度限制，大模型有时候会生成很长的回答。超过限制要截断或者分多条发。我设了个阈值，超过就截断并提示"回答过长已截断"。

加个简单的限流。防止有人刷机器人把大模型额度跑光。按用户维度做频率限制，比如每分钟最多 5 次查询。

知识库更新流程。文档会更新，RAGFlow 里的知识库也要跟着更新。我写了个脚本，定期扫描指定目录，有新文件就调 RAGFlow 的上传接口，有修改就重新解析。这块目前还是半自动，但比手动传强多了。

最后分享一个我在实际使用中体会最深的点：这套系统的瓶颈从来不是模型，而是知识库的质量。文档切块切得烂、过期文档没清理、格式混乱，再强的模型也救不回来。花在整理文档和调切块参数上的时间，回报远高于换模型。我现在的做法是，每上线一批新文档，先拿十几个典型问题测一遍召回，召回不准就回去调切块，反复几轮再放给用户用。