Saving My Company $5,000 in Just Two Weekends: Refactoring a Knowledge Base from a Multi-Database Architecture to an AI-Native Database

Source: Bailu Diyishuai. Reproduction without authorization is strictly prohibited; infringement will be pursued!

🛠️ Want to experience that “one database does it all” satisfaction described in this article? seekdb is now open source on GitHub—come try it at https://github.com/oceanbase/seekdb. Your next project might just save a bundle too~

Introduction

Traditional AI applications often combine multiple databases: PostgreSQL for structured data, Elasticsearch for full-text search, Milvus for vector retrieval, and Redis for caching. This “patchwork” architecture brings problems such as complex data synchronization, high costs, and difficult maintenance.

On November 18, 2025, OceanBase open-sourced its AI-native database, seekdb. Over two weekends, I used seekdb to refactor an enterprise knowledge base system, simplifying a complex four-database architecture into a single-database solution. Query latency dropped from 120ms to 58ms, a performance improvement of 50%+, and we saved $450 a month in cloud service costs.

This article is based on my complete hands-on experience building an enterprise knowledge base with seekdb—from setting up the environment to integrating a RAG application—documenting every technical detail and pitfall, with plenty of runnable code examples to help developers get up to speed with seekdb quickly.

1. First Encounter with seekdb: From Predicament to Turning Point

This October, I received a request: add a smart Q&A feature to the company’s internal documentation system. As an engineer working on big data and large-model application development, I’d seen plenty of requests like this, but only after I started did I realize the problem was far more complex than I’d imagined.

The initial plan looked like this:

• Use PostgreSQL to store document metadata.
• Use Elasticsearch for full-text search.
• Use Pinecone (managed) for vector data.
• Use Redis to cache hot data.

The old architecture had four main pain points.

• Complex data synchronization: four databases had to stay consistent.
• High cost: Pinecone cost $300+ per month.
• Poor performance: cross-system queries had high latency.
• Hard to maintain: managing multiple databases is a burden.

The result? Keeping data in sync across four databases drove me crazy, Pinecone cost $300 a month, and cross-system joins performed terribly. Worse still, whenever I wanted to filter search results by structured fields like document creation time or department permissions, I had to do a second round of filtering at the application layer, and code complexity shot up.

Sample data-synchronization code from the old architecture:

import psycopg2
from elasticsearch import Elasticsearch
import pinecone
import redis
import json

class MultiDBSync:
    """Multi-database sync manager - the pain point of the old architecture"""
    
    def __init__(self):
        # Initialize connections to 4 databases
        self.pg_conn = psycopg2.connect(
            host="localhost", database="docs", 
            user="admin", password="password"
        )
        self.es = Elasticsearch(['http://localhost:9200'])
        pinecone.init(api_key="your-key", environment="us-west1-gcp")
        self.pinecone_index = pinecone.Index("documents")
        self.redis_client = redis.Redis(host='localhost', port=6379)
    
    def insert_document(self, doc_id, title, content, metadata, embedding):
        """Insert a document into 4 databases - consistency must be guaranteed"""
        try:
            # 1. Store metadata in PostgreSQL
            cursor = self.pg_conn.cursor()
            cursor.execute("""
                INSERT INTO documents (id, title, created_at, department_id)
                VALUES (%s, %s, %s, %s)
            """, (doc_id, title, metadata['created_at'], metadata['department_id']))
            self.pg_conn.commit()
            
            # 2. Store full text in Elasticsearch
            self.es.index(index='documents', id=doc_id, body={
                'title': title,
                'content': content,
                'created_at': metadata['created_at']
            })
            
            # 3. Store vector in Pinecone
            self.pinecone_index.upsert([(
                str(doc_id),
                embedding,
                {'title': title, 'department_id': metadata['department_id']}
            )])
            
            # 4. Cache hot data in Redis
            self.redis_client.setex(
                f"doc:{doc_id}",
                3600,
                json.dumps({'title': title, 'content': content[:200]})
            )
            
            return True
        except Exception as e:
            # Rolling back is hard; each database must be cleaned up manually
            print(f"Sync failed: {e}")
            self._rollback(doc_id)
            return False
    
    def _rollback(self, doc_id):
        """Rollback operation - very complex and error-prone"""
        try:
            cursor = self.pg_conn.cursor()
            cursor.execute("DELETE FROM documents WHERE id = %s", (doc_id,))
            self.pg_conn.commit()
        except: pass
        
        try:
            self.es.delete(index='documents', id=doc_id)
        except: pass
        
        try:
            self.pinecone_index.delete(ids=[str(doc_id)])
        except: pass
        
        try:
            self.redis_client.delete(f"doc:{doc_id}")
        except: pass
    
    def search(self, query, filters):
        """Joint query - results must be aggregated at the application layer"""
        # 1. Vector search
        vector_results = self.pinecone_index.query(
            vector=query['embedding'],
            top_k=20,
            filter={'department_id': filters.get('department_id')}
        )
        
        # 2. Full-text search
        es_results = self.es.search(index='documents', body={
            'query': {'match': {'content': query['text']}},
            'size': 20
        })
        
        # 3. Merge results at the application layer - poor performance and complex
        merged_results = self._merge_results(vector_results, es_results)
        
        # 4. Fetch full metadata from PostgreSQL
        final_results = self._enrich_metadata(merged_results)
        
        return final_results
    
    def _merge_results(self, vector_results, es_results):
        """Merge results from different data sources - complex algorithm"""
        # Complex ranking and deduplication logic is needed here
        # Code omitted...
        pass
    
    def _enrich_metadata(self, results):
        """Enrich metadata - extra database queries"""
        # Code omitted...
        pass

