基于LangChain和PAI-EAS搭建RAG知识库安全系统_人工智能平台 PAI(PAI)-阿里云帮助中心

RAG（Retrieval-Augmented Generation）知识库包含企业重要的私域或专有数据，这些数据以明文形式存储在向量数据库中，并在网络传输过程中以明文形式存在于提示词中，存在数据泄露的隐患。因此，为确保数据安全，需要采用加密等技术，对存储在向量数据库中的文本块（chunk）和嵌入向量进行加密。当需要执行推理操作时，您可以在PAI-EAS提供的安全环境中解密数据，系统会将解密后的数据和用户问题输入给大语言模型（LLM）进行推理。本文为您介绍如何基于LangChain与阿里云PAI-EAS搭建RAG知识库安全系统。

背景信息

本方案基于EAS和LangChain搭建RAG知识库安全系统，以下介绍各模块功能：

EAS简介
EAS（Elastic Algorithm Service）是PAI的模型在线服务平台，支持将模型部署为在线推理服务和AI-Web应用。在本方案中，EAS将提供安全的解密和推理环境。关于EAS更详细的内容介绍，请参见EAS模型服务概述。
RAG简介
RAG通过将非结构化的外部知识库（如文档）与大模型相结合，弥补模型在知识专业性和时效性上的不足。在生成内容前先检索相关信息，RAG在确保数据安全的前提下，充分融合了领域知识和私有数据，从而减少生成不确定性。RAG系统通常包含以下三个流程：
- 索引：通过处理多种来源的知识库，提取其中文本并将其切分为标准长度的文本块（chunk），然后生成嵌入向量（embedding），最后将文本块（chunk）和嵌入向量一起存储到向量数据库中。
- 检索：将用户输入的查询（query）转化为向量表示，通过相似度匹配从向量数据库中检索出最相关的文本块（chunk）。
- 生成：将检索到的相关文本和用户问题一起放入提示词（prompt）中，输入到大语言模型（LLM）生成精确且具备上下文关联的内容。
LangChain简介
LangChain是一个开源框架，帮助开发者将大语言模型（LLM）与外部工具（数据库、API、文档等）连接，构建智能应用。它通过模块化组件（链、代理、记忆系统）实现任务编排与上下文管理，支持智能客服、数据分析等场景，突破单一模型的输入输出限制，扩展AI实际应用能力。

前提条件

创建专有网络VPC、交换机和安全组。具体操作，请参见搭建IPv4专有网络和创建安全组。
创建DSW实例，并为该实例选择专有网络。本方案将以DSW环境为例作为代码执行环境，您可以按照以下操作步骤进入DSW开发环境，或者选择使用其他环境。
1. 在交互式建模（DSW）页面，单击目标实例操作列下的打开。
2. 在Notebook页签，单击创建Notebook。

部署PAI-EAS模型服务

PAI-EAS服务提供解密和推理的安全环境。在部署服务前，请参照控制台快速入门创建对象存储OSS Bucket和相关目录，用来存储大语言模型文件和代码文件。

大语言模型文件：用于模型推理。以Qwen2.5-3B-Instruct模型为例，下载模型文件并将其上传到OSS目录（例如oss://examplebucket/ragtest/Qwen/Qwen2.5-3B-Instruct）中。您可以使用ossutil工具，或其他自己的方法上传模型文件。
代码文件：该代码基于Flask框架实现了Web服务应用程序，主要功能是使用rai_sam解密加密的文本内容，并将其输入到大语言模型（LLM）中进行推理。下载并更新代码文件（app.py）中的以下参数，并将代码文件上传到OSS目录（oss://examplebucket/ragtest/code/）中。
- sam_key_sets：定义知识库密钥ID（长度为4 ~ 48字节）和密钥（长度为4 ~ 128字节）。用来加密和解密知识库。
- model_name：配置为大语言模型文件在EAS容器中的绝对路径。例如/mnt/data/Qwen/Qwen2.5-3B-Instruct，您的路径以实际为准。

参照以下操作步骤部署PAI-EAS服务：

登录PAI控制台，在页面上方选择目标地域，并在右侧选择目标工作空间，然后单击进入EAS。
在模型在线服务（EAS）页面，单击部署服务，然后在自定义模型部署区域，单击自定义部署。

在自定义部署页面，配置以下参数，参数配置完成后，单击部署。

当服务状态为运行中时，表明服务已成功部署。

参数		描述

