AnalyticDB:Getting started: Build a RAG application by using OpenAPI

Architecture

The RAG architecture uses an information retrieval system to extend the capabilities of a large language model (LLM) with relevant context, such as industry-specific or private documents. The following figure shows the RAG architecture.

This getting-started guide is based on FastANN, the in-house vector engine of AnalyticDB for PostgreSQL. The OpenAPI encapsulates the following AI Service capabilities:

• Multi-tenant management

• Document processing: load, split, embedding, and multi-modal processing

• Retrieval: vector search, full-text search, and reranking

Prerequisites

• An Alibaba Cloud account. If you do not have one, register for an account on the Alibaba Cloud official website.

• If you use AnalyticDB for PostgreSQL for the first time, you must authorize a service-linked role in the console:

  1. Log on to the AnalyticDB for PostgreSQL console.

  2. In the upper-right corner of the page, click Create Instance.

  3. In the Create Service Linked Role dialog box, click OK.

• Your Alibaba Cloud account or RAM user must have the permissions to manage AnalyticDB for PostgreSQL (AliyunGPDBFullAccess).

• An AnalyticDB for PostgreSQL instance. For instructions, see Create an instance.

Step 1: Create an initial account

AnalyticDB for PostgreSQL provides two types of users:

• Privileged user — The initial account has the RDS_SUPERUSER role with full database permissions.

• Standard user — No default permissions. A privileged user or a user with the GRANT permission must grant access to database objects. For more information, see Create and manage users.

  To create the initial account, perform the following steps:

1. Log on to the AnalyticDB for PostgreSQL console and go to the details page of your instance.

2. In the left-side navigation pane, click Account Management.

3. Click Create Account.

4. In the Create Account window, enter an account name and set a password, and then click OK.

   The following table describes the parameters.

Step 2: Prepare the development environment

Check the Python environment

This tutorial uses the Python 3 SDK. Run the following commands to check that Python 3.9 or later and pip are installed:

python3 -V
pip --version

If Python is not installed or the version is earlier than 3.9, install Python.

Install the SDK

Install the alibabacloud_gpdb20160503 and alibabacloud_tea_openapi SDKs. The sample code uses them for authentication and client creation.

pip install --upgrade alibabacloud_gpdb20160503 alibabacloud_tea_openapi

Configure environment variables

Store sensitive information such as credentials and the instance ID in environment variables to prevent leaks caused by hardcoding.

REM Replace access_key_id with the AccessKey ID of the RAM user
set ALIBABA_CLOUD_ACCESS_KEY_ID=access_key_id

REM Replace access_key_secret with the AccessKey secret of the RAM user
set ALIBABA_CLOUD_ACCESS_KEY_SECRET=access_key_secret

REM Replace instance_id with the ID of the AnalyticDB for PostgreSQL instance, for example, gp-bp166cyrtr4p*****
set ADBPG_INSTANCE_ID=instance_id

REM Replace instance_region with the ID of the region where the AnalyticDB for PostgreSQL instance resides, for example, cn-hangzhou
set ADBPG_INSTANCE_REGION=instance_region

Step 3: Prepare the database environment

Preparation process

1. Build a client that you use for operations such as creating a vector database.

2. Initialize the vector database. All vector data is stored in the fixed database named knowledgebase, so you must run the initialization once for each instance. Initializing the vector database does the following:

   • Creates the knowledgebase database and grants read and write permissions on it.

   • Creates the Chinese tokenizer and full-text search features. These features work at the database level.

3. Create a namespace, which you use to create document collections.

4. Create a document collection (DocumentCollection) to store chunk text and vector data.

Sample code

Before you run the code, replace account and account_password with your actual database account and password. Modify the other settings as needed.

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

# --- Get the credentials and instance information from the environment variables ---
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']

# Build and return an API client for AnalyticDB for PostgreSQL
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.alibabacloud.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)


# Initialize the vector database
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}")

# Create a 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}")

