Claude 企业知识库搭建｜附完整命令

在企业内部，文档、制度、项目资料、产品手册、客户案例、会议纪要、研发规范往往分散在飞书、钉钉、Confluence、SharePoint、网盘、Git 仓库以及各种 PDF、Word、Markdown 文件中。员工想要查找一个准确答案，经常需要跨多个系统搜索，效率低、信息不一致，还容易遗漏关键上下文。

基于 Claude 搭建企业知识库，可以把企业内部文档转化为可检索、可问答、可追溯来源的智能知识系统。员工只需要用自然语言提问，例如：

“我们公司报销差旅费的标准是什么？”
“某产品的 API 限流规则在哪里？”
“给我总结一下 Q3 客户反馈中的主要问题。”
“根据内部销售手册，如何向金融行业客户介绍我们的方案？”

系统会先从企业知识库中检索相关资料，再把检索结果作为上下文交给 Claude，由 Claude 生成结构化回答，并附带引用来源。这类架构通常被称为 RAG：Retrieval-Augmented Generation，检索增强生成。

本文将从零开始，带你搭建一个可运行的 Claude 企业知识库系统，并提供完整命令。

一、整体架构说明

一个基础的 Claude 企业知识库通常包含以下模块：

企业文档
  ↓
文档解析
  ↓
文本切分 Chunking
  ↓
向量化 Embedding
  ↓
向量数据库 Qdrant
  ↓
用户提问
  ↓
问题向量化
  ↓
相似内容检索
  ↓
Claude 生成回答
  ↓
返回答案 + 引用来源

本文采用的技术栈如下：

说明：Claude 主要负责理解、总结、推理和生成答案；文档向量化部分本文使用本地中文 Embedding 模型，适合企业私有化或半私有化部署场景。

二、准备环境

本文默认你已经安装：

• macOS / Linux / WSL
• Python 3.10+
• Docker
• Docker Compose
• Anthropic Claude API Key

检查基础环境：

python3 --version
docker --version
docker compose version

如果你还没有安装 uv，可以执行：

curl -LsSf https://astral.sh/uv/install.sh | sh

安装完成后，重新加载终端配置：

source ~/.bashrc

如果你使用的是 zsh：

source ~/.zshrc

检查 uv：

uv --version

三、创建项目目录

执行以下命令创建项目：

mkdir claude-enterprise-kb
cd claude-enterprise-kb

初始化 Python 项目：

uv init

创建必要目录：

mkdir -p app data docs scripts

目录结构如下：

claude-enterprise-kb/
├── app/
│   ├── main.py
│   ├── config.py
│   ├── ingest.py
│   ├── rag.py
│   └── splitter.py
├── data/
├── docs/
├── scripts/
├── docker-compose.yml
├── pyproject.toml
└── .env

四、安装依赖

执行：

uv add fastapi uvicorn qdrant-client sentence-transformers anthropic python-dotenv pypdf python-docx markdown beautifulsoup4

这些依赖分别用于：

五、启动 Qdrant 向量数据库

创建 docker-compose.yml：

cat > docker-compose.yml <<'EOF'
services:
  qdrant:
    image: qdrant/qdrant:latest
    container_name: claude-enterprise-qdrant
    ports:
      - "6333:6333"
      - "6334:6334"
    volumes:
      - ./data/qdrant:/qdrant/storage
    environment:
      QDRANT__SERVICE__HTTP_PORT: 6333
      QDRANT__SERVICE__GRPC_PORT: 6334
EOF

启动 Qdrant：

docker compose up -d

查看容器状态：

docker ps

访问 Qdrant 控制台：

http://localhost:6333/dashboard

如果能打开页面，说明向量数据库已经启动成功。

六、配置 Claude API Key

创建 .env 文件：

cat > .env <<'EOF'
ANTHROPIC_API_KEY=你的_Claude_API_Key
CLAUDE_MODEL=claude-3-5-sonnet-20240620
QDRANT_HOST=localhost
QDRANT_PORT=6333
QDRANT_COLLECTION=enterprise_kb
EMBEDDING_MODEL=BAAI/bge-small-zh-v1.5
EOF

请将：

你的_Claude_API_Key

替换为你自己的 Anthropic API Key。

注意：Claude 模型名称可能会随官方更新变化。如果调用失败，请以 Anthropic 官方控制台中显示的最新模型名称为准。

七、编写配置文件

创建 app/config.py：

cat > app/config.py <<'EOF'
import os
from dotenv import load_dotenv

load_dotenv()

