feat(core): add set_defaults() and langsmith_metadata in configure path

Add LangChainTracer.set_defaults() for per-key "set if absent" metadata
merging, and thread langsmith_metadata through CallbackManager.configure()
so callers can inject default metadata that only applies to LangChainTracer
handlers without affecting other callback handlers.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

libs/core/langchain_core/callbacks/manager.py

@@ -7,7 +7,7 @@ import atexit
 import functools
 import logging
 from abc import ABC, abstractmethod
-from collections.abc import Callable
+from collections.abc import Callable, Mapping
 from concurrent.futures import ThreadPoolExecutor
 from contextlib import asynccontextmanager, contextmanager
 from contextvars import copy_context
@@ -1614,6 +1614,8 @@ class CallbackManager(BaseCallbackManager):
         local_tags: list[str] | None = None,
         inheritable_metadata: dict[str, Any] | None = None,
         local_metadata: dict[str, Any] | None = None,
+        *,
+        langsmith_metadata: Mapping[str, str] | None = None,
     ) -> CallbackManager:
         """Configure the callback manager.
 
@@ -1625,6 +1627,8 @@ class CallbackManager(BaseCallbackManager):
             local_tags: The local tags.
             inheritable_metadata: The inheritable metadata.
             local_metadata: The local metadata.
+            langsmith_metadata: Default metadata applied to any
+                `LangChainTracer` handlers via `set_defaults`.
 
         Returns:
             The configured callback manager.
@@ -1638,6 +1642,7 @@ class CallbackManager(BaseCallbackManager):
             inheritable_metadata,
             local_metadata,
             verbose=verbose,
+            langsmith_metadata=langsmith_metadata,
         )
 
 
@@ -2134,6 +2139,8 @@ class AsyncCallbackManager(BaseCallbackManager):
         local_tags: list[str] | None = None,
         inheritable_metadata: dict[str, Any] | None = None,
         local_metadata: dict[str, Any] | None = None,
+        *,
+        langsmith_metadata: Mapping[str, str] | None = None,
     ) -> AsyncCallbackManager:
         """Configure the async callback manager.
 
@@ -2145,6 +2152,8 @@ class AsyncCallbackManager(BaseCallbackManager):
             local_tags: The local tags.
             inheritable_metadata: The inheritable metadata.
             local_metadata: The local metadata.
+            langsmith_metadata: Default metadata applied to any
+                `LangChainTracer` handlers via `set_defaults`.
 
         Returns:
             The configured async callback manager.
@@ -2158,6 +2167,7 @@ class AsyncCallbackManager(BaseCallbackManager):
             inheritable_metadata,
             local_metadata,
             verbose=verbose,
+            langsmith_metadata=langsmith_metadata,
         )
 
 
