opentelemetry-instrumentation-anthropic

import os from anthropic import Anthropic from opentelemetry.instrumentation.anthropic import AnthropicInstrumentor from opentelemetry import trace from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter # --- OpenTelemetry Setup --- # 1. Configure the OpenTelemetry TracerProvider resource = Resource.create({"service.name": "anthropic-llm-app"}) provider = TracerProvider(resource=resource) trace.set_tracer_provider(provider) # 2. Configure an OTLP exporter to send traces (e.g., to an OTLP collector or a service like SigNoz/Arize) # Default OTLP HTTP endpoint is http://localhost:4318/v1/traces otlp_exporter = OTLPSpanExporter() span_processor = BatchSpanProcessor(otlp_exporter) provider.add_span_processor(span_processor) # 3. Instrument the Anthropic library AnthropicInstrumentor().instrument() # --- Anthropic API Call --- ANTHROPIC_API_KEY = os.environ.get("ANTHROPIC_API_KEY", "YOUR_ANTHROPIC_API_KEY") if ANTHROPIC_API_KEY == "YOUR_ANTHROPIC_API_KEY": print("WARNING: ANTHROPIC_API_KEY not set or placeholder. API calls will fail without a valid key.") else: try: client = Anthropic(api_key=ANTHROPIC_API_KEY) print("Making an Anthropic API call...") response = client.messages.create( model="claude-3-opus-20240229", # Or another suitable model max_tokens=100, messages=[ {"role": "user", "content": "Explain the concept of quantum entanglement in a sentence."} ], ) print("Anthropic API call successful.") print(f"Response: {response.content[0].text[:50]}...") except Exception as e: print(f"Error during Anthropic API call: {e}") # Ensure all spans are exported before the application exits provider.force_flush()