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

在 AI 编程逐渐普及的今天，企业内部知识库正在成为越来越重要的基础设施。无论是研发团队的技术文档、产品团队的需求资料、客服团队的问答手册，还是销售团队的方案资料，企业每天都会产生大量非结构化文档。如果这些知识仍然散落在飞书、企业微信、Confluence、语雀、网盘、PDF、Word、Excel 等不同系统中，员工在查找资料时就会浪费大量时间。

传统搜索主要依赖关键词匹配，面对“表达不一致但语义相同”的问题时效果并不好。例如员工搜索“客户续费流程”，文档里可能写的是“合同到期后的商务跟进步骤”；搜索“服务器报警处理”，文档里可能写的是“线上故障告警响应规范”。这类问题用传统搜索很难命中，但 AI 知识库可以通过语义理解，把用户问题与文档内容进行相似度匹配，再结合大语言模型生成自然语言答案。

本文将从实际开发角度，完整介绍如何用 AI 编程方式搭建一个企业知识库系统，并附上可运行的核心源码示例。

一、企业知识库的核心价值

企业知识库并不是简单地把文件放到一个系统里，而是要解决以下几个问题：

1. 资料统一管理
   将 PDF、Word、Markdown、TXT、网页等资料统一接入，形成企业内部可检索的知识资产。

2. 自然语言问答
   员工不再需要输入精确关键词，而是可以直接提问，例如：“新员工入职需要准备哪些材料？”、“线上事故如何升级处理？”

3. 提升知识复用效率
   老员工沉淀的经验、项目复盘、解决方案可以被新人快速获取，减少重复沟通。

4. 降低培训成本
   企业制度、产品说明、售后流程等内容可以通过 AI 助手自动解答，降低人工培训和答疑成本。

5. 保护企业数据安全
   企业可以选择私有化部署，结合权限系统，控制不同员工能访问的知识范围。

二、AI 企业知识库的基本架构

一个典型的 AI 企业知识库一般由以下几个模块组成：

文档上传
   ↓
文档解析
   ↓
文本切分
   ↓
向量化 Embedding
   ↓
向量数据库存储
   ↓
用户提问
   ↓
问题向量化
   ↓
相似文档检索
   ↓
大模型生成答案
   ↓
返回结果与引用来源

其中最关键的是 RAG 架构。

RAG 的全称是 Retrieval-Augmented Generation，即检索增强生成。它的核心思想是：大模型本身不直接记住企业所有内部知识，而是在用户提问时，先从知识库中检索相关内容，再把检索结果作为上下文交给大模型，由大模型根据这些内容生成答案。

这种方式有几个好处：

• 不需要频繁微调大模型；
• 可以快速更新企业知识；
• 回答内容有依据，便于追溯来源；
• 可以降低大模型胡编乱造的概率；
• 更适合企业私有知识场景。

三、技术选型

本文使用一个轻量级方案进行演示，方便理解核心原理。

1. 后端框架

使用 FastAPI，原因是：

• 性能较好；
• 写法简洁；
• 支持异步；
• 自动生成接口文档；
• 适合快速搭建 AI 应用服务。

2. 向量数据库

使用 ChromaDB 作为本地向量数据库。它适合原型开发和中小型知识库场景。如果后续进入生产环境，也可以替换为 Milvus、Qdrant、Weaviate、Elasticsearch Vector Search 或 PGVector。

3. 文本向量模型

可以使用 OpenAI Embedding，也可以使用本地开源模型。为了方便演示，本文代码以 OpenAI 兼容接口为例，同时保留可替换空间。

4. 大语言模型

可以接入：

• OpenAI GPT 系列；
• 通义千问；
• DeepSeek；
• 智谱 GLM；
• 月之暗面 Kimi；
• 私有化部署的 Qwen、Llama、Yi 等模型。

只要支持 OpenAI 兼容接口，代码改动都比较小。

