在构建企业级AI应用时，如何让大模型准确回答特定领域问题一直是技术难点。传统大模型虽然具备强大的通用知识，但在处理企业内部文档、产品手册等私有数据时往往表现不佳。RAG（检索增强生成）技术通过将私有知识库与大模型结合，有效解决了这一痛点。

本文将手把手教你从零搭建完整的RAG知识库系统，涵盖核心原理、环境搭建、代码实战到生产部署全流程。无论你是AI初学者还是有经验的开发者，都能通过本文学会构建企业级私有知识库应用。

1. RAG技术核心原理与架构设计

1.1 什么是RAG技术

RAG（Retrieval-Augmented Generation，检索增强生成）是一种将信息检索与大语言模型生成相结合的技术框架。其核心思想是：在回答用户问题时，先从知识库中检索相关文档片段，然后将这些片段作为上下文提供给大模型，最终生成准确、有依据的回答。

与传统大模型直接生成相比，RAG具有三大优势：

• 准确性更高：基于真实文档内容生成，减少幻觉现象
• 可追溯性：答案来源清晰，便于验证和审计
• 实时更新：知识库内容可随时更新，模型无需重新训练

1.2 RAG系统架构详解

一个完整的RAG系统包含以下核心组件：

用户问题 → 向量化编码 → 向量数据库检索 → 相关文档召回 → 大模型生成 → 最终答案

文档处理流水线（离线）：

1. 文档解析：支持PDF、Word、Excel、TXT等多种格式
2. 文本切片：将长文档切分为语义完整的片段（chunk）
3. 向量化：使用Embedding模型将文本转换为向量
4. 向量存储：将向量存入向量数据库

问答推理流水线（在线）：

1. 问题向量化：将用户问题转换为向量
2. 相似度检索：在向量数据库中查找最相关的文档片段
3. 提示词构建：将检索结果组装成大模型可理解的提示词
4. 答案生成：大模型基于上下文生成最终答案

1.3 关键技术组件选型

向量模型选择：

• 中文场景：text2vec-large-chinese、m3e-large
• 多语言场景：text-embedding-ada-002、bge-large-en
• 视觉理解：qwen-vl-embedding（支持图文多模态）

向量数据库选型：

• 轻量级：Chroma、FAISS
• 生产级：Milvus、Weaviate、Qdrant
• 云服务：阿里云ADB-PG、腾讯云VectorDB

大模型选型：

• 开源：Qwen、ChatGLM、Baichuan
• 商用API：OpenAI GPT、文心一言、通义千问

2. 环境准备与工具安装

2.1 基础环境配置

本文以Python 3.8+为例，推荐使用conda管理环境：

# 创建虚拟环境 conda create -n rag-tutorial python=3.10 conda activate rag-tutorial # 安装核心依赖 pip install langchain==0.1.0 pip install chromadb==0.4.15 pip install sentence-transformers==2.2.2 pip install pypdf==3.17.0 pip install python-docx==1.1.0

2.2 向量数据库安装

选择ChromaDB作为本地向量数据库，安装简单且功能完善：

pip install chromadb # 验证安装 python -c "import chromadb; print('ChromaDB安装成功')"

2.3 嵌入模型准备

使用Sentence-BERT中文模型进行文本向量化：

from sentence_transformers import SentenceTransformer # 下载中文嵌入模型（首次运行会自动下载） model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')

3. 文档解析与预处理实战

3.1 支持多种文档格式解析

创建文档加载器，支持PDF、Word、TXT等常见格式：