参数		描述
基本信息	服务名称	自定义服务名称，例如test_rag_sec。
环境信息	部署方式	选择镜像部署。
	镜像配置	选择官方镜像>python-inference:3.9-ubuntu2004。
	模型配置	通过对象存储OSS挂载大语言模型（LLM），单击OSS，并配置以下参数： OSS：选择大语言模型文件所在的对象存储OSS目录。例如`oss://examplebucket/ragtest/`，您的目录以实际为准。挂载路径：配置为`/mnt/data/`。
	运行命令	配置为`python /mnt/code/app.py`。
	端口号	配置为8000。
	代码配置	通过对象存储OSS挂载代码文件，单击OSS，并配置以下参数： OSS：选择代码文件所在的对象存储OSS目录。例如`oss://examplebucket/ragtest/code/`，您的目录以实际为准。挂载路径：配置为`/mnt/code/`。
	三方库配置	在三方库列表下的文本编辑框中，配置执行代码所需的三方库`Flask==3.0.3 modelscope==1.21.0 transformers==4.45.2 torch==2.1.0 accelerate==0.26.0 rai_sam==1.0.0`。说明本方案已验证上述三方库版本的兼容性。如果替换成其他版本，可能会出现版本不兼容的问题。
资源部署	资源类型	选择公共资源。
资源部署	部署资源	本方案以Qwen2.5-3B-Instruct模型为例，资源规格选择ecs.gn7i-c16g1.4xlarge。使用其他模型时，请根据显存需求调整资源规格。
专有网络	专有网络（VPC）	选择与向量检索库一致的专有网络、交换机和安全组。
	交换机
	安全组名称

在模型在线服务（EAS）页面，单击目标服务的服务方式列下的调用信息，查询EAS服务的访问地址和TOKEN。
说明
您可以选择使用公网地址或VPC内网地址。如果使用VPC内网地址，调用客户端必须与EAS服务位于同一个专有网络内。

创建RAG知识库并构建索引

本方案以PAI-DSW开发环境为例，请在Notebook中按顺序依次执行以下操作步骤创建RAG知识库。

1.准备工作

安装运行代码所需依赖。

!pip install -U langchain langchain-community langchain_huggingface
!pip install -U pypdf
!pip install -U modelscope
!pip install -U dashscope
!pip install -U pymilvus
!pip install -U alibabacloud-ha3engine-vector
!pip install -U elasticsearch
!pip install -U rai_sam

下载嵌入模型，以便后续将知识库内容转换成向量。

from modelscope import snapshot_download
model_dir = snapshot_download('BAAI/bge-large-zh-v1.5', cache_dir='.')

配置知识库的加密密钥。需要与PAI-EAS代码文件app.py中配置的知识库密钥相同。

import os

# SAM内容和向量加密的密钥
os.environ["SAM_KEY_ID"] = "LD_ID_123456"
os.environ["SAM_KEY_SECRET"] = "LD_Secret_0123456789"

2.加载RAG知识库文件并解析分块

加载RAG知识库（例如大模型安全实践白皮书2024.pdf）。执行代码前，请将知识库文件上传到执行代码的环境中。本方案以DSW开发环境为例，具体上传方法请参见上传与下载数据文件。
```
from langchain_community.document_loaders import PyPDFLoader

doc_name = "./大模型安全实践白皮书2024.pdf"

file_path = (
    doc_name
)

loader = PyPDFLoader(file_path)
docs = loader.load()

print(len(docs))
```

RAG文本解析分块，即将已加载的文档对象进一步拆分成更小的文本块（chunk）。

from langchain.text_splitter import RecursiveCharacterTextSplitter

text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
split_docs = text_splitter.split_documents(docs)

print(len(split_docs))

3.RAG文本向量生成和加密

将RAG文本块（chunk）转换成嵌入向量，并对嵌入向量和原始文本内容进行加密。

import os
from langchain_huggingface import HuggingFaceEmbeddings
from rai_sam.engine.packager.content import SamContentPackager
from rai_sam.engine.packager.vector import SamVectorPackager

# Load embedding model
embeddings_model = HuggingFaceEmbeddings(model_name="./BAAI/bge-large-zh-v1.5")

# Split documents into chunks
documents = []
for i in range(len(split_docs)):
    documents.append(split_docs[i].page_content)

# embedding chunks
embeddings = embeddings_model.embed_documents(documents)
print(len(embeddings), len(embeddings[0]))

