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

优质文章学习记录

07 Apr 2026 — 13 min read

文章目录

**第一部分：总体架构与技术选型**
- **1.1 核心架构图**
- **1.2 技术选型说明**
**第二部分：详细实现步骤**
**第三部分：部署方案**
**第四部分：安全、监控与维护**

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

1.1 核心架构图

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

知识库AI处理管道HTTP Request高置信度匹配HTTP Response向量化处理加载至内存低置信度/未匹配生成式答案Markdown文档FAQ列表答案生成
大语言模型LLM语义检索
Embedding模型+向量数据库用户提问Web应用前端
React/Next.js后端API网关
FastAPIFAQ匹配模块返回标准答案

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"}# 使用余弦相似度)defclean_markdown_text(text:str)->str:"""将Markdown转换为纯文本，并清理多余的空格和换行""" html = markdown.markdown(text) soup = BeautifulSoup(html,"html.parser")return soup.get_text().strip()defsplit_markdown_into_chunks(file_path:str, chunk_size:int=500, chunk_overlap:int=50)->list[str]:"""将Markdown文件按标题和固定大小进行分块"""withopen(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 inenumerate(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:iflen(chunk)> chunk_size:# 简单的按句号、换行分句，然后按长度合并 sentences = re.split(r'(?<=[。！？.!?\n])', chunk) temp_chunk =""for sentence in sentences:iflen(temp_chunk)+len(sentence)> chunk_size: final_chunks.append(temp_chunk) temp_chunk = sentence # 重叠机制：保留上一块的最后 overlap 个字符if chunk_overlap >0andlen(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 defadd_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 inenumerate(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数据集withopen('./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模型（请求/响应体）---classQueryRequest(BaseModel): question:str user_id: Optional[str]=None# 可用于记录用户历史classFAQAnswer(BaseModel): answer:str confidence:floattype:str="faq"classAIAnswer(BaseModel): answer:str sources: List[str]# 引用的源文件列表type:str="ai"classQueryResponse(BaseModel): answer: Union[FAQAnswer, AIAnswer]# 联合类型，可以是FAQ或AI的答案# 也可以设计成一个包含所有信息的统一响应体# --- 工具函数 ---defsearch_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 =Nonefor 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:returnNonedefretrieve_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[]defgenerate_answer_with_llm(question:str, context_chunks: List[str])->str:""" 使用LLM根据检索到的上下文生成答案。 这里以OpenAI API为例。 """ifnot 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:returnf"生成答案时出现错误: {str(e)}"# --- API路由 [email protected]("/")asyncdefroot():return{"message":"FAQ+AI Assistant API is running"}@app.post("/query", response_model=QueryResponse)asyncdefquery_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)ifnot 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 ifhasattr(chunk,'metadata')]))# 注意：上一步需要修改retrieve_relevant_chunks函数以返回元数据，此处为演示简化。return QueryResponse( answer=AIAnswer( answer=ai_generated_answer, sources=source_files ))# 健康检查端点@app.get("/health")asyncdefhealth_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的更新，定期评估新模型的效果。