AI知识库搭建与智能问答完全教程

1. AI知识库系统架构设计

企业内部积累了大量非结构化知识——技术文档、产品手册、会议纪要、客服对话记录。传统关键词搜索早已无法满足精准检索的需求。AI知识库系统通过 RAG（Retrieval-Augmented Generation） 架构，将文档检索与大语言模型结合，实现"先查再答"的智能问答体验。

核心架构分为五层：

┌─────────────────────────────────────────┐
│           应用层 (Application)           │
│   Web UI / API Gateway / Chat Widget    │
├─────────────────────────────────────────┤
│           对话层 (Conversation)          │
│   多轮对话管理 / 上下文窗口 / 引用溯源  │
├─────────────────────────────────────────┤
│           检索层 (Retrieval)             │
│   混合检索 / 重排序 / 过滤              │
├─────────────────────────────────────────┤
│           存储层 (Storage)               │
│   向量数据库 / 元数据存储 / 文档索引    │
├─────────────────────────────────────────┤
│           处理层 (Ingestion)             │
│   文档解析 / 分块 / Embedding           │
└─────────────────────────────────────────┘

技术栈选型参考：

2. 多格式文档处理（PDF/Word/Excel/网页/图片）

企业知识源格式多样，统一的文档处理管线是第一步。

2.1 PDF解析

PDF是最常见的知识载体，但也是最难处理的格式之一。推荐使用 Unstructured 库，它能自动识别标题、段落、表格、图片等元素。

from unstructured.partition.pdf import partition_pdf

# 基础解析
elements = partition_pdf(
    filename="company_report.pdf",
    strategy="hi_res",           # 高精度模式，适合复杂排版
    infer_table_structure=True,  # 推断表格结构
    extract_images_in_pdf=True,  # 提取嵌入图片
    chunking_strategy="by_title" # 按标题分块
)

for el in elements:
    print(f"[{el.category}] {el.text[:100]}")

2.2 Word/Excel处理

from docx import Document
import openpyxl

def parse_docx(filepath):
    doc = Document(filepath)
    sections = []
    current_heading = ""
    current_text = []

    for para in doc.paragraphs:
        if para.style.name.startswith("Heading"):
            if current_text:
                sections.append({
                    "heading": current_heading,
                    "content": "\n".join(current_text)
                })
            current_heading = para.text
            current_text = []
        else:
            current_text.append(para.text)

    if current_text:
        sections.append({
            "heading": current_heading,
            "content": "\n".join(current_text)
        })
    return sections

def parse_excel(filepath):
    wb = openpyxl.load_workbook(filepath)
    all_data = []
    for sheet in wb.sheetnames:
        ws = wb[sheet]
        headers = [cell.value for cell in ws[1]]
        for row in ws.iter_rows(min_row=2, values_only=True):
            record = dict(zip(headers, row))
            all_data.append({
                "source": sheet,
                "content": " | ".join(f"{k}: {v}" for k, v in record.items() if v)
            })
    return all_data

2.3 网页抓取与解析

from bs4 import BeautifulSoup
import requests

def parse_webpage(url):
    resp = requests.get(url, timeout=10)
    soup = BeautifulSoup(resp.text, "html.parser")

    # 移除噪音元素
    for tag in soup(["script", "style", "nav", "footer", "header"]):
        tag.decompose()

    # 提取正文
    title = soup.title.string if soup.title else ""
    body = soup.get_text(separator="\n", strip=True)
    return {"title": title, "body": body, "url": url}

2.4 图片OCR

import pytesseract
from PIL import Image

def ocr_image(image_path):
    img = Image.open(image_path)
    text = pytesseract.image_to_string(img, lang="chi_sim+eng")
    return text.strip()

3. 智能分块策略与元数据提取

分块质量直接决定检索效果。块太大，噪声多；块太小，语义断裂。

3.1 递归字符分块（通用方案）

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=64,
    separators=["\n\n", "\n", "。", "；", ".", " ", ""],
    length_function=len
)

chunks = splitter.split_text(document_text)

3.2 语义分块（按语义边界切分）

from langchain_experimental.text_splitter import SemanticChunker
from langchain_community.embeddings import HuggingFaceEmbeddings

embeddings = HuggingFaceEmbeddings(model_name="BAAI/bge-m3")

semantic_splitter = SemanticChunker(
    embeddings,
    breakpoint_threshold_type="percentile",
    breakpoint_threshold_amount=85
)

semantic_chunks = semantic_splitter.split_text(document_text)

3.3 元数据提取

每个分块应附带丰富的元数据，用于后续过滤和溯源：

import hashlib
from datetime import datetime

