A Step-by-Step Tutorial — Building a RAG Application with OceanBase seekdb

Posted on In Views: Word count in article: 2.3k Reading time ≈ 8 mins.
This is a step-by-step tutorial that walks you hands-on through building a RAG (Retrieval-Augmented Generation) system with OceanBase seekdb, covering the full flow of environment setup, configuring three Embedding function types, data import, vector retrieval, and launching a Streamlit app.

📚 Every step in this tutorial is something you can reproduce hands-on with seekdb! Come follow along at https://github.com/oceanbase/seekdb—two hours from now you might just have your own RAG application~

This is another step-by-step tutorial, showing you in detail how to build a RAG (Retrieval-Augmented Generation) system with OceanBase seekdb.

A Step-by-Step Tutorial — Building a RAG Application with OceanBase seekdb

A RAG system combines a retrieval system with a generative model, generating new text based on a given prompt. The system first uses seekdb’s native vector search to retrieve relevant documents from a corpus, then uses a generative model to produce new text based on the retrieved documents.

Prerequisites

  • Python 3.11 or higher installed
  • uv installed
  • An LLM API Key ready

Preparation

Clone the Code

1
2
git clone https://github.com/oceanbase/pyseekdb.git
cd pyseekdb/demo/rag

Set Up the Environment

Install Dependencies

Basic installation (for the default or api embedding types):

1
uv sync

Local model (for the local embedding type):

1
uv sync --extra local

Tips:

  • The local extra includes sentence-transformers and its related dependencies (about 2–3 GB).
  • If you are in mainland China, you can use a domestic mirror source to speed up downloads:
    • Basic installation (Tsinghua mirror): uv sync --index-url https://pypi.tuna.tsinghua.edu.cn/simple
    • Basic installation (Aliyun mirror): uv sync --index-url https://mirrors.aliyun.com/pypi/simple
    • Local model (Tsinghua mirror): uv sync --extra local --index-url https://pypi.tuna.tsinghua.edu.cn/simple
    • Local model (Aliyun mirror): uv sync --extra local --index-url https://mirrors.aliyun.com/pypi/simple

Set Environment Variables

Step 1: Copy the environment variable template

1
cp .env.example .env

Step 2: Edit the .env file and set the environment variables

This system supports three types of Embedding functions, which you can choose based on your needs:

  1. default (the default, recommended for beginners)
    • Uses pyseekdb’s built-in DefaultEmbeddingFunction (based on ONNX).
    • Automatically downloads the model on first use; no API Key configuration needed.
    • Suitable for local development and testing.
  2. local (local model)
    • Uses a custom sentence-transformers model.
    • Requires installing the sentence-transformers library.
    • The model name and device (CPU/GPU) can be configured.
  3. api (API service)
    • Uses an OpenAI-compatible Embedding API (such as DashScope, OpenAI, etc.).
    • Requires configuring the API Key and model name.
    • Suitable for production environments.

The following uses Tongyi Qianwen as an example (using the api type):

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# Embedding Function type: api, local, default
EMBEDDING_FUNCTION_TYPE=api

# LLM configuration (used to generate answers)
OPENAI_API_KEY=sk-your-dashscope-key
OPENAI_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1
OPENAI_MODEL_NAME=qwen-plus

# Embedding API configuration (only needed when EMBEDDING_FUNCTION_TYPE=api)
EMBEDDING_API_KEY=sk-your-dashscope-key
EMBEDDING_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1
EMBEDDING_MODEL_NAME=text-embedding-v4

# Local model configuration (only needed when EMBEDDING_FUNCTION_TYPE=local)
SENTENCE_TRANSFORMERS_MODEL_NAME=all-mpnet-base-v2
SENTENCE_TRANSFORMERS_DEVICE=cpu

# seekdb configuration
SEEKDB_DIR=./data/seekdb_rag
SEEKDB_NAME=test
COLLECTION_NAME=embeddings

Environment variable reference:

Variable Description Default/Example Required When
EMBEDDING_FUNCTION_TYPE Embedding function type default (options: api, local, default) Always required
OPENAI_API_KEY LLM API Key (supports OpenAI, Tongyi Qianwen, and other compatible services) - Always required (used to generate answers)
OPENAI_BASE_URL LLM API base URL https://dashscope.aliyuncs.com/compatible-mode/v1 Optional
OPENAI_MODEL_NAME Language model name qwen-plus Optional
EMBEDDING_API_KEY Embedding API Key - Required when EMBEDDING_FUNCTION_TYPE=api
EMBEDDING_BASE_URL Embedding API base URL https://dashscope.aliyuncs.com/compatible-mode/v1 Optional when EMBEDDING_FUNCTION_TYPE=api
EMBEDDING_MODEL_NAME Embedding model name text-embedding-v4 Required when EMBEDDING_FUNCTION_TYPE=api
SENTENCE_TRANSFORMERS_MODEL_NAME Local model name all-mpnet-base-v2 Optional when EMBEDDING_FUNCTION_TYPE=local
SENTENCE_TRANSFORMERS_DEVICE Device to run on cpu Optional when EMBEDDING_FUNCTION_TYPE=local
SEEKDB_DIR seekdb database directory ./data/seekdb_rag Optional
SEEKDB_NAME Database name test Optional
COLLECTION_NAME Embedding table name embeddings Optional

