RAG编程实践（DashScope+Milvus）

目录

• RAG编程实践
  • 一、RAG概述
    • RAG的价值
    • 学习资料
  • 二、系统RAG架构设计
    • 1. 整体架构图
    • 2. 技术选型
  • 三、核心组件
    • 1. 向量生成层：文本到向量的转换
    • 2. 向量存储层：Milvus数据库操作
    • 3. 混合检索实现：RRF排名融合
  • 四、RAG与智能体系统的集成
    • 1. 检索工具封装
    • 2. RAG与智能体的结合
  • 五、RAG完整工作流程
    • 1. 知识库初始化阶段
    • 2. 查询处理阶段
    • 3. 响应处理与优化
  • 六、系统优化
    • 1. 异步架构设计
    • 2. 向量检索优化
    • 3. 知识库维护

RAG编程实践

一、RAG概述

检索增强生成（Retrieval-Augmented Generation，简称RAG）是一种结合了检索系统和生成模型的先进架构，旨在解决大型语言模型（LLM）在知识时效性、准确性和专业性方面的局限性。

RAG的价值

1. 突破知识截止限制：通过外部知识库提供最新信息，解决LLM知识过时问题
2. 减少幻觉生成：基于检索到的准确信息回答问题，降低模型生成错误内容的概率
3. 领域知识专业化：可以导入特定领域的专业知识库，提升回答的专业性
4. 可追溯性：回答可以溯源到具体的知识来源，增强透明度和可信度

学习资料

Deeptoai RAG系列教程
build-rag-with-milvus文档

二、系统RAG架构设计

本文通过一个电商客服系统进行RAG实践演示。

1. 整体架构图

┌──────────────┐     ┌────────────────┐     ┌───────────────────┐     ┌─────────────────┐
│              │     │                │     │                   │     │                 │
│  用户查询     │────▶│  向量生成层     │────▶│  向量存储与检索层   │────▶│  生成增强层      │
│              │     │                │     │                   │     │                 │
└──────────────┘     └────────────────┘     └───────────────────┘     └─────────────────┘│▼┌─────────────────┐│                 ││   智能回答输出    ││                 │└─────────────────┘

2. 技术选型

• 向量模型：通过DashScope API生成稀疏向量和密集向量
• 向量数据库：Milvus，支持混合向量检索
• 索引技术：FLAT索引(密集向量) + SPARSE_INVERTED_INDEX(稀疏向量)
• 相似度计算：余弦相似度(COSINE)和内积(IP)
• 结果融合：RRF（Reciprocal Rank Fusion）排名器

三、核心组件

1. 向量生成层：文本到向量的转换

实现文本到向量的转换功能：

async def query_to_vector(query):"""将用户查询文本转换为稀疏向量与稠密向量，用于后续检索。通过 DashScope 的文本嵌入接口异步调用，避免阻塞事件循环。"""def embedding_sparse(query):# 调用DashScope API获取向量表示resp = dashscope.TextEmbedding.call(api_key=settings.EMBEDDING_API_KEY,model=dashscope.TextEmbedding.Models.text_embedding_v4,input=query,dimension=1024,output_type="dense&sparse",  # 同时获取稀疏和密集向量)if resp.status_code == 200:# 解析稀疏向量：将索引与值映射为字典sparse_vector = {item["index"]: item["value"] for item in resp.output["embeddings"][0]["sparse_embedding"]}# 直接提取稠密向量dense_vector = resp.output["embeddings"][0]["embedding"]return sparse_vector, dense_vectorreturn [], []# 使用asyncio.to_thread将同步调用放到线程池，防止阻塞sparse_vector, dense_vector = await asyncio.to_thread(embedding_sparse, query)return sparse_vector, dense_vector

技术要点：

• 密集向量：捕获语义信息，对同义词、语义相似性表现好
• 稀疏向量：保留关键词和精确匹配信息，对术语、专有名词识别准确
• 异步处理：使用asyncio.to_thread避免API调用阻塞事件循环
• 错误处理：API调用失败时返回空向量，保证系统稳定性

2. 向量存储层：Milvus数据库操作

实现基于Milvus的向量存储和检索功能：

class AsyncMilvusOperator:def __init__(self, collection_name):self.client = client.async_milvus_clientself.collection_name = collection_name# 创建向量数据库schemaasync def create_schema(self):schema = self.client.create_schema(auto_id=True, description="这是通用知识库的参数模型")schema.add_field(field_name="id", datatype=DataType.INT64, is_primary=True, description="主键")schema.add_field(field_name="dense_vector", datatype=DataType.FLOAT_VECTOR, dim=1536,description="这是密集向量")schema.add_field(field_name="sparse_vector", datatype=DataType.SPARSE_FLOAT_VECTOR,description="这是稀疏向量")schema.add_field(field_name="text", datatype=DataType.VARCHAR, max_length=4096, description="这是文本内容")return schema# 创建集合和索引async def create_collection(self):await self.client.create_collection(collection_name=self.collection_name,schema=await self.create_schema(),description="这是通用知识库")index_params = self.client.prepare_index_params()# 为密集向量创建FLAT索引，使用余弦相似度index_params.add_index(field_name="dense_vector", index_type="FLAT", metric_type="COSINE")# 为稀疏向量创建SPARSE_INVERTED_INDEX索引，使用内积index_params.add_index(field_name="sparse_vector", index_type="SPARSE_INVERTED_INDEX", metric_type="IP")await self.client.create_index(collection_name=self.collection_name, index_params=index_params)

