文章目录

• • • **第一部分：总体架构与技术选型**
    • • **1.1 核心架构图**
      • **1.2 技术选型说明**
    • **第二部分：详细实现步骤**
    • • **2.1 环境准备与项目初始化**
      • **2.2 知识库处理与向量化 (Ingestion Pipeline)**
      • **2.3 构建后端API (FastAPI Server)**
      • **2.4 构建简单前端 (Next.js)**
    • **第三部分：部署方案**
    • • **3.1 编写Dockerfile**
      • **3.2 编写docker-compose.yml**
      • **3.3 创建环境变量文件**
      • **3.4 构建和运行**
    • **第四部分：安全、监控与维护**
    • • **4.1 安全增强**
      • **4.2 监控与日志**
      • **4.3 系统维护**

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

1.1 核心架构图

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

#mermaid-svg-pSWCbcVdiuDw6ftX {font-family:\"trebuchet ms\",verdana,arial,sans-serif;font-size:16px;fill:#333;}#mermaid-svg-pSWCbcVdiuDw6ftX .error-icon{fill:#552222;}#mermaid-svg-pSWCbcVdiuDw6ftX .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-pSWCbcVdiuDw6ftX .edge-thickness-normal{stroke-width:2px;}#mermaid-svg-pSWCbcVdiuDw6ftX .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-pSWCbcVdiuDw6ftX .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-pSWCbcVdiuDw6ftX .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-pSWCbcVdiuDw6ftX .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-pSWCbcVdiuDw6ftX .marker{fill:#333333;stroke:#333333;}#mermaid-svg-pSWCbcVdiuDw6ftX .marker.cross{stroke:#333333;}#mermaid-svg-pSWCbcVdiuDw6ftX svg{font-family:\"trebuchet ms\",verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-pSWCbcVdiuDw6ftX .label{font-family:\"trebuchet ms\",verdana,arial,sans-serif;color:#333;}#mermaid-svg-pSWCbcVdiuDw6ftX .cluster-label text{fill:#333;}#mermaid-svg-pSWCbcVdiuDw6ftX .cluster-label span{color:#333;}#mermaid-svg-pSWCbcVdiuDw6ftX .label text,#mermaid-svg-pSWCbcVdiuDw6ftX span{fill:#333;color:#333;}#mermaid-svg-pSWCbcVdiuDw6ftX .node rect,#mermaid-svg-pSWCbcVdiuDw6ftX .node circle,#mermaid-svg-pSWCbcVdiuDw6ftX .node ellipse,#mermaid-svg-pSWCbcVdiuDw6ftX .node polygon,#mermaid-svg-pSWCbcVdiuDw6ftX .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-pSWCbcVdiuDw6ftX .node .label{text-align:center;}#mermaid-svg-pSWCbcVdiuDw6ftX .node.clickable{cursor:pointer;}#mermaid-svg-pSWCbcVdiuDw6ftX .arrowheadPath{fill:#333333;}#mermaid-svg-pSWCbcVdiuDw6ftX .edgePath .path{stroke:#333333;stroke-width:2.0px;}#mermaid-svg-pSWCbcVdiuDw6ftX .flowchart-link{stroke:#333333;fill:none;}#mermaid-svg-pSWCbcVdiuDw6ftX .edgeLabel{background-color:#e8e8e8;text-align:center;}#mermaid-svg-pSWCbcVdiuDw6ftX .edgeLabel rect{opacity:0.5;background-color:#e8e8e8;fill:#e8e8e8;}#mermaid-svg-pSWCbcVdiuDw6ftX .cluster rect{fill:#ffffde;stroke:#aaaa33;stroke-width:1px;}#mermaid-svg-pSWCbcVdiuDw6ftX .cluster text{fill:#333;}#mermaid-svg-pSWCbcVdiuDw6ftX .cluster span{color:#333;}#mermaid-svg-pSWCbcVdiuDw6ftX div.mermaidTooltip{position:absolute;text-align:center;max-width:200px;padding:2px;font-family:\"trebuchet ms\",verdana,arial,sans-serif;font-size:12px;background:hsl(80, 100%, 96.2745098039%);border:1px solid #aaaa33;border-radius:2px;pointer-events:none;z-index:100;}#mermaid-svg-pSWCbcVdiuDw6ftX :root{--mermaid-font-family:\"trebuchet ms\",verdana,arial,sans-serif;} 知识库 AI处理管道 HTTP Request 高置信度匹配 HTTP Response 向量化处理 加载至内存 低置信度/未匹配 生成式答案 Markdown文档 FAQ列表 答案生成
大语言模型LLM 语义检索
Embedding模型+向量数据库 用户提问 Web应用前端
React/Next.js 后端API网关
FastAPI FAQ匹配模块 返回标准答案

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-assistantcd faq-ai-assistant

2. 使用Poetry初始化项目

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

3. 安装核心依赖

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

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

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

创建脚本: ingest.py

import osimport refrom pathlib import Pathimport chromadbfrom chromadb.config import Settingsfrom sentence_transformers import SentenceTransformerimport markdownfrom 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_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)}) # 为每个块生成唯一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, HTTPExceptionfrom fastapi.middleware.cors import CORSMiddlewarefrom pydantic import BaseModelfrom typing import List, Optionalimport chromadbfrom chromadb.config import Settingsfrom sentence_transformers import SentenceTransformerimport openaifrom openai import OpenAIimport osimport 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 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服务未配置，无法回答此问题。\" # 构建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\' }}> 
智能客服助手
   setQuery(e.target.value)} placeholder=\"请输入您的问题...\" disabled={isLoading} style={{ width: \'300px\', padding: \'0.5rem\', marginRight: \'1rem\' }} />  {isLoading ? \'思考中...\' : \'提问\'}   {answer && ( <div style={{ marginTop: \'2rem\' }}> 
回答 {answerType === \'faq\' ? \'(来自FAQ)\' : \'(来自AI分析)\'}:
 
{answer}