ANTHROPIC_API_KEY = os.getenv("ANTHROPIC_API_KEY")
CLAUDE_MODEL = os.getenv("CLAUDE_MODEL", "claude-3-5-sonnet-20240620")

QDRANT_HOST = os.getenv("QDRANT_HOST", "localhost")
QDRANT_PORT = int(os.getenv("QDRANT_PORT", "6333"))
QDRANT_COLLECTION = os.getenv("QDRANT_COLLECTION", "enterprise_kb")

EMBEDDING_MODEL = os.getenv("EMBEDDING_MODEL", "BAAI/bge-small-zh-v1.5")
EOF

八、编写文本切分模块

企业文档往往很长，不能一次性全部塞给 Claude。我们需要先把文档切成较小的片段，每个片段称为一个 Chunk。

创建 app/splitter.py：

cat > app/splitter.py <<'EOF'
from typing import List


def split_text(text: str, chunk_size: int = 700, overlap: int = 120) -> List[str]:
    """
    简单文本切分器：
    - chunk_size：每段最大字符数
    - overlap：相邻片段重叠字符数，用于保留上下文
    """
    text = text.replace("\r\n", "\n").replace("\r", "\n")
    text = "\n".join([line.strip() for line in text.split("\n") if line.strip()])

    if not text:
        return []

    chunks = []
    start = 0
    length = len(text)

    while start < length:
        end = min(start + chunk_size, length)
        chunk = text[start:end]
        chunks.append(chunk)

        if end >= length:
            break

        start = max(0, end - overlap)

    return chunks
EOF

这个切分器比较简单，但足够完成第一版知识库。生产环境中，可以升级为按标题、段落、目录层级、语义边界进行切分。

九、编写文档导入程序

创建 app/ingest.py：

cat > app/ingest.py <<'EOF'
import os
import uuid
from pathlib import Path
from typing import List, Dict

from pypdf import PdfReader
from docx import Document
from bs4 import BeautifulSoup
import markdown

from qdrant_client import QdrantClient
from qdrant_client.models import VectorParams, Distance, PointStruct
from sentence_transformers import SentenceTransformer

from app.config import (
    QDRANT_HOST,
    QDRANT_PORT,
    QDRANT_COLLECTION,
    EMBEDDING_MODEL,
)
from app.splitter import split_text


def read_txt(path: Path) -> str:
    return path.read_text(encoding="utf-8", errors="ignore")


def read_pdf(path: Path) -> str:
    reader = PdfReader(str(path))
    pages = []
    for page in reader.pages:
        pages.append(page.extract_text() or "")
    return "\n".join(pages)


def read_docx(path: Path) -> str:
    doc = Document(str(path))
    return "\n".join([p.text for p in doc.paragraphs if p.text.strip()])


def read_md(path: Path) -> str:
    raw = path.read_text(encoding="utf-8", errors="ignore")
    html = markdown.markdown(raw)
    soup = BeautifulSoup(html, "html.parser")
    return soup.get_text("\n")


def load_document(path: Path) -> str:
    suffix = path.suffix.lower()

    if suffix in [".txt"]:
        return read_txt(path)

    if suffix in [".md", ".markdown"]:
        return read_md(path)

    if suffix == ".pdf":
        return read_pdf(path)

    if suffix == ".docx":
        return read_docx(path)

    raise ValueError(f"暂不支持该文件类型: {path}")


def collect_files(folder: str) -> List[Path]:
    supported = {".txt", ".md", ".markdown", ".pdf", ".docx"}
    base = Path(folder)
    files = []

    for path in base.rglob("*"):
        if path.is_file() and path.suffix.lower() in supported:
            files.append(path)

    return files


def main():
    docs_dir = "docs"

    print(f"正在加载向量模型: {EMBEDDING_MODEL}")
    model = SentenceTransformer(EMBEDDING_MODEL)
    vector_size = model.get_sentence_embedding_dimension()

    print(f"连接 Qdrant: {QDRANT_HOST}:{QDRANT_PORT}")
    client = QdrantClient(host=QDRANT_HOST, port=QDRANT_PORT)

    collections = client.get_collections().collections
    exists = any(c.name == QDRANT_COLLECTION for c in collections)

    if not exists:
        print(f"创建集合: {QDRANT_COLLECTION}")
        client.create_collection(
            collection_name=QDRANT_COLLECTION,
            vectors_config=VectorParams(
                size=vector_size,
                distance=Distance.COSINE,
            ),
        )

    files = collect_files(docs_dir)
    print(f"发现文档数量: {len(files)}")

    points = []

    for file_path in files:
        print(f"处理文档: {file_path}")
        try:
            text = load_document(file_path)
        except Exception as e:
            print(f"读取失败: {file_path}, 错误: {e}")
            continue

        chunks = split_text(text)

        for index, chunk in enumerate(chunks):
            vector = model.encode(chunk, normalize_embeddings=True).tolist()

            payload: Dict = {
                "source": str(file_path),
                "chunk_index": index,
                "text": chunk,
            }

            point = PointStruct(
                id=str(uuid.uuid4()),
                vector=vector,
                payload=payload,
            )
            points.append(point)

    if not points:
        print("没有可写入的内容。请先把文档放入 docs/ 目录。")
        return

    print(f"写入向量数量: {len(points)}")
    client.upsert(
        collection_name=QDRANT_COLLECTION,
        points=points,
    )

    print("知识库导入完成。")


