AI编程 企业知识库搭建｜附源码

在企业数字化转型过程中，知识管理一直是一个高频但又长期难以真正落地的问题。很多企业内部沉淀了大量文档：产品手册、制度文件、项目资料、会议纪要、售前方案、研发文档、运维手册、客服问答、合同模板等。这些资料往往分散在网盘、飞书、钉钉、企业微信、Confluence、语雀、本地文件夹甚至个人电脑中。

当员工需要查找某个信息时，常见场景是：

• 不知道资料在哪里；
• 找到了文档，但内容太长，无法快速定位答案；
• 文档版本混乱，不确定哪个是最新的；
• 新员工入职后，需要大量时间学习历史资料；
• 客服、售前、运营等岗位反复咨询同类问题；
• 企业知识依赖个人经验，一旦人员流动，知识随之流失。

随着大语言模型和向量数据库的发展，企业可以搭建一个基于 AI 的知识库问答系统。用户只需要用自然语言提问，系统就可以从企业内部文档中检索相关内容，并结合大模型生成准确、可读、可追溯的回答。

本文将从架构设计、技术选型、实现流程到源码示例，完整介绍如何搭建一个简易版企业 AI 知识库系统。

一、什么是企业 AI 知识库？

企业 AI 知识库可以简单理解为：

把企业内部的文档、资料、经验、流程等内容统一接入系统，通过 AI 检索和问答能力，让员工可以像聊天一样获取企业知识。

它不是简单的全文搜索，也不是传统的 FAQ 系统，而是结合了以下能力：

1. 文档解析

   • 支持 PDF、Word、Markdown、TXT、网页等多种格式；
   • 将文档内容提取为纯文本。

2. 文本切分

   • 将大文档切成多个小片段；
   • 每个片段保留语义完整性，便于后续检索。

3. 向量化

   • 使用 Embedding 模型将文本转换为向量；
   • 向量可以表示文本语义。

4. 向量检索

   • 用户提问后，也将问题转换为向量；
   • 在向量数据库中查找语义最相近的知识片段。

5. 大模型生成回答

   • 将检索到的知识片段作为上下文；
   • 让大语言模型基于上下文生成回答；
   • 尽量避免模型胡编乱造。

这种技术路线通常被称为 RAG，全称是 Retrieval-Augmented Generation，即检索增强生成。

二、为什么企业需要搭建 AI 知识库？

1. 降低信息查找成本

传统知识管理系统依赖关键词搜索。如果用户不知道准确关键词，就很难找到想要的内容。

比如用户问：

“员工转正流程怎么走？”

而文档标题可能叫：

《试用期员工考核与录用管理办法》

传统搜索可能无法很好匹配，而 AI 知识库可以理解“转正流程”和“试用期考核录用”在语义上相关，从而找到对应内容。

2. 提升员工工作效率

企业内部很多问题是重复性的，例如：

• 报销需要什么材料？
• 产品 A 支持哪些接口？
• 客户投诉如何处理？
• 某个系统部署步骤是什么？
• 项目上线前需要哪些审批？
• 新员工入职流程有哪些？

如果这些问题都需要人工询问同事或主管，就会消耗大量沟通成本。AI 知识库可以作为企业内部的“智能助手”，快速响应常见问题。

3. 沉淀组织经验

企业真正有价值的知识，往往存在于项目复盘、客户案例、技术方案、售前材料和专家经验中。AI 知识库可以将这些内容结构化、可检索化，让经验不再只存在于个人脑海中。

4. 支持客服、售前、运营等业务场景

企业知识库不仅可以服务内部员工，也可以服务外部客户。

例如：

• 客服机器人基于产品知识库回答用户问题；
• 售前顾问通过 AI 快速生成解决方案初稿；
• 运营人员快速查询活动规则；
• 技术支持人员快速定位故障处理方案。

三、系统整体架构

