AnalyticDB:quick start for building rag applications using openapi

How it works

AnalyticDB for PostgreSQL uses its self-developed FastANN vector engine to power RAG workflows. Its OpenAPI encapsulates the full AI service stack:

• Document processing: load, split, embed, and handle multi-modal content

• Search: vector search, full-text index, and reranking

• Multi-tenant management: isolate data by namespace

The RAG architecture connects an information retrieval system to a large language model (LLM), grounding model responses in your own documents.

Workflow overview

This guide covers the following steps in order:

Prerequisites

Before you begin, ensure that you have:

• An Alibaba Cloud account (register at the Alibaba Cloud official website if you don't have one)

• The AliyunGPDBFullAccess permission granted to your Alibaba Cloud account or RAM user

• An AccessKey pair created

The first time you use AnalyticDB for PostgreSQL, you must authorize the creation of a service-linked role. Log on to the AnalyticDB for PostgreSQL console, click Create Instance in the upper-right corner, and click OK in the Create Service Linked Role dialog.

Billing

Creating an instance incurs charges for compute and storage resources. For pricing details, see Pricing.

Free trial: Alibaba Cloud offers a free trial for Storage-elastic Mode instances to new users. Visit Alibaba Cloud Free Trial to apply. If you are not eligible, follow the steps below to create a pay-as-you-go instance.

Step 1: Create an instance

1. Log on to the AnalyticDB for PostgreSQL console.

2. In the upper-right corner, click Create Instance to open the purchase page.

3. Configure the following parameters. Keep the default values for other parameters. For full parameter descriptions, see Create an instance.

1. Click Buy Now, confirm the order, and click Activate Now.

2. After payment, click Management Console to view the instance list.

Instance initialization takes a few minutes. Wait for the instance status to change to Running before proceeding.

Step 2: Create an initial account

1. In the left navigation pane, click Account Management.

2. Click Create Account. In the Create Account window, configure the following parameters and click OK.

Step 3: Set up the development environment

python -V
pip --version

pip install --upgrade alibabacloud_gpdb20160503 alibabacloud_tea_openapi

set ALIBABA_CLOUD_ACCESS_KEY_ID=<your-access-key-id>
set ALIBABA_CLOUD_ACCESS_KEY_SECRET=<your-access-key-secret>
set ADBPG_INSTANCE_ID=<your-instance-id>
set ADBPG_INSTANCE_REGION=<your-region-id>

Step 4: Prepare the database environment

This step initializes the vector database, creates a namespace, and creates a document collection to store chunked text and vector data.

What this step does:

1. Build a client — authenticates with the ADBPG API

2. Initialize the vector database — creates the knowledgebase database, grants read/write permissions, and sets up Chinese tokenizer and full-text index features (run once per instance)

3. Create a namespace — a logical tenant boundary for document libraries

4. Create a document collection — stores chunked text and vector data

Before running, replace account and account_password with your actual database account and password.

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

# Read credentials and instance info from 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']


def get_client():
    """Build and return an AnalyticDB for PostgreSQL API 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
    # Shared endpoint for major Chinese mainland regions and Singapore
    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):
    """Initialize the vector database. Run once per instance."""
    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}")


def create_namespace(account, account_password, namespace, namespace_password):
    """Create a namespace (tenant boundary) for document libraries."""
    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}")


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):
    """Create a document collection to store chunked text and vector data."""
    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__':
    account = "testacc"                      # Initial database account
    account_password = "Test1234"            # Password for the initial account
    namespace = "ns1"                        # Namespace name
    namespace_password = "ns1password"       # Namespace password (used for data read/write operations)
    collection = "dc1"                       # Document collection name
    metadata = '{"title":"text", "page":"int"}'
    full_text_retrieval_fields = "title"
    embedding_model = "m3e-small"            # See Embedding models for supported options

    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

Step 5: Manage documents

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)

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 6: Retrieve content

The following example queries the document collection using plain text. ADBPG automatically converts the query to a vector and searches for similar chunks.

To compare retrieval modes, the example below shows both vector-only and hybrid retrieval side by side:

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__':
    # Vector search only (default)
    query_content('ns1', 'Ns1password', 'dc1', 10, 'What is ADBPG?')

    # Hybrid retrieval: vector search + full-text index
    query_content('ns1', 'Ns1password', 'dc1', 10, 'What is ADBPG?',
                  use_full_text_retrieval=True)

Parameters

Response fields

Step 7: Integrate LangChain

LangChain is an open-source framework for building LLM-powered applications. This section shows how to wrap ADBPG's retrieval API as a LangChain retriever and build a question-answering chain.

pip install --upgrade langchain openai tiktoken

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]

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

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

# Response:
# AnalyticDB PostgreSQL is a cloud-native Online Analytical Processing (OLAP) service provided by Alibaba Cloud. Based on the open-source PostgreSQL database extension, it delivers a high-performance, high-capacity data warehouse solution.
# It combines PostgreSQL's flexibility and compatibility with high concurrency and high-speed query capabilities for data analytics and reporting.
#
# AnalyticDB PostgreSQL is particularly well-suited for processing large-scale datasets, supporting real-time analytics and decision support. It serves as a powerful tool for enterprises to perform data mining, business intelligence (BI), reporting, and data visualization.
# As a managed service, it simplifies data warehouse management and operations and maintenance (O&M), enabling users to focus on data analytics rather than underlying infrastructure.
# Key features include the following:
#
# High-performance analytics — Uses columnar storage and Massively Parallel Processing (MPP) architecture to quickly query and analyze large volumes of data.
# Easy scalability — Resources can be easily scaled horizontally and vertically based on data volume and query performance requirements.
# PostgreSQL compatibility — Supports PostgreSQL SQL language and most tools in the PostgreSQL ecosystem, making it easy for existing PostgreSQL users to migrate and adapt.
# Security and reliability — Provides features such as data backup, recovery, and encryption to ensure data security and reliability.
# Cloud-native integration — Tightly integrated 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 enables enterprises to perform complex data analytics and reporting in cloud environments.

Appendix

What's next

• Create a document collection (API reference)

• Create and manage users

• Pricing