AI 工作流专栏 26：从 0 搭建企业级 AI 知识库

ai-knowledge-base/
├── frontend/              # React 前端
│   ├── src/
│   │   ├── App.jsx
│   │   ├── components/    # ChatBox, DocUploader
│   │   └── api/           # 调后端
│   └── package.json
├── backend/               # FastAPI 后端
│   ├── app/
│   │   ├── main.py        # 入口
│   │   ├── routers/       # chat / documents / admin
│   │   ├── rag/           # RAG 核心
│   │   ├── llm/           # 模型路由
│   │   ├── guardrails/    # 护栏
│   │   └── config.py
│   └── requirements.txt
├── docker-compose.yml
└── README.md

-- 启用 pgvector 扩展
CREATE EXTENSION IF NOT EXISTS vector;

-- 文档表
CREATE TABLE documents (
    id SERIAL PRIMARY KEY,
    title VARCHAR(200) NOT NULL,
    source_type VARCHAR(20),         -- pdf / docx / md / url
    file_path VARCHAR(500),          -- 对象存储路径
    content TEXT,                    -- 全文
    chunk_count INTEGER DEFAULT 0,
    created_at TIMESTAMP DEFAULT NOW()
);

-- 文档切片表（带向量）
CREATE TABLE document_chunks (
    id SERIAL PRIMARY KEY,
    document_id INTEGER REFERENCES documents(id) ON DELETE CASCADE,
    chunk_index INTEGER,
    content TEXT NOT NULL,
    embedding vector(1024),          -- BGE-M3 是 1024 维
    created_at TIMESTAMP DEFAULT NOW()
);

-- 对话历史表
CREATE TABLE chat_sessions (
    id SERIAL PRIMARY KEY,
    user_id VARCHAR(50),
    title VARCHAR(200),
    created_at TIMESTAMP DEFAULT NOW()
);

CREATE TABLE chat_messages (
    id SERIAL PRIMARY KEY,
    session_id INTEGER REFERENCES chat_sessions(id) ON DELETE CASCADE,
    role VARCHAR(20),                -- user / assistant
    content TEXT,
    sources JSONB,                   -- 引用的文档片段 id
    tokens_used INTEGER,
    cost DECIMAL(10,4),
    created_at TIMESTAMP DEFAULT NOW()
);

CREATE INDEX idx_chunks_embedding ON document_chunks 
USING hnsw (embedding vector_cosine_ops);

# backend/app/routers/documents.py
from fastapi import APIRouter, UploadFile, File, BackgroundTasks
from typing import List
import os, uuid

router = APIRouter(prefix="/api/documents", tags=["documents"])

@router.post("/upload")
async def upload_docs(
    background_tasks: BackgroundTasks,
    files: List[UploadFile] = File(...),
):
    """上传文档，异步处理（解析→切片→向量化→入库）"""
    doc_ids = []
    for f in files:
        # 1. 保存到对象存储（这里简化为本地）
        file_id = str(uuid.uuid4())
        path = f"./storage/{file_id}_{f.filename}"
        with open(path, "wb") as out:
            out.write(await f.read())

        # 2. 写入 documents 表
        doc_id = await db.execute(
            "INSERT INTO documents (title, source_type, file_path) VALUES ($1,$2,$3) RETURNING id",
            f.filename, f.filename.split(".")[-1], path
        )
        doc_ids.append(doc_id)

        # 3. 异步处理（不阻塞上传响应）
        background_tasks.add_task(process_document, doc_id, path)

    return {"doc_ids": doc_ids, "status": "processing"}

# backend/app/rag/pipeline.py
import pdfplumber
from langchain.text_splitter import RecursiveCharacterTextSplitter
from sentence_transformers import SentenceTransformer
import numpy as np

# 加载本地 BGE-M3 模型
embedder = SentenceTransformer("BAAI/bge-m3")

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=80,
    separators=["\n\n", "\n", "。", "！", "？", ".", " "]
)