四、项目目录结构

建议项目结构如下：

enterprise-knowledge-base/
├── app.py
├── config.py
├── requirements.txt
├── data/
│   └── docs/
│       └── example.txt
├── storage/
│   └── chroma/
└── README.md

说明：

• app.py：主程序，包含上传、入库、问答接口；
• config.py：配置文件；
• requirements.txt：依赖列表；
• data/docs/：存放上传文档；
• storage/chroma/：向量数据库持久化目录。

五、安装依赖

创建 requirements.txt：

fastapi==0.111.0
uvicorn==0.30.1
python-multipart==0.0.9
chromadb==0.5.3
openai==1.35.3
pypdf==4.2.0
python-docx==1.1.2

安装依赖：

pip install -r requirements.txt

六、配置文件源码

创建 config.py：

import os

# OpenAI 兼容接口配置
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "你的_API_KEY")
OPENAI_BASE_URL = os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")

# 模型名称，可替换为通义、DeepSeek、智谱等兼容模型
CHAT_MODEL = os.getenv("CHAT_MODEL", "gpt-4o-mini")
EMBEDDING_MODEL = os.getenv("EMBEDDING_MODEL", "text-embedding-3-small")

# 文档与向量库路径
DOC_DIR = "./data/docs"
CHROMA_DIR = "./storage/chroma"

# 文本切分参数
CHUNK_SIZE = 800
CHUNK_OVERLAP = 120

如果你使用其他模型服务，只需要修改环境变量即可：

export OPENAI_API_KEY="你的密钥"
export OPENAI_BASE_URL="https://你的模型服务地址/v1"
export CHAT_MODEL="你的聊天模型"
export EMBEDDING_MODEL="你的向量模型"

七、核心源码：FastAPI 企业知识库

创建 app.py：

import os
import uuid
from typing import List

import chromadb
from chromadb.config import Settings
from fastapi import FastAPI, UploadFile, File, HTTPException
from pydantic import BaseModel
from openai import OpenAI
from pypdf import PdfReader
from docx import Document

from config import (
    OPENAI_API_KEY,
    OPENAI_BASE_URL,
    CHAT_MODEL,
    EMBEDDING_MODEL,
    DOC_DIR,
    CHROMA_DIR,
    CHUNK_SIZE,
    CHUNK_OVERLAP,
)


app = FastAPI(title="Enterprise AI Knowledge Base")

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

os.makedirs(DOC_DIR, exist_ok=True)
os.makedirs(CHROMA_DIR, exist_ok=True)

chroma_client = chromadb.PersistentClient(
    path=CHROMA_DIR,
    settings=Settings(anonymized_telemetry=False),
)

collection = chroma_client.get_or_create_collection(
    name="enterprise_docs",
    metadata={"description": "enterprise knowledge base documents"},
)


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


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()
        if text:
            texts.append(text)
    return "\n".join(texts)


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


def load_document(file_path: str) -> str:
    """根据文件后缀解析文档"""
    suffix = file_path.lower().split(".")[-1]

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


def split_text(text: str, chunk_size: int = CHUNK_SIZE, overlap: int = CHUNK_OVERLAP) -> List[str]:
    """
    简单文本切分。
    生产环境中可以根据标题、段落、Markdown 结构进行更智能的切分。
    """
    text = text.replace("\r\n", "\n").replace("\r", "\n")
    chunks = []

    start = 0
    text_len = len(text)

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

        if chunk:
            chunks.append(chunk)

        start = end - overlap
        if start < 0:
            start = 0

        if start >= text_len:
            break

    return chunks


def get_embedding(text: str) -> List[float]:
    """调用 Embedding 模型，将文本转为向量"""
    response = client.embeddings.create(
        model=EMBEDDING_MODEL,
        input=text,
    )
    return response.data[0].embedding


