AnalyticDB：基於OpenAPI構建RAG應用快速入門

免費試用

阿里雲提供儲存彈性模式執行個體的免費試用活動，如果您是AnalyticDB PostgreSQL版的新使用者，您可以訪問阿里雲免費試用申請試用資格。如果沒有免費試用資格，按照本文操作步驟在控制台建立執行個體。

建立執行個體

1. 登入雲原生資料倉儲AnalyticDB PostgreSQL版控制台。

2. 單擊頁面右上方的新建實例，進入執行個體購買頁面。

3. 在執行個體購買頁面，配置核心參數快速完成執行個體選型，其他參數保持預設即可。如需瞭解更多參數資訊，請參見建立執行個體。

   配置項	說明	本教程樣本
   商品類型	訂用帳戶：屬於預付費，即在建立執行個體時需要支付費用。適合長期需求，價格比隨用隨付更實惠，且購買時間長度越長，折扣越多。 隨用隨付：屬於後付費，即按小時計費。適合短期需求，用完可立即釋放執行個體，節省費用。	隨用隨付
   地區和可用性區域	執行個體所在的地理位置。 購買後無法更換，建議與需要串連的ECS執行個體建立於同一個地區，以實現內網互連。	華東1（杭州）：華東1可用性區域J
   執行個體資源類型	儲存彈性模式：支援獨立磁碟擴容，支援線上平滑擴容。 Serverless Pro：只需指定需要的計算資源，無需預留儲存資源。	儲存彈性模式
   引擎版本	推薦選擇7.0標準版，以獲得更豐富的功能體驗。同時支援6.0標準版。	7.0標準版
   執行個體系列	高效能（基礎版）：適用於大部分業務分析情境。 高可用版：建議企業核心業務採用高可用版本。	高效能（基礎版）
   向量引擎最佳化	選擇開啟。	開啟
   專用網路	選擇Virtual Private Cloud的ID。 如需與同地區下的ECS執行個體內網互連，則需選擇與ECS相同的VPC。可選擇已有VPC或根據頁面提示建立VPC和交換器。	vpc-xxxx
   虛擬交換器	選擇專用網路下的交換器。如果沒有可選的交換器，說明該可用性區域暫無可用交換器資源。您可以考慮更換至其他可用性區域，或根據頁面提示在當前可用性區域內建立交換器。	vsw-xxxx

4. 單擊立即購買，確認訂單資訊並單擊立即開通。

5. 支付完成後，可單擊管理主控台返回執行個體列表查看建立執行個體。

   說明

   AnalyticDB PostgreSQL版執行個體初始化需要一定時間，待執行個體列表中的執行個體運行狀態顯示為運行中，才可進行後續操作。

建立初始帳號

1. 在左側導覽列中，單擊帳號管理。

2. 單擊建立初始帳號。在建立帳號視窗中，填寫帳號名稱並設定密碼。然後單擊確定。

   配置	說明
   資料庫帳號	初始帳號的名稱，限制如下： 由小寫字母，數字和底線組成。 以小寫字母開頭，小寫字母或數字結尾。 不能以gp開頭。 長度為2~16個字元。
   新密碼、確認密碼	初始帳號的密碼，限制如下： 由大寫字母、小寫字母、數字、特殊字元其中三種及以上組成。 支援的特殊字元包括!@#$%^&*()_+-=。 長度為8~32個字元。 重要 為保障資料安全，建議您定期更換密碼。不要使用曾經用過的密碼。

準備開發環境

1. 檢查Python環境。

   本教程基於Python3版本的SDK。請執行以下命令，確認是否已安裝Python3.9或以上版本，並安裝pip。

   如未安裝或版本不符合要求，請安裝Python。

   python -V
   pip --version

2. 安裝SDK。

   您需要安裝alibabacloud_gpdb20160503和alibabacloud_tea_openapi 的SDK，用於身份認證和構建用戶端。本教程使用的SDK版本如下。

   pip install --upgrade alibabacloud_gpdb20160503 alibabacloud_tea_openapi

