Local Doc RAG - 本地文档智能问答

纯本地运行的私密文档问答 pipeline。不上云、不泄密、用端侧算力。

核心架构

用户查询 ──▶ Qwen3.5-35B（大脑·SiliconFlow）
                  │
          指令遵循 / 规划
                  │
  ┌───────────────┴───────────────┐
  ▼                               ▼
向量化模块                    RAG 查询引擎
（本地 embedding）            （本地 FAISS）
  │                               │
  ▼                               ▼
文档切分器                    语义检索
（多格式支持）                + 重排序
  │                               │
  └───────────────┬───────────────┘
                  ▼
           Qwen 生成答案
          （含引用溯源）

工作流程

第一步：建索引（一次性）

用户说"索引我的文档"或"建知识库"时：

1. 扫描本地文档目录（支持 ~/.openclaw/workspace/docs/ 或用户指定路径）
2. 解析格式：PDF / TXT / MD / DOCX
3. 语义切分：每段 512 tokens，重叠 64 tokens
4. 用本地 embedding 模型生成向量（sentence-transformers/all-MiniLM-L6-v2）
5. 存入 FAISS 索引文件（~/.openclaw/workspace/vectorstore/）
6. 保存文档元数据（路径、页码、标题）

第二步：查询（每次问答）

用户提问时：

1. 用户 query → 本地 embedding → FAISS 语义检索 Top-K
2. K 默认=10，做粗筛
3. 重排序（Rerank）：用 cross-encoder 将 query 和所有 chunk 配对打分，取 Top-3
4. 将 Top-3 chunk 注入 prompt
5. Qwen3.5-35B 生成带引用的答案
6. 输出：答案 + 引用列表（含文件路径、页码、相关度分数）

详细指令

索引命令

# 基本用法
python3 skills/local-doc-rag/scripts/index_documents.py \
  --input /path/to/docs \
  --output ~/.openclaw/workspace/vectorstore/ \
  [--chunk-size 512] \
  [--overlap 64] \
  [--recursive]

# 示例：索引整个 workspace 的文档
python3 skills/local-doc-rag/scripts/index_documents.py \
  --input ~/openclaw/workspace/docs \
  --output ~/.openclaw/workspace/vectorstore/ \
  --recursive

输出：

index.faiss — FAISS 索引文件
metadata.json — 文档元数据（路径、页码、标题、chunk 偏移）
chunk_texts.parquet — 原始文本块（用于 Rerank）

查询命令