# embedding dimension
embedding_dimension = len(embeddings[0])

# embedding Vectors Encryption
vector_packager = SamVectorPackager()
enc_embeddings = vector_packager.SamPkgEncryptVectors(os.getenv("SAM_KEY_ID"), os.getenv("SAM_KEY_SECRET"), embeddings)

# Document Contents Encryption
content_packager = SamContentPackager()
contents = []
for i in range(len(split_docs)):
    contents.append(split_docs[i].page_content)
enc_contents = content_packager.SamPkgEncryptContents(os.getenv("SAM_KEY_ID"), os.getenv("SAM_KEY_SECRET"), contents)

# Indexed data for vector database
data = [ {"id": i, 
          "vector": enc_embeddings[i],
          "content": enc_contents[i],
          "metadata": split_docs[i].metadata["source"],
          "key_id": os.getenv("SAM_KEY_ID")
         } for i in range(len(embeddings))
       ]
print(len(data))

4.构建RAG知识库索引

准备向量检索库，用于存储加密后的RAG知识库向量。请根据实际使用的数据库，参考以下链接创建向量数据库并准备好相关配置项。创建向量数据库时，请选择与EAS服务一致的专有网络。

将加密后的知识库向量存储到向量数据库中，构建RAG知识库索引。请根据实际使用的数据库选择执行以下对应代码块。

基于阿里云Milvus向量数据库的索引构建

基于阿里云OpenSearch向量数据库的索引构建

基于阿里云ElasticSearch向量数据库的索引构建

from pymilvus import MilvusClient

demo_collection_name = "milvus_demo_collection"

client = MilvusClient(
    uri="http://c-xxxx-internal.milvus.aliyuncs.com:19530",
    token="User:Password",
    db_name="default"
)

if client.has_collection(demo_collection_name):
    client.drop_collection(demo_collection_name)

client.create_collection(
    collection_name=demo_collection_name,
    dimension=embedding_dimension
)

res = client.insert(
    collection_name=demo_collection_name,
    data=data
)
print(res)

其中关键配置说明如下：

demo_collection_name：Milvus实例的Collection名称，例如milvus_demo_collection。
uri：Milvus实例的访问地址，格式为http://<内网地址>:<port>。
token：格式为User:Password，即Milvus实例用户名:Milvus实例密码。
db_name：配置为已创建的数据库名称，例如default。

from alibabacloud_ha3engine_vector import models, client
from Tea.exceptions import TeaException, RetryError

# 实例ID
instance_id = "ha-cn-xxx"

# 表名称
table_name = "OPS_demo"

# 实例的域名，实例的用户名和密码
Config = models.Config(
    endpoint=instance_id + ".public.ha.aliyuncs.com",
    instance_id=instance_id,
    protocol="http",
    access_user_name="root",
    access_pass_word="YourPassword"
)

ha3EngineClient = client.Client(Config)

try:
    documentArrayList = []
    for i in range(len(data)):
        add2Document = {
            "fields": data[i],
            "cmd": "add"
        }
        documentArrayList.append(add2Document)

    print(len(documentArrayList))

    optionsHeaders = {}
    pushDocumentsRequest = models.PushDocumentsRequest(optionsHeaders, documentArrayList)

    pkField = "id"
    response = ha3EngineClient.push_documents(instance_id + "_" + table_name, pkField, pushDocumentsRequest)

    print(response.body)
except TeaException as e:
    print(f"send request with TeaException : {e}")
except RetryError as e:
    print(f"send request with Connection Exception  : {e}")

其中关键配置说明如下：

instance_id：配置为OpenSearch的实例ID。
table_name：配置为OpenSearch索引表名称。
access_user_name：配置为OpenSearch实例的用户名。
access_pass_word：配置为OpenSearch实例的密码。

from elasticsearch import Elasticsearch

index_name = "elasticsearch_demo"

index_config = {
    "mappings": {
        "properties": {
            "vector": {
                "type": "dense_vector",
                "dims": embedding_dimension,
                "similarity": "cosine"
            },
            "content": {
                "type": "text"
            },
           "metadata": {
                "type": "text"
            },
            "key_id": {
                "type": "text"
            }
        }
    }
}

client = Elasticsearch(
        '<Elasticsearch URL>',
        basic_auth=('elastic', '<YourPassword>')
)

exists = client.indices.exists(index=index_name)
if exists == False:
    result = client.indices.create(index=index_name, body=index_config)
    print(result)