3. 配置環境變數。

   將身份認證資訊、執行個體ID等敏感資訊配置到環境變數，避免寫入程式碼造成的資訊泄露。

   Linux&macOS

   1. 執行vim ~/.bashrc 命令，開啟~/.bashrc 檔案。

      macOS系統，請執行vim ~/.bash_profile。

   2. 在設定檔中添加以下內容。

      在RAM存取控制使用者列表頁面，單擊使用者名稱稱，擷取RAM使用者的AccessKey ID和AccessKey Secret。

      在雲原生資料倉儲AnalyticDB PostgreSQL版控制台查看執行個體ID和地區ID。

      # 用RAM使用者的AccessKey ID替換access_key_id
      export ALIBABA_CLOUD_ACCESS_KEY_ID="access_key_id"
      
      # 用RAM使用者的AccessKey Secret替換access_key_secret
      export ALIBABA_CLOUD_ACCESS_KEY_SECRET="access_key_secret"
      
      # 用AnalyticDB for PostgreSQL的執行個體ID替換instance_id，例如gp-bp166cyrtr4p*****
      export ADBPG_INSTANCE_ID="instance_id"
      
      # 用AnalyticDB for PostgreSQL的執行個體所在地區的地區ID替換instance_region，例如cn-hangzhou
      export ADBPG_INSTANCE_REGION="instance_region"

      在vim編輯器中，按下Esc，輸入:wq儲存並退出編輯器。

   3. 執行source ~/.bashrc 命令，使設定檔生效。

      macOS系統，請執行source ~/.bash_profile。

   Windows

   在當前會話中臨時設定環境變數，請在CMD中執行以下命令。

   # 用RAM使用者的AccessKey ID替換access_key_id
   set ALIBABA_CLOUD_ACCESS_KEY_ID=access_key_id
   
   # 用RAM使用者的AccessKey Secret替換access_key_secret
   set ALIBABA_CLOUD_ACCESS_KEY_SECRET=access_key_secret
   
   # 用AnalyticDB for PostgreSQL的執行個體ID替換instance_id，例如gp-bp166cyrtr4p*****
   set ADBPG_INSTANCE_ID=instance_id
   
   # 用AnalyticDB for PostgreSQL的執行個體所在地區的地區ID替換instance_region，例如cn-hangzhou
   set ADBPG_INSTANCE_REGION=instance_region

準備資料庫環境

from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_gpdb20160503.client import Client
from alibabacloud_gpdb20160503 import models as gpdb_20160503_models
import os

# --- 從環境變數中擷取認證和執行個體資訊 ---
ALIBABA_CLOUD_ACCESS_KEY_ID = os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID']
ALIBABA_CLOUD_ACCESS_KEY_SECRET = os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
ADBPG_INSTANCE_ID = os.environ['ADBPG_INSTANCE_ID']
ADBPG_INSTANCE_REGION = os.environ['ADBPG_INSTANCE_REGION']

# 構建並返回一個AnalyticDB for PostgreSQL的API用戶端
def get_client():
    config = open_api_models.Config(
        access_key_id=ALIBABA_CLOUD_ACCESS_KEY_ID,
        access_key_secret=ALIBABA_CLOUD_ACCESS_KEY_SECRET
    )
    config.region_id = ADBPG_INSTANCE_REGION
    # https://api.aliyun.com/product/gpdb
    if ADBPG_INSTANCE_REGION in ("cn-beijing", "cn-hangzhou", "cn-shanghai", "cn-shenzhen", "cn-hongkong",
                                 "ap-southeast-1"):
        config.endpoint = "gpdb.aliyuncs.com"
    else:
        config.endpoint = f'gpdb.{ADBPG_INSTANCE_REGION}.aliyuncs.com'
    return Client(config)
    

# 初始化向量資料庫
def init_vector_database(account, account_password):
    request = gpdb_20160503_models.InitVectorDatabaseRequest(
        region_id=ADBPG_INSTANCE_REGION,
        dbinstance_id=ADBPG_INSTANCE_ID,
        manager_account=account,
        manager_account_password=account_password
    )
    response = get_client().init_vector_database(request)
    print(f"init_vector_database response code: {response.status_code}, body:{response.body}")

