openinference-instrumentation-llama-index

import os from openinference.instrumentation.llama_index import LlamaIndexInstrumentor from openinference.semconv.resource import ResourceAttributes from opentelemetry import trace as trace_api from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import SimpleSpanProcessor from llama_index.core import VectorStoreIndex, SimpleDirectoryReader from llama_index.llms.openai import OpenAI from llama_index.embeddings.openai import OpenAIEmbedding from llama_index.core import Settings # Set up OpenTelemetry for tracing def setup_tracing(): collector_endpoint = os.environ.get("COLLECTOR_ENDPOINT", "http://localhost:6006/v1/traces") resource = Resource(attributes={ResourceAttributes.PROJECT_NAME: "llama-index-demo"}) tracer_provider = TracerProvider(resource=resource) span_exporter = OTLPSpanExporter(endpoint=collector_endpoint) span_processor = SimpleSpanProcessor(span_exporter=span_exporter) tracer_provider.add_span_processor(span_processor=span_processor) trace_api.set_tracer_provider(tracer_provider=tracer_provider) LlamaIndexInstrumentor().instrument() print("🔭 OpenInference LlamaIndex instrumentation enabled.") # Main application logic def main(): setup_tracing() # Ensure OPENAI_API_KEY is set if not os.environ.get("OPENAI_API_KEY"): print("Please set the OPENAI_API_KEY environment variable.") return # Configure LLM and embeddings for LlamaIndex Settings.llm = OpenAI(model="gpt-3.5-turbo", temperature=0.1) Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small") # Create a dummy data directory and file for demonstration if not os.path.exists("data"): os.makedirs("data") with open("data/sample.txt", "w") as f: f.write("The quick brown fox jumps over the lazy dog. This is a sample document for LlamaIndex.") # Load documents and create an index documents = SimpleDirectoryReader('data').load_data() index = VectorStoreIndex.from_documents(documents) # Query the index query_engine = index.as_query_engine() response = query_engine.query("What did the fox do?") print(f"Response: {response}") print("\nVisit your OpenTelemetry collector (e.g., Phoenix at http://localhost:6006) to see traces.") if __name__ == "__main__": main()