async def process_document(doc_id: int, file_path: str):
    """完整流水线：解析 → 切片 → 向量化 → 入库"""
    # 1. 解析
    text = parse_file(file_path)       # PDF/Word/Markdown 各自的解析器

    # 2. 切片
    chunks = splitter.split_text(text)

    # 3. 向量化（批量）
    embeddings = embedder.encode(chunks, normalize_embeddings=True)

    # 4. 入库
    for i, (chunk, emb) in enumerate(zip(chunks, embeddings)):
        await db.execute(
            """INSERT INTO document_chunks (document_id, chunk_index, content, embedding) 
               VALUES ($1, $2, $3, $4)""",
            doc_id, i, chunk, emb.tolist()
        )

    # 更新文档状态
    await db.execute(
        "UPDATE documents SET chunk_count=$1, content=$2 WHERE id=$3",
        len(chunks), text, doc_id
    )

def parse_file(path: str) -> str:
    """按文件类型解析"""
    if path.endswith(".pdf"):
        with pdfplumber.open(path) as pdf:
            return "\n\n".join(p.extract_text() or "" for p in pdf.pages)
    elif path.endswith(".docx"):
        from docx import Document
        return "\n".join(p.text for p in Document(path).paragraphs)
    elif path.endswith(".md"):
        with open(path) as f:
            return f.read()
    else:
        raise ValueError(f"Unsupported file type: {path}")

# backend/app/routers/chat.py
from fastapi import APIRouter
from fastapi.responses import StreamingResponse
from openai import OpenAI
import os, asyncio

router = APIRouter(prefix="/api/chat", tags=["chat"])
client = OpenAI(api_key=os.getenv("DEEPSEEK_API_KEY"),
                base_url="https://api.deepseek.com")

@router.post("/ask")
async def ask(req: AskRequest):
    """用户提问，返回流式答案：向量化 + 检索 + LLM 流式生成"""
    # 1. 把问题向量化
    q_emb = embedder.encode([req.question], normalize_embeddings=True)[0]

    # 2. 向量检索 top-5
    results = await db.fetch("""
        SELECT content, document_id, 1 - (embedding <=> $1) AS score
        FROM document_chunks
        ORDER BY embedding <=> $1
        LIMIT 5
    """, q_emb.tolist())

    # 过滤低分的
    results = [r for r in results if r["score"] > 0.5]

    if not results:
        return {"answer": "知识库中未找到相关内容，请补充文档或换个问法。"}

    # 3. 拼 Prompt
    context = "\n\n".join(f"[{i+1}] {r['content']}" 
                          for i, r in enumerate(results))
    sources = [{"doc_id": r["document_id"], "score": r["score"]} 
               for r in results]

    prompt = f"""你是企业知识库助手。请根据【参考资料】回答问题。

【参考资料】
{context}

【规则】
1. 只基于参考资料作答，不要编造
2. 句末标注引用编号，例如 [1]
3. 资料不足时明确说"资料中未提及"
4. 回答简洁专业

【问题】{req.question}
"""

    # 4. 调 LLM 流式生成
    async def stream():
        full = ""
        stream = client.chat.completions.create(
            model="deepseek-chat",
            messages=[{"role": "user", "content": prompt}],
            stream=True,
        )
        for chunk in stream:
            delta = chunk.choices[0].delta.content or ""
            if delta:
                full += delta
                yield f"data: {delta}\n\n"   # SSE 格式
        # 最后发一个来源信息
        yield f"data: __SOURCES__:{sources}\n\n"

        # 异步存对话历史（监控与成本统计用）
        await save_message(req.session_id, "user", req.question)
        await save_message(req.session_id, "assistant", full, sources)

    return StreamingResponse(stream(), media_type="text/event-stream")

// frontend/src/components/ChatBox.jsx
import { useState } from "react";