if __name__ == "__main__":
    main()
EOF

十、准备测试文档

你可以先创建一个示例文档：

cat > docs/company_policy.md <<'EOF'
# 公司差旅报销制度

## 一、适用范围

本制度适用于公司所有正式员工、试用期员工以及经批准参与公司项目的外部顾问。

## 二、交通标准

员工因公出差可以乘坐高铁二等座、动车二等座或经济舱机票。
部门总监及以上级别人员，经审批后可乘坐高铁一等座。
如遇紧急客户项目，经直属负责人和财务负责人共同审批后，可选择更高等级交通方式。

## 三、住宿标准

一线城市住宿标准不超过每晚 600 元。
新一线及省会城市住宿标准不超过每晚 500 元。
其他城市住宿标准不超过每晚 400 元。

超出标准的部分，原则上由员工个人承担；如因展会、客户指定酒店等特殊情况导致超标，需提供说明并经部门负责人审批。

## 四、餐补标准

员工出差期间，每人每天餐补 100 元。
若客户或会议主办方已提供餐食，则对应餐次不再重复报销。

## 五、报销时限

员工应在出差结束后 15 个自然日内提交报销申请。
逾期超过 30 个自然日未提交的，原则上不予报销，特殊情况需由部门负责人说明原因。
EOF

也可以把真实企业文档放入 docs/ 目录，支持：

.txt
.md
.markdown
.pdf
.docx

十一、导入文档到知识库

执行：

uv run python -m app.ingest

首次运行时会下载 Embedding 模型，耗时取决于网络环境。

看到类似输出即表示成功：

正在加载向量模型: BAAI/bge-small-zh-v1.5
连接 Qdrant: localhost:6333
创建集合: enterprise_kb
发现文档数量: 1
处理文档: docs/company_policy.md
写入向量数量: 3
知识库导入完成。

你也可以进入 Qdrant Dashboard 查看集合和向量数据。

十二、编写 RAG 问答逻辑

创建 app/rag.py：

cat > app/rag.py <<'EOF'
from typing import List, Dict, Any

from anthropic import Anthropic
from qdrant_client import QdrantClient
from sentence_transformers import SentenceTransformer

from app.config import (
    ANTHROPIC_API_KEY,
    CLAUDE_MODEL,
    QDRANT_HOST,
    QDRANT_PORT,
    QDRANT_COLLECTION,
    EMBEDDING_MODEL,
)


embedding_model = SentenceTransformer(EMBEDDING_MODEL)
qdrant = QdrantClient(host=QDRANT_HOST, port=QDRANT_PORT)
claude = Anthropic(api_key=ANTHROPIC_API_KEY)


def search_knowledge_base(query: str, top_k: int = 5) -> List[Dict[str, Any]]:
    query_vector = embedding_model.encode(
        query,
        normalize_embeddings=True,
    ).tolist()

    results = qdrant.search(
        collection_name=QDRANT_COLLECTION,
        query_vector=query_vector,
        limit=top_k,
    )

    docs = []
    for item in results:
        payload = item.payload or {}
        docs.append({
            "score": item.score,
            "source": payload.get("source"),
            "chunk_index": payload.get("chunk_index"),
            "text": payload.get("text"),
        })

    return docs


def build_context(docs: List[Dict[str, Any]]) -> str:
    blocks = []

    for i, doc in enumerate(docs, start=1):
        source = doc.get("source")
        chunk_index = doc.get("chunk_index")
        text = doc.get("text")

        blocks.append(
            f"[资料{i}]\n"
            f"来源: {source}\n"
            f"片段: {chunk_index}\n"
            f"内容:\n{text}"
        )

    return "\n\n".join(blocks)


