Python+Milvus+Langchain+大语言模型实现智能问答系统

大家好，我是咖啡哥。

今天基于RAG（检索增强生成）架构带大家实现一个智能问答系统。在项目中我们会结合向量数据库、大语言模型和网页内容检索技术，构建一个能够基于网页（或企业私有知识库）内容进行智能问答的端到端解决方案。基于这个方案可以实现传统的搜索引擎无法理解用户意图、无法准确提供上下文回答的核心问题。

一、技术栈分析

在本系统中我们使用了这些核心技术：

后端框架

FastAPI – Python异步Web框架
向量数据库

Milvus – 专业的向量相似性搜索引擎
嵌入模型

gte-large-zh中文嵌入模型
大语言模型

Llama-2-13B
文档处理

LangChain – 文档分割和处理框架
网页解析

BeautifulSoup – HTML内容提取
容器化

Docker Compose – 服务编排和部署

二、项目架构剖析

2.1 架构模式

采用微服务架构思想，通过Docker Compose编排了完整的服务栈，这种架构是云原生应用的最佳实践，实现了服务的松耦合和高可用性：

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.py def 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"]