export default function ChatBox() {
    const [question, setQuestion] = useState("");
    const [answer, setAnswer] = useState("");
    const [sources, setSources] = useState([]);
    const [loading, setLoading] = useState(false);

    const ask = async () => {
        if (!question.trim()) return;
        setLoading(true);
        setAnswer("");
        setSources([]);

        // 用 fetch + ReadableStream 消费 SSE
        const res = await fetch("/api/chat/ask", {
            method: "POST",
            headers: { "Content-Type": "application/json" },
            body: JSON.stringify({ question, session_id: 1 }),
        });
        const reader = res.body.getReader();
        const decoder = new TextDecoder();

        while (true) {
            const { done, value } = await reader.read();
            if (done) break;
            const text = decoder.decode(value);

            // 解析 SSE
            for (const line of text.split("\n")) {
                if (!line.startsWith("data: ")) continue;
                const payload = line.slice(6);
                if (payload.startsWith("__SOURCES__:")) {
                    setSources(JSON.parse(payload.slice(13)));
                } else {
                    setAnswer(prev => prev + payload);   // 打字机效果
                }
            }
        }
        setLoading(false);
    };

    return (
        <div className="max-w-3xl mx-auto p-6">
            <h1 className="text-2xl font-bold mb-4">企业知识库</h1>
            <div className="flex gap-2 mb-4">
                <input
                    value={question}
                    onChange={e => setQuestion(e.target.value)}
                    onKeyDown={e => e.key === "Enter" && ask()}
                    placeholder="问任何关于公司的问题..."
                    className="flex-1 p-3 border rounded-lg"
                />
                <button 
                    onClick={ask} 
                    disabled={loading}
                    className="px-6 py-3 bg-blue-600 text-white rounded-lg disabled:opacity-50"
                >
                    {loading ? "思考中..." : "提问"}
                </button>
            </div>

            {answer && (
                <div className="p-4 bg-gray-50 rounded-lg whitespace-pre-wrap">
                    {answer}
                </div>
            )}

            {sources.length > 0 && (
                <div className="mt-4 text-sm text-gray-500">
                    引用来源：{sources.map(s => `文档#${s.doc_id}`).join(", ")}
                </div>
            )}
        </div>
    );
}

# backend/app/llm/router.py
"""根据问题复杂度路由到不同模型：简单问题用便宜模型"""

def route_model(question: str, has_context: bool) -> str:
    """简单规则路由（生产可用 LLM 路由）"""
    # 简单问答用 DeepSeek（便宜）
    if len(question) < 30 and has_context:
        return "deepseek-chat"
    # 复杂推理用 GPT-4o
    if any(k in question for k in ["分析", "对比", "总结", "为什么"]):
        return "gpt-4o"
    # 默认用 DeepSeek
    return "deepseek-chat"

# backend/app/rag/cache.py
"""相似问题命中缓存，省 Token"""
import redis, json, numpy as np

r = redis.Redis(decode_responses=True)

async def get_cached_answer(question_emb, threshold=0.92):
    """语义缓存：找相似问题"""
    # 遍历缓存的问题向量（生产用向量库，这里简化）
    for key in r.scan_iter("qa_cache:*"):
        cached = json.loads(r.get(key))
        sim = np.dot(question_emb, cached["emb"])
        if sim > threshold:
            return cached["answer"], cached["sources"]
    return None

async def save_to_cache(question_emb, answer, sources):
    key = f"qa_cache:{hash(question_emb.tobytes())}"
    r.setex(key, 3600, json.dumps({  # 1 小时过期
        "emb": question_emb.tolist(),
        "answer": answer,
        "sources": sources
    }))

# backend/app/guardrails/input_filter.py
"""过滤不当输入 + 防 Prompt 注入"""
import re

INJECTION_PATTERNS = [
    r"忽略.{0,10}(指令|规则|前面)",
    r"ignore.{0,10}(previous|above|all)",
    r"system\s*prompt",
    r"你的(指令|提示词)是什么",
]

PII_PATTERNS = {
    "phone": r"1[3-9]\d{9}",
    "id_card": r"\d{17}[\dXx]",
    "email": r"[\w.-]+@[\w.-]+\.\w+",
}

def check_input(text: str) -> tuple[bool, str]:
    """返回 (是否通过, 原因)"""
    # 检查 Prompt 注入
    for pattern in INJECTION_PATTERNS:
        if re.search(pattern, text, re.IGNORECASE):
            return False, "检测到潜在的 Prompt 注入"

    # PII 脱敏
    for pii_type, pattern in PII_PATTERNS.items():
        text = re.sub(pattern, f"[{pii_type}已脱敏]", text)

    return True, text

# backend/app/routers/admin.py
HITL_KEYWORDS = ["薪资", "辞退", "解除合同", "理赔", "诉讼", "仲裁"]

@router.post("/ask")
async def ask(req: AskRequest):
    # ... 前面的 RAG 逻辑 ...

    # HITL 检查：高风险关键词触发人工
    if any(k in req.question for k in HITL_KEYWORDS):
        # 不直接答，转人工审核
        review_id = await create_review(req.question, context, sources)
        return {
            "answer": f"您的问题涉及敏感话题（HR/法务），已转人工审核。审核单号：{review_id}，"
                      f"预计 4 小时内回复。",
            "needs_review": True,
            "review_id": review_id
        }

    # 正常流式回答 ...