# 建立一個命名空間（Namespace）
def create_namespace(account, account_password, namespace, namespace_password):
    request = gpdb_20160503_models.CreateNamespaceRequest(
        region_id=ADBPG_INSTANCE_REGION,
        dbinstance_id=ADBPG_INSTANCE_ID,
        manager_account=account,
        manager_account_password=account_password,
        namespace=namespace,
        namespace_password=namespace_password
    )
    response = get_client().create_namespace(request)
    print(f"create_namespace response code: {response.status_code}, body:{response.body}")

# 建立一個文檔集合（Document Collection）
def create_document_collection(account,
                               account_password,
                               namespace,
                               collection,
                               metadata: str = None,
                               full_text_retrieval_fields: str = None,
                               parser: str = None,
                               embedding_model: str = None,
                               metrics: str = None,
                               hnsw_m: int = None,
                               pq_enable: int = None,
                               external_storage: int = None,):
    request = gpdb_20160503_models.CreateDocumentCollectionRequest(
        region_id=ADBPG_INSTANCE_REGION,
        dbinstance_id=ADBPG_INSTANCE_ID,
        manager_account=account,
        manager_account_password=account_password,
        namespace=namespace,
        collection=collection,
        metadata=metadata,
        full_text_retrieval_fields=full_text_retrieval_fields,
        parser=parser,
        embedding_model=embedding_model,
        metrics=metrics,
        hnsw_m=hnsw_m,
        pq_enable=pq_enable,
        external_storage=external_storage
    )
    response = get_client().create_document_collection(request)
    print(f"create_document_collection response code: {response.status_code}, body:{response.body}")



if __name__ == '__main__':
    # AnalyticDB PostgreSQL版執行個體的資料庫初始帳號。
    account = "testacc"
    # 初始帳號對應的密碼。
    account_password = "Test1234"
    # 要建立的Namespace名稱。
    namespace = "ns1"
    # Namespace對應的密碼，後續資料讀寫使用此密碼。
    namespace_password = "ns1password"
    # 要建立的文件庫名稱。
    collection = "dc1"
    metadata = '{"title":"text", "page":"int"}'
    full_text_retrieval_fields = "title"
    embedding_model = "m3e-small"

    init_vector_database(account, account_password)
    create_namespace(account, account_password, namespace, namespace_password)
    create_document_collection(account,account_password, namespace, collection, 
                               metadata=metadata, full_text_retrieval_fields=full_text_retrieval_fields, 
                               embedding_model=embedding_model)

文件管理

