排查追踪嵌套问题

​

Python

​

使用 asyncio 进行上下文传播

​

原因

​

解决方法

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,
       # config 参数（在下面的 parent_runnable 中存在）是可选的
   ):
       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 直接传递运行树：
   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,
       # 运行树可以由装饰器自动填充
       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 装饰的函数在通过 "config" 传入时可以直接接受 RunnableConfig
       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):
       # 你可以手动设置追踪上下文
       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 进行上下文传播

​

原因

​

解决方法

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. 手动提供父运行树 或者，你可以手动将父运行树传递给内部函数：
   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()