# Usage example
sync_manager = MultiDBSync()
# Every insert touches 4 databases - high failure rate
sync_manager.insert_document(
    doc_id=1,
    title="Python Best Practices",
    content="...",
    metadata={'created_at': '2024-01-01', 'department_id': 1},
    embedding=[0.1, 0.2, ...]  # 1536-dimensional vector
)

This codebase was painful to maintain:

• Data consistency was hard to guarantee; one database would often fail to update.
• The rollback logic was complex and prone to dirty data.
• Joint queries required heavy aggregation at the application layer.
• The code was large—this sync module alone exceeded 500 lines.

Then, in late November, I saw the news that seekdb had been open-sourced in the OceanBase community. On a whim, I spent a weekend refactoring the whole system. The result delighted me: not only was the architecture simpler, but performance improved by 40%, and cloud service costs dropped right away.

New architecture comparison:

1.1 Project Background and Technical Pain Points

seekdb is the AI-native database that OceanBase open-sourced on November 18, 2025. When I first saw its introduction, three things attracted me most.

1. MySQL compatibility: I didn’t need to learn a new query language; my existing MySQL clients and ORMs all worked directly.
2. Three-in-one capability: vector search, full-text search, and structured queries all in one database.
3. Lightweight deployment: a single Docker command gets it running, with no complex cluster configuration.

seekdb core features:

