一、技术栈分析
在本系统中我们使用了这些核心技术：
-
后端框架 FastAPI – Python异步Web框架 -
向量数据库 Milvus – 专业的向量相似性搜索引擎 -
嵌入模型 gte-large-zh中文嵌入模型 -
大语言模型 Llama-2-13B -
文档处理 LangChain – 文档分割和处理框架 -
网页解析 BeautifulSoup – HTML内容提取 -
容器化 Docker Compose – 服务编排和部署
二、项目架构剖析
2.1 架构模式
-
etcd: 分布式键值存储，用于Milvus集群元数据管理
-
minio: 对象存储服务，存储向量数据和索引文件
-
milvus-standalone: 向量数据库主服务
-
attu: Milvus的Web管理界面
2.2 架构模式
2.2.1 数据索引阶段
网站URL → 网页抓取 → HTML解析 → 文本分割 → 向量化 → 存储到Milvus
2.2.2 数据检索阶段
用户查询 → 向量化 → 相似性搜索 → 知识库检索 → LLM生成答案
2.3 多索引器设计模式
为了不局限一种向量数据库，我们采用了策略模式支持多向量数据库，增强系统的扩展性：
-
MilvusIndexer 专门针对Milvus优化的索引器 -
ElasticIndexer 支持Elasticsearch的索引器 -
Indexer 通用索引器基类
fmt = "n=== {:30} ===n"search_latency_fmt = "search latency = {:.4f}s"num_entities, dim = 8, 8milvus_collection_name = "ai_answer"class SearchEngine: def __init__(self): connections.connect("default", host="localhost", port="19530") has = utility.has_collection(milvus_collection_name) print(f"Does collection milvus_collection_name exist in Milvus: {has}") milvus_client = Collection(milvus_collection_name) self.milvus_client = milvus_client self.milvus_collection_name = milvus_collection_name openai.api_key = os.environ["OPENAI_API_KEY"] def query_milvus(self, embedding): result = self.milvus_client.search([embedding], "vector", {"metric_type": "L2", "offset": 1}, 1, None, None, ["id", "vector", "path", "text"]) list_of_knowledge_base = list(map(lambda match: match.entity.text, result[0])) return { 'list_of_knowledge_base': list_of_knowledge_base, } def query_vector_db(self, embedding): return self.query_milvus(embedding)
三、核心业务逻辑解析
3.1 智能索引
# indexer_by_milvus.py def add_html_to_vectordb(self, content, path): text_splitter = RecursiveCharacterTextSplitter( chunk_size = self.MODEL_CHUNK_SIZE, # 8192 chunk_overlap = math.floor(self.MODEL_CHUNK_SIZE/10) # 819 ) docs = text_splitter.create_documents([content]) for doc in docs: embedding = create_embedding(doc.page_content) self.insert_embedding(embedding, doc.page_content, path)def insert_embedding(self, embedding, text, path): try: print(fmt.format("Start inserting entities")) data = [ {"vector": np.array(embedding), "text": text, "path": path}, ] self.milvus_client.insert(data) except Exception as e: print("self.milvus_client.insert exception e: ", e) os._exit(1)
一些最佳实践说明：
-
智能分块策略 8192字符的分块大小配合10%重叠的设计（819字符重叠），确保重要信息不会在分块边界丢失，既保证了模型处理效率，又维护了上下文的连贯性 -
语义保持优化 使用递归字符分割器，按照自然的文本边界（句子、段落）进行分割， 在8192字符的分块过程中，如何保持语义完整性是一个关键
挑战：硬切分可能破坏句子和段落的完整性，使用LangChain框架的RecursiveCharacterTextSplitter，按照句子、段落等语义单位进行智能分割 -
批处理设计 每个文档片段独立处理，支持大文档的流式处理 -
数据类型转换 将Python列表转换为NumPy数组，提高向量计算效率
3.2 向量检索
# search_engine.pydef search(self, user_query):print("user_query: ", user_query)embedding = create_embedding(user_query)result = self.query_vector_db(embedding)knowledge_base = "n".join(result['list_of_knowledge_base'])response = self.ask_chatgpt(knowledge_base, user_query)return {'response': response}def ask_chatgpt(self, knowledge_base, user_query):system_content = """你是一个专业的智能问答助手，请严格遵循以下规则：1. 只能基于提供的知识库内容回答问题，不得使用知识库以外的信息；2. 如果知识库中没有相关信息或无法找到准确答案，请明确告知用户"我无法在知识库中找到相关信息来回答这个问题"；3. 回答时要客观准确，不得编造或推测信息；4. 尽量使用知识库中的原始表述，确保信息的准确性和权威性。"""user_content = f"""Knowledge Base!---{knowledge_base}---User Query: {user_query}Answer: {user_history_answer}"""system_message = {"role": "system", "content": system_content}user_message = {"role": "user", "content": user_content}chatgpt_response = create_llama2_13b(messages=[system_message, user_message])return chatgpt_response["choices"][0]["message"]["content"]
一些最佳实践说明：
-
完整的RAG工作流 检索（Retrieval）→ 增强（Augmentation）→ 生成（Generation）的标准流程 -
防幻觉设计 明确的系统提示词限制模型只基于知识库回答，避免模型一本正经的胡说八道 -
上下文构建 通过结构化的prompt模板，清晰地分离知识库内容和用户查询 -
错误边界处理 当知识库中没有相关信息时，明确告知用户"不知道答案"