@router.post("/admin/review/{review_id}")
async def review(review_id: int, decision: ReviewDecision):
    """人工审核：approve（直接发）/ edit（修改后发）/ reject（拒绝）"""
    if decision.action == "approve":
        await send_answer_to_user(review_id, decision.draft_answer)
    elif decision.action == "edit":
        await send_answer_to_user(review_id, decision.edited_answer)
    else:  # reject
        await notify_user(review_id, "您的问题暂时无法回答，请联系管理员")

    await log_audit(review_id, decision)   # 审计日志

# backend/app/rag/traced_rag.py
from langfuse import Langfuse
from langfuse.decorators import observe

langfuse = Langfuse()

@observe()    # 自动 trace 这个函数
async def rag_answer(question: str):
    # 每一步作为子 span
    with langfuse.start_as_current_span(name="embedding") as span:
        q_emb = embedder.encode([question])[0]

    with langfuse.start_as_current_span(name="retrieval") as span:
        results = await retrieve(q_emb)
        span.set_attribute("retrieved_count", len(results))

    with langfuse.start_as_current_span(name="generation") as span:
        answer = await generate(question, results)

    return answer

# backend/scripts/evaluate.py
"""每周用 Golden Set 跑评估，监控质量漂移"""
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision

GOLDEN_SET = [
    {"q": "公司年假多少天？", "ground_truth": "入职满一年 5 天，满三年 10 天..."},
    {"q": "怎么申请退款？", "ground_truth": "登录后台→我的订单→申请退款..."},
    # 50-200 条人工标注
]

async def run_evaluation():
    results = []
    for item in GOLDEN_SET:
        answer, sources = await rag_answer(item["q"])
        results.append({
            "question": item["q"],
            "answer": answer,
            "contexts": [s["content"] for s in sources],
            "ground_truth": item["ground_truth"]
        })

    # 用 RAGAS 自动评分
    scores = evaluate(
        dataset=results,
        metrics=[faithfulness, answer_relevancy, context_precision]
    )
    print(f"忠实度: {scores['faithfulness']:.2f}")
    print(f"相关性: {scores['answer_relevancy']:.2f}")

    # 低于阈值告警
    if scores['faithfulness'] < 0.8:
        await send_alert(f"告警：知识库忠实度下降到 {scores['faithfulness']:.2f}")

# docker-compose.yml
version: "3.9"
services:
  db:
    image: pgvector/pgvector:pg16     # 自带 pgvector 的 PG 镜像
    environment:
      POSTGRES_PASSWORD: ${DB_PASSWORD}
      POSTGRES_DB: knowledge_base
    volumes:
      - pgdata:/var/lib/postgresql/data
    ports: ["5432:5432"]

  redis:
    image: redis:7
    ports: ["6379:6379"]

  backend:
    build: ./backend
    environment:
      DATABASE_URL: postgres://postgres:${DB_PASSWORD}@db/knowledge_base
      REDIS_URL: redis://redis:6379/0
      DEEPSEEK_API_KEY: ${DEEPSEEK_API_KEY}
      OPENAI_API_KEY: ${OPENAI_API_KEY}
      LANGFUSE_PUBLIC_KEY: ${LANGFUSE_PUBLIC_KEY}
      LANGFUSE_SECRET_KEY: ${LANGFUSE_SECRET_KEY}
    depends_on: [db, redis]
    ports: ["8000:8000"]

  frontend:
    build: ./frontend
    ports: ["80:80"]
    depends_on: [backend]

  langfuse:
    image: langfuse/langfuse:latest
    ports: ["3000:3000"]
    environment:
      DATABASE_URL: postgres://postgres:${DB_PASSWORD}@db/langfuse
      NEXTAUTH_SECRET: ${NEXTAUTH_SECRET}

volumes:
  pgdata:

# .github/workflows/deploy.yml
name: Deploy
on:
  push:
    branches: [main]
jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Run tests
        run: |
          cd backend && pip install -r requirements.txt && pytest
  deploy:
    needs: test
    runs-on: ubuntu-latest
    steps:
      - name: Deploy to server
        uses: appleboy/ssh-action@v1
        with:
          host: ${{ secrets.SERVER_HOST }}
          key: ${{ secrets.SSH_KEY }}
          script: |
            cd /opt/knowledge-base
            git pull
            docker compose up -d --build