多源 RAG 处理方案：为 AI 场景提供高效数据管道

创建对象存储OSS Bucket

接下来您需要创建1个对象存储OSS Bucket，用于存储本方案相关资源。

1. 登录OSS管理控制台。

2. 在左侧导航栏，选择Bucket列表。

3. 在Bucket列表页面，单击创建Bucket。

4. 在创建Bucket面板，创建1个Bucket。

   项目	说明	示例值
   Bucket名称	Bucket 命名规范： 命名长度为3~63个字符。 只允许小写字母、数字、短横线（-），且不能以短横线开头或结尾。 Bucket名称在OSS范围内必须全局唯一。	test-bucket
   地域	地域选择华南1（深圳）。	华南1（深圳）
   存储类型	存储类型 ，创建成功后不支持修改。本方案以标准存储为例。	标准存储
   存储冗余类型	选择采用多可用区（AZ）冗余机制的同城冗余存储，将用户的数据分散存放在同一地域的3个可用区。当某个可用区不可用时，仍然能够保障数据的正常访问。	同城冗余存储
   读写权限	设置数据访问权限。	私有

创建向量检索服务Cluster和Collection

1. 登录向量检索服务控制台。

2. 在左侧导航栏单击Cluster列表，单击创建Cluster，配置以下信息：

   1. 商品类型选择按量付费。

   2. 地域选择华南1（深圳）。

   3. 实例类型和实例规格：新用户可以选择免费试用。除新用户外，建议选择存储型的S.small规格，副本数设置为1。

   4. 自定义Cluster名称。

3. 单击立即购买。勾选服务协议后，单击立即开通。等待Cluster创建完成。

4. 在左侧导航栏点击Cluster 列表，在顶部选择地域，找到已创建的Cluster实例，单击操作列的创建Collection。

   1. 自定义Collection名称。

   2. 选择单向量设置。

   3. 向量维度设置为1536。

   4. 距离度量方式选择Cosine。

   5. 展开高级设置，在定义Schema区域，点击新增字段，字段名称填写raw，数据类型选择STRING。

5. 单击确定，等待Collection创建完成。

获取向量检索服务Cluster Endpoint

1. 在左侧导航栏点击Cluster 列表，在顶部选择地域，找到已创建的Cluster实例，点击Cluster名称跳转到Cluster详情页。

2. 在访问端口区域，获取外网访问的Endpoint值，记录下来，后续配置中会用到。

   

创建事件流并配置Source（源）

1. 登录EventBridge控制台。

2. 在左侧导航栏单击事件流。

3. 在顶部选择地域为华南1（深圳）。

4. 单击创建事件流：

   1. 自定义任务名称。

   2. 计费模式选择按事件量计费。

5. 在任务创建页签的Source (源)区域，配置以下信息：

   1. 数据提供方选择对象存储 OSS。

   2. OSS Bucket选择之前已创建的OSS Bucket数据源。

   3. 文档加载选择TextLoader。

   4. 加载模式选择分块加载，分块标识符填写\n，实现将数据源文本按照换行符切分为多段。

   5. 角色配置：点击授权创建新角色，在弹出的访问控制快速授权页面，确保已选择权限策略中包含AliyunOSSReadOnlyAccess，单击确认授权，根据页面提示完成授权。返回当前配置页面，选择已创建的角色。

      

6. 点击下一步。

配置Filtering（过滤）

1. 在Filtering（过滤）区域，配置以下信息：

   1. 模式内容选择匹配全部事件。实际业务场景中，也可选择其他过滤规则，或选择自定义匹配来灵活定义过滤规则。

      

2. 点击下一步。

配置Transform（转换）

1. 在Transform（转换）区域，配置以下信息：

   1. 选择阿里云服务为函数计算。

   2. 使用新建函数模板创建函数，函数模板选择内容向量化。

   3. 函数代码中，将{your DASHSCOPE_API_KEY}替换为之前获取的百炼API-KEY，替换时注意保留''，不要保留{}。其余代码不变。

      

   说明

   • 本例中，经过数据转换后，输出的数据结构如下：

     {
       "status_code":200,
       "request _id":"xxx-xxxx-xxx",
       "output":{
         "embeddings":[
           {
             "embedding":[
               -0.446533203125,
               2.4246826171875,
               ...
             ],
             "text_index":0
           }
         ]
       },
       "usage":{
         "total_tokens":3
       },
       "content":"<进行embedding的数据>"
     }

   • 实际场景中可以根据业务需求修改代码，灵活定制数据转换的方式；或者选择绑定现有函数，使用已部署在函数计算上的函数来执行数据转换。

2. 点击下一步。

配置Sink（目标）

1. 在Sink（目标）区域，配置以下信息：

   1. 服务类型选择向量检索服务。

   2. DashVector Cluster选择已创建的Cluster。

   3. DashVector Collection选择已创建的Collection。

   4. 数据映射选择Upsert。

   5. 向量填写通过上游函数处理转换后得到的向量信息 $.output.embeddings[0].embedding 。

   6. Partition选择固定值Default。

   7. 属性列中，对于属性列名称raw，配置属性值为部分事件$.content。

      说明

      基于上一步Transform的输出结果，按JsonPath提取所需内容。在RAG场景中，建议配置属性值，以便在RAG召回时找到原始关联数据。

   8. 鉴权配置填写之前获取的向量检索服务API KEY。

   9. 网络配置选择公网网络。

      