Even more important, it’s open source, with the code hosted on GitHub (GitHub repo: https://github.com/oceanbase/seekdb ), which means I can confidently use it in production without worrying about vendor lock-in.

1.2 The First-Contact Experience with seekdb

When developing AI applications, we often need to use several databases at once:

• PostgreSQL for business data.
• Elasticsearch for full-text search.
• Milvus or Pinecone for vector retrieval.

This architecture not only increases system complexity but also brings problems like data synchronization and consistency maintenance. seekdb integrates these three capabilities into a single database, greatly simplifying the architecture.

1.3 seekdb’s Core Advantages

seekdb is fully compatible with MySQL, which means:

• You can use familiar SQL syntax.
• Existing MySQL tools and clients all work directly.
• The learning cost is nearly zero.

2. Migration Hands-On: From Multiple Databases to seekdb

Inheriting OceanBase’s high-performance engine, seekdb shines in vector retrieval and hybrid query scenarios. At the same time, its lightweight design dramatically lowers deployment and operations costs.

2.1 Rapid Deployment and Data Model Design

Environment Preparation and Installation

Environment Requirements

• Operating system: Linux / macOS / Windows
• Docker: 20.10+
• Memory: minimum 1C2G (official lightweight/demo); the author’s hands-on recommendation is 4GB+ to start, 8GB+ for more stability (when running embedding generation, full-text/vector indexing, and queries at the same time)
• Disk: minimum 10GB of free space

Note: 1C2G is the officially advertised minimum spec, suitable for lightweight or demo scenarios; the 4GB/8GB+ figures in this article are stability-test conclusions from local single-node Docker with parallel embedding generation and indexing, offered as a reference for real projects.

My development environment is a MacBook Pro M2 with 16GB of memory. Installing seekdb was surprisingly simple:

# Pull the image
docker pull oceanbase/seekdb:latest

# Start the container
docker run -d --name seekdb \
  -p 2881:2881 \
  -e MODE=slim \
  -v ~/seekdb_data:/root/ob \
  oceanbase/seekdb:latest

Note: I added a data volume mount (the -v option) so that data isn’t lost when the container restarts. I missed this on my first deployment and lost all my test data.

After waiting about 30 seconds, the container finished starting. Connect with a MySQL client:

mysql -h127.0.0.1 -P2881 -uroot
# The default password is empty; just press Enter

Seeing the oceanbase> prompt means the connection succeeded.

I first ran a few commands to confirm the functionality worked:

-- Check the version
SELECT VERSION(); 

-- Confirm vector functionality is available
SHOW VARIABLES LIKE '%vector%';

The output showed version 4.3.0 with vector functionality enabled. Perfect!

This process was far simpler than deploying vector databases at my previous companies, where just configuring Milvus’s dependencies took half a day.

Installing Dependencies (Python)

pip install pymysql tenacity openai

Connection Configuration and Utility Functions

To make later development easier, I wrapped a seekdb connection management class:

import pymysql
from typing import List, Dict, Optional
import logging
from contextlib import contextmanager

class SeekDBManager:
    """SeekDB connection manager"""
    
    def __init__(self, host='127.0.0.1', port=2881, user='root', 
                 password='', database='knowledge_base'):
        self.config = {
            'host': host,
            'port': port,
            'user': user,
            'password': password,
            'database': database,
            'charset': 'utf8mb4',
            'cursorclass': pymysql.cursors.DictCursor
        }
        self.logger = logging.getLogger(__name__)
    
    @contextmanager
    def get_connection(self):
        """Get a database connection (context manager)"""
        conn = pymysql.connect(**self.config)
        try:
            yield conn
            conn.commit()
        except Exception as e:
            conn.rollback()
            self.logger.error(f"Database operation failed: {e}")
            raise
        finally:
            conn.close()
    
    def execute_query(self, sql: str, params: tuple = None) -> List[Dict]:
        """Execute a query and return the results"""
        with self.get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute(sql, params or ())
            results = cursor.fetchall()
            cursor.close()
            return results
    
    def execute_update(self, sql: str, params: tuple = None) -> int:
        """Execute an update and return the number of affected rows"""
        with self.get_connection() as conn:
            cursor = conn.cursor()
            affected_rows = cursor.execute(sql, params or ())
            cursor.close()
            return affected_rows
    
    def batch_execute(self, sql: str, params_list: List[tuple]) -> int:
        """Execute SQL in batch"""
        with self.get_connection() as conn:
            cursor = conn.cursor()
            affected_rows = cursor.executemany(sql, params_list)
            cursor.close()
            return affected_rows
    
    def check_health(self) -> bool:
        """Health check"""
        try:
            result = self.execute_query("SELECT 1 as health")
            return result[0]['health'] == 1
        except Exception as e:
            self.logger.error(f"Health check failed: {e}")
            return False

# Usage example
db = SeekDBManager()

# Health check
if db.check_health():
    print("✅ SeekDB connection OK")
else:
    print("❌ SeekDB connection failed")

Data Model Design

Our knowledge base needs to store:

• The document’s title and content;
• The document’s vector representation (for semantic search);
• The document’s category and tags;
• Creation and update times;
• Access permissions (department ID).

-- Create the database
CREATE DATABASE knowledge_base;
USE knowledge_base;

-- Create the documents table
CREATE TABLE documents (
  id BIGINT PRIMARY KEY AUTO_INCREMENT,
  title VARCHAR(500) NOT NULL,
  content TEXT NOT NULL,
  category VARCHAR(100),
  tags VARCHAR(500),  -- comma-separated tags
  department_id INT,
  embedding VECTOR(1536) NOT NULL,  -- dimensionality of OpenAI text-embedding-3-small
  created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  last_accessed TIMESTAMP NULL,
  INDEX idx_category (category),
  INDEX idx_department (department_id),
  INDEX idx_created (created_at)
);

-- Create the vector index
CREATE VECTOR INDEX idx_embedding ON documents(embedding)
WITH (
  distance_metric='cosine',
  index_type='hnsw',
  m=16,
  ef_construction=200
);

-- Create the full-text index (for MATCH AGAINST)
CREATE FULLTEXT INDEX ft_content ON documents(content);

Key points explained:

• VECTOR(1536): I’m using OpenAI’s text-embedding-3-small model, which outputs 1536-dimensional vectors.
• distance_metric='cosine': cosine distance suits text vectors and is unaffected by vector length.
• index_type='hnsw': the HNSW algorithm strikes a good balance between recall and performance.
• m=16, ef_construction=200: these are my tested optimal parameters, keeping query latency <50ms at 100,000 rows.

Data Model Design Diagram

Vector Index Parameters Explained

2.2 Data Ingestion and Vectorization

I exported 200 technical documents from the company’s internal wiki, in Markdown format. First I needed to split the documents into suitable chunks, then call the OpenAI API to generate vectors.

Here’s the Python script I wrote (the key parts):

import os
import pymysql
from typing import List
from openai import OpenAI

# Configuration
client = OpenAI()  # reads the key from the OPENAI_API_KEY environment variable
DB_CONFIG = {
    'host': '127.0.0.1',
    'port': 2881,
    'user': 'root',
    'password': '',
    'database': 'knowledge_base'
}

def chunk_text(text: str, max_length: int = 1000) -> List[str]:
    """Split long text into small chunks"""
    chunks = []
    paragraphs = text.split('\n\n')
    current_chunk = ""
    
    for para in paragraphs:
        if len(current_chunk) + len(para) < max_length:
            current_chunk += para + "\n\n"
        else:
            if current_chunk:
                chunks.append(current_chunk.strip())
            current_chunk = para + "\n\n"
    
    if current_chunk:
        chunks.append(current_chunk.strip())
    
    return chunks

def get_embedding(text: str) -> List[float]:
    """Call the OpenAI API to get a vector"""
    resp = client.embeddings.create(
        model="text-embedding-3-small",
        input=text
    )
    return resp.data[0].embedding

def insert_document(conn, title: str, content: str, category: str, 
                   tags: str, department_id: int):
    """Insert a document into SeekDB"""
    # Generate the vector
    embedding = get_embedding(content)
    embedding_str = '[' + ','.join(map(str, embedding)) + ']'
    
    # Insert into the database
    cursor = conn.cursor()
    sql = """
    INSERT INTO documents (title, content, category, tags, department_id, embedding)
    VALUES (%s, %s, %s, %s, %s, %s)
    """
    cursor.execute(sql, (title, content, category, tags, department_id, embedding_str))
    conn.commit()
    cursor.close()

# Main flow
conn = pymysql.connect(**DB_CONFIG)

# Example: import one document
doc_content = """
# Python Async Programming Best Practices

When using asyncio for async programming in Python, keep the following in mind...
"""

chunks = chunk_text(doc_content)
for i, chunk in enumerate(chunks):
    insert_document(
        conn,
        title=f"Python Async Programming Best Practices - Part {i+1}",
        content=chunk,
        category="Programming Languages",
        tags="Python,async,asyncio",
        department_id=1
    )

conn.close()

Performance Optimization for Batch Import

At first I inserted documents one at a time; importing 200 documents (about 800 chunks after splitting) took 15 minutes. After switching to batch inserts, the time dropped to 3 minutes:

def batch_insert_documents(conn, documents: List[dict], batch_size: int = 50):
    """Batch-insert documents"""
    cursor = conn.cursor()
    
    for i in range(0, len(documents), batch_size):
        batch = documents[i:i+batch_size]
        
        # Build the batch INSERT SQL
        sql = """
        INSERT INTO documents (title, content, category, tags, department_id, embedding)
        VALUES """ + ','.join(['(%s, %s, %s, %s, %s, %s)'] * len(batch))
        
        # Flatten the parameters
        params = []
        for doc in batch:
            params.extend([
                doc['title'], doc['content'], doc['category'],
                doc['tags'], doc['department_id'], doc['embedding']
            ])
        
        cursor.execute(sql, params)
        conn.commit()
    
    cursor.close()

Lessons Learned

• Document chunking matters a lot: too long hurts retrieval precision, too short loses context. In my testing, 800–1200 characters is the sweet spot.
• The OpenAI API has rate limits; add retry logic and exponential backoff.
• Vector generation is the most time-consuming step; consider using a local model (such as sentence-transformers) to speed it up.

A Complete Document-Import Utility Class

import time
import openai
from typing import List, Dict
from tenacity import retry, stop_after_attempt, wait_exponential

class DocumentImporter:
    """Document import utility class"""
    
    def __init__(self, db_manager: SeekDBManager, openai_api_key: str):
        self.db = db_manager
        openai.api_key = openai_api_key
        self.batch_size = 50
    
    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
    def get_embedding_with_retry(self, text: str) -> List[float]:
        """Vector generation with retry"""
        response = openai.embeddings.create(
            model="text-embedding-3-small",
            input=text
        )
        return response.data[0].embedding
    
    def import_documents(self, documents: List[Dict]) -> Dict[str, int]:
        """Batch-import documents"""
        stats = {'success': 0, 'failed': 0, 'total': len(documents)}
        
        for i in range(0, len(documents), self.batch_size):
            batch = documents[i:i+self.batch_size]
            
            # Generate vectors in batch
            embeddings = []
            for doc in batch:
                try:
                    embedding = self.get_embedding_with_retry(doc['content'])
                    embeddings.append(embedding)
                except Exception as e:
                    print(f"Vector generation failed: {doc['title']}, error: {e}")
                    embeddings.append(None)
            
            # Insert into the database in batch
            sql = """
            INSERT INTO documents (title, content, category, tags, department_id, embedding)
            VALUES (%s, %s, %s, %s, %s, %s)
            """
            
            params_list = []
            for doc, embedding in zip(batch, embeddings):
                if embedding is None:
                    stats['failed'] += 1
                    continue
                
                embedding_str = '[' + ','.join(map(str, embedding)) + ']'
                params_list.append((
                    doc['title'],
                    doc['content'],
                    doc.get('category', ''),
                    doc.get('tags', ''),
                    doc.get('department_id', 1),
                    embedding_str
                ))
            
            try:
                self.db.batch_execute(sql, params_list)
                stats['success'] += len(params_list)
                print(f"✅ Imported {stats['success']}/{stats['total']} documents")
            except Exception as e:
                stats['failed'] += len(params_list)
                print(f"❌ Batch insert failed: {e}")
            
            # Avoid API rate limits
            time.sleep(1)
        
        return stats
    
    def import_from_markdown_files(self, file_paths: List[str]) -> Dict[str, int]:
        """Batch-import from Markdown files"""
        documents = []
        
        for file_path in file_paths:
            with open(file_path, 'r', encoding='utf-8') as f:
                content = f.read()
            
            # Extract the title (first line)
            lines = content.split('\n')
            title = lines[0].replace('#', '').strip() if lines else file_path
            
            # Split the document
            chunks = chunk_text(content, max_length=1000)
            
            for i, chunk in enumerate(chunks):
                documents.append({
                    'title': f"{title} - Part {i+1}",
                    'content': chunk,
                    'category': 'Technical Documentation',
                    'tags': 'markdown',
                    'department_id': 1
                })
        
        return self.import_documents(documents)

# Usage example
db = SeekDBManager()
importer = DocumentImporter(db, openai_api_key="your-api-key")

# Import from Markdown files
file_paths = [
    'docs/python-async.md',
    'docs/docker-guide.md',
    'docs/kubernetes-intro.md'
]
stats = importer.import_from_markdown_files(file_paths)
print(f"Import complete: {stats['success']} succeeded, {stats['failed']} failed")

2.3 Smart Search and Hybrid Retrieval

Document Processing Flow

Chunk Size Comparison Test

Implementing Semantic Search

The most basic semantic search implementation:

def semantic_search(query: str, top_k: int = 5) -> List[dict]:
    """Semantic search"""
    # 1. Convert the query into a vector
    query_embedding = get_embedding(query)
    embedding_str = '[' + ','.join(map(str, query_embedding)) + ']'

    # 2. Vector similarity search (parameterized, and update the access time)
    conn = pymysql.connect(**DB_CONFIG)
    cursor = conn.cursor(pymysql.cursors.DictCursor)

    sql = (
        "SELECT id, title, content, category, tags, "
        "       COSINE_DISTANCE(embedding, CAST(%s AS VECTOR(1536))) as distance "
        "FROM documents "
        "ORDER BY distance ASC "
        "LIMIT %s"
    )

    cursor.execute(sql, (embedding_str, top_k))
    results = cursor.fetchall()

    if results:
        ids = [row['id'] for row in results]
        update_sql = "UPDATE documents SET last_accessed = NOW() WHERE id IN (" + ",".join(["%s"]*len(ids)) + ")"
        cursor.execute(update_sql, ids)
        conn.commit()

    cursor.close()
    conn.close()

    return results

# Test
results = semantic_search("How do I use async programming in Python?")
for doc in results:
    print(f"[{doc['distance']:.4f}] {doc['title']}")

Sample output:

[0.1234] Python Async Programming Best Practices - Part 1
[0.1567] A Detailed Look at the asyncio Event Loop
[0.2103] Performance Comparison of Coroutines and Multithreading
[0.2456] A Guide to FastAPI Async Endpoint Development
[0.2789] The Complete Guide to Python Concurrent Programming

Vector Similarity Distribution

Hybrid Search: Vector + Full-text + Filtering

This is my most-used search approach, combining three retrieval capabilities:

def hybrid_search(query: str, category: str = None,
                 department_id: int = None, top_k: int = 5) -> List[dict]:
    """Hybrid search: vector similarity + full-text retrieval + structured filtering"""
    query_embedding = get_embedding(query)
    embedding_str = '[' + ','.join(map(str, query_embedding)) + ']'

    conn = pymysql.connect(**DB_CONFIG)
    cursor = conn.cursor(pymysql.cursors.DictCursor)

    where_clauses = []
    where_params = []
    if category:
        where_clauses.append("category = %s")
        where_params.append(category)
    if department_id:
        where_clauses.append("department_id = %s")
        where_params.append(department_id)
    where_sql = " AND ".join(where_clauses) if where_clauses else "1=1"

    sql = f"""
    SELECT id, title, content, category, tags, vec_distance, text_score
    FROM (
      SELECT id, title, content, category, tags,
             COSINE_DISTANCE(embedding, CAST(%s AS VECTOR(1536))) AS vec_distance,
             MATCH(content) AGAINST(%s IN NATURAL LANGUAGE MODE) AS text_score
      FROM documents
      WHERE {where_sql}
    ) t
    WHERE t.vec_distance < 0.5 OR t.text_score > 0
    ORDER BY (t.vec_distance * 0.7 + (1 - t.text_score) * 0.3) ASC
    LIMIT %s
    """

    params = [embedding_str, query] + where_params + [top_k]
    cursor.execute(sql, params)
    results = cursor.fetchall()

    if results:
        ids = [row['id'] for row in results]
        update_sql = "UPDATE documents SET last_accessed = NOW() WHERE id IN (" + ",".join(["%s"]*len(ids)) + ")"
        cursor.execute(update_sql, ids)
        conn.commit()

    cursor.close()
    conn.close()
    return results

# Test: search for Python-related documents under the Programming Languages category
results = hybrid_search(
    query="Performance optimization for async programming",
    category="Programming Languages",
    department_id=1
)

Key technical points:

• vec_distance < 0.5: filters out results with too-low similarity.
• text_score > 0: ensures there’s a keyword match.
• vec_distance * 0.7 + (1 - text_score) * 0.3: a weighted fusion of the two scores, giving vector search a higher weight.
• This weight ratio was tuned on real data; your scenario may need a different ratio.

Hybrid Search Weight Tuning

Test Results for Different Weight Ratios

Performance Test Results

I ran a load test on 800 documents (using Apache Bench):

For our scenario (an internal knowledge base with low concurrency), this performance is more than enough.

Performance Comparison Chart

Latency Distribution

2.4 RAG Application Integration and Real-World Results

The RAG Workflow

With search in place, the next step is the complete RAG (Retrieval-Augmented Generation) flow:

from openai import OpenAI

client = OpenAI()  # reads the key from the OPENAI_API_KEY environment variable (reuse the client if already created above)

def rag_query(user_question: str, department_id: int):
    """Complete RAG Q&A flow"""

    relevant_docs = hybrid_search(
        query=user_question,
        department_id=department_id,
        top_k=3
    )

    if not relevant_docs:
        return {"answer": "Sorry, I couldn't find any relevant documents.", "sources": []}

    context = "\n\n---\n\n".join([
        f"Document title: {doc['title']}\nContent: {doc['content']}"
        for doc in relevant_docs
    ])

    prompt = f"""You are a professional technical assistant. Please answer the user's question based on the following document content.
If the documents don't contain relevant information, clearly tell the user.

Reference documents:
{context}

User question: {user_question}

Please give a detailed and accurate answer:"""

    resp = client.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": "You are a professional technical assistant."},
            {"role": "user", "content": prompt}
        ],
        temperature=0.3
    )

    answer = resp.choices[0].message.content
    sources = [{"title": doc['title'], "id": doc['id']} for doc in relevant_docs]
    return {"answer": answer, "sources": sources}

