Будування системи RAG за допомогою TimescaleDB

RAG — це не лише векторний пошук | Timescale

Методика Retrieval Augmented Generation (RAG) стала потужним підходом для створення застосунків, що потребують великої кількості знань. Однак розробники часто стикаються з серйозними проблемами:

• Управління кількома базами даних для векторів та метаданих
• Забезпечення узгодженості даних між системами
• Обробка складних запитів, які поєднують векторний пошук з традиційними даними
• Масштабування систем ефективно з точки зору вартості

Хоча багато рішень зосереджуються на спеціалізованих векторних базах даних, TimescaleDB пропонує переконливу альтернативу: базу даних PostgreSQL корпоративного рівня, яка обробляє як векторні операції, так і традиційні навантаження з даними. У цій статті ми побудуємо масштабовану систему RAG за допомогою TimescaleDB, розглядаючи, як перейти від доказу концепції до вирішення реальних завдань.

Чому TimescaleDB для RAG?

Типова архітектура RAG часто передбачає використання кількох баз даних: векторного сховища для векторів та традиційної бази даних для метаданих і зв’язків. TimescaleDB пропонує переконливу альтернативу:

1. Єдине джерело правди: Зберігання векторів, метаданих і зв’язків в одній базі даних
2. Сумісність з PostgreSQL: Використання багатого екосистеми PostgreSQL
3. Функції для виробництва: Вбудоване стиснення, резервне копіювання та моніторинг
4. Економічно вигідне рішення: Не потрібно окремих баз даних для векторних та традиційних даних

Попередні вимоги

Перед тим, як почати, переконайтеся, що у вас встановлено:

# Вимоги до програмного забезпечення  
PostgreSQL >= 14.0  
TimescaleDB >= 2.11.0  
Python >= 3.8

# Залежності для Python  
pip install -r requirements.txt

Створіть файл requirements.txt з такими залежностями:

psycopg2-binary>=2.9.9  
langchain>=0.0.267  
openai>=0.28.0  
python-dotenv>=1.0.0  
numpy>=1.24.0  
cachetools>=5.3.2

Налаштуйте змінні середовища:

# файл .env  
TIMESCALE_HOST=localhost  
TIMESCALE_PORT=5432  
TIMESCALE_DB=rag_system  
TIMESCALE_USER=your_user  
TIMESCALE_PASSWORD=your_password  
OPENAI_API_KEY=your_openai_key

Архітектура системи

Наша система RAG буде включати:

• Конвеєр введення документів і їх сегментування
• Генерація векторів і їх зберігання
• Пошук подібності векторів
• Інтеграція з LLM для генерації відповідей
• Моніторинг продуктивності та оптимізація

Давайте побудуємо кожен компонент поетапно.

Налаштування TimescaleDB з pgvector

Почнемо з налаштування нашої бази даних. Нам потрібно створити дві основні таблиці: одну для документів і іншу для сегментів документів з їх векторами.

-- Увімкнення необхідних розширень  
CREATE EXTENSION IF NOT EXISTS vector;  
CREATE EXTENSION IF NOT EXISTS pgai;  

-- Створення таблиці документів  
CREATE TABLE documents (  
 id SERIAL PRIMARY KEY,  
 title TEXT NOT NULL,  
 content TEXT NOT NULL,  
 metadata JSONB,  
 created_at TIMESTAMPTZ DEFAULT NOW()  
);  

-- Створення таблиці для сегментів документів  
CREATE TABLE document_chunks (  
 id SERIAL PRIMARY KEY,  
 document_id INTEGER REFERENCES documents(id),  
 chunk_index INTEGER,  
 content TEXT NOT NULL,  
 embedding vector(1536),  
 created_at TIMESTAMPTZ DEFAULT NOW()  
);  

-- Створення індексу для пошуку подібності векторів  
CREATE INDEX ON document_chunks USING ivfflat (embedding vector_cosine_ops)  
WITH (lists = 100);

Це початкове налаштування додає підтримку векторних операцій через pgvector та можливості для штучного інтелекту через pgai.
Розширення для векторів дозволяє виконувати пошук за подібністю, а pgai допоможе нам інтегрувати різні моделі штучного інтелекту.

Цей дизайн схеми має кілька цілей:

• Таблиця documents зберігає оригінальні документи з їх метаданими
• Таблиця document_chunks містить текстові сегменти та їх вектори
• Використання типу JSONB для метаданих дозволяє зберігати гнучкі атрибути документів
• Мітки часу дають змогу відслідковувати операції та обслуговування

Індекс IVFFlat з 100 списками обраний через те, що:

• Він забезпечує хороший баланс між швидкістю пошуку та точністю для наборів даних до 1 мільйона векторів
• Кожен список містить приблизно 1% векторів, що дозволяє ефективно виконувати пошук
• Використання пам'яті залишається помірним при збереженні хорошого охоплення

Індекс пошуку подібності векторів значно прискорює пошук за подібністю. Параметр lists подібний до створення 100 кошиків для векторів, де деяка точність жертвується на користь швидкості. Для більшості застосунків 100 списків забезпечують хороший баланс, але ви можете налаштувати це залежно від розміру даних і потреб у точності.

Конвеєр обробки документів

Ось наш код Python для обробки документів:

import psycopg2  
from psycopg2.extras import execute_values  
from langchain.text_splitter import RecursiveCharacterTextSplitter  
from langchain.embeddings import OpenAIEmbeddings  
import numpy as np  

class DocumentProcessor:  
 def __init__(self, db_params, batch_size=100):  
 self.db_params = db_params  
 self.batch_size = batch_size  
 self.text_splitter = RecursiveCharacterTextSplitter(  
 chunk_size=500,  
 chunk_overlap=50  
 )  
 self.embeddings_model = OpenAIEmbeddings()  

 def process_document(self, title, content, metadata=None):  
 # Розбиваємо документ на сегменти  
 chunks = self.text_splitter.split_text(content)  

 with psycopg2.connect(**self.db_params) as conn:  
 with conn.cursor() as cur:  
 # Вставка документа  
 cur.execute("""  
 INSERT INTO documents (title, content, metadata)  
 VALUES (%s, %s, %s)  
 RETURNING id  
 """, (title, content, metadata))  
 document_id = cur.fetchone()[0]  

 # Обробка сегментів у пакетах  
 for i in range(0, len(chunks), self.batch_size):  
 batch = chunks[i:i + self.batch_size]  

 # Генерація векторів  
 embeddings = self.embeddings_model.embed_documents(batch)  

 # Підготовка даних для масового вставлення  
 chunk_data = [  
 (document_id, idx + i, chunk, embedding)  
 for idx, (chunk, embedding) in enumerate(zip(batch, embeddings))  
 ]  

 # Масове вставлення сегментів і векторів  
 execute_values(cur, """  
 INSERT INTO document_chunks   
 (document_id, chunk_index, content, embedding)  
 VALUES %s  
 """, chunk_data)  

 conn.commit()

Цей клас обробника виконує три важливі завдання:

1. Розбивка документів на зручні сегменти
2. Генерація векторів для кожного сегмента
3. Збереження всього в TimescaleDB

Розмір сегмента в 500 символів з перекриттям у 50 символів — це хороша початкова точка, але вам можливо доведеться налаштувати це залежно від вашого вмісту. Більші сегменти дають більше контексту, але збільшують час обробки та використання токенів.

Метод обробки документів відповідає за вставку документів. Ми використовуємо транзакцію, щоб забезпечити атомарність вставки документа та його сегментів — або все вдається, або нічого не змінюється.

Обробка сегментів у пакетах — ось де починається цікаве:

Чому пакетна обробка? Дві причини:

1. Ефективність пам'яті: Обробка тисяч сегментів за один раз може перевантажити вашу систему
2. Оптимізація API: Більшість API для генерації векторів працюють ефективніше з пакетами

Реалізація пошуку за векторами

Пошук за векторами — це те, де TimescaleDB дійсно сяє.
Створімо функцію пошуку, яка поєднує векторний пошук за подібністю з фільтрацією за метаданими:

