不修改 RAGFlow 前端，也不把 RAGFlow 的知识库重新导入 Open WebUI。两边通过稳定接口连接，各自继续做擅长的事情。
把 Open WebUI 接到 RAGFlow：内部知识问答平台的开发实践
这一篇进入真正的集成开发：Open WebUI 继续负责用户、会话和交互，RAGFlow 专注文档检索与引用，中间通过 Pipe Function 完成模型注册、请求转换和流式响应适配。
前面三篇分别把 RAGFlow 和 Open WebUI 跑了起来。这一篇我会从开发角度把它们接到一起，让员工在 Open WebUI 中选择“内部专业知识库”，再由 RAGFlow 完成检索、生成与引用。
到这里，两套系统各自负责什么已经比较清楚：
- RAGFlow 擅长处理文档、检索知识和返回引用；
- Open WebUI 擅长提供用户真正使用的聊天入口；
- 大模型负责理解问题和组织答案。
这一篇不再分别介绍功能，而是沿着正常开发链路完成模型注册、流式回答和引用展示。
先把组件边界固定下来
两个系统都有聊天页面，也都有一定的知识库能力。如果一开始不确定所有权，后面很容易出现两份会话、两次检索和两套权限。
我先做了下面这张职责表：
| 能力 | 负责组件 | 原因 |
|---|---|---|
| 用户登录、群组 | Open WebUI | 所有 AI 能力共用同一入口 |
| 会话列表、聊天记录 | Open WebUI | 用户侧只保留一份会话 |
| 模型选择与可见范围 | Open WebUI | 普通模型和知识助手统一管理 |
| 文档上传、解析、Chunk | RAGFlow | 保留复杂文档处理能力 |
| 检索、Rerank | RAGFlow | 参数和效果集中维护 |
| 知识助手 Prompt | RAGFlow | 与 Dataset 和检索策略一起管理 |
| 协议转换、引用映射 | Pipe Function | 隔离两边接口细节 |
这里最关键的决定是：对于这个“专业知识库模型”，Open WebUI 只负责会话和呈现，不再执行自己的 Knowledge RAG。
为什么我选择 Pipe Function
RAGFlow 已经提供 OpenAI-compatible Chat Completion API，最短路径当然是把它当作一个 OpenAI Connection 配进 Open WebUI。
这种方式适合先验证链路。但实际开发时，我还希望控制：
- Open WebUI 中显示的模型名称；
- 模型和 RAGFlow
chat_id的映射； - API Key 不暴露给前端；
- RAGFlow
reference到 Open WebUI Citation 的转换； - 请求日志、超时和统一错误返回；
- 后续按用户或群组选择不同知识助手。
因此我在两者之间加了一层很薄的 Pipe Function。
Pipe Function，不是旧 Pipelines
Open WebUI 当前已经把独立 Pipelines 标记为 legacy。新的接入更适合使用直接运行在 Open WebUI 中的 Pipe Function。
Pipe 会在模型列表中注册成一个可选模型，并接管这次请求的完整处理过程。
在 RAGFlow 准备一个可调用的助手
接入前，我先在 RAGFlow 中把知识助手本身调通：
步骤 1：准备 Dataset
文档已经解析完成，固定测试问题在 Retrieval Test 中能命中正确 Chunk。
步骤 2：创建 Chat Assistant
关联 Dataset、选择 Chat Model，配置回答规则、拒答文案和引用。
步骤 3：获取 API Key
API Key 只会保存在 Open WebUI 的管理员配置中，不写进浏览器代码，也不提交到 Git。
步骤 4：记录
chat_idRAGFlow 的 OpenAI-compatible 地址中包含 Chat Assistant ID：
POST /api/v1/openai/<chat_id>/chat/completions步骤 5：先独立验证接口
分别测试非流式、流式、多轮消息和
reference返回。这样进入 Open WebUI 联调时，我已经知道 RAGFlow 后端是正常的。
设计一个稳定的模型标识
Open WebUI 展示给员工的是业务名称，RAGFlow 使用的是内部 chat_id，两者不应该混成同一个字段。
models:
- id: internal-product-kb
name: 内部产品知识库
ragflow_chat_id: 8fxxxxxxxxxxxxxxxx
- id: internal-process-kb
name: 内部流程知识库
ragflow_chat_id: 51xxxxxxxxxxxxxxxx员工只会看到“内部产品知识库”，而 Pipe 在服务端完成 model id -> chat_id 映射。以后 RAGFlow 重建 Assistant，只需要调整映射，不必改变用户已经熟悉的模型名称。
第一版只有一个知识助手时，也可以先使用一个 Valve 保存 RAGFLOW_CHAT_ID，保持代码简单。
实现一个最小 Pipe Function
下面这段代码保留了组合开发中的主干：
- 使用 Valves 保存地址、Key 和
chat_id； - 使用异步
httpx，不阻塞 Open WebUI 事件循环； - 支持流式和非流式调用；
- 从 RAGFlow 最后的数据块中读取
reference； - 通过
sourceEvent 把资料来源交给 Open WebUI。
"""
title: RAGFlow Knowledge
author: Mars Chin
version: 0.1.0
requirements: httpx
"""
import json
from typing import Any
import httpx
from pydantic import BaseModel, Field
class Pipe:
class Valves(BaseModel):
RAGFLOW_BASE_URL: str = Field(
default="http://ragflow",
description="RAGFlow 地址，不包含末尾斜杠",
)
RAGFLOW_API_KEY: str = Field(
default="",
description="RAGFlow API Key",
)
RAGFLOW_CHAT_ID: str = Field(
default="",
description="RAGFlow Chat Assistant ID",
)
REQUEST_TIMEOUT_SECONDS: int = Field(
default=180,
description="读取完整回答的超时时间",
)
def __init__(self):
self.valves = self.Valves()
def pipes(self):
return [
{
"id": "internal-product-kb",
"name": "内部产品知识库",
}
]
def _endpoint(self) -> str:
base = self.valves.RAGFLOW_BASE_URL.rstrip("/")
chat_id = self.valves.RAGFLOW_CHAT_ID
return f"{base}/api/v1/openai/{chat_id}/chat/completions"
def _headers(self) -> dict[str, str]:
return {
"Authorization": f"Bearer {self.valves.RAGFLOW_API_KEY}",
"Content-Type": "application/json",
}
def _payload(self, body: dict, stream: bool) -> dict:
return {
# model 使用 RAGFlow Chat Assistant 中配置好的模型
"model": "model",
"messages": body.get("messages", []),
"stream": stream,
"reference": True,
"reference_metadata": {
"include": True,
"fields": ["source", "version"],
},
}
async def _emit_references(self, references, event_emitter):
if not event_emitter:
return
for reference in references or []:
document_name = reference.get("document_name", "RAGFlow 文档")
document_id = reference.get("document_id", document_name)
content = reference.get("content", "")
await event_emitter(
{
"type": "source",
"data": {
"source": {
"name": document_name,
"id": document_id,
},
"document": [content],
"metadata": [
{
"source": document_name,
"name": document_name,
"document_id": document_id,
"position": reference.get("position"),
"similarity": reference.get("similarity"),
}
],
},
}
)
async def _stream(self, payload: dict, event_emitter):
timeout = httpx.Timeout(
connect=10,
read=self.valves.REQUEST_TIMEOUT_SECONDS,
write=30,
pool=10,
)
async with httpx.AsyncClient(timeout=timeout) as client:
async with client.stream(
"POST",
self._endpoint(),
headers=self._headers(),
json=payload,
) as response:
response.raise_for_status()
async for line in response.aiter_lines():
if not line.startswith("data:"):
continue
raw = line.removeprefix("data:").strip()
if not raw or raw == "[DONE]":
continue
chunk = json.loads(raw)
delta = chunk["choices"][0].get("delta", {})
if delta.get("content"):
yield delta["content"]
if delta.get("reference"):
await self._emit_references(
delta["reference"],
event_emitter,
)
async def _complete(self, payload: dict, event_emitter) -> str:
async with httpx.AsyncClient(
timeout=self.valves.REQUEST_TIMEOUT_SECONDS
) as client:
response = await client.post(
self._endpoint(),
headers=self._headers(),
json=payload,
)
response.raise_for_status()
data = response.json()
message = data["choices"][0]["message"]
await self._emit_references(
message.get("reference"),
event_emitter,
)
return message.get("content", "")
async def pipe(
self,
body: dict,
__user__: dict | None = None,
__event_emitter__=None,
) -> Any:
stream = body.get("stream", True)
payload = self._payload(body, stream)
if stream:
return self._stream(payload, __event_emitter__)
return await self._complete(payload, __event_emitter__)代码示例的定位
这是一条清楚的最小主线，不是可以原样上线的完整网关。正式使用时，我还会补充结构化日志、Trace ID、重试边界、用户权限映射和对上游错误的统一转换。
Open WebUI Function 会在服务器中执行 Python，导入任何 Function 前都应该先审查代码。
为什么流式文本和引用分开处理
RAGFlow 的流式响应会持续返回 delta.content，引用通常出现在最后的数据块中。文本应该边收到边 yield，引用则通过 Open WebUI 的 source 或 citation Event 单独提交。
这样做比把引用 JSON 直接拼到回答末尾更自然。员工看到的是正常答案，点击引用后再查看命中的资料片段。
会话历史只保留一份
Open WebUI 会把当前多轮对话放进 messages。RAGFlow 的 OpenAI-compatible 接口也接受历史消息，因此 Pipe 可以直接传递经过筛选的 messages。
我没有再调用 RAGFlow 的 Session API 创建第二份用户会话，因为那会形成两套状态：
对于很长的会话，后续可以在 Pipe 中做消息裁剪或摘要，但仍然由 Open WebUI 作为会话事实来源。
Prompt 也需要分层
组合以后至少存在三层上下文：
- Open WebUI 模型层的基础说明；
- RAGFlow Chat Assistant 的角色与回答约束；
- RAGFlow 检索得到的知识片段。
我会把“必须根据知识回答、无依据时拒答、怎样引用”等规则放在 RAGFlow Assistant，因为它们和 Dataset、检索策略绑定。Open WebUI 只保留面向用户的简短说明，避免两边重复注入同一段系统 Prompt。
我会避免在 Open WebUI 再做一次 RAG
对于这个 Pipe 模型，我不会绑定 Open WebUI Knowledge，也会关闭不需要的 File Context 能力。
如果 Open WebUI 先检索一次，把内容注入消息，RAGFlow 收到消息后又根据 Dataset 检索一次，可能造成上下文重复、来源混乱和额外 Token 消耗。
临时上传文件是另一类需求，可以单独设计，不能默认混进专业知识库链路。
用户权限怎样落到知识助手
第一版可以通过 Open WebUI 控制模型可见范围：
这能解决“谁能看到哪个知识助手”。如果同一个 Assistant 内部还要按用户动态过滤文档，就需要让 Pipe 把可信用户信息传给后端，由适配服务转成 RAGFlow Metadata Filter 或独立 Dataset 范围。
我不会把权限判断建立在前端传来的任意字段上。用户身份应该来自 Open WebUI 已验证的 __user__，并在服务端完成映射。
部署时保持内部调用链
最终的内部部署拓扑可以保持很清楚：
RAGFlow API 和模型网关不需要直接暴露给员工浏览器。Open WebUI 是统一入口，Pipe 在服务器内部持有 RAGFlow API Key。
我会怎样验收这条开发链路
开发完成后，我不会只验证“能返回一句话”，而是按正常业务路径走一遍：
| 场景 | 预期 |
|---|---|
| 单轮知识问题 | 回答与 RAGFlow 页面结果一致 |
| 多轮追问 | 能利用 Open WebUI 传入的历史消息 |
| 无答案问题 | 使用 RAGFlow 的拒答策略 |
| 引用 | 能看到文档名和对应片段 |
| 流式响应 | 首段内容及时出现，不等待整段结束 |
| 模型切换 | 普通模型和知识模型互不影响 |
| 权限 | 非目标群组看不到知识模型 |
| 会话恢复 | 刷新页面后历史与引用仍然存在 |
到这里，两套系统的正常组合路径就完成了。真正花时间的部分，往往是接口只兼容了一半、容器网络地址写错、代理缓冲流式响应、引用字段对不上这些细节。
下一篇我会把这些问题单独整理成一份可检索的排障记录，不再打断这一篇的开发主线。