def ask_claude(question: str, docs: List[Dict[str, Any]]) -> str:
    context = build_context(docs)

    system_prompt = """
你是企业内部知识库助手。
你的任务是基于提供的企业资料回答员工问题。

必须遵守以下规则：
1. 优先依据资料内容回答，不要编造。
2. 如果资料中没有答案，请明确说明“知识库中未找到相关依据”。
3. 回答要清晰、准确、结构化。
4. 涉及制度、金额、时间、权限、流程时，要尽量引用原文依据。
5. 最后给出“参考来源”，列出用到的资料编号和文件来源。
"""

    user_prompt = f"""
以下是从企业知识库检索到的资料：

{context}

员工问题：
{question}

请基于以上资料回答。
"""

    response = claude.messages.create(
        model=CLAUDE_MODEL,
        max_tokens=1200,
        temperature=0.2,
        system=system_prompt,
        messages=[
            {
                "role": "user",
                "content": user_prompt,
            }
        ],
    )

    return response.content[0].text


def answer_question(question: str, top_k: int = 5) -> Dict[str, Any]:
    docs = search_knowledge_base(question, top_k=top_k)
    answer = ask_claude(question, docs)

    return {
        "question": question,
        "answer": answer,
        "references": docs,
    }
EOF

十三、编写 API 服务

创建 app/main.py：

cat > app/main.py <<'EOF'
from fastapi import FastAPI
from pydantic import BaseModel

from app.rag import answer_question, search_knowledge_base


app = FastAPI(
    title="Claude 企业知识库",
    description="基于 Claude + Qdrant + FastAPI 的企业 RAG 知识库服务",
    version="1.0.0",
)


class AskRequest(BaseModel):
    question: str
    top_k: int = 5


@app.get("/health")
def health():
    return {
        "status": "ok",
    }


@app.post("/ask")
def ask(req: AskRequest):
    return answer_question(req.question, req.top_k)


@app.post("/search")
def search(req: AskRequest):
    return {
        "question": req.question,
        "results": search_knowledge_base(req.question, req.top_k),
    }
EOF

十四、启动服务

执行：

uv run uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload

访问接口文档：

http://localhost:8000/docs

健康检查：

curl http://localhost:8000/health

预期返回：

{"status":"ok"}

十五、测试知识库问答

执行：

curl -X POST http://localhost:8000/ask \
  -H "Content-Type: application/json" \
  -d '{
    "question": "员工出差住宿报销标准是多少？",
    "top_k": 5
  }'

可能返回类似结果：

{
  "question": "员工出差住宿报销标准是多少？",
  "answer": "根据公司差旅报销制度，员工出差住宿标准如下：\n\n1. 一线城市：每晚不超过 600 元。\n2. 新一线及省会城市：每晚不超过 500 元。\n3. 其他城市：每晚不超过 400 元。\n\n如果超出标准，原则上由员工个人承担；如果因为展会、客户指定酒店等特殊情况导致超标，需要提供说明并经部门负责人审批。\n\n参考来源：\n- 资料1：docs/company_policy.md",
  "references": [
    {
      "score": 0.78,
      "source": "docs/company_policy.md",
      "chunk_index": 0,
      "text": "..."
    }
  ]
}

再测试一个流程类问题：

curl -X POST http://localhost:8000/ask \
  -H "Content-Type: application/json" \
  -d '{
    "question": "出差结束后多久需要提交报销？逾期怎么办？",
    "top_k": 5
  }'

十六、常用运维命令

查看 Qdrant 容器：

docker ps

停止服务：

docker compose down

重新启动：

docker compose up -d

查看 Qdrant 日志：

docker logs -f claude-enterprise-qdrant

重新导入知识库：

uv run python -m app.ingest

启动 API：

uv run uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload

查看当前项目依赖：

uv pip list

删除本地 Qdrant 数据，重新初始化：

docker compose down
rm -rf data/qdrant
docker compose up -d
uv run python -m app.ingest

十七、生产环境需要重点优化的地方

上面的系统已经可以跑通企业知识库的基本链路，但如果要真正用于生产环境，还需要进一步加强。

1. 权限控制

企业知识库最重要的问题不是“能不能答”，而是“该不该答”。

例如：

• 普通员工不能查询高管薪酬制度；
• 销售不能查询研发未发布路线图；
• 外包人员不能查询客户合同；
• 不同部门只能访问对应知识范围。

生产环境中需要在文档入库时增加权限字段，例如：

{
  "department": "finance",
  "access_level": "internal",
  "allowed_roles": ["finance_manager", "hr_admin"]
}

检索时根据当前登录用户身份过滤：

只检索用户有权限访问的文档片段

否则即使 Claude 不主动泄露，检索层也可能把敏感内容传给模型。

2. 文档增量更新

