基于 OpenClaw 构建企业知识库 RAG 问答系统

RAG（Retrieval-Augmented Generation） 的核心思想是：不要试图让模型"记住"知识，而是在回答问题时实时去检索知识，再基于检索结果来生成答案。

流程如下：

你在 Telegram 问 OpenClaw： "新员工入职流程第三步是什么？" │ ▼ OpenClaw 触发 RAG Skill │ ▼ 把问题向量化 → 去本地 Qdrant 向量库检索 │ ▼ 找到最相关的 3 段文档片段 │ ▼ 把原始问题 + 文档片段一起发给 Claude │ ▼ Claude 基于真实文档生成答案 │ ▼ OpenClaw 把答案 + 来源引用发回你的 Telegram 

这就是全链路。接下来我们来一步步实现它。

三、技术选型

四、第一步：搭建向量知识库

4.1 启动 Qdrant

docker run -d\
--name qdrant \
-p6333:6333 \
-v$(pwd)/qdrant_data:/qdrant/storage \
 qdrant/qdrant 

启动后访问 http://localhost:6333/dashboard，能看到 Web UI 就代表成功。

4.2 安装依赖

pip install qdrant-client sentence-transformers \
 langchain-text-splitters pymupdf \
 python-docx rich --break-system-packages 

4.3 文档入库脚本

这个脚本负责读取你的文档、分割成合适的片段、向量化后存入 Qdrant。

# ingest.py —— 知识库构建脚本
import uuid
import sys
from pathlib import Path
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
from sentence_transformers import SentenceTransformer
from langchain_text_splitters import RecursiveCharacterTextSplitter
import fitz # PyMuPDF
from rich.console import Console
from rich.progress import track

console = Console()
COLLECTION = "enterprise_kb"
EMBED_MODEL = "BAAI/bge-large-zh-v1.5"
VECTOR_SIZE = 1024
CHUNK_SIZE = 512
CHUNK_OVERLAP = 64

def load_pdf(path: str) -> str:
    doc = fitz.open(path)
    return "\n".join(page.get_text() for page in doc)

def load_text(path: str) -> str:
    return Path(path).read_text(encoding="utf-8")

def main(docs_dir: str):
    client = QdrantClient(url="http://localhost:6333")
    embedder = SentenceTransformer(EMBED_MODEL)
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=CHUNK_SIZE,
        chunk_overlap=CHUNK_OVERLAP,
        separators=["\n\n", "\n", "。", "！", "？", " "],
    )
    # 创建向量集合（如果不存在）
    existing = {c.name for c in client.get_collections().collections}
    if COLLECTION not in existing:
        client.create_collection(
            COLLECTION, vectors_config=VectorParams(size=VECTOR_SIZE, distance=Distance.COSINE),
        )
    console.print(f"[green]✓ 创建集合：{COLLECTION}[/green]")
    # 扫描文档目录
    supported = {".pdf", ".txt", ".md"}
    files = [f for f in Path(docs_dir).rglob("*") if f.suffix.lower() in supported]
    console.print(f"[blue]发现 {len(files)} 个文档文件[/blue]")
    total_chunks = 0
    for file in track(files, description="处理文档中..."):
        try:
            raw = load_pdf(str(file)) if file.suffix == ".pdf" else load_text(str(file))
            chunks = splitter.split_text(raw)
            points = []
            for i, chunk in enumerate(chunks):
                # bge 模型建议加指令前缀
                vec = embedder.encode(f"为这个句子生成表示以用于检索相关文章：{chunk}", normalize_embeddings=True).tolist()
                points.append(PointStruct(id=str(uuid.uuid4()), vector=vec, payload={"text": chunk, "source": file.name, "chunk_id": i},))
            client.upsert(COLLECTION, points=points)
            total_chunks += len(points)
        except Exception as e:
            console.print(f"[yellow]⚠ 跳过 {file.name}: {e}[/yellow]")
    console.print(f"\n[bold green]✓ 入库完成！共处理 {len(files)} 个文件，{total_chunks} 个片段[/bold green]")