def enrich_chunk(chunk_text, source_info):
    return {
        "id": hashlib.md5(chunk_text.encode()).hexdigest(),
        "text": chunk_text,
        "source": source_info["filename"],
        "page": source_info.get("page", 0),
        "heading": source_info.get("heading", ""),
        "doc_type": source_info.get("type", "unknown"),
        "created_at": datetime.now().isoformat(),
        "char_count": len(chunk_text)
    }

4. Embedding模型选型与对比

Embedding模型将文本映射为稠密向量，是语义检索的基础。

主流模型对比

本地部署Embedding服务

# 使用 sentence-transformers 本地部署
from sentence_transformers import SentenceTransformer
import numpy as np

model = SentenceTransformer("BAAI/bge-m3")

def get_embeddings(texts: list[str], batch_size=32) -> np.ndarray:
    embeddings = model.encode(
        texts,
        batch_size=batch_size,
        normalize_embeddings=True,  # L2归一化，便于余弦相似度
        show_progress_bar=True
    )
    return embeddings

# 测试
texts = ["什么是机器学习？", "深度学习与神经网络的关系"]
vectors = get_embeddings(texts)
print(f"向量维度: {vectors.shape}")  # (2, 1024)

使用API服务

import openai

client = openai.OpenAI(
    base_url="https://api.openai.com/v1",
    api_key="your-api-key"
)

def get_embeddings_api(texts, model="text-embedding-3-large"):
    response = client.embeddings.create(
        model=model,
        input=texts,
        dimensions=1024  # 支持自定义维度
    )
    return [item.embedding for item in response.data]

5. 向量数据库选型与部署

5.1 方案对比

5.2 Qdrant部署与使用

# docker-compose.yml
services:
  qdrant:
    image: qdrant/qdrant:latest
    ports:
      - "6333:6333"
      - "6334:6334"
    volumes:
      - qdrant_data:/qdrant/storage
    environment:
      QDRANT__SERVICE__GRPC_PORT: 6334
volumes:
  qdrant_data:

from qdrant_client import QdrantClient
from qdrant_client.models import (
    Distance, VectorParams, PointStruct,
    Filter, FieldCondition, MatchValue
)

client = QdrantClient(host="localhost", port=6333)

# 创建集合
client.create_collection(
    collection_name="knowledge_base",
    vectors_config=VectorParams(
        size=1024,
        distance=Distance.COSINE
    )
)

# 插入向量
points = [
    PointStruct(
        id=i,
        vector=embedding.tolist(),
        payload={
            "text": chunk["text"],
            "source": chunk["source"],
            "heading": chunk["heading"],
            "doc_type": chunk["doc_type"]
        }
    )
    for i, (chunk, embedding) in enumerate(zip(chunks, embeddings))
]

client.upsert(
    collection_name="knowledge_base",
    points=points,
    batch_size=100
)

5.3 Milvus部署（大规模场景）

from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType

connections.connect("default", host="localhost", port="19530")

# 定义Schema
fields = [
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
    FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=1024),
    FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=65535),
    FieldSchema(name="source", dtype=DataType.VARCHAR, max_length=512),
    FieldSchema(name="doc_type", dtype=DataType.VARCHAR, max_length=64),
]

schema = CollectionSchema(fields, description="Knowledge base collection")
collection = Collection("knowledge_base", schema)

# 创建IVF_FLAT索引（平衡速度与精度）
index_params = {
    "metric_type": "COSINE",
    "index_type": "IVF_FLAT",
    "params": {"nlist": 1024}
}
collection.create_index("embedding", index_params)

6. 混合检索与重排序

单一的向量检索在关键词匹配场景下表现不佳。混合检索结合语义检索和关键词检索，再通过重排序精排。

6.1 混合检索实现

from rank_bm25 import BM25Okapi
import numpy as np