一个基础的企业 AI 知识库系统可以分为以下几个模块：

                 ┌────────────────────┐
                 │    企业文档数据源    │
                 │ PDF/Word/Markdown   │
                 └─────────┬──────────┘
                           │
                           ▼
                 ┌────────────────────┐
                 │      文档解析        │
                 │ 提取文本、清洗内容    │
                 └─────────┬──────────┘
                           │
                           ▼
                 ┌────────────────────┐
                 │      文本切分        │
                 │ chunk 分块处理       │
                 └─────────┬──────────┘
                           │
                           ▼
                 ┌────────────────────┐
                 │     Embedding 向量化 │
                 │ 文本转语义向量       │
                 └─────────┬──────────┘
                           │
                           ▼
                 ┌────────────────────┐
                 │      向量数据库      │
                 │ FAISS/Chroma/Milvus │
                 └─────────┬──────────┘
                           │
                           ▼
用户提问 ───────► 语义检索 ───────► 召回相关片段
                           │
                           ▼
                 ┌────────────────────┐
                 │      大模型生成      │
                 │ 基于上下文回答问题   │
                 └─────────┬──────────┘
                           │
                           ▼
                       返回答案

四、技术选型

为了便于演示，本文采用 Python 实现一个轻量版本。

后端技术

为什么选择 FAISS？

FAISS 是 Meta 开源的向量检索库，适合本地快速搭建原型。它的优点是：

• 部署简单；
• 检索速度快；
• 不依赖外部数据库；
• 适合中小规模知识库验证。

如果是生产环境，建议考虑：

• Milvus；
• Qdrant；
• Elasticsearch 向量检索；
• pgvector；
• Chroma；
• Weaviate。

五、项目目录结构

下面是一个简易版企业知识库项目结构：

ai-knowledge-base/
├── app.py                 # FastAPI 主服务
├── ingest.py              # 文档入库脚本
├── requirements.txt       # 依赖包
├── config.py              # 配置文件
├── data/                  # 原始文档目录
│   ├── employee_policy.txt
│   └── product_manual.txt
├── storage/               # 向量库和元数据存储目录
│   ├── index.faiss
│   └── docs.pkl
└── README.md

六、安装依赖

创建 requirements.txt：

fastapi==0.115.0
uvicorn==0.30.6
faiss-cpu==1.8.0.post1
sentence-transformers==3.0.1
numpy==1.26.4
pydantic==2.8.2
python-multipart==0.0.9
openai==1.40.0
pypdf==4.3.1
python-docx==1.1.2

安装依赖：

pip install -r requirements.txt

七、配置文件

创建 config.py：

import os

# 文档目录
DATA_DIR = os.getenv("DATA_DIR", "data")

# 向量存储目录
STORAGE_DIR = os.getenv("STORAGE_DIR", "storage")

# FAISS 索引文件
FAISS_INDEX_PATH = os.path.join(STORAGE_DIR, "index.faiss")

# 文档元数据文件
DOCS_PATH = os.path.join(STORAGE_DIR, "docs.pkl")

# Embedding 模型
EMBEDDING_MODEL_NAME = os.getenv(
    "EMBEDDING_MODEL_NAME",
    "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
)

# 大模型配置：可使用 OpenAI 或兼容 OpenAI 协议的模型服务
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "")
OPENAI_BASE_URL = os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")
CHAT_MODEL = os.getenv("CHAT_MODEL", "gpt-4o-mini")

这里的 Embedding 模型选择了一个支持多语言的轻量模型，适合中文企业文档场景。生产环境中可以换成效果更好的中文向量模型，例如：

• bge-large-zh；
• bge-m3；
• text2vec；
• m3e；
• OpenAI text-embedding 系列；
• 其他国产模型厂商提供的 embedding 服务。

八、文档解析与文本切分

创建 ingest.py，用于读取企业文档并构建向量索引。

import os
import pickle
from typing import List, Dict

import faiss
import numpy as np
from pypdf import PdfReader
from docx import Document
from sentence_transformers import SentenceTransformer

from config import (
    DATA_DIR,
    STORAGE_DIR,
    FAISS_INDEX_PATH,
    DOCS_PATH,
    EMBEDDING_MODEL_NAME,
)


def read_txt(file_path: str) -> str:
    """读取 txt 或 markdown 文件"""
    with open(file_path, "r", encoding="utf-8") as f:
        return f.read()