def add_document_to_vector_db(file_name: str, text: str):
    """将文档切片后写入向量数据库"""
    chunks = split_text(text)

    ids = []
    documents = []
    embeddings = []
    metadatas = []

    for index, chunk in enumerate(chunks):
        chunk_id = str(uuid.uuid4())
        embedding = get_embedding(chunk)

        ids.append(chunk_id)
        documents.append(chunk)
        embeddings.append(embedding)
        metadatas.append({
            "file_name": file_name,
            "chunk_index": index,
        })

    if ids:
        collection.add(
            ids=ids,
            documents=documents,
            embeddings=embeddings,
            metadatas=metadatas,
        )

    return len(ids)


@app.get("/")
def index():
    return {
        "message": "Enterprise AI Knowledge Base is running.",
        "docs": "/docs",
    }


@app.post("/upload")
async def upload_file(file: UploadFile = File(...)):
    """
    上传企业文档并写入知识库。
    支持 txt、md、pdf、docx。
    """
    suffix = file.filename.lower().split(".")[-1]
    if suffix not in ["txt", "md", "pdf", "docx"]:
        raise HTTPException(status_code=400, detail="仅支持 txt、md、pdf、docx 文件")

    save_path = os.path.join(DOC_DIR, file.filename)

    with open(save_path, "wb") as f:
        content = await file.read()
        f.write(content)

    try:
        text = load_document(save_path)
        if not text.strip():
            raise ValueError("文档内容为空或无法解析")

        chunk_count = add_document_to_vector_db(file.filename, text)

        return {
            "message": "上传并入库成功",
            "file_name": file.filename,
            "chunk_count": chunk_count,
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.post("/ask")
def ask(request: AskRequest):
    """
    企业知识库问答接口。
    先检索相关文档片段，再调用大模型生成答案。
    """
    question = request.question.strip()

    if not question:
        raise HTTPException(status_code=400, detail="问题不能为空")

    question_embedding = get_embedding(question)

    results = collection.query(
        query_embeddings=[question_embedding],
        n_results=request.top_k,
    )

    retrieved_docs = results.get("documents", [[]])[0]
    retrieved_metadatas = results.get("metadatas", [[]])[0]

    if not retrieved_docs:
        return {
            "answer": "知识库中没有检索到相关内容，请补充相关文档后再提问。",
            "sources": [],
        }

    context_text = "\n\n".join([
        f"【资料{i + 1}】\n{doc}"
        for i, doc in enumerate(retrieved_docs)
    ])

    system_prompt = """
你是一个企业内部知识库助手。
你的任务是根据给定的企业资料回答用户问题。
要求：
1. 只能基于资料内容回答，不要编造。
2. 如果资料中没有答案，请明确说明“知识库中暂无相关信息”。
3. 回答要简洁、准确、条理清晰。
4. 如果涉及流程，请使用步骤列表。
5. 回答末尾可以提示用户查看引用来源。
"""

    user_prompt = f"""
用户问题：
{question}

企业知识库资料：
{context_text}

请根据以上资料回答用户问题。
"""

    completion = client.chat.completions.create(
        model=CHAT_MODEL,
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt},
        ],
        temperature=0.2,
    )

    answer = completion.choices[0].message.content

    sources = []
    for metadata in retrieved_metadatas:
        sources.append({
            "file_name": metadata.get("file_name"),
            "chunk_index": metadata.get("chunk_index"),
        })

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

八、启动项目

在项目根目录执行：

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

启动成功后，访问：

http://localhost:8000/docs

FastAPI 会自动生成 Swagger 接口文档。

九、准备测试文档

在 data/docs/example.txt 中写入一段企业知识：

新员工入职流程如下：

1. HR 在员工入职前三天发送入职通知。
2. 新员工需要准备身份证、银行卡、学历证明和离职证明。
3. 行政部门负责准备工位、门禁卡和办公设备。
4. IT 部门负责开通企业邮箱、代码仓库账号和内部系统权限。
5. 入职当天由 HR 组织新人培训，内容包括公司制度、考勤规范和信息安全要求。
6. 试用期为三个月，直属主管需要在试用期结束前一周完成转正评估。