class HybridRetriever:
    def __init__(self, qdrant_client, collection_name, embeddings_model):
        self.client = qdrant_client
        self.collection = collection_name
        self.model = embeddings_model
        self.corpus_texts = []
        self.bm25 = None

    def build_bm25_index(self):
        """从向量库中加载所有文本，构建BM25索引"""
        # 滚动读取所有文本
        results = self.client.scroll(
            collection_name=self.collection,
            limit=10000,
            with_payload=["text"]
        )
        self.corpus_texts = [p.payload["text"] for p in results[0]]
        tokenized = [list(text) for text in self.corpus_texts]  # 中文需分词
        self.bm25 = BM25Okapi(tokenized)

    def search(self, query: str, top_k: int = 10, alpha: float = 0.7):
        """混合检索：alpha为向量检索权重"""
        # 1. 向量检索
        query_vec = self.model.encode(query, normalize_embeddings=True)
        vector_results = self.client.search(
            collection_name=self.collection,
            query_vector=query_vec.tolist(),
            limit=top_k * 2
        )

        # 2. BM25关键词检索
        tokenized_query = list(query)
        bm25_scores = self.bm25.get_scores(tokenized_query)
        bm25_top_idx = np.argsort(bm25_scores)[::-1][:top_k * 2]

        # 3. 融合排序 (RRF - Reciprocal Rank Fusion)
        score_map = {}
        for rank, hit in enumerate(vector_results):
            doc_id = hit.id
            rrf_score = 1.0 / (60 + rank)
            score_map[doc_id] = score_map.get(doc_id, 0) + alpha * rrf_score

        for rank, idx in enumerate(bm25_top_idx):
            rrf_score = 1.0 / (60 + rank)
            score_map[idx] = score_map.get(idx, 0) + (1 - alpha) * rrf_score

        # 按融合分数排序
        sorted_ids = sorted(score_map, key=score_map.get, reverse=True)[:top_k]
        return sorted_ids, score_map

6.2 Cross-Encoder重排序

from sentence_transformers import CrossEncoder

reranker = CrossEncoder("BAAI/bge-reranker-v2-m3", max_length=512)

def rerank(query: str, candidates: list[dict], top_k: int = 5):
    """用Cross-Encoder精排"""
    pairs = [(query, c["text"]) for c in candidates]
    scores = reranker.predict(pairs)

    ranked = sorted(
        zip(candidates, scores),
        key=lambda x: x[1],
        reverse=True
    )
    return [item[0] for item in ranked[:top_k]]

7. LLM问答引擎集成

检索到相关文档后，需要将它们注入LLM的Prompt中生成最终答案。

7.1 标准RAG问答

from openai import OpenAI

client = OpenAI()

SYSTEM_PROMPT = """你是一个企业知识库助手。根据提供的参考资料回答用户问题。

规则：
1. 只基于参考资料回答，如果资料中没有相关信息，明确告知用户
2. 引用来源时标注文档名称
3. 回答要准确、简洁、有条理"""

def answer_question(query: str, context_chunks: list[dict]) -> str:
    context = "\n\n".join(
        f"【来源: {c['source']}】\n{c['text']}"
        for c in context_chunks
    )

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": f"参考资料：\n{context}\n\n用户问题：{query}"}
        ],
        temperature=0.1,
        max_tokens=2000
    )
    return response.choices[0].message.content

7.2 带引用溯源的问答

import json

CITATION_PROMPT = """你是一个企业知识库助手。根据参考资料回答问题，并在回答中标注引用来源。

输出JSON格式：
{
  "answer": "你的回答内容，在关键处用[1][2]标注引用",
  "citations": [
    {"index": 1, "source": "文档名", "excerpt": "引用的原文片段"}
  ],
  "confidence": "high/medium/low"
}

如果资料中没有相关信息，confidence设为low并说明原因。"""

def answer_with_citations(query, chunks):
    context = "\n\n".join(
        f"[{i+1}] 来源: {c['source']}\n{c['text']}"
        for i, c in enumerate(chunks)
    )

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": CITATION_PROMPT},
            {"role": "user", "content": f"参考资料：\n{context}\n\n问题：{query}"}
        ],
        temperature=0.1,
        response_format={"type": "json_object"}
    )

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

8. 多轮对话与上下文管理

真实场景中用户往往需要多轮追问。需要维护对话历史并进行上下文压缩。

8.1 对话历史管理

from dataclasses import dataclass, field
from typing import Optional

@dataclass
class Conversation:
    session_id: str
    history: list[dict] = field(default_factory=list)
    max_turns: int = 10

    def add_message(self, role: str, content: str, citations: Optional[list] = None):
        self.history.append({
            "role": role,
            "content": content,
            "citations": citations or [],
            "timestamp": datetime.now().isoformat()
        })
        # 保留最近N轮
        if len(self.history) > self.max_turns * 2:
            self.history = self.history[-self.max_turns * 2:]

    def get_context_window(self) -> list[dict]:
        """获取对话上下文窗口"""
        return [{"role": m["role"], "content": m["content"]} for m in self.history]

8.2 查询改写（Query Rewriting）

用户追问常带有指代（"它"、"这个"），需要先改写为独立问题。

