基于 AnalyticDB 与通义千问搭建 AI 智能客服

一、引言

基于检索增强生成（Retrieval-Augmented Generation, RAG）技术的 AI 智能客服能够高效地检索企业私域知识库，并利用大语言模型理解问题的上下文和意图，生成准确、贴切的答案。开发者将企业私域知识上传到智能问答系统后，企业业务人员就能通过提问快速获取公司政策、操作流程、专业知识等信息，客户也能快速得到产品知识、售后问题的答案。

本方案基于 AnalyticDB for PostgreSQL 的高效向量引擎与阿里云自主研发的通义千问 LLM 模型，构建一个高性能的检索增强生成（RAG）应用，实现企业的 AI 智能客服。

二、搭建 AI 智能客服方案概览

大模型虽然可以理解并生成自然语言，但由于缺乏特定领域的专业知识、没有企业的私域知识、知识更新不及时等问题，无法直接用于企业的 AI 智能客服。为了解决这一问题，企业可以采用 RAG 技术，从外部知识库（企业的私域知识）检索知识。召回的知识将和原始查询融合成 prompt，为大模型提供更丰富的上下文信息，从而生成更加准确的回答。

1. 方案优势

简单易用：只需简单的页面点击，即可实现知识检索增强，使大模型有更丰富的上下文信息并生成更准确的答案。同时，本方案提供智能客服的示例代码，方便您快速体验 AI 智能客服。
灵活且安全管理：向量数据存储在 AnalyticDB for PostgreSQL，企业可灵活管理数据。配套的审计、权限管理等功能可满足企业安全合规需求。
性价比高：完成本方案的部署及体验，预计产生费用较低（具体取决于资源规格与运行时间）。

2. 方案架构

在阿里云上搭建的云上私有网络环境包括以下基础设施和云服务：

专有网络 VPC：将云原生数据仓库 AnalyticDB PostgreSQL 版部署在专有网络中。
交换机：将云服务器 ECS 实例和云原生数据仓库 AnalyticDB PostgreSQL 版实例部署在交换机中，实现它们之间的网络通信。
云服务器 ECS：用于部署智能客服应用及 Python 环境。
云原生数据仓库 AnalyticDB PostgreSQL 版实例：提供向量数据存储和检索服务。
百炼平台：一站式的企业专属大模型生产平台。在本方案中，提供通义千问 LLM 模型的调用以及知识索引功能。

三、部署准备

开始部署前，请按以下指引完成账号申请、账号充值和授权。

1. 准备账号

如果您还没有阿里云账号，请访问阿里云账号注册页面，根据页面提示完成注册并进行个人实名认证。阿里云账号是您使用云资源的付费实体，因此是部署方案的必要前提。

2. 账号充值

本方案的云资源支持按量付费，且默认设置均采用按量付费引导操作。如果确定任何一个云资源采用按量付费方式部署，账户余额都必须大于等于 100 元。

四、一键部署

资源编排（ROS）可以让您通过 YAML 或 JSON 文件清晰简洁地描述所需的云资源及其依赖关系，然后自动化地创建和配置这些资源。您可以通过下方提供的 ROS 一键部署链接，来自动化地完成这些资源的创建和配置。

部署 1 个专有网络 VPC。
部署 1 台交换机。
部署 1 台云服务器 ECS，并在服务器上部署 Python3.9 环境和 AI 智能客服的示例代码。
部署 1 个云原生数据仓库 AnalyticDB PostgreSQL 版实例。

五、核心代码实现

为了帮助您更好地理解 RAG 流程，以下提供基于 Python 的核心逻辑示例。该代码展示了如何连接 AnalyticDB 进行向量检索，并调用通义千问 API 生成回答。

import os
from aliyunsdkcore.client import AcsClient
 aliyunsdkdashscope.api_call  api_call
 dashscope  Embedding
 dashscope  Generation


DASHSCOPE_API_KEY = os.getenv()
ADB_HOST = os.getenv()
ADB_PORT = os.getenv()
ADB_USER = os.getenv()
ADB_PASSWORD = os.getenv()


 ():
    response = Embedding.call(
        model=,
        =text
    )
     response.status_code == :
         response.output.embeddings[].embedding
    :
         Exception()


 ():
    
    
    ()
    
     [] 


 ():
    response = Generation.call(
        model=,
        messages=[
            {: , : },
            {: , : prompt}
        ]
    )
     response.status_code == :
         response.output.choices[].message.content
    :
         


 __name__ == :
    user_query = 
    
    
    query_vec = get_embedding(user_query)
    
    
    retrieved_docs = search_vector(query_vec)
    
    
    context = .join([doc[]  doc  retrieved_docs])
    final_prompt = 
    
    
    answer = generate_answer(final_prompt)
    (answer)