然后通过 /upload 接口上传该文件。

也可以使用命令行测试：

curl -X POST "http://localhost:8000/upload" \
  -F "file=@data/docs/example.txt"

返回结果类似：

{
  "message": "上传并入库成功",
  "file_name": "example.txt",
  "chunk_count": 1
}

十、测试知识库问答

调用 /ask 接口：

curl -X POST "http://localhost:8000/ask" \
  -H "Content-Type: application/json" \
  -d '{
    "question": "新员工入职需要准备哪些材料？",
    "top_k": 3
  }'

可能返回：

{
  "answer": "新员工入职需要准备以下材料：\n\n1. 身份证；\n2. 银行卡；\n3. 学历证明；\n4. 离职证明。\n\n可查看引用来源了解更多入职流程信息。",
  "sources": [
    {
      "file_name": "example.txt",
      "chunk_index": 0
    }
  ]
}

这样，一个最小可用的企业 AI 知识库就搭建完成了。

十一、文本切分的优化思路

在实际企业场景中，文本切分质量会直接影响问答效果。简单按照固定长度切分虽然容易实现，但可能会把完整语义切断。例如一个流程说明被切成两半，大模型拿到的上下文就可能不完整。

更推荐的切分策略包括：

1. 按标题切分

对于 Markdown、语雀、Confluence 等文档，可以根据一级标题、二级标题进行分段。这样每个知识块都有明确主题。

2. 按段落切分

对于制度类、流程类文档，可以按照自然段落切分，再把相邻短段落合并。

3. 保留上下文重叠

切分时设置一定 overlap，例如 100 到 200 字，可以避免关键上下文丢失。

4. 增加元数据

每个知识块应该记录：

• 文件名；
• 文档标题；
• 所属部门；
• 创建人；
• 更新时间；
• 权限范围；
• 原始链接；
• 章节标题。

这些元数据可以用于过滤、权限控制和引用展示。

十二、企业级权限控制设计

企业知识库最容易被忽视的问题是权限。不是所有员工都应该看到所有资料。例如：

• HR 薪酬制度只允许 HR 和管理层访问；
• 财务报表只允许财务部门访问；
• 研发设计文档只允许研发团队访问；
• 客户合同和报价方案只允许销售、法务或项目相关成员访问。

因此，生产环境中应该在文档入库时写入权限信息：

metadatas.append({
    "file_name": file_name,
    "chunk_index": index,
    "department": "研发部",
    "permission": "dev_team",
})

检索时根据当前登录用户的权限进行过滤。例如：

results = collection.query(
    query_embeddings=[question_embedding],
    n_results=request.top_k,
    where={
        "permission": "dev_team"
    }
)

如果用户拥有多个权限，可以根据实际向量数据库能力进行组合过滤。对于更复杂的权限系统，可以在检索前先根据用户身份获取可访问文档 ID，然后只在这些文档范围内查询。

十三、避免大模型幻觉

企业知识库问答最重要的是准确性，而不是“看起来很聪明”。如果知识库中没有相关资料，大模型应该明确说不知道，而不是凭经验回答。

可以从以下几个方面降低幻觉：

1. 系统提示词约束
   明确要求模型只能根据资料回答。

2. 设置较低 temperature
   如 temperature=0.1 或 0.2，减少随机性。

3. 返回引用来源
   让用户知道答案来自哪些文档。

4. 增加相似度阈值
   如果检索结果相似度太低，就不要交给大模型回答。

5. 答案审计
   对高风险场景，如法务、财务、人事制度，可以增加人工审核或审批流程。

十四、生产环境增强方案

本文代码是一个最小可用版本，如果要用于正式生产，还需要进一步增强。

1. 用户登录与鉴权