# 基本查询
python3 skills/local-doc-rag/scripts/rag_pipeline.py \
  --query "这份文档主要讲了什么？" \
  [--top-k 10] \
  [--rerank-top 3] \
  [--model Qwen/Qwen3.1-8B] \
  [--api-key YOUR_SILICONFLOW_KEY] \
  [--base-url https://api.siliconflow.cn/v1]

# 交互式问答
python3 skills/local-doc-rag/scripts/rag_pipeline.py \
  --interactive \
  [--system-prompt "你是一个专业的技术文档助手。"] \
  [--show-citations]

参数说明：

| 参数 | 默认值 | 说明 | |------|--------|------| | --query | 必填 | 用户问题 | | --top-k | 10 | FAISS 初步检索数量 | | --rerank-top | 3 | 重排后返回数量（送入 Qwen） | | --model | Qwen/Qwen3.5-35B-A3B | SiliconFlow 模型 ID | | --api-key | 环境变量 SILICONFLOW_API_KEY | SiliconFlow 密钥 | | --base-url | https://api.siliconflow.cn/v1 | API 地址 | | --temperature | 0.3 | 生成温度（偏低保持准确性） | | --show-citations | False | 是否在输出中显示引用块 |

验证命令

# 运行内置测试集
python3 skills/local-doc-rag/scripts/evaluate_skill.py \
  --vectorstore ~/.openclaw/workspace/vectorstore/ \
  --test-queries tests/test_queries.json

核心脚本说明

index_documents.py

多格式文档解析与向量化索引构建：

PDF：用 PyPDF2 提取文本，保留页码和段落结构
TXT/MD：按换行符+段落切分
DOCX：用 python-docx 提取段落
切分策略：递归字符切分（RecursivelySplit），优先保持段落完整
Embedding：使用 sentence-transformers/all-MiniLM-L6-v2（384维，~90MB，本地运行）
索引：FAISS IndexFlatIP（Inner Product，余弦相似度等价）

关键代码片段：

from sentence_transformers import SentenceTransformer
import faiss
import numpy as np

# 加载本地 embedding 模型
model = SentenceTransformer('all-MiniLM-L6-v2')
embeddings = model.encode(chunks, normalize_embeddings=True)
embeddings = embeddings.astype('float32')

# 构建 FAISS 索引
index = faiss.IndexFlatIP(embeddings.shape[1])
faiss.normalize_L2(embeddings)
index.add(embeddings)
faiss.write_index(index, "index.faiss")

rag_pipeline.py

完整 RAG 查询 pipeline（含重排序）：

# Step 1: 语义检索
query_vec = model.encode([query], normalize_embeddings=True).astype('float32')
faiss.normalize_L2(query_vec)
_, indices = index.search(query_vec, top_k)

# Step 2: Rerank（使用 cross-encoder 精细打分）
from sentence_transformers import CrossEncoder
reranker = CrossEncoder('BAAI/bge-reranker-base')
pairs = [(query, chunks[i]) for i in indices[0]]
scores = reranker.predict(pairs)
ranked = sorted(zip(indices[0], scores), key=lambda x: x[1], reverse=True)
top_chunks = [chunks[i] for i, _ in ranked[:rerank_top]]

# Step 3: Qwen 生成
messages = [
    {"role": "system", "content": SYSTEM_PROMPT},
    {"role": "user", "content": f"参考文档：\n{chr(10).join(top_chunks)}\n\n问题：{query}"}
]
# → OpenAI 兼容格式调用 SiliconFlow

evaluate_skill.py

自动化 Skill 验证脚本：

索引完整性检查：验证 FAISS 索引是否包含足够 chunks
检索质量测试：给定已知问题，验证相关文档是否被检索到
端到端 RAG 测试：运行完整 pipeline，验证答案质量（可选 LLM judge）
并发查询压测：模拟多并发请求，验证响应时间 <5s（不含 API 延迟）

模型配置

SiliconFlow API（由用户提供）：

Base URL: https://api.siliconflow.cn/v1
Model: Qwen/Qwen3.5-35B-A3B（默认）或 Qwen/Qwen3.1-8B（更便宜）
支持 OpenAI 兼容格式（/v1/chat/completions）

本地模型：

Embedding: sentence-transformers/all-MiniLM-L6-v2（384维，~90MB）
Reranker: BAAI/bge-reranker-base（~220MB，需额外下载）

首次运行会自动下载模型，后续纯本地运行。

技术亮点

1. 隐私优先架构

文档数据 100% 本地存储：向量化在本地完成，文档不离开机器
仅将 query 和检索结果 发送给 SiliconFlow API
用户可配置 --no-external-api 完全离线模式（使用本地 Qwen Ollama 版本）

2. 语义 chunk 优化

递归切分（RecursivelySplit）：先按段落，再按句子，避免截断语义单元
重叠 chunk（Overlap）：相邻 chunk 有 64 token 重叠，保证跨 chunk 的信息连续性
元数据保留：每个 chunk 携带来源文件、页码、行号，便于溯源

3. 两阶段检索

粗筛（FAISS）：从海量 chunk 中快速找到 Top-10 候选
精排（Cross-Encoder Rerank）：用语义理解模型对每个候选重新打分，取 Top-3
双重保障：平衡速度与精度

4. 引用溯源机制

{
  "answer": "Transformer 的核心机制是自注意力...",
  "citations": [
    {
      "source": "~/docs/attention_is_all_you_need.pdf",
      "page": 1,
      "chunk_id": 42,
      "relevance_score": 0.94,
      "excerpt": "Attention Is All You Need. We propose a new architecture..."
    }
  ]
}

5. 指令模板（System Prompt）

你是一个专业的文档智能助手，基于提供的参考文档片段回答用户问题。

要求：
1. 严格基于参考文档内容回答，不要编造
2. 如果文档中没有相关内容，明确说明"根据提供的文档，无法回答该问题"
3. 回答时用序号标注引用，例如：[1] [2]
4. 在回答末尾附上完整的引用列表（格式：来源文件 - 页码 - 相关度）
5. 保持回答简洁、有条理，适当使用列表或小标题
6. 对于模糊或多义词问题，分别列出不同文档中的不同观点

目录结构

skills/local-doc-rag/
├── SKILL.md                          ← 本文件
├── scripts/
│   ├── index_documents.py           ← 文档索引脚本
│   ├── rag_pipeline.py               ← RAG 查询脚本（核心）
│   ├── evaluate_skill.py             ← 验证脚本
│   └── requirements.txt              ← Python 依赖
└── references/
    ├── model-selection.md            ← 模型选型参考
    └── openvino-guide.md             ← OpenVINO 优化指南

依赖安装

pip install \
  sentence-transformers \
  faiss-cpu \
  torch \
  PyPDF2 \
  python-docx \
  openai \
  transformers \
  accelerate \
  cross-fetch \
  rank-bm25  # 可选：混合检索（BM25 + 向量）

限制与注意事项

文档大小：单文件建议 <100MB，超大 PDF 会导致内存压力
并发限制：SiliconFlow API 有速率限制，大批量查询请加 --delay 0.5
Embedding 模型下载：首次运行需要网络下载模型（~90MB + ~220MB）
Chunk 丢失处理：如果某个 chunk 无法解析（图片 PDF），会跳过并在日志中记录

使用示例

用户：帮我建一个知识库，索引我桌面的文档

Agent：好的，我来扫描并索引你的文档。

python3 skills/local-doc-rag/scripts/index_documents.py \
  --input ~/Desktop/docs \
  --output ~/.openclaw/workspace/vectorstore/ \
  --recursive

索引完成：已处理 23 个文档，生成 847 个 chunks，存放在 ~/.openclaw/workspace/vectorstore/。

用户：这份合同里关于违约条款是怎么写的？

Agent：

python3 skills/local-doc-rag/scripts/rag_pipeline.py \
  --query "合同中关于违约条款是如何规定的？" \
  --show-citations

回答：

根据文档，违约条款的规定如下 [1] [2]：

违约金标准：一方违约时，需按合同总金额的 20% 支付违约金……
履约保函：合同签署后 5 个工作日内，乙方需提交合同金额 10% 的履约保函……
免责条款：因不可抗力导致无法履约，双方互不承担违约责任……

引用：

[1] ~/Desktop/docs/采购合同.pdf - 第5页 - 相关度 0.96
[2] ~/Desktop/docs/采购合同.pdf - 第7页 - 相关度 0.89