else:
    print("{0} has existed".format(index_name))


for i in range(len(data)):
    document = data[i]
    client.index(
        index=index_name,
        id = document['id'],
        document=document
)
print("Documents indexed successfully")

其中关键配置说明如下：

index_name：在Elasticsearch实例页面，更新YML文件配置为允许自动创建索引后，即可自定义索引名称。
<Elasticsearch URL>：配置为Elasticsearch实例访问地址，格式为http://<私网地址>:<私网端口>。
<YourPassword>：配置为Elasticsearch实例的登录密码。

索引构建完成后，您可以前往对应向量数据库中查看已创建的索引。具体操作，请参见Elasticsearch：快速访问与配置、milvus：Attu工具管理或Opensearch：查询数据。

向量检索和推理

本方案以PAI-DSW开发环境为例，请在Notebook中按顺序依次执行以下步骤，进行向量检索和推理。

1.准备工作

安装运行代码所需依赖。

!pip install -U langchain langchain-community langchain_huggingface
!pip install -U pymilvus
!pip install -U alibabacloud-ha3engine-vector
!pip install -U elasticsearch
!pip install -U rai_sam

配置加密密钥，在进行向量检索时，用于解密存储在数据库中的加密向量。该密钥必须与知识库密钥相同。
```
import os

os.environ["SAM_KEY_ID"] = "LD_ID_123456"
os.environ["SAM_KEY_SECRET"] = "LD_Secret_0123456789"
```
配置PAI-EAS服务的访问地址和TOKEN。
```
os.environ["EAS_SERVICE_URL"] = "<service_url>/model"
os.environ["EAS_SERVICE_TOKEN"] = "<service_token>"
```
其中<service_url>和<service_token>需替换为EAS服务的访问地址和TOKEN。参考查询访问地址和TOKEN，进行获取。

2.用户输入向量转化和加密

执行以下代码，将用户查询内容转换为向量并进行加密。

from langchain_huggingface import HuggingFaceEmbeddings
from rai_sam.engine.packager.vector import SamVectorPackager

query = "大模型安全建设的指导思想是什么?"

# Load embedding model and embedding query
embeddings_model = HuggingFaceEmbeddings(model_name="./BAAI/bge-large-zh-v1.5")
query_embedding = embeddings_model.embed_query(query)

# Query embedding encryption
vector_packager = SamVectorPackager()
enc_query_embedding = vector_packager.SamPkgEncryptVectors(os.getenv("SAM_KEY_ID"), os.getenv("SAM_KEY_SECRET"), [query_embedding])[0]

print(len(enc_query_embedding))

3.检索相关知识片段

在向量数据库中检索相关的知识片段。请根据实际使用的数据库选择执行以下对应代码块。

基于阿里云Milvus向量数据库的知识检索

基于阿里云OpenSearch向量数据库的知识检索

基于阿里云ElasticSearch向量数据库的索引构建

from pymilvus import MilvusClient

demo_collection_name = "milvus_demo_collection"

client = MilvusClient(
    uri="http://c-xxx-internal.milvus.aliyuncs.com:19530",
    token="User:YourPassword",
    db_name="default"
)

# Use encrypted query embedding to retrieve
search_res = client.search(
    collection_name=demo_collection_name,
    data=[enc_query_embedding],
    limit=3,
    output_fields=["content", "key_id"],
)

# Print retrieved results
for res in search_res[0]:
    print("Index:", res["id"])
    print("Distance:", res["distance"])
    print("Content:", res["entity"]["content"])
    print("KeyID:", res["entity"]["key_id"])
    print("\n")

retrieved_contents = [
    (res["entity"]["content"]) for res in search_res[0]
]
key_ids = [
    (res["entity"]["key_id"]) for res in search_res[0]
]

其中关键配置说明如下：

demo_collection_name：Milvus实例的Collection名称，例如milvus_demo_collection。
uri：Milvus实例的访问地址，格式为http://<内网地址>:<port>。
token：格式为User:Password，即Milvus实例用户名:Milvus实例密码。
db_name：配置为已创建的数据库名称，例如default。

import json
from alibabacloud_ha3engine_vector import models, client
from alibabacloud_ha3engine_vector.client import Client
from alibabacloud_ha3engine_vector.models import Config
from alibabacloud_ha3engine_vector.models import FetchRequest, QueryRequest

# 实例ID
instance_id = "ha-cn-xxx"