def semantic_search(query, metadata_filter=None, limit=5):  
 query_embedding = embeddings_model.embed_query(query)  

 with psycopg2.connect(**db_params) as conn:  
 with conn.cursor() as cur:  
 # Основний запит з векторним пошуком за подібністю  
 sql = """  
 SELECT   
 dc.content,  
 d.title,  
 d.metadata,  
 1 - (dc.embedding <=> %s) as similarity  
 FROM document_chunks dc  
 JOIN documents d ON d.id = dc.document_id  
 WHERE 1=1  
 """  
 params = [query_embedding]  

 # Додаємо фільтри метаданих, якщо вони вказані  
 if metadata_filter:  
 sql += " AND d.metadata @> %s"  
 params.append(json.dumps(metadata_filter))  

 # Сортуємо за подібністю та обмежуємо кількість результатів  
 sql += """  
 ORDER BY dc.embedding <=> %s  
 LIMIT %s  
 """  
 params.extend([query_embedding, limit])  

 cur.execute(sql, params)  
 results = cur.fetchall()  

 return [  
 {  
 'content': r[0],  
 'title': r[1],  
 'metadata': r[2],  
 'similarity': r[3]  
 }  
 for r in results  
 ]

Оператор <=> обчислює косинусну відстань. Ми віднімаємо від 1, щоб перетворити відстань на подібність (чем вище значення, тим краще). Цей запит поєднує:

• Пошук за векторною подібністю
• Метадані документів
• Оцінку релевантності

Додавання фільтрів метаданих робить наш пошук потужнішим:

Оператор @> перевіряє, чи містять метадані конкретні поля, використовуючи можливості JSONB у PostgreSQL. Це дозволяє фільтрувати результати за будь-яким полем метаданих без зміни схеми.

Управління налаштуваннями

Створимо менеджер налаштувань для обробки налаштувань бази даних та API:

import os  
from typing import Dict, Any  
from dotenv import load_dotenv  

class Config:  
 def __init__(self):  
 load_dotenv()  

 @property  
 def db_params(self) -> Dict[str, Any]:  
 """Отримати параметри підключення до бази даних."""  
 return {  
 'host': os.getenv('TIMESCALE_HOST', 'localhost'),  
 'port': int(os.getenv('TIMESCALE_PORT', 5432)),  
 'database': os.getenv('TIMESCALE_DB', 'rag_system'),  
 'user': os.getenv('TIMESCALE_USER'),  
 'password': os.getenv('TIMESCALE_PASSWORD')  
 }  

 @property  
 def openai_config(self) -> Dict[str, str]:  
 """Отримати налаштування для OpenAI API."""  
 return {  
 'api_key': os.getenv('OPENAI_API_KEY')  
 }

Оптимізація для продакшн середовища

1. Впровадження смарт-кешування

class CachedEmbeddingSearch:  
 def __init__(self, cache_ttl=3600, cache_maxsize=10000):  
 self.cache = TTLCache(maxsize=cache_maxsize, ttl=cache_ttl)

Цей кеш має дві основні мети:

• Знижує витрати на API для векторів, кешуючи часті запити
• Покращує час відгуку для звичайних пошуків

2. Створення гіпертаблиці для відслідковування метрик пошуку:

CREATE TABLE search_metrics (  
 id SERIAL PRIMARY KEY,  
 query TEXT,  
 result_count INTEGER,  
 execution_time_ms FLOAT,  
 timestamp TIMESTAMPTZ DEFAULT NOW()  
);  

SELECT create_hypertable('search_metrics', 'timestamp');

Використання гіпертаблиці дозволяє ефективно зберігати та аналізувати продуктивність пошуку з часом. Ми можемо відслідковувати:

• Патерни запитів
• Час відгуку
• Якість результатів

3. Оптимізація для звичайних запитів

Для часто запитуваних даних створимо матеріалізоване представлення:

CREATE MATERIALIZED VIEW common_searches AS  
SELECT   
 dc.embedding,  
 dc.content,  
 d.title,  
 d.metadata  
FROM document_chunks dc  
JOIN documents d ON d.id = dc.document_id  
WHERE d.metadata->>'importance' = 'high';

Цей підхід:

• Преобчислює з'єднання для часто запитуваних запитів
• Зменшує обчислювальне навантаження під час виконання
• Покращує час відгуку для важливого контенту

Кращі практики та отримані уроки

1. Стратегія розбиття на частини