Tips:

  • If you use the default type, you only need to configure EMBEDDING_FUNCTION_TYPE=default and the LLM-related variables.
  • If you use the api type, you need to additionally configure the Embedding API-related variables.
  • If you use the local type, you need to install the sentence-transformers library, and you may optionally configure the model name.

Main Modules Used

Initializing the LLM Client

We initialize the LLM client by loading the environment variables:

1
2
3
4
5
6
def get_llm_client() -> OpenAI:
    """Initialize LLM client using OpenAI-compatible API."""
    return OpenAI(
        api_key=os.getenv("OPENAI_API_KEY"),
        base_url=os.getenv("OPENAI_BASE_URL"),
    )

Creating a Database Connection

1
2
3
4
5
6
7
8
def get_seekdb_client(db_dir: str = "./seekdb_rag", db_name: str = "test"):
    """Initialize seekdb client (embedded mode)."""
    cache_key = (db_dir, db_name)
    if cache_key not in _client_cache:
        print(f"Connecting to seekdb: path={db_dir}, database={db_name}")
        _client_cache[cache_key] = Client(path=db_dir, database=db_name)
        print("seekdb client connected successfully")
    return _client_cache[cache_key]

A Factory Pattern for Custom Embedding Models

In the .env file, you can configure EMBEDDING_FUNCTION_TYPE to use different embedding_functions. You can also refer to this example to customize your own embedding_function.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
from pyseekdb import EmbeddingFunction, DefaultEmbeddingFunction
from typing import List, Union
import os
from openai import OpenAI

Documents = Union[str, List[str]]
Embeddings = List[List[float]]

class SentenceTransformerCustomEmbeddingFunction(EmbeddingFunction[Documents]):
    """
    A custom embedding function using sentence-transformers with a specific model.
    """

    def __init__(self, model_name: str = "all-mpnet-base-v2", device: str = "cpu"):  # TODO: your own model name and device
        """
        Initialize the sentence-transformer embedding function.

        Args:
            model_name: Name of the sentence-transformers model to use
            device: Device to run the model on ('cpu' or 'cuda')
        """
        self.model_name = model_name or os.environ.get('SENTENCE_TRANSFORMERS_MODEL_NAME')
        self.device = device or os.environ.get('SENTENCE_TRANSFORMERS_DEVICE')
        self._model = None
        self._dimension = None

    def _ensure_model_loaded(self):
        """Lazy load the embedding model"""
        if self._model is None:
            try:
                from sentence_transformers import SentenceTransformer
                self._model = SentenceTransformer(self.model_name, device=self.device)
                # Get dimension from model
                test_embedding = self._model.encode(["test"], convert_to_numpy=True)
                self._dimension = len(test_embedding[0])
            except ImportError:
                raise ImportError(
                    "sentence-transformers is not installed. "
                    "Please install it with: pip install sentence-transformers"
                )

    @property
    def dimension(self) -> int:
        """Get the dimension of embeddings produced by this function"""
        self._ensure_model_loaded()
        return self._dimension

    def __call__(self, input: Documents) -> Embeddings:
        """
        Generate embeddings for the given documents.

        Args:
            input: Single document (str) or list of documents (List[str])

        Returns:
            List of embedding vectors
        """
        self._ensure_model_loaded()

        # Handle single string input
        if isinstance(input, str):
            input = [input]

        # Handle empty input
        if not input:
            return []

        # Generate embeddings
        embeddings = self._model.encode(
            input,
            convert_to_numpy=True,
            show_progress_bar=False
        )

        # Convert numpy arrays to lists
        return [embedding.tolist() for embedding in embeddings]