1. 上傳文檔。

   本文以通過非同步方式上傳本地文檔為例，範例程式碼如下：

   import time
   import io
   from typing import Dict, List, Any
   from alibabacloud_tea_util import models as util_models
   from alibabacloud_gpdb20160503 import models as gpdb_20160503_models
   
   def upload_document_async(
           namespace,
           namespace_password,
           collection,
           file_name,
           file_path,
           metadata: Dict[str, Any] = None,
           chunk_overlap: int = None,
           chunk_size: int = None,
           document_loader_name: str = None,
           text_splitter_name: str = None,
           dry_run: bool = None,
           zh_title_enhance: bool = None,
           separators: List[str] = None):
       with open(file_path, 'rb') as f:
           file_content_bytes = f.read()
       request = gpdb_20160503_models.UploadDocumentAsyncAdvanceRequest(
           region_id=ADBPG_INSTANCE_REGION,
           dbinstance_id=ADBPG_INSTANCE_ID,
           namespace=namespace,
           namespace_password=namespace_password,
           collection=collection,
           file_name=file_name,
           metadata=metadata,
           chunk_overlap=chunk_overlap,
           chunk_size=chunk_size,
           document_loader_name=document_loader_name,
           file_url_object=io.BytesIO(file_content_bytes),
           text_splitter_name=text_splitter_name,
           dry_run=dry_run,
           zh_title_enhance=zh_title_enhance,
           separators=separators,
       )
       response = get_client().upload_document_async_advance(request, util_models.RuntimeOptions())
       print(f"upload_document_async response code: {response.status_code}, body:{response.body}")
       return response.body.job_id
   
   
   def wait_upload_document_job(namespace, namespace_password, collection, job_id):
       def job_ready():
           request = gpdb_20160503_models.GetUploadDocumentJobRequest(
               region_id=ADBPG_INSTANCE_REGION,
               dbinstance_id=ADBPG_INSTANCE_ID,
               namespace=namespace,
               namespace_password=namespace_password,
               collection=collection,
               job_id=job_id,
           )
           response = get_client().get_upload_document_job(request)
           print(f"get_upload_document_job response code: {response.status_code}, body:{response.body}")
           return response.body.job.completed
       while True:
           if job_ready():
               print("successfully load document")
               break
           time.sleep(2)
   
   
   if __name__ == '__main__':
       job_id = upload_document_async("ns1", "Ns1password", "dc1",
                                      "test.pdf", "/root/test.pdf")
       wait_upload_document_job("ns1", "Ns1password", "dc1", job_id)

   參數說明

   參數	說明
   namespace	文件庫所在的Namespace名稱。
   namespace_password	Namespace的密碼。
   collection	文檔要存入的文件庫名稱。
   file_name	文檔名稱，帶有類型尾碼。
   file_path	本地的文檔路徑，檔案最大200 MB。
   metadata	文檔的中繼資料，需要和建立文件庫時指定的中繼資料一致。
   chunk_overlap	處理大型資料的切分策略。在分塊處理時，連續的塊之間重疊的資料量，最大值不能超過chunk_size。
   chunk_size	處理大型資料的切分策略，資料拆分成較小的部分時每個塊的大小，最大值為2048。
   document_loader_name	建議無需指定，會按照副檔名自動匹配載入器。自動匹配載入器的詳情，請參見文檔理解。
   text_splitter_name	切分器名稱。文檔切分詳情，請參見文檔切分。
   dry_run	是否只進行文檔理解和切分，不進行向量化和入庫。取值說明： true：只進行文檔理解和切分。 false（預設）：先進行文檔理解和切分，然後進行向量化和入庫。
   zh_title_enhance	是否開啟中文標題加強。取值說明： true：開啟中文標題加強。 false：關閉中文標題加強。
   separators	處理大型資料切分策略的分隔字元，一般無需指定。

2. （可選）其他文件管理操作。

   查看文檔列表

   def list_documents(namespace, namespace_password, collection):
       request = gpdb_20160503_models.ListDocumentsRequest(
           region_id=ADBPG_INSTANCE_REGION,
           dbinstance_id=ADBPG_INSTANCE_ID,
           namespace=namespace,
           namespace_password=namespace_password,
           collection=collection,
       )
       response = get_client().list_documents(request)
       print(f"list_documents response code: {response.status_code}, body:{response.body}")
   
   
   if __name__ == '__main__':
       list_documents("ns1", "Ns1password", "dc1")

   參數說明

   參數	說明
   namespace	文件庫所在的Namespace名稱。
   namespace_password	Namespace的密碼。
   collection	文件庫名稱。

   查看文檔詳情

   def describe_document(namespace, namespace_password, collection, file_name):
       request = gpdb_20160503_models.DescribeDocumentRequest(
           region_id=ADBPG_INSTANCE_REGION,
           dbinstance_id=ADBPG_INSTANCE_ID,
           namespace=namespace,
           namespace_password=namespace_password,
           collection=collection,
           file_name=file_name
       )
       response = get_client().describe_document(request)
       print(f"describe_document response code: {response.status_code}, body:{response.body}")
   
   
   if __name__ == '__main__':
       describe_document("ns1", "Ns1password", "dc1", "test.pdf")

   參數說明

   參數	說明
   namespace	文件庫所在的Namespace名稱。
   namespace_password	Namespace的密碼。
   collection	文件庫名稱。
   file_name	文檔名稱。

   返回資訊

   參數	說明
   DocsCount	文檔被切分的塊數量。
   TextSplitter	文檔切分器名稱。
   DocumentLoader	文檔Loader名稱。
   FileExt	文檔副檔名。
   FileMd5	文檔MD5 HASH值。
   FileMtime	文檔最新上傳時間。
   FileSize	檔案大小，單位為位元組。
   FileVersion	文檔版本，INT類型，代表此文檔被上傳更新了多少次。

   刪除文檔

   def delete_document(namespace, namespace_password, collection, file_name):
       request = gpdb_20160503_models.DeleteDocumentRequest(
           region_id=ADBPG_INSTANCE_REGION,
           dbinstance_id=ADBPG_INSTANCE_ID,
           namespace=namespace,
           namespace_password=namespace_password,
           collection=collection,
           file_name=file_name
       )
       response = get_client().delete_document(request)
       print(f"delete_document response code: {response.status_code}, body:{response.body}")
   
   
   if __name__ == '__main__':
       delete_document("ns1", "Ns1password", "dc1", "test.pdf")

   參數說明

   參數	說明
   namespace	文件庫所在的Namespace名稱。
   namespace_password	Namespace的密碼。
   collection	文件庫名稱。
   file_name	文檔名稱。

