Работаем с Chroma — самой простой векторной базой данных. Фильтрация по метаданным, персистентность, клиент-серверный режим.
Работаем с Chroma — самой простой векторной базой данных. Фильтрация по метаданным, персистентность, клиент-серверный режим.
За 45 минут вы:
Chroma — векторная база данных с акцентом на простоту использования.
Преимущества:
Недостатки:
Когда использовать:
pip install chromadbПроверка:
import chromadb
print(chromadb.__version__)import chromadb
# Клиент в памяти (данные исчезнут после завершения)
client = chromadb.Client()
# Создание коллекции
collection = client.create_collection(
name="documents",
metadata={"description": "Моя первая коллекция"}
)
print(f"Коллекция создана: {collection.name}")Chroma может сама генерировать эмбеддинги:
# Добавление документов (Chroma сама создаст эмбеддинги)
collection.add(
documents=[
"Python — язык программирования общего назначения",
"Django — веб-фреймворк на Python",
"Flask — микрофреймворк для веб-приложений",
"FastAPI — современный фреймворк для API"
],
ids=["doc1", "doc2", "doc3", "doc4"],
metadatas=[
{"category": "programming", "author": "Alice"},
{"category": "web", "author": "Bob"},
{"category": "web", "author": "Charlie"},
{"category": "web", "author": "Alice"}
]
)
print(f"Добавлено документов: {collection.count()}")from sentence_transformers import SentenceTransformer
model = SentenceTransformer('all-MiniLM-L6-v2')
documents = [
"Python — язык программирования",
"Django — веб-фреймворк",
"Flask — микрофреймворк"
]
# Генерация эмбеддингов
embeddings = model.encode(documents).tolist()
# Добавление с готовыми эмбеддингами
collection.add(
embeddings=embeddings,
ids=["doc1", "doc2", "doc3"],
documents=documents,
metadatas=[
{"category": "programming"},
{"category": "web"},
{"category": "web"}
]
)# Chroma сама создаст эмбеддинг для запроса
results = collection.query(
query_texts=["веб-разработка на Python"],
n_results=2
)
print(f"Найдено документов: {len(results['documents'][0])}")
print(f"Тексты: {results['documents'][0]}")
print(f"ID: {results['ids'][0]}")
print(f"Сходство: {results['distances'][0]}")# Генерация эмбеддинга запроса
query_embedding = model.encode(["веб-разработка"]).tolist()
results = collection.query(
query_embeddings=[query_embedding],
n_results=2
)
print(f"Тексты: {results['documents'][0]}")# Возврат метаданных в результатах
results = collection.query(
query_texts=["веб-фреймворк"],
n_results=2,
include=["documents", "metadatas", "distances"]
)
for i, doc in enumerate(results['documents'][0]):
print(f"\nРезультат {i+1}:")
print(f" Текст: {doc}")
print(f" Метаданные: {results['metadatas'][0][i]}")
print(f" Расстояние: {results['distances'][0][i]:.3f}")# Поиск только в категории "web"
results = collection.query(
query_texts=["фреймворк"],
n_results=2,
where={"category": "web"}
)
print(f"Результаты в категории 'web': {results['documents'][0]}")# Операторы сравнения
results = collection.query(
query_texts=["фреймворк"],
n_results=2,
where={
"author": {"$ne": "Bob"} # Не Bob
}
)
# $ne — не равно
# $eq — равно
# $gt — больше
# $lt — меньше
# $gte — больше или равно
# $lte — меньше или равно# AND: категория "web" И автор "Alice"
results = collection.query(
query_texts=["фреймворк"],
n_results=2,
where={
"$and": [
{"category": "web"},
{"author": "Alice"}
]
}
)
# OR: категория "web" ИЛИ "api"
results = collection.query(
query_texts=["фреймворк"],
n_results=5,
where={
"$or": [
{"category": "web"},
{"category": "api"}
]
}
)| Оператор | Описание | Пример |
|---|---|---|
$eq | Равно | {"age": {"$eq": 25}} |
$ne | Не равно | {"status": {"$ne": "deleted"}} |
$gt | Больше | {"score": {"$gt": 0.7}} |
$gte | Больше или равно | {"rating": {"$gte": 4}} |
$lt | Меньше | {"price": {"$lt": 100}} |
$lte | Меньше или равно | {"count": {"$lte": 10}} |
$in | В списке | {"category": {"$in": ["web", "api"]}} |
$nin | Не в списке | {"status": {"$nin": ["draft"]}} |
$and | И | {"$and": [{"a": 1}, {"b": 2}]} |
$or | ИЛИ | {"$or": [{"a": 1}, {"b": 2}]} |
# Клиент с сохранением на диск
client = chromadb.PersistentClient(path="./chroma_data")
# Все коллекции сохраняются в папке ./chroma_data
collection = client.get_or_create_collection("documents")
# Добавление документов
collection.add(
documents=["документ 1", "документ 2"],
ids=["doc1", "doc2"]
)
# После завершения данные сохранятся# При следующем запуске
client = chromadb.PersistentClient(path="./chroma_data")
# Получение существующей коллекции
collection = client.get_collection("documents")
print(f"Документов в коллекции: {collection.count()}")# Запуск сервера Chroma
chroma run --path ./chroma_data --port 8000# HTTP клиент
client = chromadb.HttpClient(host="localhost", port=8000)
# Работа как обычно
collection = client.get_or_create_collection("documents")
collection.add(
documents=["документ"],
ids=["doc1"]
)import asyncio
import chromadb
async def main():
client = chromadb.AsyncHttpClient(host="localhost", port=8000)
collection = await client.get_or_create_collection("documents")
await collection.add(
documents=["документ"],
ids=["doc1"]
)
results = await collection.query(
query_texts=["запрос"],
n_results=2
)
await client.close()
return results
# Запуск
results = asyncio.run(main())# Создание
collection = client.create_collection("my_collection")
# Получение существующей
collection = client.get_collection("my_collection")
# Получение или создание
collection = client.get_or_create_collection("my_collection")collections = client.list_collections()
for col in collections:
print(f"Коллекция: {col.name}")
print(f" Документов: {col.count()}")
print(f" Метаданные: {col.metadata}")client.delete_collection("my_collection")# Удаление по ID
collection.delete(ids=["doc1", "doc2"])
# Удаление по фильтру
collection.delete(
where={"category": "temp"}
)# Обновление метаданных
collection.update(
ids=["doc1"],
metadatas=[{"category": "updated", "author": "New Author"}]
)
# Обновление документа
collection.update(
ids=["doc1"],
documents=["Обновлённый текст документа"]
)# chroma_search.py
import chromadb
from pathlib import Path
class ChromaSearch:
"""Поисковик на Chroma с метаданными."""
def __init__(self, persist_dir: str = None):
if persist_dir:
self.client = chromadb.PersistentClient(path=persist_dir)
else:
self.client = chromadb.Client()
self.collection = None
def create_collection(self, name: str):
"""Создать коллекцию."""
self.collection = self.client.create_collection(name)
return self.collection
def get_collection(self, name: str):
"""Получить коллекцию."""
self.collection = self.client.get_collection(name)
return self.collection
def add_documents(
self,
documents: list[str],
ids: list[str] = None,
metadatas: list[dict] = None
):
"""Добавить документы."""
if self.collection is None:
raise ValueError("Сначала создайте или получите коллекцию")
if ids is None:
ids = [f"doc{i}" for i in range(len(documents))]
self.collection.add(
documents=documents,
ids=ids,
metadatas=metadatas
)
def search(
self,
query: str,
k: int = 5,
filter: dict = None
) -> list[dict]:
"""Поиск с фильтрацией."""
results = self.collection.query(
query_texts=[query],
n_results=k,
where=filter,
include=["documents", "metadatas", "distances"]
)
formatted = []
for i in range(len(results['documents'][0])):
formatted.append({
"text": results['documents'][0][i],
"metadata": results['metadatas'][0][i],
"distance": results['distances'][0][i],
"id": results['ids'][0][i]
})
return formatted
def get_stats(self) -> dict:
"""Статистика коллекции."""
if self.collection is None:
return {"error": "Коллекция не выбрана"}
return {
"name": self.collection.name,
"count": self.collection.count(),
"metadata": self.collection.metadata
}
# Использование
if __name__ == "__main__":
search = ChromaSearch(persist_dir="./chroma_data")
# Создание коллекции
search.create_collection("tech_docs")
# Добавление документов
search.add_documents(
documents=[
"Python — язык программирования общего назначения",
"Django — веб-фреймворк на Python для быстрой разработки",
"Flask — микрофреймворк для создания веб-приложений",
"FastAPI — современный фреймворк для создания API",
"NumPy — библиотека для научных вычислений"
],
ids=["python", "django", "flask", "fastapi", "numpy"],
metadatas=[
{"category": "programming", "level": "beginner"},
{"category": "web", "level": "intermediate"},
{"category": "web", "level": "beginner"},
{"category": "web", "level": "intermediate"},
{"category": "data", "level": "advanced"}
]
)
# Поиск без фильтра
print("Поиск 'веб-фреймворк':")
results = search.search("веб-фреймворк", k=3)
for r in results:
print(f" [{r['distance']:.3f}] {r['text']} ({r['metadata']['category']})")
# Поиск с фильтром
print("\nПоиск 'фреймворк' только в категории 'web':")
results = search.search(
"фреймворк",
k=3,
filter={"category": "web"}
)
for r in results:
print(f" [{r['distance']:.3f}] {r['text']}")
# Статистика
print(f"\nСтатистика: {search.get_stats()}")# ❌ Ошибка: одинаковые ID
collection.add(
documents=["doc1", "doc2"],
ids=["same_id", "same_id"] # Ошибка!
)
# ✅ Уникальные ID
collection.add(
documents=["doc1", "doc2"],
ids=["doc1", "doc2"]
)# ❌ Ошибка: разная длина
collection.add(
documents=["doc1", "doc2"],
ids=["doc1"], # Только один ID!
metadatas=[{"a": 1}, {"b": 2}]
)
# ✅ Одинаковая длина
collection.add(
documents=["doc1", "doc2"],
ids=["doc1", "doc2"],
metadatas=[{"a": 1}, {"b": 2}]
)# ❌ Неправильно
collection.query(
query_texts=["запрос"],
where={"category" == "web"} # Python-оператор!
)
# ✅ Правильно
collection.query(
query_texts=["запрос"],
where={"category": "web"}
)# ❌ Метаданные не включены в результат
results = collection.query(
query_texts=["запрос"],
n_results=2
)
# results['metadatas'] будет пустым!
# ✅ Явно указать include
results = collection.query(
query_texts=["запрос"],
n_results=2,
include=["documents", "metadatas", "distances"]
)Теперь вы умеете работать с Chroma. Следующий шаг — Qdrant:
my_chroma_search.py с поисковиком на ChromaГотовы? Открывайте qdrant!
Далее: Qdrant: полноценная векторная БД