def rewrite_query(current_query: str, chat_history: list[dict]) -> str:
    """将带指代的追问改写为独立问题"""
    history_text = "\n".join(
        f"{'用户' if m['role'] == 'user' else '助手'}: {m['content'][:200]}"
        for m in chat_history[-6:]  # 最近3轮
    )

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "将用户最新问题改写为独立的、不依赖上下文的问题。只输出改写后的问题，不要解释。"},
            {"role": "user", "content": f"对话历史：\n{history_text}\n\n最新问题：{current_query}"}
        ],
        temperature=0,
        max_tokens=200
    )
    return response.choices[0].message.content.strip()

8.3 上下文压缩

当对话过长时，对历史进行摘要压缩：

def compress_history(history: list[dict], max_tokens: int = 2000) -> str:
    """压缩对话历史为摘要"""
    full_text = "\n".join(f"{m['role']}: {m['content']}" for m in history)

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "将以下对话压缩为简洁摘要，保留关键信息和决策。"},
            {"role": "user", "content": full_text}
        ],
        temperature=0,
        max_tokens=max_tokens
    )
    return response.choices[0].message.content

9. 知识库更新与版本管理

知识库不是一次性构建的。文档会更新、新增、删除，需要一套完整的生命周期管理。

9.1 增量更新策略

import hashlib

class KnowledgeBaseManager:
    def __init__(self, qdrant_client, collection_name):
        self.client = qdrant_client
        self.collection = collection_name
        self._doc_hashes = {}  # 文档路径 -> 内容哈希

    def compute_file_hash(self, filepath: str) -> str:
        with open(filepath, "rb") as f:
            return hashlib.md5(f.read()).hexdigest()

    def incremental_update(self, doc_dir: str):
        """增量更新：只处理新增或修改的文档"""
        import os
        current_files = {}

        for root, _, files in os.walk(doc_dir):
            for f in files:
                if f.endswith((".pdf", ".docx", ".xlsx", ".md")):
                    path = os.path.join(root, f)
                    file_hash = self.compute_file_hash(path)
                    current_files[path] = file_hash

        # 检测变更
        new_files = set(current_files) - set(self._doc_hashes)
        deleted_files = set(self._doc_hashes) - set(current_files)
        modified_files = {
            f for f in current_files
            if f in self._doc_hashes and current_files[f] != self._doc_hashes[f]
        }

        # 处理新增和修改
        for filepath in new_files | modified_files:
            self._reindex_document(filepath)

        # 处理删除
        for filepath in deleted_files:
            self._remove_document(filepath)

        self._doc_hashes = current_files
        return {
            "new": len(new_files),
            "modified": len(modified_files),
            "deleted": len(deleted_files)
        }

    def _remove_document(self, filepath: str):
        """删除某文档的所有向量"""
        self.client.delete(
            collection_name=self.collection,
            points_selector=Filter(
                must=[FieldCondition(
                    key="source",
                    match=MatchValue(value=filepath)
                )]
            )
        )

9.2 版本快照

def create_snapshot(collection_name: str, snapshot_dir: str):
    """创建知识库快照，支持回滚"""
    import shutil
    from datetime import datetime

    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    snapshot_path = os.path.join(snapshot_dir, f"snapshot_{timestamp}")
    shutil.copytree(
        f"/qdrant/storage/collections/{collection_name}",
        snapshot_path
    )
    return snapshot_path

10. 实战案例：企业知识库问答系统

下面将上述所有模块整合为一个完整的FastAPI服务。

10.1 完整服务代码

from fastapi import FastAPI, UploadFile, File
from pydantic import BaseModel
import uvicorn

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

# 全局组件
retriever = None  # HybridRetriever实例
conversation_store = {}  # session_id -> Conversation

class QueryRequest(BaseModel):
    question: str
    session_id: str = "default"
    top_k: int = 5

class QueryResponse(BaseModel):
    answer: str
    citations: list[dict]
    confidence: str

@app.on_event("startup")
async def startup():
    global retriever
    # 初始化检索器、加载索引等
    retriever = HybridRetriever(...)

@app.post("/query", response_model=QueryResponse)
async def query(req: QueryRequest):
    # 1. 获取或创建会话
    conv = conversation_store.setdefault(
        req.session_id,
        Conversation(session_id=req.session_id)
    )
    conv.add_message("user", req.question)

    # 2. 查询改写（如果有历史）
    search_query = req.question
    if len(conv.history) > 2:
        search_query = rewrite_query(req.question, conv.get_context_window())

    # 3. 混合检索
    doc_ids, scores = retriever.search(search_query, top_k=req.top_k * 2)
    candidates = retriever.get_documents(doc_ids)

    # 4. 重排序
    reranked = rerank(search_query, candidates, top_k=req.top_k)

    # 5. 生成回答
    result = answer_with_citations(req.question, reranked)

    conv.add_message("assistant", result["answer"], result["citations"])

    return QueryResponse(
        answer=result["answer"],
        citations=result["citations"],
        confidence=result["confidence"]
    )

