37e RAG 知识库搭建

# 安装 Chroma
pip install chromadb
 
# 启动 Chroma Server（生产模式）
chroma run --host 0.0.0.0 --port 8000 --path /data/chroma

import chromadb
 
client = chromadb.HttpClient(host="localhost", port=8000)
 
# 创建 Collection（按知识分类）
tech_collection = client.get_or_create_collection(
    name="tech_docs",
    metadata={"description": "技术架构和开发规范文档"}
)
 
business_collection = client.get_or_create_collection(
    name="business_docs",
    metadata={"description": "业务知识和产品文档"}
)
 
experience_collection = client.get_or_create_collection(
    name="experience_docs",
    metadata={"description": "排错经验和复盘报告"}
)

-- 安装 pgvector 扩展
CREATE EXTENSION IF NOT EXISTS vector;
 
-- 创建知识库表
CREATE TABLE knowledge_chunks (
  id SERIAL PRIMARY KEY,
  source_type VARCHAR(50) NOT NULL,     -- feishu_wiki, feishu_doc, github_readme
  source_id VARCHAR(200) NOT NULL,       -- 飞书文档 token 或 GitHub 路径
  source_title VARCHAR(500),
  chunk_index INTEGER NOT NULL,
  content TEXT NOT NULL,
  embedding vector(1536),                -- OpenAI text-embedding-3-small 维度
  metadata JSONB,                        -- 额外元数据
  created_at TIMESTAMP DEFAULT NOW(),
  updated_at TIMESTAMP DEFAULT NOW()
);
 
-- 创建向量索引（IVFFlat，适合中等数据量）
CREATE INDEX idx_knowledge_embedding ON knowledge_chunks
  USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100);
 
-- 创建元数据索引
CREATE INDEX idx_knowledge_source ON knowledge_chunks(source_type, source_id);

飞书知识库/文档
      │
      ▼
  内容提取（lark-mcp）
      │
      ▼
  文本清洗（去除格式标记、空白）
      │
      ▼
  文本分块（按标题/段落/滑动窗口）
      │
      ▼
  生成 Embedding（OpenAI API）
      │
      ▼
  存入向量数据库（Chroma/pgvector）

"""
飞书知识库 → Chroma 向量数据库 Embedding 管道
"""
import os
import re
import chromadb
from openai import OpenAI
 
# 初始化客户端
openai_client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
chroma_client = chromadb.HttpClient(host="localhost", port=8000)
collection = chroma_client.get_or_create_collection("tech_docs")
 
 
def extract_feishu_doc(doc_token: str) -> dict:
    """
    通过 lark-mcp 提取飞书文档内容。
    实际使用时通过 OpenClaw Skill 调用 lark-mcp 工具。
    这里展示数据结构。
    """
    return {
        "token": doc_token,
        "title": "文档标题",
        "content": "文档正文内容...",
        "url": f"https://xxx.feishu.cn/docx/{doc_token}"
    }
 
 
def chunk_by_heading(content: str, max_tokens: int = 800) -> list[dict]:
    """按标题分块，超长块二次切分"""
    sections = re.split(r'\n(#{1,3}\s+)', content)
    chunks = []
    current_heading = ""
    current_text = ""
 
    for i, section in enumerate(sections):
        if re.match(r'^#{1,3}\s+', section):
            current_heading = section.strip()
        else:
            text = f"{current_heading}\n{section.strip()}" if current_heading else section.strip()
            if not text:
                continue
 
            # 粗略估算 token 数（中文约 1.5 字/token）
            estimated_tokens = len(text) / 1.5
            if estimated_tokens > max_tokens:
                # 滑动窗口二次切分
                words = text.split('\n')
                sub_chunk = ""
                for line in words:
                    if len(sub_chunk + line) / 1.5 > max_tokens:
                        if sub_chunk:
                            chunks.append({"heading": current_heading, "content": sub_chunk.strip()})
                        sub_chunk = line + "\n"
                    else:
                        sub_chunk += line + "\n"
                if sub_chunk.strip():
                    chunks.append({"heading": current_heading, "content": sub_chunk.strip()})
            else:
                chunks.append({"heading": current_heading, "content": text})
 
    return chunks
 
 
def generate_embeddings(texts: list[str]) -> list[list[float]]:
    """批量生成 Embedding"""
    response = openai_client.embeddings.create(
        model="text-embedding-3-small",
        input=texts
    )
    return [item.embedding for item in response.data]
 
 
def index_document(doc: dict):
    """索引单个文档"""
    chunks = chunk_by_heading(doc["content"])
 
    if not chunks:
        return
 
    # 批量生成 Embedding
    texts = [chunk["content"] for chunk in chunks]
    embeddings = generate_embeddings(texts)
 
    # 批量写入 Chroma
    collection.upsert(
        ids=[f"{doc['token']}_chunk_{i}" for i in range(len(chunks))],
        embeddings=embeddings,
        documents=texts,
        metadatas=[{
            "source_type": "feishu_wiki",
            "source_id": doc["token"],
            "source_title": doc["title"],
            "source_url": doc["url"],
            "chunk_heading": chunk["heading"],
            "chunk_index": i
        } for i, chunk in enumerate(chunks)]
    )
 
    print(f"✅ Indexed {doc['title']}: {len(chunks)} chunks")
 
 
