codekingpro/portable-devtools

1"""OpenTelemetry span processor and exporter for LangSmith."""2 3import logging4import warnings5from typing import Any, Optional6from urllib.parse import urljoin7 8from langsmith import utils as ls_utils9 10try:11    from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter12    from opentelemetry.sdk.trace.export import BatchSpanProcessor13 14    OTEL_AVAILABLE = True15except ImportError:16    warnings.warn(17        "OpenTelemetry packages are not installed. "18        "Install optional OpenTelemetry dependencies with: "19        "pip install langsmith[otel]",20        UserWarning,21        stacklevel=2,22    )23 24    class OTLPSpanExporter:  # type: ignore[no-redef]25        """Mock otlp span exporter class."""26 27        def __init__(self, *args, **kwargs):28            """Mock init method."""29            raise ImportError(30                "OpenTelemetry packages are not installed. "31                "Install optional OpenTelemetry dependencies with: "32                "pip install langsmith[otel]"33            )34 35    class BatchSpanProcessor:  # type: ignore[no-redef]36        """Mock batch span processor class."""37 38        def __init__(self, *args, **kwargs):39            """Mock init method."""40            raise ImportError(41                "OpenTelemetry packages are not installed. "42                "Install optional OpenTelemetry dependencies with: "43                "pip install langsmith[otel]"44            )45 46    class trace:47        """Mock trace class."""48 49        @staticmethod50        def get_tracer_provider():51            """Mock get tracer provider method."""52            raise ImportError(53                "OpenTelemetry packages are not installed. "54                "Install optional OpenTelemetry dependencies with: "55                "pip install langsmith[otel]"56            )57 58    OTEL_AVAILABLE = False59 60LANGSMITH_METADATA_PREFIX = "langsmith.metadata"61 62 63class OtelExporter(OTLPSpanExporter):64    """A subclass of `OTLPSpanExporter` configured for LangSmith.65 66    Environment Variables:67 68    - `LANGSMITH_API_KEY`: Your LangSmith API key.69    - `LANGSMITH_ENDPOINT`: Base URL for LangSmith API (defaults to `https://api.smith.langchain.com`).70    - `LANGSMITH_PROJECT`: Project identifier.71    """72 73    def __init__(74        self,75        url: Optional[str] = None,76        api_key: Optional[str] = None,77        project: Optional[str] = None,78        headers: Optional[dict[str, str]] = None,79        **kwargs,80    ):81        """Initialize the `OtelExporter`.82 83        Args:84            url: OTLP endpoint URL. Defaults to `{LANGSMITH_ENDPOINT}/otel/v1/traces`.85            api_key: LangSmith API key. Defaults to `LANGSMITH_API_KEY` env var.86            parent: Parent identifier (e.g., `'project_name:test'`).87 88                Defaults to `LANGSMITH_PARENT` env var.89            headers: Additional headers to include in requests.90            **kwargs: Additional arguments passed to `OTLPSpanExporter`.91        """92        base_url = ls_utils.get_api_url(None)93        # Ensure base_url ends with / for proper joining94        if not base_url.endswith("/"):95            base_url += "/"96        endpoint = url or urljoin(base_url, "otel/v1/traces")97        api_key = api_key or ls_utils.get_api_key(None)98        project = project or ls_utils.get_tracer_project()99        headers = headers or {}100 101        if not api_key:102            raise ValueError(103                "API key is required. Provide it via api_key parameter or "104                "LANGSMITH_API_KEY environment variable."105            )106 107        if not project:108            project = "default"109            logging.info(110                "No project specified, using default. "111                "Configure with LANGSMITH_PROJECT environment variable or "112                "project parameter."113            )114 115        exporter_headers = {116            "x-api-key": api_key,117            **headers,118        }119 120        if project:121            exporter_headers["Langsmith-Project"] = project122 123        self.project = project124 125        super().__init__(endpoint=endpoint, headers=exporter_headers, **kwargs)126 127 128class OtelSpanProcessor:129    """A span processor for adding LangSmith to OpenTelemetry setups.130 131    This class combines the `OtelExporter` and `BatchSpanProcessor`132    into a single processor that can be added to any `TracerProvider`.133 134    Use this when:135 136    1. You already have OpenTelemetry initialized with other tools137    2. You want to add LangSmith alongside existing OTEL exporters138 139    Examples:140        # Fresh OpenTelemetry setup (LangSmith only):141        from langsmith.integrations.otel import configure142        configure(api_key="your-key", project="your-project")143 144        # Add LangSmith to existing OpenTelemetry setup:145        from opentelemetry import trace146        from langsmith.integrations.otel.processor import OtelSpanProcessor147 148        # Get your existing TracerProvider (already set by other tools)149        provider = trace.get_tracer_provider()150 151        # Add LangSmith processor alongside existing processors152        langsmith_processor = OtelSpanProcessor(153            project="your-project",154        )155        provider.add_span_processor(langsmith_processor)156    """157 158    def __init__(159        self,160        api_key: Optional[str] = None,161        project: Optional[str] = None,162        url: Optional[str] = None,163        headers: Optional[dict[str, str]] = None,164        SpanProcessor: Optional[type] = None,165    ):166        """Initialize the `OtelSpanProcessor`.167 168        Args:169            api_key: LangSmith API key. Defaults to `LANGSMITH_API_KEY` env var.170            project: Project identifier. Defaults to `LANGSMITH_PROJECT` env var.171            url: Base URL for LangSmith API. Defaults to `LANGSMITH_ENDPOINT` env var172                or `https://api.smith.langchain.com`.173            headers: Additional headers to include in requests.174            SpanProcessor: Optional span processor class. Defaults to175                `BatchSpanProcessor`.176        """177        # Create the exporter178        # Convert url to the full endpoint URL that OtelExporter expects179        exporter_url = None180        if url:181            exporter_url = f"{url.rstrip('/')}/otel/v1/traces"182 183        self._exporter = OtelExporter(184            url=exporter_url, api_key=api_key, project=project, headers=headers185        )186 187        # Create the processor chain188        if not OTEL_AVAILABLE:189            raise ImportError(190                "OpenTelemetry packages are not installed. "191                "Install optional OpenTelemetry dependencies with: "192                "pip install langsmith[otel]"193            )194 195        if SpanProcessor is None:196            SpanProcessor = BatchSpanProcessor197 198        self._processor = SpanProcessor(self._exporter)199        self._metadata: dict[str, Any] = {}200 201    def set_metadata(self, metadata: dict[str, Any]) -> None:202        """Set metadata attributes to propagate to all spans.203 204        Use this to ensure metadata like ``thread_id`` appears on every span205        in a trace, not just the root span.  This is required for LangSmith206        features (e.g. the threads view) that query runs by metadata fields.207        """208        self._metadata = metadata.copy()209 210    def on_start(self, span, parent_context=None):211        """Forward span start events to the inner processor."""212        if self._metadata:213            for key, value in self._metadata.items():214                if value is not None:215                    span.set_attribute(f"{LANGSMITH_METADATA_PREFIX}.{key}", value)216        self._processor.on_start(span, parent_context)217 218    def on_end(self, span):219        """Forward span end events to the inner processor."""220        self._processor.on_end(span)221 222    def _on_ending(self, span):223        """Forward span ending events to the inner processor."""224        if hasattr(self._processor, "_on_ending"):225            self._processor._on_ending(span)226 227    def shutdown(self):228        """Shutdown processor."""229        self._processor.shutdown()230 231    def force_flush(self, timeout_millis=30000):232        """Force flush the inner processor."""233        return self._processor.force_flush(timeout_millis)234 235    @property236    def exporter(self):237        """The underlying OtelExporter."""238        return self._exporter239 240    @property241    def processor(self):242        """The underlying span processor."""243        return self._processor244