排查追踪嵌套问题

使用 LangSmith SDK、LangGraph 和 LangChain 进行追踪时，追踪应自动传播正确的上下文，使父追踪内执行的代码在 UI 中呈现于预期位置。

如果你看到子 run 进入单独的追踪（并出现在顶层），可能是由以下已知 "边缘情况" 之一引起的。

Python

以下概述使用 Python 构建时出现 "分裂" 追踪的常见原因。

使用 asyncio 传播上下文

在 Python 版本 < 3.11 中使用异步调用（尤其是流式调用）时，可能会遇到追踪嵌套问题。这是因为 Python 的 asyncio 在 3.11 版本才添加对上下文传递的完整支持。

原因

LangChain 和 LangSmith SDK 使用 contextvars 隐式传播追踪信息。在 Python 3.11 及以上版本中，这可以无缝工作。但在更早的版本（3.8、3.9、3.10）中，asyncio 任务缺少适当的 contextvar 支持，可能导致追踪断开。

解决方法

1. 升级 Python 版本（推荐） 如果可能，升级到 Python 3.11 或更高版本以自动传播上下文。

2. 手动传播上下文 如果无法升级，你需要手动传播追踪上下文。方法因设置而异：

   a) 使用 LangGraph 或 LangChain 将父 config 传递给子调用：

   import asyncio
   from langchain_core.runnables import RunnableConfig, RunnableLambda
   
   @RunnableLambda
   async def my_child_runnable(
       inputs: str,
       # The config arg (present in parent_runnable below) is optional
   ):
       yield "A"
       yield "response"
   
   @RunnableLambda
   async def parent_runnable(inputs: str, config: RunnableConfig):
       async for chunk in my_child_runnable.astream(inputs, config):
           yield chunk
   
   async def main():
       return [val async for val in parent_runnable.astream("call")]
   
   asyncio.run(main())

   b) 直接使用 LangSmith 直接传递 run tree：

   import asyncio
   import langsmith as ls
   
   @ls.traceable
   async def my_child_function(inputs: str):
       yield "A"
       yield "response"
   
   @ls.traceable
   async def parent_function(
       inputs: str,
       # The run tree can be auto-populated by the decorator
       run_tree: ls.RunTree,
   ):
       async for chunk in my_child_function(inputs, langsmith_extra={"parent": run_tree}):
           yield chunk
   
   async def main():
       return [val async for val in parent_function("call")]
   
   asyncio.run(main())

   c) 将装饰代码与 LangGraph/LangChain 结合 使用多种技术组合进行手动交接：

   import asyncio
   import langsmith as ls
   from langchain_core.runnables import RunnableConfig, RunnableLambda
   
   @RunnableLambda
   async def my_child_runnable(inputs: str):
       yield "A"
       yield "response"
   
   @ls.traceable
   async def my_child_function(inputs: str, run_tree: ls.RunTree):
       with ls.tracing_context(parent=run_tree):
           async for chunk in my_child_runnable.astream(inputs):
               yield chunk
   
   @RunnableLambda
   async def parent_runnable(inputs: str, config: RunnableConfig):
       # @traceable decorated functions can directly accept a RunnableConfig when passed in via "config"
       async for chunk in my_child_function(inputs, langsmith_extra={"config": config}):
           yield chunk
   
   @ls.traceable
   async def parent_function(inputs: str, run_tree: ls.RunTree):
       # You can set the tracing context manually
       with ls.tracing_context(parent=run_tree):
           async for chunk in parent_runnable.astream(inputs):
               yield chunk
   
   async def main():
       return [val async for val in parent_function("call")]
   
   asyncio.run(main())

使用 threading 传播上下文

通常，我们会在开始追踪后希望在单个追踪内对子任务应用一些并行处理。Python 标准库的 ThreadPoolExecutor 默认会破坏追踪。

原因

Python 的 contextvars 在新线程中初始为空。以下是两种保持追踪连续性的方法：

解决方法

1. 使用 LangSmith 的 ContextThreadPoolExecutor

   LangSmith 提供了 ContextThreadPoolExecutor，可自动处理上下文传播：

   from langsmith.utils import ContextThreadPoolExecutor
   from langsmith import traceable
   
   @traceable
   def outer_func():
       with ContextThreadPoolExecutor() as executor:
           inputs = [1, 2]
           r = list(executor.map(inner_func, inputs))
   
   @traceable
   def inner_func(x):
       print(x)
   
   outer_func()

2. 手动提供父 run tree

   或者，你可以手动将父 run tree 传递给内部函数：

   from langsmith import traceable, get_current_run_tree
   from concurrent.futures import ThreadPoolExecutor
   
   @traceable
   def outer_func():
       rt = get_current_run_tree()
       with ThreadPoolExecutor() as executor:
           r = list(
               executor.map(
                   lambda x: inner_func(x, langsmith_extra={"parent": rt}), [1, 2]
               )
           )
   
   @traceable
   def inner_func(x):
       print(x)
   
   outer_func()

在这种方法中，我们使用 get_current_run_tree() 获取当前 run tree，并使用 langsmith_extra 参数将其传递给内部函数。

这两种方法都能确保内部函数调用正确聚合到初始追踪栈下，即使它们在单独的线程中执行。

---