文檔檢索

本節以使用純文字檢索為例，範例程式碼如下：

def query_content(namespace, namespace_password, collection, top_k,
                  content,
                  filter_str: str = None,
                  metrics: str = None,
                  use_full_text_retrieval: bool = None):
      request = gpdb_20160503_models.QueryContentRequest(
          region_id=ADBPG_INSTANCE_REGION,
          dbinstance_id=ADBPG_INSTANCE_ID,
          namespace=namespace,
          namespace_password=namespace_password,
          collection=collection,
          content=content,
          filter=filter_str,
          top_k=top_k,
          metrics=metrics,
          use_full_text_retrieval=use_full_text_retrieval,
      )
      response = get_client().query_content(request)
      print(f"query_content response code: {response.status_code}, body:{response.body}")


if __name__ == '__main__':
    query_content('ns1', 'Ns1password', 'dc1', 10, 'ADBPG是什嗎？')

參數說明

返回資訊

整合LangChain

LangChain是一套基於大語言模型（LLM）構建應用的開源架構，可實現通過一整套介面和工具將模型和外部資料串連。下文將展示如何將AnalyticDB PostgreSQL版的檢索能力整合到LangChain中實現一個問答系統。

1. 安裝模組。

   pip install --upgrade langchain openai tiktoken

2. 構建AdbpgRetriever。

   from langchain_core.retrievers import BaseRetriever
   from langchain_core.callbacks import CallbackManagerForRetrieverRun
   from langchain_core.documents import Document
   
   class AdbpgRetriever(BaseRetriever):
       namespace: str = None
       namespace_password: str = None
       collection: str = None
       top_k: int = None
       use_full_text_retrieval: bool = None
   
       def query_content(self, content) -> List[gpdb_20160503_models.QueryContentResponseBodyMatchesMatchList]:
           request = gpdb_20160503_models.QueryContentRequest(
               region_id=ADBPG_INSTANCE_REGION,
               dbinstance_id=ADBPG_INSTANCE_ID,
               namespace=self.namespace,
               namespace_password=self.namespace_password,
               collection=self.collection,
               content=content,
               top_k=self.top_k,
               use_full_text_retrieval=self.use_full_text_retrieval,
           )
           response = get_client().query_content(request)
           return response.body.matches.match_list
   
       def _get_relevant_documents(
               self, query: str, *, run_manager: CallbackManagerForRetrieverRun
       ) -> List[Document]:
           match_list = self.query_content(query)
           return [Document(page_content=i.content) for i in match_list]

3. 建立Chain。

   from langchain_openai import ChatOpenAI
   from langchain_core.prompts import ChatPromptTemplate
   from langchain.schema import StrOutputParser
   from langchain_core.runnables import RunnablePassthrough
   
   OPENAI_API_KEY = "YOUR_OPENAI_API_KEY"
   os.environ["OPENAI_API_KEY"] = OPENAI_API_KEY
   
   template = """Answer the question based only on the following context:
   
   {context}
   
   Question: {question}
   """
   prompt = ChatPromptTemplate.from_template(template)
   model = ChatOpenAI()
   
   
   def format_docs(docs):
       return "\n\n".join([d.page_content for d in docs])
   
   retriever = AdbpgRetriever(namespace='ns1', namespace_password='Ns1password', collection='dc1', top_k=10, use_full_text_retrieval=True)
   chain = (
       {"context": retriever | format_docs, "question": RunnablePassthrough()}
       | prompt
       | model
       | StrOutputParser()
   )