class OpenAIEmbeddingFunction(EmbeddingFunction[Documents]):
    """
    A custom embedding function using Embedding API.
    """

    def __init__(self, model_name: str = "", api_key: str = "", base_url: str = ""):
        """
        Initialize the Embedding API embedding function.

        Args:
            model_name: Name of the Embedding API embedding model
            api_key: Embedding API key (if not provided, uses EMBEDDING_API_KEY env var)
        """
        self.model_name = model_name or os.environ.get('EMBEDDING_MODEL_NAME')
        self.api_key = api_key or os.environ.get('EMBEDDING_API_KEY')
        self.base_url = base_url or os.environ.get('EMBEDDING_BASE_URL')
        self._dimension = None
        if not self.api_key:
            raise ValueError("Embedding API key is required")

    def _ensure_model_loaded(self):
        """Lazy load the Embedding API model"""
        try:
            client = OpenAI(
                api_key=self.api_key,
                base_url=self.base_url
            )
            response = client.embeddings.create(
                model=self.model_name,
                input=["test"]
            )
            self._dimension = len(response.data[0].embedding)
        except Exception as e:
            raise ValueError(f"Failed to load Embedding API model: {e}")

    @property
    def dimension(self) -> int:
        """Get the dimension of embeddings produced by this function"""
        self._ensure_model_loaded()
        return self._dimension

    def __call__(self, input: Documents) -> Embeddings:
        """
        Generate embeddings using Embedding API.

        Args:
            input: Single document (str) or list of documents (List[str])

        Returns:
            List of embedding vectors
        """
        # Handle single string input
        if isinstance(input, str):
            input = [input]

        # Handle empty input
        if not input:
            return []

        # Call Embedding API
        client = OpenAI(
            api_key=self.api_key,
            base_url=self.base_url
        )
        response = client.embeddings.create(
            model=self.model_name,
            input=input
        )

        # Extract Embedding API embeddings
        embeddings = [item.embedding for item in response.data]
        return embeddings


def create_embedding_function() -> EmbeddingFunction:
    embedding_function_type = os.environ.get('EMBEDDING_FUNCTION_TYPE')
    if embedding_function_type == "api":
        print("Using OpenAI Embedding API embedding function")
        return OpenAIEmbeddingFunction()
    elif embedding_function_type == "local":
        print("Using SentenceTransformer embedding function")
        return SentenceTransformerCustomEmbeddingFunction()
    elif embedding_function_type == "default":
        print("Using Default embedding function")
        return DefaultEmbeddingFunction()
    else:
        raise ValueError(f"Unsupported embedding function type: {embedding_function_type}")

Creating a Collection

In the get_or_create_collection() method we pass in an embedding_function. Afterward, when using this collection’s add() and query() methods, you no longer need to pass in vectors—just the text, and the vectors will be generated automatically by the embedding_function.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
def get_seekdb_collection(client, collection_name: str = "embeddings",
                  embedding_function: Optional[EmbeddingFunction] = DefaultEmbeddingFunction(),
                  drop_if_exists: bool = True):
    """
    Get or create a collection using pyseekdb's get_or_create_collection.

    Args:
        client: seekdb client instance
        collection_name: Name of the collection
        embedding_function: Embedding function (required for automatic embedding generation)
        drop_if_exists: Whether to drop existing collection if it exists

    Returns:
        Collection object
    """
    if drop_if_exists and client.has_collection(collection_name):
        print(f"Collection '{collection_name}' already exists, deleting old data...")
        client.delete_collection(collection_name)

    if embedding_function is None:
        raise ValueError("embedding_function is required")

    # Use pyseekdb's native get_or_create_collection
    collection = client.get_or_create_collection(
        name=collection_name,
        embedding_function=embedding_function
    )

    print(f"Collection '{collection_name}' ready!")
    return collection

The Core Data-Insertion Function

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
def insert_embeddings(collection, data: List[Dict[str, Any]]):
    """
    Insert data into collection. Embeddings are automatically generated by collection's embedding_function.

    Args:
        collection: Collection object (must have embedding_function configured)
        data: List of data dictionaries containing 'text', 'source_file', 'chunk_index'
    """
    try:
        ids = [f"{item['source_file']}_{item.get('chunk_index', 0)}" for item in data]
        documents = [item['text'] for item in data]
        metadatas = [{'source_file': item['source_file'],
                     'chunk_index': item.get('chunk_index', 0)} for item in data]

        # Collection's embedding_function will automatically generate embeddings from documents
        collection.add(
            ids=ids,
            documents=documents,
            metadatas=metadatas
        )

        print(f"Inserted {len(data)} items successfully")
    except Exception as e:
        print(f"Error inserting data: {e}")
        raise
1
2
3
4
5
results = collection.query(
                    query_texts=[question],
                    n_results=3,
                    include=["documents", "metadatas", "distances"]
                )

