Indeksowanie: tworzenie embeddingów i przechowywanie fragmentów
Uczestnicy utworzą embedding każdego fragmentu za pomocą API OpenAI embeddings, wykonają upsert wynikowych wektorów z metadanymi do magazynu wektorowego oraz zbudują przeszukiwalny indeks dokumentów.
Indeksowanie: tworzenie embeddingów i przechowywanie fragmentów to bezpłatna lekcja AI Engineering Academy na CoddyKit. To lekcja 3 z 4. Możesz przeczytać całą lekcję poniżej za darmo — a potem ćwiczyć ją interaktywnie w przeglądarce z wbudowanym edytorem kodu i tutorem AI dostępnym 24/7. To część ścieżki edukacyjnej AI Engineering Academy, a Twój postęp synchronizuje się między webem a aplikacją CoddyKit. Kurs AI Engineering Academy zawiera 4 lekcji w sumie.
Etap indeksowania: przegląd
Po wczytaniu i podzieleniu dokumentów następuje etap indeksowania: konwersja fragmentów tekstu na osadzenia wektorowe i zapisanie ich w przeszukiwalnej bazie wektorowej. To ostatni krok offline przed rozpoczęciem obsługi zapytań. Jakość osadzeń oraz efektywność strategii przechowywania i indeksowania bezpośrednio decydują o szybkości i dokładności systemu RAG podczas obsługi zapytań.
Generowanie osadzeń za pomocą OpenAI API
Najczęściej stosowane podejście polega na wywołaniu API osadzeń OpenAI z tekstem fragmentu. Model text-embedding-3-small tworzy wektory o 1536 wymiarach i kosztuje 0,02 USD za milion tokenów — jest to niezwykle tanie w przypadku większości obciążeń. Należy wysyłać wiele tekstów w ramach jednego wywołania API (do 2048 danych wejściowych), aby zmaksymalizować przepustowość. Odpowiedź zawiera po jednym wektorze osadzenia dla każdego tekstu wejściowego, w tej samej kolejności.
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')Przetwarzanie partiami na potrzeby wydajności
Podczas indeksowania tysięcy fragmentów wydajność ma znaczenie. Należy przetwarzać fragmenty w partiach po 100–500, aby zachować równowagę między przepustowością a zużyciem pamięci. Należy śledzić pozycję, aby po awarii można było wznowić pracę bez ponownego tworzenia osadzeń dla już przetworzonych fragmentów. Postęp należy regularnie rejestrować. W przypadku 100 000 fragmentów i 500 fragmentów na partię zostanie wykonanych 200 wywołań API — zwykle kończy się to w ciągu kilku minut.
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 embeddedObsługa limitów szybkości podczas indeksowania
API osadzeń OpenAI ma limity szybkości mierzone w tokenach na minutę (TPM). Duże zadania indeksowania osiągają te limity i otrzymują błąd RateLimitError. Należy zaimplementować wykładnicze wycofywanie z losowym opóźnieniem: po wystąpieniu błędu limitu szybkości należy odczekać krótki, losowy czas przed ponowieniem próby, podwajając czas oczekiwania po każdej kolejnej porażce. Rozkłada to ponowienia w czasie i zapobiega jednoczesnemu przeciążaniu API przez wszystkich równoległych workerów.
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 []Wstawianie wektorów do Pinecone
Po wygenerowaniu embeddingów należy wstawić je do magazynu wektorowego za pomocą operacji upsert. Upsertowanie oznacza wstawianie nowych wektorów lub aktualizowanie istniejących, jeśli istnieje już identyfikator o tej samej wartości — z założenia jest to operacja idempotentna. W Pinecone każdy rekord przesłany za pomocą upsert zawiera identyfikator wektora, wartości embeddingu oraz słownik metadanych z polami, według których można później filtrować lub które można wyświetlać. Aby uzyskać optymalną przepustowość, należy przesyłać partie zawierające maksymalnie 100 rekordów w jednym wywołaniu.
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)}')Przechowywanie w pgvector
W przypadku pgvector embeddingi są wstawiane bezpośrednio do tabeli PostgreSQL za pomocą standardowego języka SQL. Typ danych vector akceptuje listę liczb zmiennoprzecinkowych w Pythonie, zserializowaną jako ciąg znaków. Po wstawieniu wszystkich wierszy należy utworzyć indeks HNSW, aby przyspieszyć wyszukiwanie przybliżonych najbliższych sąsiadów. Indeksowanie istniejącej tabeli zawierającej miliony wierszy może potrwać kilka minut, dlatego indeks należy budować po masowym wstawieniu danych, a nie przed nim.
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()Budowanie indeksu HNSW
HNSW (Hierarchical Navigable Small World) to typ indeksu umożliwiający szybkie wyszukiwanie przybliżonych najbliższych sąsiadów. W przeciwieństwie do wyszukiwania metodą brute force, które porównuje wektor zapytania z każdym zapisanym wektorem, HNSW tworzy wielowarstwową strukturę grafową ograniczającą obszar wyszukiwania. Parametr m określa liczbę połączeń każdego węzła (większa wartość oznacza lepszy recall, ale większe zużycie pamięci), a ef_construction określa jakość indeksu podczas jego budowania.
-- 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';Projektowanie schematu metadanych
Metadane przechowywane obok każdego wektora umożliwiają zaawansowane filtrowane wyszukiwanie. Schemat metadanych należy zaprojektować przed indeksowaniem — późniejsze dodanie nowych pól wymaga ponownego indeksowania. Należy uwzględnić pola, według których będzie odbywać się filtrowanie (dział, typ dokumentu, zakres dat), pola wyświetlane w cytowaniach (tytuł, strona, autor) oraz pola przydatne podczas debugowania (chunk_index, total_chunks, indexed_at). Wartości metadanych powinny być proste: ciągi znaków, liczby i wartości logiczne są efektywnie indeksowane i filtrowane, natomiast obiekty zagnieżdżone — nie.
# 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'
}Zapisywanie punktów kontrolnych długotrwałych zadań indeksowania
Indeksowanie dużego korpusu może trwać wiele godzin. Awaria w połowie pracy powoduje utratę całego postępu. Należy zaimplementować plik punktu kontrolnego, który zapisuje informacje o pomyślnie zindeksowanych fragmentach. Po ponownym uruchomieniu trzeba pominąć fragmenty już zindeksowane i kontynuować pracę od miejsca przerwania. Dzięki temu zadanie indeksowania jest idempotentne i można je bezpiecznie wznowić. Punkt kontrolny należy przechowywać jako zbiór przetworzonych identyfikatorów fragmentów w pliku JSON lub tabeli bazy danych.
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)Weryfikowanie kompletności indeksu
Po zakończeniu indeksowania należy sprawdzić, czy wszystkie fragmenty trafiły do magazynu wektorowego. Należy porównać liczbę fragmentów utworzonych przez dzielnik z liczbą wektorów raportowaną przez indeks. Wykonaj zapytanie do indeksu, używając tekstu znanego dokumentu, i potwierdź, że oczekiwany wynik znajduje się w pierwszej piątce. Uruchom kilka znanych zapytań ze zbioru testów wzorcowych i sprawdź, czy precyzja osiąga oczekiwany poziom. Nigdy nie zakładaj, że indeks jest kompletny bez jego weryfikacji.
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}')Lokalne alternatywy dla embeddingów
W przypadku danych wrażliwych pod względem prywatności, które nie mogą opuścić własnej infrastruktury, należy użyć lokalnie hostowanych modeli embeddingowych. Biblioteka sentence-transformers udostępnia wysokiej jakości modele, takie jak all-MiniLM-L6-v2 (384 wymiarów, 22 MB, bardzo szybki) oraz bge-large-en-v1.5 (1024 wymiary, lepsza jakość). Modele można uruchamiać na procesorze CPU przy umiarkowanym obciążeniu lub na procesorze GPU podczas dużych zadań indeksowania. Modele lokalne eliminują koszty API i transferu danych na zewnątrz, ale wymagają zarządzania plikami modeli oraz zasobami obliczeniowymi.
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])Szybki sprawdzian
Sprawdź swoją wiedzę na temat koncepcji inżynierii AI omówionych w tej lekcji.
Podsumowanie lekcji
W tej lekcji poznano: generowanie embeddingów partiami za pomocą API OpenAI oraz obsługę limitów szybkości z wykorzystaniem wykładniczego wycofywania, upsertowanie wektorów w Pinecone i pgvector wraz z metadanymi, budowanie indeksów HNSW na potrzeby szybkiego przybliżonego wyszukiwania najbliższych sąsiadów oraz najlepsze praktyki produkcyjne, w tym wznawianie pracy na podstawie punktów kontrolnych, projektowanie schematu metadanych i weryfikowanie kompletności indeksu. Następnie zbudujemy potok zapytań, który pobiera fragmenty i generuje odpowiedzi oparte na źródłach.
Często zadawane pytania
Czy lekcja „Indeksowanie: tworzenie embeddingów i przechowywanie fragmentów” jest bezpłatna?
Tak — pełny tekst „Indeksowanie: tworzenie embeddingów i przechowywanie fragmentów” jest dostępny za darmo tutaj w sieci. Aby ćwiczyć ją interaktywnie (wbudowany edytor kodu i tutor AI dostępny 24/7) i odblokować resztę kursu AI Engineering Academy, przejdź na CoddyKit PRO. Kurs AI Engineering Academy zawiera 4 lekcji w sumie.
Co nauczysz się w „Indeksowanie: tworzenie embeddingów i przechowywanie fragmentów”?
Uczestnicy utworzą embedding każdego fragmentu za pomocą API OpenAI embeddings, wykonają upsert wynikowych wektorów z metadanymi do magazynu wektorowego oraz zbudują przeszukiwalny indeks dokumentów. Ćwiczysz AI Engineering Academy z praktycznym kodem, który uruchamiasz bezpośrednio w przeglądarce, a tutor AI dostępny 24/7 odpowiada na Twoje pytania podczas pracy nad lekcją.
Czy potrzebuję doświadczenia, aby zacząć AI Engineering Academy?
Nie wymagamy żadnego doświadczenia. AI Engineering Academy w CoddyKit jest strukturyzowany dla początkujących i zaawansowanych użytkowników, więc możesz zacząć tutaj lub od początku i uczyć się w swoim tempie. To lekcja 3 z 4.
Ile czasu zajmuje lekcja „Indeksowanie: tworzenie embeddingów i przechowywanie fragmentów”?
Większość lekcji CoddyKit trwa około 5–10 minut. Każda lekcja to mały, interaktywny krok, dzięki czemu robisz systematyczne postępy i zawsze wracasz dokładnie do tego samego miejsca — na webie i w aplikacji.
Czy mogę pisać i uruchamiać kod w tej lekcji AI Engineering Academy?
Tak. Każda lekcja AI Engineering Academy zawiera wbudowany edytor kodu, więc piszesz i uruchamiasz prawdziwy kod bezpośrednio w przeglądarce i od razu otrzymujesz sprzężenie zwrotne od AI — bez konfiguracji na komputerze.
Wszystkie lekcje w tym kursie
- Wczytywanie dokumentów i ekstrakcja tekstu
- Strategie dzielenia tekstu: stały rozmiar, zdania i rekurencja
- Indeksowanie: tworzenie embeddingów i przechowywanie fragmentów
- Zapytanie, pobieranie i generowanie