import os from typing import List, Dict from pypdf import PdfReader from docx import Document class DocumentLoader: def __init__(self): self.supported_extensions = ['.pdf', '.docx', '.txt'] def load_document(self, file_path: str) -> str: """加载单个文档并返回文本内容""" ext = os.path.splitext(file_path)[1].lower() if ext == '.pdf': return self._load_pdf(file_path) elif ext == '.docx': return self._load_docx(file_path) elif ext == '.txt': return self._load_txt(file_path) else: raise ValueError(f"不支持的文档格式: {ext}") def _load_pdf(self, file_path: str) -> str: """加载PDF文档""" text = "" with open(file_path, 'rb') as file: reader = PdfReader(file) for page in reader.pages: text += page.extract_text() + "\n" return text def _load_docx(self, file_path: str) -> str: """加载Word文档""" doc = Document(file_path) text = "" for paragraph in doc.paragraphs: text += paragraph.text + "\n" return text def _load_txt(self, file_path: str) -> str: """加载文本文件""" with open(file_path, 'r', encoding='utf-8') as file: return file.read() # 使用示例 loader = DocumentLoader() content = loader.load_document("企业产品手册.pdf") print(f"文档加载成功，共{len(content)}字符")

3.2 智能文本切片策略

文本切片是RAG系统的关键环节，直接影响检索效果：

import re from typing import List class TextSplitter: def __init__(self, chunk_size: int = 500, chunk_overlap: int = 50): self.chunk_size = chunk_size self.chunk_overlap = chunk_overlap def split_by_sentence(self, text: str) -> List[str]: """按句子切分，保持语义完整性""" # 中文句子分割符 sentence_endings = r'[。！？!?]' sentences = re.split(sentence_endings, text) sentences = [s.strip() for s in sentences if s.strip()] chunks = [] current_chunk = "" for sentence in sentences: # 如果当前块加上新句子不超过chunk_size if len(current_chunk) + len(sentence) <= self.chunk_size: current_chunk += sentence + "。" else: if current_chunk: chunks.append(current_chunk.strip()) current_chunk = sentence + "。" if current_chunk: chunks.append(current_chunk.strip()) return chunks def split_by_fixed_size(self, text: str) -> List[str]: """固定大小切分，确保每个chunk长度均匀""" chunks = [] start = 0 text_length = len(text) while start < text_length: end = start + self.chunk_size if end > text_length: end = text_length chunk = text[start:end] # 确保不在句子中间切断 if end < text_length: last_period = max( chunk.rfind('。'), chunk.rfind('！'), chunk.rfind('？'), chunk.rfind('.'), chunk.rfind('!'), chunk.rfind('?') ) if last_period != -1 and last_period > self.chunk_size * 0.3: chunk = chunk[:last_period + 1] end = start + len(chunk) chunks.append(chunk) start = end - self.chunk_overlap return chunks # 测试文本切片 splitter = TextSplitter(chunk_size=300, chunk_overlap=30) sample_text = "这是一段测试文本。用于验证文本切片功能是否正常工作。我们需要确保每个切片都保持语义的完整性，避免在句子中间切断。这是非常重要的质量要求。" chunks = splitter.split_by_sentence(sample_text) for i, chunk in enumerate(chunks): print(f"切片 {i+1}: {chunk} (长度: {len(chunk)})")

3.3 元数据提取与增强

为文本切片添加元数据，提升检索准确性：

import hashlib from datetime import datetime class MetadataExtractor: def __init__(self): self.patterns = { 'date': r'\d{4}年\d{1,2}月\d{1,2}日|\d{4}-\d{2}-\d{2}', 'product_code': r'[A-Z]{2,}-\d{3,}', 'version': r'v\d+\.\d+(\.\d+)?', } def extract_metadata(self, text: str, filename: str) -> Dict: """从文本中提取元数据""" metadata = { 'filename': filename, 'chunk_id': hashlib.md5(text.encode()).hexdigest()[:8], 'timestamp': datetime.now().isoformat(), 'length': len(text), } # 提取特定模式的元数据 for key, pattern in self.patterns.items(): matches = re.findall(pattern, text) if matches: metadata[key] = matches return metadata # 元数据增强示例 extractor = MetadataExtractor() sample_chunk = "产品编号：XY-2024发布版本v2.1.0，发布日期2024年12月15日" metadata = extractor.extract_metadata(sample_chunk, "产品手册.pdf") print("提取的元数据:", metadata)