本文示例是全量导入。真实企业中，文档会不断变化，需要支持：

• 新增文档自动入库；
• 修改文档自动更新；
• 删除文档同步删除向量；
• 保留文档版本；
• 支持回滚历史版本。

常见做法是为每个文档计算哈希值：

sha256sum docs/company_policy.md

如果哈希没有变化，就不重新切分和向量化；如果哈希变化，则删除旧向量并重新写入。

3. 更好的文本切分策略

简单按字符切分容易破坏语义，例如把一个制度条款切成两半。生产环境建议按照文档结构切分：

• Markdown 按标题层级切；
• PDF 按页码和段落切；
• Word 按标题、表格、段落切；
• API 文档按接口切；
• 合同文档按条款切；
• FAQ 按问答对切。

更好的 Chunk 会明显提升检索准确率。

4. 引用来源和可追溯性

企业知识库必须支持答案溯源。每个回答最好都能展示：

• 文件名；
• 文档版本；
• 页码；
• 标题路径；
• 片段编号；
• 原文内容；
• 更新时间；
• 负责人。

否则员工很难判断答案是否可靠。

5. 防止模型幻觉

Claude 的推理和表达能力很强，但只要是大模型，就可能在上下文不足时生成看似合理但没有依据的内容。

可以通过以下方式降低风险：

1. 系统提示词中明确要求“无依据则回答不知道”；
2. 检索分数低于阈值时，不调用模型直接提示未找到；
3. 回答中强制输出引用来源；
4. 对金额、日期、审批权限等敏感字段做规则校验；
5. 对高风险问题增加人工确认流程。

例如可以设置：

如果最高检索分数低于 0.45，则返回“知识库中未找到足够相关的资料”

6. 日志与审计

企业场景中，需要记录：

• 谁问了什么问题；
• 检索到了哪些资料；
• Claude 返回了什么答案；
• 是否命中敏感内容；
• 是否被用户采纳；
• 是否有人工纠错。

这些日志可以用于：

• 安全审计；
• 知识库优化；
• 热点问题分析；
• 发现文档缺口；
• 改进问答质量。

十八、可扩展方向

当基础版本稳定后，可以继续扩展：

1. 接入企业 IM

可以把知识库接入：

• 飞书机器人；
• 企业微信机器人；
• 钉钉机器人；
• Slack；
• Microsoft Teams。

员工无需打开单独系统，在聊天工具中直接提问即可。

2. 接入企业文档系统

可以同步以下数据源：

• Confluence；
• Notion；
• 飞书云文档；
• 语雀；
• SharePoint；
• Google Drive；
• GitLab / GitHub；
• 企业网盘；
• CRM；
• 工单系统。

3. 增加混合检索

仅使用向量检索有时不够，尤其是涉及：

• 产品型号；
• 合同编号；
• 客户名称；
• API 参数；
• 错误码；
• 专有名词。

建议增加关键词检索，例如 Elasticsearch / OpenSearch，再与向量检索结果融合，形成混合检索：

最终结果 = 向量召回 + 关键词召回 + 重排序

4. 增加重排序模型

向量检索召回后，可以使用 reranker 对候选结果重新排序，提高上下文质量。对于企业制度、合同、技术文档等场景，重排序通常能明显改善答案准确率。

5. 多轮对话

基础版每次只处理一个问题。实际使用中，员工经常连续追问：

Q：住宿标准是多少？
A：一线城市 600 元，新一线及省会 500 元，其他城市 400 元。

Q：如果超标怎么办？
A：……

这就需要引入会话上下文，但要注意不能无限累积历史消息，否则会增加成本和误导检索。更合理的做法是：

• 保留最近几轮对话；
• 将追问改写成完整问题；
• 再进行知识库检索。

十九、总结

本文完成了一个基于 Claude 的企业知识库最小可用版本，包含：

• Qdrant 向量数据库；
• 企业文档解析；
• 文本切分；
• 本地中文 Embedding；
• 文档向量入库；
• Claude RAG 问答；
• FastAPI 接口；
• 完整命令和代码。

核心流程可以概括为：

企业文档 → 切分 → 向量化 → 入库 → 检索 → Claude 生成回答 → 返回引用来源

如果只是内部试点，本文方案已经足够用于验证价值；如果要用于正式生产，还需要重点补齐权限控制、文档同步、审计日志、检索优化、答案评估和安全治理。

Claude 擅长复杂语义理解、长文本总结、结构化表达和多步骤推理，结合企业内部知识库后，可以显著降低员工查资料成本，提高组织知识流转效率。对于文档密集型企业来说，这类系统很可能会成为未来内部办公的重要基础设施。