AI工具 企业知识库搭建｜附源码

在数字化办公逐渐深入的今天，企业内部每天都会产生大量知识资产：产品文档、项目方案、客户资料、制度流程、技术手册、会议纪要、培训材料、FAQ问答等。这些资料如果只是分散存放在个人电脑、网盘、企业微信、飞书、钉钉或各种系统中，就很容易出现“找不到、看不懂、用不上、传不下去”的问题。

而随着大语言模型的发展，企业知识库不再只是一个简单的文档管理系统，而是可以升级为一个具备自然语言问答能力的智能知识助手。员工可以像聊天一样提问，例如：

“公司报销流程是什么？”
“某个产品的技术参数有哪些？”
“上一版项目方案里客户最关注的问题是什么？”
“新员工入职需要完成哪些事项？”
“请根据已有资料生成一份客户答疑话术。”

这类能力背后的核心技术，通常被称为 RAG，即 Retrieval-Augmented Generation，中文一般翻译为“检索增强生成”。简单来说，就是先从企业知识库中检索出与问题最相关的资料片段，再把这些资料片段连同用户问题一起交给大模型，让大模型基于企业内部资料进行回答。

本文将从企业知识库的整体架构、技术选型、核心流程、源码实现、部署建议等方面，完整介绍如何搭建一个可落地的 AI 企业知识库系统。

一、为什么企业需要 AI 知识库？

传统企业知识管理常见的问题主要有以下几类。

1. 文档分散，查找成本高

很多企业的资料分散在不同平台中：

• Word、PDF、Excel、PPT 文件；
• 企业网盘；
• Wiki 系统；
• 邮件附件；
• 聊天群历史消息；
• CRM、ERP、OA 等业务系统；
• 本地共享文件夹。

当员工想找某个资料时，往往不知道应该去哪个系统搜索。即使找到相关文档，也可能需要打开多个文件逐一阅读，效率非常低。

2. 知识依赖个人经验

企业中很多关键知识存在于老员工脑中，例如客户偏好、项目经验、故障处理方法、售前方案模板等。如果这些知识没有沉淀下来，一旦人员离职或岗位调整，就可能造成经验断层。

3. 新员工培训周期长

新员工入职后，需要学习大量制度、产品、流程和业务知识。传统方式通常依赖人工培训和文档阅读，但文档内容多、结构复杂，新人很难快速找到自己需要的信息。

4. 文档有了，但不会用

企业并不缺文档，真正缺的是“让知识被准确调用”的能力。文档沉淀只是第一步，让员工可以通过自然语言快速获取答案，才是知识库真正产生价值的关键。

AI 企业知识库的价值就在于：把企业原本静态、分散、难检索的资料，变成一个可以对话、可以推理、可以辅助工作的智能系统。

二、AI 企业知识库的核心原理

一个典型的 AI 知识库系统主要包含以下几个步骤：

1. 文档上传；
2. 文档解析；
3. 文本切分；
4. 向量化；
5. 存入向量数据库；
6. 用户提问；
7. 相似度检索；
8. 构造 Prompt；
9. 调用大模型生成回答；
10. 返回答案和引用来源。

这个过程可以用一句话概括：

先把企业文档转换成可检索的向量，再在用户提问时找到相关内容，最后让大模型基于这些内容生成答案。

1. 什么是向量化？

大模型或 Embedding 模型可以把文本转换成一组数字，例如：

“公司报销流程” → [0.123, -0.456, 0.891, ...]

这些数字表示文本的语义特征。语义相近的文本，在向量空间中的距离也会更近。

比如：

• “如何申请报销？”
• “费用报销流程是什么？”
• “差旅费怎么走审批？”

虽然表述不同，但语义相近，因此它们对应的向量距离也会比较近。系统就可以通过向量相似度找到最相关的知识片段。

2. 为什么需要文本切分？

企业文档通常比较长，比如一份 PDF 可能有几十页。如果直接把整篇文档交给大模型，可能会遇到以下问题：

• 超出模型上下文长度限制；
• 检索不精准；
• 成本高；
• 回答容易混入无关内容。

因此需要把文档切成多个较小片段。例如每段 500～1000 个中文字符，并保留一定重叠内容，保证上下文连续性。

3. RAG 与微调有什么区别？

很多人会问：企业知识库为什么不用模型微调？

简单来说：

• RAG 更适合知识频繁变化、需要引用来源、成本较低的场景；
• 微调 更适合让模型学习特定风格、格式或任务能力，但不适合频繁注入大量企业资料。