def read_pdf(file_path: str) -> str:
    """读取 PDF 文件"""
    reader = PdfReader(file_path)
    texts = []
    for page in reader.pages:
        text = page.extract_text() or ""
        texts.append(text)
    return "\n".join(texts)


def read_docx(file_path: str) -> str:
    """读取 Word 文档"""
    doc = Document(file_path)
    texts = []
    for p in doc.paragraphs:
        if p.text.strip():
            texts.append(p.text.strip())
    return "\n".join(texts)


def load_document(file_path: str) -> str:
    """根据文件类型加载文档"""
    ext = os.path.splitext(file_path)[1].lower()

    if ext in [".txt", ".md"]:
        return read_txt(file_path)
    elif ext == ".pdf":
        return read_pdf(file_path)
    elif ext == ".docx":
        return read_docx(file_path)
    else:
        raise ValueError(f"暂不支持的文件类型: {ext}")


def split_text(text: str, chunk_size: int = 500, overlap: int = 80) -> List[str]:
    """
    简单文本切分函数。
    chunk_size：每个文本块最大长度
    overlap：相邻文本块重叠字符数，避免上下文断裂
    """
    text = text.replace("\r\n", "\n").replace("\r", "\n")
    paragraphs = [p.strip() for p in text.split("\n") if p.strip()]

    chunks = []
    current = ""

    for para in paragraphs:
        if len(current) + len(para) <= chunk_size:
            current += "\n" + para if current else para
        else:
            if current:
                chunks.append(current)

            if len(para) > chunk_size:
                start = 0
                while start < len(para):
                    end = start + chunk_size
                    chunks.append(para[start:end])
                    start = end - overlap
            else:
                current = para

    if current:
        chunks.append(current)

    return chunks


def build_index():
    """构建向量索引"""
    os.makedirs(STORAGE_DIR, exist_ok=True)

    model = SentenceTransformer(EMBEDDING_MODEL_NAME)

    docs: List[Dict] = []

    for root, _, files in os.walk(DATA_DIR):
        for file in files:
            file_path = os.path.join(root, file)
            ext = os.path.splitext(file_path)[1].lower()

            if ext not in [".txt", ".md", ".pdf", ".docx"]:
                continue

            print(f"正在处理文档: {file_path}")

            try:
                content = load_document(file_path)
                chunks = split_text(content)

                for idx, chunk in enumerate(chunks):
                    docs.append({
                        "source": file_path,
                        "chunk_id": idx,
                        "content": chunk
                    })
            except Exception as e:
                print(f"处理失败: {file_path}, 错误: {e}")

    if not docs:
        raise RuntimeError("没有可入库的文档，请检查 data 目录。")

    texts = [doc["content"] for doc in docs]

    print("正在生成向量...")
    embeddings = model.encode(
        texts,
        normalize_embeddings=True,
        show_progress_bar=True
    )

    embeddings = np.array(embeddings).astype("float32")
    dim = embeddings.shape[1]

    index = faiss.IndexFlatIP(dim)
    index.add(embeddings)

    faiss.write_index(index, FAISS_INDEX_PATH)

    with open(DOCS_PATH, "wb") as f:
        pickle.dump(docs, f)

    print(f"入库完成，共写入 {len(docs)} 个文本块。")
    print(f"FAISS 索引保存到: {FAISS_INDEX_PATH}")
    print(f"文档元数据保存到: {DOCS_PATH}")


if __name__ == "__main__":
    build_index()

执行文档入库：

python ingest.py

九、准备示例企业文档

在 data/employee_policy.txt 中写入：

员工报销制度

1. 员工因公产生的交通费、住宿费、餐饮费可申请报销。
2. 报销时需要提供合法有效发票，发票抬头应为公司全称。
3. 单笔金额超过 1000 元的报销，需要部门负责人审批。
4. 差旅报销应在出差结束后 7 个工作日内提交。
5. 如发票遗失，需提交情况说明，并由直属上级确认。

在 data/product_manual.txt 中写入：

产品接口说明

本系统支持通过 REST API 接入第三方业务系统。

接口认证方式采用 Bearer Token。
调用方需要在 HTTP Header 中携带 Authorization 字段。
格式为：Authorization: Bearer 