• Розділяйте на частини від 300 до 500 символів для збереження балансу між контекстом
• Використовуйте семантичні межі (параграфи, речення), коли це можливо
• Зважайте на тип контенту при налаштуванні перекриття

2. Управління індексами

• Регулярно запускайте VACUUM ANALYZE після масових вставок
• Слідкуйте за використанням індексів через pgstatuser_indexes
• Оцінюйте та налаштовуйте параметри IVFFlat в міру зростання даних

3.

Управління ресурсами

• Реалізуйте пулінг з'єднань за допомогою pgbouncer
• Встановіть відповідні таймоути для запитів
• Моніторте використання пам'яті та кількість з'єднань

Практичний приклад: Система підтримки клієнтів

Давайте побудуємо практичний приклад: систему підтримки клієнтів, яка поєднує історичні квитки, документацію про продукти та взаємодії з клієнтами в реальному часі.

class SupportSystem:  
 def __init__(self):  
 self.db = TimescaleDBClient(  
 host=os.environ['TIMESCALE_HOST'],  
 database='support_system'  
 )  

 async def handle_customer_query(self, query: str, customer_id: str):  
 # Спочатку отримуємо контекст клієнта  
 customer_context = await self.db.fetch_one("""  
 SELECT   
 json_build_object(  
 'recent_tickets', (  
 SELECT json_agg(t.*)  
 FROM tickets t  
 WHERE customer_id = $1  
 ORDER BY created_at DESC  
 LIMIT 5  
 ),  
 'subscription', (  
 SELECT json_build_object('plan', plan, 'status', status)  
 FROM subscriptions  
 WHERE customer_id = $1  
 )  
 ) as context  
 """, customer_id)  

 # Потім виконуємо векторний пошук з контекстом  
 results = await self.db.fetch_all("""  
 WITH query_embedding AS (  
 SELECT ai.embed($1) as embedding  
 )  
 SELECT   
 dc.content,  
 1 - (dc.embedding <=> (SELECT embedding FROM query_embedding)) as similarity,  
 d.source,  
 d.metadata  
 FROM document_chunks dc  
 JOIN documents d ON d.id = dc.document_id  
 WHERE d.metadata->>'product' = $2  
 ORDER BY dc.embedding <=> (SELECT embedding FROM query_embedding)  
 LIMIT 5  
 """, query, customer_context['subscription']['plan'])  

 return self.format_response(results, customer_context)

Цей практичний приклад демонструє кілька переваг TimescaleDB:

• Складні SQL-з'єднання з векторними операціями
• Багаті можливості роботи з JSON для метаданих
• Підтримка транзакцій для забезпечення узгодженості даних
• Не потрібно використовувати кілька баз даних

Керівництво з вибору: Коли використовувати TimescaleDB для RAG

✅ Виберіть TimescaleDB, коли вам потрібно:

• Єдине джерело правди для всіх даних
• Складні запити, що поєднують вектори та традиційні дані
• Надійність рівня підприємств та резервне копіювання
• Масштабування за вигідною ціною
• Багаті можливості моніторингу та спостережності

❌ Розгляньте альтернативи, коли:

• Вам потрібен лише простий пошук за векторною подібністю
• Ваші дані не мають часових аспектів
• Вам не потрібна підтримка ACID

Пам'ятайте: справжня вартість бази даних для векторів не лише в зберіганні — це ціла екосистема та зусилля з інженерії, необхідні для підтримки синхронізації кількох баз даних.

Висновок

TimescaleDB забезпечує надійний фундамент для побудови готових до виробництва систем RAG. Його поєднання можливостей векторного пошуку та функціоналу традиційної бази даних робить його відмінним вибором для організацій, що прагнуть впровадити RAG без керування кількома базами даних.

Основні висновки:

• Уніфіковане зберігання спрощує архітектуру
• Вбудовані інструменти моніторингу сприяють оптимізації
• Сумісність з PostgreSQL дозволяє виконувати складні запити
• Масштабованість від PoC до продакшн

Наступні кроки

Готові створити свою систему RAG? Ось кілька ресурсів для початку:

• Документація TimescaleDB
• Документація pgvector

Не соромтесь звертатися з питаннями або поділитися своїм досвідом у коментарях нижче!

Перекладено з: Building a RAG System with TimescaleDB