[Core] Update Tracing Interops (#22318)

LangSmith and LangChain context var handling evolved in parallel since
originally we didn't expect people to want to interweave the decorator
and langchain code.

Once we get a new langsmith release, this PR will let you seemlessly
hand off between @Traceable context and runnable config context so you
can arbitrarily nest code.

It's expected that this fails right now until we get another release of
the SDK

libs/core/langchain_core/callbacks/manager.py

@@ -1918,7 +1918,7 @@ def _configure(
     )
 
     run_tree = get_run_tree_context()
-    parent_run_id = None if run_tree is None else getattr(run_tree, "id")
+    parent_run_id = None if run_tree is None else run_tree.id
     callback_manager = callback_manager_cls(handlers=[], parent_run_id=parent_run_id)
     if inheritable_callbacks or local_callbacks:
         if isinstance(inheritable_callbacks, list) or inheritable_callbacks is None:
@@ -1929,10 +1929,22 @@ def _configure(
                 parent_run_id=parent_run_id,
             )
         else:
+            parent_run_id_ = inheritable_callbacks.parent_run_id
+            # Break ties between the external tracing context and inherited context
+            if parent_run_id is not None:
+                if parent_run_id_ is None:
+                    parent_run_id_ = parent_run_id
+                # If the LC parent has already been reflected
+                # in the run tree, we know the run_tree is either the
+                # same parent or a child of the parent.
+                elif run_tree and str(parent_run_id_) in run_tree.dotted_order:
+                    parent_run_id_ = parent_run_id
+                # Otherwise, we assume the LC context has progressed
+                # beyond the run tree and we should not inherit the parent.
             callback_manager = callback_manager_cls(
                 handlers=inheritable_callbacks.handlers.copy(),
                 inheritable_handlers=inheritable_callbacks.inheritable_handlers.copy(),
-                parent_run_id=inheritable_callbacks.parent_run_id,
+                parent_run_id=parent_run_id_,
                 tags=inheritable_callbacks.tags.copy(),
                 inheritable_tags=inheritable_callbacks.inheritable_tags.copy(),
                 metadata=inheritable_callbacks.metadata.copy(),

libs/core/langchain_core/runnables/base.py

@@ -45,6 +45,7 @@
 from langchain_core.pydantic_v1 import BaseModel, Field
 from langchain_core.runnables.config import (
     RunnableConfig,
+    _set_config_context,
     acall_func_with_variable_args,
     call_func_with_variable_args,
     ensure_config,
@@ -55,7 +56,6 @@
     merge_configs,
     patch_config,
     run_in_executor,
-    var_child_runnable_config,
 )
 from langchain_core.runnables.graph import Graph
 from langchain_core.runnables.schema import StreamEvent
@@ -1503,7 +1503,7 @@ def _call_with_config(
         try:
             child_config = patch_config(config, callbacks=run_manager.get_child())
             context = copy_context()
-            context.run(var_child_runnable_config.set, child_config)
+            context.run(_set_config_context, child_config)
             output = cast(
                 Output,
                 context.run(
@@ -1551,7 +1551,7 @@ async def _acall_with_config(
         try:
             child_config = patch_config(config, callbacks=run_manager.get_child())
             context = copy_context()
-            context.run(var_child_runnable_config.set, child_config)
+            context.run(_set_config_context, child_config)
             coro = acall_func_with_variable_args(
                 func, input, config, run_manager, **kwargs
             )
@@ -1760,7 +1760,7 @@ def _transform_stream_with_config(
             if accepts_run_manager(transformer):
                 kwargs["run_manager"] = run_manager
             context = copy_context()
-            context.run(var_child_runnable_config.set, child_config)
+            context.run(_set_config_context, child_config)
             iterator = context.run(transformer, input_for_transform, **kwargs)  # type: ignore[arg-type]
             if stream_handler := next(
                 (
@@ -1860,7 +1860,7 @@ async def _atransform_stream_with_config(
             if accepts_run_manager(transformer):
                 kwargs["run_manager"] = run_manager
             context = copy_context()
-            context.run(var_child_runnable_config.set, child_config)
+            context.run(_set_config_context, child_config)
             iterator = context.run(transformer, input_for_transform, **kwargs)  # type: ignore[arg-type]
 
             if stream_handler := next(

libs/core/langchain_core/runnables/config.py

@@ -109,6 +109,25 @@ class RunnableConfig(TypedDict, total=False):
 )
 
 
+def _set_config_context(config: RunnableConfig) -> None:
+    """Set the child runnable config + tracing context
+
+    Args:
+        config (RunnableConfig): The config to set.
+    """
+    from langsmith import (
+        RunTree,  # type: ignore
+        run_helpers,  # type: ignore
+    )
+
+    var_child_runnable_config.set(config)
+    if hasattr(RunTree, "from_runnable_config"):
+        # import _set_tracing_context, get_tracing_context
+        rt = RunTree.from_runnable_config(dict(config))
+        tc = run_helpers.get_tracing_context()
+        run_helpers._set_tracing_context({**tc, "parent": rt})
+
+
 def ensure_config(config: Optional[RunnableConfig] = None) -> RunnableConfig:
     """Ensure that a config is a dict with all keys present.
 

libs/core/langchain_core/tools.py

@@ -65,9 +65,9 @@
     ensure_config,
 )
 from langchain_core.runnables.config import (
+    _set_config_context,
     patch_config,
     run_in_executor,
-    var_child_runnable_config,
 )
 from langchain_core.runnables.utils import accepts_context
 
@@ -402,7 +402,7 @@ def run(
                 callbacks=run_manager.get_child(),
             )
             context = copy_context()
-            context.run(var_child_runnable_config.set, child_config)
+            context.run(_set_config_context, child_config)
             parsed_input = self._parse_input(tool_input)
             tool_args, tool_kwargs = self._to_args_and_kwargs(parsed_input)
             observation = (
@@ -502,7 +502,7 @@ async def arun(
                 callbacks=run_manager.get_child(),
             )
             context = copy_context()
-            context.run(var_child_runnable_config.set, child_config)
+            context.run(_set_config_context, child_config)
             coro = (
                 context.run(
                     self._arun, *tool_args, run_manager=run_manager, **tool_kwargs