“FAQ + AI”智能助手全栈实现方案-EW帮帮网

在这里插入图片描述

文章目录

第一部分：总体架构与技术选型

1.1 核心架构图

整个系统的工作流程如下，它清晰地展示了用户问题如何被处理，以及FAQ模块和AI模块如何协同工作：

1.2 技术选型说明

前端 (Frontend): React 或 Next.js。选择Next.js是因为它支持服务端渲染(SSR)，对SEO更友好，且API Routes功能可以简化全栈开发。
后端 (Backend): FastAPI。性能极高，自带自动交互式API文档，异步支持好，非常适合此类AI应用。
向量数据库 (Vector Database): Chroma。轻量级、开源、易于使用和嵌入到应用程序中，非常适合原型和中小规模项目。
Embedding 模型 (Embedding Model): BGE-large-en-v1.5 或 m3e-large。开源、强大的中英文双语模型，可以本地部署，避免数据泄露风险。
大语言模型 (LLM): OpenAI GPT-3.5-Turbo 或 ChatGLM3-6B。前者是API调用，开发简单，效果顶尖但涉及费用和数据出境；后者可本地部署，数据安全但需要GPU资源。
项目依赖管理: Poetry。优于pip，能更好地管理虚拟环境和依赖版本。
部署: Docker + Docker Compose。实现环境隔离、一键部署和水平扩展。

第二部分：详细实现步骤

2.1 环境准备与项目初始化

1. 创建项目目录

mkdir faq-ai-assistant
cd faq-ai-assistant

2. 使用Poetry初始化项目

poetry init
# 根据提示填写项目信息，或一路回车使用默认值

3. 安装核心依赖

poetry add fastapi uvicorn chromadb openai sentence-transformers
poetry add --group dev pytest httpx

2.2 知识库处理与向量化 (Ingestion Pipeline)

这是最关键的离线处理步骤。我们创建一个脚本，将Markdown文档读取、分块、并存入向量数据库。

创建脚本: ingest.py

import os
import re
from pathlib import Path
import chromadb
from chromadb.config import Settings
from sentence_transformers import SentenceTransformer
import markdown
from bs4 import BeautifulSoup

# 配置参数
MARKDOWN_DOCS_PATH = "./docs"  # 你的Markdown文档存放的文件夹
CHROMA_DB_PATH = "./chroma_db"
EMBEDDING_MODEL_NAME = "BAAI/bge-large-zh-v1.5"  # 使用中文模型
COLLECTION_NAME = "knowledge_base"

# 初始化Embedding模型
print("Loading embedding model...")
embed_model = SentenceTransformer(EMBEDDING_MODEL_NAME)

# 初始化Chroma客户端，持久化到磁盘
chroma_client = chromadb.PersistentClient(path=CHROMA_DB_PATH)

# 创建一个集合（Collection）
collection = chroma_client.get_or_create_collection(
    name=COLLECTION_NAME,
    metadata={"hnsw:space": "cosine"}  # 使用余弦相似度
)

def clean_markdown_text(text: str) -> str:
    """将Markdown转换为纯文本，并清理多余的空格和换行"""
    html = markdown.markdown(text)
    soup = BeautifulSoup(html, "html.parser")
    return soup.get_text().strip()

def split_markdown_into_chunks(file_path: str, chunk_size: int = 500, chunk_overlap: int = 50) -> list[str]:
    """将Markdown文件按标题和固定大小进行分块"""
    with open(file_path, 'r', encoding='utf-8') as f:
        content = f.read()

    # 首先尝试按标题分割（## 标题）
    pattern = r'(?m)^(## .+)$'
    splits = re.split(pattern, content)
    chunks = []
    current_chunk = ""
    header = ""

    for i, split in enumerate(splits):
        if i % 2 == 0:  # 内容部分
            current_chunk += split
        else: # 标题部分
            # 如果当前块已经有内容，先保存上一个块
            if current_chunk:
                cleaned_chunk = clean_markdown_text(header + current_chunk)
                if cleaned_chunk:
                    chunks.append(cleaned_chunk)
                current_chunk = ""
            header = split + "\n\n"  # 新标题

    # 处理最后一个块
    if current_chunk:
        cleaned_chunk = clean_markdown_text(header + current_chunk)
        if cleaned_chunk:
            chunks.append(cleaned_chunk)

    # 如果按标题分块后块太大，再按固定大小进行二次分块
    final_chunks = []
    for chunk in chunks:
        if len(chunk) > chunk_size:
            # 简单的按句号、换行分句，然后按长度合并
            sentences = re.split(r'(?<=[。！？.!?\n])', chunk)
            temp_chunk = ""
            for sentence in sentences:
                if len(temp_chunk) + len(sentence) > chunk_size:
                    final_chunks.append(temp_chunk)
                    temp_chunk = sentence
                    # 重叠机制：保留上一块的最后 overlap 个字符
                    if chunk_overlap > 0 and len(temp_chunk) > chunk_overlap:
                        overlap_text = temp_chunk[:chunk_overlap]
                        temp_chunk = temp_chunk[chunk_overlap:]
                        final_chunks[-1] += overlap_text
                else:
                    temp_chunk += sentence
            if temp_chunk:
                final_chunks.append(temp_chunk)
        else:
            final_chunks.append(chunk)
    return final_chunks