企业制度、产品文档、项目资料经常更新，如果每次更新都微调模型，成本和复杂度都很高。而 RAG 只需要更新知识库索引即可，更适合大多数企业场景。

三、系统技术架构设计

本文示例采用一个轻量但完整的架构，适合中小企业或团队快速落地。

技术选型

当然，在生产环境中也可以替换为：

• Milvus、Qdrant、Weaviate、Elasticsearch Vector Search；
• PostgreSQL + pgvector；
• Redis Vector；
• 企业私有化大模型；
• LangChain、LlamaIndex 等框架。

本文为了便于理解，尽量不依赖复杂框架，而是手写核心逻辑。

四、项目目录结构

推荐目录如下：

ai-knowledge-base/
├── app.py                  # FastAPI 主服务
├── requirements.txt        # 依赖文件
├── data/
│   ├── uploads/            # 上传的原始文档
│   ├── index.faiss         # FAISS 向量索引
│   └── docs.json           # 文档片段元数据
├── static/
│   └── index.html          # 前端页面
└── README.md

五、安装依赖

创建 requirements.txt：

fastapi==0.115.0
uvicorn==0.30.6
python-multipart==0.0.9
pymupdf==1.24.9
python-docx==1.1.2
sentence-transformers==3.0.1
faiss-cpu==1.8.0.post1
numpy==1.26.4
requests==2.32.3

安装依赖：

pip install -r requirements.txt

如果你使用的是国内网络环境，可以指定镜像源：

pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple

六、核心源码实现

下面是一个完整可运行的简化版后端源码。它支持：

• 上传 PDF、DOCX、TXT；
• 自动解析文本；
• 文本切分；
• 生成向量；
• 存入 FAISS；
• 用户提问；
• 检索相关知识片段；
• 调用大模型生成回答；
• 返回引用来源。

创建 app.py：

import os
import json
import uuid
import shutil
from typing import List

import fitz
import docx
import faiss
import numpy as np
import requests

from fastapi import FastAPI, UploadFile, File
from fastapi.staticfiles import StaticFiles
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from sentence_transformers import SentenceTransformer


BASE_DIR = os.path.dirname(os.path.abspath(__file__))
DATA_DIR = os.path.join(BASE_DIR, "data")
UPLOAD_DIR = os.path.join(DATA_DIR, "uploads")
INDEX_PATH = os.path.join(DATA_DIR, "index.faiss")
DOCS_PATH = os.path.join(DATA_DIR, "docs.json")

os.makedirs(DATA_DIR, exist_ok=True)
os.makedirs(UPLOAD_DIR, exist_ok=True)

app = FastAPI(title="AI 企业知识库")

# 你也可以替换为 bge-small-zh-v1.5、bge-base-zh-v1.5 等中文向量模型
embedding_model = SentenceTransformer("sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2")
EMBEDDING_DIM = 384

docs = []

if os.path.exists(DOCS_PATH):
    with open(DOCS_PATH, "r", encoding="utf-8") as f:
        docs = json.load(f)

if os.path.exists(INDEX_PATH):
    index = faiss.read_index(INDEX_PATH)
else:
    index = faiss.IndexFlatIP(EMBEDDING_DIM)


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


def save_docs():
    with open(DOCS_PATH, "w", encoding="utf-8") as f:
        json.dump(docs, f, ensure_ascii=False, indent=2)


def save_index():
    faiss.write_index(index, INDEX_PATH)


def normalize_vectors(vectors: np.ndarray) -> np.ndarray:
    norms = np.linalg.norm(vectors, axis=1, keepdims=True)
    return vectors / np.clip(norms, 1e-12, None)


def embed_texts(texts: List[str]) -> np.ndarray:
    vectors = embedding_model.encode(texts, convert_to_numpy=True)
    vectors = vectors.astype("float32")
    vectors = normalize_vectors(vectors)
    return vectors


def read_pdf(path: str) -> str:
    text = ""
    pdf = fitz.open(path)
    for page in pdf:
        text += page.get_text() + "\n"
    return text


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


def read_txt(path: str) -> str:
    with open(path, "r", encoding="utf-8", errors="ignore") as f:
        return f.read()


def parse_document(path: str, filename: str) -> str:
    lower_name = filename.lower()
    if lower_name.endswith(".pdf"):
        return read_pdf(path)
    elif lower_name.endswith(".docx"):
        return read_docx(path)
    elif lower_name.endswith(".txt") or lower_name.endswith(".md"):
        return read_txt(path)
    else:
        raise ValueError("暂不支持该文件格式")


