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

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

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 += splitelse: # 标题部分# 如果当前块已经有内容，先保存上一个块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_textelse:temp_chunk += sentenceif temp_chunk:final_chunks.append(temp_chunk)else:final_chunks.append(chunk)return final_chunksdef 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)})# 为每个块生成唯一IDall_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: struser_id: Optional[str] = None  # 可用于记录用户历史class FAQAnswer(BaseModel):answer: strconfidence: floattype: str = "faq"class AIAnswer(BaseModel):answer: strsources: 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 = 0best_match = Nonefor faq in FAQ_LIST:faq_embedding = embed_model.encode([faq['question']]).tolist()[0]# 计算余弦相似度 (使用点积，因为向量是归一化的)from numpy import dotfrom numpy.linalg import normsimilarity = dot(question_embedding, faq_embedding) / (norm(question_embedding) * norm(faq_embedding))if similarity > max_similarity:max_similarity = similaritybest_match = faqif best_match and max_similarity > threshold:return {"question": best_match['question'], "answer": best_match['answer'], "confidence": max_similarity}else:return Nonedef 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服务未配置，无法回答此问题。"# 构建Promptcontext = "\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.contentexcept 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. 先检查FAQ2. 如果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 uvicornuvicorn.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}><inputtype="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-backendports:- "8000:8000"volumes:# 持久化向量数据库和数据文件- ./chroma_db:/app/chroma_db- ./data:/app/data- ./docs:/app/docsenvironment:- 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 安全增强

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

4.2 监控与日志

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

4.3 系统维护

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

---