if __name__ == "__main__":
    docs_dir = sys.argv[1] if len(sys.argv) > 1 else "./docs"
    main(docs_dir)

把你的文档放到 ./docs/ 目录下，然后运行：

python ingest.py ./docs 

几分钟后，你的文档就全进向量库了。

五、第二步：编写 RAG 检索脚本

这是核心检索逻辑，负责接收问题、返回最相关的文档片段：

# search.py —— RAG 检索脚本（供 OpenClaw Skill 调用）
import sys
import json
from qdrant_client import QdrantClient
from sentence_transformers import SentenceTransformer, CrossEncoder

COLLECTION = "enterprise_kb"
EMBED_MODEL = "BAAI/bge-large-zh-v1.5"
RERANK_MODEL = "BAAI/bge-reranker-large"

client = QdrantClient(url="http://localhost:6333")
embedder = SentenceTransformer(EMBED_MODEL)
reranker = CrossEncoder(RERANK_MODEL)

def search(query: str, top_k_recall: int = 8, top_k_final: int = 3) -> list[dict]:
    """两阶段检索：向量召回 → Rerank 精排"""
    # 阶段一：向量检索（召回候选集）
    query_vec = embedder.encode(f"为这个句子生成表示以用于检索相关文章：{query}", normalize_embeddings=True).tolist()
    results = client.search(
        collection_name=COLLECTION,
        query_vector=query_vec,
        limit=top_k_recall,
        score_threshold=0.4, # 过滤掉明显不相关的结果
    )
    if not results:
        return []
    # 阶段二：Rerank 精排（Cross-Encoder，准确率更高）
    docs = [r.payload["text"] for r in results]
    sources = [r.payload.get("source", "未知") for r in results]
    pairs = [(query, doc) for doc in docs]
    scores = reranker.predict(pairs)
    ranked = sorted(zip(scores, docs, sources), key=lambda x: x[0], reverse=True)
    return [{"text": doc, "source": src, "score": round(float(score), 4)} for score, doc, src in ranked[:top_k_final]]

def format_output(query: str, docs: list[dict]) -> str:
    """格式化为易读的文本，供 OpenClaw 拼入 Prompt"""
    if not docs:
        return f"在知识库中未找到与「{query}」相关的内容。"
    parts = [f"以下是从企业知识库中检索到的相关内容，请基于这些内容回答问题：\n"]
    for i, doc in enumerate(docs, 1):
        parts.append(f"【参考资料 {i}】（来源：{doc['source']}）\n{doc['text']}\n")
    sources = list({d["source"] for d in docs})
    parts.append(f"\n---\n📎 引用来源：{' | '.join(sources)}")
    return "\n".join(parts)

if __name__ == "__main__":
    query = " ".join(sys.argv[1:]) if len(sys.argv) > 1 else ""
    if not query:
        print(json.dumps({"error": "请提供查询词"}, ensure_ascii=False))
        sys.exit(1)
    docs = search(query)
    print(format_output(query, docs))

测试一下：

python search.py "新员工入职需要准备哪些材料"

如果能看到相关文档片段，说明检索链路通了。

六、第三步：编写 OpenClaw Skill

现在是最关键的部分——把上面的检索能力"封装"成一个 OpenClaw Skill，让 OpenClaw 学会什么时候去用它。

6.1 创建 Skill 文件夹

mkdir -p ~/.openclaw/skills/enterprise-kb 

6.2 编写 SKILL.md

---
name: enterprise-kb
description: 在企业内部知识库中检索信息。当用户询问公司内部政策、产品手册、操作规范、历史工单、合规文件等内部知识时，必须调用此技能获取准确信息，禁止凭空作答。
metadata: {"openclaw":{"emoji":"📚","requires":{"bins":["python3"],"env":[]}}}
---
# Enterprise Knowledge Base Search

## 使用场景
每当用户提问涉及以下类型内容时，你必须通过此技能检索，不得直接回答：
- 公司内部政策、人事制度、行政规定
- 产品操作手册、技术规范、API 文档
- 历史工单、解决方案、经验总结
- 法律合规文件、合同模板
- 任何你不确定是否有内部文档的专业问题