# 表名称
table_name = "OPS_demo"

# 实例的域名，实例的用户名和密码
Config = models.Config(
    endpoint=instance_id + ".public.ha.aliyuncs.com",
    instance_id=instance_id,
    protocol="http",
    access_user_name="root",
    access_pass_word="YourPassword"
)

ha3EngineClient = client.Client(Config)

# Use encrypted query embedding to retrieve
request = QueryRequest(
    table_name=table_name,
    vector=enc_query_embedding,
    search_params="{\\\"qc.searcher.scan_ratio\\\":0.01}",
    top_k=3,
    output_fields=["content", "key_id"],
    sort = "__vs_vector_score__")

response = ha3EngineClient.query(request)
search_res = json.loads(response.body)

# Print retrieved results
for res in search_res['result']:
    print("Index:", res["id"])
    print("Distance:", res["score"])
    print("Content:", res["fields"]["content"])
    print("KeyId:", res["fields"]["key_id"])
    print("\n")

retrieved_contents = [
    (res["fields"]["content"]) for res in search_res['result']
]
key_ids = [
    (res["fields"]["key_id"]) for res in search_res['result']
]

其中关键配置说明如下：

instance_id：配置为OpenSearch的实例ID。
table_name：配置为OpenSearch索引表名称。
access_user_name：配置为OpenSearch实例的用户名。
access_pass_word：配置为OpenSearch实例的密码。

from elasticsearch import Elasticsearch

index_name = "elasticsearch_demo"

client = Elasticsearch(
        '<Elasticsearch URL>',
        basic_auth=('elastic', '<YourPassword>')
)
# Use encrypted query embedding to retrieve
response = client.search(
    index = index_name,
    query = {
        "knn": {
            "field": "vector",
            "query_vector": enc_query_embedding,
            "k": 3
        }
    },
    fields=["content", "key_id"]
)
search_res = response["hits"]

# Print retrieved results
for res in search_res["hits"]:
    print("Index:", res["_id"])
    print("Score:", res["_score"])
    print("Content:", res["_source"]["content"])
    print("KeyId:", res["_source"]["key_id"])
    print("\n")

retrieved_contents = [
    (res["_source"]["content"]) for res in search_res["hits"]
]
key_ids = [
    (res["_source"]["key_id"]) for res in search_res["hits"]
]

其中关键配置说明如下：

index_name：Elasticsearch实例的索引名称。
<Elasticsearch URL>：配置为Elasticsearch实例访问地址，格式为http://<私网地址>:<私网端口>。
<YourPassword>：配置为Elasticsearch实例的登录密码。

4.使用PAI-EAS进行知识片段解密和推理

使用PAI-EAS解密知识片段，然后将解密后的内容和相关问题输入到大语言模型（LLM）中进行推理，生成推理结果。

import os
import json
from langchain import hub
from langchain_community.llms.pai_eas_endpoint import PaiEasEndpoint
from langchain_core.output_parsers import StrOutputParser

rai_context_start = "<|rai_sam_encrypted_context_start|>"
rai_context_end = "<|rai_sam_encrypted_context_end|>"

llm = PaiEasEndpoint(
    eas_service_url=os.getenv("EAS_SERVICE_URL"),
    eas_service_token=os.getenv("EAS_SERVICE_TOKEN"),
)

prompt = hub.pull("rlm/rag-prompt")

chain = prompt | llm | StrOutputParser()

contents = {
    "contents": retrieved_contents,
    "key_ids": key_ids
}

content_str = json.dumps(contents)
context = rai_context_start + content_str + rai_context_end

print("context: ", context)
print("\n")

response = chain.invoke({"context": context, "question": query})
print(response)

单击此处，查看代码执行示例结果，您的结果以实际为准

context:  <|rai_sam_encrypted_context_start|>{"contents": [******], "key_ids": ["LD_ID_****", "LD_ID_****", "LD_ID_****"]}<|rai_sam_encrypted_context_end|>
大模型安全建设的核心指导思想是以人为本，确保技术发展既符合伦理道德，又能为人类社会带来积极影响。这意味着在大模型的技术和应用过程中，始终将人的利益、需求和安全置于首位。以人为本的理念要求所有参与者，包括设计者、开发者和使用者，都要保持这种思维，并切实保障用户和社会的安全与利益。如果偏离这一核心，可能会导致安全风险和挑战，引发诸如侵犯隐私、社会不公平及伦理道德冲突等不可预见的问题。