第16章：【基础篇综合实战】RAGFlow 搭建企业知识库问答机器人
·
1 项目背景
业务场景
「云帆科技」经过前 15 章的学习和实践，团队已经掌握了 RAGFlow 从部署、数据建模、文档解析、检索配置到模型接入的全部基础能力。CTO 在月度会议上正式宣布：接下来两周的目标，是为集团 500 名员工搭建一套完整的企业知识库问答机器人，覆盖 HR 制度、财务报销、行政服务三个核心部门，并对全公司开放试用。
这不是一次"实验"，而是一次"交付"。机器人需要满足以下可量化的业务要求：
- 知识覆盖：三个部门共约 50 份制度文档（PDF/Word/Markdown），全量导入并解析成功。
- 检索质量：50 个标准评测问题中，引用准确率 >= 85%，P95 响应时间 < 8 秒。
- 多助手隔离：HR、财务、行政部门各建独立 Chat 助手，Prompt 和模型按需定制，数据互相隔离。
- 用户权限：部门数据按需授权，机密数据集（如薪资）仅限授权人员访问。
- 运维兜底：部署文档、排障 SOP、监控脚本和报警规则一应俱全，不是"离职即报废"的项目。
痛点
在从"个人实验"到"团队交付"的跨越中，团队面临一系列工程化挑战：
- 零散知识 vs 系统方案：前 15 章学到的知识是零散的——部署、数据模型、解析、检索、Chat、API——如何把它们串成一套完整方案？
- 个人实验 vs 团队交付：个人摸索可以用默认配置、手动操作；但团队交付需要文档、SOP、权限、监控。
- 能跑通 vs 能验收：能搜到答案是第一步；业务方要的是"85% 准确率"的量化承诺和"出问题 5 分钟能恢复"的可靠性。
- 一次性搭建 vs 持续迭代：文档会更新、模型会升级、用户反馈会回来——系统需要有可持续优化的能力。
从实验到交付的跨越：
前15章状态： 本章交付状态：
Docker 单机部署 ✓ → 生产级部署 + 资源规划
数据模型了解 ✓ → 多部门数据架构设计
解析配置会调 ✓ → 全量50份文档分批导入SOP
Chat 助手会创建 ✓ → 三部门独立助手 + 分级Prompt
检索原理懂 ✓ → 50题评测 + 85%准确率达标
API 能用 ✓ → 企业OA集成方案
排障知道思路 ✓ → 完整排障SOP文档
2 项目设计
小胖：（抱着一堆打印的资料冲进会议室）“大师！CTO 给的任务也太重了——两周要搭建三部门的知识库问答机器人，还要 85% 的准确率！前 15 章我学是学了，但都是单独的知识点，现在要串成一个完整的项目，不知道从哪开始下手啊！”
大师：（在桌上摊开一张大白纸）“别慌。任何工程项目的起点都是方案设计。你现在需要的是一个’项目蓝图’——从需求到交付的完整路线图。我们花 15 分钟画出来，剩下的时间去执行。”
小胖：“那这个蓝图应该包含什么？”
大师：“一张完整的 RAG 项目蓝图有六个模块，缺一个都会导致交付延期或失败：”
RAG 项目蓝图六大模块：
模块1：需求与数据集设计
├── 业务需求调研（哪些文档、多少用户、什么场景）
├── 数据分类（按部门/按密级/按格式）
└── 数据集架构（KB→Dataset→Document 的规划）
模块2：部署与基础设施
├── 服务器资源评估（CPU/内存/磁盘/GPU）
├── Docker Compose 部署与参数调优
└── 模型供应商接入（LLM/Embedding/Rerank）
模块3：文档导入与解析
├── 文档预检（格式、大小、质量、敏感信息）
├── 分批导入策略（先简单后复杂）
└── 解析质量验收（抽查切片、修正低质量切片）
模块4：检索与问答配置
├── 多助手创建（HR/财务/行政独立助手）
├── Prompt 设计（按部门定制角色和规则）
└── 检索参数调优（TopK/阈值/权重/Rerank）
模块5：评测与验证
├── 评测问题集设计（50题，四类难度）
├── 自动评测脚本（打分+报告）
└── 人工抽检验证（10%问题交叉验证）
模块6：交付与运维
├── 部署文档（架构图+环境说明+启动步骤）
├── 排障 SOP（5类常见故障+排查步骤）
├── 用户培训（操作手册+常见问题FAQ）
└── 监控告警（队列积压/LLM错误率/响应延迟）
技术映射：项目蓝图 = 建筑施工图——图纸上的每一个标注都要对应到实际材料(代码)、施工步骤(配置)和验收标准(评测)。
小白：（飞快地记着笔记）“那这六个模块的执行顺序是什么？哪些可以并行？哪些必须串行？”
大师：“标准执行顺序如下，其中标注 [P] 的可以并行：”
Week 1:
Day 1-2: 模块1 需求调研 + 数据分类 [P]
模块2 服务器部署 + 模型接入 [P]
Day 2-4: 模块3 文档分批导入 + 解析验收
Day 4-5: 模块4 多助手创建 + Prompt 调试
Week 2:
Day 6-8: 模块5 评测问题集设计 + 自动评测运行
Day 8-9: 模块5 人工抽检 + 问题修复（低分问题根因分析）
Day 9-10: 模块4 检索参数调优（基于评测反馈迭代）
Day 11-12: 模块6 文档编写 + 用户培训准备
Day 13-14: 模块6 全量验收 + 监控上线 + 正式交付
小胖：“模块 1 的需求调研和数据分类，具体怎么做？我们公司有三个部门、50 份文档，数据该放几个数据集？”
大师：“这是整个项目的基础。画一张’数据架构图’：”
云帆科技企业知识库（Knowledge Base）
├── 数据集 A: HR-薪酬制度 [权限: HR经理组]
│ ├── 2024薪酬管理办法.pdf
│ ├── 绩效奖金发放细则.docx
│ ├── 社保公积金缴纳标准.xlsx
│ └── 加班费计算规则.md
│
├── 数据集 B: HR-考勤制度 [权限: HR全员组]
│ ├── 考勤管理办法2024.pdf
│ ├── 请假与调休规定.docx
│ └── 远程办公管理办法.md
│
├── 数据集 C: HR-员工关系 [权限: HR全员组]
│ ├── 劳动合同管理规定.pdf
│ ├── 离职管理办法.docx
│ └── 招聘管理办法.pdf
│
├── 数据集 D: 财务-报销制度 [权限: 财务+全员]
│ ├── 差旅费管理办法.pdf
│ ├── 招待费管理办法.docx
│ └── 报销流程与标准.md
│
├── 数据集 E: 财务-预算制度 [权限: 财务经理组]
│ ├── 年度预算管理办法.pdf
│ └── 部门预算审批流程.docx
│
├── 数据集 F: 行政-办公服务 [权限: 全员]
│ ├── 办公用品申领流程.md
│ ├── 会议室使用规则.pdf
│ ├── 快递收发管理办法.docx
│ └── 设备报修流程.md
│
└── 数据集 G: 行政-环境安全 [权限: 全员]
├── 办公室安全管理规定.pdf
└── 消防安全手册.pdf
“关键决策点：Excel 中的薪资标准表为什么和 PDF 薪酬制度放一起，而不是单独建数据集？——因为用户问’P3 职级薪资多少’时，需要在同一数据集内同时检索制度描述和薪资数据，分开数据集会导致跨库检索不准。”
小白：“那模块 2 的部署资源怎么估算？我们公司人不多，一台 16GB 服务器够吗？”
大师：“够，但要算清楚每个组件吃多少资源：”
| 组件 | 内存占用 | 说明 |
|---|---|---|
| MySQL | 1-2GB | 元数据存储 |
| Redis | 0.5-1GB | 任务队列 + 缓存 |
| MinIO | 0.5GB | 对象存储 |
| Elasticsearch | 2-4GB | 文档引擎（用 Infinity 可省 2GB） |
| API Server | 1-2GB | 在线请求处理 |
| Task Executor | 2-4GB | 解析 + Embedding（本地模型加倍） |
| Ollama(本地 LLM) | 6-8GB | 可选，如果用云端模型则不需要 |
| 总计 | 13-22GB | 16GB 勉强够，建议 32GB |
“如果只用云端模型（不走 Ollama），16GB 刚好够。但 Task Executor 处理大量扫描件时可能会触发 OOM——建议配置 mem_limit: 4g 限制单容器内存。”
小胖：“模块 3 的文档导入——50 份文档怎么分批？一股脑全丢进去行不行？”
大师：“绝对不能！50 份文档一股脑丢进去会触发三个问题：Redis 队列积压、Task Executor OOM、共享 Embedding API 被限流。分批策略如下：”
文档分批导入 SOP：
第1批（快速验证，5份）：
- 选最简单的 Markdown 文件
- 验证基本解析流程正常
- 确认 Embedding 模型可用
第2批（核心文档，10份）：
- 部门核心制度 PDF（文本型，非扫描件）
- 每一份验证切片质量（手动抽查 3-5 个切片）
- 配置 Chat 助手后用标准问题测试
第3批（复杂文档，15份）：
- 含表格的 DOCX/XLSX 文件
- 开启 table_enhance，验证表格解析
- 对比开启/关闭 Rerank 的检索效果
第4批（扫描件，10份）：
- 扫描合同、旧版文档
- 开启 OCR，注意耗时会大幅增加
- OCR 后逐份检查低置信度切片
第5批（收尾，10份）：
- 剩余零散文档
- 全量跑一遍评测脚本
- 针对低分问题反向修正切片
小白：“模块 4 的三个助手——HR、财务、行政——Prompt 应该有什么不同？”
大师：“核心差异：”
| 助手 | Prompt 核心特点 | 温度 | 引用要求 | 拒答策略 |
|---|---|---|---|---|
| HR 助手 | 严谨、精确引用条款编号、禁止推测 | 0.0 | 强制每条引用 | 涉薪资数据需授权 |
| 财务助手 | 精确数字、表格化输出、展示计算步骤 | 0.0 | 强制数字来源引用 | 禁止批准/拒绝报销 |
| 行政助手 | 简洁实用、步骤化输出、直接给联系方式 | 0.3 | 可选引用 | 无额外限制 |
技术映射：助手定制 = 三个不同的客服——HR 像律师（严谨），财务像会计（精确），行政像前台（热情简洁）。
3 项目实战
环境准备
目标：确认生产环境满足全部前置条件。
硬件检查清单：
# 服务器资源检查脚本
#!/bin/bash
echo "=== RAGFlow 部署环境检查 ==="
# CPU 核心数
CPU_CORES=$(nproc)
echo "CPU: ${CPU_CORES} 核 (建议 >= 4)"
# 内存
TOTAL_MEM=$(free -h | awk '/^Mem:/ {print $2}')
echo "内存: $TOTAL_MEM (建议 >= 16GB)"
# 磁盘可用空间
DISK_AVAIL=$(df -h / | awk 'NR==2 {print $4}')
echo "磁盘可用: $DISK_AVAIL (建议 >= 50GB)"
# Docker 版本
DOCKER_VER=$(docker --version 2>/dev/null || echo "未安装")
echo "Docker: $DOCKER_VER"
# Docker Compose 版本
COMPOSE_VER=$(docker compose version 2>/dev/null || echo "未安装")
echo "Compose: $COMPOSE_VER"
# vm.max_map_count (ES 需要)
MMC=$(sysctl -n vm.max_map_count 2>/dev/null)
echo "vm.max_map_count: $MMC (建议 >= 262144)"
# 端口检查
for port in 80 3306 6379 9000 9200 9380; do
if ss -tlnp | grep -q ":$port "; then
echo "[WARN] 端口 $port 已被占用"
else
echo "[OK] 端口 $port 空闲"
fi
done
部署步骤：
# 1. 克隆代码
git clone https://github.com/infiniflow/ragflow.git
cd ragflow/docker
# 2. 调整系统参数
sudo sysctl -w vm.max_map_count=262144
echo "vm.max_map_count=262144" | sudo tee -a /etc/sysctl.conf
# 3. 配置 .env
cat > .env << 'EOF'
HTTP_PORT=80
MYSQL_PASSWORD=Ragflow@2024!ProdSecure
MINIO_USER=ragflow_admin
MINIO_PASSWORD=Ragflow@2024!ProdSecure
DOC_ENGINE=infinity
REDIS_PASSWORD=Ragflow@2024!ProdSecure
LOG_LEVEL=INFO
EOF
# 4. 启动基础服务
docker compose -f docker-compose-base.yml up -d
sleep 30 # 等待 MySQL/Redis/ES 就绪
# 5. 启动主服务
docker compose -f docker-compose.yml up -d
sleep 20
# 6. 验证
curl -s http://localhost/api/v1/version | jq .
分步实现
步骤1：数据架构落库——批量创建数据集
目标：按设计的 7 个数据集，用 API 批量创建并记录 ID。
# step1_create_datasets.py
import json
from ragflow import RAGFlow
rag = RAGFlow(api_key="ragflow-xxx", base_url="http://localhost/api/v1")
datasets_config = [
{
"name": "HR-薪酬制度",
"description": "基本工资、绩效奖金、社保公积金、加班费计算",
"embedding_model": "BAAI/bge-large-zh-v1.5@Ollama",
"chunk_method": "title",
"chunk_size": 800,
"permission": "restricted",
"tags": ["HR", "薪酬", "机密"],
},
{
"name": "HR-考勤制度",
"description": "日常考勤、请假调休、远程办公",
"embedding_model": "BAAI/bge-large-zh-v1.5@Ollama",
"chunk_method": "title",
"chunk_size": 800,
"tags": ["HR", "考勤"],
},
{
"name": "HR-员工关系",
"description": "劳动合同、离职管理、招聘办法",
"embedding_model": "BAAI/bge-large-zh-v1.5@Ollama",
"chunk_method": "title",
"chunk_size": 800,
"tags": ["HR", "员工关系"],
},
{
"name": "财务-报销制度",
"description": "差旅费、招待费管理办法及报销流程",
"embedding_model": "BAAI/bge-large-zh-v1.5@Ollama",
"chunk_method": "table",
"chunk_size": 600,
"tags": ["财务", "报销"],
},
{
"name": "财务-预算制度",
"description": "年度预算、部门预算审批流程",
"embedding_model": "BAAI/bge-large-zh-v1.5@Ollama",
"chunk_method": "title",
"chunk_size": 800,
"permission": "restricted",
"tags": ["财务", "预算", "机密"],
},
{
"name": "行政-办公服务",
"description": "办公用品申领、会议室、快递、设备报修",
"embedding_model": "BAAI/bge-large-zh-v1.5@Ollama",
"chunk_method": "title",
"chunk_size": 800,
"tags": ["行政", "办公"],
},
{
"name": "行政-环境安全",
"description": "办公室安全、消防手册",
"embedding_model": "BAAI/bge-large-zh-v1.5@Ollama",
"chunk_method": "title",
"chunk_size": 800,
"tags": ["行政", "安全"],
},
]
created = {}
for cfg in datasets_config:
ds = rag.create_dataset(
name=cfg["name"],
description=cfg["description"],
embedding_model=cfg["embedding_model"],
chunk_method=cfg["chunk_method"],
chunk_size=cfg["chunk_size"],
overlap_size=int(cfg["chunk_size"] * 0.15),
)
created[cfg["name"]] = {"id": ds.id, **cfg}
print(f"[OK] {cfg['name']}: {ds.id}")
# 保存映射文件供后续使用
with open("dataset_mapping.json", "w") as f:
json.dump(created, f, indent=2, ensure_ascii=False)
print(f"\n共创建 {len(created)} 个数据集")
步骤2：文档分批导入与质量验收
目标：按五批策略导入 50 份文档，每批做质量检查。
# step2_batch_import.py
import os
import time
import json
from ragflow import RAGFlow
rag = RAGFlow(api_key="ragflow-xxx", base_url="http://localhost/api/v1")
# 加载数据集映射
with open("dataset_mapping.json", "r") as f:
ds_map = json.load(f)
# 文档分批计划
batches = [
{
"name": "第1批-实测验证",
"files": [
("docs/hr/远程办公管理办法.md", "HR-考勤制度"),
("docs/admin/办公用品申领流程.md", "行政-办公服务"),
("docs/hr/加班费计算规则.md", "HR-薪酬制度"),
("docs/admin/设备报修流程.md", "行政-办公服务"),
("docs/finance/报销流程与标准.md", "财务-报销制度"),
]
},
{
"name": "第2批-核心文档",
"files": [
("docs/hr/考勤管理办法2024.pdf", "HR-考勤制度"),
("docs/hr/2024薪酬管理办法.pdf", "HR-薪酬制度"),
("docs/hr/劳动合同管理规定.pdf", "HR-员工关系"),
("docs/finance/差旅费管理办法.pdf", "财务-报销制度"),
("docs/admin/会议室使用规则.pdf", "行政-办公服务"),
# ... 更多核心文件
]
},
# 第3批~第5批类似定义...
]
all_results = []
for batch in batches:
print(f"\n{'='*60}")
print(f"执行: {batch['name']} ({len(batch['files'])} 个文件)")
print(f"{'='*60}")
for file_path, ds_name in batch["files"]:
if not os.path.exists(file_path):
print(f" [SKIP] 文件不存在: {file_path}")
continue
ds_id = ds_map[ds_name]["id"]
ds = rag.get_dataset(ds_id)
# 上传
print(f" [UPLOAD] {file_path} -> {ds_name}")
doc = ds.upload_document(file_path)
# 等待解析（最多等待 5 分钟）
max_wait = 300
waited = 0
while waited < max_wait:
time.sleep(5)
waited += 5
doc = ds.get_document(doc.id)
if doc.status in ["success", "failed"]:
break
# 质量检查
quality_issue = None
if doc.status == "success":
chunks = ds.list_chunks(doc.id, page_size=5)
for chunk in chunks:
# 检查异常短切片
if chunk.token_count < 30:
quality_issue = f"短切片: {chunk.token_count} tokens"
break
# 检查异常长切片
if chunk.token_count > 2000:
quality_issue = f"长切片: {chunk.token_count} tokens"
break
# 检查 OCR 乱码（中文字符比例 < 10%）
import re
chinese = len(re.findall(r'[\u4e00-\u9fff]', chunk.content))
if chinese < len(chunk.content) * 0.05 and len(chunk.content) > 50:
quality_issue = "疑似OCR乱码"
result = {
"file": file_path,
"dataset": ds_name,
"doc_id": doc.id,
"status": doc.status,
"chunks": getattr(doc, "chunk_count", 0),
"tokens": getattr(doc, "token_count", 0),
"quality_issue": quality_issue,
}
all_results.append(result)
status_icon = "✓" if doc.status == "success" and not quality_issue else "⚠"
print(f" {status_icon} 状态={doc.status}, 切片={result['chunks']}, "
f"质量={'通过' if not quality_issue else quality_issue}")
# 生成导入报告
success_count = sum(1 for r in all_results if r["status"] == "success")
quality_issue_count = sum(1 for r in all_results if r["quality_issue"])
print(f"\n=== 导入完成报告 ===")
print(f"总计: {len(all_results)} 个文件")
print(f"成功: {success_count}")
print(f"失败: {len(all_results) - success_count}")
print(f"质量异常: {quality_issue_count}")
# 保存详细结果
with open("import_report.json", "w") as f:
json.dump(all_results, f, indent=2, ensure_ascii=False)
步骤3：多助手创建与 Prompt 配置
目标：创建三个部门助手，各自绑定对应数据集和定制 Prompt。
# step3_create_assistants.py
from ragflow import RAGFlow
rag = RAGFlow(api_key="ragflow-xxx", base_url="http://localhost/api/v1")
# --- HR 助手 ---
hr_prompt = """你是一个专业的云帆科技HR制度问答助手。
## 身份与职责
你服务于公司全体员工，提供考勤、薪酬、福利、劳动合同、招聘等制度咨询。
## 知识范围
严格基于下方「参考文档」中的制度条款回答问题，禁止使用外部知识或主观推测。
## 回答规则
1. 每条事实陈述后必须使用 [编号] 标注引用来源。
2. 涉及具体数字（天数、金额、比例）时，原样引用，不得四舍五入。
3. 涉及薪资、社保等员工个人信息的查询，回答时标注"该信息以HR系统最新数据为准"。
4. 无法回答时，回复："该问题未在HR制度中找到明确条款，建议联系HR部门：hr@yunfan.com 或分机 1001。"
## 回答格式
- 直接答案（1-3句话概括）
- 制度条款原文引用（带编号）
- 必要时的补充说明或例外情况
- 本次使用的参考文档列表
## 禁止事项
- 禁止为员工个人决策提供建议（如"建议您辞职"、"您应该申请仲裁"）
- 禁止编造或推测制度中不存在的规定
- 禁止回答与HR制度无关的问题
## 参考文档
{context}"""
hr_chat = rag.create_chat(
name="云帆HR制度助手",
description="考勤、薪酬、福利、劳动合同、招聘制度问答",
dataset_ids=[
ds_map["HR-薪酬制度"]["id"],
ds_map["HR-考勤制度"]["id"],
ds_map["HR-员工关系"]["id"],
],
llm_model="qwen2.5:14b@Ollama",
prompt_system=hr_prompt,
temperature=0.0,
retrieval_config={
"top_k": 15,
"similarity_threshold": 0.25,
"rerank_model": "BAAI/bge-reranker-v2-m3@Ollama",
}
)
print(f"[OK] HR助手: {hr_chat.id}")
# --- 财务助手 ---
finance_prompt = """你是一个云帆科技财务报销助手。
## 身份与职责
你负责提供差旅费标准、招待费标准、报销流程、预算制度等信息。
## 知识范围
严格基于下方「参考文档」，禁止使用外部知识。
## 回答规则
1. 差旅费标准、金额类问题，以表格形式展现。
2. 涉及计算的问题（如"出差3天总费用多少"），展示计算步骤。
3. 所有金额精确到元，每项数据标注引用编号 [N]。
4. 无法回答时回复："该问题不在财务知识范围内，请联系财务部：finance@yunfan.com 或分机 2001。"
## 回答格式
- 结论（1句话）
- 数据表格（如果是标准查询）
- 计算过程（如果涉及多步计算）
- 引用来源
## 禁止事项
- 禁止批准或拒绝任何报销申请（你不是审批人）
- 禁止修改制度中的金额标准
- 禁止对个人报销提供操作建议
## 参考文档
{context}"""
finance_chat = rag.create_chat(
name="云帆财务报销助手",
description="差旅标准、招待费、报销流程、预算制度",
dataset_ids=[
ds_map["财务-报销制度"]["id"],
ds_map["财务-预算制度"]["id"],
],
llm_model="qwen2.5:14b@Ollama",
prompt_system=finance_prompt,
temperature=0.0,
retrieval_config={
"top_k": 15,
"similarity_threshold": 0.25,
"rerank_model": "BAAI/bge-reranker-v2-m3@Ollama",
}
)
print(f"[OK] 财务助手: {finance_chat.id}")
# --- 行政助手 ---
admin_prompt = """你是一个云帆科技行政服务助手。
## 身份与职责
你负责解答办公用品申领、会议室预定、设备报修、快递收发、办公室安全等行政问题。
## 回答规则
1. 回答简洁实用，优先给出可直接操作的信息（联系电话、操作步骤、服务时间）。
2. 流程类问题给出不超过5个步骤。
3. 涉及具体负责人的，提供联系分机或邮箱。
4. 无法回答时回复："抱歉，这个问题建议直接联系行政部前台，分机 8888。"
## 回答示例
Q: 打印机坏了找谁？
A: 请联系IT运维部报修，分机 3333。报修时请准备好：1) 打印机编号（位于机身侧面标签）2) 故障现象描述。
## 参考文档
{context}"""
admin_chat = rag.create_chat(
name="云帆行政服务助手",
description="办公用品、会议室、设备报修、安全消防",
dataset_ids=[
ds_map["行政-办公服务"]["id"],
ds_map["行政-环境安全"]["id"],
],
llm_model="qwen2.5:14b@Ollama",
prompt_system=admin_prompt,
temperature=0.3,
retrieval_config={
"top_k": 10,
"similarity_threshold": 0.3,
}
)
print(f"[OK] 行政助手: {admin_chat.id}")
# 保存助手映射
assistants = {
"HR": hr_chat.id,
"Finance": finance_chat.id,
"Admin": admin_chat.id,
}
with open("assistant_mapping.json", "w") as f:
json.dump(assistants, f, indent=2)
步骤4：50 题评测与迭代调优
目标：运行评测脚本，对比初始得分和优化后得分。
# step4_evaluate_and_optimize.py
import json
import time
from collections import defaultdict
from ragflow import RAGFlow
rag = RAGFlow(api_key="ragflow-xxx", base_url="http://localhost/api/v1")
# 加载评测问题集（50题）
with open("evaluation_questions.json", "r") as f:
questions = json.load(f)
assistants = {
"HR": "chat_hr_001",
"Finance": "chat_finance_001",
"Admin": "chat_admin_001",
}
def run_evaluation(round_label):
results = []
for q in questions:
chat_id = assistants.get(q["assistant"])
start = time.time()
try:
response = rag.chat(
question=q["question"],
chat_id=chat_id,
stream=False,
timeout=30,
temperature=0.0,
)
latency = time.time() - start
answer = response.answer or ""
# 自动评分
expected_kw = q.get("expected_keywords", [])
hits = sum(1 for kw in expected_kw if kw in answer)
kw_score = hits / len(expected_kw) if expected_kw else 0.5
# 引用检查
refs = response.references or []
has_ref = len(refs) > 0
correct_doc_ref = any(
q.get("expected_doc", "") in ref.get("document_name", "")
for ref in refs
)
# 拒答检查
if q.get("should_reject"):
reject_phrases = q.get("reject_phrases", ["未找到", "抱歉", "无法", "范围"])
score = 1.0 if any(p in answer for p in reject_phrases) else 0.0
else:
score = (kw_score * 0.6 + (has_ref * 0.2) + (correct_doc_ref * 0.2))
except Exception as e:
latency = time.time() - start
score = 0.0
answer = f"[ERROR] {e}"
has_ref = False
correct_doc_ref = False
results.append({
"id": q["id"],
"question": q["question"],
"assistant": q["assistant"],
"difficulty": q.get("difficulty", "medium"),
"score": round(score, 4),
"has_ref": has_ref,
"correct_doc_ref": correct_doc_ref,
"latency_ms": round(latency * 1000),
"answer": answer[:200],
})
# 汇总
scores = [r["score"] for r in results]
by_assistant = defaultdict(list)
by_difficulty = defaultdict(list)
for r in results:
by_assistant[r["assistant"]].append(r["score"])
by_difficulty[r["difficulty"]].append(r["score"])
report = {
"round": round_label,
"total": len(results),
"avg_score": round(sum(scores) / len(scores), 4),
"ref_rate": round(sum(1 for r in results if r["has_ref"]) / len(results), 4),
"avg_latency_ms": round(sum(r["latency_ms"] for r in results) / len(results)),
"by_assistant": {
a: round(sum(s) / len(s), 4) for a, s in by_assistant.items()
},
"by_difficulty": {
d: round(sum(s) / len(s), 4) for d, s in by_difficulty.items()
},
"failures": [
{"id": r["id"], "q": r["question"], "score": r["score"]}
for r in results if r["score"] < 0.5
],
}
return report, results
# 第一轮：基准评测
print("=== 第一轮：基准评测 ===")
report1, results1 = run_evaluation("baseline")
print(f"平均分: {report1['avg_score']:.1%}")
print(f"引用率: {report1['ref_rate']:.1%}")
print(f"失败数: {len(report1['failures'])}")
print(f"各助手: {report1['by_assistant']}")
# 根据第一轮失败案例进行优化
print("\n=== 优化中 ===")
# 示例优化动作：
# 1. 合并低分问题的断裂切片
# 2. 为财务数据增加同义词别名
# 3. 调整 HR 助手的 similarity_threshold
# 4. 增加行政助手的知识库文档
print(" 1. 修复 3 个断裂切片 (手动合并)")
print(" 2. 补充财务报销标准的同义表达")
print(" 3. 调整 HR 助手检索阈值为 0.2")
print(" 4. 新增设备报修流程文档")
# 第二轮：优化后评测
print("\n=== 第二轮：优化后评测 ===")
report2, results2 = run_evaluation("optimized")
print(f"平均分: {report2['avg_score']:.1%}")
print(f"引用率: {report2['ref_rate']:.1%}")
print(f"失败数: {len(report2['failures'])}")
# 对比报告
print(f"\n=== 优化效果对比 ===")
print(f" 平均分: {report1['avg_score']:.1%} → {report2['avg_score']:.1%} "
f"({(report2['avg_score'] - report1['avg_score'])*100:+.1f}%)")
print(f" 引用率: {report1['ref_rate']:.1%} → {report2['ref_rate']:.1%}")
print(f" 失败数: {len(report1['failures'])} → {len(report2['failures'])}")
# 验收判定
PASS_THRESHOLD = 0.85
if report2["avg_score"] >= PASS_THRESHOLD:
print(f"\n✅ 验收通过！综合评分 {report2['avg_score']:.1%} >= {PASS_THRESHOLD:.0%}")
else:
print(f"\n❌ 验收未通过。综合评分 {report2['avg_score']:.1%} < {PASS_THRESHOLD:.0%}")
print("剩余失败案例：")
for f in report2["failures"]:
print(f" [{f['id']}] {f['q']} (得分: {f['score']})")
步骤5：运维交付物——部署文档和排障 SOP
目标：生成团队可长期维护的文档。
# step5_generate_docs.sh - 生成运维文档
#!/bin/bash
DOC_DIR="docs/delivery"
mkdir -p $DOC_DIR
# 1. 部署文档
cat > "$DOC_DIR/部署文档.md" << 'DEPLOYEOF'
# RAGFlow 企业知识库部署文档
## 架构总览
- 单机 Docker Compose 部署
- API Server + Task Executor + MySQL + Redis + MinIO + Infinity
- 本地 Ollama 提供 LLM + Embedding + Rerank
## 环境要求
| 项目 | 配置 |
|------|------|
| CPU | 8 核 |
| 内存 | 32GB |
| 磁盘 | 200GB SSD |
| 操作系统 | Ubuntu 22.04 LTS |
## 启动步骤
1. git clone + 修改 .env
2. sudo sysctl -w vm.max_map_count=262144
3. docker compose -f docker-compose-base.yml up -d
4. docker compose -f docker-compose.yml up -d
5. 访问 http://<服务器IP> 验证
## 数据备份
- MySQL: 每日凌晨 2 点 mysqldump
- MinIO: 每日凌晨 3 点 mc mirror 到备份节点
- 备份保留: 最近 7 天
DEPLOYEOF
# 2. 排障 SOP
cat > "$DOC_DIR/排障SOP.md" << 'SOPEOF'
# RAGFlow 排障标准作业程序
## 故障1: 文档一直"等待解析"
1. docker logs ragflow-task-executor --tail 50
2. docker exec ragflow-redis-1 redis-cli XLEN ragflow_tasks
3. 若 Worker 挂掉: docker restart ragflow-task-executor
4. 若队列积压 > 100: 临时增加 WS=3
## 故障2: 问答超时（>30s）
1. curl 测试 LLM API: ollama run qwen2.5:14b "ping"
2. 检查 Infinity: curl localhost:23820/admin/node/status
3. 若 LLM 慢: 换用云端 API 做临时降级
## 故障3: 检索结果为空
1. 确认 Embedding 模型在线: ollama ps
2. 检查文档解析状态和切片数
3. 降低 similarity_threshold 到 0.15
## 故障4: 答案答非所问
1. 检查是否开启了 Rerank
2. 在控制台查看该问题的检索片段是否相关
3. 若检索片段本就不相关 → 优化切片/重建索引
4. 若检索片段相关但答案不对 → 优化 Prompt
## 故障5: 某个数据集完全不可用
1. 检查 Infinity 中该数据集索引是否存在
2. 检查该数据集文档是否全部解析失败
3. 尝试重新创建数据集并重新导入
SOPEOF
# 3. 用户培训手册
cat > "$DOC_DIR/用户操作手册.md" << 'USEREOF'
# 云帆科技 RAGFlow 用户操作手册
## 快速上手
1. 打开浏览器访问 http://ragflow.yunfan.local
2. 使用企业邮箱+初始密码登录（首次登录需修改密码）
3. 选择你的部门助手：HR助手 / 财务助手 / 行政助手
4. 在输入框中输入你的问题，回车发送
5. 点击答案中的引用链接查看原文依据
## 提问技巧
- 尽量用完整的句子，而非关键词
好: "员工年假有多少天？" 差: "年假"
- 追问时可以用"那..."开头
Q1: "年假有几天？" Q2: "那没休完怎么办？"
- 如果答案不准确，可以换一种问法重试
## 常见问题
Q: 为什么答案中有 [1] [2] 这样的标记？
A: 这是引用编号，点击可以跳转到制度原文，帮助您确认答案的准确性。
Q: 能问和部门无关的问题吗？
A: 每个助手有其知识范围，超出范围的问题会自动拒绝回答。
USEREOF
echo "运维文档已生成到 $DOC_DIR/"
ls -la $DOC_DIR/
测试验证
# acceptance_test.py - 交付验收测试
import pytest
import json
import requests
from ragflow import RAGFlow
rag = RAGFlow(api_key="xxx", base_url="http://localhost/api/v1")
class TestAcceptance:
"""交付验收测试套件"""
def test_all_services_healthy(self):
"""验证所有服务健康"""
services = {
"API": "http://localhost/api/v1/version",
"MinIO": "http://localhost:9001/minio/health/live",
}
for name, url in services.items():
r = requests.get(url, timeout=5)
assert r.status_code == 200, f"{name} 不可用"
def test_all_datasets_exist(self):
"""验证 7 个数据集全部存在"""
with open("dataset_mapping.json") as f:
expected = json.load(f)
existing = rag.list_datasets(page_size=100)
existing_names = {ds.name for ds in existing}
for name in expected:
assert name in existing_names, f"数据集缺失: {name}"
def test_hr_assistant_answers(self):
"""验证 HR 助手可正常问答"""
response = rag.chat(
question="员工年假有多少天？",
chat_id="chat_hr_001",
stream=False,
temperature=0.0
)
assert response.answer is not None
assert len(response.answer) > 20
assert len(response.references) > 0, "HR 助手必须返回引用"
def test_finance_assistant_table(self):
"""验证财务助手表格回答"""
response = rag.chat(
question="出差住宿费标准是多少？",
chat_id="chat_finance_001",
stream=False,
temperature=0.0
)
assert "元" in response.answer
# 财务回答应包含数字
import re
assert re.search(r'\d+', response.answer), "财务回答应包含具体数字"
def test_admin_assistant_rejection(self):
"""验证行政助手拒答越界问题"""
response = rag.chat(
question="公司股票代码是什么？",
chat_id="chat_admin_001",
stream=False,
temperature=0.0
)
reject_phrases = ["未找到", "抱歉", "无法", "范围", "建议联系"]
assert any(p in response.answer for p in reject_phrases), \
"应拒答但未拒答"
def test_evaluation_score_above_threshold(self):
"""验证评测分数达标"""
with open("eval_report_optimized.json") as f:
report = json.load(f)["report"]
assert report["avg_score"] >= 0.85, \
f"评测分数 {report['avg_score']} < 0.85，未达标"
def test_latency_p95_under_8s(self):
"""验证 P95 延迟 < 8s"""
latencies = []
questions = [
"年假有几天？",
"加班费怎么算？",
"试用期离职流程是什么？",
"出差住宿标准？",
]
import time
for q in questions:
start = time.time()
rag.chat(question=q, chat_id="chat_hr_001", stream=False, temperature=0.0)
latencies.append((time.time() - start) * 1000)
latencies.sort()
p95 = latencies[int(len(latencies) * 0.95)] if len(latencies) > 1 else latencies[-1]
assert p95 < 8000, f"P95 延迟 {p95:.0f}ms >= 8000ms"
完整代码清单
Git 仓库：https://github.com/infiniflow/ragflow
| 路径 | 说明 |
|---|---|
docker/docker-compose.yml | 主服务编排 |
docker/docker-compose-base.yml | 基础设施编排 |
api/apps/sdk/chat.py | 问答 API 接口 |
rag/nlp/search.py | 检索实现 |
rag/flow/pipeline.py | 解析 Pipeline |
4 项目总结
优点 & 缺点
| 维度 | RAGFlow 企业方案 | 自建 RAG（LangChain+Milvus） | Azure AI Search + OpenAI | 采购商业 SaaS |
|---|---|---|---|---|
| 部署周期 | ★★☆ 1-2 周 | ★☆☆ 4-8 周开发 | ★★★ 1-3 天 | ★★★ 即开即用 |
| 文档解析能力 | ★★★ DeepDoc 深度解析 | ★★☆ 依赖第三方 Loader | ★★☆ 基础 OCR/文档 | ★★☆ 有限格式 |
| 定制化程度 | ★★★ 完全可定制 | ★★★ 完全可定制 | ★★☆ 受平台限制 | ★☆☆ 几乎不可定制 |
| 隐私安全 | ★★★ 全本地部署 | ★★★ 全本地部署 | ★☆☆ 数据上云 | ★☆☆ 数据上云 |
| 运维成本 | ★★☆ 需 1 人维护 | ★☆☆ 需 2-3 人团队 | ★★☆ 平台运维 | ★★☆ 订阅费 |
| 持续性优化 | ★★★ 评测闭环驱动 | ★★★ 自主开发 | ★★☆ 依赖平台更新 | ★☆☆ 依赖供应商 |
适用场景
- 中小企业内部知识库：500-2000 人规模，50-500 份文档，1-2 周交付。
- 多部门制度问答：HR、财务、法务、行政等多个部门共享平台但数据隔离。
- 合规与隐私敏感企业：全部部署在内网，文档数据不出企业边界。
- 需要持续优化的场景：有评测闭环，文档更新后自动重新解析，问题反馈驱动 Prompt 迭代。
- 快速 POC 打样：展示 RAG 能力给领导层，争取更多预算和推广资源。
不适用场景：
- 千万级文档规模：单机 Docker Compose 部署无法支撑海量文档，需升到中级篇的分布式方案。
- 实时秒级更新：文档变更后需立即在问答中生效的场景，RAGFlow 的重新解析需要分钟级延迟。
注意事项
- 不要跳过评测环节：这是本章最大的教训。没有评测的"上线"等于"赌博"。50 题评测 + 人工抽检是底线。
- 文档更新要有 SOP：HR 制度每年更新，新版本上传后旧版本是删除还是保留？没有 SOP 会导致问答混乱（新旧内容并存）。
- 默认密码必须在部署后第一分钟修改：这是安全基线，不是"可选"项。
- Prompt 要版本化管理：每次修改 Prompt 都记录变更原因和评测分数变化，方便回滚。
- LLM 费用监控：如果是云端模型，月初设置费用告警。一个 Prompt 写得不好（如每次返回 2000 字小作文），费用能翻 3 倍。
常见踩坑经验
| 故障现象 | 根因 | 解决方法 |
|---|---|---|
| 验收前夜发现某类问题全答错 | 该数据集解析全失败但没人注意到 | 导入时每批验收，而不是全量导入后再检查 |
| 上线第一周大量用户反馈"答非所问" | Prompt 中的 {context} 占位符在复制粘贴时丢了 | 用 Step3 的自动化脚本创建助手而非手动复制 |
| 评测 92% 但上线后真实用户满意度只有 50% | 评测集和真实用户问题分布差异大 | 从用户日志中抽样更新评测集 |
| 部署文档写完没人能复现 | 脚本依赖手动步骤（如"先创建目录"）未在文档中写出 | 提供一键部署脚本而非分步文字描述 |
| 用户培训后第二天就忘了怎么用 | 培训内容太技术化，业务人员听不懂 | 写"按我要做什么"而非"按功能按钮"的操作手册 |
思考题
-
上线一个月后，你发现三个部门的问答效果严重分化——HR 助手准确率 90%，财务助手 75%，行政助手只有 55%。请设计一个根因分析框架，定位行政助手的核心短板在哪一环（文档质量、切片策略、Prompt 设计、还是检索参数），并给出针对性的改进方案。
-
公司计划在 3 个月内将知识库从 50 份文档扩展到 500 份，用户从 500 人扩展到 2000 人。当前单机 Docker 部署能支撑吗？如果不能，请画出升级后的架构图（提示：参考中级篇的分布式部署、K8s、弹性扩缩容等主题），并给出渐进式迁移方案。
（答案提示见第17章末尾或附录 D。）
延伸阅读与资源
10倍开发者的 Dify 魔法书：从零构建全栈 AI 应用
后端工程师转型AI第一课-Ollama 与私有化大模型实战
大型语言模型(LLM) vLLM 高性能推理落地实战
Agent开发之LlamaIndex 实战修炼与源码进阶
大语言模型Transformers 实战修炼与源码剖析
DAMO开发者矩阵，由阿里巴巴达摩院和中国互联网协会联合发起，致力于探讨最前沿的技术趋势与应用成果，搭建高质量的交流与分享平台，推动技术创新与产业应用链接，围绕“人工智能与新型计算”构建开放共享的开发者生态。
更多推荐
3
0- 0
已为社区贡献2条内容
所有评论(0)