4. 問答。

   chain.invoke("AnalyticDB PostgreSQL是什麼?")
   
   # 回答：
   # AnalyticDB PostgreSQL是阿里雲提供的一種雲原生線上分析處理（OLAP）服務，它基於開源的PostgreSQL資料庫擴充，提供了高效能、高容量的資料倉儲解決方案。
   # 它結合了PostgreSQL的靈活性和相容性以及用於資料分析和報告的高並發和高速查詢能力。
   #
   # AnalyticDB PostgreSQL特別適合處理大規模資料集，支援即時分析和決策支援，是企業進行資料採礦、商業智慧（BI）、報告和資料視覺效果的有力工具。
   # 作為一種託管服務，它簡化了資料倉儲的管理和營運，讓使用者能夠專註於資料分析而不是底層基礎設施。
   # 主要特點包括：
   # 
   # 高效能分析 - 使用列式儲存和大規模平行處理（MPP）架構來快速查詢和分析大量資料。
   # 易於擴充 - 根據資料量和查詢效能要求，容易橫向和縱向擴充資源。
   # 相容 PostgreSQL - 支援PostgreSQL SQL語言和生態系統中的大部分工具，便於現有 PostgreSQL 使用者移轉和適應。
   # 安全和可靠 - 提供資料備份、恢複和加密等功能，確保資料的安全性和可靠性。
   # 雲原生整合 - 與阿里雲的其他服務如Data Integration、資料視覺化工具等緊密整合。
   # 總之，AnalyticDB PostgreSQL是一個高效能、可擴充的雲資料倉儲服務，允許企業在雲環境中進行複雜的資料分析和報告。

全文檢索索引

為了提高檢索的精度，除了向量相似性外，AnalyticDB PostgreSQL版還支援全文檢索索引，並且能和向量相似性檢索同時使用達到雙路召回效果。

1. 定義全文檢索索引欄位。

   在使用全文檢索索引前，首先需指定哪些欄位用於全文檢索索引的資料來源，文件庫的介面已經預設使用content欄位，您還可以指定其他的Metadata自訂欄位。

2. 分詞

   建立文件庫時可以指定Parser欄位作為分詞器，一般情境下，使用預設的中文zh_ch即可，如果有特殊的分詞字元要求，請聯絡阿里雲支援人員。

   在插入資料時，分詞器會將全文檢索索引指定欄位的資料按照分詞符切分，儲存到to_tsvector中，供後續全文檢索索引使用。

Embedding模型

AnalyticDB for PostgreSQL支援如下Embedding模型：

向量索引

向量索引支援設定如下參數：

文檔理解

您可以根據文件類型選用合適的loader：

• UnstructuredHTMLLoader：.html

• UnstructuredMarkdownLoader：.md

• PyMuPDFLoader：.pdf

• PyPDFLoader：.pdf

• RapidOCRPDFLoader：.pdf

• JSONLoader：.json

• CSVLoader：.csv

• RapidOCRLoader：.png、.jpg、.jpeg或.bmp

• UnstructuredFileLoader：.eml、.msg、.rst、.txt、.xml、.docx、.epub、.odt、.pptx或.tsv

如果未指定document_loader_name，會根據文檔名尾碼自動決定使用的loader。當一個文件類型有多個loader時，如pdf，可以指定其中任一個。

文檔切分

文檔切分的效果由chunk_overlap、chunk_size、text_splitter_name幾部分決定，其中text_splitter_name取值說明如下：

• ChineseRecursiveTextSplitter：繼承於RecursiveCharacterTextSplitter，預設以["\n\n","\n", "。|！|？","\.\s|\!\s|\?\s", "；|;\s", "，|,\s"]作為分隔字元，使用正則匹配，中文比 RecursiveCharacterTextSplitter效果更好一些。

• SpacyTextSplitter：預設以["\n\n", "\n", " ", ""]作為分隔字元。可以支援c++、go、java、js、php、proto、python、rst、ruby、rust、scala、swift、markdown、latex、html、sol、csharp等多種代碼語言的切分。

• RecursiveCharacterTextSplitter：預設分隔符號為\n\n，使用Spacy庫的en_core_web_sm模型來分隔，對全英文文檔支援較好。

• MarkdownHeaderTextSplitter：針對markdown類型，使用[ ("#", "head1"), ("##", "head2"), ("###", "head3"), ("####", "head4") ]來切分。