可以接入企业微信、飞书、钉钉、LDAP、OAuth2 或 SSO 单点登录，识别用户身份和部门权限。

2. 文档增量同步

企业文档经常更新，不能只靠手动上传。可以开发定时任务，同步以下来源：

• 飞书知识库；
• 企业微信微盘；
• Confluence；
• Notion；
• 语雀；
• Git 仓库；
• 对象存储；
• 内部 CMS。

3. 文档删除与更新

当文档被删除或更新时，向量库中对应的 chunk 也要删除或重新生成，否则可能产生过期答案。

4. 多轮对话

实际用户经常会连续追问，例如：

用户：新员工入职需要准备什么？
用户：这些材料什么时候提交？
用户：如果没有离职证明怎么办？

这时需要保留会话历史，并结合当前问题进行改写和检索。

5. 混合检索

单纯向量检索有时会漏掉关键词很强的内容，比如制度编号、错误码、接口名。可以结合：

• 向量检索；
• BM25 关键词检索；
• 标签过滤；
• 热门问题库；
• 重排序模型 Reranker。

混合检索通常比单一向量检索效果更稳定。

6. 可观测性

需要记录：

• 用户问题；
• 命中文档；
• 模型回答；
• 用户反馈；
• 响应耗时；
• Token 消耗；
• 错误日志。

这些数据可以帮助持续优化知识库质量。

十五、前端页面示例

如果想快速做一个简单页面，可以创建 index.html：

  
  
  


  
企业 AI 知识库

  
  

  提问

  


  

这个页面可以直接打开测试。如果部署到生产环境，建议使用 Vue、React 或 Next.js，并增加登录、上传、文档管理、反馈按钮等功能。

十六、常见问题

1. 为什么上传了文档，问答还是不准确？

可能原因包括：

• 文档切分不合理；
• 文档内容本身不完整；
• Embedding 模型效果不好；
• 检索 top_k 太小；
• 问题表达与文档差异较大；
• 没有使用 reranker；
• 文档中存在大量表格或图片，解析效果差。

2. PDF 表格和扫描件怎么处理？

普通 PDF 可以提取文本，但扫描件需要 OCR。可以接入 PaddleOCR、Tesseract 或云厂商 OCR 服务。表格类文档建议转换成 Markdown 表格或结构化 JSON 后再入库。

3. 是否必须微调大模型？

大多数企业知识库场景不需要微调。RAG 已经可以解决大部分“基于企业资料回答”的需求。只有当企业有特殊表达风格、专业推理能力或固定格式输出要求时，才考虑微调。

4. 数据会不会泄露？

如果使用公有云模型，需要关注数据合规和供应商协议。对安全要求高的企业，建议使用私有化模型和内网部署向量数据库，同时做好访问控制、日志审计和数据加密。

十七、总结

本文完整介绍了如何通过 AI 编程搭建一个企业知识库系统。核心流程是：文档上传、文档解析、文本切分、向量化存储、问题检索、大模型生成答案。通过 RAG 架构，企业可以在不微调大模型的情况下，让 AI 助手基于内部资料进行可靠问答。

本文提供的源码虽然是最小可用版本，但已经覆盖了企业知识库最重要的技术链路。后续可以继续扩展权限控制、文档同步、多轮对话、混合检索、重排序、用户反馈和可观测性能力。对于企业来说，真正有价值的不是单纯接入一个大模型，而是把企业内部长期沉淀的知识资产组织起来、管理起来，并通过 AI 让这些知识被更高效地使用。

当企业知识库建设成熟后，它不仅是一个问答机器人，更会成为企业内部的智能工作入口：新人可以用它学习制度，研发可以用它查技术方案，客服可以用它找标准回复，销售可以用它生成客户方案，管理层也可以用它快速了解组织沉淀的经验。AI 编程的价值，正是在于用更低成本、更快速度，把这些能力真正落地到业务场景中。