@app.post("/upload")
async def upload_document(file: UploadFile = File(...)):
    """上传文档并入库"""
    filepath = f"./uploads/{file.filename}"
    with open(filepath, "wb") as f:
        f.write(await file.read())

    # 解析 -> 分块 -> Embedding -> 入库
    chunks = process_document(filepath)
    embeddings = get_embeddings([c["text"] for c in chunks])
    store_to_qdrant(chunks, embeddings)

    return {"status": "ok", "chunks": len(chunks)}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

10.2 部署架构

# docker-compose.yml
services:
  api:
    build: .
    ports:
      - "8000:8000"
    depends_on:
      - qdrant
      - redis
    environment:
      - QDRANT_HOST=qdrant
      - REDIS_URL=redis://redis:6379

  qdrant:
    image: qdrant/qdrant:latest
    volumes:
      - qdrant_data:/qdrant/storage

  redis:
    image: redis:7-alpine
    # 用于会话缓存和速率限制

  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf
      - ./frontend:/usr/share/nginx/html

volumes:
  qdrant_data:

11. 评估指标与持续优化

11.1 检索质量评估

def evaluate_retrieval(test_cases: list[dict], retriever, k_values=[1, 3, 5, 10]):
    """评估检索质量"""
    results = {}

    for k in k_values:
        hits = 0
        mrr_sum = 0

        for case in test_cases:
            query = case["query"]
            expected_sources = set(case["relevant_sources"])

            doc_ids, _ = retriever.search(query, top_k=k)
            retrieved = retriever.get_documents(doc_ids)
            retrieved_sources = {d["source"] for d in retrieved}

            # Hit Rate
            if retrieved_sources & expected_sources:
                hits += 1

            # MRR (Mean Reciprocal Rank)
            for rank, doc in enumerate(retrieved):
                if doc["source"] in expected_sources:
                    mrr_sum += 1.0 / (rank + 1)
                    break

        n = len(test_cases)
        results[f"Hit@{k}"] = hits / n
        results[f"MRR@{k}"] = mrr_sum / n

    return results

11.2 端到端评估

def evaluate_e2e(test_cases: list[dict], pipeline):
    """端到端问答质量评估"""
    from rouge_score import rouge_scorer

    scorer = rouge_scorer.RougeScorer(["rougeL"], use_stemmer=True)
    scores = []

    for case in test_cases:
        result = pipeline.query(case["query"])
        score = scorer.score(case["reference_answer"], result["answer"])
        scores.append({
            "query": case["query"],
            "rouge_l": score["rougeL"].fmeasure,
            "correct_citation": any(
                c["source"] in case["relevant_sources"]
                for c in result["citations"]
            )
        })

    avg_rouge = sum(s["rouge_l"] for s in scores) / len(scores)
    citation_acc = sum(s["correct_citation"] for s in scores) / len(scores)

    return {"avg_rouge_l": avg_rouge, "citation_accuracy": citation_acc}

11.3 持续优化策略

分块策略优化：在测试集上对比不同 chunk_size（256/512/1024）和 chunk_overlap（0/32/64/128）的检索效果，选择最优组合。

Embedding模型切换：当业务场景变化时（如新增英文文档），需要重新评估多语言模型的效果。切换模型后需要重新生成所有向量。

检索参数调优：混合检索的 alpha 参数（语义vs关键词权重）对不同类型查询影响显著。技术类查询倾向关键词，概念类查询倾向语义。可以按查询类型动态调整。

用户反馈闭环：在问答界面增加"有用/无用"按钮，收集真实反馈数据，用于评估和优化。

@app.post("/feedback")
async def submit_feedback(session_id: str, message_index: int, helpful: bool):
    """收集用户反馈"""
    feedback = {
        "session_id": session_id,
        "message_index": message_index,
        "helpful": helpful,
        "timestamp": datetime.now().isoformat()
    }
    await save_feedback(feedback)  # 存入数据库
    return {"status": "ok"}

定期重索引：随着文档库增长，建议每季度对全量文档重新索引，更新Embedding和分块策略。

以上就是AI知识库搭建的完整技术方案。从文档处理到智能问答，从单次检索到多轮对话，每个环节都有对应的最佳实践。关键是根据实际业务场景做取舍——数据量小可以用Chroma快速验证，规模大了再迁移到Milvus；中文场景优先BGE-M3，多语言场景考虑text-embedding-3-large。先跑通MVP，再逐步优化。