4. 向量化与向量数据库实战

4.1 嵌入模型配置与优化

import numpy as np from sentence_transformers import SentenceTransformer class EmbeddingService: def __init__(self, model_name: str = 'paraphrase-multilingual-MiniLM-L12-v2'): self.model = SentenceTransformer(model_name) self.dimension = self.model.get_sentence_embedding_dimension() def encode_text(self, text: str) -> List[float]: """将文本编码为向量""" if not text.strip(): return [0.0] * self.dimension # 对长文本进行智能处理 if len(text) > 512: sentences = text.split('。') chunks = [] current_chunk = "" for sentence in sentences: if len(current_chunk) + len(sentence) < 500: current_chunk += sentence + "。" else: if current_chunk: chunks.append(current_chunk) current_chunk = sentence + "。" if current_chunk: chunks.append(current_chunk) # 对每个chunk编码后取平均 embeddings = [self.model.encode(chunk) for chunk in chunks] return np.mean(embeddings, axis=0).tolist() else: return self.model.encode(text).tolist() def batch_encode(self, texts: List[str]) -> List[List[float]]: """批量编码文本""" return [self.encode_text(text) for text in texts] # 测试嵌入服务 embedding_service = EmbeddingService() sample_texts = ["这是一个测试句子", "这是另一个测试句子"] embeddings = embedding_service.batch_encode(sample_texts) print(f"向量维度: {len(embeddings[0])}") print(f"第一个向量的前10维: {embeddings[0][:10]}")

4.2 ChromaDB向量数据库集成

import chromadb from chromadb.config import Settings class VectorDatabase: def __init__(self, persist_directory: str = "./chroma_db"): self.client = chromadb.PersistentClient( path=persist_directory, settings=Settings(allow_reset=True) ) self.collection = None def create_collection(self, collection_name: str): """创建向量集合""" try: self.collection = self.client.get_collection(collection_name) print(f"使用现有集合: {collection_name}") except: self.collection = self.client.create_collection( name=collection_name, metadata={"description": "RAG知识库数据"} ) print(f"创建新集合: {collection_name}") def add_documents(self, documents: List[str], metadatas: List[Dict], ids: List[str]): """添加文档到向量数据库""" if not self.collection: raise ValueError("请先创建集合") # 生成嵌入向量 embeddings = embedding_service.batch_encode(documents) self.collection.add( embeddings=embeddings, documents=documents, metadatas=metadatas, ids=ids ) print(f"成功添加 {len(documents)} 个文档") def search_similar(self, query: str, n_results: int = 3) -> List[Dict]: """相似度搜索""" if not self.collection: raise ValueError("请先创建集合") query_embedding = embedding_service.encode_text(query) results = self.collection.query( query_embeddings=[query_embedding], n_results=n_results ) return results # 初始化向量数据库 vector_db = VectorDatabase() vector_db.create_collection("enterprise_knowledge") # 测试数据添加 test_documents = [ "公司产品支持多种支付方式，包括支付宝、微信支付和银联", "技术支持服务时间为工作日9:00-18:00，紧急问题可拨打400热线", "最新产品版本v3.2.1增加了AI智能客服功能" ] test_metadatas = [ {"source": "产品手册", "page": 1}, {"source": "服务协议", "page": 5}, {"source": "更新日志", "version": "v3.2.1"} ] test_ids = ["doc1", "doc2", "doc3"] vector_db.add_documents(test_documents, test_metadatas, test_ids)

4.3 高级检索功能实现