def add_documents_to_collection(docs_dir: str):
    """将目录下的所有Markdown文件处理并添加到向量数据库"""
    doc_files = list(Path(docs_dir).glob("**/*.md"))
    all_chunks = []
    all_metadatas = []
    all_ids = []

    for doc_path in doc_files:
        print(f"Processing: {doc_path}")
        chunks = split_markdown_into_chunks(str(doc_path))
        for idx, chunk in enumerate(chunks):
            all_chunks.append(chunk)
            # 元数据，记录来源文件，便于追溯
            all_metadatas.append({"source": str(doc_path)})
            # 为每个块生成唯一ID
            all_ids.append(f"{doc_path.stem}_{idx}")

    # 为所有文本块生成向量
    print(f"Generating embeddings for {len(all_chunks)} chunks...")
    embeddings = embed_model.encode(all_chunks).tolist()

    # 批量添加到集合
    print("Adding to Chroma collection...")
    collection.add(
        documents=all_chunks,
        embeddings=embeddings,
        metadatas=all_metadatas,
        ids=all_ids
    )
    print(f"Successfully added {len(all_chunks)} chunks from {len(doc_files)} files.")

if __name__ == "__main__":
    add_documents_to_collection(MARKDOWN_DOCS_PATH)

运行此脚本：

poetry run python ingest.py

2.3 构建后端API (FastAPI Server)

创建主应用文件: main.py

from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import List, Optional
import chromadb
from chromadb.config import Settings
from sentence_transformers import SentenceTransformer
import openai
from openai import OpenAI
import os
import json

# --- 配置 ---
EMBEDDING_MODEL_NAME = "BAAI/bge-large-zh-v1.5"
CHROMA_DB_PATH = "./chroma_db"
COLLECTION_NAME = "knowledge_base"
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")  # 从环境变量读取
# 如果使用本地模型，如ChatGLM，这里需要修改为本地API的base_url
# LOCAL_MODEL_BASE_URL = "http://localhost:8000/v1" 

# 加载FAQ数据集
with open('./data/faqs.json', 'r', encoding='utf-8') as f:
    FAQ_LIST = json.load(f)

# --- 初始化组件 ---
app = FastAPI(title="FAQ+AI Assistant API")

# 允许跨域，方便前端调用
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 初始化Embedding模型
print("Loading embedding model...")
embed_model = SentenceTransformer(EMBEDDING_MODEL_NAME)

# 初始化Chroma客户端
chroma_client = chromadb.PersistentClient(path=CHROMA_DB_PATH)
collection = chroma_client.get_collection(COLLECTION_NAME)

# 初始化OpenAI客户端（用于GPT模型）
if OPENAI_API_KEY:
    openai_client = OpenAI(api_key=OPENAI_API_KEY)
else:
    openai_client = None

# --- Pydantic模型（请求/响应体）---
class QueryRequest(BaseModel):
    question: str
    user_id: Optional[str] = None  # 可用于记录用户历史

class FAQAnswer(BaseModel):
    answer: str
    confidence: float
    type: str = "faq"

class AIAnswer(BaseModel):
    answer: str
    sources: List[str]  # 引用的源文件列表
    type: str = "ai"

class QueryResponse(BaseModel):
    answer: Union[FAQAnswer, AIAnswer]  # 联合类型，可以是FAQ或AI的答案
    # 也可以设计成一个包含所有信息的统一响应体