@@ -2304,6 +2314,7 @@ def _configure(
     local_metadata: dict[str, Any] | None = None,
     *,
     verbose: bool = False,
+    langsmith_metadata: Mapping[str, str] | None = None,
 ) -> T:
     """Configure the callback manager.
 
@@ -2316,6 +2327,8 @@ def _configure(
         inheritable_metadata: The inheritable metadata.
         local_metadata: The local metadata.
         verbose: Whether to enable verbose mode.
+        langsmith_metadata: Default metadata applied to any
+            `LangChainTracer` handlers via `set_defaults`.
 
     Raises:
         RuntimeError: If `LANGCHAIN_TRACING` is set but `LANGCHAIN_TRACING_V2` is not.
@@ -2479,6 +2492,10 @@ def _configure(
                 for handler in callback_manager.handlers
             ):
                 callback_manager.add_handler(var_handler, inheritable)
+    if langsmith_metadata:
+        for handler in callback_manager.handlers:
+            if isinstance(handler, LangChainTracer):
+                handler.set_defaults(metadata=langsmith_metadata)
     return callback_manager
 
 

libs/core/langchain_core/tracers/langchain.py

@@ -157,7 +157,24 @@ class LangChainTracer(BaseTracer):
         self.tags = tags or []
         self.latest_run: Run | None = None
         self.run_has_token_event_map: dict[str, bool] = {}
-        self.tracing_metadata = metadata
+        self.tracing_metadata: dict[str, str] | None = (
+            dict(metadata) if metadata is not None else None
+        )
+
+    def set_defaults(self, *, metadata: Mapping[str, str] | None = None) -> None:
+        """Set default tracer values, only filling in keys not already present.
+
+        Args:
+            metadata: Default metadata to include on runs. Keys already present
+                in `tracing_metadata` are not overwritten.
+        """
+        if metadata is not None:
+            if self.tracing_metadata is None:
+                self.tracing_metadata = dict(metadata)
+            else:
+                for k, v in metadata.items():
+                    if k not in self.tracing_metadata:
+                        self.tracing_metadata[k] = v
 
     def _start_trace(self, run: Run) -> None:
         if self.project_name:

libs/core/tests/unit_tests/runnables/test_tracing_interops.py

@@ -13,6 +13,7 @@ from langsmith.run_helpers import tracing_context
 from langsmith.utils import get_env_var
 
 from langchain_core.callbacks.base import BaseCallbackHandler
+from langchain_core.callbacks.manager import CallbackManager
 from langchain_core.runnables.base import RunnableLambda, RunnableParallel
 from langchain_core.tracers.langchain import LangChainTracer
 
@@ -678,3 +679,91 @@ class TestTracerMetadataThroughInvoke:
         assert len(posts) == 1
         md = posts[0].get("extra", {}).get("metadata", {})
         assert md["config_key"] == "config_val"
+
+
+class TestLangsmithMetadataInConfigure:
+    """Tests for `langsmith_metadata` parameter in `CallbackManager.configure()`."""
+
+    def test_langsmith_metadata_applied_via_configure(self) -> None:
+        """langsmith_metadata flows through configure to LangChainTracer."""
+        tracer = _create_tracer_with_mocked_client()
+        cm = CallbackManager.configure(
+            inheritable_callbacks=[tracer],
+            langsmith_metadata={"env": "prod", "service": "api"},
+        )
+        # The tracer should have set_defaults called with the metadata
+        lc_tracers = [h for h in cm.handlers if isinstance(h, LangChainTracer)]
+        assert len(lc_tracers) == 1
+        assert lc_tracers[0].tracing_metadata == {"env": "prod", "service": "api"}
+
+    def test_langsmith_metadata_does_not_overwrite_tracer_metadata(self) -> None:
+        """Tracer's own metadata takes precedence over langsmith_metadata."""
+        tracer = _create_tracer_with_mocked_client(metadata={"env": "staging"})
+        CallbackManager.configure(
+            inheritable_callbacks=[tracer],
+            langsmith_metadata={"env": "prod", "service": "api"},
+        )
+        assert tracer.tracing_metadata == {"env": "staging", "service": "api"}
+
+    def test_langsmith_metadata_end_to_end(self) -> None:
+        """langsmith_metadata in configure propagates to posted runs."""
+        tracer = _create_tracer_with_mocked_client()
+
+        @RunnableLambda
+        def my_func(x: int) -> int:
+            return x
+
+        # Use langsmith_metadata through the config callbacks path
+        cm = CallbackManager.configure(
+            inheritable_callbacks=[tracer],
+            langsmith_metadata={"env": "prod"},
+        )
+        my_func.invoke(1, {"callbacks": cm})
+
+        posts = _get_posts(tracer.client)
+        assert len(posts) == 1
+        md = posts[0].get("extra", {}).get("metadata", {})
+        assert md["env"] == "prod"
+
+    def test_langsmith_metadata_does_not_affect_non_tracer_handlers(self) -> None:
+        """langsmith_metadata only applies to LangChainTracer, not other handlers."""
+        tracer = _create_tracer_with_mocked_client()
+
+        received_metadata: list[dict[str, Any]] = []
+
+        class MetadataCapture(BaseCallbackHandler):
+            def on_chain_start(self, *_args: Any, **kwargs: Any) -> None:
+                received_metadata.append(dict(kwargs.get("metadata", {})))
+
+        capture = MetadataCapture()
+        cm = CallbackManager.configure(
+            inheritable_callbacks=[tracer, capture],
+            langsmith_metadata={"tracer_only": "yes"},
+        )
+
+        @RunnableLambda
+        def my_func(x: int) -> int:
+            return x
+
+        my_func.invoke(1, {"callbacks": cm})
+
+        # Non-tracer handler should NOT see langsmith_metadata
+        assert len(received_metadata) >= 1
+        for md in received_metadata:
+            assert "tracer_only" not in md
+
+        # But the tracer's posted runs SHOULD have it
+        posts = _get_posts(tracer.client)
+        assert len(posts) >= 1
+        for post in posts:
+            post_md = post.get("extra", {}).get("metadata", {})
+            assert post_md["tracer_only"] == "yes"
+
+    def test_no_langsmith_metadata_is_noop(self) -> None:
+        """Passing langsmith_metadata=None does not alter tracer state."""
+        tracer = _create_tracer_with_mocked_client()
+        CallbackManager.configure(
+            inheritable_callbacks=[tracer],
+            langsmith_metadata=None,
+        )
+        assert tracer.tracing_metadata is None

libs/core/tests/unit_tests/tracers/test_langchain.py

@@ -805,3 +805,40 @@ class TestPatchMissingMetadata:
 
         assert run.metadata["env"] == "staging"
         assert run.metadata["extra"] == "from_tracer"
+
+
+class TestSetDefaults:
+    """Tests for `LangChainTracer.set_defaults()`."""
+
+    @staticmethod
+    def _make_tracer(
+        metadata: dict[str, str] | None = None,
+    ) -> LangChainTracer:
+        client = unittest.mock.MagicMock(spec=Client)
+        client.tracing_queue = None
+        return LangChainTracer(client=client, metadata=metadata)
+
+    def test_sets_metadata_when_none(self) -> None:
+        """Fills in metadata when tracer has no prior metadata."""
+        tracer = self._make_tracer()
+        tracer.set_defaults(metadata={"env": "prod"})
+        assert tracer.tracing_metadata == {"env": "prod"}
+
+    def test_does_not_overwrite_existing_keys(self) -> None:
+        """Existing keys are preserved; only missing keys are added."""
+        tracer = self._make_tracer(metadata={"env": "staging"})
+        tracer.set_defaults(metadata={"env": "prod", "service": "api"})
+        assert tracer.tracing_metadata == {"env": "staging", "service": "api"}
+
+    def test_noop_when_defaults_is_none(self) -> None:
+        """No-op when metadata=None is passed."""
+        tracer = self._make_tracer(metadata={"env": "prod"})
+        tracer.set_defaults(metadata=None)
+        assert tracer.tracing_metadata == {"env": "prod"}
+
+    def test_multiple_calls_accumulate(self) -> None:
+        """Successive calls fill in disjoint keys."""
+        tracer = self._make_tracer()
+        tracer.set_defaults(metadata={"a": "1"})
+        tracer.set_defaults(metadata={"b": "2", "a": "overwrite"})
+        assert tracer.tracing_metadata == {"a": "1", "b": "2"}