class AdvancedRetriever: def __init__(self, vector_db: VectorDatabase): self.vector_db = vector_db self.collection = vector_db.collection def semantic_search(self, query: str, n_results: int = 5, metadata_filter: Dict = None) -> List[Dict]: """语义搜索 with 元数据过滤""" query_embedding = embedding_service.encode_text(query) where_clause = None if metadata_filter: where_clause = metadata_filter results = self.collection.query( query_embeddings=[query_embedding], n_results=n_results, where=where_clause ) formatted_results = [] for i in range(len(results['documents'][0])): formatted_results.append({ 'document': results['documents'][0][i], 'metadata': results['metadatas'][0][i], 'distance': results['distances'][0][i], 'id': results['ids'][0][i] }) return formatted_results def hybrid_search(self, query: str, keyword: str = None, n_results: int = 5): """混合搜索：语义 + 关键词""" # 语义搜索 semantic_results = self.semantic_search(query, n_results * 2) # 如果有关键词，进行过滤 if keyword: filtered_results = [ result for result in semantic_results if keyword.lower() in result['document'].lower() ] # 如果过滤后结果太少，返回原始语义结果 if len(filtered_results) >= n_results // 2: return filtered_results[:n_results] return semantic_results[:n_results] # 测试高级检索 retriever = AdvancedRetriever(vector_db) # 语义搜索 results = retriever.semantic_search("支付方式有哪些", n_results=2) print("语义搜索结果:") for result in results: print(f"- {result['document']} (相似度: {1 - result['distance']:.3f})") # 混合搜索 hybrid_results = retriever.hybrid_search("客服时间", keyword="工作日", n_results=2) print("\n混合搜索结果:") for result in hybrid_results: print(f"- {result['document']}")

5. 大模型集成与提示词工程

5.1 本地大模型部署（Ollama）

import requests import json class LocalLLMClient: def __init__(self, base_url: str = "http://localhost:11434"): self.base_url = base_url self.model_name = "qwen:7b" # 可根据需要调整模型 def generate(self, prompt: str, max_tokens: int = 1000) -> str: """调用本地大模型生成回答""" try: response = requests.post( f"{self.base_url}/api/generate", json={ "model": self.model_name, "prompt": prompt, "stream": False, "options": { "temperature": 0.3, "top_p": 0.9, "max_tokens": max_tokens } }, timeout=60 ) response.raise_for_status() return response.json()["response"] except Exception as e: return f"模型调用失败: {str(e)}" def chat_completion(self, messages: List[Dict]) -> str: """对话式接口""" prompt = self._format_messages(messages) return self.generate(prompt) def _format_messages(self, messages: List[Dict]) -> str: """格式化消息为提示词""" prompt = "" for msg in messages: if msg["role"] == "system": prompt += f"系统指令: {msg['content']}\n\n" elif msg["role"] == "user": prompt += f"用户问题: {msg['content']}\n\n" elif msg["role"] == "assistant": prompt += f"助手回答: {msg['content']}\n\n" return prompt + "助手回答:" # 测试本地模型 llm_client = LocalLLMClient() # 简单测试 test_prompt = "请用中文简要介绍人工智能的发展历史" response = llm_client.generate(test_prompt) print("模型响应:", response)

5.2 提示词模板设计