def split_text(text: str, chunk_size: int = 800, overlap: int = 120) -> List[str]:
    text = text.replace("\r\n", "\n").replace("\r", "\n")
    text = "\n".join([line.strip() for line in text.split("\n") if line.strip()])

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

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

        if chunk:
            chunks.append(chunk)

        if end >= length:
            break

        start = end - overlap

    return chunks


def call_llm(question: str, contexts: List[str]) -> str:
    """
    这里以 OpenAI 兼容接口为例。
    如果使用本地模型、企业私有模型或其他厂商模型，只需要替换该函数。
    """

    api_key = os.getenv("OPENAI_API_KEY", "")
    base_url = os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")
    model = os.getenv("OPENAI_MODEL", "gpt-4o-mini")

    context_text = "\n\n".join(
        [f"资料片段{i + 1}：\n{ctx}" for i, ctx in enumerate(contexts)]
    )

    prompt = f"""
你是企业内部知识库助手。请严格基于下面提供的企业资料回答用户问题。

要求：
1. 如果资料中有明确答案，请给出准确、简洁、结构化的回答；
2. 如果资料不足以回答，请说明“根据当前知识库资料，暂未找到明确答案”；
3. 不要编造企业内部不存在的信息；
4. 可以在回答中列出依据；
5. 回答使用中文。

企业资料：
{context_text}

用户问题：
{question}
"""

    if not api_key:
        return "未配置 OPENAI_API_KEY。已完成知识库检索，但无法调用大模型生成最终回答。"

    url = f"{base_url}/chat/completions"

    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }

    payload = {
        "model": model,
        "messages": [
            {"role": "system", "content": "你是一个严谨的企业知识库问答助手。"},
            {"role": "user", "content": prompt}
        ],
        "temperature": 0.2
    }

    response = requests.post(url, headers=headers, json=payload, timeout=60)
    response.raise_for_status()

    result = response.json()
    return result["choices"][0]["message"]["content"]


@app.post("/upload")
async def upload_file(file: UploadFile = File(...)):
    file_id = str(uuid.uuid4())
    save_name = f"{file_id}_{file.filename}"
    save_path = os.path.join(UPLOAD_DIR, save_name)

    with open(save_path, "wb") as buffer:
        shutil.copyfileobj(file.file, buffer)

    try:
        text = parse_document(save_path, file.filename)
        chunks = split_text(text)

        if not chunks:
            return JSONResponse({"message": "文档内容为空，无法入库"}, status_code=400)

        vectors = embed_texts(chunks)
        index.add(vectors)

        for i, chunk in enumerate(chunks):
            docs.append({
                "id": str(uuid.uuid4()),
                "filename": file.filename,
                "path": save_path,
                "chunk_index": i,
                "content": chunk
            })

        save_docs()
        save_index()

        return {
            "message": "上传并入库成功",
            "filename": file.filename,
            "chunks": len(chunks)
        }

    except Exception as e:
        return JSONResponse({"message": f"处理失败：{str(e)}"}, status_code=500)


@app.post("/ask")
async def ask(req: AskRequest):
    if index.ntotal == 0:
        return JSONResponse({"message": "知识库为空，请先上传文档"}, status_code=400)

    query_vector = embed_texts([req.question])
    scores, ids = index.search(query_vector, req.top_k)

    contexts = []
    references = []

    for score, idx in zip(scores[0], ids[0]):
        if idx == -1:
            continue

        item = docs[idx]
        contexts.append(item["content"])
        references.append({
            "filename": item["filename"],
            "chunk_index": item["chunk_index"],
            "score": float(score),
            "content": item["content"][:300]
        })

    answer = call_llm(req.question, contexts)

    return {
        "question": req.question,
        "answer": answer,
        "references": references
    }


@app.get("/documents")
async def list_documents():
    filenames = {}
    for item in docs:
        name = item["filename"]
        filenames[name] = filenames.get(name, 0) + 1

    return {
        "total_chunks": len(docs),
        "documents": [
            {"filename": name, "chunks": count}
            for name, count in filenames.items()
        ]
    }


app.mount("/", StaticFiles(directory="static", html=True), name="static")

七、前端页面源码

创建 static/index.html：

  
  
  


  
AI 企业知识库

  

    
上传文档
    
    

    上传并入库
    

  

  

    
知识库问答
    
    

    提交问题
    
回答
    

    
引用来源
    

  

  

    
文档列表
    刷新文档列表
    

  

  

八、启动项目

在项目根目录执行：

uvicorn app:app --host 0.0.0.0 --port 8000 --reload

浏览器访问：

http://localhost:8000

如果需要调用 OpenAI 兼容接口，需要配置环境变量：