# Test
result = rag_query("How do I handle exceptions in Python async programming?", department_id=1)
print(result['answer'])
print("\nReference documents:")
for source in result['sources']:
    print(f"- {source['title']} (ID: {source['id']})")

Real-World Results and User Feedback

I ran a one-week canary test inside the company and collected user feedback.

Positive feedback:

• “The search results are much more accurate than before—it understands what I mean.”
• “It responds very fast, basically instant.”
• “The documents it cites are all relevant, unlike before when it often missed the point.”

Problems encountered:

• Hallucination. The large model sometimes fabricates content that doesn’t exist.
  • Solution: emphasize in the prompt “answer only based on the provided documents” and lower the temperature to 0.3.
• Stale documents. Some users reported that the documents found were old versions.
  • Solution: added document version management and return the latest version first when searching.
• Cross-document synthesis questions. When the answer requires synthesizing multiple documents, the results weren’t ideal.
  • Solution: increased top_k to 5 and tuned the prompt so the model integrates information better.

RAG Improvement Comparison

Summary of Optimization Measures

2.5 Pitfalls and Lessons from the Migration

Vector Index Parameter Tuning

At first I used the default parameters, and once the data grew to 5,000 rows, query latency shot up to 300ms+. I later found the HNSW index’s ef_search parameter was too small.