class PromptEngineer: def __init__(self): self.templates = { "qa": """基于以下上下文信息，请回答用户的问题。如果上下文信息不足以回答问题，请如实告知。 上下文： {context} 问题：{question} 请根据上下文提供准确、简洁的回答：""", "summary": """请根据以下文档内容，生成一个简洁的摘要： 文档内容： {context} 要求： 1. 摘要长度在100-200字之间 2. 突出关键信息点 3. 保持客观准确 摘要：""", "analysis": """请分析以下文本内容，并按要求提供分析结果： 文本内容： {context} 分析要求： {requirements} 请按以下格式回复： 1. 主要观点 2. 支持论据 3. 潜在问题 4. 改进建议 分析结果：""" } def build_qa_prompt(self, context: str, question: str) -> str: """构建问答提示词""" return self.templates["qa"].format( context=context, question=question ) def build_prompt_with_history(self, context: str, question: str, history: List[Dict]) -> str: """构建带历史对话的提示词""" history_text = "" for i, exchange in enumerate(history[-3:]): # 最近3轮对话 history_text += f"第{i+1}轮对话:\n" history_text += f"用户: {exchange['user']}\n" history_text += f"助手: {exchange['assistant']}\n\n" prompt = f"""以下是之前的对话历史： {history_text} 当前问题的上下文信息： {context} 请基于以上信息回答当前问题：{question} 回答：""" return prompt # 提示词工程测试 prompt_engineer = PromptEngineer() context = "公司产品支持多种支付方式，包括支付宝、微信支付和银联。技术支持服务时间为工作日9:00-18:00。" question = "周末能获得技术支持吗？" prompt = prompt_engineer.build_qa_prompt(context, question) print("生成的提示词:") print(prompt)

5.3 完整的RAG问答系统

class RAGSystem: def __init__(self, vector_db: VectorDatabase, llm_client: LocalLLMClient): self.vector_db = vector_db self.llm_client = llm_client self.retriever = AdvancedRetriever(vector_db) self.prompt_engineer = PromptEngineer() self.conversation_history = [] def ask_question(self, question: str, use_history: bool = True) -> Dict: """核心问答方法""" # 1. 检索相关文档 search_results = self.retriever.semantic_search(question, n_results=3) if not search_results: return { "answer": "抱歉，知识库中没有找到相关信息。", "sources": [], "confidence": 0.0 } # 2. 构建上下文 context = "\n\n".join([ f"来源 {i+1}: {result['document']}" for i, result in enumerate(search_results) ]) # 3. 构建提示词 if use_history and self.conversation_history: prompt = self.prompt_engineer.build_prompt_with_history( context, question, self.conversation_history ) else: prompt = self.prompt_engineer.build_qa_prompt(context, question) # 4. 调用大模型生成答案 answer = self.llm_client.generate(prompt) # 5. 更新对话历史 self.conversation_history.append({ "user": question, "assistant": answer }) # 保持历史记录不超过10轮 if len(self.conversation_history) > 10: self.conversation_history = self.conversation_history[-10:] # 6. 返回结果 return { "answer": answer, "sources": [ { "content": result['document'], "metadata": result['metadata'], "similarity": 1 - result['distance'] } for result in search_results ], "confidence": 1 - search_results[0]['distance'] if search_results else 0.0 } def clear_history(self): """清空对话历史""" self.conversation_history = [] # 完整的RAG系统测试 rag_system = RAGSystem(vector_db, llm_client) # 测试问答 questions = [ "支付方式有哪些？", "技术支持时间是什么时候？", "最新版本有什么新功能？" ] for question in questions: print(f"\n问题: {question}") result = rag_system.ask_question(question) print(f"回答: {result['answer']}") print(f"置信度: {result['confidence']:.3f}") print("来源:") for i, source in enumerate(result['sources']): print(f" {i+1}. {source['content'][:50]}... (相似度: {source['similarity']:.3f})")

6. 系统优化与性能调优

6.1 检索效果优化策略