# Create a 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__':
    # The initial database account of the AnalyticDB for PostgreSQL instance.
    account = "testacc"
    # The password of the initial account.
    account_password = "Test1234"
    # The name of the namespace to create.
    namespace = "ns1"
    # The password of the namespace. Use this password for subsequent data read and write operations.
    namespace_password = "Ns1password"
    # The name of the document collection to create.
    collection = "dc1"
    metadata = '{"title":"text", "page":"int"}'
    full_text_retrieval_fields = "title"
    embedding_model = "m3e-base"

    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)

Parameters

View the table schema

After the preceding code runs, log on to the database to view the table schema:

1. Log on to the AnalyticDB for PostgreSQL console.

2. In the upper-right corner of the instance details page, click Log On to Database.

3. On the Logon Instance page, enter the Account and Database Password, and then click Log On.

   After you log on, you can see a new database named knowledgebase in the instance, a schema named ns1 in the knowledgebase database, and a table named dc1 in that schema. The following table describes the schema of the dc1 table.

Step 4: Manage documents

The sample code in this section uses the environment variables and the get_client() function defined in Step 3: Prepare the database environment. Run it in the same script as the code from that section.

Upload a document

The following sample code uploads a local document asynchronously:

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)

Parameters

(Optional) Other document management operations

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")

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")

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")

Step 5: Retrieve documents

The sample code in this section uses the environment variables and the get_client() function defined in Step 3: Prepare the database environment. Run it in the same script as the code from that section.

This section uses plain text retrieval as an example. The following sample code shows how to retrieve documents:

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, 'What is ADBPG?')

Parameters

Response parameters

Step 6: Integrate with LangChain

LangChain is an open source framework for building applications on large language models (LLMs). It connects models to external data through a complete set of interfaces and tools. The following example integrates the retrieval capabilities of AnalyticDB for PostgreSQL into LangChain to implement a question-answering system.

The sample code in this section uses the environment variables and the get_client() function defined in Step 3: Prepare the database environment. Run it in the same script as the code from that section.

This example uses an OpenAI model as the large language model. Replace YOUR_OPENAI_API_KEY in the following code with your OpenAI API key.

Install the modules

Run the following command to install the modules:

pip install --upgrade langchain langchain-openai openai tiktoken

Build the AdbpgRetriever

The following class wraps the query_content operation as a LangChain retriever:

from typing import List

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]

Create the chain

The following code creates a retrieval-augmented question-answering chain:

import os

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()
)

Ask a question

Call the chain to ask a question:

chain.invoke("What is AnalyticDB PostgreSQL?")

# Answer:
# AnalyticDB PostgreSQL is a cloud-native online analytical processing (OLAP) service from Alibaba Cloud. It extends the open source PostgreSQL database and provides a high-performance, high-capacity data warehouse solution.
# It combines the flexibility and compatibility of PostgreSQL with the high-concurrency, high-speed query capabilities required for data analytics and reporting.
#
# AnalyticDB PostgreSQL suits large-scale datasets, supports real-time analytics and decision support, and helps enterprises with data mining, business intelligence (BI), reporting, and data visualization.
# As a managed service, it simplifies data warehouse management and O&M, so you can focus on data analytics instead of the underlying infrastructure.
# Main features:
#
# High-performance analytics - Uses columnar storage and a massively parallel processing (MPP) architecture to query and analyze large volumes of data quickly.
# Easy scaling - Scales resources horizontally and vertically based on data volume and query performance requirements.
# PostgreSQL compatibility - Supports the PostgreSQL SQL language and most tools in the PostgreSQL ecosystem, so existing PostgreSQL users can migrate and adapt easily.
# Security and reliability - Provides data backup, restoration, and encryption to keep data secure and reliable.
# Cloud-native integration - Integrates closely with other Alibaba Cloud services, such as data integration and data visualization tools.
# In summary, AnalyticDB PostgreSQL is a high-performance, scalable cloud data warehouse service that lets enterprises run complex data analytics and reporting in the cloud.

Appendix