SiliconCloud x Milvus:高效构建高性能RAG系统
检索增强生成（RAG）系统被设计用来提升大型语言模型（LLM）的响应质量，能显著减少LLM的幻觉。当用户提交查询时，RAG系统从向量数据库中提取相关信息，然后通过Embedding、Reranker模型，提升内容相关度与精细度，最终通过LLM为用户生成准确的响应。
在RAG流水线中，Milvus（https://milvus.io） 这样的高性能向量数据库在检索最相关数据方面起着至关重要的作用，能帮助LLM生成上下文相关且信息充分的回应。Milvus能够执行快速且精确的相似度搜索，使其成为需要快速访问大型数据集任务中的理想选择。该向量数据库由Zilliz团队开发和维护，已为全球许多AI应用提供支持。
作为集合顶尖大模型的一站式云服务平台，硅基流动SiliconCloud（https://cloud.siliconflow.cn/）致力于为开发者提供极速响应、价格亲民、品类齐全、体验丝滑的模型API服务。平台已上架数十种大语言模型、向量&重排序模型、图片/视频生成大模型（包括20+免费模型），免去开发者模型部署门槛，自由切换适合不同应用场景的模型，高效提升你的使用体验。
现在，开发者可利用Milvus以及硅基流动 SiliconCloud上的Qwen2.5、DeepSeek V2.5等语言模型以及BCE与BGE等Embedding、Reranker模型构建属于自己的完整RAG系统。
本教程将分享如何使用Milvus和硅基流动SiliconCloud构建RAG（检索增强生成）流水线。
准备工作
1. 环境设置和依赖项
pip install --upgrade pymilvus openai requests tqdmSiliconCloud提供类似于OpenAI的API密匙（https://docs.siliconflow.cn/quickstart）。你可以在其官方网站登录，然后准备一个API密匙SILICON_FLOW_API_KEY作为环境变量。
import os
os.environ["SILICON_FLOW_API_KEY"] = "***********"2. 准备数据
我们将使用Milvus文档2.4.x（https://github.com/milvus-io/milvus-docs/releases/download/v2.4.6-preview/milvus_docs_2.4.x_en.zip）中的FAQ页面作为RAG的专有知识，这对于简单的RAG流水线来说是很好的数据源。
下载zip文件并将其解压缩到名为milvus_docs的文件夹中。
wget https://github.com/milvus-io/milvus-docs/releases/download/v2.4.6-preview/milvus_docs_2.4.x_en.zip
unzip -q milvus_docs_2.4.x_en.zip -d milvus_docs我们从milvus_docs/en/faq文件夹中加载所有Markdown文件。对于每个文档，我们使用“#”作为分隔符来大致分割Markdown文件中的各个主要部分。
from glob import glob
text_lines = []
for file_path in glob("milvus_docs/en/faq/*.md", recursive=True):
with open(file_path, "r") as file:
file_text = file.read()
text_lines += file_text.split("# ")3. 准备嵌入模型
初始化客户端以准备嵌入模型。SiliconCloud支持OpenAI形式的API，因此你能够使用相同接口，并结合必要调整，来调用嵌入模型和服务。
from openai import OpenAI
siliconflow_client = OpenAI(
api_key=os.environ["SILICON_FLOW_API_KEY"], base_url="https://api.siliconflow.cn/v1"
)定义函数，使用客户端生成文本嵌入。我们选用BAAI/bge-large-en-v1.5模型作为例子。
def emb_text(text):
return (
siliconflow_client.embeddings.create(input=text, model="BAAI/bge-large-en-v1.5")
.data[0]
.embedding
)创建嵌入测试，并输出其维度和前部元素。
test_embedding = emb_text("This is a test")
embedding_dim = len(test_embedding)
print(embedding_dim)
print(test_embedding[:10])1024
[0.011475468054413795, 0.02982141077518463, 0.0038535362109541893, 0.035921916365623474, -0.0159175843000412, -0.014918108470737934, -0.018094222992658615, -0.002937349723652005, 0.030917132273316383, 0.03390815854072571]将数据加载到Milvus
1. 创建集合
from pymilvus import MilvusClient
milvus_client = MilvusClient(uri="./milvus_demo.db")
collection_name = "my_rag_collection"注意
有关MilvusClient参数：
- 将URI设置为本地文件，例如./milvus.db是最便捷的方法，因为这将自动使用Milvus Lite在文件中保存所有数据。
- 如果你有大量数据，可以设置一个Milvus服务器，它能在Docker或Kubernetes上更有效的运行。在这个设置中，请使用服务器的URI，例如 http://localhost:19530。
- 如果你想使用Zilliz Cloud，即Milvus的完全管理云服务，请调整URI和token，它们对应Zilliz Cloud中的公共端点和API密钥（https://docs.zilliz.com/docs/on-zilliz-cloud-console#free-cluster-details）。
在添加新集合前，请确认当前集合不存在，如存在则删除以避免重复创建。
if milvus_client.has_collection(collection_name):
milvus_client.drop_collection(collection_name)创建具有指定参数的新集合。
如果未指定任何字段信息，Milvus将自动创建默认id字段作为主key，并且创建Vector字段来存储向量数据。保留JSON字段将用于存储未定义架构字段及其数值。
milvus_client.create_collection(
collection_name=collection_name,
dimension=embedding_dim,
metric_type="IP", # Inner product distance
consistency_level="Strong", # Strong consistency level
)2. 插入数据
遍历文本行以创建嵌入，并将生成数据插入Milvus。
在此过程中，我们将引入一个新的text字段，其在集合模式中属于未定义字段。该字段将被自动追加到保留JSON动态字段中，在高级别操作中可被处理为常规字段。
from tqdm import tqdm
data = []
for i, line in enumerate(tqdm(text_lines, desc="Creating embeddings")):
data.append({"id": i, "vector": emb_text(line), "text": line})
milvus_client.insert(collection_name=collection_name, data=data)Creating embeddings: 100%|██████████| 72/72 [00:04<00:00, 16.97it/s]
{'insert_count': 72, 'ids': [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71], 'cost': 0}构建RAG系统
1.获取查询所需数据
让我们来明确一个关于Milvus的常见问题。
question = "How is data stored in milvus?"在集合中搜索问题，并检索语义上的前三个匹配项。
search_res = milvus_client.search(
collection_name=collection_name,
data=[
emb_text(question)
], # Use the `emb_text` function to convert the question to an embedding vector
limit=3, # Return top 3 results
search_params={"metric_type": "IP", "params": {}}, # Inner product distance
output_fields=["text"], # Return the text field
)让我们来看看查询的搜索结果。
import json
retrieved_lines_with_distances = [
(res["entity"]["text"], res["distance"]) for res in search_res[0]
]
print(json.dumps(retrieved_lines_with_distances, indent=4))[
[
" Where does Milvus store data?\n\nMilvus deals with two types of data, inserted data and metadata. \n\nInserted data, including vector data, scalar data, and collection-specific schema, are stored in persistent storage as incremental log. Milvus supports multiple object storage backends, including [MinIO](https://min.io/), [AWS S3](https://aws.amazon.com/s3/?nc1=h_ls), [Google Cloud Storage](https://cloud.google.com/storage?hl=en#object-storage-for-companies-of-all-sizes) (GCS), [Azure Blob Storage](https://azure.microsoft.com/en-us/products/storage/blobs), [Alibaba Cloud OSS](https://www.alibabacloud.com/product/object-storage-service), and [Tencent Cloud Object Storage](https://www.tencentcloud.com/products/cos) (COS).\n\nMetadata are generated within Milvus. Each Milvus module has its own metadata that are stored in etcd.\n\n###",
0.833885133266449
],
[
"How does Milvus flush data?\n\nMilvus returns success when inserted data are loaded to the message queue. However, the data are not yet flushed to the disk. Then Milvus' data node writes the data in the message queue to persistent storage as incremental logs. If `flush()` is called, the data node is forced to write all data in the message queue to persistent storage immediately.\n\n###",
0.812842607498169
],
[
"Does the query perform in memory? What are incremental data and historical data?\n\nYes. When a query request comes, Milvus searches both incremental data and historical data by loading them into memory. Incremental data are in the growing segments, which are buffered in memory before they reach the threshold to be persisted in storage engine, while historical data are from the sealed segments that are stored in the object storage. Incremental data and historical data together constitute the whole dataset to search.\n\n###",
0.7714196443557739
]
]2.使用LLM构建RAG系统响应
将检索到的文档转换为字符串格式。
context = "\n".join(
[line_with_distance[0] for line_with_distance in retrieved_lines_with_distances]
)定义语言模型的系统和用户提示词。这个提示词是由Milvus检索到的文档组成。
SYSTEM_PROMPT = """
Human: You are an AI assistant. You are able to find answers to the questions from the contextual passage snippets provided.
"""
USER_PROMPT = f"""
Use the following pieces of information enclosed in <context> tags to provide an answer to the question enclosed in <question> tags.
<context>
{context}
</context>
<question>
{question}
</question>
"""我们可以使用SiliconCloud提供的各种大语言模型，这里以deepseek-ai/DeepSeek-V2.5模型为例，根据提示词生成响应。
response = siliconflow_client.chat.completions.create(
model="deepseek-ai/DeepSeek-V2.5",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": USER_PROMPT},
],
)
print(response.choices[0].message.content)In Milvus, data is stored in two main categories: inserted data and metadata.
- **Inserted Data**: This includes vector data, scalar data, and collection-specific schema, which are stored in persistent storage as incremental logs. Milvus supports various object storage backends such as MinIO, AWS S3, Google Cloud Storage (GCS), Azure Blob Storage, Alibaba Cloud OSS, and Tencent Cloud Object Storage (COS).至此，利用高性能向量数据库Milvus，以及大模型云服务平台SiliconCloud的语言模型、Embedding、Reranker等各类API服务，我们已经成功构建了RAG流水线，这将为你的生成式AI应用带来更高效、准确的内容输出。
Milvus文档：https://blog.milvus.io/docs/build_RAG_with_milvus_and_siliconflow.md
让超级产品开发者实现“Token自由”
邀请好友体验SiliconCloud，狂送2000万Token/人
邀请越多，Token奖励越多
siliconflow.cn/zh-cn/siliconcloud
加入用户交流群请后台私信