系统默认接口调用频率限制为每分钟 600 次。
如需提升调用额度，需要联系客户成功经理申请。

然后执行：

python ingest.py

十、问答服务源码

创建 app.py：

import pickle
from typing import List

import faiss
import numpy as np
from fastapi import FastAPI
from pydantic import BaseModel
from sentence_transformers import SentenceTransformer
from openai import OpenAI

from config import (
    FAISS_INDEX_PATH,
    DOCS_PATH,
    EMBEDDING_MODEL_NAME,
    OPENAI_API_KEY,
    OPENAI_BASE_URL,
    CHAT_MODEL,
)

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

embedding_model = SentenceTransformer(EMBEDDING_MODEL_NAME)

index = faiss.read_index(FAISS_INDEX_PATH)

with open(DOCS_PATH, "rb") as f:
    docs = pickle.load(f)

client = OpenAI(
    api_key=OPENAI_API_KEY,
    base_url=OPENAI_BASE_URL
)


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


class SourceItem(BaseModel):
    source: str
    chunk_id: int
    content: str
    score: float


class AskResponse(BaseModel):
    answer: str
    sources: List[SourceItem]


def retrieve(question: str, top_k: int = 5):
    """向量检索"""
    query_embedding = embedding_model.encode(
        [question],
        normalize_embeddings=True
    )

    query_embedding = np.array(query_embedding).astype("float32")

    scores, ids = index.search(query_embedding, top_k)

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

        doc = docs[idx]
        results.append({
            "source": doc["source"],
            "chunk_id": doc["chunk_id"],
            "content": doc["content"],
            "score": float(score)
        })

    return results


def build_prompt(question: str, contexts: List[dict]) -> str:
    """构造提示词"""
    context_text = "\n\n".join([
        f"【资料{i + 1}】\n来源：{item['source']}\n内容：{item['content']}"
        for i, item in enumerate(contexts)
    ])

    prompt = f"""
你是企业内部知识库助手。请严格根据给定资料回答用户问题。

要求：
1. 如果资料中有明确答案，请直接回答。
2. 如果资料中没有相关信息，请说明“根据当前知识库资料，无法确定”。
3. 不要编造制度、数字、流程或接口信息。
4. 回答要清晰、简洁，必要时使用条目。
5. 回答末尾请列出参考来源文件名。

用户问题：
{question}

知识库资料：
{context_text}
"""
    return prompt


def generate_answer(question: str, contexts: List[dict]) -> str:
    """调用大模型生成答案"""
    prompt = build_prompt(question, contexts)

    completion = client.chat.completions.create(
        model=CHAT_MODEL,
        messages=[
            {
                "role": "system",
                "content": "你是严谨的企业知识库问答助手，只能基于提供的上下文回答。"
            },
            {
                "role": "user",
                "content": prompt
            }
        ],
        temperature=0.2
    )

    return completion.choices[0].message.content


@app.post("/ask", response_model=AskResponse)
def ask(req: AskRequest):
    contexts = retrieve(req.question, req.top_k)
    answer = generate_answer(req.question, contexts)

    return {
        "answer": answer,
        "sources": contexts
    }


@app.get("/")
def index_page():
    return {
        "message": "企业 AI 知识库服务已启动",
        "usage": "POST /ask，参数：question, top_k"
    }

启动服务：

uvicorn app:app --reload --port 8000

十一、接口测试

使用 curl 测试：

curl -X POST "http://127.0.0.1:8000/ask" \
  -H "Content-Type: application/json" \
  -d '{
    "question": "差旅报销需要多久内提交？",
    "top_k": 3
  }'

可能返回：

{
  "answer": "差旅报销应在出差结束后 7 个工作日内提交。\n\n参考来源：data/employee_policy.txt",
  "sources": [
    {
      "source": "data/employee_policy.txt",
      "chunk_id": 0,
      "content": "员工报销制度\n\n1. 员工因公产生的交通费、住宿费、餐饮费可申请报销...",
      "score": 0.83
    }
  ]
}

再测试一个产品相关问题：

curl -X POST "http://127.0.0.1:8000/ask" \
  -H "Content-Type: application/json" \
  -d '{
    "question": "接口认证方式是什么？",
    "top_k": 3
  }'