# --- 工具函数 ---
def search_faq(question: str, threshold: float = 0.8) -> Optional[dict]:
    """
    在FAQ列表中做语义相似度搜索。
    返回最匹配的问题和答案，以及置信度。
    """
    question_embedding = embed_model.encode([question]).tolist()[0]
    # 这里简化处理：实际应将FAQ列表也向量化并存入向量库进行搜索。
    # 为演示，我们直接计算与每个FAQ的相似度
    max_similarity = 0
    best_match = None

    for faq in FAQ_LIST:
        faq_embedding = embed_model.encode([faq['question']]).tolist()[0]
        # 计算余弦相似度 (使用点积，因为向量是归一化的)
        from numpy import dot
        from numpy.linalg import norm
        similarity = dot(question_embedding, faq_embedding) / (norm(question_embedding) * norm(faq_embedding))
        if similarity > max_similarity:
            max_similarity = similarity
            best_match = faq

    if best_match and max_similarity > threshold:
        return {"question": best_match['question'], "answer": best_match['answer'], "confidence": max_similarity}
    else:
        return None

def retrieve_relevant_chunks(question: str, n_results: int = 3) -> List[str]:
    """从向量数据库中检索最相关的文本片段"""
    question_embedding = embed_model.encode([question]).tolist()[0]
    results = collection.query(
        query_embeddings=[question_embedding],
        n_results=n_results
    )
    # results 结构: {'ids': [[...]], 'documents': [[doc1, doc2, doc3]], ...}
    return results['documents'][0] if results['documents'] else []

def generate_answer_with_llm(question: str, context_chunks: List[str]) -> str:
    """
    使用LLM根据检索到的上下文生成答案。
    这里以OpenAI API为例。
    """
    if not openai_client:
        return "抱歉，AI服务未配置，无法回答此问题。"

    # 构建Prompt
    context = "\n\n".join(context_chunks)
    prompt = f"""
    你是一个专业的客服助手，请严格根据以下提供的上下文信息来回答用户的问题。
    如果上下文信息中没有答案，或者信息不相关，请直接回答“根据现有资料，我无法找到相关信息来回答这个问题。”
    不要编造任何未知的信息。

    上下文信息：
    {context}

    用户问题：{question}

    请给出准确、有帮助的回答：
    """

    try:
        response = openai_client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[
                {"role": "system", "content": "你是一个乐于助人的客服助手。"},
                {"role": "user", "content": prompt}
            ],
            temperature=0.1  # 低温度，保证答案更确定、更基于事实
        )
        return response.choices[0].message.content
    except Exception as e:
        return f"生成答案时出现错误: {str(e)}"

# --- API路由 ---
@app.get("/")
async def root():
    return {"message": "FAQ+AI Assistant API is running"}

@app.post("/query", response_model=QueryResponse)
async def query_knowledge_base(request: QueryRequest):
    """
    核心查询接口。
    1. 先检查FAQ
    2. 如果FAQ不匹配，则检索知识库并用AI生成答案
    """
    # 1. FAQ 匹配
    faq_result = search_faq(request.question)
    if faq_result:
        return QueryResponse(
            answer=FAQAnswer(
                answer=faq_result['answer'],
                confidence=faq_result['confidence']
            )
        )

    # 2. AI 处理流程
    relevant_chunks = retrieve_relevant_chunks(request.question)
    if not relevant_chunks:
        raise HTTPException(status_code=404, detail="未找到相关信息")

    ai_generated_answer = generate_answer_with_llm(request.question, relevant_chunks)

    # 提取来源文件（从元数据中获取，这里简化处理）
    source_files = list(set([chunk.metadata.get('source', 'Unknown') for chunk in relevant_chunks if hasattr(chunk, 'metadata')]))
    # 注意：上一步需要修改retrieve_relevant_chunks函数以返回元数据，此处为演示简化。

    return QueryResponse(
        answer=AIAnswer(
            answer=ai_generated_answer,
            sources=source_files
        )
    )

# 健康检查端点
@app.get("/health")
async def health_check():
    return {"status": "healthy"}

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

2.4 构建简单前端 (Next.js)

创建文件: frontend/pages/index.js

import { useState } from 'react';

