Индексирование: создание и сохранение фрагментов
Создайте эмбеддинг для каждого фрагмента с помощью API эмбеддингов OpenAI и добавьте полученные векторы с метаданными в векторное хранилище, построив индекс документов для поиска.
«Индексирование: создание и сохранение фрагментов» — бесплатный урок AI Engineering Academy на CoddyKit. Это урок 3 из 4. Ты можешь прочитать весь урок бесплатно ниже — а потом практиковать его прямо в браузере с встроенным редактором кода и ИИ-репетитором 24/7. Это часть пути обучения AI Engineering Academy, и твой прогресс синхронизируется между веб-версией и приложением CoddyKit. Курс AI Engineering Academy содержит 4 уроков всего.
Этап индексирования: обзор
После загрузки документов и разбиения их на фрагменты вы переходите к этапу индексирования: преобразованию текстовых фрагментов в векторные представления и сохранению их в доступной для поиска векторной базе данных. Это последний автономный этап перед обработкой запросов. Качество векторных представлений, а также эффективность хранения и индексирования напрямую определяют скорость и точность вашей системы RAG во время выполнения запросов.
Создание векторных представлений через API OpenAI
Наиболее распространённый подход — отправлять текст фрагментов в API векторных представлений OpenAI. Модель text-embedding-3-small создаёт векторы размерностью 1536 и стоит 0,02 доллара за миллион токенов, что чрезвычайно дёшево для большинства задач. Отправляйте несколько текстов в одном вызове API (до 2048 входных данных), чтобы максимально увеличить пропускную способность. Ответ содержит по одному вектору для каждого входного текста в том же порядке.
from openai import OpenAI
client = OpenAI()
def embed_batch(texts, model='text-embedding-3-small'):
response = client.embeddings.create(
model=model,
input=texts # up to 2048 texts per call
)
return [item.embedding for item in response.data]
# Embed one batch of 100 chunk texts
texts = [chunk['text'] for chunk in chunks[:100]]
vectors = embed_batch(texts)
print(f'Embedding dimension: {len(vectors[0])}')
print(f'Embedded {len(vectors)} chunks')Пакетная обработка для повышения эффективности
При индексировании тысяч фрагментов важна эффективность. Обрабатывайте фрагменты пакетами по 100–500, чтобы сбалансировать пропускную способность и использование памяти. Отслеживайте текущую позицию, чтобы после сбоя продолжить работу, не создавая заново векторные представления уже обработанных фрагментов. Регулярно записывайте ход выполнения в журнал. Для 100 000 фрагментов при размере пакета 500 потребуется 200 вызовов API — обычно это занимает несколько минут.
def embed_all_chunks(chunks, batch_size=200):
embedded = []
total = len(chunks)
for i in range(0, total, batch_size):
batch = chunks[i:i+batch_size]
texts = [c['text'] for c in batch]
vectors = embed_batch(texts)
for chunk, vector in zip(batch, vectors):
embedded.append({
**chunk,
'embedding': vector
})
if (i // batch_size) % 10 == 0:
print(f'Progress: {min(i+batch_size, total)}/{total}')
return embeddedОбработка ограничений частоты при индексировании
API векторных представлений OpenAI имеет ограничения частоты, измеряемые в токенах в минуту (TPM). При больших задачах индексирования эти ограничения достигаются, и возникает RateLimitError. Реализуйте экспоненциальную задержку с добавлением случайного разброса: при возникновении ошибки ограничения частоты подождите небольшой случайный промежуток времени перед повторной попыткой, увеличивая время ожидания вдвое после каждой последующей ошибки. Это распределяет повторные попытки во времени и не позволяет всем параллельным рабочим процессам одновременно перегружать API.
import time
import random
from openai import RateLimitError
def embed_batch_with_retry(texts, max_retries=5):
for attempt in range(max_retries):
try:
return embed_batch(texts)
except RateLimitError:
if attempt == max_retries - 1:
raise
wait = (2 ** attempt) + random.uniform(0, 1)
print(f'Rate limited. Waiting {wait:.1f}s...')
time.sleep(wait)
return []Добавление векторов в Pinecone
После создания эмбеддингов загрузите их в векторное хранилище с помощью upsert. Upsert означает добавление новых векторов или обновление существующих, если такой же ID уже есть, — по своей сути эта операция идемпотентна. В Pinecone каждая добавляемая запись содержит ID вектора, значения эмбеддинга и словарь метаданных с полями, по которым Вы хотите фильтровать или которые хотите отображать позже. Для оптимальной пропускной способности выполняйте upsert пакетами не более чем по 100 записей за вызов.
import pinecone
pc = pinecone.Pinecone(api_key='YOUR_KEY')
index = pc.Index('rag-index')
def upsert_to_pinecone(embedded_chunks, batch_size=100):
for i in range(0, len(embedded_chunks), batch_size):
batch = embedded_chunks[i:i+batch_size]
vectors = [
(
chunk['id'],
chunk['embedding'],
{
'text': chunk['text'],
'source': chunk['metadata']['source'],
'page': chunk['metadata'].get('page', 0)
}
)
for chunk in batch
]
index.upsert(vectors=vectors)
print(f'Upserted {min(i+batch_size, len(embedded_chunks))}/{len(embedded_chunks)}')Хранение в pgvector
В pgvector эмбеддинги вставляются непосредственно в таблицу PostgreSQL с помощью стандартного SQL. Тип данных vector принимает список чисел с плавающей точкой Python, сериализованный в виде строки. После вставки всех строк создайте индекс HNSW для быстрых приближённых запросов ближайших соседей. Индексация существующей таблицы с миллионами строк может занять несколько минут, поэтому создавайте индекс после массовой вставки, а не до неё.
import psycopg2
from psycopg2.extras import execute_values
def upsert_to_pgvector(conn, embedded_chunks):
with conn.cursor() as cur:
records = [
(
chunk['id'],
chunk['text'],
chunk['metadata']['source'],
chunk['metadata'].get('page', 0),
chunk['embedding'] # list of floats
)
for chunk in embedded_chunks
]
execute_values(cur, '''
INSERT INTO document_chunks (id, text, source, page, embedding)
VALUES %s
ON CONFLICT (id) DO UPDATE
SET text = EXCLUDED.text, embedding = EXCLUDED.embedding
''', records)
conn.commit()Создание индекса HNSW
HNSW (Hierarchical Navigable Small World) — это тип индекса, обеспечивающий быстрый приближённый поиск ближайших соседей. В отличие от поиска полным перебором, при котором вектор запроса сравнивается с каждым сохранённым вектором, HNSW строит многоуровневую графовую структуру, сокращающую пространство поиска. Параметр m задаёт количество связей каждого узла: большее значение повышает полноту поиска, но требует больше памяти. Параметр ef_construction определяет качество индекса во время его построения.
-- Build HNSW index after bulk insertion
CREATE INDEX CONCURRENTLY ON document_chunks
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);
-- Set ef_search at query time to trade recall vs speed
SET hnsw.ef_search = 100;
-- Verify index was created
SELECT indexname, indexdef
FROM pg_indexes
WHERE tablename = 'document_chunks';Проектирование схемы метаданных
Метаданные, хранящиеся вместе с каждым вектором, обеспечивают мощный поиск с фильтрацией. Спроектируйте схему метаданных до индексации: добавление новых полей позже потребует переиндексации. Включите поля, по которым Вы будете фильтровать (подразделение, тип документа, диапазон дат), поля, которые будете отображать в цитатах (название, страница, автор), и поля, полезные для отладки (индекс фрагмента, общее количество фрагментов, время индексации). Используйте простые значения метаданных: строки, числа и логические значения эффективно индексируются и фильтруются, а вложенные объекты — нет.
# Well-designed metadata schema
METADATA_SCHEMA = {
# For filtering at retrieval time
'department': 'HR', # string
'doc_type': 'policy', # string
'year': 2025, # integer
'is_active': True, # boolean
# For display in citations
'title': 'Employee Handbook 2025',
'author': 'HR Team',
'page': 12,
'source': 's3://docs/handbook_2025.pdf',
# For debugging and updates
'chunk_index': 3,
'total_chunks': 24,
'indexed_at': '2025-09-01T10:00:00Z'
}Контрольные точки для длительных заданий индексации
Индексация большого корпуса может занимать несколько часов. Сбой в середине работы приводит к потере всего прогресса. Реализуйте файл контрольной точки, в котором записывается, какие фрагменты успешно проиндексированы. После перезапуска пропускайте уже проиндексированные фрагменты и продолжайте с того места, на котором остановились. Благодаря этому задание индексации становится идемпотентным и его безопасно возобновлять. Храните контрольную точку как множество обработанных ID фрагментов в JSON-файле или таблице базы данных.
import json
from pathlib import Path
CHECKPOINT_FILE = '/tmp/index_checkpoint.json'
def load_checkpoint():
if Path(CHECKPOINT_FILE).exists():
return set(json.loads(Path(CHECKPOINT_FILE).read_text()))
return set()
def save_checkpoint(indexed_ids):
Path(CHECKPOINT_FILE).write_text(json.dumps(list(indexed_ids)))
def index_with_checkpoint(chunks, index):
done = load_checkpoint()
remaining = [c for c in chunks if c['id'] not in done]
print(f'Resuming: {len(done)} done, {len(remaining)} remaining')
for chunk in remaining:
upsert_to_pinecone([chunk], index)
done.add(chunk['id'])
save_checkpoint(done)Проверка полноты индекса
После индексации убедитесь, что все фрагменты попали в векторное хранилище. Сравните количество фрагментов, созданных средством разделения, с количеством векторов, о котором сообщает индекс. Выполните запрос к индексу с текстом известного документа и убедитесь, что ожидаемый результат входит в первые 5. Выполните несколько известных запросов из эталонного набора тестов и проверьте, что точность находится на ожидаемом уровне. Никогда не считайте индекс полным без проверки.
def verify_index(index, chunks, sample_size=10):
index_stats = index.describe_index_stats()
total_vectors = index_stats.total_vector_count
expected = len(chunks)
print(f'Index vectors: {total_vectors}, Expected: {expected}')
if total_vectors != expected:
print('WARNING: mismatch — some chunks may not have been indexed')
# Spot-check retrieval
import random
sample = random.sample(chunks, sample_size)
for chunk in sample:
vec = embed_batch([chunk['text']])[0]
results = index.query(vector=vec, top_k=1, include_metadata=True)
top_id = results.matches[0].id if results.matches else None
if top_id != chunk['id']:
print(f'WARNING: expected {chunk["id"]}, got {top_id}')Локальные альтернативы для создания эмбеддингов
Для данных, требующих конфиденциальности и не покидающих Вашу инфраструктуру, используйте локально размещённые модели эмбеддингов. Библиотека sentence-transformers предоставляет высококачественные модели, такие как all-MiniLM-L6-v2 (384 измерения, 22 МБ, очень высокая скорость) и bge-large-en-v1.5 (1024 измерения, более высокое качество). Запускайте их на CPU при умеренной нагрузке или на GPU при больших заданиях индексации. Локальные модели устраняют расходы на API и передачу данных за пределы инфраструктуры, но требуют управления файлами моделей и вычислительными ресурсами.
from sentence_transformers import SentenceTransformer
# Load once at startup
model = SentenceTransformer('sentence-transformers/all-MiniLM-L6-v2')
def embed_locally(texts, batch_size=64):
# encode() handles batching internally
embeddings = model.encode(
texts,
batch_size=batch_size,
show_progress_bar=True,
convert_to_numpy=True
)
return embeddings.tolist() # convert numpy array to Python list
vectors = embed_locally([c['text'] for c in chunks])Быстрая проверка
Проверьте, насколько хорошо Вы усвоили концепции AI Engineering из этого урока.
Итоги урока
В этом уроке Вы узнали, как создавать эмбеддинги пакетами с помощью API OpenAI и обрабатывать ограничения частоты запросов с помощью экспоненциальной задержки, добавлять векторы через upsert в Pinecone и pgvector вместе с метаданными, создавать индексы HNSW для быстрого приближённого поиска ближайших соседей, а также применять рекомендации для промышленной эксплуатации, включая возобновление работы по контрольным точкам, проектирование схемы метаданных и проверку полноты индекса. Далее мы создадим конвейер запросов, который извлекает фрагменты и генерирует ответы, основанные на источниках.
Часто задаваемые вопросы
Урок «Индексирование: создание и сохранение фрагментов» бесплатный?
Да — полный текст урока «Индексирование: создание и сохранение фрагментов» бесплатно доступен здесь в веб-версии. Чтобы практиковать его интерактивно (встроенный редактор кода и ИИ-репетитор 24/7) и разблокировать остальной курс AI Engineering Academy, подпишись на CoddyKit PRO. Курс AI Engineering Academy содержит 4 уроков всего.
Чему я научусь в уроке «Индексирование: создание и сохранение фрагментов»?
Создайте эмбеддинг для каждого фрагмента с помощью API эмбеддингов OpenAI и добавьте полученные векторы с метаданными в векторное хранилище, построив индекс документов для поиска. Ты практикуешь AI Engineering Academy с помощью реального кода, который запускаешь прямо в браузере, и ИИ-репетитор 24/7 отвечает на твои вопросы во время урока.
Нужен ли мне опыт, чтобы начать AI Engineering Academy?
Предыдущий опыт не требуется. AI Engineering Academy на CoddyKit структурирован для всех уровней — от новичков до продвинутых, поэтому ты можешь начать отсюда или с самого начала и учиться в своем темпе. Это урок 3 из 4.
Сколько времени занимает урок «Индексирование: создание и сохранение фрагментов»?
Большинство уроков CoddyKit занимают около 5–10 минут. Каждый из них компактный и интерактивный, поэтому ты постоянно делаешь прогресс и продолжаешь с того же места в веб-версии и приложении.
Можно ли писать и запускать код в этом уроке AI Engineering Academy?
Да. Каждый урок AI Engineering Academy включает встроенный редактор кода, поэтому ты пишешь и запускаешь реальный код прямо в браузере и получаешь моментальную обратную связь от AI — локальная установка не требуется.
Все уроки этого курса
- Загрузка документов и извлечение текста
- Стратегии разбиения: фиксированные фрагменты, предложения и рекурсия
- Индексирование: создание и сохранение фрагментов
- Запрос, поиск и генерация