返回结果可能是：

{
  "answer": "接口认证方式采用 Bearer Token。调用方需要在 HTTP Header 中携带 Authorization 字段，格式为：Authorization: Bearer 。\n\n参考来源：data/product_manual.txt",
  "sources": [...]
}

十二、前端页面简单实现

如果希望提供一个简单网页，可以新增 static/index.html：

  
  
  


  
企业 AI 知识库

  
  

  提问

  

    
回答
    
等待提问...
  

  

    
参考资料
    

  

  

如果需要让 FastAPI 托管静态页面，可以修改 app.py：

from fastapi.staticfiles import StaticFiles

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

然后访问：

http://127.0.0.1:8000/static/index.html

十三、关键实现细节解析

1. 文本切分为什么重要？

很多初学者搭建知识库时，会直接把整篇文档向量化。这样做通常效果不好，因为一篇文档可能包含多个主题，大模型检索时难以精确命中。

合理的做法是将文档拆成多个 chunk。每个 chunk 控制在几百字到一千字左右，同时保留一定重叠内容。这样既能保证语义完整，又能提升召回准确率。

常见切分策略有：

• 按字符长度切分；
• 按段落切分；
• 按标题层级切分；
• 按 Markdown 结构切分；
• 按语义边界切分；
• 按表格、章节、条款单独切分。

企业制度类文档建议按条款切分；产品手册建议按标题和小节切分；代码文档建议按函数、类、模块切分。

2. 为什么要使用 overlap？

如果切分时完全不重叠，可能会出现上下文断裂。

例如原文：

接口认证方式采用 Bearer Token。
调用方需要在 Header 中携带 Authorization 字段。

如果刚好被切成两个块，用户问“Authorization 字段怎么用”，系统可能只召回第二句，却缺少认证方式说明。加入 overlap 后，相邻文本块会有部分重复内容，可以减少信息丢失。

3. 如何降低大模型幻觉？

企业知识库最怕的问题是：AI 一本正经地胡说八道。

降低幻觉可以从以下方面入手：

1. 提示词约束

   • 明确要求只基于知识库资料回答；
   • 未检索到答案时必须说明无法确定。

2. 引用来源

   • 每个答案附带来源文档；
   • 方便用户核验。

3. 设置较低 temperature

   • 例如 0.1 或 0.2；
   • 降低模型发散程度。

4. 检索结果过滤

   • 对低相似度结果进行过滤；
   • 避免无关上下文误导模型。

5. 权限控制

   • 不同用户只能检索其有权限访问的文档；
   • 防止越权回答。

十四、生产环境需要补充哪些能力？

本文源码适合学习和原型验证。如果要在企业真实环境中使用，还需要补充以下能力。

1. 用户认证与权限控制

企业知识库中可能包含敏感信息，例如：

• 薪酬制度；
• 合同数据；
• 客户资料；
• 财务数据；
• 研发方案；
• 商业报价。

因此必须支持基于用户、部门、角色、项目的权限控制。检索时不能只查向量相似度，还要过滤用户可访问的文档范围。

2. 文档增量更新

企业文档会持续变化。生产系统需要支持：

• 新文档自动入库；
• 旧文档更新后重新向量化；
• 删除文档后同步删除向量；
• 文档版本管理；
• 定时扫描网盘或知识平台。

3. 更好的文档解析能力

真实企业文档往往很复杂：

• PDF 扫描件；
• 表格；
• 图片；
• 合同；
• PPT；
• Excel；
• 带目录和页眉页脚的 Word；
• 混合中英文内容。

因此可能需要加入：

• OCR；
• 表格解析；
• 版面分析；
• 图片理解；
• 文档结构识别。

4. 问答日志与反馈闭环

企业知识库上线后，需要持续优化。建议记录：

• 用户问题；
• 检索片段；
• AI 回答；
• 用户点赞/点踩；
• 未命中问题；
• 高频问题；
• 响应耗时。

这些日志可以帮助管理员发现：

• 哪些知识缺失；
• 哪些文档过期；
• 哪些问题经常被问；
• 哪些回答质量较差。

5. 混合检索