export default function Home() {
  const [query, setQuery] = useState('');
  const [answer, setAnswer] = useState('');
  const [isLoading, setIsLoading] = useState(false);
  const [answerType, setAnswerType] = useState('');

  const handleSubmit = async (e) => {
    e.preventDefault();
    if (!query.trim()) return;

    setIsLoading(true);
    setAnswer('');
    setAnswerType('');

    try {
      const response = await fetch('http://localhost:8000/query', {
        method: 'POST',
        headers: {
          'Content-Type': 'application/json',
        },
        body: JSON.stringify({ question: query }),
      });

      if (!response.ok) {
        throw new Error(`HTTP error! status: ${response.status}`);
      }

      const data = await response.json();
      setAnswer(data.answer.answer);
      setAnswerType(data.answer.type);
    } catch (error) {
      console.error('Error:', error);
      setAnswer('抱歉，查询过程中出现了错误。');
    } finally {
      setIsLoading(false);
    }
  };

  return (
    <div style={{ padding: '2rem' }}>
      <h1>智能客服助手</h1>
      <form onSubmit={handleSubmit}>
        <input
          type="text"
          value={query}
          onChange={(e) => setQuery(e.target.value)}
          placeholder="请输入您的问题..."
          disabled={isLoading}
          style={{ width: '300px', padding: '0.5rem', marginRight: '1rem' }}
        />
        <button type="submit" disabled={isLoading}>
          {isLoading ? '思考中...' : '提问'}
        </button>
      </form>
      {answer && (
        <div style={{ marginTop: '2rem' }}>
          <h2>回答 {answerType === 'faq' ? '(来自FAQ)' : '(来自AI分析)'}:</h2>
          <p>{answer}</p>
        </div>
      )}
    </div>
  );
}

第三部分：部署方案

我们使用Docker容器化应用，确保环境一致性。

3.1 编写Dockerfile

创建文件: Dockerfile

# 使用官方Python运行时作为父镜像
FROM python:3.11-slim

# 设置工作目录
WORKDIR /app

# 复制项目文件
COPY . .

# 安装系统依赖（如果需要编译某些Python包）
RUN apt-get update && apt-get install -y \
    gcc \
    g++ \
    && rm -rf /var/lib/apt/lists/*

# 安装Poetry
RUN pip install poetry

# 配置Poetry不创建虚拟环境（直接在当前环境安装）
RUN poetry config virtualenvs.create false

# 使用Poetry安装项目依赖
RUN poetry install --no-dev

# 下载Embedding模型（也可以在启动时下载，但提前下载好镜像更大但启动更快）
# RUN python -c "from sentence_transformers import SentenceTransformer; SentenceTransformer('BAAI/bge-large-zh-v1.5')"

# 暴露端口
EXPOSE 8000

# 启动应用
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

3.2 编写docker-compose.yml

创建文件: docker-compose.yml

version: '3.8'

services:
  faq-ai-backend:
    build: .
    container_name: faq-ai-backend
    ports:
      - "8000:8000"
    volumes:
      # 持久化向量数据库和数据文件
      - ./chroma_db:/app/chroma_db
      - ./data:/app/data
      - ./docs:/app/docs
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}  # 从.env文件或宿主机环境变量传入
    restart: unless-stopped

  # 如果需要，可以添加一个Nginx服务作为反向代理和静态文件服务器
  # nginx:
  #   image: nginx:alpine
  #   ports:
  #     - "80:80"
  #   volumes:
  #     - ./nginx.conf:/etc/nginx/conf.d/default.conf
  #   depends_on:
  #     - faq-ai-backend

# 定义卷，用于持久化数据（上面已经使用了主机绑定，这里也可用命名卷）
# volumes:
#   chroma_data:
#   app_data:

3.3 创建环境变量文件

创建文件: .env

OPENAI_API_KEY=your_openai_api_key_here

3.4 构建和运行

# 构建镜像
docker-compose build

# 启动服务
docker-compose up -d

# 查看日志
docker-compose logs -f

现在，后端API将在 http://localhost:8000 运行。前端Next.js应用需要另外部署或使用docker-compose集成。

在这里插入图片描述

第四部分：安全、监控与维护

4.1 安全增强

API密钥管理: 永远不要将密钥硬编码在代码中。使用环境变量或 secrets 管理工具（如Vault）。
速率限制 (Rate Limiting): 在FastAPI中添加slowapi等中间件，防止API被滥用。
输入验证与清理: 对用户输入进行严格的验证和清理，防止Prompt注入等攻击。
HTTPS: 在生产环境使用Nginx反向代理并配置SSL证书。
认证与授权 (Optional): 为API添加API Key或JWT认证。

4.2 监控与日志

日志: 使用Python的logging模块记录所有查询、错误和信息。
健康检查: 已经实现了/health端点，可以集成到Kubernetes或监控系统中。
性能监控: 使用Prometheus、Grafana等工具监控API响应时间和资源使用情况。

4.3 系统维护

知识库更新:
- 修改docs目录下的Markdown文件。
- 重新运行ingest.py脚本（可以将其也容器化，定期执行或通过API触发）。
- 或者编写一个/admin/ingest API端点来触发更新。
FAQ更新: 直接修改data/faqs.json文件，重启服务或实现热重载逻辑。
模型更新: 关注Embedding模型和LLM的更新，定期评估新模型的效果。

“FAQ + AI”智能助手全栈实现方案