## 调用方式
使用 bash 工具执行以下命令：
```bash
python3 {baseDir}/search.py <用户的完整问题>

将 <用户的完整问题> 替换为用户提问的自然语言表述，保持完整，不要简化。

结果处理规则

1. 有检索结果时：完全基于返回的参考资料回答问题，在答案末尾注明"📎 来源：xxx"
2. 无检索结果时：明确告知用户"企业知识库中暂无相关记录"，不要用通用知识补充
3. 永远不要在没有调用此技能的情况下，自行回答内部知识相关的问题

示例

用户问："报销需要提交哪些材料？"
→ 调用：python3 {baseDir}/search.py 报销需要提交哪些材料
→ 基于检索结果回答，注明来源文件名

#### 6.3 复制脚本文件

```bash
cp search.py ~/.openclaw/skills/enterprise-kb/
cp requirements.txt ~/.openclaw/skills/enterprise-kb/
# 安装依赖
cd ~/.openclaw/skills/enterprise-kb
pip install -r requirements.txt 

6.4 验证 Skill 加载

重启 OpenClaw 或开启新会话后，发送 /skills 命令，确认 enterprise-kb 出现在列表里。

七、实际体验：在 Telegram 里问公司文档

完成上面的步骤后，你就可以打开 Telegram，直接向 OpenClaw 提问了。

场景一：HR 政策查询

🧑 你：年假天数怎么算的，我工作 3 年了

🦞 OpenClaw：（调用 enterprise-kb 技能）
根据公司《员工手册》第 4.2 节，司龄满 1 年不足 3 年享有 5 天年假，满 3 年不足 10 年享有 10 天年假。您工作满 3 年，今年可享有 10 天带薪年假。
📎 来源：员工手册_v2024.pdf

场景二：技术文档查询

🧑 你：订单接口返回 ERROR_4023 是什么意思怎么处理

🦞 OpenClaw：（调用 enterprise-kb 技能）
根据订单系统 API 文档，ERROR_4023 表示"库存预占失败"，通常原因是商品库存不足或并发扣减冲突。建议处理步骤：1. 检查商品实际库存；2. 重试间隔 500ms 后最多重试 3 次；3. 若持续失败，通知仓储团队检查 SKU 状态。
📎 来源：订单 API 文档_v3.2.md

场景三：知识库中找不到

🧑 你：竞争对手 XX 公司的产品定价是多少

🦞 OpenClaw：企业知识库中暂无该竞争对手定价的相关记录。如需了解，建议联系市场部门或通过官方渠道获取。

这第三个场景同样重要——OpenClaw 不会用通用知识或猜测来填充答案。这正是企业场景最需要的行为。

八、进阶优化：让检索更聪明

基础版跑起来了，但距离生产级还有几个值得做的优化点。

8.1 查询改写：理解用户真实意图

用户的提问往往不够精确，或者夹带了上下文缩写。可以在 SKILL.md 里增加一条指令：

## 查询优化（在调用搜索前）
如果用户的问题较短或含有缩写，先在心里将其扩展为完整的检索词。
例如："报销流程" → "员工费用报销申请审批流程及所需材料"

这个"扩展"是由 Claude/GPT 在 Agent 推理阶段完成的，不需要额外代码。

8.2 多轮对话的 Query 改写

当用户在多轮对话中追问时，上下文很重要：

你：年假怎么算？
OpenClaw：司龄 3 年以下 5 天，3 年以上 10 天…
你：那病假呢？

"那病假呢"这个问题如果单独检索，效果会很差。可以在 SKILL.md 里加一条：

## 多轮对话处理
如果当前问题是一个代词性追问（如"那...呢"、"它的...是什么"），
请先根据对话历史补全问题的完整含义，再调用检索。
如上例应改写为"病假天数规定是什么"再进行检索。

8.3 元数据过滤：检索范围精细化

如果你的知识库覆盖多个部门，可以在入库时打上部门标签，检索时只搜特定范围：

