feat: OTEL_ENABLED mode sends to both otel and ls, while allowing otel only through OTEL_ONLY var (#1762)

# Hybrid Tracing: Overview

### Before: Choose One
```
LANGSMITH_OTEL_ENABLED = "true"
traceable_function() 
    │
    ▼
┌─────────────┐
│ Background  │ 
│   Thread    │
└─────┬───────┘
      │
      ▼
   Either:
 ┌──────────┐  OR  ┌─────────────┐
 │   OTEL   │      │ LangSmith   │
 │ Endpoint │      │  Endpoint   │
 └──────────┘      └─────────────┘
```

### After: 
Default behavior of LANGSMITH_OTEL_ENABLED: Get Both
```

traceable_function() 
    │
    ▼
LANGSMITH_OTEL_ENABLED = "true"
┌─────────────┐
│ Background  
│   Thread    │
└─────┬───────┘
      │
      ▼
   Both:
 ┌──────────┐ AND ┌─────────────┐
 │   OTEL   │     │ LangSmith   │
 │ Endpoint │     │  Endpoint   │
 └──────────┘     └─────────────┘
```

User can do Otel-only if insist to, through additional env var
LANGSMITH_OTEL_ONLY
```

traceable_function() 
    │
    ▼
LANGSMITH_OTEL_ENABLED = "true" and LANGSMITH_OTEL_ONLY = "true"
┌─────────────┐
│ Background  
│   Thread    │
└─────┬───────┘
      │
      ▼
   ONLY OTEL:
 ┌──────────┐ 
 │   OTEL   │     
 │ Endpoint │   
 └──────────┘    
```

## Single Function Call → Dual Observability
###  ** Increase Tracing Coverage**
- Get the best of both worlds
- No more choosing between tools
- Future-proof observability

###  **Risk Mitigation** 
- If one service is down, you still have traces
- Backup observability data

## Technical detail for reviewer
- A new `_hybrid_tracing_thread_handle_batch()` to make sure call
task_done() once per item, regardless of how many exporters succeeded
- Sub-thread spawning logic also adopted this. 
- Future improvement can work on the special case of feedback handling,
and whether needs to tune up performance of these two parallel exports
by using two worker threads etc.

## Test plan
- 11 test cases covering three main scenarios: hybrid mode
(OTEL_ENABLED=true) that exports to both OTEL and LangSmith
simultaneously, OTEL-only mode (OTEL_ONLY=true) that exports exclusively
to OTEL, and LangSmith-only mode (default) that exports only to
LangSmith.
- Also i did a local E2E test, setting up a jaegar and tested the (three
mode) * 2 way (from
https://docs.smith.langchain.com/observability/how_to_guides/trace_langchain_with_otel#configuring-alternate-otlp-endpoints)
and confirmed results through manually inspection , i have
https://docs.google.com/document/d/1hgS-YPAqvavbkYN0DyrFkjd91lD-n0n6kAedAfAiCFM/edit?tab=t.0
which contains all the scenarios i tested

Author: EugeneJinXin

python/langsmith/__init__.py

@@ -20,7 +20,8 @@
     from langsmith.utils import ContextThreadPoolExecutor
 
 # Avoid calling into importlib on every call to __version__
-__version__ = "0.3.45"
+
+__version__ = "0.4.0"
 version = __version__  # for backwards compatibility
 
 

python/langsmith/_internal/_background_thread.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import concurrent.futures as cf
+import copy
 import functools
 import io
 import logging
@@ -48,8 +49,8 @@ class TracingQueueItem:
 
     Attributes:
         priority (str): The priority of the item.
-        action (str): The action associated with the item.
         item (Any): The item itself.
+        otel_context (Optional[Context]): The OTEL context of the item.
     """
 
     priority: str
@@ -158,9 +159,27 @@ def _tracing_thread_handle_batch(
     tracing_queue: Queue,
     batch: list[TracingQueueItem],
     use_multipart: bool,
+    mark_task_done: bool = True,
+    ops: Optional[
+        list[Union[SerializedRunOperation, SerializedFeedbackOperation]]
+    ] = None,
 ) -> None:
+    """Handle a batch of tracing queue items by sending them to LangSmith.
+
+    Args:
+        client: The LangSmith client to use for sending data.
+        tracing_queue: The queue containing tracing items (used for task_done calls).
+        batch: List of tracing queue items to process.
+        use_multipart: Whether to use multipart endpoint for sending data.
+        mark_task_done: Whether to mark queue tasks as done after processing.
+            Set to False when called from parallel execution to avoid double counting.
+        ops: Pre-combined serialized operations to use instead of combining from batch.
+            If None, operations will be combined from the batch items.
+    """
     try:
-        ops = combine_serialized_queue_operations([item.item for item in batch])
+        if ops is None:
+            ops = combine_serialized_queue_operations([item.item for item in batch])
+
         if use_multipart:
             client._multipart_ingest_ops(ops)
         else:
@@ -180,22 +199,45 @@ def _tracing_thread_handle_batch(
             "Error details:",
             exc_info=True,
         )
-        # exceptions are logged elsewhere, but we need to make sure the
-        # background thread continues to run
-        pass
     finally:
-        for _ in batch:
-            tracing_queue.task_done()
+        if mark_task_done:
+            for _ in batch:
+                try:
+                    tracing_queue.task_done()
+                except ValueError as e:
+                    if "task_done() called too many times" in str(e):
+                        # This can happen during shutdown when multiple threads
+                        # process the same queue items. It's harmless.
+                        logger.debug(
+                            f"Ignoring harmless task_done error during shutdown: {e}"
+                        )
+                    else:
+                        raise
 
 
 def _otel_tracing_thread_handle_batch(
     client: Client,
     tracing_queue: Queue,
     batch: list[TracingQueueItem],
+    mark_task_done: bool = True,
+    ops: Optional[
+        list[Union[SerializedRunOperation, SerializedFeedbackOperation]]
+    ] = None,
 ) -> None:
-    """Handle a batch of tracing queue items by exporting them to OTEL."""
+    """Handle a batch of tracing queue items by exporting them to OTEL.
+
+    Args:
+        client: The LangSmith client containing the OTEL exporter.
+        tracing_queue: The queue containing tracing items (used for task_done calls).
+        batch: List of tracing queue items to process.
+        mark_task_done: Whether to mark queue tasks as done after processing.
+            Set to False when called from parallel execution to avoid double counting.
+        ops: Pre-combined serialized operations to use instead of combining from batch.
+            If None, operations will be combined from the batch items.
+    """
     try:
-        ops = combine_serialized_queue_operations([item.item for item in batch])
+        if ops is None:
+            ops = combine_serialized_queue_operations([item.item for item in batch])
 
         run_ops = [op for op in ops if isinstance(op, SerializedRunOperation)]
         otel_context_map = {
@@ -215,17 +257,129 @@ def _otel_tracing_thread_handle_batch(
 
     except Exception:
         logger.error(
-            "LangSmith tracing error: Failed to submit OTEL trace data.\n"
+            "OTEL tracing error: Failed to submit trace data.\n"
             "This does not affect your application's runtime.\n"
             "Error details:",
             exc_info=True,
         )
-        # Exceptions are logged elsewhere, but we need to make sure the
-        # background thread continues to run
     finally:
-        # Mark all items in the batch as done
+        if mark_task_done:
+            for _ in batch:
+                try:
+                    tracing_queue.task_done()
+                except ValueError as e:
+                    if "task_done() called too many times" in str(e):
+                        # This can happen during shutdown when multiple threads
+                        # process the same queue items. It's harmless.
+                        logger.debug(
+                            f"Ignoring harmless task_done error during shutdown: {e}"
+                        )
+                    else:
+                        raise
+
+
+def _hybrid_tracing_thread_handle_batch(
+    client: Client,
+    tracing_queue: Queue,
+    batch: list[TracingQueueItem],
+    use_multipart: bool,
+    mark_task_done: bool = True,
+) -> None:
+    """Handle a batch of tracing queue items by sending to both both LangSmith and OTEL.
+
+    Args:
+        client: The LangSmith client to use for sending data.
+        tracing_queue: The queue containing tracing items (used for task_done calls).
+        batch: List of tracing queue items to process.
+        use_multipart: Whether to use multipart endpoint for LangSmith.
+        mark_task_done: Whether to mark queue tasks as done after processing.
+            Set to False primarily for testing when items weren't actually queued.
+    """
+    # Combine operations once to avoid race conditions
+    ops = combine_serialized_queue_operations([item.item for item in batch])
+
+    # Create copies for each thread to avoid shared mutation
+    langsmith_ops = copy.deepcopy(ops)
+    otel_ops = copy.deepcopy(ops)
+
+    # Use ThreadPoolExecutor for parallel execution
+    with cf.ThreadPoolExecutor(max_workers=2) as executor:
+        # Submit both tasks
+        future_langsmith = executor.submit(
+            _tracing_thread_handle_batch,
+            client,
+            tracing_queue,
+            batch,
+            use_multipart,
+            False,  # Don't mark tasks done - we'll do it once at the end
+            langsmith_ops,
+        )
+        future_otel = executor.submit(
+            _otel_tracing_thread_handle_batch,
+            client,
+            tracing_queue,
+            batch,
+            False,  # Don't mark tasks done - we'll do it once at the end
+            otel_ops,
+        )
+
+        # Wait for both to complete
+        future_langsmith.result()
+        future_otel.result()
+
+    # Mark all tasks as done once, only if requested
+    if mark_task_done:
         for _ in batch:
-            tracing_queue.task_done()
+            try:
+                tracing_queue.task_done()
+            except ValueError as e:
+                if "task_done() called too many times" in str(e):
+                    # This can happen during shutdown when multiple threads
+                    # process the same queue items. It's harmless.
+                    logger.debug(
+                        f"Ignoring harmless task_done error during shutdown: {e}"
+                    )
+                else:
+                    raise
+
+
+def _is_using_internal_otlp_provider(client: Client) -> bool:
+    """Check if client is using LangSmith's internal OTLP provider.
+
+    Returns True if using LangSmith's internal provider, False if user
+    provided their own.
+    """
+    if not hasattr(client, "otel_exporter") or client.otel_exporter is None:
+        return False
+
+    try:
+        # Use OpenTelemetry's standard API to get the global TracerProvider
+        # Check if OTEL is available
+        if not ls_utils.is_truish(ls_utils.get_env_var("OTEL_ENABLED")):
+            return False
+
+        # Get the global TracerProvider and check its resource attributes
+        from opentelemetry import trace  # type: ignore[import]
+
+        tracer_provider = trace.get_tracer_provider()
+        if hasattr(tracer_provider, "resource") and hasattr(
+            tracer_provider.resource, "attributes"
+        ):
+            is_internal = tracer_provider.resource.attributes.get(
+                "langsmith.internal_provider", False
+            )
+            logger.debug(
+                f"TracerProvider resource check: "
+                f"langsmith.internal_provider={is_internal}"
+            )
+            return is_internal
+
+        return False
+    except Exception as e:
+        logger.debug(
+            f"Could not determine TracerProvider type: {e}, assuming user-provided"
+        )
+        return False
 
 
 def get_size_limit_from_env() -> Optional[int]:
@@ -267,6 +421,29 @@ def _ensure_ingest_config(
         return default_config
 
 
+def get_tracing_mode() -> tuple[bool, bool]:
+    """Get the current tracing mode configuration.
+
+    Returns:
+        tuple[bool, bool]:
+            - hybrid_otel_and_langsmith: True if both OTEL and LangSmith tracing
+              are enabled, which is default behavior if OTEL_ENABLED is set to
+              true and OTEL_ONLY is not set to true
+            - is_otel_only: True if only OTEL tracing is enabled
+    """
+    otel_enabled = ls_utils.is_truish(ls_utils.get_env_var("OTEL_ENABLED"))
+    otel_only = ls_utils.is_truish(ls_utils.get_env_var("OTEL_ONLY"))
+
+    # If OTEL is not enabled, neither mode should be active
+    if not otel_enabled:
+        return False, False
+
+    hybrid_otel_and_langsmith = not otel_only
+    is_otel_only = otel_only
+
+    return hybrid_otel_and_langsmith, is_otel_only
+
+
 def tracing_control_thread_func(client_ref: weakref.ref[Client]) -> None:
     client = client_ref()
     if client is None:
@@ -351,21 +528,41 @@ def keep_thread_active() -> bool:
             )
             sub_threads.append(new_thread)
             new_thread.start()
+
+        hybrid_otel_and_langsmith, is_otel_only = get_tracing_mode()
         if next_batch := _tracing_thread_drain_queue(tracing_queue, limit=size_limit):
-            if client.otel_exporter is not None:
+            if hybrid_otel_and_langsmith:
+                # Hybrid mode: both OTEL and LangSmith
+                _hybrid_tracing_thread_handle_batch(
+                    client, tracing_queue, next_batch, use_multipart
+                )
+            elif is_otel_only:
+                # OTEL-only mode
                 _otel_tracing_thread_handle_batch(client, tracing_queue, next_batch)
             else:
+                # LangSmith-only mode
                 _tracing_thread_handle_batch(
                     client, tracing_queue, next_batch, use_multipart
                 )
 
-    # drain the queue on exit
+    # drain the queue on exit - apply same logic
+    hybrid_otel_and_langsmith, is_otel_only = get_tracing_mode()
     while next_batch := _tracing_thread_drain_queue(
         tracing_queue, limit=size_limit, block=False
     ):
-        if client.otel_exporter is not None:
+        if hybrid_otel_and_langsmith:
+            # Hybrid mode cleanup
+            logger.debug("Hybrid mode cleanup")
+            _hybrid_tracing_thread_handle_batch(
+                client, tracing_queue, next_batch, use_multipart
+            )
+        elif is_otel_only:
+            # OTEL-only cleanup
+            logger.debug("OTEL-only cleanup")
             _otel_tracing_thread_handle_batch(client, tracing_queue, next_batch)
         else:
+            # LangSmith-only cleanup
+            logger.debug("LangSmith-only cleanup")
             _tracing_thread_handle_batch(
                 client, tracing_queue, next_batch, use_multipart
             )
@@ -378,7 +575,7 @@ def tracing_control_thread_func_compress_parallel(
     client = client_ref()
     if client is None:
         return
-
+    logger.debug("Tracing control thread func compress parallel called")
     if (
         client.compressed_traces is None
         or client._data_available_event is None
@@ -542,22 +739,41 @@ def _tracing_sub_thread_func(
     ):
         if next_batch := _tracing_thread_drain_queue(tracing_queue, limit=size_limit):
             seen_successive_empty_queues = 0
-            if client.otel_exporter is not None:
+
+            hybrid_otel_and_langsmith, is_otel_only = get_tracing_mode()
+            if hybrid_otel_and_langsmith:
+                # Hybrid mode: both OTEL and LangSmith
+                _hybrid_tracing_thread_handle_batch(
+                    client, tracing_queue, next_batch, use_multipart
+                )
+            elif is_otel_only:
+                # OTEL-only mode
                 _otel_tracing_thread_handle_batch(client, tracing_queue, next_batch)
             else:
+                # LangSmith-only mode
                 _tracing_thread_handle_batch(
                     client, tracing_queue, next_batch, use_multipart
                 )
         else:
             seen_successive_empty_queues += 1
 
-    # drain the queue on exit
+    # drain the queue on exit - apply same logic
+    hybrid_otel_and_langsmith, is_otel_only = get_tracing_mode()
     while next_batch := _tracing_thread_drain_queue(
         tracing_queue, limit=size_limit, block=False
     ):
-        if client.otel_exporter is not None:
+        if hybrid_otel_and_langsmith:
+            # Hybrid mode cleanup
+            _hybrid_tracing_thread_handle_batch(
+                client, tracing_queue, next_batch, use_multipart
+            )
+        elif is_otel_only:
+            # OTEL-only cleanup
+            logger.debug("OTEL-only cleanup")
             _otel_tracing_thread_handle_batch(client, tracing_queue, next_batch)
         else:
+            # LangSmith-only cleanup
+            logger.debug("LangSmith-only cleanup")
             _tracing_thread_handle_batch(
                 client, tracing_queue, next_batch, use_multipart
             )

python/langsmith/_internal/otel/_otel_client.py

@@ -63,7 +63,13 @@ def get_otlp_tracer_provider() -> "TracerProvider":
         os.environ["OTEL_EXPORTER_OTLP_HEADERS"] = headers
 
     service_name = os.environ.get("OTEL_SERVICE_NAME", "langsmith")
-    resource = Resource(attributes={SERVICE_NAME: service_name})
+    resource = Resource(
+        attributes={
+            SERVICE_NAME: service_name,
+            # Marker to identify LangSmith's internal provider
+            "langsmith.internal_provider": True,
+        }
+    )
 
     tracer_provider = TracerProvider(resource=resource)
 

python/langsmith/run_trees.py

@@ -217,7 +217,7 @@ def set(
         by the @traceable decorator.
 
         If your LangChain or LangGraph versions are sufficiently up-to-date,
-        this will also override the default behavior LangChainTracer.
+        this will also override the default behavior of LangChainTracer.
 
         Args:
             inputs: The inputs to set.