OpenTelemetry ChromaDB Instrumentation

0.58.0 · active · verified Fri Apr 10

This library provides OpenTelemetry auto-instrumentation for the ChromaDB vector database client, allowing developers to trace operations like collection creation, document addition, and queries. It's part of the OpenLLMetry project, ensuring compatibility with evolving GenAI semantic conventions. It is actively maintained with frequent releases, currently at version 0.58.0.

Warnings

Install

Imports

Quickstart

This quickstart demonstrates how to set up OpenTelemetry with `opentelemetry-instrumentation-chromadb`. It configures a simple console exporter, initializes the ChromaDB instrumentor, and then performs basic ChromaDB operations (create, add, query, delete collection) using an in-memory client, with traces printed to the console.

import os
from opentelemetry import trace
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import ConsoleSpanExporter, SimpleSpanProcessor
from opentelemetry.instrumentation.chromadb import ChromaDBInstrumentor
import chromadb

# Configure OpenTelemetry
resource = Resource.create({"service.name": "chromadb-app-instrumented"})
provider = TracerProvider(resource=resource)
processor = SimpleSpanProcessor(ConsoleSpanExporter())
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)

# Initialize ChromaDB instrumentation BEFORE using chromadb client
ChromaDBInstrumentor().instrument()

# Use ChromaDB (in-memory client for quickstart)
client = chromadb.Client()

try:
    collection_name = "my_vector_collection"
    collection = client.create_collection(name=collection_name)
    print(f"Collection '{collection_name}' created.")

    collection.add(
        documents=["This is a document about dogs.", "This is a document about cats."],
        metadatas=[{"source": "wiki"}, {"source": "wiki"}],
        ids=["doc1", "doc2"]
    )
    print("Documents added.")

    results = collection.query(
        query_texts=["What are dogs?"],
        n_results=1
    )
    print(f"Query results: {results}")

    client.delete_collection(name=collection_name)
    print(f"Collection '{collection_name}' deleted.")

except Exception as e:
    print(f"An error occurred: {e}")

finally:
    provider.shutdown()

view raw JSON →