-- Adjust the search parameter (this is session-level)
SET ef_search = 100;

After testing, ef_search=100 was the best value in my scenario, with recall of 99%+ and latency kept within 50ms.

ef_search Parameter Tuning Test

Parameter Comparison Table

Tokenization Issues

seekdb’s full-text search uses a generic tokenizer by default, which handles technical terms poorly. For example, “Kubernetes” gets split into “Kuber” and “netes.”

The solution is to tokenize properly before insertion, or to use a strategy that prioritizes vector search with full-text search as a supplement.

Tokenization Issue Example

Choosing Vector Dimensionality

I originally used OpenAI’s text-embedding-ada-002 (1536 dimensions), then switched to text-embedding-3-small (also 1536 dimensions) and found the results clearly improved—and it’s cheaper too.

Embedding Model Comparison

Recommendations:

• For Chinese scenarios, consider a local model such as bge-large-zh (1024 dimensions).
• If you’re cost-sensitive, text-embedding-3-small is a great choice.
• Don’t blindly chase high dimensionality; the higher the dimensions, the greater the storage and compute cost.

The Relationship Between Dimensionality and Performance

Designing a Sensible Chunk Strategy

Document chunking greatly affects retrieval quality. My strategy is:

• Split by paragraph to preserve semantic integrity.
• Each chunk is 800–1200 characters.
• Keep a 100-character overlap between chunks to avoid cutting off key information.
• Keep each chunk’s original document title and metadata.