# 使用示例
if __name__ == "__main__":
    # 实际使用时，文档列表从飞书知识库 API 获取
    doc_tokens = ["doccnxxxxxxxxxx", "doccnyyyyyyyyyy"]
    for token in doc_tokens:
        doc = extract_feishu_doc(token)
        index_document(doc)

# knowledge-rag-search
 
根据用户问题检索知识库，结合检索结果生成回答。
 
## 触发条件
- 用户消息包含"知识库"、"查一下"、"有没有文档"、"之前怎么做的"
- 其他 Skill 无法匹配时的兜底（知识库问答）
 
## 执行步骤
1. 提取用户问题的核心关键词
2. 将用户问题转换为 Embedding 向量
3. 在向量数据库中检索最相关的 5 个文档块（top_k=5，相似度阈值 > 0.7）
4. 将检索到的文档块作为上下文，结合用户问题生成回答
5. 在回答末尾附上参考文档链接
6. 将本次问答写入记忆，积累常见问题
 
## 检索参数
- top_k: 5
- 相似度阈值: 0.7（低于此阈值的结果不使用）
- 搜索范围: 所有 Collection（tech_docs, business_docs, experience_docs）
 
## Prompt 模板
你是 HackQuest 团队的知识助手。根据以下检索到的文档内容回答用户问题。
 
规则：
1. 只基于检索到的文档内容回答，不要编造信息
2. 如果检索结果不足以回答问题，明确告知用户"知识库中暂无相关信息"
3. 回答末尾列出参考文档链接
4. 使用简洁清晰的中文回答
 
检索到的文档：
{retrieved_chunks}
 
用户问题：{user_question}
 
## 输出格式
回答正文（2-5 段）
 
📚 参考文档：
- [文档标题1](链接1)
- [文档标题2](链接2)
 
## 示例输出
**用户问题**: 我们的支付系统用的什么方案？
 
**回答**:
HackQuest 的支付系统采用 Stripe 作为主要支付网关，支持信用卡和加密货币支付。
 
核心架构：
- 前端使用 Stripe Elements 组件
- 后端通过 Stripe Node.js SDK 处理支付意图
- Webhook 接收支付状态回调，更新订单状态
- 加密货币支付通过 Coinbase Commerce API 集成
 
详细的技术方案见架构文档。
 
📚 参考文档：
- [支付系统架构设计](https://xxx.feishu.cn/docx/xxx)
- [Stripe 集成指南](https://xxx.feishu.cn/wiki/xxx)
 
## 权限要求
- 向量数据库读取权限
- OpenAI Embedding API 调用权限
 
## 错误处理
- 向量数据库连接失败：回复"知识库服务暂时不可用，请稍后再试"
- 检索结果为空：回复"知识库中暂无相关信息，建议联系 [相关负责人]"
- Embedding API 调用失败：使用关键词匹配作为降级方案

[系统 Prompt]
你是 HackQuest 团队的知识助手...

[检索结果 - 按相似度降序]
--- 文档 1 (相似度: 0.92) ---
来源: 支付系统架构设计
内容: ...

--- 文档 2 (相似度: 0.85) ---
来源: Stripe 集成指南
内容: ...

[用户问题]
我们的支付系统用的什么方案？

飞书文档变更 Webhook
        │
        ▼
OpenClaw 接收事件
        │
        ▼
knowledge-auto-update Skill
        │
        ├─→ 提取变更文档内容
        ├─→ 删除旧的文档块
        ├─→ 重新分块和 Embedding
        └─→ 写入向量数据库

# knowledge-auto-update
 
监听飞书文档变更事件，自动更新向量数据库中的对应文档。
 
## 触发条件
飞书文档变更事件（文档编辑保存后触发）。
 
## 执行步骤
1. 从事件中提取文档 token
2. 通过 lark-mcp 读取文档最新内容
3. 删除向量数据库中该文档的所有旧块（通过 source_id 匹配）
4. 对新内容执行分块和 Embedding
5. 将新块写入向量数据库
6. 记录更新日志到记忆
 
## 过滤规则
- 只处理知识库和指定文件夹下的文档
- 忽略草稿状态的文档
- 忽略 5 分钟内的重复变更（防抖）
 
## 权限要求
- 飞书文档读取权限
- 向量数据库读写权限
- OpenAI Embedding API

# knowledge-full-rebuild
 
每周日凌晨 2:00 全量重建知识库索引。
 
## 触发条件
定时触发：每周日 02:00
 
## 执行步骤
1. 通过 lark-mcp 获取知识库所有文档列表
2. 清空向量数据库中所有 Collection
3. 逐个文档执行分块 → Embedding → 入库
4. 统计重建结果（文档数、块数、耗时）
5. 发送重建报告到运维群
 
## 预计耗时
- 200 篇文档约 30 分钟
- 主要耗时在 Embedding API 调用（受限于 API rate limit）

[
  {
    "question": "我们的支付系统用的什么方案？",
    "expected_sources": ["支付系统架构设计"],
    "expected_answer_contains": ["Stripe", "Coinbase Commerce"],
    "category": "tech_architecture"
  },
  {
    "question": "新员工入职第一天需要做什么？",
    "expected_sources": ["入职流程指南"],
    "expected_answer_contains": ["飞书账号", "GitHub 权限", "开发环境"],
    "category": "process"
  },
  {
    "question": "上次数据库连接池满了怎么解决的？",
    "expected_sources": ["2026-06 数据库故障复盘"],
    "expected_answer_contains": ["连接池", "max_connections", "pgbouncer"],
    "category": "experience"
  }
]