export OPENAI_API_KEY="你的API_KEY"
export OPENAI_BASE_URL="https://api.openai.com/v1"
export OPENAI_MODEL="gpt-4o-mini"

Windows PowerShell 可以使用：

$env:OPENAI_API_KEY="你的API_KEY"
$env:OPENAI_BASE_URL="https://api.openai.com/v1"
$env:OPENAI_MODEL="gpt-4o-mini"

如果使用的是其他厂商的大模型，只要接口兼容 OpenAI 格式，通常只需要修改 OPENAI_BASE_URL 和 OPENAI_MODEL 即可。

九、关键功能说明

1. 文档解析

示例中支持 PDF、DOCX、TXT、MD。实际企业场景中，还可能需要支持：

• Excel；
• PPT；
• HTML；
• 扫描版 PDF；
• 图片；
• 飞书文档；
• 企业微信文档；
• 数据库记录；
• API 数据。

对于扫描版 PDF 和图片，需要接入 OCR，例如 PaddleOCR、Tesseract 或云厂商 OCR 服务。

2. 文本切分策略

本文代码采用固定长度切分，这种方式简单直接，但在复杂文档中可能会切断语义。生产环境中可以改进为：

• 按标题层级切分；
• 按段落切分；
• 按 Markdown 标题切分；
• 按 FAQ 问答对切分；
• 按表格行切分；
• 按业务字段切分。

一个好的切分策略，会显著影响知识库问答质量。

3. 相似度检索

示例使用 FAISS 的 IndexFlatIP，并对向量做归一化。归一化后，内积相似度近似等价于余弦相似度。

对于中小规模知识库，FAISS 本地索引足够使用。如果企业文档量达到百万级片段，可以考虑：

• Milvus；
• Qdrant；
• Elasticsearch；
• pgvector；
• 云厂商向量数据库。

4. Prompt 设计

Prompt 是 RAG 系统中非常关键的一环。一个好的 Prompt 应该明确告诉模型：

• 只能基于知识库资料回答；
• 资料不足时要承认不知道；
• 不要编造；
• 输出格式要清晰；
• 必要时给出引用依据。

本文示例中的 Prompt 已经包含这些要求，但在生产环境中还可以进一步针对企业场景优化。

十、企业级优化建议

如果要把这个系统真正用于企业内部，还需要考虑更多工程化能力。

1. 权限控制

企业资料通常有权限等级，例如：

• 全员可见；
• 部门可见；
• 管理层可见；
• 项目组可见；
• 仅个人可见。

因此知识库问答时，不能只做语义检索，还必须结合用户权限过滤。否则可能出现普通员工通过 AI 问答获取敏感资料的问题。

常见做法是在每个知识片段的元数据中加入：

{
  "department": "财务部",
  "permission": ["finance", "manager"],
  "owner": "user_001"
}

检索时先根据用户身份过滤，再进行向量召回或重排序。

2. 知识更新机制

企业文档会不断变化，知识库必须支持：

• 新增文档；
• 删除文档；
• 替换文档；
• 版本管理；
• 定时同步；
• 增量索引；
• 失效内容清理。

本文示例为了简化，只实现了追加入库。生产环境需要设计文档 ID、版本号和索引删除机制。

3. 回答可追溯

企业知识库不能只给答案，还要给依据。否则用户无法判断答案是否可靠。

建议每次回答都返回：

• 来源文件；
• 文档标题；
• 页码或章节；
• 原文片段；
• 相似度分数；
• 知识更新时间。

这样用户可以点击跳转原文，进一步核验。

4. 多轮对话

真实使用中，用户经常会连续追问。例如：

用户：销售合同审批流程是什么？
用户：如果金额超过 100 万呢？
用户：需要哪些附件？

这就需要系统保留会话历史，并在检索和生成时结合上下文。可以将最近几轮对话加入 Prompt，也可以对追问进行问题改写，把省略信息补全后再检索。

5. 检索增强

基础向量检索有时会遗漏关键词精确匹配内容。例如制度编号、产品型号、客户名称等。建议采用混合检索：

• 向量检索：解决语义相似；
• 关键词检索：解决精确匹配；
• 重排序模型：提高最终召回质量。

常见组合是：

BM25 关键词检索 + 向量检索 + Rerank 重排序

这样可以显著提升复杂问题的回答准确率。

十一、常见问题与解决方案

问题 1：模型回答胡编怎么办？

解决方法：

1. Prompt 中明确要求“资料不足则回答不知道”；
2. 限制模型只能基于检索内容回答；
3. 提高检索质量；
4. 设置较低 temperature；
5. 返回引用来源；
6. 对高风险问题增加人工确认。