2. 点击保存。等待事件流创建成功，状态应显示运行中。

验证数据处理结果

创建Web函数

1. 登录函数计算控制台。

2. 在左侧导航栏单击函数管理 > 函数列表，地域选择华南1（深圳）。

3. 单击创建函数，选择Web函数，单击创建Web函数。

   项目	示例值	说明
   函数名称	test-rag	只能包含字母、数字、下划线和中划线。不能以数字、中划线开头。长度在 1-64 之间。
   规格方案	vCPU 0.35，内存 512 MB，磁盘 512 MB	函数实例的规格。
   最小实例数	0	如果一段时间内函数未被调用，则仅保留最小实例数的实例。
   单实例并发度	20	单个函数实例最多同时处理的请求数。
   运行环境	自定义运行时 / Python / Python 3.10	函数代码的运行时环境。
   代码上传方式	当前使用示例代码。后续创建函数后，再使用实际的业务代码来替代。	选择函数代码的上传方式。
   启动命令	python3 app.py	指定函数实例启动时的执行命令。
   监听端口	9000	函数实例监听请求的端口。
   执行超时时间	600	如果函数在这个时间内未能成功执行，函数计算会返回超时错误，请设置大小合适的超时时间，避免函数执行超时。

4. 点击创建，等待函数创建成功。

配置函数运行时环境

在函数计算控制台，点击已创建的Web函数（注意和之前创建的用于处理数据流的事件函数区分），进入函数详情页。在配置页签的高级配置区域，点击编辑。

编写函数代码

1. 在已创建的Web函数详情页，在代码页签，将app.py文件内容替换为以下代码：

   from flask import Flask
   from flask import request
   
   import os
   
   import dashscope
   from dashscope import TextEmbedding, Generation
   
   from dashvector import Client, Doc
   
   REQUEST_ID_HEADER = 'x-fc-request-id'
   
   app = Flask(__name__)
   
   dashscope_api_key = os.getenv('DASHSCOPE_API_KEY')
   dashvector_api_key = os.getenv('DASHVECTOR_API_KEY')
   dashvector_cluster_endpoint = os.getenv('dashvector_cluster_endpoint')
   collection_name = os.getenv('collection_name')
   
   @app.route('/', defaults={'path': ''})
   @app.route('/<path:path>', methods=['GET', 'POST', 'PUT', 'DELETE'])
   def handler(path):
       rid = request.headers.get(REQUEST_ID_HEADER)
       print("FC Invoke Start RequestId: " + rid)
       # print("Path: " + path)
       # print("Data: " + str(data))
       # print("FC Invoke End RequestId: " + rid)
       data = request.stream.read()
       question = str(data.decode('utf-8'))
       print("User Input: " + question)
       answer = answer_question(question, search_relevant_info(question))
       return answer
   
   def answer_question(question, context):
       prompt = f'''请基于```内的内容回答问题。
           ```
           {context}
           ```
           我的问题是：{question}。
       '''
       rsp = Generation.call(model='qwen-turbo', prompt=prompt)
       return rsp.output.text
   
   def search_relevant_info(question):
       # 初始化 dashvector client
       client = Client(
           api_key = dashvector_api_key,
           endpoint = dashvector_cluster_endpoint
       )
   
       # 获取刚刚存入的集合
       collection = client.get(collection_name)
       assert collection
   
       # 向量检索：指定 topk = 1
       rsp = collection.query(generate_embeddings(question), output_fields=['raw'],
                              topk=1)
       assert rsp
       relevant_info = rsp.output[0].fields['raw']
       print("relevant_info: ", relevant_info)
       return relevant_info
   
   def generate_embeddings(question):
       rsp = TextEmbedding.call(
           model=TextEmbedding.Models.text_embedding_v1,
           input=question
       )
       embeddings = [record['embedding'] for record in rsp.output['embeddings']]
       return embeddings if isinstance(question, list) else embeddings[0]
   
   
   if __name__ == '__main__':
           app.run(host='0.0.0.0',port=9000)
   

2. 点击部署代码，等待代码部署成功。

验证效果

1. 在已创建的Web函数详情页，在代码页签，点击测试函数旁边的下拉框，点击配置测试参数。

   

2. 创建新测试事件，事件模板选择hello-world，自定义事件名称，在页面下方的代码框中输入以下测试文本，点击确定：

   百炼X1手机的分辨率是多少？

3. 点击测试函数。页面显示执行成功。展开详细信息，在返回结果页签中，查看大模型根据客户问题，查询知识库后生成的回答。在日志输出页签中，查看向量检索召回的文本。

   

   

4. 验证RAG应用对不同问题的回答和召回效果：更换测试参数中的测试文本，然后重新执行测试。