Chunk Overlap Strategy Illustration

Monitoring and Logging

I added detailed logging and monitoring in production:

import time
import logging

def semantic_search_with_logging(query: str, top_k: int = 5):
    start_time = time.time()
    
    try:
        results = semantic_search(query, top_k)
        
        # Record the query log
        logging.info({
            "query": query,
            "top_k": top_k,
            "result_count": len(results),
            "latency_ms": (time.time() - start_time) * 1000,
            "top_distance": results[0]['distance'] if results else None
        })
        
        return results
    except Exception as e:
        logging.error(f"Search failed: {e}")
        raise

These logs helped me uncover many problems—for example, certain queries being especially slow, or certain documents never being retrieved.

A Complete Monitoring and Performance-Analysis Tool

import time
from datetime import datetime, timedelta
from collections import defaultdict
import json

class SeekDBMonitor:
    """SeekDB monitoring tool"""
    
    def __init__(self, db_manager: SeekDBManager):
        self.db = db_manager
        self.query_stats = defaultdict(list)
    
    def log_query(self, query_type: str, query: str, latency_ms: float, 
                  result_count: int, metadata: dict = None):
        """Record a query log"""
        log_entry = {
            'timestamp': datetime.now().isoformat(),
            'query_type': query_type,
            'query': query[:100],  # only record the first 100 characters
            'latency_ms': latency_ms,
            'result_count': result_count,
            'metadata': metadata or {}
        }
        self.query_stats[query_type].append(log_entry)
        
        # Record to a file
        with open('seekdb_query.log', 'a', encoding='utf-8') as f:
            f.write(json.dumps(log_entry, ensure_ascii=False) + '\n')
    
    def get_performance_report(self, hours: int = 24) -> Dict:
        """Generate a performance report"""
        cutoff_time = datetime.now() - timedelta(hours=hours)
        
        report = {
            'period': f'Last {hours} hours',
            'query_types': {}
        }
        
        for query_type, logs in self.query_stats.items():
            recent_logs = [
                log for log in logs
                if datetime.fromisoformat(log['timestamp']) > cutoff_time
            ]
            
            if not recent_logs:
                continue
            
            latencies = [log['latency_ms'] for log in recent_logs]
            latencies.sort()
            
            report['query_types'][query_type] = {
                'total_queries': len(recent_logs),
                'avg_latency': sum(latencies) / len(latencies),
                'p50_latency': latencies[len(latencies) // 2],
                'p95_latency': latencies[int(len(latencies) * 0.95)],
                'p99_latency': latencies[int(len(latencies) * 0.99)],
                'max_latency': max(latencies),
                'min_latency': min(latencies)
            }
        
        return report
    
    def check_slow_queries(self, threshold_ms: float = 100) -> List[Dict]:
        """Check for slow queries"""
        slow_queries = []
        
        for query_type, logs in self.query_stats.items():
            for log in logs:
                if log['latency_ms'] > threshold_ms:
                    slow_queries.append(log)
        
        # Sort by latency
        slow_queries.sort(key=lambda x: x['latency_ms'], reverse=True)
        return slow_queries[:20]  # return the 20 slowest
    
    def analyze_document_coverage(self) -> Dict:
        """Analyze document coverage"""
        # Query the total number of documents
        total_docs = self.db.execute_query(
            "SELECT COUNT(*) as count FROM documents"
        )[0]['count']
        
        # Query the number of documents retrieved in the last 30 days
        accessed_docs = self.db.execute_query("""
            SELECT COUNT(DISTINCT id) as count 
            FROM documents 
            WHERE last_accessed > DATE_SUB(NOW(), INTERVAL 30 DAY)
        """)
        
        coverage = (accessed_docs[0]['count'] / total_docs * 100) if total_docs > 0 else 0
        
        return {
            'total_documents': total_docs,
            'accessed_documents': accessed_docs[0]['count'],
            'coverage_percentage': round(coverage, 2),
            'unused_documents': total_docs - accessed_docs[0]['count']
        }
    
    def get_system_metrics(self) -> Dict:
        """Get system metrics"""
        metrics = {}
        
        # Database size
        size_result = self.db.execute_query("""
            SELECT 
                table_schema as db_name,
                SUM(data_length + index_length) / 1024 / 1024 as size_mb
            FROM information_schema.tables
            WHERE table_schema = 'knowledge_base'
            GROUP BY table_schema
        """)
        metrics['database_size_mb'] = size_result[0]['size_mb'] if size_result else 0
        
        # Table statistics
        table_stats = self.db.execute_query("""
            SELECT 
                table_name,
                table_rows,
                ROUND((data_length + index_length) / 1024 / 1024, 2) as size_mb
            FROM information_schema.tables
            WHERE table_schema = 'knowledge_base'
        """)
        metrics['tables'] = table_stats
        
        # Index usage
        index_stats = self.db.execute_query("""
            SELECT 
                table_name,
                index_name,
                cardinality
            FROM information_schema.statistics
            WHERE table_schema = 'knowledge_base'
        """)
        metrics['indexes'] = index_stats
        
        return metrics
    
    def print_report(self):
        """Print the monitoring report"""
        print("=" * 60)
        print("SeekDB Performance Monitoring Report")
        print("=" * 60)
        
        # Performance report
        perf_report = self.get_performance_report(24)
        print(f"\n📊 Query Performance ({perf_report['period']})")
        for query_type, stats in perf_report['query_types'].items():
            print(f"\n  {query_type}:")
            print(f"    Total queries: {stats['total_queries']}")
            print(f"    Avg latency: {stats['avg_latency']:.2f}ms")
            print(f"    P95 latency: {stats['p95_latency']:.2f}ms")
            print(f"    P99 latency: {stats['p99_latency']:.2f}ms")
        
        # Slow queries
        slow_queries = self.check_slow_queries(100)
        if slow_queries:
            print(f"\n⚠️  Slow queries (>{100}ms):")
            for i, query in enumerate(slow_queries[:5], 1):
                print(f"  {i}. {query['latency_ms']:.2f}ms - {query['query']}")
        
        # Document coverage
        coverage = self.analyze_document_coverage()
        print(f"\n📚 Document coverage:")
        print(f"    Total documents: {coverage['total_documents']}")
        print(f"    Accessed: {coverage['accessed_documents']}")
        print(f"    Coverage: {coverage['coverage_percentage']}%")
        print(f"    Unused: {coverage['unused_documents']}")
        
        # System metrics
        metrics = self.get_system_metrics()
        print(f"\n💾 System metrics:")
        print(f"    Database size: {metrics['database_size_mb']:.2f}MB")
        print(f"    Table count: {len(metrics['tables'])}")
        
        print("\n" + "=" * 60)

# Usage example
db = SeekDBManager()
monitor = SeekDBMonitor(db)

# Log during a query
start_time = time.time()
results = semantic_search("Python async programming")
latency = (time.time() - start_time) * 1000

monitor.log_query(
    query_type='semantic_search',
    query='Python async programming',
    latency_ms=latency,
    result_count=len(results),
    metadata={'top_k': 5}
)

# Generate the report
monitor.print_report()

Monitoring Dashboard

Key Monitoring Metrics

3. Real-World Results in Production

Beyond knowledge-base Q&A, seekdb can also be used in many AI scenarios.

3.1 Performance Comparison Data

Our team built a customer-service bot with seekdb, storing historical tickets and standard answers. When a user asks a question, the system retrieves similar historical cases and then generates an answer. This solution was later rolled out across several business lines inside the company, performing far better than the previous keyword matching.

Intelligent Customer Service Architecture

# Search for the customer-service scenario
def search_similar_tickets(user_question: str, top_k: int = 3):
    query_embedding = get_embedding(user_question)
    
    sql = """
    SELECT ticket_id, question, answer, resolution_time
    FROM support_tickets
    WHERE status = 'resolved'
    ORDER BY COSINE_DISTANCE(question_embedding, %s)
    LIMIT %s
    """
    # Returns similar historical tickets

Customer Service Results Comparison

3.2 Practice in Typical Application Scenarios

Another interesting application is code search. We vectorized the company’s codebase and stored it in seekdb, so developers can search for code snippets using natural language.

For example, searching “how to connect to Redis and set an expiration time” finds the relevant code examples. This feature is very popular with the engineering team and greatly speeds up newcomers’ ramp-up.

Code Search Workflow:

Code Search Results:

3.3 Improved Operations Experience

In e-commerce scenarios, you can turn users’ browsing history and purchase records into vectors, then use seekdb to find similar products to recommend.

Recommendation System Architecture:

-- Recommend products based on the user-interest vector
SELECT product_id, product_name, price
FROM products
WHERE stock > 0 AND category IN ('electronics', 'books')
ORDER BY COSINE_DISTANCE(product_vector, '[user interest vector]')
LIMIT 20;

Improved Recommendation Results:

4. Technology Selection Comparison

Before deciding on seekdb, I compared a few mainstream options.

Option 1: PostgreSQL + pgvector.

• Pros: a mature ecosystem; the pgvector plugin is free and open source.
• Cons: mediocre vector-search performance, weak full-text search, and poor Chinese support.
• Test result: on 5,000 rows, query latency was around 150ms—2–3x slower than SeekDB.

Option 2: Milvus.

• Pros: a professional vector database with strong performance, supporting multiple index algorithms.
• Cons: complex to deploy (requires configuring dependencies like etcd and MinIO), no SQL support, and weak structured-query capability.
• Experience: just getting Milvus running took half a day, and it still needs MySQL alongside it.

Option 3: Elasticsearch.

• Pros: powerful full-text search, a mature ecosystem, and rich tooling.
• Cons: no vector search, high memory usage, and complex query syntax.

Comparison Summary:

Conclusion: seekdb reaches production-ready levels across vector search, full-text search, and structured queries, and is simple to deploy with a low learning curve—making it the best choice for AI applications.

5. Production Operating Data and Outlook

5.1 Architecture Comparison and Real Data

Our knowledge base now has:

• Documents: 1,200
• Document chunks: 4,800
• Daily queries: about 500
• Concurrent users: 20–30

Query Volume Distribution (by time of day):

At this scale, seekdb runs on a single 4-core, 8G cloud server with average CPU usage of 15% and memory usage of about 3GB—more than enough.

Resource Usage Monitoring:

Daily Operating Metrics:

5.2 Follow-up Plans and Outlook

Building on seekdb, I plan to keep optimizing and expanding the functionality.

Short-term plans (1–2 months)

• Multimodal support: add vectorization and retrieval of images and tables.
• Personalized recommendation: optimize search-result ranking based on user history.
• Document version management: support version tracking and rollback for documents.

Mid-term plans (3–6 months)

• Knowledge graph: build relationships between documents.
• Auto-annotation: use a large model to automatically extract document tags and summaries.
• Multi-tenant isolation: support data isolation across departments.

Technical exploration

• Try replacing the OpenAI API with a local embedding model (such as bge-large-zh) to cut costs.
• Study seekdb’s distributed deployment options to prepare for data growth.
• Explore integration with frameworks like LangChain and LlamaIndex.

A Few Final Words

From first encountering seekdb to finishing the system refactor, I spent only two weekends. The process drove home how much a good tool can boost development efficiency. What impressed me most about seekdb isn’t how advanced its technology is, but that it truly understands the pain points of AI application developers: we need vector search, but don’t want to bring in a complex specialized database for it; we need full-text search, but don’t want to maintain an Elasticsearch cluster; we need structured queries, but don’t want to sync data across multiple databases. seekdb integrates these needs into one lightweight database, letting me focus on business logic instead of wrestling with infrastructure.

As a veteran who’s been writing technical blogs since 2015, I’ve witnessed the rise and fall of countless technologies. What truly lasts is often not the flashiest technology, but the tools that genuinely solve problems and lower barriers. seekdb is exactly that kind of tool.

If you’re also building AI applications, and you’re also plagued by a multi-database architecture, give seekdb a try. It’s open source, the code is on GitHub, the official docs are thorough, and the community is active. I’ll keep sharing my seekdb experience and best practices on my CSDN blog. Thanks to the OceanBase team for open-sourcing such an excellent project. I look forward to seekdb’s future, and to more developers joining this ecosystem.

Summary

seekdb’s greatest value lies in lowering the barrier to AI application development: MySQL compatibility lets existing tools work directly, the lightweight design makes deployment and operations simple, the open-source nature removes vendor-lock-in concerns, and—most importantly—it integrates capabilities scattered across multiple systems, letting developers focus on business logic rather than infrastructure.

Key lessons from real-world use include:

• Design a sensible chunk strategy (800–1200 characters is best).
• Choose the right embedding model (text-embedding-3-small offers the best value).
• Tune HNSW index parameters (ef_search=100 is best in production).
• Build a solid monitoring system.

These lessons helped our knowledge base run stably at a scale of 1,200 documents and 4,800 chunks, raising user satisfaction from 70% to 92%. If you’re also building a RAG application, recommendation system, or smart search engine, I strongly recommend trying seekdb. As an emerging AI-native database, seekdb is iterating fast, and I believe it will become the go-to database for AI application development.

I’m Bailu, a programmer who keeps striving. I hope this article helps you—feel free to like, comment, and share! If you have other questions, suggestions, or additions, leave a comment below the article. Thanks for your support!

About the author: Guo Jing (pen name “Bailu Diyishuai”) is currently a big-data and large-model development engineer at a major internet company. He has worked at several well-known internet companies and cloud vendors, with rich experience in enterprise big-data development and large-model applications.

As a figure on the annual list of influential Chinese developers, Guo Jing has been creating technical content continuously for 11 years, from 2015 to today. His personal CSDN blog has published over 300 technical articles and reviews, with more than 60,000 followers across all platforms and total views exceeding 1,500,000. He has earned multiple technical-community certifications, including CSDN “Blog Expert” and “Quality Java Creator,” OSCHINA “Outstanding Original Author,” Tencent Cloud TDP, Alibaba Cloud “Expert Blogger,” and Huawei Cloud “Huawei Cloud Expert,” and has become a member of the top-tier internet technical guild “Polaris Club.”