class RetrievalOptimizer: def __init__(self, rag_system: RAGSystem): self.rag_system = rag_system def evaluate_retrieval(self, test_questions: List[str], ground_truth: Dict[str, List[str]]) -> Dict: """评估检索效果""" results = { 'precision': [], 'recall': [], 'mrr': [] # Mean Reciprocal Rank } for question, relevant_docs in ground_truth.items(): if question not in test_questions: continue search_results = self.rag_system.retriever.semantic_search(question, n_results=5) retrieved_docs = [result['document'] for result in search_results] # 计算精确率 relevant_retrieved = len(set(retrieved_docs) & set(relevant_docs)) precision = relevant_retrieved / len(retrieved_docs) if retrieved_docs else 0 results['precision'].append(precision) # 计算召回率 recall = relevant_retrieved / len(relevant_docs) if relevant_docs else 0 results['recall'].append(recall) # 计算MRR for rank, doc in enumerate(retrieved_docs, 1): if doc in relevant_docs: results['mrr'].append(1.0 / rank) break else: results['mrr'].append(0.0) # 计算平均值 avg_metrics = {} for metric, values in results.items(): avg_metrics[f'avg_{metric}'] = sum(values) / len(values) if values else 0 return avg_metrics def optimize_chunk_size(self, test_data: Dict, chunk_sizes: List[int] = [200, 300, 500, 800]) -> Dict: """优化文本切片大小""" best_size = None best_score = 0 results = {} for chunk_size in chunk_sizes: # 重新处理文档（实际应用中需要重新构建向量库） print(f"测试chunk大小: {chunk_size}") # 这里简化演示，实际需要重新切分文档和构建向量库 score = 0.7 # 模拟评估分数 results[chunk_size] = score if score > best_score: best_score = score best_size = chunk_size return { 'best_chunk_size': best_size, 'best_score': best_score, 'all_results': results } # 优化器测试 optimizer = RetrievalOptimizer(rag_system) # 模拟测试数据 test_questions = { "支付方式有哪些？": ["公司产品支持多种支付方式，包括支付宝、微信支付和银联"], "技术支持时间？": ["技术支持服务时间为工作日9:00-18:00，紧急问题可拨打400热线"] } metrics = optimizer.evaluate_retrieval(list(test_questions.keys()), test_questions) print("检索效果评估:", metrics)

6.2 缓存与性能优化

import time from functools import lru_cache import hashlib class PerformanceOptimizer: def __init__(self): self.embedding_cache = {} @lru_cache(maxsize=1000) def get_cached_embedding(self, text: str) -> List[float]: """带缓存的嵌入计算""" text_hash = hashlib.md5(text.encode()).hexdigest() if text_hash in self.embedding_cache: return self.embedding_cache[text_hash] # 计算新嵌入 embedding = embedding_service.encode_text(text) self.embedding_cache[text_hash] = embedding return embedding def benchmark_retrieval(self, rag_system: RAGSystem, questions: List[str], iterations: int = 10) -> Dict: """性能基准测试""" times = [] for i in range(iterations): start_time = time.time() for question in questions: rag_system.ask_question(question, use_history=False) end_time = time.time() times.append(end_time - start_time) avg_time = sum(times) / len(times) qps = len(questions) / avg_time # 每秒处理问题数 return { 'avg_time_per_batch': avg_time, 'questions_per_second': qps, 'min_time': min(times), 'max_time': max(times) } # 性能测试 performance_optimizer = PerformanceOptimizer() # 测试缓存效果 test_text = "这是一个测试文本" start_time = time.time() embedding1 = performance_optimizer.get_cached_embedding(test_text) time1 = time.time() - start_time start_time = time.time() embedding2 = performance_optimizer.get_cached_embedding(test_text) # 应该从缓存获取 time2 = time.time() - start_time print(f"首次嵌入时间: {time1:.4f}s") print(f"缓存嵌入时间: {time2:.4f}s") print(f"加速比: {time1/time2:.1f}x")

7. 生产环境部署方案

7.1 FastAPI Web服务部署

from fastapi import FastAPI, HTTPException from pydantic import BaseModel import uvicorn from typing import List, Optional app = FastAPI(title="RAG知识库API", version="1.0.0") # 请求响应模型 class QuestionRequest(BaseModel): question: str use_history: bool = True n_results: int = 3 class SourceDocument(BaseModel): content: str metadata: dict similarity: float class AnswerResponse(BaseModel): answer: str sources: List[SourceDocument] confidence: float processing_time: float class HealthResponse(BaseModel): status: str model_ready: bool database_ready: bool # 全局RAG系统实例 rag_system = None @app.on_event("startup") async def startup_event(): """启动时初始化RAG系统""" global rag_system try: # 初始化各组件 vector_db = VectorDatabase() vector_db.create_collection("production_knowledge") llm_client = LocalLLMClient() rag_system = RAGSystem(vector_db, llm_client) print("RAG系统初始化完成") except Exception as e: print(f"系统初始化失败: {e}") raise @app.get("/health", response_model=HealthResponse) async def health_check(): """健康检查端点""" if rag_system is None: return HealthResponse( status="unhealthy", model_ready=False, database_ready=False ) # 简单的组件健康检查 try: # 测试向量数据库 rag_system.vector_db.collection.count() db_ready = True except: db_ready = False try: # 测试大模型 rag_system.llm_client.generate("测试") model_ready = True except: model_ready = False status = "healthy" if (db_ready and model_ready) else "degraded" return HealthResponse( status=status, model_ready=model_ready, database_ready=db_ready ) @app.post("/ask", response_model=AnswerResponse) async def ask_question(request: QuestionRequest): """问答接口""" if rag_system is None: raise HTTPException(status_code=503, detail="系统未就绪") start_time = time.time() try: result = rag_system.ask_question( question=request.question, use_history=request.use_history ) processing_time = time.time() - start_time return AnswerResponse( answer=result["answer"], sources=[ SourceDocument( content=source["content"], metadata=source["metadata"], similarity=source["similarity"] ) for source in result["sources"] ], confidence=result["confidence"], processing_time=processing_time ) except Exception as e: raise HTTPException(status_code=500, detail=f"处理问题时出错: {str(e)}") @app.post("/knowledge/upload") async def upload_documents(documents: List[str], metadata: List[dict]): """上传文档到知识库""" try: # 文档预处理和向量化 ids = [f"doc_{hashlib.md5(doc.encode()).hexdigest()[:8]}" for doc in documents] rag_system.vector_db.add_documents(documents, metadata, ids) return {"message": f"成功上传 {len(documents)} 个文档", "ids": ids} except Exception as e: raise HTTPException(status_code=500, detail=f"文档上传失败: {str(e)}") if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000)

7.2 Docker容器化部署

创建Dockerfile：

FROM python:3.10-slim WORKDIR /app # 安装系统依赖 RUN apt-get update && apt-get install -y \ gcc \ g++ \ && rm -rf /var/lib/apt/lists/* # 复制依赖文件 COPY requirements.txt . # 安装Python依赖 RUN pip install --no-cache-dir -r requirements.txt # 复制应用代码 COPY . . # 创建知识库存储目录 RUN mkdir -p /app/chroma_db # 暴露端口 EXPOSE 8000 # 启动命令 CMD ["python", "-m", "uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

创建docker-compose.yml：

version: '3.8' services: rag-api: build: . ports: - "8000:8000" volumes: - ./chroma_db:/app/chroma_db - ./knowledge_docs:/app/knowledge_docs environment: - OLLAMA_HOST=ollama-service:11434 depends_on: - ollama-service ollama-service: image: ollama/ollama:latest ports: - "11434:11434" volumes: - ollama_data:/root/.ollama command: ["serve"] volumes: ollama_data:

7.3 监控与日志配置

import logging from logging.handlers import RotatingFileHandler import json class MonitoringSystem: def __init__(self, log_file: str = "rag_system.log"): self.setup_logging(log_file) self.usage_stats = { "total_questions": 0, "successful_answers": 0, "average_response_time": 0, "error_count": 0 } def setup_logging(self, log_file: str): """配置日志系统""" logger = logging.getLogger("rag_system") logger.setLevel(logging.INFO) # 文件处理器（自动轮转） file_handler = RotatingFileHandler( log_file, maxBytes=10*1024*1024, backupCount=5 ) file_handler.setFormatter(logging.Formatter( '%(asctime)s -