技术要点：

• schema设计：支持不同类型的向量和元数据存储
• 索引优化：针对不同类型向量选择最适合的索引类型
• 异步接口：全面异步实现，提高并发性能

3. 混合检索实现：RRF排名融合

最关键的检索功能通过mix_search_vector方法实现：

async def mix_search_vector(self, dense_vector, sparse_vector):# 加载集合到内存await self.client.load_collection(collection_name=self.collection_name)# 密集向量搜索请求req_dense = AnnSearchRequest(data=[dense_vector],anns_field="dense_vector",param={"metric_type": "COSINE"},limit=10,)# 稀疏向量搜索请求req_sparse = AnnSearchRequest(data=[sparse_vector],anns_field="sparse_vector",param={"metric_type": "IP"},limit=10,)# 合并请求并使用RRF排名器reqs = [req_dense, req_sparse]ranker = RRFRanker(k=10)res = await self.client.hybrid_search(collection_name=self.collection_name, reqs=reqs,output_fields=["text"],ranker=ranker)return res

技术要点：

• 混合检索：同时使用两种向量检索，优势互补
• RRF排名：使用倒数排名融合算法整合两个检索结果列表
• 性能优化：加载集合到内存提升检索速度

四、RAG与智能体系统的集成

1. 检索工具封装

将RAG检索功能封装为可重用工具：

class SearchTool(BaseTool):name: str = "search_tool"description: str = "这是通过用户的指令来检索知识库的工具"args_schema: Type[BaseModel] = SearchArgsdef _run(self, query: str) -> str:"""同步执行工具（通过异步包装）"""loop = asyncio.get_event_loop()return loop.run_until_complete(self._arun(query))async def _arun(self, query: str):# 初始化Milvus操作器，指定集合名称milvus_operator = AsyncMilvusOperator(collection_name="general_knowledge")# 生成查询向量sparse_vector, dense_vector = await query_to_vector(query)# 执行混合向量搜索knowledge_info = await milvus_operator.mix_search_vector(dense_vector=dense_vector, sparse_vector=sparse_vector)return knowledge_info

2. RAG与智能体的结合

将检索工具集成到通用智能体中：

class GeneralAgent:def __init__(self):self.prompt = """**# 角色**你是电商平台的**通用客服助手**。你负责处理无法归类到其他专业智能体的一般性问题..."""async def generate_general_agent(self, messages):# 创建智能体，将SearchTool作为可用工具传入order_agent = create_react_agent(model=client.async_openai_chat, tools=[SearchTool()], prompt=self.prompt)# 执行智能体推理result = await order_agent.ainvoke({"messages": messages})return result

五、RAG完整工作流程

1. 知识库初始化阶段

1. 创建数据库表：基于MySQL创建知识库元数据表
2. 创建向量集合：应用启动时初始化Milvus集合和索引
3. 知识导入：
   • 收集和整理知识文档
   • 将文档切分为适当大小的文本块
   • 使用query_to_vector生成向量表示
   • 通过insert_vector方法将向量和文本存储到Milvus

2. 查询处理阶段

1. 用户查询接收：通过FastAPI端点接收用户查询
2. 智能体路由：RouterAgent将请求路由到相应专业智能体
3. 向量生成：将用户查询转换为稀疏向量和密集向量
4. 混合检索：执行混合向量搜索并使用RRF排名器合并结果
5. 知识增强生成：
   • 将检索到的知识作为上下文提供给LLM
   • LLM基于增强上下文生成回答

3. 响应处理与优化

1. 智能体处理：专业智能体利用检索到的知识生成精准回答
2. 记忆保存：将交互保存到用户记忆系统
3. 结果返回：格式化并返回最终回答

六、系统优化

1. 异步架构设计

• 全面使用异步处理，提高系统并发能力
• 使用线程池避免API调用阻塞事件循环

2. 向量检索优化

• 预加载集合到内存提升检索速度
• 合理设置检索参数（limit、排名策略等）
• 针对不同类型内容选择合适的向量模型

3. 知识库维护

• 定期更新知识库内容，确保信息时效性
• 监控检索效果，优化向量模型和检索参数
• 建立反馈机制，持续改进知识库质量