问题 2：检索出来的内容不相关怎么办？

可以从以下方面优化：

• 更换更适合中文的 Embedding 模型；
• 调整 chunk_size 和 overlap；
• 使用混合检索；
• 加入 Rerank 模型；
• 清洗低质量文档；
• 针对标题、摘要、关键词建立额外索引。

问题 3：上传 PDF 后内容乱码怎么办？

可能原因包括：

• PDF 是扫描件，需要 OCR；
• PDF 字体编码特殊；
• 文档本身加密；
• 提取工具兼容性不足。

可以尝试更换解析库，或者对扫描件走 OCR 流程。

问题 4：企业数据安全如何保障？

建议：

• 优先私有化部署；
• 使用内网模型或私有大模型 API；
• 对敏感字段脱敏；
• 做访问权限控制；
• 记录审计日志；
• 对上传和问答内容加密存储；
• 限制外部接口调用范围。

十二、Docker 部署示例

可以创建 Dockerfile：

FROM python:3.10-slim

WORKDIR /app

COPY requirements.txt /app/requirements.txt

RUN pip install --no-cache-dir -r requirements.txt \
    -i https://pypi.tuna.tsinghua.edu.cn/simple

COPY . /app

EXPOSE 8000

CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]

构建镜像：

docker build -t ai-knowledge-base .

运行容器：

docker run -d \
  --name ai-kb \
  -p 8000:8000 \
  -e OPENAI_API_KEY="你的API_KEY" \
  -e OPENAI_BASE_URL="https://api.openai.com/v1" \
  -e OPENAI_MODEL="gpt-4o-mini" \
  -v $(pwd)/data:/app/data \
  ai-knowledge-base

访问：

http://服务器IP:8000

十三、适用业务场景

AI 企业知识库可以应用在很多实际业务中。

1. 内部制度问答

员工可以直接询问：

• 年假怎么申请？
• 报销需要哪些发票？
• 出差补贴标准是什么？
• 加班调休规则是什么？

2. 售前和客服支持

客服或销售可以快速查询：

• 产品功能说明；
• 价格政策；
• 常见客户异议；
• 竞品对比；
• 交付周期；
• 售后服务条款。

3. 技术支持

研发、运维和实施团队可以查询：

• 系统部署文档；
• 故障排查手册；
• API 接口说明；
• 数据库字段含义；
• 历史问题解决方案。

4. 项目管理

项目成员可以查询：

• 项目背景；
• 客户需求；
• 会议纪要；
• 风险事项；
• 任务分工；
• 验收标准。

5. 新员工助手

新员工可以通过知识库快速了解公司情况，包括组织架构、办公系统、制度流程、产品知识和岗位操作手册，大幅降低培训成本。

十四、后续可扩展方向

本文提供的是一个轻量级源码示例，如果要继续扩展，可以考虑以下方向：

1. 增加登录系统和用户权限；
2. 增加文档删除与重新索引；
3. 支持 Excel、PPT、图片和网页解析；
4. 接入 OCR；
5. 增加 Rerank 模型；
6. 支持多知识库空间；
7. 支持按部门、项目、标签分类；
8. 增加回答评价功能；
9. 增加问答日志和命中率统计；
10. 增加流式输出；
11. 接入企业微信、飞书、钉钉机器人；
12. 支持私有化大模型部署；
13. 增加知识冲突检测；
14. 支持文档自动同步和定时更新。

十五、总结

AI 企业知识库的本质，不是简单地把文档上传到系统里，而是让企业知识真正变成可检索、可问答、可复用、可追溯的智能资产。

本文通过 FastAPI、FAISS、Embedding 模型和大模型接口，实现了一个完整的企业知识库原型。它具备文档上传、文本解析、向量化、语义检索、智能问答和引用来源展示等核心能力。虽然示例代码相对简化，但已经覆盖了 RAG 知识库系统的主要流程。

对于企业来说，搭建 AI 知识库的关键不只在技术，还包括知识治理、权限管理、内容更新、数据安全和业务流程融合。只有将 AI 能力与企业实际知识体系结合起来，才能真正提升组织效率。

如果你是技术团队，可以基于本文源码快速搭建 Demo，再逐步补充权限、多知识库、重排序、日志审计、私有化模型等能力；如果你是业务团队，也可以先从高频问答、制度查询、客服知识库、项目文档助手等场景切入，逐步扩大使用范围。

AI 企业知识库不是一次性项目，而是企业知识管理升级的长期工程。越早沉淀知识、越早建立智能检索能力，企业在协作效率、经验复用和组织学习方面就越容易形成优势。