单纯向量检索有时会漏掉精确关键词，例如：

• 产品型号；
• 合同编号；
• API 名称；
• 错误码；
• 专有名词；
• 人名；
• 项目代号。

因此生产环境建议使用混合检索：

向量检索 + 关键词检索 + 重排序模型

典型流程为：

1. 向量检索召回一批候选；
2. BM25 关键词检索召回一批候选；
3. 合并去重；
4. 使用 rerank 模型重新排序；
5. 取前 N 个片段给大模型。

这样可以显著提升企业知识库的准确率。

十五、可扩展优化方向

1. 接入企业微信、飞书或钉钉

企业知识库最好的入口不是单独的网站，而是员工每天使用的协作工具。可以将 /ask 接口封装成机器人，接入：

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

员工在群里或私聊中直接提问，AI 自动返回答案和来源。

2. 支持多知识库

不同部门需要不同知识库：

• 人事知识库；
• 财务知识库；
• 产品知识库；
• 研发知识库；
• 客服知识库；
• 运维知识库。

系统可以为每个知识库建立独立索引，也可以在统一向量库中增加 knowledge_base_id 字段，通过元数据过滤实现隔离。

3. 支持文件上传入库

目前示例中是将文件放入 data 目录后运行脚本。实际系统可以提供文件上传接口：

POST /upload

上传后自动解析、切分、向量化，并写入向量数据库。

4. 支持流式输出

大模型回答可能需要几秒钟。为了提升体验，可以使用 SSE 或 WebSocket 实现流式输出，让用户看到答案逐字生成。

5. 增加答案可信度评分

可以基于以下指标给答案打分：

• 检索结果最高相似度；
• 多个片段之间是否一致；
• 是否命中文档标题；
• rerank 分数；
• 大模型自评置信度。

当可信度较低时，系统可以提示：

当前答案可信度较低，建议人工核验。

十六、常见问题

Q1：企业知识库是否必须微调大模型？

不一定。大多数企业知识库场景优先使用 RAG，而不是微调。

原因是企业文档经常变化，如果每次文档更新都微调模型，成本高且维护困难。RAG 可以实时更新知识，只需要重新入库或增量更新向量索引。

微调更适合解决模型风格、特定任务格式、复杂业务推理流程等问题，而不是单纯记忆企业文档。

Q2：为什么有时检索不到正确内容？

可能原因包括：

• 文档切分不合理；
• embedding 模型中文效果不好；
• 用户问题和文档表达差异太大；
• top_k 设置太小；
• 文档解析丢失内容；
• 没有使用关键词检索；
• 没有 rerank；
• 文档本身内容不完整。

优化方向是：调整切分策略、更换 embedding 模型、增加混合检索、引入重排序模型、补充知识文档。

Q3：数据安全如何保障？

建议从以下几个方面考虑：

1. 私有化部署大模型和向量数据库；
2. 对敏感文档做权限控制；
3. 对日志进行脱敏；
4. 接口增加鉴权；
5. 对上传文件进行安全扫描；
6. 禁止将敏感数据发送到不可信第三方接口；
7. 建立知识库审计机制。

十七、总结

企业 AI 知识库是大语言模型在企业内部非常实用的落地场景。它可以帮助企业降低知识查找成本、提升员工效率、沉淀组织经验，并为客服、售前、运维、人事、财务等多个部门提供智能问答能力。

本文给出了一个基于 FastAPI、FAISS、sentence-transformers 和大模型接口的简易实现，完整覆盖了：

• 文档解析；
• 文本切分；
• 向量化；
• 向量检索；
• Prompt 构造；
• 大模型问答；
• 来源引用；
• Web 接口；
• 简单前端页面。

如果只是做原型验证，本文源码已经可以跑通核心流程。如果要进入生产环境，还需要进一步补充权限控制、增量更新、混合检索、重排序、日志反馈、安全审计和企业系统集成等能力。

AI 知识库的关键不只是“接一个大模型”，而是要围绕企业知识资产建立完整的数据治理和应用闭环。只有当文档可靠、权限清晰、检索准确、答案可追溯、反馈可优化时，它才能真正成为企业内部可长期使用的智能知识基础设施。