W dzisiejszych czasach nie wystarczy nam wiedza wbudowana w modele AI. Musimy często sięgnąć po nowe informacje/bazy wiedzy/specjalistyczne ebooki, w celu jej uzupełnienia. Ponowny trening samych modeli, przekracza nasze domowe możliwości mocy obliczeniowej (wymaga potężnych centrów danych) lub jest totalnie nie opłacalny. Każdy model który został wytrenowany, posiada wiedzę aktualną do czasu rozpoczęcia jego treningu.
Aby móc efektycznie przeszukiwać nasze dane i dostarczać je jako kontekst dla naszego zapytania, nie wystarczy wkleić całą naszą baze wiedzy. Jest to podejście bardzo nie efektywne, ze względu na to że:
- same modele mają ograniczone okno kontekstowe
- płacimy za każdy token
- w przypadku przesłania zbyt dużego kontektu, modele potrafią zgubić to o co naprawdę je zapytaliśmy
W naszym przykładzie wykorzystamy bibliotekę ChromaDB jako bazę danych, która przeszuka i zwróci nam tylko zawężony zakres informacji, podobnych (czyli zbliżonych kontekstowo) do naszego zapytania. Dzięki temu dostaniemy zamiast np tekstu na 200-stron, tylko 3 akapity które będą zawierać odpowiedź.
A teraz do dzieła :)
Przygotowanie środowiska
Zacznijmy od przygotowania środowiska na którym będziemy pracować:
1. Utwórzmy nasze środowisko wirtualne:
python -m venv .venv
source .venv/bin/activate
2. Następnie tworzymy plik z wymaganymi zależnościami:
requirements.txt
openai
langchain
faiss-cpu
chromadb
openai
tiktoken
python-dotenv
sentence-transformers
requests
bs4
rich
gradio
ruff
Które następnie zainstalujemy po przez:
pip install -r requirements.txt*Każdy pythonowy projekt, powinien opierać się na wirtualnych środowiskach. Pozwoli nam to uniknąć konfliktów pomiędzy pakietami systemowymi a wymaganymi dla danej aplikacji.
Baza wektorowa
Aby można było wyszukiwać z powodzeniem w naszej bazie powiązanych zagadnień. Należy zamienić nasze słowa, na wektory (embedingi), aby następnie móc określić, które z nich są do siebie semantycznie podobne.
Musimy użyć do tego dedykowanej bazy danych (wektorowej). Najprostszym, a zarazem bardzo mocno rozwijanym rozwiązaniem, jest użycie silnika baz danych o nazwie ChromaDb. Wykorzystamy go w naszym przykładzie.
Sama inicjalizacja naszej bazy wygląda następująco:
client = PersistentClient(path="chroma-db")
collection = client.get_or_create_collection(name="documents")
Flow naszej aplikacji
Teraz przejdźmy do naszego flow aplikacji. Chcemy:
- Pobrać nasze dokumenty (np w formacie json).
- Podzielić nasze dane w mniejsze porcje. Dzięki temu unikniemy przesyłania np 200 stronicowego dokumentu jako kontekst do zapytania, a prześlemy tylko te informacje które są użyteczne.
- Wygenerować dla nich embedingi (wektorowe postacie) z użyciem dedykowanego modelu np. all-mpnet-base-v2 (lokalnie) lub text-embedding-3-small (z użyciem OpenAI).
1. Import danych
Na samym początku, musimy wygenerować dane na których będziemy pracować. Utwórzmy więc funkcję pomocniczą zwracającą listę naszych dokumentów w formacie json.
def load_sample_data():
json_data = """
[
{"id":"doc1", "doc":"Chroma to silnik dla bazy wektorowej.", "meta":{"source":"notatki"}},
{"id":"doc2", "doc":"OpenAI embeddings pozwalają na zamianę tekstu na wektory.", "meta":{"source":"blog"}},
{"id":"doc3", "doc":"LangChain ułatwia tworzenie aplikacji opartych na LLM-ach.", "meta":{"source":"dokumentacja"}},
{"id":"doc4", "doc":"Bazy wektorowe pozwalają na efektywne przeszukiwanie dużych zbiorów danych tekstowych.", "meta":{"source":"artykuł"}},
{"id":"doc5", "doc":"RAG łączy generowanie tekstu z odzyskiwaniem informacji w czasie rzeczywistym.", "meta":{"source":"prezentacja"}}
]
"""
return json_data
*W następnych artykułach rozbudujemy ją od pełny import stron www :) (w tym przypadku również będziemy musieli oczyścić dane/przekonwertować je do markdown, aby pozbyć się zbędnych meta danych lub informacji o np css-ach).
2. Chunkowanie - dzielenie na mniejsze porcje
Zostaje nam tylko stworzenia własnego systemu dzielenia tekstu. Możemy do tego wykorzystać poniższą metodę która, poza tym że dzieli tekst na mniejsze kawałki, zachowuje również część informacji z poprzedniego jak i następnego fragmentu.
from typing import List
def chunk_text(text: str, chunk_size: int = 500, overlap: int = 50) -> List[str]:
if chunk_size <= 0:
raise ValueError("chunk_size must be > 0")
if overlap < 0 or overlap >= chunk_size:
raise ValueError("overlap must be >= 0 and smaller than chunk_size")
chunks: List[str] = []
step = chunk_size - overlap
start = 0
while start < len(text):
end = start + chunk_size
if end >= len(text):
chunk = text[-chunk_size:]
chunks.append(chunk)
break
chunks.append(text[start:end])
start += step
return chunks
Poniżej możesz zobaczyć jak wygląda jej przykładowe wywołanie na podanym ciągu znaków:
text = "1234567890ABCDEFGHIJ"
print(chunk_text(text, 6, 1)),
# Zwraca: ['123456', '67890A', 'ABCDEF', 'EFGHIJ']
print(chunk_text(text, 6, 2))
# Zwraca: ['123456', '567890', '90ABCD', 'CDEFGH', 'EFGHIJ']
3. Embeding - tworzenie wektorowej postaci danych
Z racji że chcemy przechowywać reprezentację wektorową naszych danych, potrzebujemy modelu który przekonwertuje konkretne słowa, na ich liczbowe reprezentacje.
Poniżej przestawiam kod loadera, który pozwoli wybrać nam jakim modelem chcemy je generować. Zwróci nam metodę go wywołującą, jako zwykłą funkcję.
from typing import List
from openai import OpenAI
from sentence_transformers import SentenceTransformer
from dotenv import load_dotenv
import os
load_dotenv()
def get_openai_embeddings(texts: List[str], model_name: str) -> List[List[float]]:
openai_client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
response = openai_client.embeddings.create(
input=texts,
model=model_name
)
return [r.embedding for r in response.data]
def get_local_embeddings(texts: List[str], model_name: str) -> List[List[float]]:
model = SentenceTransformer(model_name)
return model.encode(texts, show_progress_bar=False).tolist()
def embeder_loader(provider: str, model: str) -> List[List[float]]:
if provider == "openai":
def embeder(texts):
return get_openai_embeddings(texts, model_name = model)
else:
def embeder(texts):
return get_local_embeddings(texts, model_name = model)
return embeder
Samo załadowanie funkcji opiera się na wywołaniu:
embed_fn = embeder_loader("local", "sdadas/st-polish-paraphrase-from-distilroberta")
lub
embed_fn = embeder_loader("openai", "text-embedding-3-small")
4. Zapis danych w bazie wektorowej
Stworzymy teraz funkcję, która będzie iterować po każdym wierszu naszego json-a. Dla każdego naszego "row" dokonamy podzielenia na mniejsze porcje (patrz pkt 2), wygenerowania dla nich wartości wektorowych (patrz pkt 3) oraz załadowania do naszej bazy tylko brakujących dokumentów.
import json
from chromadb import PersistentClient
from typing import Callable, List
def import_data(chroma_collection: PersistentClient, json_data: json, embed_fn: Callable[[List[str]], List[List[float]]] = None, chunk_size: int = 500, overlap: int = 50):
items = json.loads(json_data) if isinstance(json_data, str) else json_data
for item in items:
doc_id = item["id"]
doc_text = item["doc"]
meta = item.get("meta", {})
chunks = chunk_text(doc_text, chunk_size, overlap)
for i, chunk in enumerate(chunks):
chunk_id = f"{doc_id}_chunk{i}"
if not chroma_collection.get(ids=[chunk_id]).get("ids"):
chroma_collection.add(
ids=[chunk_id],
documents=[chunk],
embeddings=embed_fn([chunk])[0] if embed_fn else None,
metadatas=[meta]
)
5. Czas na uruchomienie importu
Poniżej przedstawiam pełne wywołanie naszego programu:
# pkt 1
client = PersistentClient(path="chroma-db")
collection = client.get_or_create_collection(name="documents")
# pkt 2
documents = load_sample_data()
# pkt 3
embed_fn = embeder_loader("openai", "text-embedding-3-small")
chunk_size = 100
overlap = 10
# pkt 4
import_data(
collection,
documents,
embed_fn,
chunk_size,
overlap
)
Pełny kod:
import json
import os
from typing import Callable, List, Union
from dotenv import load_dotenv
from chromadb import PersistentClient
from openai import OpenAI
from sentence_transformers import SentenceTransformer
load_dotenv()
def load_sample_data() -> str:
"""
Ładuje przykładowe dane w formacie JSON.
Returns:
str: JSON zawierający listę dokumentów z ID, tekstem i metadanymi.
"""
json_data = """
[
{"id":"doc1", "doc":"Chroma to silnik dla bazy wektorowej.", "meta":{"source":"notatki"}},
{"id":"doc2", "doc":"OpenAI embeddings pozwalają na zamianę tekstu na wektory.", "meta":{"source":"blog"}},
{"id":"doc3", "doc":"LangChain ułatwia tworzenie aplikacji opartych na LLM-ach.", "meta":{"source":"dokumentacja"}},
{"id":"doc4", "doc":"Bazy wektorowe pozwalają na efektywne przeszukiwanie dużych zbiorów danych tekstowych.", "meta":{"source":"artykuł"}},
{"id":"doc5", "doc":"RAG łączy generowanie tekstu z odzyskiwaniem informacji w czasie rzeczywistym.", "meta":{"source":"prezentacja"}}
]
"""
return json_data
def chunk_text(text: str, chunk_size: int = 500, overlap: int = 50) -> List[str]:
"""
Dzieli tekst na mniejsze fragmenty z określonym rozmiarem i nakładką.
Args:
text (str): Tekst do podziału.
chunk_size (int): Maksymalna długość pojedynczego fragmentu.
overlap (int): Liczba znaków nakładających się między fragmentami.
Returns:
List[str]: Lista fragmentów tekstu.
Raises:
ValueError: Jeśli chunk_size <= 0 lub overlap >= chunk_size.
"""
if chunk_size <= 0:
raise ValueError("chunk_size must be > 0")
if overlap < 0 or overlap >= chunk_size:
raise ValueError("overlap must be >= 0 and smaller than chunk_size")
chunks: List[str] = []
step = chunk_size - overlap
start = 0
while start < len(text):
end = start + chunk_size
if end >= len(text):
chunk = text[-chunk_size:]
chunks.append(chunk)
break
chunks.append(text[start:end])
start += step
return chunks
def get_openai_embeddings(texts: List[str], model_name: str) -> List[List[float]]:
"""
Pobiera embeddingi dla listy tekstów przy użyciu OpenAI API.
Args:
texts (List[str]): Lista tekstów.
model_name (str): Nazwa modelu embeddingów OpenAI.
Returns:
List[List[float]]: Lista embeddingów dla każdego tekstu.
"""
def get_openai_embeddings(texts: List[str], model_name: str) -> List[List[float]]:
"""
Pobiera embeddingi dla listy tekstów przy użyciu OpenAI API.
Args:
texts (List[str]): Lista tekstów.
model_name (str): Nazwa modelu embeddingów OpenAI.
Returns:
List[List[float]]: Lista embeddingów dla każdego tekstu.
"""
try:
openai_client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
response = openai_client.embeddings.create(input=texts, model=model_name)
return [r.embedding for r in response.data]
except ValueError as e:
print(f"ValueError przy pobieraniu embeddingów OpenAI: {e}")
except TypeError as e:
print(f"TypeError przy pobieraniu embeddingów OpenAI: {e}")
except ConnectionError as e:
print(f"Błąd połączenia z OpenAI: {e}")
except TimeoutError as e:
print(f"Timeout przy połączeniu z OpenAI: {e}")
except Exception as e:
print(f"Inny błąd przy pobieraniu embeddingów OpenAI: {e}")
return []
def get_local_embeddings(texts: List[str], model_name: str) -> List[List[float]]:
"""
Pobiera embeddingi dla listy tekstów przy użyciu lokalnego modelu SentenceTransformer.
Args:
texts (List[str]): Lista tekstów.
model_name (str): Nazwa modelu SentenceTransformer.
Returns:
List[List[float]]: Lista embeddingów dla każdego tekstu.
"""
try:
model = SentenceTransformer(model_name)
return model.encode(texts, show_progress_bar=False).tolist()
except Exception as e:
print(f"Błąd przy pobieraniu embeddingów lokalnych: {e}")
return []
def embeder_loader(provider: str, model: str) -> Callable[[List[str]], List[List[float]]]:
"""
Tworzy funkcję embedingu na podstawie wybranego providera.
Args:
provider (str): "openai" lub "local".
model (str): Nazwa modelu do użycia.
Returns:
Callable[[List[str]], List[List[float]]]: Funkcja przyjmująca listę tekstów i zwracająca embeddingi.
"""
if provider == "openai":
def embeder(texts: List[str]) -> List[List[float]]:
return get_openai_embeddings(texts, model_name=model)
else:
def embeder(texts: List[str]) -> List[List[float]]:
return get_local_embeddings(texts, model_name=model)
return embeder
def import_data(
chroma_collection: PersistentClient,
json_data: Union[str, List[dict]],
embed_fn: Callable[[List[str]], List[List[float]]] = None,
chunk_size: int = 500,
overlap: int = 50,
):
"""
Importuje dane do kolekcji Chroma, dzieląc tekst na fragmenty i obliczając embeddingi.
Args:
chroma_collection (PersistentClient): Kolekcja Chroma do której dodajemy dokumenty.
json_data (Union[str, List[dict]]): Dane JSON jako string lub lista słowników.
embed_fn (Callable, optional): Funkcja zwracająca embeddingi. Defaults to None.
chunk_size (int, optional): Rozmiar fragmentów tekstu. Defaults to 500.
overlap (int, optional): Nakładka między fragmentami. Defaults to 50.
"""
try:
items = json.loads(json_data) if isinstance(json_data, str) else json_data
except json.JSONDecodeError as e:
print(f"Błąd parsowania JSON: {e}")
return
for item in items:
try:
doc_id = item["id"]
doc_text = item["doc"]
meta = item.get("meta", {})
chunks = chunk_text(doc_text, chunk_size, overlap)
for i, chunk in enumerate(chunks):
chunk_id = f"{doc_id}_chunk{i}"
try:
if not chroma_collection.get(ids=[chunk_id]).get("ids"):
chroma_collection.add(
ids=[chunk_id],
documents=[chunk],
embeddings=embed_fn([chunk])[0] if embed_fn else None,
metadatas=[meta],
)
print(f"Dodano chunk: {chunk_id}")
except Exception as e:
print(f"Błąd dodawania chunku {chunk_id}: {e}")
except KeyError as e:
print(f"Brakuje klucza w danych dokumentu: {e}")
except Exception as e:
print(f"Inny błąd przy przetwarzaniu dokumentu {item.get('id', 'unknown')}: {e}")
if __name__ == "__main__":
try:
# Inicjalizacja klienta bazy danych
client = PersistentClient(path="chroma-db")
collection = client.get_or_create_collection(name="documents")
except Exception as e:
print(f"Błąd inicjalizacji Chroma: {e}")
exit(1)
# Załaduj przykładowe dane
documents = load_sample_data()
# Wybór funkcji embeddingu
# embed_fn = embeder_loader("openai", "text-embedding-3-small")
embed_fn = embeder_loader("local", "all-MiniLM-L6-v2")
# Parametry dzielenia tekstu
chunk_size = 100
overlap = 10
# Import danych do kolekcji
import_data(collection, documents, embed_fn, chunk_size, overlap)
W następnej części zajmiemy się przeszukiwaniem baz wektorowych i przekazywaniem z nich kontektstu do naszego modelu LLM.
Komentarze