# 入库时增加 metadata
payload = {
    "text": chunk,
    "source": file.name,
    "department": "HR", # ← 增加部门标签
    "year": 2024,
}
# 检索时过滤
from qdrant_client.models import Filter, FieldCondition, MatchValue
results = client.search(
    collection_name=COLLECTION,
    query_vector=query_vec,
    query_filter=Filter(
        must=[FieldCondition(key="department", match=MatchValue(value="HR"))],
    ),
    limit=top_k_recall,
)

8.4 定期增量更新

企业文档是会更新的，你需要一个增量入库机制。最简单的方法是记录文件的最后修改时间：

import hashlib
def file_hash(path: str) -> str:
    return hashlib.md5(Path(path).read_bytes()).hexdigest()
# 对比 hash，只处理有变化的文件
# 删旧 points（按 source 过滤删除），插新 points

或者更简单粗暴：每次重新入库，因为向量化速度够快，1000 页文档也就几分钟。

8.5 设置心跳提醒（Heartbeat）

OpenClaw 支持定时心跳任务，你可以让它每天早上主动提醒你知识库状态：

# 在 openclaw.json 里配置
{
    "heartbeat": {
        "enabled": true,
        "cronExpression": "0 9 * * 1-5",
        "message": "早上好！企业知识库今日状态：运行正常，共索引 XX 篇文档。有什么我能帮你查的吗？"
    }
}

九、安全与隐私：这套方案为什么适合企业

这是企业用户最关心的问题，也是这套方案的核心优势。

数据从不离开你的机器。 Qdrant 跑在你本地，文档向量化在本地完成。唯一出去的是：你的问题（Query）和检索到的文档片段，打包成 Prompt 发给 LLM API。

如果连这个也不想出去，可以接入本地模型：

# 用 Ollama 跑本地模型
ollama pull qwen2.5:32b
# 在 openclaw.json 里切换
{
    "model": {
        "provider": "ollama",
        "name": "qwen2.5:32b"
    }
}

这样整个链路 100% 本地，完全离线，企业数据安全级别拉满。

OpenClaw 官方的安全理念也与此一致，项目主页有一句话说得很好：

"Not enterprise. Not hosted. Infrastructure you control. This is what personal AI should feel like."

十、完整项目结构回顾

enterprise-kb-skill/
│ ├── SKILL.md # OpenClaw Skill 描述（必须）
│ ├── search.py # RAG 检索主脚本
│ ├── ingest.py # 知识库构建脚本（一次性使用）
│ ├── requirements.txt # Python 依赖
│ └── docs/ # 你的企业文档放这里
│     ├── 员工手册_v2024.pdf
│     ├── 产品 API 文档_v3.md
│     ├── 合规指引_2024Q4.docx
│     └── ...

三步完成部署：

# 1. 启动向量数据库
docker run -d -p6333:6333 -v ./data:/qdrant/storage qdrant/qdrant
# 2. 文档入库
python ingest.py ./docs
# 3. 安装 Skill 并重启 OpenClaw
cp -r enterprise-kb-skill ~/.openclaw/skills/enterprise-kb
openclaw restart # 或者开启新会话

然后打开 Telegram，开问。

十一、总结与思考

我们做了什么：

• 搞清楚了 OpenClaw 的本质：跑在你电脑上的开源私人 AI 助手，通过 Skills 可以无限扩展
• 用 Qdrant + bge-large-zh + Cross-Encoder Reranker 搭了一套两阶段 RAG 检索链路
• 把检索能力封装成一个 OpenClaw Skill，只需要一个 SKILL.md + 一个 Python 脚本
• 在 Telegram 里实现了"企业文档问答"的完整闭环

这套方案的根本价值是什么？

它把"企业知识库"从一个需要 IT 部门立项的大工程，变成了一个任何开发者下午就能跑起来的个人工具。你可以先给自己用，积累效果后再推给团队，最终形成团队共享的知识助手——而整个过程，你对数据有完全的掌控权。

OpenClaw 还很年轻（2026 年初才出现），Skills 生态正在快速成长。现在入场，还能赶上最有趣的那段时间。