在上一篇文章中，我们深入学习了 OpenClaw 的核心机制，开发了多格式文档解析技能，并学会了通过 API 调用 OpenClaw 技能。这篇文章，我们将集成向量数据库 Chroma，开发文档向量化和检索技能，完成 RAG 系统的核心检索模块。
摘要
- 理解向量数据库的工作原理与在 RAG 系统中的作用
- 掌握 Chroma 向量数据库的安装与基本操作
- 在 FastAPI 后端实现文档向量化、存储与检索接口
- 开发 OpenClaw RAG 检索技能，实现自动知识检索
- 完成完整的"文档上传→解析→向量化→检索→问答"基础流程
1. 向量数据库：RAG 系统的"知识仓库"
1.1 为什么我们需要向量数据库？
传统的关系型数据库擅长存储结构化数据和精确匹配查询，但无法处理语义相似性搜索。当用户提问"如何重置密码？“时，我们需要找到所有与"密码重置"语义相关的文档片段，而不仅仅是包含这几个关键词的文档。
向量数据库专门解决这个问题：
- 将文本转换为高维向量（Embedding）
- 存储这些向量并建立高效的索引
- 支持快速的相似度搜索，找到与查询向量最接近的向量
1.2 为什么选择 Chroma？
在众多向量数据库中，我们选择 Chroma 作为本地开发的首选：
- 零配置：无需额外部署服务，直接作为 Python 库使用
- 开发友好：API 简单直观，文档完善
- 性能足够：对于中小规模知识库（百万级向量）完全够用
- 开源免费：MIT 协议，可自由使用和修改
- 易于迁移：如果后续需要扩展，可以轻松迁移到 Pinecone、Weaviate 等生产级向量数据库
1.3 嵌入模型选择
嵌入模型负责将文本转换为向量，它的质量直接影响 RAG 系统的检索效果。我们选择：
- 默认：OpenAI
text-embedding-3-small（效果好，价格便宜，速度快） - 备选：BGE-m3（开源本地模型，效果接近 OpenAI 模型）
- 生产环境：根据预算和性能需求选择合适的模型
2. FastAPI 后端：向量存储与检索接口开发
我们将在 FastAPI 后端实现所有向量相关的操作，因为 Python 在 AI 和数据处理方面有最好的生态支持。
2.1 安装依赖
进入 backend 目录，激活虚拟环境，安装所需依赖：
cd backend
source venv/bin/activate # Windows 使用 venv\Scripts\activate
pip install chromadb openai python-dotenv langchain langchain-text-splitters
创建 .env 文件，添加你的 OpenAI API Key：
OPENAI_API_KEY=your-openai-api-key-here
CHROMA_PERSIST_DIRECTORY=./chroma_db
2.2 文本分块策略
在向量化之前，我们需要将长文档拆分成适合模型处理的小块。这是 RAG 系统中最关键的步骤之一，直接影响检索效果。
我们采用固定长度分块+重叠的策略：
- 块大小：512 个 token（约 300-400 个汉字）
- 重叠大小：50 个 token（确保上下文不被切断）
创建 utils/text_splitter.py：
from langchain_text_splitters import RecursiveCharacterTextSplitter
def split_text(text: str, chunk_size: int = 512, chunk_overlap: int = 50) -> list[str]:
"""
将长文本拆分成小块
"""
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
length_function=len,
is_separator_regex=False,
)
chunks = text_splitter.split_text(text)
return chunks
2.3 向量数据库服务
创建 services/vector_service.py，实现向量的添加、检索和删除功能：
import os
import chromadb
from chromadb.utils import embedding_functions
from dotenv import load_dotenv
from utils.text_splitter import split_text
load_dotenv()
class VectorService:
def __init__(self):
self.persist_directory = os.getenv("CHROMA_PERSIST_DIRECTORY", "./chroma_db")
self.client = chromadb.PersistentClient(path=self.persist_directory)
# 使用 OpenAI 嵌入模型
self.embedding_function = embedding_functions.OpenAIEmbeddingFunction(
api_key=os.getenv("OPENAI_API_KEY"),
model_name="text-embedding-3-small"
)
# 获取或创建集合
self.collection = self.client.get_or_create_collection(
name="knowledge_base",
embedding_function=self.embedding_function,
metadata={"description": "企业级智能知识库向量集合"}
)
def add_document(self, document_id: str, content: str, metadata: dict = None) -> dict:
"""
添加文档到向量数据库
"""
# 拆分文本
chunks = split_text(content)
# 生成块 ID
chunk_ids = [f"{document_id}_chunk_{i}" for i in range(len(chunks))]
# 准备元数据
chunk_metadatas = []
for i, chunk in enumerate(chunks):
chunk_metadata = {
"document_id": document_id,
"chunk_index": i,
"chunk_count": len(chunks),
"content_preview": chunk[:100] + "..." if len(chunk) > 100 else chunk
}
if metadata:
chunk_metadata.update(metadata)
chunk_metadatas.append(chunk_metadata)
# 添加到向量数据库
self.collection.add(
ids=chunk_ids,
documents=chunks,
metadatas=chunk_metadatas
)
return {
"document_id": document_id,
"chunk_count": len(chunks),
"success": True
}
def search(self, query: str, top_k: int = 5) -> list[dict]:
"""
搜索与查询最相关的文档片段
"""
results = self.collection.query(
query_texts=[query],
n_results=top_k
)
# 格式化结果
formatted_results = []
for i in range(len(results["ids"][0])):
formatted_results.append({
"id": results["ids"][0][i],
"document": results["documents"][0][i],
"metadata": results["metadatas"][0][i],
"distance": results["distances"][0][i] if results["distances"] else None
})
return formatted_results
def delete_document(self, document_id: str) -> dict:
"""
删除文档及其所有块
"""
# 查找该文档的所有块
results = self.collection.get(
where={"document_id": document_id}
)
if not results["ids"]:
return {
"document_id": document_id,
"deleted_count": 0,
"success": True,
"message": "文档不存在"
}
# 删除所有块
self.collection.delete(ids=results["ids"])
return {
"document_id": document_id,
"deleted_count": len(results["ids"]),
"success": True
}
# 创建全局实例
vector_service = VectorService()
2.4 添加 API 接口
修改 main.py，添加向量相关的 API 接口：
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from services.vector_service import vector_service
app = FastAPI(title="企业级智能知识库 API", version="0.1.0")
# 配置 CORS
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # 开发环境允许所有来源，生产环境请限制为具体域名
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 请求模型
class AddDocumentRequest(BaseModel):
document_id: str
content: str
metadata: dict = None
class SearchRequest(BaseModel):
query: str
top_k: int = 5
class DeleteDocumentRequest(BaseModel):
document_id: str
@app.get("/")
async def root():
return {"message": "企业级智能知识库后端已启动", "version": "0.1.0"}
@app.get("/api/health")
async def health_check():
return {"status": "healthy"}
@app.post("/api/vector/add-document")
async def add_document(request: AddDocumentRequest):
try:
result = vector_service.add_document(
document_id=request.document_id,
content=request.content,
metadata=request.metadata
)
return result
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/vector/search")
async def search_documents(request: SearchRequest):
try:
results = vector_service.search(
query=request.query,
top_k=request.top_k
)
return {"results": results, "count": len(results)}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.delete("/api/vector/delete-document")
async def delete_document(request: DeleteDocumentRequest):
try:
result = vector_service.delete_document(document_id=request.document_id)
return result
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
2.5 测试接口
重启 FastAPI 服务：
uvicorn main:app --reload --port 8000
打开 http://localhost:8000/docs，使用 Swagger UI 测试接口：
- 调用
/api/vector/add-document添加一个测试文档 - 调用
/api/vector/search搜索相关内容 - 验证搜索结果是否符合预期
3. OpenClaw RAG 检索技能开发
现在我们来开发 OpenClaw 的 RAG 检索技能，它将调用 FastAPI 的检索接口，为大模型提供相关的知识上下文。
3.1 创建 RAG 检索技能目录
cd ../openclaw
mkdir skills/rag-retrieval
3.2 编写技能定义文件
创建 skills/rag-retrieval/skill.yaml：
name: rag-retrieval
description: 从企业知识库中检索与用户问题相关的文档片段。当用户询问企业内部知识、产品信息、规章制度等问题时，必须先调用此技能获取相关信息。
author: Your Name
version: 0.1.0
category: 知识库
parameters:
type: object
properties:
query:
type: string
description: 用户的问题，用于检索相关文档
top_k:
type: integer
description: 返回的最相关文档数量
default: 5
minimum: 1
maximum: 20
required:
- query
returns:
type: object
properties:
success:
type: boolean
description: 是否检索成功
results:
type: array
description: 检索到的文档片段列表
items:
type: object
properties:
document:
type: string
description: 文档片段内容
metadata:
type: object
description: 文档元数据
distance:
type: number
description: 相似度距离（越小越相似）
count:
type: integer
description: 返回的结果数量
3.3 编写技能执行代码
创建 skills/rag-retrieval/index.ts：
import { SkillContext, SkillResult } from 'openclaw';
import axios from 'axios';
interface RagRetrievalParams {
query: string;
top_k?: number;
}
export default async function execute(
params: RagRetrievalParams,
context: SkillContext
): Promise<SkillResult> {
const { query, top_k = 5 } = params;
try {
context.logger.info(`开始检索知识库，查询: "${query}", top_k: ${top_k}`);
// 调用 FastAPI 检索接口
const response = await axios.post('http://localhost:8000/api/vector/search', {
query,
top_k
});
const { results, count } = response.data;
context.logger.info(`检索完成，找到 ${count} 个相关文档片段`);
// 记录检索结果（用于调试）
results.forEach((result: any, index: number) => {
context.logger.debug(`结果 ${index+1}: 距离=${result.distance?.toFixed(4)}, 来源=${result.metadata?.fileName || '未知'}`);
});
return {
success: true,
data: {
success: true,
results,
count
}
};
} catch (error: any) {
const errorMessage = error.response?.data?.detail || error.message;
context.logger.error(`知识库检索失败: ${errorMessage}`);
return {
success: false,
error: `知识库检索失败: ${errorMessage}`
};
}
}
3.4 安装依赖并注册技能
安装 axios 依赖：
pnpm add axios
编辑 .claw/config.yaml，添加 RAG 检索技能：
skills:
- path: ../skills/hello-world
enabled: true
- path: ../skills/document-parser
enabled: true
- path: ../skills/rag-retrieval
enabled: true
3.5 配置 OpenClaw 自动调用检索技能
为了让 OpenClaw 在回答用户问题时自动调用 RAG 检索技能，我们需要修改系统提示词。
编辑 .claw/config.yaml，添加以下内容：
system_prompt: |
你是一个专业的企业知识库助手。你的任务是基于企业内部知识库的内容，准确、清晰地回答用户的问题。
重要规则：
1. 当用户询问任何与企业内部知识、产品信息、规章制度、流程规范相关的问题时，**必须先调用 rag-retrieval 技能**获取相关信息。
2. 你的回答必须严格基于检索到的文档内容，不得编造信息。
3. 如果检索结果中没有相关信息，请明确告诉用户"知识库中没有找到相关信息"，不要猜测。
4. 回答时要引用来源，格式为：[来源: 文件名]
5. 保持回答简洁明了，重点突出。
3.6 测试完整 RAG 流程
现在我们来测试完整的"文档添加→检索→问答"流程：
-
添加测试文档：
使用 Swagger UI 调用/api/vector/add-document接口，添加一个测试文档：{ "document_id": "test-doc-001", "content": "公司员工请假制度：\n1. 员工享有每年 10 天的带薪年假，工作满 1 年以上可享受。\n2. 病假需要提供医院开具的病假证明，病假期间工资按 80% 发放。\n3. 事假需要提前 1 天申请，事假期间不发放工资。\n4. 加班可以调休，调休需要在加班后 3 个月内使用。", "metadata": { "fileName": "员工请假制度.md", "department": "人力资源部", "version": "1.0" } } -
重启 OpenClaw 网关：
pnpm run start -
测试问答：
打开 OpenClaw Web 管理界面，输入问题：员工每年有多少天带薪年假？你应该会看到 OpenClaw 自动调用
rag-retrieval技能，然后基于检索结果给出回答：正在调用 rag-retrieval 技能... 技能调用成功，找到 1 个相关文档片段。 根据公司员工请假制度，员工享有每年 10 天的带薪年假，工作满 1 年以上可享受。 [来源: 员工请假制度.md]
恭喜！你已经成功实现了 RAG 系统的核心功能。
4. 项目结构更新
至此，我们的项目目录已经更新为：
enterprise-knowledge-base/
├── .gitignore
├── backend/
│ ├── venv/
│ ├── .env
│ ├── main.py
│ ├── requirements.txt
│ ├── utils/
│ │ └── text_splitter.py
│ └── services/
│ └── vector_service.py
├── openclaw/
│ ├── node_modules/
│ ├── package.json
│ ├── .claw/
│ │ └── config.yaml
│ └── skills/
│ ├── hello-world/
│ ├── document-parser/
│ └── rag-retrieval/
│ ├── skill.yaml
│ └── index.ts
└── frontend/
├── node_modules/
├── src/
├── index.html
├── package.json
└── tsconfig.json
5. 小结
本篇文章我们完成了以下事情：
- 理解了向量数据库的工作原理和在 RAG 系统中的核心作用
- 集成了 Chroma 向量数据库，实现了文档向量化、存储和检索功能
- 设计并实现了合理的文本分块策略
- 开发了 OpenClaw RAG 检索技能，实现了自动知识检索
- 配置了系统提示词，让大模型基于检索结果回答问题
- 测试了完整的"文档添加→检索→问答"基础 RAG 流程