Gathering Statistics About the Data in a Collection

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
def get_database_stats(collection) -> Dict[str, Any]:
    """Get statistics about the collection."""
    try:
        results = collection.get(limit=10000, include=["metadatas"])
        ids = results.get('ids', []) if isinstance(results, dict) else []
        metadatas = results.get('metadatas', []) if isinstance(results, dict) else []

        unique_files = {m.get('source_file') for m in metadatas if m and m.get('source_file')}

        return {
            "total_embeddings": len(ids),
            "unique_source_files": len(unique_files)
        }
    except Exception as e:
        print(f"Error getting database stats: {e}")
        return {"total_embeddings": 0, "unique_source_files": 0}

Building the RAG System

This module implements the retrieval functionality of the RAG system. By converting a user’s question into an embedding vector and leveraging the native vector search capability provided by seekdb, it quickly retrieves the document chunks most relevant to the question, supplying the necessary context for the downstream generative model.

Importing Data

We use pyseekdb’s SDK documentation as an example, but you can also use your own Markdown documents or directory.

Run the data import script:

1
2
3
4
5
# import a single document
uv run python seekdb_insert.py ../../README.md

# or import all Markdown documents under a directory
uv run python seekdb_insert.py path/to/your_dir

Launching the Application

In the pyseekdb/demo/rag path, run the following command to launch the application via Streamlit:

1
uv run streamlit run seekdb_app.py --server.port your_port

Use the IP and port (default 8501, customizable via the --server.port option) to open the RAG interface in your browser.

Tip: If you use uv as your package manager, prefix commands with uv run to ensure the correct Python environment and dependencies are used.

A Step-by-Step Tutorial — Building a RAG Application with OceanBase seekdb

A Real-World Example Run

Below is a real query example, demonstrating the actual results of the seekdb RAG system:

User question: “What is seekdb’s vector search?”

Retrieved relevant documents:

1
2
3
4
5
6
7
8
1. 30.experience-vector-retrieval.md
   Vector search overview In today's era of information explosion, users often need to quickly retrieve the information they need from massive amounts o...

2. 20.using-seekdb-in-python-mode.md
   Experience vector search SeekDB supports up to 16,000 dimensions of float-type dense vectors, sparse vectors, and various types of vector distance ca...

3. 70.experience-hybrid-vector-index.md
   Summary Through this tutorial, you have mastered the core features of SeekDB's hybrid vector index: Simplified usage process: Achieve semantic ret...

The answer generated by the LLM:

SeekDB’s vector search is a powerful feature that enables efficient and accurate retrieval of data by encoding various types of data—such as text, images, and audio—into mathematical vectors and performing searches in a vector space. This approach captures deep semantic relationships within the data, going beyond traditional keyword-based search methods to deliver more relevant results.

Key capabilities of SeekDB’s vector search include:

  • High-dimensional vector support: It supports up to 16,000 dimensions for float-type dense vectors and sparse vectors.
  • Multiple distance metrics: It supports various similarity or distance calculations, including Manhattan distance, Euclidean distance, inner product, and cosine distance.
  • Advanced indexing methods: Vector indexes can be created using HNSW (Hierarchical Navigable Small World) or IVF (Iterative Virtual File), enabling fast approximate nearest neighbor searches.
  • Incremental updates: The system allows for real-time insertion and deletion of vectors without compromising search accuracy or recall.
  • Hybrid search with scalar filtering: Users can combine vector similarity search with traditional attribute-based (scalar) filtering for more precise results.
  • Flexible access interfaces: SeekDB supports SQL access via MySQL protocol clients in multiple programming languages, as well as a Python SDK.
  • Automatic embedding and hybrid indexing: With hybrid vector index features, users can store raw text directly—the system automatically converts it into vectors and builds indexes.

In summary, SeekDB’s vector search provides a comprehensive, high-performance solution for semantic search, particularly valuable in AI applications involving large-scale unstructured data.

This example demonstrates:

  • ✅ Accurate information retrieval: the system successfully found relevant information in the documents.
  • ✅ Multi-document integration: extracting and integrating information from 3 different documents.
  • ✅ Semantic matching: accurately matching documents related to “vector search.”
  • ✅ Structured answers: the AI organized the retrieved information into a clear structure.
  • ✅ Completeness: covering the main features of seekdb’s vector search.
  • ✅ Professionalism: the answer includes technical details and practical application value.

Retrieval quality analysis:

  • Most relevant document: experience-vector-retrieval.md - vector search overview.
  • Technical details: using-seekdb-in-python-mode.md - specific technical specifications.
  • Advanced features: experience-hybrid-vector-index.md - hybrid vector index functionality.

A Quick Try

To quickly try out the seekdb RAG system, refer to Quick Deployment.

References